From 5adbabbad59977424407211888247649d0d77c89 Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Mon, 3 Aug 2026 11:01:28 -0700 Subject: [PATCH 1/8] feat(tools): add TraceIndex for progressive trace disclosure to judge agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Large agent trajectories overflow a judge model's context window when inlined into the evaluation prompt, forcing Experiment error-isolation to record the case as score:0 / test_pass:False — a false failure for a correct agent. TraceIndex builds an in-memory index over a Session and exposes: - overview(): one compact line per span (index, type, tool, sizes, preview) that always fits the judge context, substituted for the full trajectory. - three discovery tools the judge calls on demand: list_spans / get_span / search_spans (mirrors MLflow's ListSpans/GetSpan/SearchTraceRegex). get_span pages oversized spans via max_read_chars + offset so no single tool return can itself overflow the judge. Backend-agnostic: consumes any Session produced by a provider/mapper. --- src/strands_evals/tools/trace_index.py | 177 +++++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 src/strands_evals/tools/trace_index.py diff --git a/src/strands_evals/tools/trace_index.py b/src/strands_evals/tools/trace_index.py new file mode 100644 index 00000000..1569cf54 --- /dev/null +++ b/src/strands_evals/tools/trace_index.py @@ -0,0 +1,177 @@ +"""Progressive trace disclosure for judge agents. + +A large agent trajectory does not fit in a judge's context window. Rather than +inlining the whole Session into the evaluation prompt (which overflows and gets +scored as a failure), `TraceIndex` builds a small in-memory index over the trace +and gives the judge two things: + +1. An `overview()` — one line per span (index, type, tool name, sizes, truncated + preview) — cheap enough to always fit in context, and +2. **Lookup tools** the judge calls to load only the spans it needs to verify the + rubric: `list_spans`, `get_span`, `search_spans`. + +This is the same list / get / search shape used to query any indexed collection, +and the same progressive-disclosure pattern skills use: the overview is the +"name + description" line; the tools load the full content on demand. + +Example:: + + from strands_evals.evaluators import TrajectoryEvaluator + from strands_evals.tools.trace_index import TraceIndex + + index = TraceIndex(session) + evaluator = TrajectoryEvaluator( + rubric="Every claim in the final response must be supported by a tool result.", + tools=index.tools, + ) + # Compose the prompt with index.overview() instead of the full trajectory. +""" + +import json +import re + +from strands import tool + +from ..types.trace import ( + AgentInvocationSpan, + InferenceSpan, + Session, + SpanUnion, + ToolExecutionSpan, +) + +_PREVIEW_CHARS = 120 +_DEFAULT_MAX_READ_CHARS = 8_000 + + +def _flatten_spans(session: Session) -> list[SpanUnion]: + """Flatten all spans across traces in start_time order.""" + spans = [span for trace in session.traces for span in trace.spans] + spans.sort(key=lambda s: s.span_info.start_time) + return spans + + +def _span_text(span: SpanUnion) -> str: + """Full text content of a span, for search and retrieval.""" + if isinstance(span, ToolExecutionSpan): + return json.dumps( + { + "tool_call": span.tool_call.model_dump(), + "tool_result": span.tool_result.model_dump(), + }, + default=str, + ) + if isinstance(span, AgentInvocationSpan): + return json.dumps( + {"user_prompt": span.user_prompt, "agent_response": span.agent_response}, + default=str, + ) + if isinstance(span, InferenceSpan): + return json.dumps([m.model_dump() for m in span.messages], default=str) + return json.dumps(span.model_dump(), default=str) + + +def _preview(text: str, limit: int = _PREVIEW_CHARS) -> str: + text = re.sub(r"\s+", " ", text).strip() + return text if len(text) <= limit else text[: limit - 3] + "..." + + +def _describe(span: SpanUnion) -> str: + """One overview line describing a span without its full payload.""" + if isinstance(span, ToolExecutionSpan): + args = json.dumps(span.tool_call.arguments, default=str) + result_size = len(str(span.tool_result.content)) + return ( + f"TOOL {span.tool_call.name}({_preview(args, 80)}) " + f"-> result: {result_size} chars: {_preview(str(span.tool_result.content))}" + ) + if isinstance(span, AgentInvocationSpan): + return ( + f"AGENT prompt: {_preview(span.user_prompt, 80)} " + f"-> response: {len(span.agent_response)} chars: {_preview(span.agent_response)}" + ) + if isinstance(span, InferenceSpan): + return f"INFERENCE {len(span.messages)} messages" + return f"{type(span).__name__}" + + +class TraceIndex: + """Read-only list / get / search index over a Session for judge agents. + + Attributes: + session: The Session being evaluated. + max_read_chars: Cap on any single tool return, so a huge span can't + overflow the judge's context in one call. Oversized content is + windowed and the tool reports how to page through it. + """ + + def __init__(self, session: Session, max_read_chars: int = _DEFAULT_MAX_READ_CHARS): + self.session = session + self.max_read_chars = max_read_chars + self._spans = _flatten_spans(session) + + # Bind instance state into plain functions so @tool sees clean signatures. + # `this` (not `index`) so the public get_span(index=...) arg name is free. + this = self + + @tool + def list_spans() -> str: + """List every span in the trace: one line per span with its index, type, + tool name, argument preview, and result size. Call this first to decide + which spans to inspect.""" + return this.overview() + + @tool + def get_span(index: int, offset: int = 0) -> str: + """Get the full content of one span by its index from the span list. + Large spans are windowed; the response says how to page with offset. + + Args: + index: Span index as shown by list_spans. + offset: Character offset for paging through oversized spans. + """ + if not 0 <= index < len(this._spans): + return f"ERROR: index {index} out of range (0..{len(this._spans) - 1})" + return this._window(_span_text(this._spans[index]), offset) + + @tool + def search_spans(pattern: str, max_matches: int = 20) -> str: + """Search all span content for a regex or literal string. Returns matching + span indices with a short excerpt around each match. Use get_span to load + a matching span in full. + + Args: + pattern: Regex (or literal text) to search for. + max_matches: Maximum matches to return. + """ + try: + rx = re.compile(pattern, re.IGNORECASE) + except re.error: + rx = re.compile(re.escape(pattern), re.IGNORECASE) + hits = [] + for i, span in enumerate(this._spans): + text = _span_text(span) + m = rx.search(text) + if m: + start = max(0, m.start() - 60) + hits.append(f"[{i}] ...{_preview(text[start : m.end() + 60], 160)}...") + if len(hits) >= max_matches: + break + return "\n".join(hits) if hits else f"No matches for {pattern!r}" + + self.tools = [list_spans, get_span, search_spans] + + def overview(self) -> str: + """Compact one-line-per-span overview of the session.""" + lines = [f"Trace overview: {len(self._spans)} spans (session {self.session.session_id})"] + lines += [f"[{i}] {_describe(span)}" for i, span in enumerate(self._spans)] + return "\n".join(lines) + + def _window(self, text: str, offset: int) -> str: + if offset >= len(text): + return f"ERROR: offset {offset} beyond content length {len(text)}" + window = text[offset : offset + self.max_read_chars] + if offset + len(window) < len(text): + remaining = len(text) - offset - len(window) + window += f"\n[TRUNCATED: {remaining} chars remain; call again with offset={offset + len(window)}]" + return window From ad264d2a55b72863eb0b889d6f5d6504da8d4417 Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Mon, 3 Aug 2026 11:01:28 -0700 Subject: [PATCH 2/8] test(tools): unit, cross-pattern, and evaluator-integration tests for TraceIndex Covers overview() formatting, list_spans/get_span/search_spans behavior, offset paging on oversized spans, and end-to-end use through OutputEvaluator with tools=index.tools. --- tests/strands_evals/tools/__init__.py | 0 tests/strands_evals/tools/test_trace_index.py | 130 ++++++++++++ .../test_trace_index_evaluator_integration.py | 176 ++++++++++++++++ .../tools/test_trace_index_patterns.py | 196 ++++++++++++++++++ 4 files changed, 502 insertions(+) create mode 100644 tests/strands_evals/tools/__init__.py create mode 100644 tests/strands_evals/tools/test_trace_index.py create mode 100644 tests/strands_evals/tools/test_trace_index_evaluator_integration.py create mode 100644 tests/strands_evals/tools/test_trace_index_patterns.py diff --git a/tests/strands_evals/tools/__init__.py b/tests/strands_evals/tools/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/strands_evals/tools/test_trace_index.py b/tests/strands_evals/tools/test_trace_index.py new file mode 100644 index 00000000..8a02940c --- /dev/null +++ b/tests/strands_evals/tools/test_trace_index.py @@ -0,0 +1,130 @@ +from datetime import datetime, timezone + +import pytest + +from strands_evals.tools.trace_index import TraceIndex +from strands_evals.types.trace import ( + AgentInvocationSpan, + Session, + SpanInfo, + ToolCall, + ToolExecutionSpan, + ToolResult, + Trace, +) + + +def _span_info(second: int) -> SpanInfo: + return SpanInfo( + session_id="s1", + span_id=f"sp{second}", + start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=timezone.utc), + end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=timezone.utc), + ) + + +@pytest.fixture +def session(): + spans = [ + AgentInvocationSpan( + span_info=_span_info(0), + user_prompt="Look up ticket TKT-1042", + agent_response="Ticket TKT-1042 was refunded $150.", + available_tools=[], + ), + ToolExecutionSpan( + span_info=_span_info(1), + tool_call=ToolCall(name="lookup_ticket", arguments={"id": "TKT-1042"}), + tool_result=ToolResult(content="x" * 20_000 + " refund_amount=$150"), + ), + ToolExecutionSpan( + span_info=_span_info(2), + tool_call=ToolCall(name="get_customer", arguments={"id": "C-7"}), + tool_result=ToolResult(content="customer name: Alex"), + ), + ] + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +def test_overview_is_compact_and_ordered(session): + index = TraceIndex(session) + overview = index.overview() + + lines = overview.splitlines() + assert "3 spans" in lines[0] + assert lines[1].startswith("[0] AGENT") + assert "lookup_ticket" in lines[2] + assert "get_customer" in lines[3] + # Manifest must not inline the 20K-char tool result + assert len(overview) < 2_000 + + +def test_get_span_returns_full_content_for_small_span(session): + index = TraceIndex(session) + get_span = index.tools[1] + + content = get_span(index=2) + + assert "customer name: Alex" in content + assert "TRUNCATED" not in content + + +def test_get_span_windows_oversized_content_and_pages(session): + index = TraceIndex(session, max_read_chars=5_000) + get_span = index.tools[1] + + first = get_span(index=1) + assert "TRUNCATED" in first + assert "offset=5000" in first + + second = get_span(index=1, offset=5_000) + assert second.startswith("x") or '"' in second # continuation, not a restart + assert first[:100] != second[:100] + + +def test_get_span_index_out_of_range(session): + index = TraceIndex(session) + get_span = index.tools[1] + + assert "ERROR" in get_span(index=99) + assert "ERROR" in get_span(index=-1) + + +def test_search_spans_finds_span_by_content(session): + index = TraceIndex(session) + search_spans = index.tools[2] + + result = search_spans(pattern=r"refund_amount=\$150") + + assert result.startswith("[1]") + assert "refund_amount" in result + + +def test_search_spans_falls_back_to_literal_on_bad_regex(session): + index = TraceIndex(session) + search_spans = index.tools[2] + + result = search_spans(pattern="refund_amount=$150[") + + assert "No matches" in result or result.startswith("[") + + +def test_search_spans_no_matches(session): + index = TraceIndex(session) + search_spans = index.tools[2] + + assert "No matches" in search_spans(pattern="nonexistent-zzz") + + +def test_list_spans_tool_matches_overview(session): + index = TraceIndex(session) + list_spans = index.tools[0] + + assert list_spans() == index.overview() + + +def test_tools_are_strands_tools(session): + index = TraceIndex(session) + + for t in index.tools: + assert hasattr(t, "tool_spec") or hasattr(t, "TOOL_SPEC") or callable(t) diff --git a/tests/strands_evals/tools/test_trace_index_evaluator_integration.py b/tests/strands_evals/tools/test_trace_index_evaluator_integration.py new file mode 100644 index 00000000..5e15c128 --- /dev/null +++ b/tests/strands_evals/tools/test_trace_index_evaluator_integration.py @@ -0,0 +1,176 @@ +"""Integration: OutputEvaluator + TraceIndex — skill-style progressive discovery. + +The judge receives the compact overview in its prompt and the index's +discovery tools via the evaluators' `tools=` parameter. These tests use a +scripted fake Agent to verify the full loop deterministically: the "judge" +must call the tools to find evidence before scoring. +""" + +from datetime import datetime, timezone +from unittest.mock import Mock, patch + +from strands_evals.evaluators import OutputEvaluator +from strands_evals.tools.trace_index import TraceIndex +from strands_evals.types import EvaluationData, EvaluationOutput +from strands_evals.types.trace import ( + AgentInvocationSpan, + Session, + SpanInfo, + ToolCall, + ToolExecutionSpan, + ToolResult, + Trace, +) + + +def _span_info(second: int) -> SpanInfo: + return SpanInfo( + session_id="s1", + span_id=f"sp{second}", + start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=timezone.utc), + end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=timezone.utc), + ) + + +def _big_session(refund_amount: str = "$150") -> Session: + """A session whose tool results are too big to inline: 30 spans x ~11K chars.""" + spans = [ + AgentInvocationSpan( + span_info=_span_info(0), + user_prompt="What was the refund for TKT-1042?", + agent_response=f"Ticket TKT-1042 was refunded {refund_amount}.", + available_tools=[], + ) + ] + for i in range(1, 30): + content = ("filler row data " * 700) + (f" refund_amount={refund_amount} TKT-1042" if i == 17 else "") + spans.append( + ToolExecutionSpan( + span_info=_span_info(i), + tool_call=ToolCall(name="query_db", arguments={"page": i}), + tool_result=ToolResult(content=content), + ) + ) + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +def test_evaluator_receives_index_tools_and_overview_fits(): + session = _big_session() + index = TraceIndex(session) + + evaluator = OutputEvaluator( + rubric="Every numeric claim must be supported by a tool result in the trace.", + tools=index.tools, + ) + + assert evaluator.tools == index.tools + # The overview replaces the inline trajectory and is context-safe + inline = len(str(session.model_dump())) + assert inline > 300_000 + assert len(index.overview()) < 15_000 + + +@patch("strands_evals.evaluators.output_evaluator.Agent") +def test_judge_agent_constructed_with_discovery_tools(mock_agent_class): + session = _big_session() + index = TraceIndex(session) + mock_agent = Mock() + result = Mock() + result.structured_output = EvaluationOutput(score=1.0, test_pass=True, reason="grounded") + mock_agent.return_value = result + mock_agent_class.return_value = mock_agent + + evaluator = OutputEvaluator(rubric="Claims must be grounded.", tools=index.tools) + data = EvaluationData( + input="What was the refund for TKT-1042?", + actual_output="Ticket TKT-1042 was refunded $150.", + ) + + evaluator.evaluate(data) + + kwargs = mock_agent_class.call_args[1] + assert kwargs["tools"] == index.tools + tool_names = {getattr(t, "tool_name", getattr(t, "__name__", "")) for t in kwargs["tools"]} + assert {"list_spans", "get_span", "search_spans"} <= tool_names + + +def test_scripted_judge_finds_evidence_via_discovery(): + """Simulate the judge's tool-use loop: overview -> search -> get_span. + + This is the skill-discovery flow: the overview says *what exists*, the + tools load *what is needed*, and the judge never sees the full 300K trace. + """ + session = _big_session(refund_amount="$150") + index = TraceIndex(session) + overview, get_span, search_spans = index.tools + + judge_context_chars = 0 + + # Step 1: judge reads the overview + overview = overview() + judge_context_chars += len(overview) + assert "query_db" in overview + + # Step 2: judge searches for the claim from the agent's answer + hits = search_spans(pattern=r"refund_amount=\$150") + judge_context_chars += len(hits) + assert hits.startswith("["), "evidence must be locatable" + evidence_index = int(hits.split("]")[0][1:]) + assert evidence_index == 17 + + # Step 3: judge loads the evidence span, paging when told to + span_content = get_span(index=evidence_index) + judge_context_chars += len(span_content) + offset = 0 + while "refund_amount=$150" not in span_content and "TRUNCATED" in span_content: + offset += index.max_read_chars + span_content = get_span(index=evidence_index, offset=offset) + judge_context_chars += len(span_content) + assert "refund_amount=$150" in span_content + + # The judge verified the claim while reading a fraction of the trace + full_trace_chars = len(str(session.model_dump())) + assert judge_context_chars < full_trace_chars / 10 + + +def test_scripted_judge_detects_fabrication(): + """The agent claims $999 but the trace only supports $150 — discovery + exposes the fabrication where an overflowed inline judge would score 0 + or a truncated one might miss the evidence entirely.""" + session = _big_session(refund_amount="$150") + # Overwrite the agent's claim with a fabricated amount + agent_span = session.traces[0].spans[0] + fabricated = AgentInvocationSpan( + span_info=agent_span.span_info, + user_prompt=agent_span.user_prompt, + agent_response="Ticket TKT-1042 was refunded $999.", + available_tools=[], + ) + session.traces[0].spans[0] = fabricated + + index = TraceIndex(session) + _, _, search_spans = index.tools + + # Judge searches for the claimed amount in tool evidence: not found + claimed = search_spans(pattern=r"refund_amount=\$999") + assert "No matches" in claimed + + # But the actual amount is present: the claim contradicts the evidence + actual = search_spans(pattern=r"refund_amount=\$150") + assert actual.startswith("[") + + +def test_overview_prompt_composition_pattern(): + """The documented usage: overview into the prompt, tools onto the evaluator.""" + session = _big_session() + index = TraceIndex(session) + + prompt = ( + "Evaluate whether the agent's answer is grounded in the trace.\n" + f"\n{index.overview()}\n\n" + "Ticket TKT-1042 was refunded $150.\n" + "Use get_span/search_spans to verify before scoring." + ) + + # Stays well inside any judge's context window (~4 chars/token heuristic) + assert len(prompt) / 4 < 10_000 diff --git a/tests/strands_evals/tools/test_trace_index_patterns.py b/tests/strands_evals/tools/test_trace_index_patterns.py new file mode 100644 index 00000000..0137ad3f --- /dev/null +++ b/tests/strands_evals/tools/test_trace_index_patterns.py @@ -0,0 +1,196 @@ +"""Cross-pattern tests: TraceIndex over Sessions produced by every trace source. + +Verifies the index's overview/discovery behavior is identical whether the +Session came from: +- Strands-native OTEL spans (gen_ai semconv, StrandsInMemorySessionMapper) +- Langfuse observations (LangfuseProvider conversion) +- OpenInference spans (OpenInferenceSessionMapper, ADOT fixture) +""" + +import json +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.trace import SpanContext, SpanKind, TraceFlags + +from strands_evals.mappers import OpenInferenceSessionMapper, StrandsInMemorySessionMapper +from strands_evals.tools.trace_index import TraceIndex +from strands_evals.types.trace import Session + +_FIXTURES_DIR = Path(__file__).parent.parent / "mappers" / "fixtures" + +LARGE_RESULT = json.dumps({"rows": [{"ticket": f"TKT-{i}", "status": "resolved"} for i in range(500)]}) + + +# --- Strands-native OTEL (gen_ai semconv) --- + + +def _otel_span(provider, trace_id, span_id, parent_id, operation, attributes, events_fn): + tracer = provider.get_tracer(__name__) + with tracer.start_as_current_span(operation, kind=SpanKind.CLIENT) as s: + for k, v in attributes.items(): + s.set_attribute(k, v) + events_fn(s) + return ReadableSpan( + name=operation, + context=SpanContext(trace_id, span_id, False, TraceFlags(0x01)), + parent=SpanContext(trace_id, parent_id, False, TraceFlags(0x01)) if parent_id else None, + resource=provider.resource, + attributes=attributes, + events=tuple(s._events), + start_time=1700000000000000000, + end_time=1700000001000000000, + ) + + +@pytest.fixture +def strands_native_session() -> Session: + provider = TracerProvider() + agent_span = _otel_span( + provider, + 0xAAA, + 0xBB1, + None, + "invoke_agent", + {"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "support-agent"}, + lambda s: ( + s.add_event("gen_ai.user.message", {"content": '[{"text": "Check ticket TKT-42"}]'}), + s.add_event("gen_ai.choice", {"message": '[{"text": "TKT-42 is resolved."}]'}), + ), + ) + tool_result_message = json.dumps([{"text": LARGE_RESULT}]) + tool_span = _otel_span( + provider, + 0xAAA, + 0xBB2, + 0xBB1, + "execute_tool lookup_ticket", + {"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "lookup_ticket"}, + lambda s: ( + s.add_event("gen_ai.tool.message", {"content": '{"id": "TKT-42"}', "id": "call-1"}), + s.add_event("gen_ai.choice", {"message": tool_result_message, "id": "call-1"}), + ), + ) + return StrandsInMemorySessionMapper().map_to_session([agent_span, tool_span], "native-session") + + +# --- Langfuse observations --- + + +def _lf_obs(obs_id, trace_id, obs_type, name=None, obs_input=None, obs_output=None, parent=None, start=None): + o = MagicMock() + o.id, o.trace_id, o.type, o.name = obs_id, trace_id, obs_type, name + o.start_time = start or datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc) + o.end_time = datetime(2025, 1, 15, 10, 0, 5, tzinfo=timezone.utc) + o.input, o.output = obs_input, obs_output + o.parent_observation_id = parent + o.metadata, o.model = {}, None + o.level, o.usage, o.usage_details = "DEFAULT", None, None + return o + + +@pytest.fixture +def langfuse_session() -> Session: + import strands_evals.providers.langfuse_provider as lf_module + + with patch.object(lf_module, "Langfuse", return_value=MagicMock()): + provider = lf_module.LangfuseProvider(public_key="pk-test", secret_key="sk-test") + + observations = [ + _lf_obs( + "obs-agent", + "trace-1", + "SPAN", + name="invoke_agent support-agent", + obs_input=[{"text": "Check ticket TKT-42"}], + obs_output="TKT-42 is resolved.", + start=datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc), + ), + _lf_obs( + "obs-tool", + "trace-1", + "TOOL", + name="lookup_ticket", + obs_input={"id": "TKT-42"}, + obs_output=LARGE_RESULT, + parent="obs-agent", + start=datetime(2025, 1, 15, 10, 0, 1, tzinfo=timezone.utc), + ), + ] + spans = provider._convert_observations(observations, "lf-session") + spans = [s for s in spans if s is not None] + if not spans: + pytest.skip("Langfuse conversion produced no spans for this synthetic shape") + from strands_evals.types.trace import Trace + + return Session(traces=[Trace(spans=spans, trace_id="trace-1", session_id="lf-session")], session_id="lf-session") + + +# --- OpenInference (ADOT fixture from the repo) --- + + +@pytest.fixture +def openinference_session() -> Session: + fixture = _FIXTURES_DIR / "openinference_adot_spans.json" + if not fixture.exists(): + pytest.skip("ADOT fixture not present") + with open(fixture) as f: + spans = json.load(f) + return OpenInferenceSessionMapper().map_to_session(spans, "oi-session") + + +# --- Shared assertions across patterns --- + + +def _assert_index_works(session: Session): + index = TraceIndex(session) + list_spans, get_span, search_spans = index.tools + + overview = index.overview() + n_spans = sum(len(t.spans) for t in session.traces) + assert f"{n_spans} spans" in overview.splitlines()[0] + # Overview stays compact regardless of payload size + assert len(overview) < 400 * max(n_spans, 1) + 200 + + # Every span index is retrievable + for i in range(n_spans): + content = get_span(index=i) + assert not content.startswith("ERROR"), f"span {i} failed: {content[:80]}" + + assert isinstance(list_spans(), str) + + +def test_index_on_strands_native_session(strands_native_session): + _assert_index_works(strands_native_session) + + index = TraceIndex(strands_native_session) + _, get_span, search_spans = index.tools + + # Content-level checks: the judge can find the ticket in the tool result + hits = search_spans(pattern="TKT-42") + assert hits.startswith("[") + + +def test_index_on_langfuse_session(langfuse_session): + _assert_index_works(langfuse_session) + + index = TraceIndex(langfuse_session) + _, _, search_spans = index.tools + assert search_spans(pattern="TKT-42").startswith("[") + + +def test_index_on_openinference_session(openinference_session): + _assert_index_works(openinference_session) + + +def test_overview_compression_on_large_native_trace(strands_native_session): + """The overview must be dramatically smaller than the inline serialization.""" + index = TraceIndex(strands_native_session) + inline_size = len(str(strands_native_session.model_dump())) + overview_size = len(index.overview()) + + assert inline_size > 10_000 # LARGE_RESULT made it big + assert overview_size < inline_size / 10 From b209029af7ca497b40b0bcf4ea6405b3c5878c1c Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Mon, 3 Aug 2026 11:01:28 -0700 Subject: [PATCH 3/8] test(integ): judge-reliability A/B for inline vs TraceIndex judging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compares OutputEvaluator judging grounded vs fabricated claims two ways — full trajectory inlined vs overview + discovery tools — asserting the index-equipped judge separates grounded from fabricated where inline overflows. Skips without live Bedrock credentials. --- .../test_trace_index_judge_reliability.py | 156 ++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 tests_integ/test_trace_index_judge_reliability.py diff --git a/tests_integ/test_trace_index_judge_reliability.py b/tests_integ/test_trace_index_judge_reliability.py new file mode 100644 index 00000000..9dc2ad6d --- /dev/null +++ b/tests_integ/test_trace_index_judge_reliability.py @@ -0,0 +1,156 @@ +"""Judge-reliability A/B: inline trajectory vs TraceIndex progressive discovery. + +Compares OutputEvaluator judging the same cases two ways: + +- **inline**: the full trajectory serialized into the prompt (status quo) — + overflows or forces the judge to skim on large traces. +- **index**: compact overview in the prompt + discovery tools + (list_spans / get_span / search_spans) via `tools=`. + +Cases seed known ground truth: grounded claims (judge should pass) and +fabricated claims whose evidence is buried mid-trace (judge should fail). +A reliable judge scores grounded high and fabricated low; an unreliable one +collapses both to the same score. + +Requires live Bedrock credentials; skips otherwise. +Run: pytest tests_integ/test_trace_index_judge_reliability.py -v -s +""" + +import json +import logging +from datetime import datetime, timezone + +import pytest + +from strands_evals.evaluators import OutputEvaluator +from strands_evals.tools.trace_index import TraceIndex +from strands_evals.types import EvaluationData +from strands_evals.types.trace import ( + AgentInvocationSpan, + Session, + SpanInfo, + ToolCall, + ToolExecutionSpan, + ToolResult, + Trace, +) + +logger = logging.getLogger(__name__) + +RUBRIC = ( + "Score whether every factual claim in the output (amounts, ticket ids, dates, statuses) " + "is supported by evidence in the execution trace. 1.0 = every claim has direct tool-result " + "evidence; 0.0 = a central claim contradicts or is absent from the tool results. " + "Verify claims against the trace evidence before scoring; do not assume." +) + + +def _bedrock_available() -> bool: + try: + import boto3 + + return boto3.client("sts").get_caller_identity() is not None + except Exception: + return False + + +pytestmark = pytest.mark.skipif(not _bedrock_available(), reason="Bedrock credentials not available") + + +def _span_info(second: int) -> SpanInfo: + return SpanInfo( + session_id="s1", + span_id=f"sp{second}", + start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=timezone.utc), + end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=timezone.utc), + ) + + +def _make_session(n_tool_spans: int, evidence: str, evidence_at: int, claim: str) -> Session: + """Session with `n_tool_spans` bulky tool results; `evidence` buried at one index.""" + spans = [ + AgentInvocationSpan( + span_info=_span_info(0), + user_prompt="Summarize the resolution for ticket TKT-1042.", + agent_response=claim, + available_tools=[], + ) + ] + filler_rows = [{"ticket": f"TKT-{2000 + j}", "status": "open", "note": "unrelated backlog item"} for j in range(80)] + for i in range(1, n_tool_spans + 1): + payload = {"page": i, "rows": filler_rows} + if i == evidence_at: + payload["rows"] = [*filler_rows, {"ticket": "TKT-1042", "resolution": evidence}] + spans.append( + ToolExecutionSpan( + span_info=_span_info(i), + tool_call=ToolCall(name="query_tickets", arguments={"page": i}), + tool_result=ToolResult(content=json.dumps(payload)), + ) + ) + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +GROUNDED_CLAIM = "Ticket TKT-1042 was resolved with a $150 refund." +FABRICATED_CLAIM = "Ticket TKT-1042 was resolved with a $975 refund." +EVIDENCE = "refunded $150 to customer" + +CASES = [ + ("grounded", GROUNDED_CLAIM, True), + ("fabricated", FABRICATED_CLAIM, False), +] + +# ~30 spans x ~8K chars: large enough to stress a judge, small enough to run cheaply. +N_SPANS = 30 +EVIDENCE_AT = 17 + + +def _judge_inline(session: Session, claim: str) -> dict: + evaluator = OutputEvaluator(rubric=RUBRIC) + trace_text = str(session.model_dump()) + data = EvaluationData( + input="Summarize the resolution for ticket TKT-1042.", + actual_output=f"{claim}\n\n{trace_text}", + ) + try: + out = evaluator.evaluate(data)[0] + return {"score": out.score, "reason": out.reason, "error": None} + except Exception as e: + return {"score": None, "reason": None, "error": f"{type(e).__name__}: {e}"} + + +def _judge_explore(session: Session, claim: str) -> dict: + index = TraceIndex(session) + evaluator = OutputEvaluator(rubric=RUBRIC, tools=index.tools) + data = EvaluationData( + input="Summarize the resolution for ticket TKT-1042.", + actual_output=f"{claim}\n\n\n{index.overview()}\n", + ) + try: + out = evaluator.evaluate(data)[0] + return {"score": out.score, "reason": out.reason, "error": None} + except Exception as e: + return {"score": None, "reason": None, "error": f"{type(e).__name__}: {e}"} + + +def test_judge_reliability_inline_vs_explore(): + results = {} + for label, claim, should_pass in CASES: + session = _make_session(N_SPANS, EVIDENCE, EVIDENCE_AT, claim) + results[label] = { + "expected_pass": should_pass, + "inline": _judge_inline(session, claim), + "index": _judge_explore(session, claim), + } + + logger.info("results=<%s> | judge reliability inline vs index", json.dumps(results, indent=2, default=str)) + + # The index judge must separate grounded from fabricated. + tk_grounded = results["grounded"]["index"]["score"] + tk_fabricated = results["fabricated"]["index"]["score"] + assert tk_grounded is not None and tk_fabricated is not None, "index judge must not error" + assert tk_grounded > tk_fabricated, ( + f"index judge failed to separate grounded ({tk_grounded}) from fabricated ({tk_fabricated})" + ) + assert tk_grounded >= 0.7 + assert tk_fabricated <= 0.5 From 6017825e61435a0f8447867a368c4bccff2f859f Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Mon, 3 Aug 2026 11:01:28 -0700 Subject: [PATCH 4/8] docs: add progressive trace disclosure example to README --- README.md | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/README.md b/README.md index 583da35c..d1e93372 100644 --- a/README.md +++ b/README.md @@ -200,6 +200,31 @@ evaluator = TrajectoryEvaluator( ) ``` +### Evaluating Large Traces with Progressive Disclosure + +When a session is too large to inline into a judge prompt (large tool results, +many turns), give the judge a compact overview plus discovery tools instead of +the full trajectory. The judge loads only the spans the rubric requires: + +```python +from strands_evals.evaluators import OutputEvaluator +from strands_evals.tools.trace_index import TraceIndex + +index = TraceIndex(session) # session: a Session from any provider/mapper + +evaluator = OutputEvaluator( + rubric="Every factual claim must be supported by tool-result evidence in the trace.", + tools=index.tools, # list_spans, get_span, search_spans +) + +# Compose the prompt with the compact overview instead of the full trajectory +evaluation_output = f"{agent_answer}\n\n\n{index.overview()}\n" +``` + +The overview is one line per span (index, type, tool name, sizes, preview); +`get_span` pages through oversized spans so no single tool return can overflow +the judge's context. + ### Trace-based Helpfulness Evaluation Evaluate agent helpfulness using OpenTelemetry traces with seven-level scoring: From 44b6c1f7acef8bd601a5b5a332d842d4861d8d22 Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Thu, 27 Aug 2026 12:34:36 -0700 Subject: [PATCH 5/8] =?UTF-8?q?fix(tools):=20address=20TraceIndex=20review?= =?UTF-8?q?=20=E2=80=94=20page=20overview,=20literal=20search,=20expose=20?= =?UTF-8?q?system=5Fprompt/tools,=20tz-safe=20sort,=20add=20for=5Fjudge()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 27 +- src/strands_evals/tools/trace_index.py | 230 ++++++++++++++---- tests/strands_evals/tools/test_trace_index.py | 221 +++++++++++++++-- .../test_trace_index_evaluator_integration.py | 6 +- 4 files changed, 414 insertions(+), 70 deletions(-) diff --git a/README.md b/README.md index d1e93372..a89eeb53 100644 --- a/README.md +++ b/README.md @@ -209,21 +209,36 @@ the full trajectory. The judge loads only the spans the rubric requires: ```python from strands_evals.evaluators import OutputEvaluator from strands_evals.tools.trace_index import TraceIndex +from strands_evals.types import EvaluationData index = TraceIndex(session) # session: a Session from any provider/mapper +# for_judge() hands back both halves together so neither is forgotten: +# the overview to put next to the answer, and the discovery tools for the judge. +prompt_section, tools = index.for_judge() # tools: list_spans, get_span, search_spans + evaluator = OutputEvaluator( - rubric="Every factual claim must be supported by tool-result evidence in the trace.", - tools=index.tools, # list_spans, get_span, search_spans + rubric=( + "Every factual claim must be supported by tool-result evidence in the trace. " + "Use the trace tools to verify each claim against the evidence before scoring." + ), + tools=tools, ) -# Compose the prompt with the compact overview instead of the full trajectory -evaluation_output = f"{agent_answer}\n\n\n{index.overview()}\n" +# Compose the overview into the judged output instead of the full trajectory: +judged_output = f"{agent_answer}\n{prompt_section}" +evaluator.evaluate(EvaluationData(input=user_prompt, actual_output=judged_output)) ``` The overview is one line per span (index, type, tool name, sizes, preview); -`get_span` pages through oversized spans so no single tool return can overflow -the judge's context. +`list_spans` and `get_span` page through long traces and oversized spans so no +single tool return can overflow the judge's context. The rubric must tell the +judge to verify claims with the tools — otherwise it scores off the previews alone. + +> **Note:** this composes with `OutputEvaluator`, whose prompt is caller-controlled. +> It does **not** work with `TrajectoryEvaluator`, which inlines the full +> `actual_trajectory` unconditionally and would re-create the overflow this pattern +> exists to prevent. ### Trace-based Helpfulness Evaluation diff --git a/src/strands_evals/tools/trace_index.py b/src/strands_evals/tools/trace_index.py index 1569cf54..259a6bc6 100644 --- a/src/strands_evals/tools/trace_index.py +++ b/src/strands_evals/tools/trace_index.py @@ -6,25 +6,35 @@ and gives the judge two things: 1. An `overview()` — one line per span (index, type, tool name, sizes, truncated - preview) — cheap enough to always fit in context, and + preview) — paged so it always fits in context, and 2. **Lookup tools** the judge calls to load only the spans it needs to verify the rubric: `list_spans`, `get_span`, `search_spans`. -This is the same list / get / search shape used to query any indexed collection, -and the same progressive-disclosure pattern skills use: the overview is the -"name + description" line; the tools load the full content on demand. +This is the same list / get / search shape used to query any indexed collection: +the overview is the "name + description" line; the tools load the full content on +demand. + +`TraceIndex` composes with `OutputEvaluator`, whose prompt is caller-controlled — +put the overview in the judged output and pass the tools. It does **not** compose +with `TrajectoryEvaluator`, which inlines the full `actual_trajectory` +unconditionally and so re-creates the overflow this class exists to prevent. Example:: - from strands_evals.evaluators import TrajectoryEvaluator + from strands_evals.evaluators import OutputEvaluator from strands_evals.tools.trace_index import TraceIndex index = TraceIndex(session) - evaluator = TrajectoryEvaluator( - rubric="Every claim in the final response must be supported by a tool result.", - tools=index.tools, + prompt_section, tools = index.for_judge() # overview + tools together + evaluator = OutputEvaluator( + rubric=( + "Every claim in the final response must be supported by a tool result. " + "Use the trace tools to verify each claim before scoring." + ), + tools=tools, ) - # Compose the prompt with index.overview() instead of the full trajectory. + # Put the compact overview next to the answer instead of the full trajectory: + judged_output = f"{agent_answer}\n{prompt_section}" """ import json @@ -32,6 +42,7 @@ from strands import tool +from ..extractors.trace_extractor import _to_aware_utc from ..types.trace import ( AgentInvocationSpan, InferenceSpan, @@ -45,14 +56,19 @@ def _flatten_spans(session: Session) -> list[SpanUnion]: - """Flatten all spans across traces in start_time order.""" + """Flatten all spans across traces in start_time order. + + Timestamps are normalized to timezone-aware UTC before sorting so a mix of + naive and aware `start_time` values (produced by different mappers) can't + raise `TypeError` at construction time. + """ spans = [span for trace in session.traces for span in trace.spans] - spans.sort(key=lambda s: s.span_info.start_time) + spans.sort(key=lambda s: _to_aware_utc(s.span_info.start_time)) return spans def _span_text(span: SpanUnion) -> str: - """Full text content of a span, for search and retrieval.""" + """Full text content of a span, for retrieval via get_span.""" if isinstance(span, ToolExecutionSpan): return json.dumps( { @@ -63,7 +79,12 @@ def _span_text(span: SpanUnion) -> str: ) if isinstance(span, AgentInvocationSpan): return json.dumps( - {"user_prompt": span.user_prompt, "agent_response": span.agent_response}, + { + "user_prompt": span.user_prompt, + "agent_response": span.agent_response, + "system_prompt": span.system_prompt, + "available_tools": [t.model_dump() for t in span.available_tools], + }, default=str, ) if isinstance(span, InferenceSpan): @@ -71,6 +92,34 @@ def _span_text(span: SpanUnion) -> str: return json.dumps(span.model_dump(), default=str) +def _search_haystack(span: SpanUnion) -> str: + """Plain-text rendering of a span for search matching. + + Unlike `_span_text`, this joins the raw field values without JSON escaping so + a literal a judge copies from the overview (``$150``, ``refund_amount``) matches + the bytes it sees, and every human-visible field — including `system_prompt` + and `available_tools` — is searchable. + """ + if isinstance(span, ToolExecutionSpan): + parts = [ + span.tool_call.name, + json.dumps(span.tool_call.arguments, default=str), + str(span.tool_result.content), + ] + if span.tool_result.error: + parts.append(str(span.tool_result.error)) + return "\n".join(parts) + if isinstance(span, AgentInvocationSpan): + parts = [span.user_prompt, span.agent_response] + if span.system_prompt: + parts.append(span.system_prompt) + parts += [f"{t.name}: {t.description or ''}" for t in span.available_tools] + return "\n".join(parts) + if isinstance(span, InferenceSpan): + return "\n".join(str(m.model_dump()) for m in span.messages) + return str(span.model_dump()) + + def _preview(text: str, limit: int = _PREVIEW_CHARS) -> str: text = re.sub(r"\s+", " ", text).strip() return text if len(text) <= limit else text[: limit - 3] + "..." @@ -81,17 +130,23 @@ def _describe(span: SpanUnion) -> str: if isinstance(span, ToolExecutionSpan): args = json.dumps(span.tool_call.arguments, default=str) result_size = len(str(span.tool_result.content)) - return ( + status = "ERROR" if span.tool_result.error else "ok" + line = ( f"TOOL {span.tool_call.name}({_preview(args, 80)}) " - f"-> result: {result_size} chars: {_preview(str(span.tool_result.content))}" + f"-> [{status}] result: {result_size} chars: {_preview(str(span.tool_result.content))}" ) + if span.tool_result.error: + line += f" | error: {_preview(str(span.tool_result.error), 80)}" + return line if isinstance(span, AgentInvocationSpan): return ( f"AGENT prompt: {_preview(span.user_prompt, 80)} " f"-> response: {len(span.agent_response)} chars: {_preview(span.agent_response)}" ) if isinstance(span, InferenceSpan): - return f"INFERENCE {len(span.messages)} messages" + size = sum(len(str(m.model_dump())) for m in span.messages) + preview = _preview(" ".join(str(m.model_dump()) for m in span.messages)) + return f"INFERENCE {len(span.messages)} messages, {size} chars: {preview}" return f"{type(span).__name__}" @@ -100,26 +155,37 @@ class TraceIndex: Attributes: session: The Session being evaluated. - max_read_chars: Cap on any single tool return, so a huge span can't - overflow the judge's context in one call. Oversized content is - windowed and the tool reports how to page through it. + max_read_chars: Cap on any single tool return, so a large span or a long + overview can't overflow the judge's context in one call. Oversized + content is windowed and the tool reports how to page through it. """ def __init__(self, session: Session, max_read_chars: int = _DEFAULT_MAX_READ_CHARS): + if max_read_chars < 1: + raise ValueError(f"max_read_chars must be >= 1, got {max_read_chars}") self.session = session self.max_read_chars = max_read_chars self._spans = _flatten_spans(session) + # Precompute per-span overview lines once (recomputing per list_spans call + # is measurable on large sessions). + self._describe_lines = [f"[{i}] {_describe(span)}" for i, span in enumerate(self._spans)] # Bind instance state into plain functions so @tool sees clean signatures. # `this` (not `index`) so the public get_span(index=...) arg name is free. this = self @tool - def list_spans() -> str: - """List every span in the trace: one line per span with its index, type, - tool name, argument preview, and result size. Call this first to decide - which spans to inspect.""" - return this.overview() + def list_spans(offset: int = 0) -> str: + """List spans in the trace: one line per span with its index, type, tool + name, argument preview, and result size. Call this first to decide which + spans to inspect. Long traces are paged; the response says how to page with + offset. Previews are truncated — load a span with get_span before you rely + on its content to score. + + Args: + offset: Span index to start the listing from, for paging long traces. + """ + return this.overview(offset) @tool def get_span(index: int, offset: int = 0) -> str: @@ -135,39 +201,111 @@ def get_span(index: int, offset: int = 0) -> str: return this._window(_span_text(this._spans[index]), offset) @tool - def search_spans(pattern: str, max_matches: int = 20) -> str: - """Search all span content for a regex or literal string. Returns matching - span indices with a short excerpt around each match. Use get_span to load - a matching span in full. + def search_spans(pattern: str, max_matches: int = 20, is_regex: bool = False) -> str: + """Search all span content for a literal string (default) or a regex. + Matching is case-insensitive and covers every visible field, including + system prompts and tool configs. Returns matching span indices with a + short excerpt and per-span match count. Use get_span to load a match in full. Args: - pattern: Regex (or literal text) to search for. - max_matches: Maximum matches to return. + pattern: Text to search for. Treated literally unless is_regex=True. + max_matches: Maximum number of matching spans to return. + is_regex: Set True to treat pattern as a regular expression. """ - try: - rx = re.compile(pattern, re.IGNORECASE) - except re.error: - rx = re.compile(re.escape(pattern), re.IGNORECASE) - hits = [] + if is_regex: + try: + rx = re.compile(pattern, re.IGNORECASE) + except re.error as e: + return f"ERROR: invalid regex {pattern!r}: {e}. Retry with is_regex=False for a literal search." + matcher = lambda text: [(m.start(), m.end()) for m in rx.finditer(text)] # noqa: E731 + else: + needle = pattern.lower() + + def matcher(text: str) -> list[tuple[int, int]]: + spans, low, start = [], text.lower(), 0 + while (i := low.find(needle, start)) != -1: + spans.append((i, i + len(needle))) + start = i + max(len(needle), 1) + return spans + + hits, truncated = [], False for i, span in enumerate(this._spans): - text = _span_text(span) - m = rx.search(text) - if m: - start = max(0, m.start() - 60) - hits.append(f"[{i}] ...{_preview(text[start : m.end() + 60], 160)}...") if len(hits) >= max_matches: + truncated = True break - return "\n".join(hits) if hits else f"No matches for {pattern!r}" + text = _search_haystack(span) + positions = matcher(text) + if not positions: + continue + s, e = positions[0] + excerpt = _preview(text[max(0, s - 60) : e + 60], 160) + count = len(positions) + suffix = f" ({count} matches)" if count > 1 else "" + hits.append(f"[{i}]{suffix} ...{excerpt}...") + if not hits: + return f"No matches for {pattern!r}" + if truncated: + hits.append(f"[stopped at {max_matches} spans; refine the pattern or raise max_matches for more]") + return "\n".join(hits) self.tools = [list_spans, get_span, search_spans] - def overview(self) -> str: - """Compact one-line-per-span overview of the session.""" - lines = [f"Trace overview: {len(self._spans)} spans (session {self.session.session_id})"] - lines += [f"[{i}] {_describe(span)}" for i, span in enumerate(self._spans)] - return "\n".join(lines) + def overview(self, offset: int = 0) -> str: + """Compact one-line-per-span overview of the session, paged by span index. + + Args: + offset: Span index to start from. The listing is capped at + `max_read_chars`; if it doesn't fit, the response says the next offset. + """ + total = len(self._spans) + if offset < 0: + return f"ERROR: offset {offset} is negative; use offset >= 0" + if total and offset >= total: + return f"ERROR: offset {offset} beyond last span index {total - 1}" + + header = ( + f"Trace overview: {total} spans (session {self.session.session_id}). " + f"Previews are truncated; call get_span/search_spans to load full content " + f"(up to {self.max_read_chars} chars per call) and verify claims before scoring." + ) + lines, used, end = [], len(header), offset + for i in range(offset, total): + line = self._describe_lines[i] + if lines and used + len(line) + 1 > self.max_read_chars: + break + lines.append(line) + used += len(line) + 1 + end = i + 1 + shown = f"Showing spans {offset}-{end - 1} of {total}." if lines else f"0 spans (of {total})." + parts = [header, shown, *lines] + if end < total: + parts.append(f"[MORE: {total - end} spans remain; call again with offset={end}]") + return "\n".join(parts) + + def for_judge(self) -> tuple[str, list]: + """Return the two pieces a judge needs, together, so neither is forgotten. + + Composing a `TraceIndex` into an evaluator has two halves — the overview must + go into the judged output, and the discovery tools must be passed to the + evaluator — and doing only one silently degrades the judge (previews with no + way to drill in, or a bare answer with no trace map). This hands back both: + + prompt_section, tools = index.for_judge() + evaluator = OutputEvaluator(rubric="...", tools=tools) + output = f"{agent_answer}\n{prompt_section}" + evaluator.evaluate(EvaluationData(input=..., actual_output=output)) + + Returns: + A ``(prompt_section, tools)`` pair. ``prompt_section`` is the overview + wrapped in a ```` block ready to concatenate onto the judged + output; ``tools`` is `self.tools`. + """ + prompt_section = f"\n{self.overview()}\n" + return prompt_section, self.tools def _window(self, text: str, offset: int) -> str: + if offset < 0: + return f"ERROR: offset {offset} is negative; use offset >= 0" if offset >= len(text): return f"ERROR: offset {offset} beyond content length {len(text)}" window = text[offset : offset + self.max_read_chars] diff --git a/tests/strands_evals/tools/test_trace_index.py b/tests/strands_evals/tools/test_trace_index.py index 8a02940c..3be6479c 100644 --- a/tests/strands_evals/tools/test_trace_index.py +++ b/tests/strands_evals/tools/test_trace_index.py @@ -8,18 +8,19 @@ Session, SpanInfo, ToolCall, + ToolConfig, ToolExecutionSpan, ToolResult, Trace, ) -def _span_info(second: int) -> SpanInfo: +def _span_info(second: int, tz: timezone | None = timezone.utc) -> SpanInfo: return SpanInfo( session_id="s1", span_id=f"sp{second}", - start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=timezone.utc), - end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=timezone.utc), + start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=tz), + end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=tz), ) @@ -30,7 +31,8 @@ def session(): span_info=_span_info(0), user_prompt="Look up ticket TKT-1042", agent_response="Ticket TKT-1042 was refunded $150.", - available_tools=[], + system_prompt="You are a support agent. Never issue a refund over $500.", + available_tools=[ToolConfig(name="lookup_ticket", description="Look up a ticket by id")], ), ToolExecutionSpan( span_info=_span_info(1), @@ -52,11 +54,73 @@ def test_overview_is_compact_and_ordered(session): lines = overview.splitlines() assert "3 spans" in lines[0] - assert lines[1].startswith("[0] AGENT") - assert "lookup_ticket" in lines[2] - assert "get_customer" in lines[3] - # Manifest must not inline the 20K-char tool result + # Header + "Showing spans" banner precede the span lines. + span_lines = [ln for ln in lines if ln.startswith("[")] + assert span_lines[0].startswith("[0] AGENT") + assert "lookup_ticket" in span_lines[1] + assert "get_customer" in span_lines[2] + # Overview must not inline the 20K-char tool result. assert len(overview) < 2_000 + # Result size of the big span is surfaced exactly. + assert "20019 chars" in span_lines[1] + + +def test_overview_header_tells_the_model_previews_are_truncated(session): + index = TraceIndex(session) + header = index.overview().splitlines()[0] + assert "truncated" in header.lower() + assert "verify" in header.lower() + assert "8000" in header # max_read_chars surfaced so N-chars counts are actionable + + +def test_overview_pages_when_it_exceeds_max_read_chars(session): + # Tiny budget forces paging across the three spans. + index = TraceIndex(session, max_read_chars=200) + first = index.overview() + assert "Showing spans 0-" in first + assert "MORE:" in first + + # Follow the offset the tool reported, not a hand-computed one. + next_offset = int(first.split("offset=")[1].split("]")[0]) + assert next_offset > 0 + second = index.overview(offset=next_offset) + assert f"Showing spans {next_offset}-" in second + + # Every span shows up exactly once across the pages. + seen = set() + offset, guard = 0, 0 + while True: + page = index.overview(offset=offset) + for ln in page.splitlines(): + idx = ln[1 : ln.index("]")] if ln.startswith("[") and "]" in ln else "" + if idx.isdigit(): + seen.add(int(idx)) + if "MORE:" not in page: + break + offset = int(page.split("offset=")[1].split("]")[0]) + guard += 1 + assert guard < 10, "paging did not terminate" + assert seen == {0, 1, 2} + + +def test_overview_rejects_out_of_range_offset(session): + index = TraceIndex(session) + assert "ERROR" in index.overview(offset=99) + assert "ERROR" in index.overview(offset=-1) + + +def test_list_spans_tool_is_paged(session): + index = TraceIndex(session, max_read_chars=200) + list_spans = index.tools[0] + assert list_spans(offset=0) == index.overview(0) + assert "MORE:" in list_spans(offset=0) + + +def test_max_read_chars_must_be_positive(session): + with pytest.raises(ValueError, match="max_read_chars"): + TraceIndex(session, max_read_chars=0) + with pytest.raises(ValueError, match="max_read_chars"): + TraceIndex(session, max_read_chars=-5) def test_get_span_returns_full_content_for_small_span(session): @@ -69,6 +133,16 @@ def test_get_span_returns_full_content_for_small_span(session): assert "TRUNCATED" not in content +def test_get_span_exposes_system_prompt_and_tools(session): + index = TraceIndex(session) + get_span = index.tools[1] + + content = get_span(index=0) + + assert "Never issue a refund over $500" in content + assert "lookup_ticket" in content + + def test_get_span_windows_oversized_content_and_pages(session): index = TraceIndex(session, max_read_chars=5_000) get_span = index.tools[1] @@ -90,32 +164,139 @@ def test_get_span_index_out_of_range(session): assert "ERROR" in get_span(index=-1) -def test_search_spans_finds_span_by_content(session): +def test_get_span_rejects_negative_offset(session): + index = TraceIndex(session) + get_span = index.tools[1] + out = get_span(index=2, offset=-10) + assert "ERROR" in out + assert "-10" not in out.split("offset=")[-1] if "offset=" in out else True + + +def test_search_finds_literal_the_judge_would_quote(session): + """The flagship case: a literal '$150' the judge copies from the overview must + match without the judge knowing to escape regex metacharacters.""" index = TraceIndex(session) search_spans = index.tools[2] - result = search_spans(pattern=r"refund_amount=\$150") + result = search_spans(pattern="$150") + + assert result.startswith("[0]") or "\n[0]" in result # agent_response has "$150" + assert "[1]" in result # tool result has "refund_amount=$150" + assert "No matches" not in result + +def test_search_literal_does_not_treat_pattern_as_regex(session): + index = TraceIndex(session) + search_spans = index.tools[2] + # refund_amount=$150 as a literal matches the tool result exactly. + result = search_spans(pattern="refund_amount=$150") assert result.startswith("[1]") - assert "refund_amount" in result -def test_search_spans_falls_back_to_literal_on_bad_regex(session): +def test_search_regex_opt_in(session): + index = TraceIndex(session) + search_spans = index.tools[2] + result = search_spans(pattern=r"refund_amount=\$\d+", is_regex=True) + assert "[1]" in result + + +def test_search_invalid_regex_reports_error_not_silent_miss(session): index = TraceIndex(session) search_spans = index.tools[2] + result = search_spans(pattern="refund[", is_regex=True) + assert "ERROR" in result + assert "is_regex=False" in result # actionable hint - result = search_spans(pattern="refund_amount=$150[") - assert "No matches" in result or result.startswith("[") +def test_search_covers_system_prompt(session): + index = TraceIndex(session) + search_spans = index.tools[2] + result = search_spans(pattern="Never issue a refund") + assert result.startswith("[0]") -def test_search_spans_no_matches(session): +def test_search_does_not_false_positive_on_serialization_artifacts(session): + """'error' must not match the JSON key that every successful tool span carries.""" + index = TraceIndex(session) + search_spans = index.tools[2] + assert "No matches" in search_spans(pattern="error") + + +def test_search_reports_per_span_match_count(session): + spans = [ + ToolExecutionSpan( + span_info=_span_info(0), + tool_call=ToolCall(name="t", arguments={}), + tool_result=ToolResult(content="foo foo foo bar"), + ), + ] + sess = Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + search_spans = TraceIndex(sess).tools[2] + result = search_spans(pattern="foo") + assert "3 matches" in result + + +def test_search_signals_truncation_at_max_matches(session): + # Build many matching spans, cap below the total. + spans = [ + ToolExecutionSpan( + span_info=_span_info(i), + tool_call=ToolCall(name="t", arguments={}), + tool_result=ToolResult(content="needle here"), + ) + for i in range(5) + ] + sess = Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + search_spans = TraceIndex(sess).tools[2] + result = search_spans(pattern="needle", max_matches=2) + assert "stopped at 2" in result + assert result.count("[") >= 3 # 2 hits + the truncation marker line + + +def test_search_no_matches(session): index = TraceIndex(session) search_spans = index.tools[2] assert "No matches" in search_spans(pattern="nonexistent-zzz") +def test_overview_flags_tool_errors(session): + spans = [ + ToolExecutionSpan( + span_info=_span_info(0), + tool_call=ToolCall(name="fetch", arguments={}), + tool_result=ToolResult(content="", error="ConnectionError: timed out"), + ), + ] + sess = Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + overview = TraceIndex(sess).overview() + assert "[ERROR]" in overview + assert "ConnectionError" in overview + + +def test_flatten_tolerates_mixed_naive_and_aware_timestamps(): + """Different mappers can emit naive and aware start_time; construction must not crash.""" + spans = [ + ToolExecutionSpan( + span_info=_span_info(1, tz=None), # naive + tool_call=ToolCall(name="a", arguments={}), + tool_result=ToolResult(content="first"), + ), + ToolExecutionSpan( + span_info=_span_info(0, tz=timezone.utc), # aware, earlier + tool_call=ToolCall(name="b", arguments={}), + tool_result=ToolResult(content="second"), + ), + ] + sess = Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + index = TraceIndex(sess) # must not raise TypeError + overview = index.overview() + # Sorted by normalized time: the aware second==0 span comes first. + span_lines = [ln for ln in overview.splitlines() if ln.startswith("[")] + assert "TOOL b" in span_lines[0] + assert "TOOL a" in span_lines[1] + + def test_list_spans_tool_matches_overview(session): index = TraceIndex(session) list_spans = index.tools[0] @@ -123,6 +304,16 @@ def test_list_spans_tool_matches_overview(session): assert list_spans() == index.overview() +def test_for_judge_returns_overview_block_and_tools(session): + index = TraceIndex(session) + prompt_section, tools = index.for_judge() + + assert prompt_section.startswith("") + assert prompt_section.endswith("") + assert index.overview() in prompt_section + assert tools is index.tools + + def test_tools_are_strands_tools(session): index = TraceIndex(session) diff --git a/tests/strands_evals/tools/test_trace_index_evaluator_integration.py b/tests/strands_evals/tools/test_trace_index_evaluator_integration.py index 5e15c128..06a020c1 100644 --- a/tests/strands_evals/tools/test_trace_index_evaluator_integration.py +++ b/tests/strands_evals/tools/test_trace_index_evaluator_integration.py @@ -112,7 +112,7 @@ def test_scripted_judge_finds_evidence_via_discovery(): assert "query_db" in overview # Step 2: judge searches for the claim from the agent's answer - hits = search_spans(pattern=r"refund_amount=\$150") + hits = search_spans(pattern="refund_amount=$150") judge_context_chars += len(hits) assert hits.startswith("["), "evidence must be locatable" evidence_index = int(hits.split("]")[0][1:]) @@ -152,11 +152,11 @@ def test_scripted_judge_detects_fabrication(): _, _, search_spans = index.tools # Judge searches for the claimed amount in tool evidence: not found - claimed = search_spans(pattern=r"refund_amount=\$999") + claimed = search_spans(pattern="refund_amount=$999") assert "No matches" in claimed # But the actual amount is present: the claim contradicts the evidence - actual = search_spans(pattern=r"refund_amount=\$150") + actual = search_spans(pattern="refund_amount=$150") assert actual.startswith("[") From 05c5f2a174977239becae3e22175dfae2588eb2a Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Thu, 3 Sep 2026 12:51:22 -0700 Subject: [PATCH 6/8] fix(tools): bound search output, fix inference error search, honor regex anchors Addresses the adversarial review's three residual yellows on TraceIndex, plus the pre-existing mypy failures on this branch (CI runs ruff + mypy; only ruff was green before). - search_spans now caps its accumulated output at max_read_chars (not max_matches alone), so a generous max_matches on a long trace can no longer return 13x the per-call budget. Emits a 'budget reached' marker distinct from the max_matches marker. - Inference-span search rendered messages via model_dump(), leaking 'error': None into the haystack (false positive on every inference-span 'error' search) while a genuinely failed tool inside an inference span was unfindable. Messages are now rendered from their fields (role + text + tool name/args/result), and tool errors carry a searchable 'error:' token in both the tool and inference branches, matching the overview's [ERROR] vocabulary. - Regex search compiles with re.MULTILINE so ^/$ anchor to line boundaries in the newline-joined haystack instead of silently never matching. - mypy: annotate hits/lines empty-list locals and rename the regex exception variable so it no longer collides with the match-position unpacking. README and search_spans docstring updated to state search is bounded too. --- README.md | 7 +- src/strands_evals/tools/trace_index.py | 82 ++++++++++--- tests/strands_evals/tools/test_trace_index.py | 109 +++++++++++++++++- 3 files changed, 179 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index a89eeb53..da0e2bfe 100644 --- a/README.md +++ b/README.md @@ -231,9 +231,10 @@ evaluator.evaluate(EvaluationData(input=user_prompt, actual_output=judged_output ``` The overview is one line per span (index, type, tool name, sizes, preview); -`list_spans` and `get_span` page through long traces and oversized spans so no -single tool return can overflow the judge's context. The rubric must tell the -judge to verify claims with the tools — otherwise it scores off the previews alone. +`list_spans`, `get_span`, and `search_spans` all page or cap their output at +`max_read_chars` so no single tool return can overflow the judge's context. The +rubric must tell the judge to verify claims with the tools — otherwise it scores +off the previews alone. > **Note:** this composes with `OutputEvaluator`, whose prompt is caller-controlled. > It does **not** work with `TrajectoryEvaluator`, which inlines the full diff --git a/src/strands_evals/tools/trace_index.py b/src/strands_evals/tools/trace_index.py index 259a6bc6..a51075da 100644 --- a/src/strands_evals/tools/trace_index.py +++ b/src/strands_evals/tools/trace_index.py @@ -45,10 +45,15 @@ from ..extractors.trace_extractor import _to_aware_utc from ..types.trace import ( AgentInvocationSpan, + AssistantMessage, InferenceSpan, Session, SpanUnion, + TextContent, + ToolCallContent, ToolExecutionSpan, + ToolResultContent, + UserMessage, ) _PREVIEW_CHARS = 120 @@ -92,6 +97,28 @@ def _span_text(span: SpanUnion) -> str: return json.dumps(span.model_dump(), default=str) +def _render_message(msg: UserMessage | AssistantMessage) -> str: + """Plain-text rendering of one inference message from its fields. + + Renders role + each content block by field (text, or tool name/args/result/ + error) like the tool and agent branches, rather than dumping the pydantic dict. + A raw `model_dump()` repr leaks structural keys such as ``'error': None`` into the + text, which turns an "error" search into a false positive on every inference span + and hides genuine failures behind serialization noise. + """ + parts = [msg.role.value] + for block in msg.content: + if isinstance(block, TextContent): + parts.append(block.text) + elif isinstance(block, ToolCallContent): + parts.append(f"{block.name}({json.dumps(block.arguments, default=str)})") + elif isinstance(block, ToolResultContent): + parts.append(str(block.content)) + if block.error: + parts.append(f"error: {block.error}") + return " ".join(p for p in parts if p) + + def _search_haystack(span: SpanUnion) -> str: """Plain-text rendering of a span for search matching. @@ -107,7 +134,10 @@ def _search_haystack(span: SpanUnion) -> str: str(span.tool_result.content), ] if span.tool_result.error: - parts.append(str(span.tool_result.error)) + # Prefix with an "error:" token so a judge searching the overview's + # [ERROR] vocabulary finds the failed tool; the raw value alone + # (e.g. "CardDeclined: insufficient funds") carries no such token. + parts.append(f"error: {span.tool_result.error}") return "\n".join(parts) if isinstance(span, AgentInvocationSpan): parts = [span.user_prompt, span.agent_response] @@ -116,7 +146,7 @@ def _search_haystack(span: SpanUnion) -> str: parts += [f"{t.name}: {t.description or ''}" for t in span.available_tools] return "\n".join(parts) if isinstance(span, InferenceSpan): - return "\n".join(str(m.model_dump()) for m in span.messages) + return "\n".join(_render_message(m) for m in span.messages) return str(span.model_dump()) @@ -144,8 +174,9 @@ def _describe(span: SpanUnion) -> str: f"-> response: {len(span.agent_response)} chars: {_preview(span.agent_response)}" ) if isinstance(span, InferenceSpan): - size = sum(len(str(m.model_dump())) for m in span.messages) - preview = _preview(" ".join(str(m.model_dump()) for m in span.messages)) + rendered = [_render_message(m) for m in span.messages] + size = sum(len(r) for r in rendered) + preview = _preview(" ".join(rendered)) return f"INFERENCE {len(span.messages)} messages, {size} chars: {preview}" return f"{type(span).__name__}" @@ -204,19 +235,24 @@ def get_span(index: int, offset: int = 0) -> str: def search_spans(pattern: str, max_matches: int = 20, is_regex: bool = False) -> str: """Search all span content for a literal string (default) or a regex. Matching is case-insensitive and covers every visible field, including - system prompts and tool configs. Returns matching span indices with a - short excerpt and per-span match count. Use get_span to load a match in full. + system prompts and tool configs. Regex anchors ^/$ match line boundaries. + Returns matching span indices with a short excerpt and per-span match + count, capped at max_read_chars total; use get_span to load a match in full. Args: pattern: Text to search for. Treated literally unless is_regex=True. - max_matches: Maximum number of matching spans to return. + max_matches: Maximum number of matching spans to return. Results are + also capped at max_read_chars total, whichever comes first. is_regex: Set True to treat pattern as a regular expression. """ if is_regex: try: - rx = re.compile(pattern, re.IGNORECASE) - except re.error as e: - return f"ERROR: invalid regex {pattern!r}: {e}. Retry with is_regex=False for a literal search." + # MULTILINE so ^/$ anchor to line boundaries in the newline-joined + # haystack — LLMs write anchored regexes and would otherwise read a + # silent "No matches" as "claim unsupported". + rx = re.compile(pattern, re.IGNORECASE | re.MULTILINE) + except re.error as exc: + return f"ERROR: invalid regex {pattern!r}: {exc}. Retry with is_regex=False for a literal search." matcher = lambda text: [(m.start(), m.end()) for m in rx.finditer(text)] # noqa: E731 else: needle = pattern.lower() @@ -228,10 +264,12 @@ def matcher(text: str) -> list[tuple[int, int]]: start = i + max(len(needle), 1) return spans - hits, truncated = [], False + hits: list[str] = [] + stop_reason: str | None = None + used = 0 for i, span in enumerate(this._spans): if len(hits) >= max_matches: - truncated = True + stop_reason = "max_matches" break text = _search_haystack(span) positions = matcher(text) @@ -241,11 +279,24 @@ def matcher(text: str) -> list[tuple[int, int]]: excerpt = _preview(text[max(0, s - 60) : e + 60], 160) count = len(positions) suffix = f" ({count} matches)" if count > 1 else "" - hits.append(f"[{i}]{suffix} ...{excerpt}...") + line = f"[{i}]{suffix} ...{excerpt}..." + # Bound the whole response by max_read_chars, not max_matches alone: + # a generous max_matches on a long trace would otherwise blow past the + # per-call budget every other tool honors. Always keep at least one hit. + if hits and used + len(line) + 1 > this.max_read_chars: + stop_reason = "budget" + break + hits.append(line) + used += len(line) + 1 if not hits: return f"No matches for {pattern!r}" - if truncated: + if stop_reason == "max_matches": hits.append(f"[stopped at {max_matches} spans; refine the pattern or raise max_matches for more]") + elif stop_reason == "budget": + hits.append( + f"[budget reached at {this.max_read_chars} chars ({len(hits)} spans shown); " + f"refine the pattern to narrow results]" + ) return "\n".join(hits) self.tools = [list_spans, get_span, search_spans] @@ -268,7 +319,8 @@ def overview(self, offset: int = 0) -> str: f"Previews are truncated; call get_span/search_spans to load full content " f"(up to {self.max_read_chars} chars per call) and verify claims before scoring." ) - lines, used, end = [], len(header), offset + lines: list[str] = [] + used, end = len(header), offset for i in range(offset, total): line = self._describe_lines[i] if lines and used + len(line) + 1 > self.max_read_chars: diff --git a/tests/strands_evals/tools/test_trace_index.py b/tests/strands_evals/tools/test_trace_index.py index 3be6479c..37fe6cf1 100644 --- a/tests/strands_evals/tools/test_trace_index.py +++ b/tests/strands_evals/tools/test_trace_index.py @@ -1,17 +1,21 @@ -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import pytest from strands_evals.tools.trace_index import TraceIndex from strands_evals.types.trace import ( AgentInvocationSpan, + InferenceSpan, Session, SpanInfo, + TextContent, ToolCall, ToolConfig, ToolExecutionSpan, ToolResult, + ToolResultContent, Trace, + UserMessage, ) @@ -260,6 +264,109 @@ def test_search_no_matches(session): assert "No matches" in search_spans(pattern="nonexistent-zzz") +def _many_matching_spans(n: int, content: str) -> Session: + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + spans = [ + ToolExecutionSpan( + span_info=SpanInfo( + session_id="s1", + span_id=f"sp{i}", + start_time=base + timedelta(seconds=i), + end_time=base + timedelta(seconds=i + 1), + ), + tool_call=ToolCall(name="t", arguments={}), + tool_result=ToolResult(content=content), + ) + for i in range(n) + ] + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +def test_search_bounded_by_max_read_chars(session): + """Finding #1: search output must honor max_read_chars, not max_matches alone. + + With a generous max_matches on a long trace, the accumulated excerpts must not + blow past the per-call budget every other tool respects. + """ + sess = _many_matching_spans(800, "needle in a reasonably sized tool result payload") + search_spans = TraceIndex(sess, max_read_chars=2_000).tools[2] + + result = search_spans(pattern="needle", max_matches=800) + + assert len(result) <= 2_000 + 200 # bounded (+ slack for the closing marker line) + assert "budget reached" in result + # Far fewer than 800 spans are shown because the budget, not max_matches, stops it. + assert result.count("[") < 800 + + +def test_search_still_signals_max_matches_when_budget_not_hit(session): + """The max_matches marker (not the budget one) fires when the cap is the limit.""" + sess = _many_matching_spans(5, "needle here") + search_spans = TraceIndex(sess).tools[2] # default 8000-char budget, tiny lines + + result = search_spans(pattern="needle", max_matches=2) + + assert "stopped at 2" in result + assert "budget reached" not in result + + +def test_search_regex_anchors_match_line_boundaries(session): + """Finding #3: ^/$ must anchor to lines in the newline-joined haystack (MULTILINE).""" + index = TraceIndex(session) + search_spans = index.tools[2] + + # "lookup_ticket" is the tool name on its own line of the agent span's haystack. + assert search_spans(pattern=r"^lookup_ticket", is_regex=True).startswith("[") + # And a value ending a line is reachable with $. + assert "No matches" not in search_spans(pattern=r"\$150\.?$", is_regex=True) + + +def _inference_session(*results: ToolResult) -> Session: + """A session with one inference span carrying tool-result content blocks.""" + msg = UserMessage( + content=[TextContent(text="checking payment")] + + [ToolResultContent(content=r.content, error=r.error, tool_call_id="tc") for r in results] + ) + spans = [InferenceSpan(span_info=_span_info(0), messages=[msg])] + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +def test_search_error_no_false_positive_on_inference_span(): + """Finding #2a: a successful inference span must not match 'error'. + + Rendering messages via model_dump() leaks 'error': None into the haystack, so + every inference span with a tool result falsely matches an "error" search. + """ + sess = _inference_session(ToolResult(content="payment approved")) # error defaults None + search_spans = TraceIndex(sess).tools[2] + + assert "No matches" in search_spans(pattern="error") + + +def test_search_finds_genuine_inference_tool_error(): + """Finding #2b: a real tool failure inside an inference span must be findable.""" + sess = _inference_session(ToolResult(content="", error="CardDeclined: insufficient funds")) + search_spans = TraceIndex(sess).tools[2] + + assert search_spans(pattern="error").startswith("[") + assert search_spans(pattern="CardDeclined").startswith("[") + + +def test_search_finds_failed_tool_execution_via_error_token(): + """A failed ToolExecutionSpan is reachable by the same 'error' vocabulary as the overview.""" + spans = [ + ToolExecutionSpan( + span_info=_span_info(0), + tool_call=ToolCall(name="charge", arguments={}), + tool_result=ToolResult(content="", error="CardDeclined: insufficient funds"), + ), + ] + sess = Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + search_spans = TraceIndex(sess).tools[2] + + assert search_spans(pattern="error").startswith("[") + + def test_overview_flags_tool_errors(session): spans = [ ToolExecutionSpan( From bb5d37739b2505c60d1cc3ad39aaaf29bd7fd176 Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Wed, 9 Sep 2026 01:18:25 -0700 Subject: [PATCH 7/8] refactor(tools): extract shared progressive-disclosure helpers Pull TraceIndex's paging, windowing, and budget-bounded search into a _progressive module so a sibling index can reuse identical behavior. No observable change to TraceIndex: overview(), get_span, and search_spans now delegate to paged_listing/window/search_matches, and the local _window, _preview, and constants are removed. --- src/strands_evals/tools/_progressive.py | 160 ++++++++++++++++++++++++ src/strands_evals/tools/trace_index.py | 128 ++++--------------- 2 files changed, 187 insertions(+), 101 deletions(-) create mode 100644 src/strands_evals/tools/_progressive.py diff --git a/src/strands_evals/tools/_progressive.py b/src/strands_evals/tools/_progressive.py new file mode 100644 index 00000000..5629bd4e --- /dev/null +++ b/src/strands_evals/tools/_progressive.py @@ -0,0 +1,160 @@ +"""Shared machinery for progressive-disclosure indexes. + +`TraceIndex` (over a `Session`) and `ReferenceIndex` (over a keyed reference +corpus) expose the same list / get / search shape to a judge agent. The paging, +windowing, and search logic that keeps every tool return inside a `max_read_chars` +budget is identical between them and lives here so the two indexes can't drift. + +None of these helpers know about spans or entries — callers pass in the already +rendered lines and per-item text accessors, plus the noun to use in the paging +markers (``span``/``spans`` or ``entry``/``entries``). +""" + +import re +from typing import Callable + +PREVIEW_CHARS = 120 +DEFAULT_MAX_READ_CHARS = 8_000 + + +def preview(text: str, limit: int = PREVIEW_CHARS) -> str: + """Collapse whitespace and truncate ``text`` to ``limit`` chars with an ellipsis.""" + text = re.sub(r"\s+", " ", text).strip() + return text if len(text) <= limit else text[: limit - 3] + "..." + + +def window(text: str, offset: int, max_read_chars: int) -> str: + """Return ``text`` from ``offset``, capped at ``max_read_chars``, with a paging marker. + + Oversized content is windowed rather than returned whole so a single large + item can't overflow the judge's context; the marker says the next offset. + """ + if offset < 0: + return f"ERROR: offset {offset} is negative; use offset >= 0" + if offset >= len(text): + return f"ERROR: offset {offset} beyond content length {len(text)}" + win = text[offset : offset + max_read_chars] + if offset + len(win) < len(text): + remaining = len(text) - offset - len(win) + win += f"\n[TRUNCATED: {remaining} chars remain; call again with offset={offset + len(win)}]" + return win + + +def paged_listing( + header: str, + lines_all: list[str], + offset: int, + max_read_chars: int, + *, + unit_sg: str, + unit_pl: str, +) -> str: + """Render a one-line-per-item overview, paged by item index to fit ``max_read_chars``. + + Args: + header: A leading line describing the whole collection. + lines_all: Pre-rendered per-item lines (already prefixed with ``[id]``). + offset: Item index to start from. + max_read_chars: Cap on the returned listing. + unit_sg / unit_pl: Singular/plural noun for markers (``span``/``spans``). + """ + total = len(lines_all) + if offset < 0: + return f"ERROR: offset {offset} is negative; use offset >= 0" + if total and offset >= total: + return f"ERROR: offset {offset} beyond last {unit_sg} index {total - 1}" + + lines: list[str] = [] + used, end = len(header), offset + for i in range(offset, total): + line = lines_all[i] + if lines and used + len(line) + 1 > max_read_chars: + break + lines.append(line) + used += len(line) + 1 + end = i + 1 + shown = f"Showing {unit_pl} {offset}-{end - 1} of {total}." if lines else f"0 {unit_pl} (of {total})." + parts = [header, shown, *lines] + if end < total: + parts.append(f"[MORE: {total - end} {unit_pl} remain; call again with offset={end}]") + return "\n".join(parts) + + +def search_matches( + count: int, + haystack_of: Callable[[int], str], + ident_of: Callable[[int], str], + pattern: str, + max_matches: int, + is_regex: bool, + max_read_chars: int, + *, + unit_pl: str, +) -> str: + """Search ``count`` items, returning matching identifiers with excerpts. + + Case-insensitive; literal by default, regex when ``is_regex`` (with ``re.MULTILINE`` + so ``^``/``$`` anchor to line boundaries in the newline-joined haystacks). Output is + bounded by ``max_read_chars`` as well as ``max_matches``, whichever comes first. + + Args: + count: Number of items to search (indices ``0..count-1``). + haystack_of: Maps an item index to its searchable text. + ident_of: Maps an item index to the identifier shown in brackets (span index or key). + unit_pl: Plural noun for the stop markers. + """ + if is_regex: + try: + # MULTILINE so ^/$ anchor to line boundaries in the newline-joined haystack — + # LLMs write anchored regexes and would otherwise read a silent "No matches" + # as "claim unsupported". + rx = re.compile(pattern, re.IGNORECASE | re.MULTILINE) + except re.error as exc: + return f"ERROR: invalid regex {pattern!r}: {exc}. Retry with is_regex=False for a literal search." + + def matcher(text: str) -> list[tuple[int, int]]: + return [(m.start(), m.end()) for m in rx.finditer(text)] + else: + needle = pattern.lower() + + def matcher(text: str) -> list[tuple[int, int]]: + out, low, start = [], text.lower(), 0 + while (i := low.find(needle, start)) != -1: + out.append((i, i + len(needle))) + start = i + max(len(needle), 1) + return out + + hits: list[str] = [] + stop_reason: str | None = None + used = 0 + for i in range(count): + if len(hits) >= max_matches: + stop_reason = "max_matches" + break + text = haystack_of(i) + positions = matcher(text) + if not positions: + continue + s, e = positions[0] + excerpt = preview(text[max(0, s - 60) : e + 60], 160) + n = len(positions) + suffix = f" ({n} matches)" if n > 1 else "" + line = f"[{ident_of(i)}]{suffix} ...{excerpt}..." + # Bound the whole response by max_read_chars, not max_matches alone: a generous + # max_matches on a large collection would otherwise blow past the per-call + # budget every other tool honors. Always keep at least one hit. + if hits and used + len(line) + 1 > max_read_chars: + stop_reason = "budget" + break + hits.append(line) + used += len(line) + 1 + if not hits: + return f"No matches for {pattern!r}" + if stop_reason == "max_matches": + hits.append(f"[stopped at {max_matches} {unit_pl}; refine the pattern or raise max_matches for more]") + elif stop_reason == "budget": + hits.append( + f"[budget reached at {max_read_chars} chars ({len(hits)} {unit_pl} shown); " + f"refine the pattern to narrow results]" + ) + return "\n".join(hits) diff --git a/src/strands_evals/tools/trace_index.py b/src/strands_evals/tools/trace_index.py index a51075da..cec304fb 100644 --- a/src/strands_evals/tools/trace_index.py +++ b/src/strands_evals/tools/trace_index.py @@ -38,7 +38,6 @@ """ import json -import re from strands import tool @@ -55,9 +54,7 @@ ToolResultContent, UserMessage, ) - -_PREVIEW_CHARS = 120 -_DEFAULT_MAX_READ_CHARS = 8_000 +from ._progressive import DEFAULT_MAX_READ_CHARS, paged_listing, preview, search_matches, window def _flatten_spans(session: Session) -> list[SpanUnion]: @@ -150,11 +147,6 @@ def _search_haystack(span: SpanUnion) -> str: return str(span.model_dump()) -def _preview(text: str, limit: int = _PREVIEW_CHARS) -> str: - text = re.sub(r"\s+", " ", text).strip() - return text if len(text) <= limit else text[: limit - 3] + "..." - - def _describe(span: SpanUnion) -> str: """One overview line describing a span without its full payload.""" if isinstance(span, ToolExecutionSpan): @@ -162,22 +154,21 @@ def _describe(span: SpanUnion) -> str: result_size = len(str(span.tool_result.content)) status = "ERROR" if span.tool_result.error else "ok" line = ( - f"TOOL {span.tool_call.name}({_preview(args, 80)}) " - f"-> [{status}] result: {result_size} chars: {_preview(str(span.tool_result.content))}" + f"TOOL {span.tool_call.name}({preview(args, 80)}) " + f"-> [{status}] result: {result_size} chars: {preview(str(span.tool_result.content))}" ) if span.tool_result.error: - line += f" | error: {_preview(str(span.tool_result.error), 80)}" + line += f" | error: {preview(str(span.tool_result.error), 80)}" return line if isinstance(span, AgentInvocationSpan): return ( - f"AGENT prompt: {_preview(span.user_prompt, 80)} " - f"-> response: {len(span.agent_response)} chars: {_preview(span.agent_response)}" + f"AGENT prompt: {preview(span.user_prompt, 80)} " + f"-> response: {len(span.agent_response)} chars: {preview(span.agent_response)}" ) if isinstance(span, InferenceSpan): rendered = [_render_message(m) for m in span.messages] size = sum(len(r) for r in rendered) - preview = _preview(" ".join(rendered)) - return f"INFERENCE {len(span.messages)} messages, {size} chars: {preview}" + return f"INFERENCE {len(span.messages)} messages, {size} chars: {preview(' '.join(rendered))}" return f"{type(span).__name__}" @@ -191,7 +182,7 @@ class TraceIndex: content is windowed and the tool reports how to page through it. """ - def __init__(self, session: Session, max_read_chars: int = _DEFAULT_MAX_READ_CHARS): + def __init__(self, session: Session, max_read_chars: int = DEFAULT_MAX_READ_CHARS): if max_read_chars < 1: raise ValueError(f"max_read_chars must be >= 1, got {max_read_chars}") self.session = session @@ -229,7 +220,7 @@ def get_span(index: int, offset: int = 0) -> str: """ if not 0 <= index < len(this._spans): return f"ERROR: index {index} out of range (0..{len(this._spans) - 1})" - return this._window(_span_text(this._spans[index]), offset) + return window(_span_text(this._spans[index]), offset, this.max_read_chars) @tool def search_spans(pattern: str, max_matches: int = 20, is_regex: bool = False) -> str: @@ -245,59 +236,16 @@ def search_spans(pattern: str, max_matches: int = 20, is_regex: bool = False) -> also capped at max_read_chars total, whichever comes first. is_regex: Set True to treat pattern as a regular expression. """ - if is_regex: - try: - # MULTILINE so ^/$ anchor to line boundaries in the newline-joined - # haystack — LLMs write anchored regexes and would otherwise read a - # silent "No matches" as "claim unsupported". - rx = re.compile(pattern, re.IGNORECASE | re.MULTILINE) - except re.error as exc: - return f"ERROR: invalid regex {pattern!r}: {exc}. Retry with is_regex=False for a literal search." - matcher = lambda text: [(m.start(), m.end()) for m in rx.finditer(text)] # noqa: E731 - else: - needle = pattern.lower() - - def matcher(text: str) -> list[tuple[int, int]]: - spans, low, start = [], text.lower(), 0 - while (i := low.find(needle, start)) != -1: - spans.append((i, i + len(needle))) - start = i + max(len(needle), 1) - return spans - - hits: list[str] = [] - stop_reason: str | None = None - used = 0 - for i, span in enumerate(this._spans): - if len(hits) >= max_matches: - stop_reason = "max_matches" - break - text = _search_haystack(span) - positions = matcher(text) - if not positions: - continue - s, e = positions[0] - excerpt = _preview(text[max(0, s - 60) : e + 60], 160) - count = len(positions) - suffix = f" ({count} matches)" if count > 1 else "" - line = f"[{i}]{suffix} ...{excerpt}..." - # Bound the whole response by max_read_chars, not max_matches alone: - # a generous max_matches on a long trace would otherwise blow past the - # per-call budget every other tool honors. Always keep at least one hit. - if hits and used + len(line) + 1 > this.max_read_chars: - stop_reason = "budget" - break - hits.append(line) - used += len(line) + 1 - if not hits: - return f"No matches for {pattern!r}" - if stop_reason == "max_matches": - hits.append(f"[stopped at {max_matches} spans; refine the pattern or raise max_matches for more]") - elif stop_reason == "budget": - hits.append( - f"[budget reached at {this.max_read_chars} chars ({len(hits)} spans shown); " - f"refine the pattern to narrow results]" - ) - return "\n".join(hits) + return search_matches( + len(this._spans), + lambda i: _search_haystack(this._spans[i]), + str, + pattern, + max_matches, + is_regex, + this.max_read_chars, + unit_pl="spans", + ) self.tools = [list_spans, get_span, search_spans] @@ -309,30 +257,19 @@ def overview(self, offset: int = 0) -> str: `max_read_chars`; if it doesn't fit, the response says the next offset. """ total = len(self._spans) - if offset < 0: - return f"ERROR: offset {offset} is negative; use offset >= 0" - if total and offset >= total: - return f"ERROR: offset {offset} beyond last span index {total - 1}" - header = ( f"Trace overview: {total} spans (session {self.session.session_id}). " f"Previews are truncated; call get_span/search_spans to load full content " f"(up to {self.max_read_chars} chars per call) and verify claims before scoring." ) - lines: list[str] = [] - used, end = len(header), offset - for i in range(offset, total): - line = self._describe_lines[i] - if lines and used + len(line) + 1 > self.max_read_chars: - break - lines.append(line) - used += len(line) + 1 - end = i + 1 - shown = f"Showing spans {offset}-{end - 1} of {total}." if lines else f"0 spans (of {total})." - parts = [header, shown, *lines] - if end < total: - parts.append(f"[MORE: {total - end} spans remain; call again with offset={end}]") - return "\n".join(parts) + return paged_listing( + header, + self._describe_lines, + offset, + self.max_read_chars, + unit_sg="span", + unit_pl="spans", + ) def for_judge(self) -> tuple[str, list]: """Return the two pieces a judge needs, together, so neither is forgotten. @@ -354,14 +291,3 @@ def for_judge(self) -> tuple[str, list]: """ prompt_section = f"\n{self.overview()}\n" return prompt_section, self.tools - - def _window(self, text: str, offset: int) -> str: - if offset < 0: - return f"ERROR: offset {offset} is negative; use offset >= 0" - if offset >= len(text): - return f"ERROR: offset {offset} beyond content length {len(text)}" - window = text[offset : offset + self.max_read_chars] - if offset + len(window) < len(text): - remaining = len(text) - offset - len(window) - window += f"\n[TRUNCATED: {remaining} chars remain; call again with offset={offset + len(window)}]" - return window From 0a67c55291efa9b4ef0ee3fc0653f7bbc7345b4a Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Wed, 9 Sep 2026 01:18:26 -0700 Subject: [PATCH 8/8] feat(tools): add KnowledgeIndex for progressive disclosure over reference knowledge Some rubrics must check the trace against knowledge that lives outside it (the catalog of tools/skills the agent could have used, API specs, policies). That corpus overflows the judge for the same reason a large trace does, and a Session can't answer capability questions since it only holds what a run invoked. KnowledgeIndex applies TraceIndex's list/get/search shape to an arbitrary keyed corpus, on the shared _progressive core, with two modes: - render(keys): deterministic retrieve-then-inject as a bounded block (one LLM call) for when the relevant keys are derivable from the trace; - for_judge(): overview + discovery tools for open-ended lookup. Adds unit tests (paging, get/missing key, literal/regex/anchored/budget search, render slicing/dedupe/bounds, coercion, validation) and a Bedrock A/B integ test showing inline-vs-render-vs-explore judge reliability on an oversized catalog. --- README.md | 29 ++ src/strands_evals/tools/knowledge_index.py | 274 ++++++++++++++++++ .../tools/test_knowledge_index.py | 195 +++++++++++++ .../test_knowledge_index_judge_reliability.py | 172 +++++++++++ 4 files changed, 670 insertions(+) create mode 100644 src/strands_evals/tools/knowledge_index.py create mode 100644 tests/strands_evals/tools/test_knowledge_index.py create mode 100644 tests_integ/test_knowledge_index_judge_reliability.py diff --git a/README.md b/README.md index da0e2bfe..f50427b2 100644 --- a/README.md +++ b/README.md @@ -241,6 +241,35 @@ off the previews alone. > `actual_trajectory` unconditionally and would re-create the overflow this pattern > exists to prevent. +#### Large reference knowledge with `KnowledgeIndex` + +Some rubrics need the judge to check the trace against knowledge that lives +**outside** it — the catalog of tools/skills the agent could have used, API +specs, or policy documents. That corpus overflows the judge for the same reason a +large trace does, and the `Session` can't answer capability questions (it only +holds what a run *actually invoked*). `KnowledgeIndex` gives the same list / get / +search treatment to any keyed corpus, in two modes: + +```python +from strands_evals.tools.knowledge_index import KnowledgeIndex + +index = KnowledgeIndex(tool_catalog) # {key: document text}, e.g. tool schemas by name + +# --- Mode A: retrieve-then-inject (default; deterministic, one LLM call) --- +# The metric selects the trace-relevant keys and injects only that bounded slice. +knowledge_block = index.render(keys=plan_tool_names) # a block +judged_output = f"{agent_answer}\n{knowledge_block}" + +# --- Mode B: agentic discovery (reserve for open-ended lookup) --- +# When the needed entry can't be predetermined ("does *any* tool cover this?"). +prompt_section, tools = index.for_judge() # overview + list/get/search +``` + +Use **Mode A** whenever the relevant keys are derivable from the trace (the tools +the plan called, the domains it touched) — it's one call and deterministic. Reach +for **Mode B** only when the lookup is genuinely open-ended. Both honor the same +`max_read_chars` bound and case-insensitive literal-or-regex search as `TraceIndex`. + ### Trace-based Helpfulness Evaluation Evaluate agent helpfulness using OpenTelemetry traces with seven-level scoring: diff --git a/src/strands_evals/tools/knowledge_index.py b/src/strands_evals/tools/knowledge_index.py new file mode 100644 index 00000000..ac9c120e --- /dev/null +++ b/src/strands_evals/tools/knowledge_index.py @@ -0,0 +1,274 @@ +"""Progressive disclosure over a large keyed knowledge corpus. + +`TraceIndex` gives a judge progressive disclosure over a large *trace*. Some +rubrics also need the judge to consult knowledge that lives **outside** the +trace — a catalog of the tools/skills the agent could have used, API specs, +policy or guideline documents. That corpus overflows the judge's context for the +same reason a large trajectory does, and inlining all of it degrades judge +accuracy even when it fits (lost-in-the-middle, position bias). + +`KnowledgeIndex` applies the same list / get / search treatment to an arbitrary +collection of **keyed** documents, and adds `render()` for the common +retrieve-then-inject case where the metric already knows which entries are +relevant from the trace. + +Two usage modes: + +**Mode A — retrieve-then-inject (default, deterministic, one LLM call).** The +metric selects the trace-relevant keys and injects only those:: + + index = KnowledgeIndex(catalog) # key -> document text + knowledge_block = index.render(keys=plan_tool_names) + judged_output = f"{agent_answer}\n{knowledge_block}" + evaluator = OutputEvaluator(rubric="...") + +**Mode B — agentic discovery (reserve for open-ended lookup).** When the needed +knowledge can't be predetermined ("does *any* entry cover this request?"), hand +the judge the tools and let it look things up:: + + prompt_section, tools = index.for_judge() + evaluator = OutputEvaluator(rubric="...", tools=tools) + output = f"{agent_answer}\n{prompt_section}" + +Like `TraceIndex`, this composes with `OutputEvaluator` (caller-controlled +output) and **not** with `TrajectoryEvaluator`, which inlines the full trajectory +unconditionally. +""" + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass + +from strands import tool + +from ._progressive import DEFAULT_MAX_READ_CHARS, paged_listing, preview, search_matches, window + + +@dataclass +class KnowledgeEntry: + """One document in a knowledge corpus. + + Attributes: + key: Stable identifier the judge uses to fetch the entry (tool name, + skill domain, doc id). Must be unique within an index. + content: Full document text. + description: Optional one-line summary shown in the overview so the judge + can decide whether to load the full content. + """ + + key: str + content: str + description: str | None = None + + +def _coerce_entries( + entries: "Mapping[str, str] | Iterable[KnowledgeEntry] | Iterable[tuple[str, str]]", +) -> list[KnowledgeEntry]: + """Normalize the accepted input shapes into a list of KnowledgeEntry. + + Accepts a ``{key: content}`` mapping, an iterable of `KnowledgeEntry`, or an + iterable of ``(key, content)`` pairs — whichever is convenient for the caller + assembling the corpus. + """ + if isinstance(entries, Mapping): + return [KnowledgeEntry(key=k, content=v) for k, v in entries.items()] + out: list[KnowledgeEntry] = [] + for item in entries: + if isinstance(item, KnowledgeEntry): + out.append(item) + else: + key, content = item # (key, content) pair + out.append(KnowledgeEntry(key=key, content=content)) + return out + + +def _describe(entry: KnowledgeEntry) -> str: + """One overview line describing an entry without its full content.""" + size = len(entry.content) + if entry.description: + return f"{entry.key} — {preview(entry.description, 80)} [{size} chars]: {preview(entry.content)}" + return f"{entry.key} [{size} chars]: {preview(entry.content)}" + + +def _search_haystack(entry: KnowledgeEntry) -> str: + """Plain-text rendering of an entry for search matching (key + description + content).""" + parts = [entry.key] + if entry.description: + parts.append(entry.description) + parts.append(entry.content) + return "\n".join(parts) + + +class KnowledgeIndex: + """Read-only list / get / search index over a keyed knowledge corpus. + + Attributes: + max_read_chars: Cap on any single tool return, so a large entry or a long + overview can't overflow the judge's context in one call. Oversized + content is windowed and the tool reports how to page through it. + """ + + def __init__( + self, + entries: "Mapping[str, str] | Iterable[KnowledgeEntry] | Iterable[tuple[str, str]]", + max_read_chars: int = DEFAULT_MAX_READ_CHARS, + ): + if max_read_chars < 1: + raise ValueError(f"max_read_chars must be >= 1, got {max_read_chars}") + self.max_read_chars = max_read_chars + + coerced = _coerce_entries(entries) + self._entries: dict[str, KnowledgeEntry] = {} + for entry in coerced: + if entry.key in self._entries: + raise ValueError(f"duplicate knowledge key {entry.key!r}") + self._entries[entry.key] = entry + # Sorted key order gives the overview and search a stable, predictable + # ordering independent of insertion/dict order. + self._keys = sorted(self._entries) + self._describe_lines = [f"[{k}] {_describe(self._entries[k])}" for k in self._keys] + + # Bind instance state into plain functions so @tool sees clean signatures. + this = self + + @tool + def list_entries(offset: int = 0) -> str: + """List knowledge entries: one line per entry with its key, optional + description, size, and a truncated preview. Call this first to decide which + entries to load. Long corpora are paged; the response says how to page with + offset. Previews are truncated — load an entry with get_entry before you rely + on its content to score. + + Args: + offset: Entry index to start the listing from, for paging long corpora. + """ + return this.overview(offset) + + @tool + def get_entry(key: str, offset: int = 0) -> str: + """Get the full content of one knowledge entry by its key. + Large entries are windowed; the response says how to page with offset. + + Args: + key: Entry key as shown by list_entries. + offset: Character offset for paging through oversized entries. + """ + entry = this._entries.get(key) + if entry is None: + return f"ERROR: no entry with key {key!r}. Call list_entries to see available keys." + return window(entry.content, offset, this.max_read_chars) + + @tool + def search_entries(pattern: str, max_matches: int = 20, is_regex: bool = False) -> str: + """Search all knowledge entries for a literal string (default) or a regex. + Matching is case-insensitive and covers the key, description, and content. + Regex anchors ^/$ match line boundaries. Returns matching keys with a short + excerpt and per-entry match count, capped at max_read_chars total; use + get_entry to load a match in full. + + Args: + pattern: Text to search for. Treated literally unless is_regex=True. + max_matches: Maximum number of matching entries to return. Results are + also capped at max_read_chars total, whichever comes first. + is_regex: Set True to treat pattern as a regular expression. + """ + return search_matches( + len(this._keys), + lambda i: _search_haystack(this._entries[this._keys[i]]), + lambda i: this._keys[i], + pattern, + max_matches, + is_regex, + this.max_read_chars, + unit_pl="entries", + ) + + self.tools = [list_entries, get_entry, search_entries] + + @property + def keys(self) -> list[str]: + """The entry keys, in sorted order.""" + return list(self._keys) + + def overview(self, offset: int = 0) -> str: + """Compact one-line-per-entry overview of the corpus, paged by entry index. + + Args: + offset: Entry index to start from. The listing is capped at + `max_read_chars`; if it doesn't fit, the response says the next offset. + """ + total = len(self._keys) + header = ( + f"Knowledge overview: {total} entries. Previews are truncated; call " + f"get_entry/search_entries to load full content (up to {self.max_read_chars} " + f"chars per call) before scoring." + ) + return paged_listing( + header, + self._describe_lines, + offset, + self.max_read_chars, + unit_sg="entry", + unit_pl="entries", + ) + + def render(self, keys: Iterable[str], *, max_chars: int | None = None) -> str: + """Render selected entries as a bounded ```` block for inline injection. + + This is the deterministic, tool-free retrieve-then-inject path (Mode A): the + metric picks the trace-relevant keys and injects only those, so the judge scores + in one LLM call with no discovery round-trips. Unknown keys are reported inline + (rather than raising) so a metric deriving keys from the trace degrades to a + visible note instead of a crash. The whole block is capped at ``max_chars`` + (defaults to `max_read_chars`); if the selected entries don't fit, later entries + are truncated with a marker rather than silently dropped. + + Args: + keys: Entry keys to include, in the order given. + max_chars: Cap on the rendered block. Defaults to `max_read_chars`. + + Returns: + A ``...`` block containing the selected entries. + """ + budget = self.max_read_chars if max_chars is None else max_chars + seen: set[str] = set() + blocks: list[str] = [] + used = len("\n") + truncated = 0 + for key in keys: + if key in seen: + continue + seen.add(key) + entry = self._entries.get(key) + if entry is None: + block = f"[{key}] ERROR: no such knowledge entry" + else: + block = f"[{key}]\n{entry.content}" + # Always emit at least the first block; otherwise stop once the budget is + # spent and report how many entries were dropped. + if blocks and used + len(block) + 1 > budget: + truncated += 1 + continue + blocks.append(block) + used += len(block) + 1 + body = "\n".join(blocks) + if truncated: + body += f"\n[TRUNCATED: {truncated} more selected entries omitted at {budget} chars]" + return f"\n{body}\n" + + def for_judge(self) -> tuple[str, list]: + """Return the overview block and the discovery tools together (Mode B). + + Composing a `KnowledgeIndex` into an evaluator has two halves — the overview must + go into the judged output and the discovery tools must be passed to the evaluator + — and doing only one silently degrades the judge. This hands back both:: + + prompt_section, tools = index.for_judge() + evaluator = OutputEvaluator(rubric="...", tools=tools) + output = f"{agent_answer}\n{prompt_section}" + + Returns: + A ``(prompt_section, tools)`` pair. ``prompt_section`` is the overview wrapped + in a ```` block; ``tools`` is `self.tools`. + """ + prompt_section = f"\n{self.overview()}\n" + return prompt_section, self.tools diff --git a/tests/strands_evals/tools/test_knowledge_index.py b/tests/strands_evals/tools/test_knowledge_index.py new file mode 100644 index 00000000..f9cc168d --- /dev/null +++ b/tests/strands_evals/tools/test_knowledge_index.py @@ -0,0 +1,195 @@ +import pytest + +from strands_evals.tools.knowledge_index import KnowledgeEntry, KnowledgeIndex + + +@pytest.fixture +def catalog(): + return { + "issue_refund": "Refund a payment. args: order_id, amount. Refuses amounts over $500.", + "lookup_order": "Look up an order by id. args: order_id. Returns status and total.", + "escalate": "Escalate a case to a human agent. args: case_id, reason.", + } + + +def test_overview_is_compact_and_sorted(catalog): + index = KnowledgeIndex(catalog) + overview = index.overview() + lines = overview.splitlines() + + assert "3 entries" in lines[0] + # Keys appear in sorted order, one line each, bracketed. + entry_lines = [ln for ln in lines if ln.startswith("[")] + assert [ln.split("]")[0][1:] for ln in entry_lines] == ["escalate", "issue_refund", "lookup_order"] + # Overview is far smaller than the full corpus. + assert len(overview) < sum(len(v) for v in catalog.values()) + 500 + + +def test_description_shown_in_overview(): + index = KnowledgeIndex([KnowledgeEntry(key="k1", content="body", description="a short summary")]) + assert "a short summary" in index.overview() + + +def test_get_entry_returns_full_content(catalog): + index = KnowledgeIndex(catalog) + _, tools = index.for_judge() + get_entry = _tool(tools, "get_entry") + assert catalog["issue_refund"] in get_entry(key="issue_refund") + + +def test_get_entry_unknown_key_is_error(catalog): + index = KnowledgeIndex(catalog) + get_entry = _tool(index.tools, "get_entry") + out = get_entry(key="nope") + assert "ERROR" in out and "nope" in out + + +def test_get_entry_windows_oversized_content(): + big = "A" * 25_000 + index = KnowledgeIndex({"big": big}, max_read_chars=8_000) + get_entry = _tool(index.tools, "get_entry") + page1 = get_entry(key="big") + assert "TRUNCATED" in page1 + assert len(page1) < 8_200 + # Paging via the reported offset continues the content. + page2 = get_entry(key="big", offset=8_000) + assert page2.startswith("A") + + +def test_search_literal_finds_entry(catalog): + index = KnowledgeIndex(catalog) + search = _tool(index.tools, "search_entries") + out = search(pattern="$500") + assert "issue_refund" in out + assert "lookup_order" not in out + + +def test_search_is_case_insensitive(catalog): + index = KnowledgeIndex(catalog) + search = _tool(index.tools, "search_entries") + assert "escalate" in search(pattern="ESCALATE") + + +def test_search_regex_anchors_match_line_boundaries(): + index = KnowledgeIndex({"doc": "first line\nSTATUS: ok\nlast line"}) + search = _tool(index.tools, "search_entries") + assert "doc" in search(pattern=r"^STATUS:", is_regex=True) + + +def test_search_invalid_regex_is_reported(): + index = KnowledgeIndex({"doc": "text"}) + search = _tool(index.tools, "search_entries") + out = search(pattern="[", is_regex=True) + assert "ERROR" in out and "regex" in out + + +def test_search_no_match(catalog): + index = KnowledgeIndex(catalog) + search = _tool(index.tools, "search_entries") + assert "No matches" in search(pattern="zzz-not-present") + + +def test_search_bounded_by_max_read_chars(): + entries = {f"k{i:04d}": f"needle body {i}" for i in range(800)} + index = KnowledgeIndex(entries, max_read_chars=2_000) + search = _tool(index.tools, "search_entries") + out = search(pattern="needle", max_matches=800) + assert len(out) < 2_300 + assert "budget reached" in out + + +def test_search_signals_max_matches(): + entries = {f"k{i}": "needle" for i in range(10)} + index = KnowledgeIndex(entries, max_read_chars=100_000) + search = _tool(index.tools, "search_entries") + out = search(pattern="needle", max_matches=3) + assert "stopped at 3 entries" in out + + +def test_render_injects_selected_entries(catalog): + index = KnowledgeIndex(catalog) + block = index.render(keys=["issue_refund", "escalate"]) + assert block.startswith("") and block.endswith("") + assert catalog["issue_refund"] in block + assert catalog["escalate"] in block + # Unselected entry is not injected. + assert catalog["lookup_order"] not in block + + +def test_render_reports_unknown_key(catalog): + index = KnowledgeIndex(catalog) + block = index.render(keys=["issue_refund", "ghost"]) + assert "ghost" in block and "no such knowledge entry" in block + assert catalog["issue_refund"] in block + + +def test_render_dedupes_keys(catalog): + index = KnowledgeIndex(catalog) + block = index.render(keys=["issue_refund", "issue_refund"]) + assert block.count("[issue_refund]") == 1 + + +def test_render_bounds_by_max_chars(): + entries = {f"k{i}": "B" * 5_000 for i in range(10)} + index = KnowledgeIndex(entries, max_read_chars=8_000) + block = index.render(keys=list(entries)) + assert "TRUNCATED" in block + assert len(block) < 8_300 + + +def test_for_judge_returns_overview_and_tools(catalog): + index = KnowledgeIndex(catalog) + section, tools = index.for_judge() + assert section.startswith("") + assert "3 entries" in section + assert {t.tool_name for t in tools} == {"list_entries", "get_entry", "search_entries"} + + +def test_accepts_reference_entry_iterable(): + index = KnowledgeIndex( + [ + KnowledgeEntry(key="a", content="alpha"), + KnowledgeEntry(key="b", content="beta"), + ] + ) + assert index.keys == ["a", "b"] + + +def test_accepts_pairs(): + index = KnowledgeIndex([("a", "alpha"), ("b", "beta")]) + assert index.keys == ["a", "b"] + + +def test_duplicate_key_raises(): + with pytest.raises(ValueError, match="duplicate knowledge key"): + KnowledgeIndex([("a", "1"), ("a", "2")]) + + +def test_invalid_max_read_chars_raises(catalog): + with pytest.raises(ValueError, match="max_read_chars"): + KnowledgeIndex(catalog, max_read_chars=0) + + +def test_overview_paging(): + entries = {f"k{i:03d}": "x" * 500 for i in range(50)} + index = KnowledgeIndex(entries, max_read_chars=2_000) + page1 = index.overview() + assert "MORE" in page1 + # The MORE marker reports the next offset; paging from it makes progress. + next_offset = int(page1.split("offset=")[1].split("]")[0]) + assert next_offset > 0 + page2 = index.overview(offset=next_offset) + assert "Showing entries" in page2 + + +def test_overview_offset_out_of_range(catalog): + index = KnowledgeIndex(catalog) + assert "ERROR" in index.overview(offset=99) + + +def _tool(tools, name): + """Return the @tool-decorated callable named `name` (the tool objects are callable).""" + for t in tools: + if t.tool_name == name: + return t + raise KeyError(name) diff --git a/tests_integ/test_knowledge_index_judge_reliability.py b/tests_integ/test_knowledge_index_judge_reliability.py new file mode 100644 index 00000000..60e7074d --- /dev/null +++ b/tests_integ/test_knowledge_index_judge_reliability.py @@ -0,0 +1,172 @@ +"""Judge-reliability A/B: inline catalog vs KnowledgeIndex progressive discovery. + +A capability/feasibility rubric needs the judge to check the agent's claims +against the **full catalog of tools the agent could have used** — knowledge that +lives outside the trace. That catalog is large (dozens–hundreds of tool schemas), +so inlining all of it overflows the judge exactly like a large trajectory does. + +Compares OutputEvaluator judging the same cases three ways: + +- **inline**: the whole catalog serialized into the prompt (status quo) — + overflows or forces the judge to skim. +- **render**: Mode A — the metric selects the trace-relevant keys and injects + only that bounded ```` slice (`index.render(keys=...)`). +- **explore**: Mode B — compact overview in the prompt + discovery tools + (list_entries / get_entry / search_entries) via `tools=`. + +Cases seed known ground truth. The agent claims a capability does *not* exist and +so it could not act; the catalog is the arbiter: + +- **grounded**: the agent's "no such capability" claim is true (no matching entry). + A correct judge passes it. +- **fabricated**: the agent claims no capability exists, but a matching tool is + buried mid-catalog. A correct judge fails it. + +A reliable judge separates the two; an unreliable one (inline, overflowed) +collapses both to the same score. + +Requires live Bedrock credentials; skips otherwise. +Run: pytest tests_integ/test_knowledge_index_judge_reliability.py -v -s +""" + +import json +import logging + +import pytest + +from strands_evals.evaluators import OutputEvaluator +from strands_evals.tools.knowledge_index import KnowledgeIndex +from strands_evals.types import EvaluationData + +logger = logging.getLogger(__name__) + +RUBRIC = ( + "The agent declined a request, claiming no available tool could satisfy it. " + "Using the provided tool catalog as the source of truth, score whether that claim is correct. " + "1.0 = the catalog genuinely contains no tool that could satisfy the request; " + "0.0 = the catalog does contain a tool that could satisfy the request (the agent was wrong to decline). " + "Check the catalog before scoring; do not assume." +) + + +def _bedrock_available() -> bool: + try: + import boto3 + + return boto3.client("sts").get_caller_identity() is not None + except Exception: + return False + + +pytestmark = pytest.mark.skipif(not _bedrock_available(), reason="Bedrock credentials not available") + + +def _make_catalog(n: int, *, include_escalation: bool) -> dict[str, str]: + """A large tool catalog of bulky schemas; optionally bury an escalation tool.""" + catalog: dict[str, str] = {} + for i in range(n): + catalog[f"query_widget_{i:03d}"] = json.dumps( + { + "name": f"query_widget_{i:03d}", + "description": "Query an internal widget record by id and return its fields.", + "input_schema": {"widget_id": "string", "fields": "list[string]", "region": "string"}, + "output_schema": {"widget_id": "string", "status": "string", "payload": "object"}, + "notes": "Read-only. Paginated. Unrelated to human escalation or case handoff.", + } + ) + if include_escalation: + # The one tool that satisfies "hand this off to a human", buried mid-catalog. + catalog["transfer_to_human_agent"] = json.dumps( + { + "name": "transfer_to_human_agent", + "description": "Escalate the current case to a human support agent with full context.", + "input_schema": {"case_id": "string", "reason": "string", "priority": "string"}, + "output_schema": {"ticket_id": "string", "queued": "boolean"}, + } + ) + return catalog + + +AGENT_CLAIM = ( + "I'm sorry — I don't have any tool that can hand this case off to a human agent, " + "so I'm unable to escalate it." +) +REQUEST = "Please escalate my case to a human agent." + +N_ENTRIES = 200 + +CASES = [ + # (label, include_escalation_tool, expected_pass) + ("grounded", False, True), # no escalation tool exists -> declining is correct + ("fabricated", True, False), # escalation tool exists -> declining is wrong +] + + +def _judge_inline(catalog: dict[str, str]) -> dict: + evaluator = OutputEvaluator(rubric=RUBRIC) + catalog_text = json.dumps(catalog) + data = EvaluationData( + input=REQUEST, + actual_output=f"{AGENT_CLAIM}\n\n{catalog_text}", + ) + try: + out = evaluator.evaluate(data)[0] + return {"score": out.score, "reason": out.reason, "error": None} + except Exception as e: + return {"score": None, "reason": None, "error": f"{type(e).__name__}: {e}"} + + +def _judge_render(catalog: dict[str, str]) -> dict: + """Mode A: metric selects candidate escalation-related keys and injects only those.""" + index = KnowledgeIndex(catalog) + # A real metric would derive these from the request/trace; here we select by intent. + candidates = [k for k in index.keys if "human" in k or "transfer" in k or "escalate" in k] + reference_block = index.render(keys=candidates) + evaluator = OutputEvaluator(rubric=RUBRIC) + data = EvaluationData(input=REQUEST, actual_output=f"{AGENT_CLAIM}\n\n{reference_block}") + try: + out = evaluator.evaluate(data)[0] + return {"score": out.score, "reason": out.reason, "error": None} + except Exception as e: + return {"score": None, "reason": None, "error": f"{type(e).__name__}: {e}"} + + +def _judge_explore(catalog: dict[str, str]) -> dict: + """Mode B: overview + discovery tools; judge searches the catalog itself.""" + index = KnowledgeIndex(catalog) + prompt_section, tools = index.for_judge() + evaluator = OutputEvaluator(rubric=RUBRIC, tools=tools) + data = EvaluationData(input=REQUEST, actual_output=f"{AGENT_CLAIM}\n\n{prompt_section}") + try: + out = evaluator.evaluate(data)[0] + return {"score": out.score, "reason": out.reason, "error": None} + except Exception as e: + return {"score": None, "reason": None, "error": f"{type(e).__name__}: {e}"} + + +def test_judge_reliability_inline_vs_knowledge_index(): + results = {} + for label, include_escalation, should_pass in CASES: + catalog = _make_catalog(N_ENTRIES, include_escalation=include_escalation) + results[label] = { + "expected_pass": should_pass, + "inline": _judge_inline(catalog), + "render": _judge_render(catalog), + "explore": _judge_explore(catalog), + } + + logger.info( + "results=<%s> | judge reliability inline vs reference index", json.dumps(results, indent=2, default=str) + ) + + # Both progressive-disclosure modes must separate grounded from fabricated, + # where the fabricated case hinges on a tool buried in an oversized catalog. + for mode in ("render", "explore"): + grounded = results["grounded"][mode]["score"] + fabricated = results["fabricated"][mode]["score"] + assert grounded is not None and fabricated is not None, f"{mode} judge must not error" + assert grounded > fabricated, ( + f"{mode} judge failed to separate grounded ({grounded}) from fabricated ({fabricated})" + ) + assert grounded >= 0.7 + assert fabricated <= 0.5