Namespace: llm.rag.* — RAG Retrieval Payloads

Module: spanforge.namespaces.retrieval
Added in: 2.0.12 (Phase 13)
Events covered: llm.rag.query, llm.rag.retrieved, llm.rag.generated, llm.rag.session

This namespace covers all payload types emitted during a Retrieval-Augmented Generation (RAG) pipeline trace. Use spanforge.sdk.rag to emit these events without constructing the dataclasses directly.

Classes

RetrievedChunk

A value object representing one retrieved document chunk. Used inside RetrievalResultPayload.chunks.

@dataclass
class RetrievedChunk:
    chunk_id: str
    content_hash: str
    score: float
    source: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

Methods: to_dict() -> dict, from_dict(data: dict) -> RetrievedChunk

RetrievalQueryPayload

Payload for llm.rag.query events. Captures the user query and retriever configuration without storing raw query text.

@dataclass
class RetrievalQueryPayload:
    session_id: str
    query_hash: str
    top_k: int = 5
    retriever_name: str = ""
    embedding_model: str = ""
    namespace: str = ""
    latency_ms: float = 0.0
    filters: dict[str, Any] = field(default_factory=dict)

Methods: to_dict() -> dict, from_dict(data: dict) -> RetrievalQueryPayload

RetrievalResultPayload

Payload for llm.rag.retrieved events. Contains the ordered list of chunks returned from the retriever.

@dataclass
class RetrievalResultPayload:
    session_id: str
    query_hash: str
    chunks: list[RetrievedChunk] = field(default_factory=list)
    total_found: int = 0
    latency_ms: float = 0.0
    status: Literal["ok", "partial", "error", "timeout"] = "ok"
    error_message: str | None = None

Methods: to_dict() -> dict, from_dict(data: dict) -> RetrievalResultPayload

RAGSpanPayload

Payload for llm.rag.generated events. Represents the LLM generation step that consumes retrieved context.

@dataclass
class RAGSpanPayload:
    session_id: str
    span_name: str
    model: str
    chunk_ids_used: list[str] = field(default_factory=list)
    context_tokens: int = 0
    prompt_tokens: int = 0
    output_tokens: int = 0
    latency_ms: float = 0.0
    status: Literal["ok", "error", "timeout"] = "ok"
    grounding_score: float | None = None
    error_message: str | None = None

Methods: to_dict() -> dict, from_dict(data: dict) -> RAGSpanPayload

RAGSessionPayload

Payload for llm.rag.session events. An aggregated summary of a complete RAG session, emitted when sf_rag.end_session() is called.

@dataclass
class RAGSessionPayload:
    session_id: str
    retriever_name: str
    total_queries: int = 0
    total_chunks_retrieved: int = 0
    unique_chunk_ids: list[str] = field(default_factory=list)
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    avg_grounding_score: float | None = None
    total_latency_ms: float = 0.0
    started_at: str = ""
    status: str = "ok"

Methods: to_dict() -> dict, from_dict(data: dict) -> RAGSessionPayload

Usage example

from spanforge.namespaces.retrieval import (
    RetrievedChunk,
    RetrievalQueryPayload,
    RetrievalResultPayload,
    RAGSpanPayload,
    RAGSessionPayload,
)

# Prefer the high-level client instead:
from spanforge.sdk import sf_rag
session_id = sf_rag.trace_query("What is entropy?", top_k=3)

Related

• sf-rag SDK client
• User Guide — RAG Tracing
• Feedback namespace