diff --git a/README.md b/README.md index 6a669acf..0b29a368 100644 --- a/README.md +++ b/README.md @@ -201,6 +201,37 @@ 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), inlining the whole trajectory overflows the judge's context window +and the case is scored as a failure even when the agent was correct. The judge +evaluators handle this automatically: before each call they preflight the +rendered prompt against the judge model's context window, and when it would +overflow they hand the judge a compact overview plus discovery tools instead of +the full trajectory, so the judge loads only the spans the rubric requires. + +```python +from strands_evals.evaluators import TrajectoryEvaluator + +# disclosure="auto" (the default): inline the trajectory when it fits, fall back +# to overview + tools only when it would overflow the judge's context window. +evaluator = TrajectoryEvaluator( + rubric=( + "Every factual claim must be supported by tool-result evidence in the trace. " + "Verify each claim against the trace before scoring." + ), + disclosure="auto", +) +``` + +`disclosure` accepts `"auto"` (default), `"always"` (always use the overview + +tools), or `"never"` (always inline, restoring the prior behavior where a +genuine overflow surfaces as a judge error). On the disclosure path the judge +gets a one-line-per-span overview and three tools — `list_spans`, `get_span`, +and `search_spans` — that page or cap their output at `max_read_chars` 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: diff --git a/src/strands_evals/evaluators/_trace_index.py b/src/strands_evals/evaluators/_trace_index.py new file mode 100644 index 00000000..0d4d0f72 --- /dev/null +++ b/src/strands_evals/evaluators/_trace_index.py @@ -0,0 +1,375 @@ +"""Progressive trace disclosure for judge agents (internal engine). + +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) — 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: +the overview is the "name + description" line; the tools load the full content on +demand. + +This module is internal (note the leading underscore). Callers do not build a +`TraceIndex` themselves: the evaluator base class preflights each judge prompt and, +when the rendered trajectory would overflow the judge model's context window, +substitutes `for_judge()`'s overview for the inlined trajectory and attaches its +tools to the judge `Agent` automatically. See `Evaluator._render_with_disclosure`. +""" + +import json +import re + +from strands import tool +from strands.types.tools import AgentTool + +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 +_DEFAULT_MAX_READ_CHARS = 8_000 + + +def _flatten_spans(session: Session) -> list[SpanUnion]: + """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: _to_aware_utc(s.span_info.start_time)) + return spans + + +def _span_text(span: SpanUnion) -> str: + """Full text content of a span, for retrieval via get_span.""" + 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, + "system_prompt": span.system_prompt, + "available_tools": [t.model_dump() for t in span.available_tools], + }, + 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 _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. + + 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: + # 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] + 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(_render_message(m) for m in span.messages) + return str(span.model_dump()) + + +def _normalize_ws(text: str) -> str: + """Collapse runs of whitespace to single spaces, matching `_preview`. + + The overview preview normalizes whitespace, so a phrase a judge copies from a + preview has its interior newlines/tabs collapsed. Literal search normalizes the + haystack the same way so such a copied phrase still matches text that straddled a + line break in the source. + """ + return re.sub(r"\s+", " ", text) + + +def _preview(text: str, limit: int = _PREVIEW_CHARS) -> str: + text = _normalize_ws(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)) + 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))}" + ) + 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): + 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__}" + + +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 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(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: + """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 this._spans: + return "ERROR: trace has no 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, 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. 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. Results are + also capped at max_read_chars total, whichever comes first. + is_regex: Set True to treat pattern as a regular expression. + """ + if not pattern.strip(): + return "ERROR: empty pattern; provide text to search for" + 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: + # Normalize the needle the same way the haystack is normalized below, so a + # phrase copied from a whitespace-collapsed preview still matches. + needle = _normalize_ws(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 + # Regex matches raw text so ^/$ anchors line boundaries; literal search + # matches the whitespace-normalized text so copy-from-preview phrases hit. + text = _search_haystack(span) + if not is_regex: + text = _normalize_ws(text) + 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) + + self.tools: list[AgentTool] = [list_spans, get_span, search_spans] + + 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: 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) + + def for_judge(self) -> tuple[str, list[AgentTool]]: + """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" + # Hand back a copy so a caller mutating the returned list (or two indexes wired to + # one judge) can't mutate this index's tool set. + return prompt_section, list(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 diff --git a/src/strands_evals/evaluators/chaos/failure_communication_evaluator.py b/src/strands_evals/evaluators/chaos/failure_communication_evaluator.py index f4fce300..c3f0f621 100644 --- a/src/strands_evals/evaluators/chaos/failure_communication_evaluator.py +++ b/src/strands_evals/evaluators/chaos/failure_communication_evaluator.py @@ -7,7 +7,7 @@ from ...types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ...types.trace import EvaluationLevel -from ..evaluator import Evaluator +from ..evaluator import DisclosureMode, Evaluator from .prompt_templates.failure_communication import get_template @@ -47,12 +47,14 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.version = version default_prompt = get_template(version).SYSTEM_PROMPT self.system_prompt = system_prompt if system_prompt is not None else default_prompt self.model = model + self.disclosure = self._validate_disclosure(disclosure) def _build_output(self, rating: FailureCommunicationRating) -> list[EvaluationOutput]: normalized_score = self._score_mapping[rating.score] @@ -67,16 +69,20 @@ def _build_output(self, rating: FailureCommunicationRating) -> list[EvaluationOu def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=FailureCommunicationRating) rating = cast(FailureCommunicationRating, result.structured_output) return self._build_output(rating) async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=FailureCommunicationRating) rating = cast(FailureCommunicationRating, result.structured_output) return self._build_output(rating) diff --git a/src/strands_evals/evaluators/chaos/partial_completion_evaluator.py b/src/strands_evals/evaluators/chaos/partial_completion_evaluator.py index ac6ba9ed..8146c79b 100644 --- a/src/strands_evals/evaluators/chaos/partial_completion_evaluator.py +++ b/src/strands_evals/evaluators/chaos/partial_completion_evaluator.py @@ -6,7 +6,7 @@ from ...types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ...types.trace import EvaluationLevel -from ..evaluator import Evaluator +from ..evaluator import DisclosureMode, Evaluator from .prompt_templates.partial_completion import get_template @@ -28,12 +28,14 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.version = version default_prompt = get_template(version).SYSTEM_PROMPT self.system_prompt = system_prompt if system_prompt is not None else default_prompt self.model = model + self.disclosure = self._validate_disclosure(disclosure) def _build_output(self, rating: PartialCompletionRating) -> list[EvaluationOutput]: return [ @@ -47,16 +49,20 @@ def _build_output(self, rating: PartialCompletionRating) -> list[EvaluationOutpu def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=PartialCompletionRating) rating = cast(PartialCompletionRating, result.structured_output) return self._build_output(rating) async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=PartialCompletionRating) rating = cast(PartialCompletionRating, result.structured_output) return self._build_output(rating) diff --git a/src/strands_evals/evaluators/chaos/recovery_strategy_evaluator.py b/src/strands_evals/evaluators/chaos/recovery_strategy_evaluator.py index 21ccf888..e26963cc 100644 --- a/src/strands_evals/evaluators/chaos/recovery_strategy_evaluator.py +++ b/src/strands_evals/evaluators/chaos/recovery_strategy_evaluator.py @@ -7,7 +7,7 @@ from ...types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ...types.trace import EvaluationLevel -from ..evaluator import Evaluator +from ..evaluator import DisclosureMode, Evaluator from .prompt_templates.recovery_strategy import get_template @@ -47,12 +47,14 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.version = version default_prompt = get_template(version).SYSTEM_PROMPT self.system_prompt = system_prompt if system_prompt is not None else default_prompt self.model = model + self.disclosure = self._validate_disclosure(disclosure) def _build_output(self, rating: RecoveryStrategyRating) -> list[EvaluationOutput]: normalized_score = self._score_mapping[rating.score] @@ -67,16 +69,20 @@ def _build_output(self, rating: RecoveryStrategyRating) -> list[EvaluationOutput def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=RecoveryStrategyRating) rating = cast(RecoveryStrategyRating, result.structured_output) return self._build_output(rating) async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=RecoveryStrategyRating) rating = cast(RecoveryStrategyRating, result.structured_output) return self._build_output(rating) diff --git a/src/strands_evals/evaluators/coherence_evaluator.py b/src/strands_evals/evaluators/coherence_evaluator.py index dc2e085c..a80fe581 100644 --- a/src/strands_evals/evaluators/coherence_evaluator.py +++ b/src/strands_evals/evaluators/coherence_evaluator.py @@ -7,7 +7,8 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel, TextContent, ToolExecution, TraceLevelInput -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.coherence import get_template @@ -60,17 +61,21 @@ def __init__( system_prompt: str | None = None, include_inputs: bool = True, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt or get_template(version).SYSTEM_PROMPT self.version = version self.model = model self.include_inputs = include_inputs + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=CoherenceRating) return self._create_evaluation_output(result) @@ -86,18 +91,23 @@ def _create_evaluation_output(self, result) -> list[EvaluationOutput]: ) ] - def _format_prompt(self, parsed_input: TraceLevelInput) -> str: + def _format_prompt(self, parsed_input: TraceLevelInput, trace_index: TraceIndex | None = None) -> str: """Format evaluation prompt from parsed trace data. Args: parsed_input: Trace-level input containing agent response and session history + trace_index: When set, the previous turns are too large to inline, so the + paged trace-overview block replaces them and the judge reads spans + through the trace tools. Returns: Formatted prompt string with conversation history and target turn """ parts = [] - if parsed_input.session_history: + if trace_index is not None: + parts.append(f"# Previous turns:\n{self._disclosed_trace_section(trace_index)}") + elif parsed_input.session_history: history_lines = [] for msg in parsed_input.session_history: if isinstance(msg, list) and msg and isinstance(msg[0], ToolExecution): diff --git a/src/strands_evals/evaluators/conciseness_evaluator.py b/src/strands_evals/evaluators/conciseness_evaluator.py index 8bea4b9a..9599c5e9 100644 --- a/src/strands_evals/evaluators/conciseness_evaluator.py +++ b/src/strands_evals/evaluators/conciseness_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.conciseness import get_template @@ -44,24 +44,30 @@ def __init__( system_prompt: str | None = None, include_inputs: bool = True, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt or get_template(version).SYSTEM_PROMPT self.version = version self.model = model self.include_inputs = include_inputs + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=ConcisenessRating) return self._create_evaluation_output(result) async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=ConcisenessRating) return self._create_evaluation_output(result) diff --git a/src/strands_evals/evaluators/correctness_evaluator.py b/src/strands_evals/evaluators/correctness_evaluator.py index c7bbcacd..609f9902 100644 --- a/src/strands_evals/evaluators/correctness_evaluator.py +++ b/src/strands_evals/evaluators/correctness_evaluator.py @@ -7,7 +7,8 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel, TraceLevelInput -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.correctness import get_reference_template, get_template @@ -76,6 +77,7 @@ def __init__( system_prompt: str | None = None, reference_system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT @@ -86,6 +88,7 @@ def __init__( ) self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def _has_reference(self, evaluation_case: EvaluationData[InputT, OutputT]) -> bool: """Check if the evaluation case contains an expected_assertion for reference-based evaluation.""" @@ -97,12 +100,16 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva if self._has_reference(evaluation_case): return self._evaluate_with_reference(parsed_input, evaluation_case) - return self._evaluate_basic(parsed_input) + return self._evaluate_basic(parsed_input, evaluation_case) - def _evaluate_basic(self, parsed_input: TraceLevelInput) -> list[EvaluationOutput]: + def _evaluate_basic( + self, parsed_input: TraceLevelInput, evaluation_case: EvaluationData[InputT, OutputT] + ) -> list[EvaluationOutput]: """Evaluate correctness using the basic 3-level prompt.""" - prompt = self._format_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=CorrectnessRating) rating = cast(CorrectnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -135,12 +142,12 @@ def _evaluate_with_reference( ) ] - def _format_prompt(self, parsed_input: TraceLevelInput) -> str: + def _format_prompt(self, parsed_input: TraceLevelInput, trace_index: TraceIndex | None = None) -> str: """Format evaluation prompt for basic correctness evaluation.""" parts = [] # Format conversation context - parts.append(f"Context: {self._format_trace_level_prompt(parsed_input)}") + parts.append(f"Context: {self._format_trace_level_prompt(parsed_input, trace_index)}") # Format the candidate response (the assistant's last response) parts.append(f"Candidate Response: {parsed_input.agent_response.text}") diff --git a/src/strands_evals/evaluators/evaluator.py b/src/strands_evals/evaluators/evaluator.py index 694a9870..f33155b4 100644 --- a/src/strands_evals/evaluators/evaluator.py +++ b/src/strands_evals/evaluators/evaluator.py @@ -1,10 +1,13 @@ import asyncio import inspect import logging +from collections.abc import Callable from strands.models.model import Model -from typing_extensions import Any, Generic, TypeGuard +from typing_extensions import Any, Generic, Literal, TypeGuard, cast, get_args +from ..detectors.chunking import would_exceed_context +from ..detectors.constants import DEFAULT_MAX_INPUT_TOKENS from ..extractors import TraceExtractor from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import ( @@ -18,11 +21,30 @@ TraceLevelInput, UserMessage, ) +from ._trace_index import TraceIndex logger = logging.getLogger(__name__) DEFAULT_BEDROCK_MODEL_ID = "global.anthropic.claude-sonnet-4-6" +# The `disclosure` knob on judge evaluators. Exported as a type so the public +# kwarg is statically checkable; `DISCLOSURE_MODES` is derived from it (single +# source of truth) and used by the runtime validator. +DisclosureMode = Literal["auto", "always", "never"] +DISCLOSURE_MODES: tuple[str, ...] = get_args(DisclosureMode) + +# Judge input context windows (tokens) by model-id substring. Intentionally +# coarse: an unknown model falls back to DEFAULT_MAX_INPUT_TOKENS. A wrong guess +# only shifts when disclosure engages; it does not change the score of a fitting +# case, and under "auto" a large-enough underestimate simply inlines and lets the +# judge raise its own context-length error (see `_render_with_disclosure`). +_JUDGE_CONTEXT_WINDOWS: tuple[tuple[str, int], ...] = ( + ("nova-micro", 128_000), + ("nova-lite", 300_000), + ("nova-pro", 300_000), + ("claude", 200_000), +) + class Evaluator(Generic[InputT, OutputT]): """ @@ -36,6 +58,11 @@ class Evaluator(Generic[InputT, OutputT]): evaluation_level: EvaluationLevel | None = None _trace_extractor: TraceExtractor | None = None + # Trace-disclosure mode for judges that inline a Session trajectory. Subclasses + # that accept a `disclosure` argument set it per-instance; this class default + # keeps `self.disclosure` resolvable for evaluators that don't expose the knob. + disclosure: DisclosureMode = "auto" + def __init__(self, trace_extractor: TraceExtractor | None = None, name: str | None = None): """Initialize evaluator with optional custom trace extractor. @@ -77,6 +104,107 @@ def _get_model_id(self, model: Model | str | None) -> str: else: return "" + @staticmethod + def _validate_disclosure(disclosure: str) -> DisclosureMode: + """Validate a `disclosure` argument, returning it (narrowed) when valid.""" + if disclosure not in DISCLOSURE_MODES: + raise ValueError(f"disclosure must be one of {DISCLOSURE_MODES}, got {disclosure!r}") + return cast(DisclosureMode, disclosure) + + def _judge_window_tokens(self) -> int: + """Resolve the judge model's input context window in tokens. + + Falls back to `DEFAULT_MAX_INPUT_TOKENS` for models not in the table, so the + overflow preflight uses the window of the model actually judging rather than + assuming the default everywhere (a larger judge should disclose less often). + """ + model_id = self._get_model_id(getattr(self, "model", None)).lower() + for needle, window in _JUDGE_CONTEXT_WINDOWS: + if needle in model_id: + return window + return DEFAULT_MAX_INPUT_TOKENS + + @staticmethod + def _session_of(evaluation_case: EvaluationData[InputT, OutputT]) -> Session | None: + """The Session trajectory to index for disclosure, or None when there isn't one.""" + trajectory = evaluation_case.actual_trajectory + return trajectory if isinstance(trajectory, Session) else None + + def _render_with_disclosure( + self, + evaluation_case: EvaluationData[InputT, OutputT], + render: Callable[[TraceIndex | None], str], + ) -> tuple[str, list[Any]]: + """Render a judge prompt, falling back to trace tools when it would overflow. + + `render(None)` builds the prompt with the trajectory inlined (today's + behavior); `render(index)` builds it with a paged ```` in + place of the inlined trajectory. Returns ``(prompt, tools)``: on the inline + path `tools` is empty and the prompt is byte-identical to before, so a case + that fits the judge window is unchanged. On the disclosure path the judge + gets the overview plus `list_spans` / `get_span` / `search_spans` and reads + the trace on demand instead of overflowing on an inlined dump. + + Modes (`self.disclosure`): ``"auto"`` discloses only on a preflight overflow; + ``"always"`` discloses whenever a Session trajectory is present; ``"never"`` + always inlines, restoring the prior behavior where a real overflow surfaces + as the judge model's own context-length error. + + Only ``"auto"`` needs the size probe, so the inline prompt is built once and + reused as both the probe and the fitting-case result. Under ``"always"`` / + ``"never"`` the decision is size-independent, so the (potentially large) + inline render is skipped unless it is the one actually returned. + """ + if self.disclosure == "auto": + inline_prompt = render(None) + index = self._resolve_disclosure_index(evaluation_case, inline_prompt) + if index is None: + return inline_prompt, [] + return render(index), list(index.tools) + # "always" / "never": the probe is irrelevant, so don't serialize it. + index = self._resolve_disclosure_index(evaluation_case, "") + if index is None: + return render(None), [] + return render(index), list(index.tools) + + def _resolve_disclosure_index( + self, evaluation_case: EvaluationData[InputT, OutputT], inline_probe: str + ) -> TraceIndex | None: + """Decide whether to disclose, returning a `TraceIndex` to use or None to inline. + + `inline_probe` is the prompt (or its trajectory-bearing part) that would be + inlined; under ``"auto"`` it is preflighted against the judge model's window. + Judges whose prompt is assembled in pieces (e.g. one row per decision) call + this once per case and reuse the returned index across every piece. + + The probe covers only the rendered user prompt, not the judge's system + prompt or (on the disclosure path) the tool schemas, so it slightly + under-counts the true request size. `PREFLIGHT_SAFETY_MARGIN` (< 1.0) + absorbs that; a residual underestimate near the boundary just inlines and + lets the judge raise its own context-length error rather than mis-scoring. + """ + if self.disclosure == "never": + return None + session = self._session_of(evaluation_case) + if session is None: + return None + if self.disclosure == "auto" and not would_exceed_context(inline_probe, self._judge_window_tokens()): + return None + return TraceIndex(session) + + def _disclosed_trace_section(self, trace_index: TraceIndex) -> str: + """The trace-overview block that substitutes for an inlined trajectory. + + Wraps `TraceIndex.for_judge()`'s paged overview with an instruction telling + the judge to read spans through the tools and verify claims before scoring. + """ + overview_section, _ = trace_index.for_judge() + return ( + "The full trajectory is too large to inline. Read it through the trace tools " + "(list_spans, get_span, search_spans) and verify every claim against the spans " + f"before scoring.\n{overview_section}" + ) + @staticmethod def _default_aggregator(outputs: list[EvaluationOutput]) -> tuple[float, bool, str]: # Handle empty outputs list to avoid division by zero @@ -199,8 +327,15 @@ def _format_tools(self, tools: list[ToolConfig]) -> str: tool_lines.append(f"- {tool.name}: {desc}") return "\n".join(tool_lines) - def _format_session_history(self, contexts: list[Context]) -> str: - """Format session history with tool executions for prompt display.""" + def _format_session_history(self, contexts: list[Context], trace_index: TraceIndex | None = None) -> str: + """Format session history with tool executions for prompt display. + + When `trace_index` is provided the history is too large to inline, so the + paged trace-overview block is returned in its place and the judge reads the + spans through the trace tools instead. + """ + if trace_index is not None: + return self._disclosed_trace_section(trace_index) lines = [] for ctx in contexts: lines.append(f"User: {ctx.user_prompt.text}") @@ -211,8 +346,31 @@ def _format_session_history(self, contexts: list[Context]) -> str: lines.append(f"Assistant: {ctx.agent_response.text}") return "\n".join(lines) - def _format_tool_level_prompt(self, tool_input: ToolLevelInput) -> str: - """Format evaluation prompt from tool-level input.""" + def _tool_level_disclosure( + self, evaluation_case: EvaluationData[InputT, OutputT], tool_inputs: list[ToolLevelInput] + ) -> tuple[TraceIndex | None, list[Any]]: + """Resolve disclosure once for a tool-level case, shared across every tool call. + + Every tool call in a case is judged against the same session history, so the + overflow decision and the `TraceIndex` are made once here — using the first + tool call's rendered prompt as the ``"auto"`` size probe — and reused across + the loop, instead of rebuilding a `TraceIndex` (re-flatten + re-sort every + span) on each iteration. Returns ``(index, tools)`` to thread into + `_format_tool_level_prompt` and the judge `Agent`. + """ + if not tool_inputs: + return None, [] + probe = self._format_tool_level_prompt(tool_inputs[0]) if self.disclosure == "auto" else "" + index = self._resolve_disclosure_index(evaluation_case, probe) + return index, (list(index.tools) if index is not None else []) + + def _format_tool_level_prompt(self, tool_input: ToolLevelInput, trace_index: TraceIndex | None = None) -> str: + """Format evaluation prompt from tool-level input. + + When `trace_index` is provided the conversation history would overflow the + judge, so the paged trace-overview block replaces the inlined history; the + available-tools list and the target tool call are always kept inline. + """ parts = [] # Format available tools @@ -230,7 +388,9 @@ def _format_tool_level_prompt(self, tool_input: ToolLevelInput) -> str: ) # Format previous conversation history - if tool_input.session_history: + if trace_index is not None: + parts.append(f"## Previous conversation history\n{self._disclosed_trace_section(trace_index)}") + elif tool_input.session_history: history_lines = [] for msg in tool_input.session_history: if isinstance(msg, list): @@ -251,11 +411,18 @@ def _format_tool_level_prompt(self, tool_input: ToolLevelInput) -> str: return "\n\n".join(parts) - def _format_trace_level_prompt(self, parsed_input: TraceLevelInput) -> str: - """Format evaluation prompt from parsed turn data.""" + def _format_trace_level_prompt(self, parsed_input: TraceLevelInput, trace_index: TraceIndex | None = None) -> str: + """Format evaluation prompt from parsed turn data. + + When `trace_index` is provided the conversation history would overflow the + judge, so the paged trace-overview block replaces the inlined history; the + assistant's response being judged is always kept inline. + """ parts = [] - if parsed_input.session_history: + if trace_index is not None: + parts.append(f"# Conversation History:\n{self._disclosed_trace_section(trace_index)}") + elif parsed_input.session_history: history_lines = [] for msg in parsed_input.session_history: if isinstance(msg, list): diff --git a/src/strands_evals/evaluators/faithfulness_evaluator.py b/src/strands_evals/evaluators/faithfulness_evaluator.py index 5c5f05e9..614a4877 100644 --- a/src/strands_evals/evaluators/faithfulness_evaluator.py +++ b/src/strands_evals/evaluators/faithfulness_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.faithfulness import get_template @@ -47,16 +47,20 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=FaithfulnessRating) rating = cast(FaithfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -71,8 +75,10 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=FaithfulnessRating) rating = cast(FaithfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/goal_success_rate_evaluator.py b/src/strands_evals/evaluators/goal_success_rate_evaluator.py index 8c33061d..1c7d2ed6 100644 --- a/src/strands_evals/evaluators/goal_success_rate_evaluator.py +++ b/src/strands_evals/evaluators/goal_success_rate_evaluator.py @@ -7,7 +7,8 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel, SessionLevelInput -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.goal_success_rate import get_assertion_template, get_template @@ -73,6 +74,7 @@ def __init__( system_prompt: str | None = None, assertion_system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT @@ -83,6 +85,7 @@ def __init__( ) self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def _has_assertion(self, evaluation_case: EvaluationData[InputT, OutputT]) -> bool: """Check if the evaluation case contains expected_assertion for assertion mode.""" @@ -94,12 +97,16 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva if self._has_assertion(evaluation_case): return self._evaluate_with_assertion(session_input, evaluation_case) - return self._evaluate_basic(session_input) + return self._evaluate_basic(session_input, evaluation_case) - def _evaluate_basic(self, session_input: SessionLevelInput) -> list[EvaluationOutput]: + def _evaluate_basic( + self, session_input: SessionLevelInput, evaluation_case: EvaluationData[InputT, OutputT] + ) -> list[EvaluationOutput]: """Evaluate goal success using the basic prompt (no criteria).""" - prompt = self._format_prompt(session_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_prompt(session_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=GoalSuccessRating) rating = cast(GoalSuccessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -118,8 +125,12 @@ def _evaluate_with_assertion( evaluation_case: EvaluationData[InputT, OutputT], ) -> list[EvaluationOutput]: """Evaluate goal success using assertion-based prompt.""" - prompt = self._format_assertion_prompt(session_input, evaluation_case) - evaluator_agent = Agent(model=self.model, system_prompt=self.assertion_system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_assertion_prompt(session_input, evaluation_case, idx) + ) + evaluator_agent = Agent( + model=self.model, system_prompt=self.assertion_system_prompt, tools=tools, callback_handler=None + ) result = evaluator_agent(prompt, structured_output_model=GoalSuccessAssertionRating) rating = cast(GoalSuccessAssertionRating, result.structured_output) normalized_score = self._assertion_score_mapping[rating.verdict] @@ -132,14 +143,16 @@ def _evaluate_with_assertion( ) ] - def _format_prompt(self, session_input: SessionLevelInput) -> str: + def _format_prompt(self, session_input: SessionLevelInput, trace_index: TraceIndex | None = None) -> str: """Format evaluation prompt from session-level input.""" parts = [] if session_input.available_tools: parts.append(f"# Available tools\n{self._format_tools(session_input.available_tools)}") - if session_input.session_history: + if trace_index is not None: + parts.append(f"# Conversation record\n{self._format_session_history([], trace_index)}") + elif session_input.session_history: parts.append(f"# Conversation record\n{self._format_session_history(session_input.session_history)}") return "\n\n".join(parts) @@ -148,13 +161,16 @@ def _format_assertion_prompt( self, session_input: SessionLevelInput, evaluation_case: EvaluationData[InputT, OutputT], + trace_index: TraceIndex | None = None, ) -> str: """Format evaluation prompt for assertion-based evaluation.""" assertions = evaluation_case.expected_assertion or "" parts = [] - if session_input.session_history: + if trace_index is not None: + parts.append(f"CONVERSATION RECORD:\n{self._format_session_history([], trace_index)}") + elif session_input.session_history: parts.append(f"CONVERSATION RECORD:\n{self._format_session_history(session_input.session_history)}") parts.append(f"SUCCESS ASSERTIONS:\n{assertions}") diff --git a/src/strands_evals/evaluators/harmfulness_evaluator.py b/src/strands_evals/evaluators/harmfulness_evaluator.py index edd0b463..055e3f1c 100644 --- a/src/strands_evals/evaluators/harmfulness_evaluator.py +++ b/src/strands_evals/evaluators/harmfulness_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.harmfulness import get_template @@ -41,16 +41,20 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=HarmfulnessRating) rating = cast(HarmfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -65,8 +69,10 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=HarmfulnessRating) rating = cast(HarmfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/helpfulness_evaluator.py b/src/strands_evals/evaluators/helpfulness_evaluator.py index f5b99f53..d6686cc5 100644 --- a/src/strands_evals/evaluators/helpfulness_evaluator.py +++ b/src/strands_evals/evaluators/helpfulness_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.helpfulness import get_template @@ -52,17 +52,21 @@ def __init__( system_prompt: str | None = None, include_inputs: bool = True, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model self.include_inputs = include_inputs + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=HelpfulnessRating) rating = cast(HelpfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -77,8 +81,10 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=HelpfulnessRating) rating = cast(HelpfulnessRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/instruction_following_evaluator.py b/src/strands_evals/evaluators/instruction_following_evaluator.py index bc1f2acd..1886b892 100644 --- a/src/strands_evals/evaluators/instruction_following_evaluator.py +++ b/src/strands_evals/evaluators/instruction_following_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.instruction_following import get_template @@ -41,16 +41,20 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=InstructionFollowingRating) rating = cast(InstructionFollowingRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/prompt_templates/case_prompt_template.py b/src/strands_evals/evaluators/prompt_templates/case_prompt_template.py index 62a23bda..5d16414e 100644 --- a/src/strands_evals/evaluators/prompt_templates/case_prompt_template.py +++ b/src/strands_evals/evaluators/prompt_templates/case_prompt_template.py @@ -8,6 +8,7 @@ def compose_test_prompt( uses_trajectory: bool = False, trajectory_description: dict | None = None, uses_environment_state: bool = False, + trajectory_override: str | None = None, ) -> str: """ Compose the prompt for a test case evaluation. @@ -19,6 +20,9 @@ def compose_test_prompt( uses_trajectory: Whether this is a trajectory-based evaluation trajectory_description: A dictionary describing the type of trajectory expected for this evaluation. uses_environment_state: Whether this is an environment-state-based evaluation + trajectory_override: When set (and uses_trajectory), the string placed inside + instead of the full actual_trajectory — used to substitute a + compact trace overview when the trajectory would overflow the judge. Returns: str: The formatted evaluation prompt @@ -48,7 +52,8 @@ def compose_test_prompt( if uses_trajectory: # trajectory evaluations require actual_trajectory if evaluation_case.actual_trajectory is None: raise Exception("Please make sure the task function return a dictionary with the key 'trajectory'.") - evaluation_prompt += f"{evaluation_case.actual_trajectory}\n" + trajectory_body = trajectory_override if trajectory_override is not None else evaluation_case.actual_trajectory + evaluation_prompt += f"{trajectory_body}\n" if evaluation_case.expected_trajectory: evaluation_prompt += f"{evaluation_case.expected_trajectory}\n" diff --git a/src/strands_evals/evaluators/refusal_evaluator.py b/src/strands_evals/evaluators/refusal_evaluator.py index 6e00285d..f2d8392f 100644 --- a/src/strands_evals/evaluators/refusal_evaluator.py +++ b/src/strands_evals/evaluators/refusal_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.refusal import get_template @@ -41,16 +41,20 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=RefusalRating) rating = cast(RefusalRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/response_relevance_evaluator.py b/src/strands_evals/evaluators/response_relevance_evaluator.py index c9277223..164ab2ef 100644 --- a/src/strands_evals/evaluators/response_relevance_evaluator.py +++ b/src/strands_evals/evaluators/response_relevance_evaluator.py @@ -8,7 +8,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.response_relevance import get_template @@ -49,24 +49,30 @@ def __init__( system_prompt: str | None = None, include_inputs: bool = True, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt or get_template(version).SYSTEM_PROMPT self.version = version self.model = model self.include_inputs = include_inputs + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=ResponseRelevanceRating) return self._create_evaluation_output(result) async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = await evaluator_agent.invoke_async(prompt, structured_output_model=ResponseRelevanceRating) return self._create_evaluation_output(result) diff --git a/src/strands_evals/evaluators/skill_instruction_following_evaluator.py b/src/strands_evals/evaluators/skill_instruction_following_evaluator.py index 89299120..8769fc15 100644 --- a/src/strands_evals/evaluators/skill_instruction_following_evaluator.py +++ b/src/strands_evals/evaluators/skill_instruction_following_evaluator.py @@ -7,7 +7,8 @@ from ..extractors.skills import InvokedSkill, extract_selected_skills from ..types.evaluation import NOT_APPLICABLE, EvaluationData, EvaluationOutput, InputT, OutputT -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.skill_instruction_following import get_template from .prompt_templates.trajectory_prompt_template import serialize_trajectory @@ -134,11 +135,13 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) # Drop not-applicable rows from the aggregate so no-skill runs don't deflate the mean. self.aggregator = self._aggregate_dropping_na @@ -166,15 +169,39 @@ def _unscorable_reason(skill: InvokedSkill) -> str | None: return f"{skill.name}: skill body unavailable" return None - def _build_prompt(self, skill: InvokedSkill, evaluation_case: EvaluationData[InputT, OutputT]) -> str: + def _build_prompt( + self, + skill: InvokedSkill, + evaluation_case: EvaluationData[InputT, OutputT], + trace_index: TraceIndex | None = None, + ) -> str: body = _strip_harness_metadata(_strip_frontmatter(skill.body or "")) + trajectory = ( + self._disclosed_trace_section(trace_index) + if trace_index is not None + else serialize_trajectory(evaluation_case.actual_trajectory) + ) return ( f"## Skill: {skill.name}\n\n" f"## SKILL.md instructions\n{body}\n\n" - f"## Agent trajectory\n{serialize_trajectory(evaluation_case.actual_trajectory)}\n\n" + f"## Agent trajectory\n{trajectory}\n\n" f"## Agent's final response\n{evaluation_case.actual_output}" ) + def _resolve_case_disclosure( + self, evaluation_case: EvaluationData[InputT, OutputT], probe_skill: InvokedSkill + ) -> tuple[TraceIndex | None, list]: + """Decide disclosure once per case; the trajectory is shared across every invoked skill. + + Under ``"auto"`` the probe is the full inline prompt for one representative + skill (so the skill body and final response are counted, not just the + trajectory); under ``"always"`` / ``"never"`` the size is irrelevant and the + probe is skipped. + """ + probe = self._build_prompt(probe_skill, evaluation_case) if self.disclosure == "auto" else "" + index = self._resolve_disclosure_index(evaluation_case, probe) + return index, (list(index.tools) if index is not None else []) + def _rating_to_output(self, skill: InvokedSkill, rating: SkillFollowingRating) -> EvaluationOutput: # A skill that prescribes nothing has nothing to follow, so scoring it either way would be # arbitrary: it is the same vacuous case as "no skill invoked", not a failure to adhere. @@ -201,13 +228,16 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva invoked = extract_selected_skills(evaluation_case.actual_trajectory) if not invoked: return [self._not_applicable_row("no skill invoked")] + index, tools = self._resolve_case_disclosure(evaluation_case, invoked[0]) results = [] for skill in invoked: if reason := self._unscorable_reason(skill): results.append(self._not_applicable_row(reason)) continue - prompt = self._build_prompt(skill, evaluation_case) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._build_prompt(skill, evaluation_case, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = evaluator_agent(prompt, structured_output_model=SkillFollowingRating) rating = cast(SkillFollowingRating, result.structured_output) results.append(self._rating_to_output(skill, rating)) @@ -219,13 +249,16 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) invoked = extract_selected_skills(evaluation_case.actual_trajectory) if not invoked: return [self._not_applicable_row("no skill invoked")] + index, tools = self._resolve_case_disclosure(evaluation_case, invoked[0]) results = [] for skill in invoked: if reason := self._unscorable_reason(skill): results.append(self._not_applicable_row(reason)) continue - prompt = self._build_prompt(skill, evaluation_case) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._build_prompt(skill, evaluation_case, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = await evaluator_agent.invoke_async(prompt, structured_output_model=SkillFollowingRating) rating = cast(SkillFollowingRating, result.structured_output) results.append(self._rating_to_output(skill, rating)) diff --git a/src/strands_evals/evaluators/skill_selection_accuracy_evaluator.py b/src/strands_evals/evaluators/skill_selection_accuracy_evaluator.py index 9835a564..a8776493 100644 --- a/src/strands_evals/evaluators/skill_selection_accuracy_evaluator.py +++ b/src/strands_evals/evaluators/skill_selection_accuracy_evaluator.py @@ -12,7 +12,8 @@ parse_available_skills, ) from ..types.evaluation import NOT_APPLICABLE, EvaluationData, EvaluationOutput, InputT, OutputT -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.skill_selection_accuracy import get_template from .prompt_templates.trajectory_prompt_template import serialize_trajectory @@ -51,11 +52,13 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) # A case with nothing to select from contributes a placeholder 0.0 row; averaging it in # would report a run that had no decision to make as a failed one. self.aggregator = self._aggregate_dropping_na @@ -79,17 +82,23 @@ def _available_str(self, evaluation_case: EvaluationData[InputT, OutputT]) -> st def _has_catalog(self, evaluation_case: EvaluationData[InputT, OutputT]) -> bool: return bool(parse_available_skills(evaluation_case.actual_trajectory)) - def _case_context(self, evaluation_case: EvaluationData[InputT, OutputT]) -> tuple[str, str]: + def _case_context( + self, evaluation_case: EvaluationData[InputT, OutputT], trace_index: TraceIndex | None = None + ) -> tuple[str, str]: """The two halves of the prompt that do not depend on which decision is being judged. Built once per case: the skill catalog and the serialized trajectory are the same for every invoked skill, and serializing a long trajectory once per skill is wasted work. + When `trace_index` is provided the trajectory would overflow the judge, so the paged + trace-overview block replaces the inlined serialization. """ head = f"## Task\n{evaluation_case.input}\n\n## Available skills\n{self._available_str(evaluation_case)}\n\n" - tail = ( - f"## Agent trajectory\n{serialize_trajectory(evaluation_case.actual_trajectory)}\n\n" - f"## Agent's final response\n{evaluation_case.actual_output}" + trajectory = ( + self._disclosed_trace_section(trace_index) + if trace_index is not None + else serialize_trajectory(evaluation_case.actual_trajectory) ) + tail = f"## Agent trajectory\n{trajectory}\n\n## Agent's final response\n{evaluation_case.actual_output}" return head, tail @staticmethod @@ -134,21 +143,22 @@ def _rating_to_output(self, rating: SkillSelectionRating, decision: str) -> Eval label=rating.score.value, ) - def _new_judge(self) -> Agent: + def _new_judge(self, tools: list | None = None) -> Agent: """A fresh judge per decision. Each skill is judged independently, so reusing one `Agent` across the loop would both carry the previous verdicts into the next prompt as conversation history and resend the - whole trajectory on top of it, growing every request. + whole trajectory on top of it, growing every request. `tools` carries the trace-disclosure + tools when the trajectory is read on demand instead of inlined. """ - return Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + return Agent(model=self.model, system_prompt=self.system_prompt, tools=tools or [], callback_handler=None) - def _judge(self, prompt: str) -> SkillSelectionRating: - result = self._new_judge()(prompt, structured_output_model=SkillSelectionRating) + def _judge(self, prompt: str, tools: list | None = None) -> SkillSelectionRating: + result = self._new_judge(tools)(prompt, structured_output_model=SkillSelectionRating) return cast(SkillSelectionRating, result.structured_output) - async def _judge_async(self, prompt: str) -> SkillSelectionRating: - result = await self._new_judge().invoke_async(prompt, structured_output_model=SkillSelectionRating) + async def _judge_async(self, prompt: str, tools: list | None = None) -> SkillSelectionRating: + result = await self._new_judge(tools).invoke_async(prompt, structured_output_model=SkillSelectionRating) return cast(SkillSelectionRating, result.structured_output) @staticmethod @@ -182,10 +192,10 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva invoked = extract_selected_skills(evaluation_case.actual_trajectory) if not invoked: return [self._no_invocation_row(self._has_catalog(evaluation_case))] - context = self._case_context(evaluation_case) + context, tools = self._case_context_and_tools(evaluation_case) results = [] for skill in invoked: - rating = self._judge(self._prompt_for(context, skill)) + rating = self._judge(self._prompt_for(context, skill), tools) results.append(self._rating_to_output(rating, decision=skill.name)) return results @@ -195,9 +205,21 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) invoked = extract_selected_skills(evaluation_case.actual_trajectory) if not invoked: return [self._no_invocation_row(self._has_catalog(evaluation_case))] - context = self._case_context(evaluation_case) + context, tools = self._case_context_and_tools(evaluation_case) results = [] for skill in invoked: - rating = await self._judge_async(self._prompt_for(context, skill)) + rating = await self._judge_async(self._prompt_for(context, skill), tools) results.append(self._rating_to_output(rating, decision=skill.name)) return results + + def _case_context_and_tools(self, evaluation_case: EvaluationData[InputT, OutputT]) -> tuple[tuple[str, str], list]: + """Case context plus the disclosure tools, deciding disclosure once for all skills. + + The trajectory is the same for every invoked skill, so the overflow decision and the + trace tools are resolved once here and reused across the per-skill prompts. + """ + inline_context = self._case_context(evaluation_case) + index = self._resolve_disclosure_index(evaluation_case, inline_context[1]) + if index is None: + return inline_context, [] + return self._case_context(evaluation_case, index), list(index.tools) diff --git a/src/strands_evals/evaluators/stereotyping_evaluator.py b/src/strands_evals/evaluators/stereotyping_evaluator.py index 99f46295..bda11469 100644 --- a/src/strands_evals/evaluators/stereotyping_evaluator.py +++ b/src/strands_evals/evaluators/stereotyping_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.stereotyping import get_template @@ -41,16 +41,20 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: parsed_input = self._get_last_turn(evaluation_case) - prompt = self._format_trace_level_prompt(parsed_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt, tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._format_trace_level_prompt(parsed_input, idx) + ) + evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None) result = evaluator_agent(prompt, structured_output_model=StereotypingRating) rating = cast(StereotypingRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/tool_parameter_accuracy_evaluator.py b/src/strands_evals/evaluators/tool_parameter_accuracy_evaluator.py index 40ee7f1e..a14767a1 100644 --- a/src/strands_evals/evaluators/tool_parameter_accuracy_evaluator.py +++ b/src/strands_evals/evaluators/tool_parameter_accuracy_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.tool_parameter_accuracy import get_template @@ -41,19 +41,24 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: tool_inputs = self._parse_trajectory(evaluation_case) + index, tools = self._tool_level_disclosure(evaluation_case, tool_inputs) results = [] for tool_input in tool_inputs: - prompt = self._format_tool_level_prompt(tool_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._format_tool_level_prompt(tool_input, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = evaluator_agent(prompt, structured_output_model=ToolParameterAccuracyRating) rating = cast(ToolParameterAccuracyRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -70,11 +75,14 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: tool_inputs = self._parse_trajectory(evaluation_case) + index, tools = self._tool_level_disclosure(evaluation_case, tool_inputs) results = [] for tool_input in tool_inputs: - prompt = self._format_tool_level_prompt(tool_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._format_tool_level_prompt(tool_input, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = await evaluator_agent.invoke_async(prompt, structured_output_model=ToolParameterAccuracyRating) rating = cast(ToolParameterAccuracyRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/tool_selection_accuracy_evaluator.py b/src/strands_evals/evaluators/tool_selection_accuracy_evaluator.py index ced8756c..4eb72381 100644 --- a/src/strands_evals/evaluators/tool_selection_accuracy_evaluator.py +++ b/src/strands_evals/evaluators/tool_selection_accuracy_evaluator.py @@ -7,7 +7,7 @@ from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT from ..types.trace import EvaluationLevel -from .evaluator import Evaluator +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.tool_selection_accuracy import get_template @@ -41,19 +41,24 @@ def __init__( model: Model | str | None = None, system_prompt: str | None = None, name: str | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT self.version = version self.model = model + self.disclosure = self._validate_disclosure(disclosure) def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: tool_inputs = self._parse_trajectory(evaluation_case) + index, tools = self._tool_level_disclosure(evaluation_case, tool_inputs) results = [] for tool_input in tool_inputs: - prompt = self._format_tool_level_prompt(tool_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._format_tool_level_prompt(tool_input, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = evaluator_agent(prompt, structured_output_model=ToolSelectionRating) rating = cast(ToolSelectionRating, result.structured_output) normalized_score = self._score_mapping[rating.score] @@ -70,11 +75,14 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: tool_inputs = self._parse_trajectory(evaluation_case) + index, tools = self._tool_level_disclosure(evaluation_case, tool_inputs) results = [] for tool_input in tool_inputs: - prompt = self._format_tool_level_prompt(tool_input) - evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None) + prompt = self._format_tool_level_prompt(tool_input, index) + evaluator_agent = Agent( + model=self.model, system_prompt=self.system_prompt, tools=tools, callback_handler=None + ) result = await evaluator_agent.invoke_async(prompt, structured_output_model=ToolSelectionRating) rating = cast(ToolSelectionRating, result.structured_output) normalized_score = self._score_mapping[rating.score] diff --git a/src/strands_evals/evaluators/trajectory_evaluator.py b/src/strands_evals/evaluators/trajectory_evaluator.py index d46c5870..27c25f41 100644 --- a/src/strands_evals/evaluators/trajectory_evaluator.py +++ b/src/strands_evals/evaluators/trajectory_evaluator.py @@ -6,7 +6,8 @@ from ..tools.evaluation_tools import any_order_match_scorer, exact_match_scorer, in_order_match_scorer from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT -from .evaluator import Evaluator +from ._trace_index import TraceIndex +from .evaluator import DisclosureMode, Evaluator from .prompt_templates.case_prompt_template import compose_test_prompt from .prompt_templates.prompt_templates import judge_trajectory_template_tools as SYSTEM_PROMPT @@ -36,6 +37,7 @@ def __init__( include_inputs: bool = True, name: str | None = None, tools: list[Any] | None = None, + disclosure: DisclosureMode = "auto", ): super().__init__(name=name) self.rubric = rubric @@ -49,6 +51,7 @@ def __init__( *(tools or []), ] self.system_prompt = system_prompt + self.disclosure = self._validate_disclosure(disclosure) def update_trajectory_description(self, new_description: dict) -> None: """ @@ -69,14 +72,14 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva Returns: The results of the evaluation as EvaluationOutput. """ - evaluator_agent = Agent( - model=self.model, system_prompt=self.system_prompt, tools=self._tools, callback_handler=None + evaluation_prompt, disclosure_tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._compose(evaluation_case, idx) ) - evaluation_prompt = compose_test_prompt( - evaluation_case=evaluation_case, - rubric=self.rubric, - include_inputs=self.include_inputs, - uses_trajectory=True, + evaluator_agent = Agent( + model=self.model, + system_prompt=self.system_prompt, + tools=[*self._tools, *disclosure_tools], + callback_handler=None, ) result = evaluator_agent(evaluation_prompt, structured_output_model=EvaluationOutput) return [cast(EvaluationOutput, result.structured_output)] @@ -91,14 +94,25 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) Returns: The results of the evaluation as EvaluationOutput. """ + evaluation_prompt, disclosure_tools = self._render_with_disclosure( + evaluation_case, lambda idx: self._compose(evaluation_case, idx) + ) evaluator_agent = Agent( - model=self.model, system_prompt=self.system_prompt, tools=self._tools, callback_handler=None + model=self.model, + system_prompt=self.system_prompt, + tools=[*self._tools, *disclosure_tools], + callback_handler=None, ) - evaluation_prompt = compose_test_prompt( + result = await evaluator_agent.invoke_async(evaluation_prompt, structured_output_model=EvaluationOutput) + return [cast(EvaluationOutput, result.structured_output)] + + def _compose(self, evaluation_case: EvaluationData[InputT, OutputT], trace_index: TraceIndex | None) -> str: + """Compose the trajectory judge prompt, substituting the trace overview on overflow.""" + override = self._disclosed_trace_section(trace_index) if trace_index is not None else None + return compose_test_prompt( evaluation_case=evaluation_case, rubric=self.rubric, include_inputs=self.include_inputs, uses_trajectory=True, + trajectory_override=override, ) - result = await evaluator_agent.invoke_async(evaluation_prompt, structured_output_model=EvaluationOutput) - return [cast(EvaluationOutput, result.structured_output)] diff --git a/tests/strands_evals/evaluators/test_evaluator_disclosure.py b/tests/strands_evals/evaluators/test_evaluator_disclosure.py new file mode 100644 index 00000000..8bd17e46 --- /dev/null +++ b/tests/strands_evals/evaluators/test_evaluator_disclosure.py @@ -0,0 +1,223 @@ +"""Stage D: the automatic disclosure seam on the base Evaluator. + +Covers the decision (`_resolve_disclosure_index`), the render fallback +(`_render_with_disclosure`), and the mode validation shared by every judge, +plus an end-to-end pass through TrajectoryEvaluator to confirm a fitting case +is byte-identical to before and an overflowing case hands the judge the three +trace tools. +""" + +from datetime import datetime, timezone +from unittest.mock import Mock, patch + +import pytest + +from strands_evals.evaluators import Evaluator, TrajectoryEvaluator +from strands_evals.evaluators._trace_index import TraceIndex +from strands_evals.types import EvaluationData, EvaluationOutput +from strands_evals.types.trace import ( + AgentInvocationSpan, + Session, + SpanInfo, + ToolCall, + ToolExecutionSpan, + ToolResult, + Trace, +) + +# The preflight threshold is DEFAULT_MAX_INPUT_TOKENS (200K) * PREFLIGHT_SAFETY_MARGIN +# (0.65) ~= 130K tokens. A probe comfortably past that overflows under "auto" +# regardless of whether tiktoken or the char/4 fallback does the counting. +_OVERFLOW_PROBE = "token " * 300_000 +_FITS_PROBE = "a short prompt that fits any judge window" + + +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 _small_session() -> Session: + """A one-turn session that fits any judge window when inlined.""" + return Session( + traces=[ + Trace( + spans=[ + AgentInvocationSpan( + span_info=_span_info(0), + user_prompt="What is 2 + 2?", + agent_response="4", + available_tools=[], + ) + ], + trace_id="t1", + session_id="s1", + ) + ], + session_id="s1", + ) + + +def _huge_session() -> Session: + """A session whose serialized trajectory overflows a 200K-token judge.""" + spans = [ + AgentInvocationSpan( + span_info=_span_info(0), + user_prompt="Summarize the run.", + agent_response="done", + available_tools=[], + ) + ] + for i in range(1, 6): + spans.append( + ToolExecutionSpan( + span_info=_span_info(i), + tool_call=ToolCall(name="query_db", arguments={"page": i}), + tool_result=ToolResult(content="filler row " * 60_000), + ) + ) + return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1") + + +def _case(trajectory) -> EvaluationData: + return EvaluationData( + input="q", + actual_output="a", + actual_trajectory=trajectory, + ) + + +# --- _validate_disclosure -------------------------------------------------- + + +@pytest.mark.parametrize("mode", ["auto", "always", "never"]) +def test_validate_disclosure_accepts_known_modes(mode): + assert Evaluator._validate_disclosure(mode) == mode + + +def test_validate_disclosure_rejects_unknown_mode(): + with pytest.raises(ValueError, match="disclosure"): + Evaluator._validate_disclosure("sometimes") + + +def test_constructor_validates_disclosure(): + with pytest.raises(ValueError, match="disclosure"): + TrajectoryEvaluator(rubric="r", disclosure="bogus") + + +# --- _resolve_disclosure_index --------------------------------------------- + + +def test_auto_fits_does_not_disclose(): + ev = Evaluator() + ev.disclosure = "auto" + assert ev._resolve_disclosure_index(_case(_small_session()), _FITS_PROBE) is None + + +def test_auto_overflow_discloses(): + ev = Evaluator() + ev.disclosure = "auto" + index = ev._resolve_disclosure_index(_case(_small_session()), _OVERFLOW_PROBE) + assert isinstance(index, TraceIndex) + + +def test_never_does_not_disclose_even_on_overflow(): + ev = Evaluator() + ev.disclosure = "never" + assert ev._resolve_disclosure_index(_case(_small_session()), _OVERFLOW_PROBE) is None + + +def test_always_discloses_when_session_present_even_if_it_fits(): + ev = Evaluator() + ev.disclosure = "always" + index = ev._resolve_disclosure_index(_case(_small_session()), _FITS_PROBE) + assert isinstance(index, TraceIndex) + + +def test_no_session_never_discloses(): + ev = Evaluator() + for mode in ("auto", "always"): + ev.disclosure = mode + # A non-Session trajectory (e.g. a plain list of steps) has no spans to index. + assert ev._resolve_disclosure_index(_case(["step one", "step two"]), _OVERFLOW_PROBE) is None + + +# --- _render_with_disclosure ----------------------------------------------- + + +def test_render_inline_path_is_byte_identical_and_toolless(): + ev = Evaluator() + ev.disclosure = "auto" + prompt, tools = ev._render_with_disclosure( + _case(_small_session()), + lambda idx: _FITS_PROBE if idx is None else ev._disclosed_trace_section(idx), + ) + assert prompt == _FITS_PROBE + assert tools == [] + + +def test_render_disclosure_path_swaps_prompt_and_adds_three_tools(): + ev = Evaluator() + ev.disclosure = "auto" + prompt, tools = ev._render_with_disclosure( + _case(_small_session()), + lambda idx: _OVERFLOW_PROBE if idx is None else ev._disclosed_trace_section(idx), + ) + assert prompt != _OVERFLOW_PROBE + assert "too large to inline" in prompt + tool_names = {getattr(t, "tool_name", getattr(t, "__name__", "")) for t in tools} + assert {"list_spans", "get_span", "search_spans"} == tool_names + + +# --- end-to-end through TrajectoryEvaluator -------------------------------- + + +@patch("strands_evals.evaluators.trajectory_evaluator.Agent") +def test_trajectory_fits_keeps_only_scoring_tools(mock_agent_class): + mock_agent = Mock() + mock_agent.return_value = Mock(structured_output=EvaluationOutput(score=1.0, test_pass=True, reason="ok")) + mock_agent_class.return_value = mock_agent + + evaluator = TrajectoryEvaluator(rubric="r", disclosure="auto") + evaluator.evaluate(_case(_small_session())) + + tools = mock_agent_class.call_args[1]["tools"] + tool_names = {getattr(t, "tool_name", getattr(t, "__name__", "")) for t in tools} + assert not ({"list_spans", "get_span", "search_spans"} & tool_names) + + +@patch("strands_evals.evaluators.trajectory_evaluator.Agent") +def test_trajectory_overflow_adds_disclosure_tools(mock_agent_class): + mock_agent = Mock() + mock_agent.return_value = Mock(structured_output=EvaluationOutput(score=1.0, test_pass=True, reason="ok")) + mock_agent_class.return_value = mock_agent + + evaluator = TrajectoryEvaluator(rubric="r", disclosure="auto") + evaluator.evaluate(_case(_huge_session())) + + tools = mock_agent_class.call_args[1]["tools"] + tool_names = {getattr(t, "tool_name", getattr(t, "__name__", "")) for t in tools} + assert {"list_spans", "get_span", "search_spans"} <= tool_names + # The judge is handed the overview, not the inlined multi-hundred-K-token trajectory. + prompt = mock_agent.call_args[0][0] + assert "too large to inline" in prompt + + +@patch("strands_evals.evaluators.trajectory_evaluator.Agent") +def test_trajectory_never_inlines_on_overflow(mock_agent_class): + mock_agent = Mock() + mock_agent.return_value = Mock(structured_output=EvaluationOutput(score=1.0, test_pass=True, reason="ok")) + mock_agent_class.return_value = mock_agent + + evaluator = TrajectoryEvaluator(rubric="r", disclosure="never") + evaluator.evaluate(_case(_huge_session())) + + tool_names = {getattr(t, "tool_name", getattr(t, "__name__", "")) for t in mock_agent_class.call_args[1]["tools"]} + assert not ({"list_spans", "get_span", "search_spans"} & tool_names) + # "never" restores today's behavior: the full trajectory is inlined, so a real + # overflow is surfaced downstream as could-not-evaluate rather than disclosed here. + assert "too large to inline" not in mock_agent.call_args[0][0] 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..56e69ede --- /dev/null +++ b/tests/strands_evals/tools/test_trace_index.py @@ -0,0 +1,464 @@ +from datetime import datetime, timedelta, timezone + +import pytest + +from strands_evals.evaluators._trace_index import TraceIndex +from strands_evals.types.trace import ( + AgentInvocationSpan, + InferenceSpan, + Session, + SpanInfo, + TextContent, + ToolCall, + ToolConfig, + ToolExecutionSpan, + ToolResult, + ToolResultContent, + Trace, + UserMessage, +) + + +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=tz), + end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=tz), + ) + + +@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.", + 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), + 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] + # 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): + 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_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] + + 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_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="$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]") + + +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 + + +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_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 _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( + 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] + + 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 + # Same tools, but a copy: mutating the returned list must not touch the index's set. + assert tools == index.tools + assert tools is not index.tools + tools.clear() + assert len(index.tools) == 3 + + +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) + + +def test_search_matches_phrase_straddling_a_newline_in_source(): + """A literal a judge copies from the (whitespace-collapsed) preview matches source text + that had a newline where the preview shows a space.""" + spans = [ + ToolExecutionSpan( + span_info=_span_info(0), + tool_call=ToolCall(name="refund", arguments={}), + tool_result=ToolResult(content="Refund approved.\nAmount: $150 total."), + ) + ] + index = TraceIndex(Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1")) + search = index.tools[2] + + # The phrase spans the newline in the source but appears with a space in the preview. + result = search(pattern="approved. Amount") + assert "No matches" not in result + assert "[0]" in result + + +def test_search_rejects_empty_pattern(session): + search = TraceIndex(session).tools[2] + + assert search(pattern="") == "ERROR: empty pattern; provide text to search for" + assert search(pattern=" ") == "ERROR: empty pattern; provide text to search for" + + +def test_get_span_on_empty_session_reports_no_spans(): + get_span = TraceIndex(Session(traces=[], session_id="s1")).tools[1] + + assert get_span(index=0) == "ERROR: trace has no spans" 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..a4f081f1 --- /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.evaluators._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="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="refund_amount=$999") + assert "No matches" in claimed + + # But the actual amount is present: the claim contradicts the evidence + actual = search_spans(pattern="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..ca1c01f9 --- /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.evaluators._trace_index import TraceIndex +from strands_evals.mappers import OpenInferenceSessionMapper, StrandsInMemorySessionMapper +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 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..ae03dcfd --- /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.evaluators._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