Span Groups — Per-Hop Metric Localization¶

This notebook demonstrates how to use span_group() to get per-hop groundedness scores in a multi-hop RAG pipeline, instead of one pooled score across all retrieval steps.

The problem¶

A multi-hop RAG pipeline performs several retrieval steps. By default, a groundedness metric pools all retrieved contexts into one score. You can't tell which hop was ungrounded.

The fix¶

Tag each hop's spans with a span_group() label. compute_feedback_by_span_group then runs the metric once per group, giving you per-hop scores.

1. Define a simple multi-hop RAG app¶

In [ ]:

from trulens.core.otel.instrument import instrument, span_group
from trulens.otel.semconv.trace import SpanAttributes


@instrument(
    span_type=SpanAttributes.SpanType.RETRIEVAL,
    attributes={
        SpanAttributes.RETRIEVAL.QUERY_TEXT: "query",
        SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return",
    },
)
def retrieve(query: str) -> list:
    """Simulated retrieval — returns dummy contexts."""
    return [f"Context for: {query}"]


@instrument()
def multi_hop_rag(question: str) -> str:
    """A 3-hop retrieval pipeline."""
    with span_group("hop1"):
        ctx1 = retrieve(question)
    with span_group("hop2"):
        ctx2 = retrieve(f"Follow-up based on {ctx1}")
    with span_group("hop3"):
        ctx3 = retrieve(f"Final refinement from {ctx2}")

    all_context = ctx1 + ctx2 + ctx3
    return f"Answer based on {len(all_context)} contexts"

from trulens.core.otel.instrument import instrument, span_group from trulens.otel.semconv.trace import SpanAttributes @instrument( span_type=SpanAttributes.SpanType.RETRIEVAL, attributes={ SpanAttributes.RETRIEVAL.QUERY_TEXT: "query", SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return", }, ) def retrieve(query: str) -> list: """Simulated retrieval — returns dummy contexts.""" return [f"Context for: {query}"] @instrument() def multi_hop_rag(question: str) -> str: """A 3-hop retrieval pipeline.""" with span_group("hop1"): ctx1 = retrieve(question) with span_group("hop2"): ctx2 = retrieve(f"Follow-up based on {ctx1}") with span_group("hop3"): ctx3 = retrieve(f"Final refinement from {ctx2}") all_context = ctx1 + ctx2 + ctx3 return f"Answer based on {len(all_context)} contexts"

2. Record a trace¶

In [ ]:

from trulens.apps.app import TruApp
from trulens.core.session import TruSession

session = TruSession()
session.reset_database()

tru_app = TruApp(
    multi_hop_rag,
    app_name="MultiHopRAG",
    app_version="v1",
)

with tru_app as recording:
    result = multi_hop_rag("What is TruLens?")

print(result)

from trulens.apps.app import TruApp from trulens.core.session import TruSession session = TruSession() session.reset_database() tru_app = TruApp( multi_hop_rag, app_name="MultiHopRAG", app_version="v1", ) with tru_app as recording: result = multi_hop_rag("What is TruLens?") print(result)

3. Inspect the span groups¶

Each retrieval span now carries a distinct SPAN_GROUPS label.

In [ ]:

import json

import pandas as pd
import sqlalchemy as sa

session.force_flush()

db = session.connector.db
with db.session.begin() as db_session:
    events = pd.read_sql(
        sa.select(db.orm.Event).order_by(db.orm.Event.start_timestamp),
        db_session.bind,
    )


def _parse(val):
    return val if isinstance(val, dict) else json.loads(val)


for _, row in events.iterrows():
    attrs = _parse(row["record_attributes"])
    record = _parse(row["record"])
    groups = attrs.get(SpanAttributes.SPAN_GROUPS, "(none)")
    span_type = attrs.get(SpanAttributes.SPAN_TYPE, "")
    print(f"{record['name']:50s}  type={span_type:15s}  groups={groups}")

import json import pandas as pd import sqlalchemy as sa session.force_flush() db = session.connector.db with db.session.begin() as db_session: events = pd.read_sql( sa.select(db.orm.Event).order_by(db.orm.Event.start_timestamp), db_session.bind, ) def _parse(val): return val if isinstance(val, dict) else json.loads(val) for _, row in events.iterrows(): attrs = _parse(row["record_attributes"]) record = _parse(row["record"]) groups = attrs.get(SpanAttributes.SPAN_GROUPS, "(none)") span_type = attrs.get(SpanAttributes.SPAN_TYPE, "") print(f"{record['name']:50s} type={span_type:15s} groups={groups}")

4. Compute per-hop groundedness¶

compute_feedback_by_span_group runs the metric once per group. Instead of one pooled score, you get three — one per hop — plus one pooled score across all hops.

In [ ]:

from trulens.core.feedback.selector import Selector
from trulens.core.metric.metric import Metric
from trulens.feedback.computer import compute_feedback_by_span_group


def mock_groundedness(query_text: str, retrieved_contexts: list) -> float:
    """Deterministic mock keyed on query content.

    In production, replace with an LLM-based groundedness check.
    Scores are stable regardless of call order because they're
    determined by the query, which is unique per hop.
    Check "Final refinement" before "Follow-up" because hop3's
    query contains both substrings (from the nested context).
    """
    if query_text is None:
        return 0.5  # pooled call with unresolved query
    if "Final refinement" in query_text:
        return 0.8  # hop3 — okay
    if "Follow-up" in query_text:
        return 0.4  # hop2 — poorly grounded
    return 0.9  # hop1 — well grounded


f_groundedness = Metric(
    implementation=mock_groundedness,
    name="Groundedness",
    higher_is_better=True,
).on(
    {
        "query_text": Selector(
            span_attribute=SpanAttributes.RETRIEVAL.QUERY_TEXT,
        ),
        "retrieved_contexts": Selector(
            span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS,
        ),
    }
)

compute_feedback_by_span_group(events, f_groundedness)

from trulens.core.feedback.selector import Selector from trulens.core.metric.metric import Metric from trulens.feedback.computer import compute_feedback_by_span_group def mock_groundedness(query_text: str, retrieved_contexts: list) -> float: """Deterministic mock keyed on query content. In production, replace with an LLM-based groundedness check. Scores are stable regardless of call order because they're determined by the query, which is unique per hop. Check "Final refinement" before "Follow-up" because hop3's query contains both substrings (from the nested context). """ if query_text is None: return 0.5 # pooled call with unresolved query if "Final refinement" in query_text: return 0.8 # hop3 — okay if "Follow-up" in query_text: return 0.4 # hop2 — poorly grounded return 0.9 # hop1 — well grounded f_groundedness = Metric( implementation=mock_groundedness, name="Groundedness", higher_is_better=True, ).on( { "query_text": Selector( span_attribute=SpanAttributes.RETRIEVAL.QUERY_TEXT, ), "retrieved_contexts": Selector( span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS, ), } ) compute_feedback_by_span_group(events, f_groundedness)

5. View per-hop scores¶

Each eval_root span carries eval_root.span_group (the hop label) and eval_root.score (the per-hop score).

The (pooled) score is computed without span groups — it tries to collapse all three retrieval spans into one input, but query_text can't resolve to a single value across hops, so the mock receives None. That's exactly the problem span groups solve: pooling doesn't just hide which hop failed, it can't even construct a coherent input.

Expected output:

  (pooled)      score=0.5
  hop1          score=0.9
  hop2          score=0.4
  hop3          score=0.8

In [ ]:

session.force_flush()

with db.session.begin() as db_session:
    updated_events = pd.read_sql(
        sa.select(db.orm.Event).order_by(db.orm.Event.start_timestamp),
        db_session.bind,
    )

for _, row in updated_events.iterrows():
    attrs = _parse(row["record_attributes"])
    if SpanAttributes.EVAL_ROOT.SCORE in attrs:
        hop = attrs.get(SpanAttributes.EVAL_ROOT.SPAN_GROUP, "(pooled)")
        score = attrs[SpanAttributes.EVAL_ROOT.SCORE]
        print(f"  {hop:12s}  score={score}")

session.force_flush() with db.session.begin() as db_session: updated_events = pd.read_sql( sa.select(db.orm.Event).order_by(db.orm.Event.start_timestamp), db_session.bind, ) for _, row in updated_events.iterrows(): attrs = _parse(row["record_attributes"]) if SpanAttributes.EVAL_ROOT.SCORE in attrs: hop = attrs.get(SpanAttributes.EVAL_ROOT.SPAN_GROUP, "(pooled)") score = attrs[SpanAttributes.EVAL_ROOT.SCORE] print(f" {hop:12s} score={score}")

Without span_group(), the same metric produces one pooled score that can't even resolve a coherent query_text across three different hops.

With span groups, each hop gets its own score:

• hop1: 0.9 — well grounded
• hop2: 0.4 — poorly grounded ← the problem is localized
• hop3: 0.8 — okay

This is the per-segment localization that span groups enable.