Module scenario.judge_agent
Use the Judge Agent module in Scenario to evaluate conversation quality and LLM reasoning during AI agent testing.
This module provides the JudgeAgent class, which evaluates ongoing conversations between users and agents to determine if success criteria are met. The judge makes real-time decisions about whether scenarios should continue or end with success/failure verdicts.
Expand source code
"""
Use the Judge Agent module in Scenario to evaluate conversation quality and LLM reasoning during AI agent testing.
This module provides the JudgeAgent class, which evaluates ongoing conversations
between users and agents to determine if success criteria are met. The judge
makes real-time decisions about whether scenarios should continue or end with
success/failure verdicts.
"""
import json
import logging
import re
from dataclasses import dataclass, field as dataclass_field
from typing import Any, List, Optional, Sequence, Union, cast
import litellm
from litellm import Choices
from litellm.files.main import ModelResponse
from openai.types.chat import ChatCompletionMessageParam
from scenario.cache import scenario_cache
from scenario.agent_adapter import AgentAdapter
from scenario.config import ModelConfig, ScenarioConfig
from ._error_messages import agent_not_configured_error_message
from ._judge import JudgeUtils, judge_span_digest_formatter
from ._judge.estimate_tokens import estimate_tokens, DEFAULT_TOKEN_THRESHOLD
from ._judge.trace_tools import expand_trace, grep_trace
from ._judge.transcript_tools import (
build_transcript_skeleton,
expand_transcript,
grep_transcript,
)
from ._tracing import (
judge_span_collector,
JudgeSpanCollector,
remote_trace_fetcher as default_remote_trace_fetcher,
RemoteTraceFetcher,
)
from ._tracing.remote_trace_fetcher import DEFAULT_TRACE_WAIT_TIMEOUT_SECONDS
from .types import AgentInput, AgentReturnTypes, AgentRole, ScenarioResult
from .voice._transcribe import transcribe_segments
from .voice.modality_resolver import ModalityTier, resolve_modality
logger = logging.getLogger("scenario")
# `/v1/chat/completions` refuses function tools on some reasoning models unless
# reasoning is explicitly switched off:
#
# Function tools with reasoning_effort are not supported for <model> in
# /v1/chat/completions. To use function tools, use /v1/responses or set
# reasoning_effort to 'none'.
#
# The judge forces a finish_test / continue_test tool call on every graded run,
# so on such a model no run could reach a verdict (langwatch/scenario#864, and
# the same signature on the LangWatch platform judge, langwatch/langwatch#6369).
#
# Reasoning is disabled by RETRY, never preemptively: whether a model accepts
# reasoning off is not knowable up front (Gemini 2.5 Pro rejects it with
# "Budget 0 is invalid. This model only works in thinking mode."), so the call
# goes out untouched and is re-sent with reasoning off only when the provider's
# rejection asks for exactly that. Models that work today are never sent
# anything new.
_REASONING_OFF = "none"
def _rejection_asks_for_reasoning_off(error: Exception) -> bool:
"""
Whether a provider rejection is the "set reasoning_effort to 'none' to use
function tools" class, as opposed to any other bad request.
Keyed on the remediation directive, not just the field name: an error such
as "reasoning_effort 'none' is invalid for this model" mentions both tokens
but is not asking us to turn reasoning off, and retrying it with reasoning
off would replace the provider's real error.
"""
return "set reasoning_effort to 'none'" in str(error)
_DISCOVERY_TOOL_NAMES = frozenset(
{"expand_trace", "grep_trace", "expand_transcript", "grep_transcript"}
)
REMOTE_TRACES_JUDGE_RULE = (
"Criteria about the agent's internal behavior (tool calls, database "
"writes, API calls, retrievals) must be verified against the "
"<opentelemetry_traces> section, not against claims in the transcript. "
"If a span named langwatch.span_collection.error is present, read its "
"reason: when no agent spans arrived, mark criteria that depend on "
"internal behavior as inconclusive, never passed. When the trace is "
"incomplete, criteria proven by the spans that are present may pass, and "
"criteria whose evidence is missing stay inconclusive. Criteria about "
"the conversation itself are unaffected by missing traces: judge them "
"from the transcript as normal. Never mark internal-behavior criteria "
"as passed based on the transcript alone."
)
"""Rule appended to the verdict system prompt when fetch_remote_traces is on."""
DECISION_PHASE_RULE = (
"In this step, only decide whether the conversation has collected enough "
"information to evaluate the criteria: call make_verdict when it has, or "
"continue_test to let the conversation play out. Do not decide whether "
"the criteria pass or fail now: that evaluation happens in a separate "
"step after the conversation ends."
)
"""Appended to a custom system prompt on decision calls, so custom judge
personas still drive the argument-free decision tools correctly."""
REMOTE_TRACES_DECISION_RULE = (
"The agent's execution traces are fetched and verified at the verdict, "
"after the conversation ends; they are not part of this decision. Do not "
"continue the conversation only to wait for trace evidence, and do not "
"end it early to see traces sooner."
)
"""Appended to the decision system prompt when fetch_remote_traces is on."""
_DECISION_TOOL_NAMES = frozenset({"continue_test", "make_verdict"})
class _WaitForTracesRequested:
"""Sentinel type: the judge called wait_for_traces instead of a verdict."""
_WAIT_FOR_TRACES_REQUESTED = _WaitForTracesRequested()
def _build_wait_for_traces_tool() -> dict:
"""The verdict phase's one-shot extension tool.
Offered only when the remote traces are still incomplete after the
settle-wait, and withdrawn after one use: the second verdict call must
decide on the evidence it has. Byte-identical description with the
TypeScript SDK.
"""
return {
"type": "function",
"function": {
"name": "wait_for_traces",
"description": (
"The remote trace evidence is still incomplete and the "
"missing spans are essential for the verdict. Wait one more "
"period for them to arrive. Available once: after this wait "
"the verdict must be delivered on the evidence at hand. Only "
"call this when a criterion genuinely depends on the missing "
"spans; otherwise deliver the verdict now."
),
"strict": True,
"parameters": {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
},
},
}
def _response_called_wait_for_traces(response: ModelResponse) -> bool:
"""True when the completion called the wait_for_traces tool."""
if not hasattr(response, "choices") or len(response.choices) == 0:
return False
message = cast(Choices, response.choices[0]).message
return any(
tc.function.name == "wait_for_traces" for tc in (message.tool_calls or [])
)
def _distinct_message_trace_ids(messages: Sequence[Any]) -> List[str]:
"""All distinct trace ids stamped on the conversation's messages, in
first-seen order. Every turn stamps its own trace id, so this covers the
whole conversation, never only the last turn."""
seen: List[str] = []
for message in messages:
trace_id = message.get("trace_id") if isinstance(message, dict) else None
if isinstance(trace_id, str) and trace_id and trace_id not in seen:
seen.append(trace_id)
return seen
def _render_judge_content(
*, transcript: str, traces_digest: str, extra_context_section: str
) -> str:
"""Renders the judge's user-message content from its three sections."""
return f"""
<transcript>
{transcript}
</transcript>
<opentelemetry_traces>
{traces_digest}
</opentelemetry_traces>{extra_context_section}
"""
@dataclass
class _ConversationView:
"""Transcript-side context shared by the decision and verdict phases.
Built once per judge call: audio transcription and transcript sizing are
identical in both phases, while the span digest is rebuilt per phase (the
verdict phase sees the settled remote spans)."""
working_messages: List[Any]
transcript_for_prompt: str
is_large_transcript: bool
extra_context_section: str
@dataclass
class _DecisionOutcome:
"""Result of the decision phase.
``discovery_recap`` is only populated on exhaustion: the decision loop's
discovery cycles collapsed to plain-text assistant messages, so the forced
verdict call keeps the information the judge already gathered instead of
starting from a blank digest."""
decision: str # "continue" | "verdict" | "exhausted"
discovery_recap: List[dict] = dataclass_field(default_factory=list)
def _stringify_tool_output(output: Any) -> str:
"""Best-effort stringify of a tool result for a plain-text recap."""
if isinstance(output, str):
return output
if isinstance(output, dict):
value = output.get("value")
if isinstance(value, str):
return value
try:
return json.dumps(output)
except (TypeError, ValueError):
return str(output)
try:
return json.dumps(output)
except (TypeError, ValueError):
return str(output)
def _collapse_discovery_history(messages: List[dict]) -> List[dict]:
"""
Rewrites message history so every discovery cycle
(assistant tool_call for expand_trace/grep_trace → tool result)
collapses into a single plain-text assistant message recounting what
the judge called and what came back.
Required before a forced verdict so we can strip expand_trace /
grep_trace from the tool set without Anthropic rejecting the call
for referencing undefined tools, and so the model physically cannot
emit a discovery tool again.
If an assistant message mixes discovery and non-discovery tool calls,
only the discovery calls are collapsed to text; non-discovery calls
and their corresponding tool results are preserved unchanged.
Messages without any discovery content pass through unchanged.
"""
out: List[dict] = []
i = 0
while i < len(messages):
msg = messages[i]
tool_calls = msg.get("tool_calls") if isinstance(msg, dict) else None
has_discovery_call = (
msg.get("role") == "assistant"
and isinstance(tool_calls, list)
and any(
tc.get("function", {}).get("name") in _DISCOVERY_TOOL_NAMES
for tc in tool_calls
)
)
if has_discovery_call:
assert isinstance(tool_calls, list)
# Gather ALL consecutive following tool result messages (covers
# both discovery and non-discovery results).
result_by_id: dict = {}
j = i + 1
while (
j < len(messages)
and messages[j].get("role") == "tool"
and messages[j].get("tool_call_id")
):
result_by_id[messages[j]["tool_call_id"]] = messages[j].get(
"content", ""
)
j += 1
discovery_calls = [
tc for tc in tool_calls
if tc.get("function", {}).get("name") in _DISCOVERY_TOOL_NAMES
]
non_discovery_calls = [
tc for tc in tool_calls
if tc.get("function", {}).get("name") not in _DISCOVERY_TOOL_NAMES
]
discovery_ids = {tc.get("id") for tc in discovery_calls}
lines: List[str] = []
leading_text = msg.get("content") or ""
if leading_text:
lines.append(str(leading_text))
for tc in discovery_calls:
name = tc.get("function", {}).get("name", "unknown_tool")
raw_args = tc.get("function", {}).get("arguments", "")
try:
parsed = json.loads(raw_args) if raw_args else {}
args_str = json.dumps(parsed)
except (TypeError, ValueError):
args_str = str(raw_args)
body = _stringify_tool_output(result_by_id.get(tc.get("id")))
lines.append(f"[Called {name} with {args_str}]\n{body}")
new_msg: dict = {"role": "assistant", "content": "\n\n".join(lines)}
if non_discovery_calls:
new_msg["tool_calls"] = non_discovery_calls
out.append(new_msg)
# Re-emit tool result messages only for non-discovery calls so
# their tool references remain valid in the stripped tool set.
for k in range(i + 1, j):
result_msg = messages[k]
if result_msg.get("tool_call_id") not in discovery_ids:
out.append(result_msg)
i = j
continue
out.append(msg)
i += 1
return out
def _criteria_keys(criteria: Sequence[str]) -> List[str]:
"""Sanitized schema property names for each criterion.
Must stay the single source of truth for these keys: the finish_test
tool schema declares them as required properties, and _parse_response
maps the LLM's verdicts back to criteria BY these keys. If the two ever
computed them differently, every verdict would silently fail to map.
"""
return [
re.sub(r"[^a-zA-Z0-9]", "_", c.replace(" ", "_").replace("'", "").lower())[:70]
for c in criteria
]
class JudgeAgent(AgentAdapter):
"""
Agent that evaluates conversations against success criteria.
The JudgeAgent watches conversations in real-time and makes decisions about
whether the agent under test is meeting the specified criteria. It can either
allow the conversation to continue or end it with a success/failure verdict.
The judge uses function calling to make structured decisions and provides
detailed reasoning for its verdicts. It evaluates each criterion independently
and provides comprehensive feedback about what worked and what didn't.
Attributes:
role: Always AgentRole.JUDGE for judge agents
model: LLM model identifier to use for evaluation
api_base: Optional base URL where the model is hosted
api_key: Optional API key for the model provider
temperature: Sampling temperature for evaluation consistency
max_tokens: Maximum tokens for judge reasoning
criteria: List of success criteria to evaluate against
system_prompt: Custom system prompt to override default judge behavior
Example:
```
import scenario
# Basic judge agent with criteria
judge = scenario.JudgeAgent(
criteria=[
"Agent provides helpful responses",
"Agent asks relevant follow-up questions",
"Agent does not provide harmful information"
]
)
# Customized judge with specific model and behavior
strict_judge = scenario.JudgeAgent(
model="openai/gpt-4.1-mini",
criteria=[
"Code examples are syntactically correct",
"Explanations are technically accurate",
"Security best practices are mentioned"
],
temperature=0.0, # More deterministic evaluation
system_prompt="You are a strict technical reviewer evaluating code quality."
)
# Use in scenario
result = await scenario.run(
name="coding assistant test",
description="User asks for help with Python functions",
agents=[
coding_agent,
scenario.UserSimulatorAgent(),
judge
]
)
print(f"Passed criteria: {result.passed_criteria}")
print(f"Failed criteria: {result.failed_criteria}")
```
Note:
- Judge agents evaluate conversations continuously, not just at the end
- They can end scenarios early if clear success/failure conditions are met
- Provide detailed reasoning for their decisions
- Support both positive criteria (things that should happen) and negative criteria (things that shouldn't)
"""
role = AgentRole.JUDGE
model: str
api_base: Optional[str]
api_key: Optional[str]
temperature: float
max_tokens: Optional[int]
criteria: List[str]
system_prompt: Optional[str]
_extra_params: dict
_span_collector: JudgeSpanCollector
_remote_trace_fetcher: RemoteTraceFetcher
_token_threshold: int
_max_discovery_steps: int
def __init__(
self,
*,
criteria: Optional[List[str]] = None,
model: Optional[str] = None,
api_base: Optional[str] = None,
api_key: Optional[str] = None,
temperature: float = 0.0,
max_tokens: Optional[int] = None,
system_prompt: Optional[str] = None,
span_collector: Optional[JudgeSpanCollector] = None,
remote_trace_fetcher: Optional[RemoteTraceFetcher] = None,
token_threshold: int = DEFAULT_TOKEN_THRESHOLD,
max_discovery_steps: int = 10,
include_audio: Optional[bool] = None,
include_timeline: Optional[bool] = None,
include_traces: Optional[bool] = None,
modality: Optional[str] = None,
**extra_params,
):
"""
Initialize a judge agent with evaluation criteria.
Args:
criteria: List of success criteria to evaluate the conversation against.
Can include both positive requirements ("Agent provides helpful responses")
and negative constraints ("Agent should not provide personal information").
model: LLM model identifier (e.g., "openai/gpt-4.1-mini").
If not provided, uses the default model from global configuration.
api_base: Optional base URL where the model is hosted. If not provided,
uses the base URL from global configuration.
api_key: API key for the model provider. If not provided,
uses the key from global configuration or environment.
temperature: Sampling temperature for evaluation (0.0-1.0).
Lower values (0.0-0.2) recommended for consistent evaluation.
max_tokens: Maximum number of tokens for judge reasoning and explanations.
system_prompt: Custom system prompt to override default judge behavior.
Use this to create specialized evaluation perspectives.
span_collector: Optional span collector for telemetry. Defaults to global singleton.
remote_trace_fetcher: Optional fetcher for remote traces, used when
the scenario enables ``fetch_remote_traces``.
Defaults to the global singleton.
token_threshold: Estimated token count above which traces switch to
structure-only rendering with progressive discovery tools.
Defaults to 8192.
max_discovery_steps: Maximum number of expand/grep tool calls the judge
can make before being forced to return a verdict.
Defaults to 10.
modality: Explicit modality declaration for this role. Accepted values:
``"audio-in"`` (LLM receives raw audio), ``"stt-bridge"``
(audio transcribed to text before the LLM), or ``"text"``
(no audio in the stack). Complementary to ``include_audio``:
``include_audio=True/False`` takes precedence; ``modality=``
applies when ``include_audio`` is ``None``. When ``None``
(default), the modality is auto-detected from litellm capabilities.
Raises:
Exception: If no model is configured either in parameters or global config
Example:
```
# Customer service judge
cs_judge = JudgeAgent(
criteria=[
"Agent replies with the refund policy",
"Agent offers next steps for the customer",
],
temperature=0.1
)
# Technical accuracy judge
tech_judge = JudgeAgent(
criteria=[
"Agent adds a code review pointing out the code compilation errors",
"Agent adds a code review about the missing security headers"
],
system_prompt="You are a senior software engineer reviewing code for production use."
)
```
Note:
Advanced usage: Additional parameters can be passed as keyword arguments
(e.g., headers, timeout, client) for specialized configurations. These are
experimental and may not be supported in future versions.
"""
self.criteria = criteria or []
self.api_base = api_base
self.api_key = api_key
self.temperature = temperature
self.max_tokens = max_tokens
self.system_prompt = system_prompt
self._span_collector = span_collector or judge_span_collector
self._remote_trace_fetcher = remote_trace_fetcher or default_remote_trace_fetcher
self._token_threshold = token_threshold
self._max_discovery_steps = max_discovery_steps
# Voice-aware judge behaviour (§4.3). None = auto-detect based on
# conversation content and judge model capabilities.
self.include_audio = include_audio
self.include_timeline = include_timeline
self.include_traces = include_traces
self.modality = modality
if model:
self.model = model
if ScenarioConfig.default_config is not None and isinstance(
ScenarioConfig.default_config.default_model, str
):
self.model = model or ScenarioConfig.default_config.default_model
self._extra_params = extra_params
elif ScenarioConfig.default_config is not None and isinstance(
ScenarioConfig.default_config.default_model, ModelConfig
):
self.model = model or ScenarioConfig.default_config.default_model.model
self.api_base = (
api_base or ScenarioConfig.default_config.default_model.api_base
)
self.api_key = (
api_key or ScenarioConfig.default_config.default_model.api_key
)
self.temperature = (
temperature or ScenarioConfig.default_config.default_model.temperature
)
self.max_tokens = (
max_tokens or ScenarioConfig.default_config.default_model.max_tokens
)
# Extract extra params from ModelConfig
config_dict = ScenarioConfig.default_config.default_model.model_dump(
exclude_none=True
)
config_dict.pop("model", None)
config_dict.pop("api_base", None)
config_dict.pop("api_key", None)
config_dict.pop("temperature", None)
config_dict.pop("max_tokens", None)
# Merge: config extras < agent extra_params
self._extra_params = {**config_dict, **extra_params}
else:
self._extra_params = extra_params
if not hasattr(self, "model"):
raise Exception(agent_not_configured_error_message("JudgeAgent"))
# --------------------------------------------- voice auto-detection (§4.3)
def effective_include_audio(self, conversation_has_audio: bool) -> bool:
"""Resolve include_audio: explicit wins, otherwise use modality resolver.
Intentional behavior change (Bundle 3 / AC3b):
Before: gpt-4o → audio-capable (substring match).
After: gpt-4o → text path (litellm advisory returns False).
Before: gpt-audio-mini → NOT audio-capable (not in list).
After: gpt-audio-mini → audio-capable (litellm advisory returns True).
The old substring list was wrong; the resolver is the source of truth.
"""
if self.include_audio is not None:
# Explicit override always wins (AC3c)
return self.include_audio and conversation_has_audio
# Use resolver with per-role declaration (AC0, Bundle 6)
tier, warnings = resolve_modality(declaration=self.modality, model_id=self.model or "")
for w in warnings:
logger.warning(w)
return conversation_has_audio and (tier == ModalityTier.AUDIO_IN)
def effective_include_timeline(self, conversation_has_audio: bool) -> bool:
"""Default timeline True for voice, False for text — unless explicitly set."""
if self.include_timeline is not None:
return self.include_timeline
return conversation_has_audio
def effective_include_traces(self, otel_configured: bool) -> bool:
if self.include_traces is not None:
return self.include_traces
return otel_configured
# --------------------------------- audio-transcription fallback helpers
@staticmethod
def _conversation_has_audio(messages: List[Any]) -> bool:
"""Return True if any message content contains an audio part."""
for msg in messages:
content = msg.get("content") if isinstance(msg, dict) else None
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get("type") in ("input_audio", "audio"):
return True
return False
@staticmethod
def _extract_recording(input: AgentInput) -> Any:
"""Return the VoiceRecording from the executor, or None."""
scenario_state = getattr(input, "scenario_state", None)
if scenario_state is None:
return None
executor = getattr(scenario_state, "_executor", None)
if executor is None:
return None
return getattr(executor, "_voice_recording", None)
@scenario_cache()
async def call(
self,
input: AgentInput,
) -> AgentReturnTypes:
"""
Evaluate the current conversation state against the configured criteria.
The judge runs in two phases. A mid-conversation call first makes an
argument-free decision between continuing the conversation and moving
to the verdict (continue_test / make_verdict); only a make_verdict
decision triggers the verdict call, which settle-waits for remote
traces when enabled and evaluates every criterion with finish_test.
The last turn and an explicit judgment request skip the decision and
go straight to the verdict.
Args:
input: AgentInput containing conversation history and scenario context
Returns:
AgentReturnTypes: Either an empty list (continue scenario) or a
ScenarioResult (end scenario with verdict)
Raises:
Exception: If the judge cannot make a valid decision or if there's an
error in the evaluation process
Note:
- Returns empty list [] to continue the scenario
- Returns ScenarioResult to end with success/failure
- Provides detailed reasoning for all decisions
- Evaluates each criterion independently
- Can end scenarios early if clear violation or success is detected
"""
effective_criteria = (
input.judgment_request.criteria
if input.judgment_request and input.judgment_request.criteria is not None
else self.criteria
)
max_turns = input.scenario_state.config.max_turns or 10
is_last_message = (
input.scenario_state.current_turn >= max_turns - 1
)
enforce_judgment = input.judgment_request is not None
has_criteria = len(effective_criteria) > 0
if enforce_judgment and not has_criteria:
return ScenarioResult(
success=False,
messages=[],
reasoning="TestingAgent was called as a judge, but it has no criteria to judge against",
)
# A judgment is required when the conversation cannot continue past
# this call: the last turn, or an explicit judgment_request. Both go
# straight to the verdict phase; only an unforced mid-conversation
# call runs the decision phase first.
judgment_required = is_last_message or enforce_judgment
# min_turns floor (ADR-005): below the floor the decision is
# predetermined (the conversation must continue), so nothing is spent
# on it. The check runs before the conversation view is built, or a
# gated voice turn pays for a transcription it discards. The judge
# observes a 0-based current_turn: reset() overrides the initial
# _new_turn() back to 0, so the call on turn N sees current_turn N-1.
# The floor is unmet while current_turn < min_turns: with min_turns=4,
# the first decision call happens on the turn-5 call. A required
# judgment is never gated.
min_turns = getattr(input.scenario_state.config, "min_turns", None)
if (
not judgment_required
and isinstance(min_turns, int)
and input.scenario_state.current_turn < min_turns
):
return []
fetch_remote_traces, trace_wait_timeout, trace_wait_extension = (
self._remote_trace_settings(input)
)
view = await self._build_conversation_view(input)
discovery_recap: List[dict] = []
if judgment_required:
verdict_forced = True
else:
outcome = self._run_decision_phase(
input=input,
effective_criteria=effective_criteria,
view=view,
fetch_remote_traces=fetch_remote_traces,
)
if outcome.decision == "continue":
return []
# "verdict": the judge chose to end the conversation. Its verdict
# stays voluntary so an inconclusive outcome continues the
# conversation (#886). "exhausted": the decision loop burned its
# discovery steps without deciding; the verdict is forced so the
# run cannot churn through discovery again every turn.
verdict_forced = outcome.decision == "exhausted"
discovery_recap = outcome.discovery_recap
return await self._run_judgment_phase(
input=input,
effective_criteria=effective_criteria,
view=view,
fetch_remote_traces=fetch_remote_traces,
trace_wait_timeout=trace_wait_timeout,
trace_wait_extension=trace_wait_extension,
is_last_message=is_last_message,
verdict_forced=verdict_forced,
discovery_recap=discovery_recap,
)
async def _build_conversation_view(self, input: AgentInput) -> _ConversationView:
"""Builds the transcript-side context both phases share.
When the judge model can't ingest audio, transcribes agent audio and
substitutes text so the judge can evaluate the content. The transcript
is gated on its own estimated size, independent of the span digest: it
is built from input.messages regardless of whether the agent under
test routed its calls through litellm, so it can be arbitrarily large
even when the span trace stays small (issue #836).
"""
conversation_has_audio = self._conversation_has_audio(input.messages)
working_messages = input.messages
if conversation_has_audio and not self.effective_include_audio(conversation_has_audio):
recording = self._extract_recording(input)
if recording is not None:
await transcribe_segments(recording)
working_messages = _enrich_messages_with_transcripts(
input.messages, recording
)
transcript = JudgeUtils.build_transcript_from_messages(working_messages)
is_large_transcript = estimate_tokens(transcript) > self._token_threshold
if is_large_transcript:
transcript_for_prompt = (
build_transcript_skeleton(working_messages)
+ "\n\nUse expand_transcript(indices) to see full message content or grep_transcript(pattern) to search across messages. Reference messages by the index shown in brackets."
)
else:
transcript_for_prompt = transcript
extra_context = (
input.judgment_request.additional_context
if input.judgment_request and input.judgment_request.additional_context
else None
)
extra_context_section = (
f"\n<additional_context>\n{extra_context}\n</additional_context>"
if extra_context
else ""
)
return _ConversationView(
working_messages=list(working_messages),
transcript_for_prompt=transcript_for_prompt,
is_large_transcript=is_large_transcript,
extra_context_section=extra_context_section,
)
def _build_decision_system_prompt(
self,
*,
description: str,
criteria: Sequence[str],
fetch_remote_traces: bool,
) -> str:
"""System prompt for the decision phase.
The decision deliberately carries no verdict vocabulary: the judge is
told NOT to decide pass or fail yet, so nothing in this call can
pre-commit it to an outcome before the verdict phase sees the full
evidence.
"""
if self.system_prompt:
content = self.system_prompt + "\n\n" + DECISION_PHASE_RULE
if fetch_remote_traces:
content += "\n\n" + REMOTE_TRACES_DECISION_RULE
return content
criteria_str = "\n".join(
[f"{idx + 1}. {criterion}" for idx, criterion in enumerate(criteria)]
)
remote_rule = (
f"\n- {REMOTE_TRACES_DECISION_RULE}" if fetch_remote_traces else ""
)
return f"""
<role>
You are an LLM as a judge watching a simulated conversation as it plays out live to decide if it has collected enough information to evaluate the agent under test.
</role>
<goal>
Your goal is to decide if the conversation has collected enough information to evaluate the criteria, or if it should continue for longer. Do not decide whether the criteria pass or fail now: that evaluation happens in a separate step after the conversation ends. If enough information has been collected, call the make_verdict tool; if not, call the continue_test tool to let the next step play out.
</goal>
<scenario>
{description}
</scenario>
<criteria>
{criteria_str}
</criteria>
<rules>
- Call make_verdict as soon as the agent has clearly broken one of the "do not" or "should not" criteria; more conversation cannot repair a violation.
- Scenario simulations exist to exercise multi-turn conversations: while the conversation is still short, lean towards continuing, and end it only when more turns would clearly add no information for the criteria.{remote_rule}
</rules>
"""
def _build_decision_tools(self) -> List[dict]:
"""Argument-free decision tools.
No reasoning field on purpose: writing reasoning here would push the
judge to pre-commit to pass or fail before the evidence is complete,
and the text itself is wasted tokens for a binary transition.
"""
return [
{
"type": "function",
"function": {
"name": "continue_test",
"description": "Continue the test with the next step",
"strict": True,
"parameters": {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "make_verdict",
"description": (
"The conversation has collected enough information to "
"evaluate the criteria. End the conversation and move "
"to the verdict."
),
"strict": True,
"parameters": {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
},
},
},
]
def _run_decision_phase(
self,
*,
input: AgentInput,
effective_criteria: List[str],
view: _ConversationView,
fetch_remote_traces: bool,
) -> _DecisionOutcome:
"""Phase 1 of the two-phase judge: continue, or move to the verdict.
Returns a ``_DecisionOutcome`` whose decision is "continue",
"verdict", or "exhausted" (the discovery loop ran out of steps
without a decision; the collapsed discovery history rides along so
the forced verdict keeps what was gathered). Never fetches remote
traces and never produces a verdict; the span digest here holds only
what the local collector already has.
"""
spans = self._span_collector.get_spans_for_thread(input.thread_id)
digest, is_large_span_trace = self._build_trace_digest(spans)
is_large_trace = is_large_span_trace or view.is_large_transcript
messages: List[dict] = [
{
"role": "system",
"content": self._build_decision_system_prompt(
description=input.scenario_state.description,
criteria=effective_criteria,
fetch_remote_traces=fetch_remote_traces,
),
},
{
"role": "user",
"content": _render_judge_content(
transcript=view.transcript_for_prompt,
traces_digest=digest,
extra_context_section="",
),
},
]
tools = self._build_decision_tools()
if is_large_span_trace:
tools = self._build_progressive_discovery_tools() + tools
if view.is_large_transcript:
tools = self._build_transcript_discovery_tools() + tools
if not is_large_trace:
response = self._completion_with_reasoning_off_retry(
model=self.model,
messages=messages,
temperature=self.temperature,
api_key=self.api_key,
api_base=self.api_base,
max_tokens=self.max_tokens,
tools=tools,
tool_choice="required",
**self._extra_params,
)
return _DecisionOutcome(decision=self._parse_decision(response))
for _ in range(self._max_discovery_steps):
response = self._completion_with_reasoning_off_retry(
model=self.model,
messages=messages,
temperature=self.temperature,
api_key=self.api_key,
api_base=self.api_base,
max_tokens=self.max_tokens,
tools=tools,
tool_choice="required",
**self._extra_params,
)
if not hasattr(response, "choices") or len(response.choices) == 0:
raise Exception(
f"Unexpected response format from LLM: {response.__repr__()}"
)
message = cast(Choices, response.choices[0]).message
if not message.tool_calls:
raise Exception(
f"Invalid response from judge agent, tool calls not found: {message.__repr__()}"
)
terminal_call = next(
(
tc
for tc in message.tool_calls
if tc.function.name in _DECISION_TOOL_NAMES
),
None,
)
if terminal_call:
return _DecisionOutcome(
decision=(
"continue"
if terminal_call.function.name == "continue_test"
else "verdict"
)
)
messages.append({
"role": "assistant",
"content": message.content or "",
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in message.tool_calls
],
})
for tc in message.tool_calls:
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": self._execute_discovery_tool(
tc, spans, view.working_messages
),
})
logger.debug(
"decision discovery exhausted its steps without a decision - "
"forcing the verdict"
)
# Everything after the system and content messages is discovery
# cycles; collapsed to plain text they carry what the judge gathered
# into the forced verdict call.
return _DecisionOutcome(
decision="exhausted",
discovery_recap=_collapse_discovery_history(messages)[2:],
)
def _parse_decision(self, response: Any) -> str:
"""Maps a decision-phase LLM response to "continue" or "verdict"."""
if not hasattr(response, "choices") or len(response.choices) == 0:
raise Exception(
f"Unexpected response format from LLM: {response.__repr__()}"
)
message = cast(Choices, response.choices[0]).message
if not message.tool_calls:
raise Exception(
f"Invalid response from judge agent, tool calls not found: {message.__repr__()}"
)
terminal_call = next(
(
tc
for tc in message.tool_calls
if tc.function.name in _DECISION_TOOL_NAMES
),
None,
)
if terminal_call is None:
raise Exception(
f"Invalid tool call from judge agent: {message.tool_calls[0].function.name}"
)
return (
"continue"
if terminal_call.function.name == "continue_test"
else "verdict"
)
async def _run_judgment_phase(
self,
*,
input: AgentInput,
effective_criteria: List[str],
view: _ConversationView,
fetch_remote_traces: bool,
trace_wait_timeout: float,
trace_wait_extension: float,
is_last_message: bool,
verdict_forced: bool,
discovery_recap: Optional[List[dict]] = None,
wait_extension_used: bool = False,
) -> AgentReturnTypes:
"""Phase 2 of the two-phase judge: the verdict itself.
Settle-waits for the remote traces first when fetching is on (the
only fetch site), so the digest always holds the full evidence, then
makes one finish_test-pinned evaluation. When the traces are still
incomplete after the settle-wait, the call also offers a one-shot
``wait_for_traces`` tool: calling it settle-waits once more under
``trace_wait_extension`` and re-enters the phase with the tool
withdrawn (``wait_extension_used``), so the second call must decide.
``verdict_forced`` reflects the entry mode: a required judgment
(last turn, explicit judgment_request) or decision-discovery
exhaustion makes an inconclusive verdict terminal; a voluntary
make_verdict entry lets an inconclusive verdict continue the
conversation (#886), unless not one remote trace of the run ever
settled, in which case more turns cannot improve the evidence and
the verdict stands.
A non-empty ``discovery_recap`` marks the exhaustion entry: the
decision loop already spent the discovery budget, so its collapsed
cycles are replayed as context and the verdict is one pinned call
with no further discovery.
"""
remote_trace_ids: List[str] = (
_distinct_message_trace_ids(input.messages) if fetch_remote_traces else []
)
all_settled = True
if fetch_remote_traces and remote_trace_ids:
all_settled = await self._remote_trace_fetcher.settle_traces(
thread_id=input.thread_id,
trace_ids=remote_trace_ids,
collector=self._span_collector,
timeout=trace_wait_timeout,
)
elif fetch_remote_traces:
# Fetching is on and there is nothing to fetch. Without this the
# traces section is silently empty and the judge marks internal
# criteria inconclusive without a stated reason.
logger.warning(
"Remote trace fetching is on but no message carries a trace "
"id; nothing to fetch"
)
self._remote_trace_fetcher.record_missing_trace_ids(
thread_id=input.thread_id,
collector=self._span_collector,
)
# When not one remote trace of the run ever settled, more turns
# cannot produce trace evidence: a voluntary inconclusive verdict
# would loop (verdict, continue, settle, inconclusive again) all the
# way to the turn cap, paying the settle budget every turn. The
# verdict becomes terminal instead; with any settled trace, #886
# semantics stay.
evidence_exhausted = (
fetch_remote_traces
and bool(remote_trace_ids)
and self._remote_trace_fetcher.none_settled(
thread_id=input.thread_id, trace_ids=remote_trace_ids
)
)
verdict_is_terminal = verdict_forced or evidence_exhausted
# The judge's one extra wait: offered as a wait_for_traces tool while
# the traces are incomplete, consumed at most once, then withdrawn so
# the second call must decide. With no trace ids at all there is
# nothing a wait could produce, so the tool is never offered.
wait_extension_available = (
fetch_remote_traces
and bool(remote_trace_ids)
and not all_settled
and trace_wait_extension > 0
and not wait_extension_used
)
spans = self._span_collector.get_spans_for_thread(input.thread_id)
digest, is_large_span_trace = self._build_trace_digest(spans)
is_large_trace = is_large_span_trace or view.is_large_transcript
logger.debug(f"OpenTelemetry traces built: {digest[:200]}...")
content_for_judge = _render_judge_content(
transcript=view.transcript_for_prompt,
traces_digest=digest,
extra_context_section=view.extra_context_section,
)
criteria_str = "\n".join(
[f"{idx + 1}. {criterion}" for idx, criterion in enumerate(effective_criteria)]
)
remote_traces_rule = (
f"\n- {REMOTE_TRACES_JUDGE_RULE}" if fetch_remote_traces else ""
)
system_content = self.system_prompt or f"""
<role>
You are an LLM as a judge delivering the final verdict on a simulated conversation, determining if the agent under test meets the criteria or not.
</role>
<goal>
Your goal is to deliver the final verdict of the scenario below with the finish_test tool, evaluating each criterion independently against the conversation and the collected evidence.
</goal>
<scenario>
{input.scenario_state.description}
</scenario>
<criteria>
{criteria_str}
</criteria>
<rules>
- Be strict: a criterion passes only when the conversation or the collected evidence clearly shows it was met.
- DO NOT make any judgment calls that are not explicitly listed in the success or failure criteria, withhold judgement if necessary
- When the evidence for a criterion is not definitive, mark that criterion inconclusive rather than guessing; an inconclusive verdict is acceptable{remote_traces_rule}
</rules>
"""
if self.system_prompt and fetch_remote_traces:
system_content = self.system_prompt + "\n\n" + REMOTE_TRACES_JUDGE_RULE
messages: List[dict] = [
{"role": "system", "content": system_content},
{"role": "user", "content": content_for_judge},
]
if is_last_message:
messages.append(
{
"role": "user",
"content": """
System:
<finish_test>
This is the last message, conversation has reached the maximum number of turns, give your final verdict,
if you don't have enough information to make a verdict, say inconclusive with max turns reached.
</finish_test>
""",
}
)
if wait_extension_used:
messages.append(
{
"role": "user",
"content": (
"You already waited once more for the remote traces. "
"The trace evidence above is final: deliver your "
"verdict now."
),
}
)
criteria_names = _criteria_keys(effective_criteria)
tools: List[dict] = [
{
"type": "function",
"function": {
"name": "finish_test",
"description": "Complete the test with a final verdict",
"strict": True,
"parameters": {
"type": "object",
"properties": {
"criteria": {
"type": "object",
"properties": {
criteria_names[idx]: {
"type": "string",
"enum": ["true", "false", "inconclusive"],
"description": criterion,
}
for idx, criterion in enumerate(effective_criteria)
},
"required": criteria_names,
"additionalProperties": False,
"description": "Strict verdict for each criterion",
},
"reasoning": {
"type": "string",
"description": "Explanation of what the final verdict should be",
},
"verdict": {
"type": "string",
"enum": ["success", "failure", "inconclusive"],
"description": "The final verdict of the test",
},
},
"required": ["criteria", "reasoning", "verdict"],
"additionalProperties": False,
},
},
},
]
exhausted_entry = bool(discovery_recap)
if not exhausted_entry:
if is_large_span_trace:
tools = self._build_progressive_discovery_tools() + tools
if view.is_large_transcript:
tools = self._build_transcript_discovery_tools() + tools
if wait_extension_available:
tools = [_build_wait_for_traces_tool()] + tools
# finish_test is the only terminal tool of the verdict phase and the
# tool choice pins it: continuing is the decision phase's business.
# While the traces are incomplete and the extension is unused, the
# wait_for_traces tool joins the set and the pin relaxes to
# "required" so the judge can pick either. The large-trace discovery
# loop relaxes the pin to "required" on its intermediate steps so
# the judge can use discovery tools, and forces the verdict on
# exhaustion.
tool_choice: Any = (
"required"
if wait_extension_available
else {"type": "function", "function": {"name": "finish_test"}}
)
if exhausted_entry:
assert discovery_recap is not None
messages.extend(discovery_recap)
messages.append({
"role": "user",
"content": (
"You have reached the maximum number of trace exploration steps. "
"Based on the information you have gathered so far, give your final verdict now."
),
})
elif is_large_trace:
outcome = self._run_discovery_loop(
messages=messages,
tools=tools,
tool_choice=tool_choice,
spans=spans,
working_messages=view.working_messages,
effective_criteria=effective_criteria,
input_messages=input.messages,
verdict_forced=verdict_is_terminal,
wait_tool_offered=wait_extension_available,
)
if outcome is _WAIT_FOR_TRACES_REQUESTED:
return await self._extend_wait_and_rejudge(
input=input,
effective_criteria=effective_criteria,
view=view,
fetch_remote_traces=fetch_remote_traces,
trace_wait_timeout=trace_wait_timeout,
trace_wait_extension=trace_wait_extension,
is_last_message=is_last_message,
verdict_forced=verdict_forced,
discovery_recap=discovery_recap,
remote_trace_ids=remote_trace_ids,
)
return cast(AgentReturnTypes, outcome)
response = self._completion_with_reasoning_off_retry(
model=self.model,
messages=messages,
temperature=self.temperature,
api_key=self.api_key,
api_base=self.api_base,
max_tokens=self.max_tokens,
tools=tools,
tool_choice=tool_choice,
**self._extra_params,
)
if wait_extension_available and _response_called_wait_for_traces(response):
return await self._extend_wait_and_rejudge(
input=input,
effective_criteria=effective_criteria,
view=view,
fetch_remote_traces=fetch_remote_traces,
trace_wait_timeout=trace_wait_timeout,
trace_wait_extension=trace_wait_extension,
is_last_message=is_last_message,
verdict_forced=verdict_forced,
discovery_recap=discovery_recap,
remote_trace_ids=remote_trace_ids,
)
return self._parse_response(
response,
effective_criteria,
messages,
input_messages=input.messages,
verdict_forced=verdict_is_terminal,
)
async def _extend_wait_and_rejudge(
self,
*,
input: AgentInput,
effective_criteria: List[str],
view: _ConversationView,
fetch_remote_traces: bool,
trace_wait_timeout: float,
trace_wait_extension: float,
is_last_message: bool,
verdict_forced: bool,
discovery_recap: Optional[List[dict]],
remote_trace_ids: List[str],
) -> AgentReturnTypes:
"""Runs the judge's one extra wait, then re-enters the verdict.
Re-arms the failed traces, settle-waits once more under the
extension budget, and re-enters the judgment phase with the
wait_for_traces tool withdrawn, so the second call must decide.
"""
logger.debug(
"Judge requested one more wait for the remote traces (%.0fs)",
trace_wait_extension,
)
await self._remote_trace_fetcher.extend_settle(
thread_id=input.thread_id,
trace_ids=remote_trace_ids,
collector=self._span_collector,
timeout=trace_wait_extension,
)
return await self._run_judgment_phase(
input=input,
effective_criteria=effective_criteria,
view=view,
fetch_remote_traces=fetch_remote_traces,
trace_wait_timeout=trace_wait_timeout,
trace_wait_extension=trace_wait_extension,
is_last_message=is_last_message,
verdict_forced=verdict_forced,
discovery_recap=discovery_recap,
wait_extension_used=True,
)
def _remote_trace_settings(self, input: AgentInput) -> "tuple[bool, float, float]":
"""Resolves the remote trace fetching configuration for this call.
Reads ``fetch_remote_traces`` (effective default False),
``trace_wait_timeout`` (effective default 30 seconds) and
``trace_wait_extension`` (effective default: the resolved timeout)
from the scenario configuration.
"""
config = getattr(input.scenario_state, "config", None)
enabled = getattr(config, "fetch_remote_traces", None) is True
timeout = getattr(config, "trace_wait_timeout", None)
if (
not isinstance(timeout, (int, float))
or isinstance(timeout, bool)
or timeout <= 0
):
timeout = DEFAULT_TRACE_WAIT_TIMEOUT_SECONDS
# The one extra wait the judge may request via the wait_for_traces
# tool. Defaults to the wait budget itself; the platform passes its
# upper cap here so a short measured budget still gets a meaningful
# extension.
extension = getattr(config, "trace_wait_extension", None)
if (
not isinstance(extension, (int, float))
or isinstance(extension, bool)
or extension <= 0
):
extension = timeout
return enabled, float(timeout), float(extension)
def _completion_with_reasoning_off_retry(self, **kwargs: Any) -> ModelResponse:
"""
``litellm.completion``, retried once with reasoning declared off when —
and only when — the provider rejected a tool-carrying call for exactly
that reason. A caller that already asked for a specific effort keeps it
and gets the endpoint's own error, rather than having its intent
silently rewritten.
"""
try:
return cast(ModelResponse, litellm.completion(**kwargs))
except Exception as error:
if not kwargs.get("tools") or "reasoning_effort" in kwargs:
raise
if not _rejection_asks_for_reasoning_off(error):
raise
logger.debug(
"provider rejected function tools without reasoning off for %s; retrying",
kwargs.get("model"),
)
return cast(
ModelResponse,
litellm.completion(**kwargs, reasoning_effort=_REASONING_OFF),
)
def _build_trace_digest(self, spans: Sequence[Any]) -> tuple[str, bool]:
"""
Builds the trace digest, choosing between full inline rendering
and structure-only mode based on estimated token count.
Args:
spans: The spans for this thread.
Returns:
Tuple of (digest_string, is_large_trace).
"""
full_digest = judge_span_digest_formatter.format(spans)
is_large_trace = (
len(spans) > 0 and estimate_tokens(full_digest) > self._token_threshold
)
if is_large_trace:
digest = (
judge_span_digest_formatter.format_structure_only(spans)
+ "\n\nUse expand_trace(span_id) to see span details or grep_trace(pattern) to search across spans. Reference spans by the ID shown in brackets."
)
else:
digest = full_digest
logger.debug(
"Trace digest built",
extra={
"is_large_trace": is_large_trace,
"estimated_tokens": estimate_tokens(full_digest),
},
)
return digest, is_large_trace
def _build_progressive_discovery_tools(self) -> List[dict]:
"""
Builds the expand_trace and grep_trace tool definitions for litellm.
Returns:
List of tool definition dicts for litellm function calling.
"""
return [
{
"type": "function",
"function": {
"name": "expand_trace",
"description": (
"Expand one or more spans to see their full details "
"(attributes, events, content). Use the span ID shown "
"in brackets in the trace skeleton."
),
"parameters": {
"type": "object",
"properties": {
"span_ids": {
"type": "array",
"items": {"type": "string"},
"description": "Span IDs (or 8-char prefixes) to expand",
},
},
"required": ["span_ids"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "grep_trace",
"description": (
"Search across all span attributes, events, and content "
"for a pattern (case-insensitive). Returns matching spans with context."
),
"parameters": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Search pattern (case-insensitive)",
},
},
"required": ["pattern"],
"additionalProperties": False,
},
},
},
]
def _build_transcript_discovery_tools(self) -> List[dict]:
"""
Builds the expand_transcript and grep_transcript tool definitions for
litellm. Parallel to ``_build_progressive_discovery_tools``, but for
message-transcript discovery instead of span discovery (see
``transcript_tools.py`` for why the two need to be independent).
Returns:
List of tool definition dicts for litellm function calling.
"""
return [
{
"type": "function",
"function": {
"name": "expand_transcript",
"description": (
"Expand one or more messages to see their full content. "
"Use the message index shown in brackets in the transcript "
"skeleton."
),
"parameters": {
"type": "object",
"properties": {
"indices": {
"type": "array",
"items": {"type": "integer"},
"description": "0-based message indices to expand",
},
},
"required": ["indices"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "grep_transcript",
"description": (
"Search across all message content for a pattern "
"(case-insensitive). Returns matching messages with context."
),
"parameters": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Search pattern (case-insensitive)",
},
},
"required": ["pattern"],
"additionalProperties": False,
},
},
},
]
def _run_discovery_loop(
self,
*,
messages: List[dict],
tools: List[dict],
tool_choice: Any,
spans: Sequence[Any],
working_messages: Sequence[ChatCompletionMessageParam],
effective_criteria: List[str],
input_messages: Sequence[Any],
verdict_forced: bool,
wait_tool_offered: bool = False,
) -> Union[AgentReturnTypes, _WaitForTracesRequested]:
"""
Runs the multi-step discovery loop of the verdict phase for large
traces.
The judge can call expand_trace/grep_trace (spans) and/or
expand_transcript/grep_transcript (messages) tools multiple times
before calling finish_test, the only terminal tool of the verdict
phase, or hitting the max discovery steps limit, which forces the
verdict with whatever was gathered.
On intermediate steps, tool_choice is "required" so the judge can
freely pick a discovery tool. On the final step, the pinned
finish_test tool_choice is applied.
Args:
messages: The conversation messages so far.
tools: The tool definitions.
tool_choice: The tool choice constraint for the final step.
spans: The spans for executing expand_trace/grep_trace.
working_messages: The conversation messages for executing
expand_transcript/grep_transcript.
effective_criteria: The criteria to judge against.
Returns:
AgentReturnTypes from the finish_test call.
"""
for step in range(self._max_discovery_steps):
# Use "required" for intermediate steps so the judge can use
# discovery tools; only apply the forced tool_choice on the
# last allowed step.
is_last_step = step == self._max_discovery_steps - 1
step_tool_choice = tool_choice if is_last_step else "required"
response = self._completion_with_reasoning_off_retry(
model=self.model,
messages=messages,
temperature=self.temperature,
api_key=self.api_key,
api_base=self.api_base,
max_tokens=self.max_tokens,
tools=tools,
tool_choice=step_tool_choice,
**self._extra_params,
)
if not hasattr(response, "choices") or len(response.choices) == 0:
raise Exception(
f"Unexpected response format from LLM: {response.__repr__()}"
)
message = cast(Choices, response.choices[0]).message
if not message.tool_calls:
# No tool calls - try to parse as a response
return self._parse_response(
response,
effective_criteria,
messages,
input_messages=input_messages,
verdict_forced=verdict_forced,
)
if wait_tool_offered and any(
tc.function.name == "wait_for_traces" for tc in message.tool_calls
):
# The extra wait rebuilds the whole phase: the caller
# settle-waits once more and re-enters with a fresh digest
# and the tool withdrawn.
return _WAIT_FOR_TRACES_REQUESTED
terminal_call = next(
(tc for tc in message.tool_calls if tc.function.name == "finish_test"),
None,
)
if terminal_call:
return self._parse_response(
response,
effective_criteria,
messages,
input_messages=input_messages,
verdict_forced=verdict_forced,
)
# Execute discovery tools and add results to messages
# Add the assistant message with tool calls
messages.append({
"role": "assistant",
"content": message.content or "",
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in message.tool_calls
],
})
for tc in message.tool_calls:
tool_result = self._execute_discovery_tool(tc, spans, working_messages)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": tool_result,
})
return self._force_verdict(
messages=messages,
tools=tools,
effective_criteria=effective_criteria,
input_messages=input_messages,
)
def _force_verdict(
self,
*,
messages: List[dict],
tools: List[dict],
effective_criteria: List[str],
input_messages: Sequence[Any],
) -> AgentReturnTypes:
"""
Makes one final LLM call with tool_choice forced to finish_test.
Hardening (vs. a naive re-invocation with the same tool set):
- Prior discovery tool_call/tool_result pairs are rewritten in the
message history as plain-text assistant recaps. This lets us
drop expand_trace/grep_trace from the tool set without
Anthropic rejecting the call for referencing undefined tools.
- Discovery tools are then stripped so the model physically
cannot emit them, closing the leak path where tool_choice
wasn't honored and a discovery tool reached _parse_response.
"""
logger.warning(
f"Progressive discovery hit max steps ({self._max_discovery_steps}), "
"forcing verdict"
)
rewritten_messages = _collapse_discovery_history(messages)
rewritten_messages.append({
"role": "user",
"content": (
"You have reached the maximum number of trace exploration steps. "
"Based on the information you have gathered so far, give your final verdict now."
),
})
# finish_test only, not just "everything except discovery". The
# verdict phase also offers wait_for_traces while the extension is
# unused, and it would survive a deny-list. The pin below asks for
# finish_test, but a model that ignores the pin and calls
# wait_for_traces here reaches _parse_response as an invalid tool
# call. Leaving one tool closes that path.
finish_only_tools = [
t for t in tools
if t.get("function", {}).get("name") == "finish_test"
]
forced_response = self._completion_with_reasoning_off_retry(
model=self.model,
messages=rewritten_messages,
temperature=self.temperature,
api_key=self.api_key,
api_base=self.api_base,
max_tokens=self.max_tokens,
tools=finish_only_tools,
tool_choice={"type": "function", "function": {"name": "finish_test"}},
**self._extra_params,
)
return self._parse_response(
forced_response,
effective_criteria,
rewritten_messages,
input_messages=input_messages,
# The whole point of this call is a pinned finish_test — the model
# has no continue escape, so an inconclusive verdict is legitimate.
verdict_forced=True,
)
def _execute_discovery_tool(
self,
tool_call: Any,
spans: Sequence[Any],
working_messages: Sequence[ChatCompletionMessageParam],
) -> str:
"""
Executes an expand_trace, grep_trace, expand_transcript, or
grep_transcript tool call.
Args:
tool_call: The tool call from the LLM response.
spans: The spans to operate on for expand_trace/grep_trace.
working_messages: The conversation messages to operate on for
expand_transcript/grep_transcript.
Returns:
The tool result string.
"""
try:
args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
return f"Error: could not parse arguments: {tool_call.function.arguments}"
if tool_call.function.name == "expand_trace":
return expand_trace(
spans,
span_ids=args.get("span_ids", []),
)
elif tool_call.function.name == "grep_trace":
return grep_trace(spans, args.get("pattern", ""))
elif tool_call.function.name == "expand_transcript":
return expand_transcript(
working_messages,
indices=args.get("indices", []),
)
elif tool_call.function.name == "grep_transcript":
return grep_transcript(working_messages, args.get("pattern", ""))
else:
return f"Unknown tool: {tool_call.function.name}"
def _parse_response(
self,
response: Any,
effective_criteria: List[str],
messages: List[dict],
*,
input_messages: Sequence[Any],
verdict_forced: bool,
) -> AgentReturnTypes:
"""
Parses a litellm response into the appropriate return type.
Handles finish_test, continue_test, and error cases.
Args:
response: The litellm ModelResponse.
effective_criteria: The criteria to evaluate against.
messages: The judge's internal LLM messages (system prompt + transcript).
input_messages: The actual conversation messages to include in ScenarioResult.
Returns:
AgentReturnTypes: Either an empty list (continue) or ScenarioResult.
"""
if not hasattr(response, "choices") or len(response.choices) == 0:
raise Exception(
f"Unexpected response format from LLM: {response.__repr__()}"
)
message = cast(Choices, response.choices[0]).message
if not message.tool_calls:
raise Exception(
f"Invalid response from judge agent, tool calls not found: {message.__repr__()}"
)
# In multi-step mode, find the terminal tool call
terminal_call = next(
(tc for tc in message.tool_calls if tc.function.name == "finish_test"),
None,
)
tool_call = terminal_call or message.tool_calls[0]
if tool_call.function.name == "finish_test":
try:
args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
raise Exception(
f"Failed to parse tool call arguments from judge agent: {tool_call.function.arguments}"
)
verdict = args.get("verdict", "inconclusive")
reasoning = args.get("reasoning", "No reasoning provided")
# "Can't tell yet" is not a verdict (#886). When nothing forced the
# judge to finish — continue_test was freely available — an
# inconclusive finish_test used to end the run as a failure, which
# on a platform surface reads as the simulated user going silent
# mid-conversation. Treat it as continue_test and let the
# conversation play out; a FORCED judgment (last turn, an explicit
# judgment_request, discovery exhaustion) keeps its terminal
# behavior unchanged.
if not verdict_forced and verdict == "inconclusive":
logger.debug(
"finish_test returned an inconclusive verdict without a "
"forced judgment - continuing the conversation"
)
return []
criteria_verdicts = args.get("criteria", {})
# LLMs sometimes serialise the criteria object as a JSON *string*
# instead of an inline dict, especially with complex dynamic
# schemas (issue #161). Re-parse one level if that happens.
if isinstance(criteria_verdicts, str):
try:
criteria_verdicts = json.loads(criteria_verdicts)
except (json.JSONDecodeError, ValueError):
criteria_verdicts = None # unparseable — handled below
# If the criteria payload is not a usable object, we cannot trust
# any per-criterion verdict. Do NOT fall back to {}: an empty dict
# makes failed_criteria empty and lets a "success" verdict slip
# through having evaluated ZERO criteria, masking the real problem
# (issue #161 follow-up). Surface it as an explicit, fail-closed
# result instead of swallowing it.
if not isinstance(criteria_verdicts, dict):
raw = args.get("criteria")
logger.warning(
"JudgeAgent could not resolve criteria verdicts to an "
"object (got %s); failing the judgment instead of "
"reporting an unverified success.",
type(raw).__name__,
)
return ScenarioResult(
success=False,
messages=cast(Any, input_messages),
reasoning=(
"JudgeAgent could not parse the per-criterion verdicts "
"returned by the LLM, so the judgment could not be "
f"verified (raw criteria value was of type "
f"{type(raw).__name__}). Original verdict was "
f"{verdict!r}. Treating the judgment as failed."
),
passed_criteria=[],
failed_criteria=list(effective_criteria),
)
# Map each verdict back to its criterion BY the schema key we
# generated for it. Positional .values() mapping silently
# mislabels partial / reordered payloads and IndexErrors on extra
# keys; key-based lookup is robust. A criterion passes ONLY on an
# explicit "true"; anything else (false, inconclusive, missing, or
# a nested/unexpected value) is a failure, so an unevaluated
# criterion can never slip through as success.
# Map each verdict to its criterion by the schema key we generated
# for it (single source of truth: _criteria_keys). Positional
# .values() mapping silently mislabels partial / reordered / nested
# payloads and IndexErrors on extra keys; key lookup is robust.
# A criterion passes ONLY on an explicit "true"; anything else
# (false, inconclusive, missing, or a nested/unexpected value) is a
# failure, so an unevaluated criterion can never slip through as
# success. JSON booleans are coerced — some LLMs emit `true`/`false`
# instead of the enum strings.
criteria_keys = _criteria_keys(effective_criteria)
passed_criteria: List[str] = []
failed_criteria: List[str] = []
for criterion, key in zip(effective_criteria, criteria_keys):
raw_verdict = criteria_verdicts.get(key)
if isinstance(raw_verdict, bool):
raw_verdict = "true" if raw_verdict else "false"
bucket = passed_criteria if raw_verdict == "true" else failed_criteria
bucket.append(criterion)
return ScenarioResult(
success=verdict == "success" and len(failed_criteria) == 0,
messages=cast(Any, input_messages),
reasoning=reasoning,
passed_criteria=passed_criteria,
failed_criteria=failed_criteria,
)
if tool_call.function.name in _DISCOVERY_TOOL_NAMES:
logger.warning(
f"Discovery tool {tool_call.function.name} leaked past "
"discovery loop without reaching a terminal verdict"
)
return ScenarioResult(
success=False,
messages=cast(Any, input_messages),
reasoning=(
"JudgeAgent: trace discovery did not converge on a "
"verdict within the step budget"
),
passed_criteria=[],
failed_criteria=list(effective_criteria),
)
raise Exception(
f"Invalid tool call from judge agent: {tool_call.function.name}"
)
# --------------------------------------------------------------------- #
# Transcript-enrichment helper — module-level to keep JudgeAgent clean #
# --------------------------------------------------------------------- #
def _enrich_messages_with_transcripts(
messages: List[Any],
recording: Any,
) -> List[Any]:
"""
Add a transcript text part to each audio-only message (both user AND
agent), preserving the audio part so the judge still sees ``input_audio``
evidence in the message.
Why we don't REPLACE: criteria like "agent and user exchanged real audio
turns" need the audio block visible to the judge as proof the message
carried bytes, not just text. Replacing the content (the previous
behavior) made the message look text-only and the judge correctly
concluded "the assistant's turns are text-only" — which then failed
audio-presence criteria.
Strategy: insert a ``{"type": "text", "text": <transcript>}`` part at the
front of the content list, leaving the input_audio part in place.
``_truncate_base64_media`` later collapses the base64 to a placeholder
so token cost stays bounded; what survives is the **shape** evidence
(the audio block) plus the readable transcript.
Returns a new list — does not mutate input.
Matching strategy: per-role ordinal — the Nth assistant audio-only
message maps to the Nth agent segment, and the Nth user audio-only
message maps to the Nth user segment (each in temporal order). This
role-scoped matching is important when scenarios interleave turns:
user/agent counts don't have to match.
If a segment has no transcript (STT failed / unavailable), the
corresponding message is left as-is so evaluation degrades gracefully.
"""
# Gather transcripts per-role in temporal order. Both rails are
# transcribed by the same provider via ``transcribe_segments`` upstream;
# this loop just routes the resulting text into the matching message.
segments = getattr(recording, "segments", []) or []
sorted_segments = sorted(segments, key=lambda s: s.start_time)
agent_transcripts = [
s.transcript
for s in sorted_segments
if getattr(s, "speaker", None) == "agent" and s.transcript
]
user_transcripts = [
s.transcript
for s in sorted_segments
if getattr(s, "speaker", None) == "user" and s.transcript
]
enriched: List[Any] = []
agent_msg_idx = 0 # ordinal counter over assistant audio-only messages
user_msg_idx = 0 # ordinal counter over user audio-only messages
for msg in messages:
role = msg.get("role") if isinstance(msg, dict) else None
if role not in ("assistant", "user"):
enriched.append(msg)
continue
content = msg.get("content") if isinstance(msg, dict) else None
if not isinstance(content, list):
enriched.append(msg)
continue
# Check: does this message have audio but no text?
has_audio = any(
isinstance(p, dict) and p.get("type") in ("input_audio", "audio")
for p in content
)
has_text = any(
isinstance(p, dict) and p.get("type") == "text"
for p in content
)
if has_audio and not has_text:
if role == "assistant" and agent_msg_idx < len(agent_transcripts):
transcript_text = agent_transcripts[agent_msg_idx]
agent_msg_idx += 1
enriched.append({
**msg,
"content": [{"type": "text", "text": transcript_text}, *content],
})
continue
if role == "user" and user_msg_idx < len(user_transcripts):
transcript_text = user_transcripts[user_msg_idx]
user_msg_idx += 1
enriched.append({
**msg,
"content": [{"type": "text", "text": transcript_text}, *content],
})
continue
# No transcript available — consume the ordinal slot anyway so
# subsequent messages map to the right segment.
if role == "assistant":
agent_msg_idx += 1
else:
user_msg_idx += 1
enriched.append(msg)
return enrichedGlobal variables
var DECISION_PHASE_RULE-
Appended to a custom system prompt on decision calls, so custom judge personas still drive the argument-free decision tools correctly.
var REMOTE_TRACES_DECISION_RULE-
Appended to the decision system prompt when fetch_remote_traces is on.
var REMOTE_TRACES_JUDGE_RULE-
Rule appended to the verdict system prompt when fetch_remote_traces is on.
Classes
class JudgeAgent (*, criteria: List[str] | None = None, model: str | None = None, api_base: str | None = None, api_key: str | None = None, temperature: float = 0.0, max_tokens: int | None = None, system_prompt: str | None = None, span_collector: scenario._tracing.judge_span_collector.JudgeSpanCollector | None = None, remote_trace_fetcher: scenario._tracing.remote_trace_fetcher.RemoteTraceFetcher | None = None, token_threshold: int = 8192, max_discovery_steps: int = 10, include_audio: bool | None = None, include_timeline: bool | None = None, include_traces: bool | None = None, modality: str | None = None, **extra_params)-
Agent that evaluates conversations against success criteria.
The JudgeAgent watches conversations in real-time and makes decisions about whether the agent under test is meeting the specified criteria. It can either allow the conversation to continue or end it with a success/failure verdict.
The judge uses function calling to make structured decisions and provides detailed reasoning for its verdicts. It evaluates each criterion independently and provides comprehensive feedback about what worked and what didn't.
Attributes
role- Always AgentRole.JUDGE for judge agents
model- LLM model identifier to use for evaluation
api_base- Optional base URL where the model is hosted
api_key- Optional API key for the model provider
temperature- Sampling temperature for evaluation consistency
max_tokens- Maximum tokens for judge reasoning
criteria- List of success criteria to evaluate against
system_prompt- Custom system prompt to override default judge behavior
Example
import scenario # Basic judge agent with criteria judge = scenario.JudgeAgent( criteria=[ "Agent provides helpful responses", "Agent asks relevant follow-up questions", "Agent does not provide harmful information" ] ) # Customized judge with specific model and behavior strict_judge = scenario.JudgeAgent( model="openai/gpt-4.1-mini", criteria=[ "Code examples are syntactically correct", "Explanations are technically accurate", "Security best practices are mentioned" ], temperature=0.0, # More deterministic evaluation system_prompt="You are a strict technical reviewer evaluating code quality." ) # Use in scenario result = await scenario.run( name="coding assistant test", description="User asks for help with Python functions", agents=[ coding_agent, scenario.UserSimulatorAgent(), judge ] ) print(f"Passed criteria: {result.passed_criteria}") print(f"Failed criteria: {result.failed_criteria}")Note
- Judge agents evaluate conversations continuously, not just at the end
- They can end scenarios early if clear success/failure conditions are met
- Provide detailed reasoning for their decisions
- Support both positive criteria (things that should happen) and negative criteria (things that shouldn't)
Initialize a judge agent with evaluation criteria.
Args
criteria- List of success criteria to evaluate the conversation against. Can include both positive requirements ("Agent provides helpful responses") and negative constraints ("Agent should not provide personal information").
model- LLM model identifier (e.g., "openai/gpt-4.1-mini"). If not provided, uses the default model from global configuration.
api_base- Optional base URL where the model is hosted. If not provided, uses the base URL from global configuration.
api_key- API key for the model provider. If not provided, uses the key from global configuration or environment.
temperature- Sampling temperature for evaluation (0.0-1.0). Lower values (0.0-0.2) recommended for consistent evaluation.
max_tokens- Maximum number of tokens for judge reasoning and explanations.
system_prompt- Custom system prompt to override default judge behavior. Use this to create specialized evaluation perspectives.
span_collector- Optional span collector for telemetry. Defaults to global singleton.
remote_trace_fetcher- Optional fetcher for remote traces, used when
the scenario enables
fetch_remote_traces. Defaults to the global singleton. token_threshold- Estimated token count above which traces switch to structure-only rendering with progressive discovery tools. Defaults to 8192.
max_discovery_steps- Maximum number of expand/grep tool calls the judge can make before being forced to return a verdict. Defaults to 10.
modality- Explicit modality declaration for this role. Accepted values:
"audio-in"(LLM receives raw audio),"stt-bridge"(audio transcribed to text before the LLM), or"text"(no audio in the stack). Complementary toinclude_audio:include_audio=True/Falsetakes precedence;modality=applies wheninclude_audioisNone. WhenNone(default), the modality is auto-detected from litellm capabilities.
Raises
Exception- If no model is configured either in parameters or global config
Example
# Customer service judge cs_judge = JudgeAgent( criteria=[ "Agent replies with the refund policy", "Agent offers next steps for the customer", ], temperature=0.1 ) # Technical accuracy judge tech_judge = JudgeAgent( criteria=[ "Agent adds a code review pointing out the code compilation errors", "Agent adds a code review about the missing security headers" ], system_prompt="You are a senior software engineer reviewing code for production use." )Note
Advanced usage: Additional parameters can be passed as keyword arguments (e.g., headers, timeout, client) for specialized configurations. These are experimental and may not be supported in future versions.
Expand source code
class JudgeAgent(AgentAdapter): """ Agent that evaluates conversations against success criteria. The JudgeAgent watches conversations in real-time and makes decisions about whether the agent under test is meeting the specified criteria. It can either allow the conversation to continue or end it with a success/failure verdict. The judge uses function calling to make structured decisions and provides detailed reasoning for its verdicts. It evaluates each criterion independently and provides comprehensive feedback about what worked and what didn't. Attributes: role: Always AgentRole.JUDGE for judge agents model: LLM model identifier to use for evaluation api_base: Optional base URL where the model is hosted api_key: Optional API key for the model provider temperature: Sampling temperature for evaluation consistency max_tokens: Maximum tokens for judge reasoning criteria: List of success criteria to evaluate against system_prompt: Custom system prompt to override default judge behavior Example: ``` import scenario # Basic judge agent with criteria judge = scenario.JudgeAgent( criteria=[ "Agent provides helpful responses", "Agent asks relevant follow-up questions", "Agent does not provide harmful information" ] ) # Customized judge with specific model and behavior strict_judge = scenario.JudgeAgent( model="openai/gpt-4.1-mini", criteria=[ "Code examples are syntactically correct", "Explanations are technically accurate", "Security best practices are mentioned" ], temperature=0.0, # More deterministic evaluation system_prompt="You are a strict technical reviewer evaluating code quality." ) # Use in scenario result = await scenario.run( name="coding assistant test", description="User asks for help with Python functions", agents=[ coding_agent, scenario.UserSimulatorAgent(), judge ] ) print(f"Passed criteria: {result.passed_criteria}") print(f"Failed criteria: {result.failed_criteria}") ``` Note: - Judge agents evaluate conversations continuously, not just at the end - They can end scenarios early if clear success/failure conditions are met - Provide detailed reasoning for their decisions - Support both positive criteria (things that should happen) and negative criteria (things that shouldn't) """ role = AgentRole.JUDGE model: str api_base: Optional[str] api_key: Optional[str] temperature: float max_tokens: Optional[int] criteria: List[str] system_prompt: Optional[str] _extra_params: dict _span_collector: JudgeSpanCollector _remote_trace_fetcher: RemoteTraceFetcher _token_threshold: int _max_discovery_steps: int def __init__( self, *, criteria: Optional[List[str]] = None, model: Optional[str] = None, api_base: Optional[str] = None, api_key: Optional[str] = None, temperature: float = 0.0, max_tokens: Optional[int] = None, system_prompt: Optional[str] = None, span_collector: Optional[JudgeSpanCollector] = None, remote_trace_fetcher: Optional[RemoteTraceFetcher] = None, token_threshold: int = DEFAULT_TOKEN_THRESHOLD, max_discovery_steps: int = 10, include_audio: Optional[bool] = None, include_timeline: Optional[bool] = None, include_traces: Optional[bool] = None, modality: Optional[str] = None, **extra_params, ): """ Initialize a judge agent with evaluation criteria. Args: criteria: List of success criteria to evaluate the conversation against. Can include both positive requirements ("Agent provides helpful responses") and negative constraints ("Agent should not provide personal information"). model: LLM model identifier (e.g., "openai/gpt-4.1-mini"). If not provided, uses the default model from global configuration. api_base: Optional base URL where the model is hosted. If not provided, uses the base URL from global configuration. api_key: API key for the model provider. If not provided, uses the key from global configuration or environment. temperature: Sampling temperature for evaluation (0.0-1.0). Lower values (0.0-0.2) recommended for consistent evaluation. max_tokens: Maximum number of tokens for judge reasoning and explanations. system_prompt: Custom system prompt to override default judge behavior. Use this to create specialized evaluation perspectives. span_collector: Optional span collector for telemetry. Defaults to global singleton. remote_trace_fetcher: Optional fetcher for remote traces, used when the scenario enables ``fetch_remote_traces``. Defaults to the global singleton. token_threshold: Estimated token count above which traces switch to structure-only rendering with progressive discovery tools. Defaults to 8192. max_discovery_steps: Maximum number of expand/grep tool calls the judge can make before being forced to return a verdict. Defaults to 10. modality: Explicit modality declaration for this role. Accepted values: ``"audio-in"`` (LLM receives raw audio), ``"stt-bridge"`` (audio transcribed to text before the LLM), or ``"text"`` (no audio in the stack). Complementary to ``include_audio``: ``include_audio=True/False`` takes precedence; ``modality=`` applies when ``include_audio`` is ``None``. When ``None`` (default), the modality is auto-detected from litellm capabilities. Raises: Exception: If no model is configured either in parameters or global config Example: ``` # Customer service judge cs_judge = JudgeAgent( criteria=[ "Agent replies with the refund policy", "Agent offers next steps for the customer", ], temperature=0.1 ) # Technical accuracy judge tech_judge = JudgeAgent( criteria=[ "Agent adds a code review pointing out the code compilation errors", "Agent adds a code review about the missing security headers" ], system_prompt="You are a senior software engineer reviewing code for production use." ) ``` Note: Advanced usage: Additional parameters can be passed as keyword arguments (e.g., headers, timeout, client) for specialized configurations. These are experimental and may not be supported in future versions. """ self.criteria = criteria or [] self.api_base = api_base self.api_key = api_key self.temperature = temperature self.max_tokens = max_tokens self.system_prompt = system_prompt self._span_collector = span_collector or judge_span_collector self._remote_trace_fetcher = remote_trace_fetcher or default_remote_trace_fetcher self._token_threshold = token_threshold self._max_discovery_steps = max_discovery_steps # Voice-aware judge behaviour (§4.3). None = auto-detect based on # conversation content and judge model capabilities. self.include_audio = include_audio self.include_timeline = include_timeline self.include_traces = include_traces self.modality = modality if model: self.model = model if ScenarioConfig.default_config is not None and isinstance( ScenarioConfig.default_config.default_model, str ): self.model = model or ScenarioConfig.default_config.default_model self._extra_params = extra_params elif ScenarioConfig.default_config is not None and isinstance( ScenarioConfig.default_config.default_model, ModelConfig ): self.model = model or ScenarioConfig.default_config.default_model.model self.api_base = ( api_base or ScenarioConfig.default_config.default_model.api_base ) self.api_key = ( api_key or ScenarioConfig.default_config.default_model.api_key ) self.temperature = ( temperature or ScenarioConfig.default_config.default_model.temperature ) self.max_tokens = ( max_tokens or ScenarioConfig.default_config.default_model.max_tokens ) # Extract extra params from ModelConfig config_dict = ScenarioConfig.default_config.default_model.model_dump( exclude_none=True ) config_dict.pop("model", None) config_dict.pop("api_base", None) config_dict.pop("api_key", None) config_dict.pop("temperature", None) config_dict.pop("max_tokens", None) # Merge: config extras < agent extra_params self._extra_params = {**config_dict, **extra_params} else: self._extra_params = extra_params if not hasattr(self, "model"): raise Exception(agent_not_configured_error_message("JudgeAgent")) # --------------------------------------------- voice auto-detection (§4.3) def effective_include_audio(self, conversation_has_audio: bool) -> bool: """Resolve include_audio: explicit wins, otherwise use modality resolver. Intentional behavior change (Bundle 3 / AC3b): Before: gpt-4o → audio-capable (substring match). After: gpt-4o → text path (litellm advisory returns False). Before: gpt-audio-mini → NOT audio-capable (not in list). After: gpt-audio-mini → audio-capable (litellm advisory returns True). The old substring list was wrong; the resolver is the source of truth. """ if self.include_audio is not None: # Explicit override always wins (AC3c) return self.include_audio and conversation_has_audio # Use resolver with per-role declaration (AC0, Bundle 6) tier, warnings = resolve_modality(declaration=self.modality, model_id=self.model or "") for w in warnings: logger.warning(w) return conversation_has_audio and (tier == ModalityTier.AUDIO_IN) def effective_include_timeline(self, conversation_has_audio: bool) -> bool: """Default timeline True for voice, False for text — unless explicitly set.""" if self.include_timeline is not None: return self.include_timeline return conversation_has_audio def effective_include_traces(self, otel_configured: bool) -> bool: if self.include_traces is not None: return self.include_traces return otel_configured # --------------------------------- audio-transcription fallback helpers @staticmethod def _conversation_has_audio(messages: List[Any]) -> bool: """Return True if any message content contains an audio part.""" for msg in messages: content = msg.get("content") if isinstance(msg, dict) else None if isinstance(content, list): for part in content: if isinstance(part, dict) and part.get("type") in ("input_audio", "audio"): return True return False @staticmethod def _extract_recording(input: AgentInput) -> Any: """Return the VoiceRecording from the executor, or None.""" scenario_state = getattr(input, "scenario_state", None) if scenario_state is None: return None executor = getattr(scenario_state, "_executor", None) if executor is None: return None return getattr(executor, "_voice_recording", None) @scenario_cache() async def call( self, input: AgentInput, ) -> AgentReturnTypes: """ Evaluate the current conversation state against the configured criteria. The judge runs in two phases. A mid-conversation call first makes an argument-free decision between continuing the conversation and moving to the verdict (continue_test / make_verdict); only a make_verdict decision triggers the verdict call, which settle-waits for remote traces when enabled and evaluates every criterion with finish_test. The last turn and an explicit judgment request skip the decision and go straight to the verdict. Args: input: AgentInput containing conversation history and scenario context Returns: AgentReturnTypes: Either an empty list (continue scenario) or a ScenarioResult (end scenario with verdict) Raises: Exception: If the judge cannot make a valid decision or if there's an error in the evaluation process Note: - Returns empty list [] to continue the scenario - Returns ScenarioResult to end with success/failure - Provides detailed reasoning for all decisions - Evaluates each criterion independently - Can end scenarios early if clear violation or success is detected """ effective_criteria = ( input.judgment_request.criteria if input.judgment_request and input.judgment_request.criteria is not None else self.criteria ) max_turns = input.scenario_state.config.max_turns or 10 is_last_message = ( input.scenario_state.current_turn >= max_turns - 1 ) enforce_judgment = input.judgment_request is not None has_criteria = len(effective_criteria) > 0 if enforce_judgment and not has_criteria: return ScenarioResult( success=False, messages=[], reasoning="TestingAgent was called as a judge, but it has no criteria to judge against", ) # A judgment is required when the conversation cannot continue past # this call: the last turn, or an explicit judgment_request. Both go # straight to the verdict phase; only an unforced mid-conversation # call runs the decision phase first. judgment_required = is_last_message or enforce_judgment # min_turns floor (ADR-005): below the floor the decision is # predetermined (the conversation must continue), so nothing is spent # on it. The check runs before the conversation view is built, or a # gated voice turn pays for a transcription it discards. The judge # observes a 0-based current_turn: reset() overrides the initial # _new_turn() back to 0, so the call on turn N sees current_turn N-1. # The floor is unmet while current_turn < min_turns: with min_turns=4, # the first decision call happens on the turn-5 call. A required # judgment is never gated. min_turns = getattr(input.scenario_state.config, "min_turns", None) if ( not judgment_required and isinstance(min_turns, int) and input.scenario_state.current_turn < min_turns ): return [] fetch_remote_traces, trace_wait_timeout, trace_wait_extension = ( self._remote_trace_settings(input) ) view = await self._build_conversation_view(input) discovery_recap: List[dict] = [] if judgment_required: verdict_forced = True else: outcome = self._run_decision_phase( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, ) if outcome.decision == "continue": return [] # "verdict": the judge chose to end the conversation. Its verdict # stays voluntary so an inconclusive outcome continues the # conversation (#886). "exhausted": the decision loop burned its # discovery steps without deciding; the verdict is forced so the # run cannot churn through discovery again every turn. verdict_forced = outcome.decision == "exhausted" discovery_recap = outcome.discovery_recap return await self._run_judgment_phase( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, trace_wait_timeout=trace_wait_timeout, trace_wait_extension=trace_wait_extension, is_last_message=is_last_message, verdict_forced=verdict_forced, discovery_recap=discovery_recap, ) async def _build_conversation_view(self, input: AgentInput) -> _ConversationView: """Builds the transcript-side context both phases share. When the judge model can't ingest audio, transcribes agent audio and substitutes text so the judge can evaluate the content. The transcript is gated on its own estimated size, independent of the span digest: it is built from input.messages regardless of whether the agent under test routed its calls through litellm, so it can be arbitrarily large even when the span trace stays small (issue #836). """ conversation_has_audio = self._conversation_has_audio(input.messages) working_messages = input.messages if conversation_has_audio and not self.effective_include_audio(conversation_has_audio): recording = self._extract_recording(input) if recording is not None: await transcribe_segments(recording) working_messages = _enrich_messages_with_transcripts( input.messages, recording ) transcript = JudgeUtils.build_transcript_from_messages(working_messages) is_large_transcript = estimate_tokens(transcript) > self._token_threshold if is_large_transcript: transcript_for_prompt = ( build_transcript_skeleton(working_messages) + "\n\nUse expand_transcript(indices) to see full message content or grep_transcript(pattern) to search across messages. Reference messages by the index shown in brackets." ) else: transcript_for_prompt = transcript extra_context = ( input.judgment_request.additional_context if input.judgment_request and input.judgment_request.additional_context else None ) extra_context_section = ( f"\n<additional_context>\n{extra_context}\n</additional_context>" if extra_context else "" ) return _ConversationView( working_messages=list(working_messages), transcript_for_prompt=transcript_for_prompt, is_large_transcript=is_large_transcript, extra_context_section=extra_context_section, ) def _build_decision_system_prompt( self, *, description: str, criteria: Sequence[str], fetch_remote_traces: bool, ) -> str: """System prompt for the decision phase. The decision deliberately carries no verdict vocabulary: the judge is told NOT to decide pass or fail yet, so nothing in this call can pre-commit it to an outcome before the verdict phase sees the full evidence. """ if self.system_prompt: content = self.system_prompt + "\n\n" + DECISION_PHASE_RULE if fetch_remote_traces: content += "\n\n" + REMOTE_TRACES_DECISION_RULE return content criteria_str = "\n".join( [f"{idx + 1}. {criterion}" for idx, criterion in enumerate(criteria)] ) remote_rule = ( f"\n- {REMOTE_TRACES_DECISION_RULE}" if fetch_remote_traces else "" ) return f""" <role> You are an LLM as a judge watching a simulated conversation as it plays out live to decide if it has collected enough information to evaluate the agent under test. </role> <goal> Your goal is to decide if the conversation has collected enough information to evaluate the criteria, or if it should continue for longer. Do not decide whether the criteria pass or fail now: that evaluation happens in a separate step after the conversation ends. If enough information has been collected, call the make_verdict tool; if not, call the continue_test tool to let the next step play out. </goal> <scenario> {description} </scenario> <criteria> {criteria_str} </criteria> <rules> - Call make_verdict as soon as the agent has clearly broken one of the "do not" or "should not" criteria; more conversation cannot repair a violation. - Scenario simulations exist to exercise multi-turn conversations: while the conversation is still short, lean towards continuing, and end it only when more turns would clearly add no information for the criteria.{remote_rule} </rules> """ def _build_decision_tools(self) -> List[dict]: """Argument-free decision tools. No reasoning field on purpose: writing reasoning here would push the judge to pre-commit to pass or fail before the evidence is complete, and the text itself is wasted tokens for a binary transition. """ return [ { "type": "function", "function": { "name": "continue_test", "description": "Continue the test with the next step", "strict": True, "parameters": { "type": "object", "properties": {}, "required": [], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "make_verdict", "description": ( "The conversation has collected enough information to " "evaluate the criteria. End the conversation and move " "to the verdict." ), "strict": True, "parameters": { "type": "object", "properties": {}, "required": [], "additionalProperties": False, }, }, }, ] def _run_decision_phase( self, *, input: AgentInput, effective_criteria: List[str], view: _ConversationView, fetch_remote_traces: bool, ) -> _DecisionOutcome: """Phase 1 of the two-phase judge: continue, or move to the verdict. Returns a ``_DecisionOutcome`` whose decision is "continue", "verdict", or "exhausted" (the discovery loop ran out of steps without a decision; the collapsed discovery history rides along so the forced verdict keeps what was gathered). Never fetches remote traces and never produces a verdict; the span digest here holds only what the local collector already has. """ spans = self._span_collector.get_spans_for_thread(input.thread_id) digest, is_large_span_trace = self._build_trace_digest(spans) is_large_trace = is_large_span_trace or view.is_large_transcript messages: List[dict] = [ { "role": "system", "content": self._build_decision_system_prompt( description=input.scenario_state.description, criteria=effective_criteria, fetch_remote_traces=fetch_remote_traces, ), }, { "role": "user", "content": _render_judge_content( transcript=view.transcript_for_prompt, traces_digest=digest, extra_context_section="", ), }, ] tools = self._build_decision_tools() if is_large_span_trace: tools = self._build_progressive_discovery_tools() + tools if view.is_large_transcript: tools = self._build_transcript_discovery_tools() + tools if not is_large_trace: response = self._completion_with_reasoning_off_retry( model=self.model, messages=messages, temperature=self.temperature, api_key=self.api_key, api_base=self.api_base, max_tokens=self.max_tokens, tools=tools, tool_choice="required", **self._extra_params, ) return _DecisionOutcome(decision=self._parse_decision(response)) for _ in range(self._max_discovery_steps): response = self._completion_with_reasoning_off_retry( model=self.model, messages=messages, temperature=self.temperature, api_key=self.api_key, api_base=self.api_base, max_tokens=self.max_tokens, tools=tools, tool_choice="required", **self._extra_params, ) if not hasattr(response, "choices") or len(response.choices) == 0: raise Exception( f"Unexpected response format from LLM: {response.__repr__()}" ) message = cast(Choices, response.choices[0]).message if not message.tool_calls: raise Exception( f"Invalid response from judge agent, tool calls not found: {message.__repr__()}" ) terminal_call = next( ( tc for tc in message.tool_calls if tc.function.name in _DECISION_TOOL_NAMES ), None, ) if terminal_call: return _DecisionOutcome( decision=( "continue" if terminal_call.function.name == "continue_test" else "verdict" ) ) messages.append({ "role": "assistant", "content": message.content or "", "tool_calls": [ { "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, } for tc in message.tool_calls ], }) for tc in message.tool_calls: messages.append({ "role": "tool", "tool_call_id": tc.id, "content": self._execute_discovery_tool( tc, spans, view.working_messages ), }) logger.debug( "decision discovery exhausted its steps without a decision - " "forcing the verdict" ) # Everything after the system and content messages is discovery # cycles; collapsed to plain text they carry what the judge gathered # into the forced verdict call. return _DecisionOutcome( decision="exhausted", discovery_recap=_collapse_discovery_history(messages)[2:], ) def _parse_decision(self, response: Any) -> str: """Maps a decision-phase LLM response to "continue" or "verdict".""" if not hasattr(response, "choices") or len(response.choices) == 0: raise Exception( f"Unexpected response format from LLM: {response.__repr__()}" ) message = cast(Choices, response.choices[0]).message if not message.tool_calls: raise Exception( f"Invalid response from judge agent, tool calls not found: {message.__repr__()}" ) terminal_call = next( ( tc for tc in message.tool_calls if tc.function.name in _DECISION_TOOL_NAMES ), None, ) if terminal_call is None: raise Exception( f"Invalid tool call from judge agent: {message.tool_calls[0].function.name}" ) return ( "continue" if terminal_call.function.name == "continue_test" else "verdict" ) async def _run_judgment_phase( self, *, input: AgentInput, effective_criteria: List[str], view: _ConversationView, fetch_remote_traces: bool, trace_wait_timeout: float, trace_wait_extension: float, is_last_message: bool, verdict_forced: bool, discovery_recap: Optional[List[dict]] = None, wait_extension_used: bool = False, ) -> AgentReturnTypes: """Phase 2 of the two-phase judge: the verdict itself. Settle-waits for the remote traces first when fetching is on (the only fetch site), so the digest always holds the full evidence, then makes one finish_test-pinned evaluation. When the traces are still incomplete after the settle-wait, the call also offers a one-shot ``wait_for_traces`` tool: calling it settle-waits once more under ``trace_wait_extension`` and re-enters the phase with the tool withdrawn (``wait_extension_used``), so the second call must decide. ``verdict_forced`` reflects the entry mode: a required judgment (last turn, explicit judgment_request) or decision-discovery exhaustion makes an inconclusive verdict terminal; a voluntary make_verdict entry lets an inconclusive verdict continue the conversation (#886), unless not one remote trace of the run ever settled, in which case more turns cannot improve the evidence and the verdict stands. A non-empty ``discovery_recap`` marks the exhaustion entry: the decision loop already spent the discovery budget, so its collapsed cycles are replayed as context and the verdict is one pinned call with no further discovery. """ remote_trace_ids: List[str] = ( _distinct_message_trace_ids(input.messages) if fetch_remote_traces else [] ) all_settled = True if fetch_remote_traces and remote_trace_ids: all_settled = await self._remote_trace_fetcher.settle_traces( thread_id=input.thread_id, trace_ids=remote_trace_ids, collector=self._span_collector, timeout=trace_wait_timeout, ) elif fetch_remote_traces: # Fetching is on and there is nothing to fetch. Without this the # traces section is silently empty and the judge marks internal # criteria inconclusive without a stated reason. logger.warning( "Remote trace fetching is on but no message carries a trace " "id; nothing to fetch" ) self._remote_trace_fetcher.record_missing_trace_ids( thread_id=input.thread_id, collector=self._span_collector, ) # When not one remote trace of the run ever settled, more turns # cannot produce trace evidence: a voluntary inconclusive verdict # would loop (verdict, continue, settle, inconclusive again) all the # way to the turn cap, paying the settle budget every turn. The # verdict becomes terminal instead; with any settled trace, #886 # semantics stay. evidence_exhausted = ( fetch_remote_traces and bool(remote_trace_ids) and self._remote_trace_fetcher.none_settled( thread_id=input.thread_id, trace_ids=remote_trace_ids ) ) verdict_is_terminal = verdict_forced or evidence_exhausted # The judge's one extra wait: offered as a wait_for_traces tool while # the traces are incomplete, consumed at most once, then withdrawn so # the second call must decide. With no trace ids at all there is # nothing a wait could produce, so the tool is never offered. wait_extension_available = ( fetch_remote_traces and bool(remote_trace_ids) and not all_settled and trace_wait_extension > 0 and not wait_extension_used ) spans = self._span_collector.get_spans_for_thread(input.thread_id) digest, is_large_span_trace = self._build_trace_digest(spans) is_large_trace = is_large_span_trace or view.is_large_transcript logger.debug(f"OpenTelemetry traces built: {digest[:200]}...") content_for_judge = _render_judge_content( transcript=view.transcript_for_prompt, traces_digest=digest, extra_context_section=view.extra_context_section, ) criteria_str = "\n".join( [f"{idx + 1}. {criterion}" for idx, criterion in enumerate(effective_criteria)] ) remote_traces_rule = ( f"\n- {REMOTE_TRACES_JUDGE_RULE}" if fetch_remote_traces else "" ) system_content = self.system_prompt or f""" <role> You are an LLM as a judge delivering the final verdict on a simulated conversation, determining if the agent under test meets the criteria or not. </role> <goal> Your goal is to deliver the final verdict of the scenario below with the finish_test tool, evaluating each criterion independently against the conversation and the collected evidence. </goal> <scenario> {input.scenario_state.description} </scenario> <criteria> {criteria_str} </criteria> <rules> - Be strict: a criterion passes only when the conversation or the collected evidence clearly shows it was met. - DO NOT make any judgment calls that are not explicitly listed in the success or failure criteria, withhold judgement if necessary - When the evidence for a criterion is not definitive, mark that criterion inconclusive rather than guessing; an inconclusive verdict is acceptable{remote_traces_rule} </rules> """ if self.system_prompt and fetch_remote_traces: system_content = self.system_prompt + "\n\n" + REMOTE_TRACES_JUDGE_RULE messages: List[dict] = [ {"role": "system", "content": system_content}, {"role": "user", "content": content_for_judge}, ] if is_last_message: messages.append( { "role": "user", "content": """ System: <finish_test> This is the last message, conversation has reached the maximum number of turns, give your final verdict, if you don't have enough information to make a verdict, say inconclusive with max turns reached. </finish_test> """, } ) if wait_extension_used: messages.append( { "role": "user", "content": ( "You already waited once more for the remote traces. " "The trace evidence above is final: deliver your " "verdict now." ), } ) criteria_names = _criteria_keys(effective_criteria) tools: List[dict] = [ { "type": "function", "function": { "name": "finish_test", "description": "Complete the test with a final verdict", "strict": True, "parameters": { "type": "object", "properties": { "criteria": { "type": "object", "properties": { criteria_names[idx]: { "type": "string", "enum": ["true", "false", "inconclusive"], "description": criterion, } for idx, criterion in enumerate(effective_criteria) }, "required": criteria_names, "additionalProperties": False, "description": "Strict verdict for each criterion", }, "reasoning": { "type": "string", "description": "Explanation of what the final verdict should be", }, "verdict": { "type": "string", "enum": ["success", "failure", "inconclusive"], "description": "The final verdict of the test", }, }, "required": ["criteria", "reasoning", "verdict"], "additionalProperties": False, }, }, }, ] exhausted_entry = bool(discovery_recap) if not exhausted_entry: if is_large_span_trace: tools = self._build_progressive_discovery_tools() + tools if view.is_large_transcript: tools = self._build_transcript_discovery_tools() + tools if wait_extension_available: tools = [_build_wait_for_traces_tool()] + tools # finish_test is the only terminal tool of the verdict phase and the # tool choice pins it: continuing is the decision phase's business. # While the traces are incomplete and the extension is unused, the # wait_for_traces tool joins the set and the pin relaxes to # "required" so the judge can pick either. The large-trace discovery # loop relaxes the pin to "required" on its intermediate steps so # the judge can use discovery tools, and forces the verdict on # exhaustion. tool_choice: Any = ( "required" if wait_extension_available else {"type": "function", "function": {"name": "finish_test"}} ) if exhausted_entry: assert discovery_recap is not None messages.extend(discovery_recap) messages.append({ "role": "user", "content": ( "You have reached the maximum number of trace exploration steps. " "Based on the information you have gathered so far, give your final verdict now." ), }) elif is_large_trace: outcome = self._run_discovery_loop( messages=messages, tools=tools, tool_choice=tool_choice, spans=spans, working_messages=view.working_messages, effective_criteria=effective_criteria, input_messages=input.messages, verdict_forced=verdict_is_terminal, wait_tool_offered=wait_extension_available, ) if outcome is _WAIT_FOR_TRACES_REQUESTED: return await self._extend_wait_and_rejudge( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, trace_wait_timeout=trace_wait_timeout, trace_wait_extension=trace_wait_extension, is_last_message=is_last_message, verdict_forced=verdict_forced, discovery_recap=discovery_recap, remote_trace_ids=remote_trace_ids, ) return cast(AgentReturnTypes, outcome) response = self._completion_with_reasoning_off_retry( model=self.model, messages=messages, temperature=self.temperature, api_key=self.api_key, api_base=self.api_base, max_tokens=self.max_tokens, tools=tools, tool_choice=tool_choice, **self._extra_params, ) if wait_extension_available and _response_called_wait_for_traces(response): return await self._extend_wait_and_rejudge( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, trace_wait_timeout=trace_wait_timeout, trace_wait_extension=trace_wait_extension, is_last_message=is_last_message, verdict_forced=verdict_forced, discovery_recap=discovery_recap, remote_trace_ids=remote_trace_ids, ) return self._parse_response( response, effective_criteria, messages, input_messages=input.messages, verdict_forced=verdict_is_terminal, ) async def _extend_wait_and_rejudge( self, *, input: AgentInput, effective_criteria: List[str], view: _ConversationView, fetch_remote_traces: bool, trace_wait_timeout: float, trace_wait_extension: float, is_last_message: bool, verdict_forced: bool, discovery_recap: Optional[List[dict]], remote_trace_ids: List[str], ) -> AgentReturnTypes: """Runs the judge's one extra wait, then re-enters the verdict. Re-arms the failed traces, settle-waits once more under the extension budget, and re-enters the judgment phase with the wait_for_traces tool withdrawn, so the second call must decide. """ logger.debug( "Judge requested one more wait for the remote traces (%.0fs)", trace_wait_extension, ) await self._remote_trace_fetcher.extend_settle( thread_id=input.thread_id, trace_ids=remote_trace_ids, collector=self._span_collector, timeout=trace_wait_extension, ) return await self._run_judgment_phase( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, trace_wait_timeout=trace_wait_timeout, trace_wait_extension=trace_wait_extension, is_last_message=is_last_message, verdict_forced=verdict_forced, discovery_recap=discovery_recap, wait_extension_used=True, ) def _remote_trace_settings(self, input: AgentInput) -> "tuple[bool, float, float]": """Resolves the remote trace fetching configuration for this call. Reads ``fetch_remote_traces`` (effective default False), ``trace_wait_timeout`` (effective default 30 seconds) and ``trace_wait_extension`` (effective default: the resolved timeout) from the scenario configuration. """ config = getattr(input.scenario_state, "config", None) enabled = getattr(config, "fetch_remote_traces", None) is True timeout = getattr(config, "trace_wait_timeout", None) if ( not isinstance(timeout, (int, float)) or isinstance(timeout, bool) or timeout <= 0 ): timeout = DEFAULT_TRACE_WAIT_TIMEOUT_SECONDS # The one extra wait the judge may request via the wait_for_traces # tool. Defaults to the wait budget itself; the platform passes its # upper cap here so a short measured budget still gets a meaningful # extension. extension = getattr(config, "trace_wait_extension", None) if ( not isinstance(extension, (int, float)) or isinstance(extension, bool) or extension <= 0 ): extension = timeout return enabled, float(timeout), float(extension) def _completion_with_reasoning_off_retry(self, **kwargs: Any) -> ModelResponse: """ ``litellm.completion``, retried once with reasoning declared off when — and only when — the provider rejected a tool-carrying call for exactly that reason. A caller that already asked for a specific effort keeps it and gets the endpoint's own error, rather than having its intent silently rewritten. """ try: return cast(ModelResponse, litellm.completion(**kwargs)) except Exception as error: if not kwargs.get("tools") or "reasoning_effort" in kwargs: raise if not _rejection_asks_for_reasoning_off(error): raise logger.debug( "provider rejected function tools without reasoning off for %s; retrying", kwargs.get("model"), ) return cast( ModelResponse, litellm.completion(**kwargs, reasoning_effort=_REASONING_OFF), ) def _build_trace_digest(self, spans: Sequence[Any]) -> tuple[str, bool]: """ Builds the trace digest, choosing between full inline rendering and structure-only mode based on estimated token count. Args: spans: The spans for this thread. Returns: Tuple of (digest_string, is_large_trace). """ full_digest = judge_span_digest_formatter.format(spans) is_large_trace = ( len(spans) > 0 and estimate_tokens(full_digest) > self._token_threshold ) if is_large_trace: digest = ( judge_span_digest_formatter.format_structure_only(spans) + "\n\nUse expand_trace(span_id) to see span details or grep_trace(pattern) to search across spans. Reference spans by the ID shown in brackets." ) else: digest = full_digest logger.debug( "Trace digest built", extra={ "is_large_trace": is_large_trace, "estimated_tokens": estimate_tokens(full_digest), }, ) return digest, is_large_trace def _build_progressive_discovery_tools(self) -> List[dict]: """ Builds the expand_trace and grep_trace tool definitions for litellm. Returns: List of tool definition dicts for litellm function calling. """ return [ { "type": "function", "function": { "name": "expand_trace", "description": ( "Expand one or more spans to see their full details " "(attributes, events, content). Use the span ID shown " "in brackets in the trace skeleton." ), "parameters": { "type": "object", "properties": { "span_ids": { "type": "array", "items": {"type": "string"}, "description": "Span IDs (or 8-char prefixes) to expand", }, }, "required": ["span_ids"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "grep_trace", "description": ( "Search across all span attributes, events, and content " "for a pattern (case-insensitive). Returns matching spans with context." ), "parameters": { "type": "object", "properties": { "pattern": { "type": "string", "description": "Search pattern (case-insensitive)", }, }, "required": ["pattern"], "additionalProperties": False, }, }, }, ] def _build_transcript_discovery_tools(self) -> List[dict]: """ Builds the expand_transcript and grep_transcript tool definitions for litellm. Parallel to ``_build_progressive_discovery_tools``, but for message-transcript discovery instead of span discovery (see ``transcript_tools.py`` for why the two need to be independent). Returns: List of tool definition dicts for litellm function calling. """ return [ { "type": "function", "function": { "name": "expand_transcript", "description": ( "Expand one or more messages to see their full content. " "Use the message index shown in brackets in the transcript " "skeleton." ), "parameters": { "type": "object", "properties": { "indices": { "type": "array", "items": {"type": "integer"}, "description": "0-based message indices to expand", }, }, "required": ["indices"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "grep_transcript", "description": ( "Search across all message content for a pattern " "(case-insensitive). Returns matching messages with context." ), "parameters": { "type": "object", "properties": { "pattern": { "type": "string", "description": "Search pattern (case-insensitive)", }, }, "required": ["pattern"], "additionalProperties": False, }, }, }, ] def _run_discovery_loop( self, *, messages: List[dict], tools: List[dict], tool_choice: Any, spans: Sequence[Any], working_messages: Sequence[ChatCompletionMessageParam], effective_criteria: List[str], input_messages: Sequence[Any], verdict_forced: bool, wait_tool_offered: bool = False, ) -> Union[AgentReturnTypes, _WaitForTracesRequested]: """ Runs the multi-step discovery loop of the verdict phase for large traces. The judge can call expand_trace/grep_trace (spans) and/or expand_transcript/grep_transcript (messages) tools multiple times before calling finish_test, the only terminal tool of the verdict phase, or hitting the max discovery steps limit, which forces the verdict with whatever was gathered. On intermediate steps, tool_choice is "required" so the judge can freely pick a discovery tool. On the final step, the pinned finish_test tool_choice is applied. Args: messages: The conversation messages so far. tools: The tool definitions. tool_choice: The tool choice constraint for the final step. spans: The spans for executing expand_trace/grep_trace. working_messages: The conversation messages for executing expand_transcript/grep_transcript. effective_criteria: The criteria to judge against. Returns: AgentReturnTypes from the finish_test call. """ for step in range(self._max_discovery_steps): # Use "required" for intermediate steps so the judge can use # discovery tools; only apply the forced tool_choice on the # last allowed step. is_last_step = step == self._max_discovery_steps - 1 step_tool_choice = tool_choice if is_last_step else "required" response = self._completion_with_reasoning_off_retry( model=self.model, messages=messages, temperature=self.temperature, api_key=self.api_key, api_base=self.api_base, max_tokens=self.max_tokens, tools=tools, tool_choice=step_tool_choice, **self._extra_params, ) if not hasattr(response, "choices") or len(response.choices) == 0: raise Exception( f"Unexpected response format from LLM: {response.__repr__()}" ) message = cast(Choices, response.choices[0]).message if not message.tool_calls: # No tool calls - try to parse as a response return self._parse_response( response, effective_criteria, messages, input_messages=input_messages, verdict_forced=verdict_forced, ) if wait_tool_offered and any( tc.function.name == "wait_for_traces" for tc in message.tool_calls ): # The extra wait rebuilds the whole phase: the caller # settle-waits once more and re-enters with a fresh digest # and the tool withdrawn. return _WAIT_FOR_TRACES_REQUESTED terminal_call = next( (tc for tc in message.tool_calls if tc.function.name == "finish_test"), None, ) if terminal_call: return self._parse_response( response, effective_criteria, messages, input_messages=input_messages, verdict_forced=verdict_forced, ) # Execute discovery tools and add results to messages # Add the assistant message with tool calls messages.append({ "role": "assistant", "content": message.content or "", "tool_calls": [ { "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, } for tc in message.tool_calls ], }) for tc in message.tool_calls: tool_result = self._execute_discovery_tool(tc, spans, working_messages) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": tool_result, }) return self._force_verdict( messages=messages, tools=tools, effective_criteria=effective_criteria, input_messages=input_messages, ) def _force_verdict( self, *, messages: List[dict], tools: List[dict], effective_criteria: List[str], input_messages: Sequence[Any], ) -> AgentReturnTypes: """ Makes one final LLM call with tool_choice forced to finish_test. Hardening (vs. a naive re-invocation with the same tool set): - Prior discovery tool_call/tool_result pairs are rewritten in the message history as plain-text assistant recaps. This lets us drop expand_trace/grep_trace from the tool set without Anthropic rejecting the call for referencing undefined tools. - Discovery tools are then stripped so the model physically cannot emit them, closing the leak path where tool_choice wasn't honored and a discovery tool reached _parse_response. """ logger.warning( f"Progressive discovery hit max steps ({self._max_discovery_steps}), " "forcing verdict" ) rewritten_messages = _collapse_discovery_history(messages) rewritten_messages.append({ "role": "user", "content": ( "You have reached the maximum number of trace exploration steps. " "Based on the information you have gathered so far, give your final verdict now." ), }) # finish_test only, not just "everything except discovery". The # verdict phase also offers wait_for_traces while the extension is # unused, and it would survive a deny-list. The pin below asks for # finish_test, but a model that ignores the pin and calls # wait_for_traces here reaches _parse_response as an invalid tool # call. Leaving one tool closes that path. finish_only_tools = [ t for t in tools if t.get("function", {}).get("name") == "finish_test" ] forced_response = self._completion_with_reasoning_off_retry( model=self.model, messages=rewritten_messages, temperature=self.temperature, api_key=self.api_key, api_base=self.api_base, max_tokens=self.max_tokens, tools=finish_only_tools, tool_choice={"type": "function", "function": {"name": "finish_test"}}, **self._extra_params, ) return self._parse_response( forced_response, effective_criteria, rewritten_messages, input_messages=input_messages, # The whole point of this call is a pinned finish_test — the model # has no continue escape, so an inconclusive verdict is legitimate. verdict_forced=True, ) def _execute_discovery_tool( self, tool_call: Any, spans: Sequence[Any], working_messages: Sequence[ChatCompletionMessageParam], ) -> str: """ Executes an expand_trace, grep_trace, expand_transcript, or grep_transcript tool call. Args: tool_call: The tool call from the LLM response. spans: The spans to operate on for expand_trace/grep_trace. working_messages: The conversation messages to operate on for expand_transcript/grep_transcript. Returns: The tool result string. """ try: args = json.loads(tool_call.function.arguments) except json.JSONDecodeError: return f"Error: could not parse arguments: {tool_call.function.arguments}" if tool_call.function.name == "expand_trace": return expand_trace( spans, span_ids=args.get("span_ids", []), ) elif tool_call.function.name == "grep_trace": return grep_trace(spans, args.get("pattern", "")) elif tool_call.function.name == "expand_transcript": return expand_transcript( working_messages, indices=args.get("indices", []), ) elif tool_call.function.name == "grep_transcript": return grep_transcript(working_messages, args.get("pattern", "")) else: return f"Unknown tool: {tool_call.function.name}" def _parse_response( self, response: Any, effective_criteria: List[str], messages: List[dict], *, input_messages: Sequence[Any], verdict_forced: bool, ) -> AgentReturnTypes: """ Parses a litellm response into the appropriate return type. Handles finish_test, continue_test, and error cases. Args: response: The litellm ModelResponse. effective_criteria: The criteria to evaluate against. messages: The judge's internal LLM messages (system prompt + transcript). input_messages: The actual conversation messages to include in ScenarioResult. Returns: AgentReturnTypes: Either an empty list (continue) or ScenarioResult. """ if not hasattr(response, "choices") or len(response.choices) == 0: raise Exception( f"Unexpected response format from LLM: {response.__repr__()}" ) message = cast(Choices, response.choices[0]).message if not message.tool_calls: raise Exception( f"Invalid response from judge agent, tool calls not found: {message.__repr__()}" ) # In multi-step mode, find the terminal tool call terminal_call = next( (tc for tc in message.tool_calls if tc.function.name == "finish_test"), None, ) tool_call = terminal_call or message.tool_calls[0] if tool_call.function.name == "finish_test": try: args = json.loads(tool_call.function.arguments) except json.JSONDecodeError: raise Exception( f"Failed to parse tool call arguments from judge agent: {tool_call.function.arguments}" ) verdict = args.get("verdict", "inconclusive") reasoning = args.get("reasoning", "No reasoning provided") # "Can't tell yet" is not a verdict (#886). When nothing forced the # judge to finish — continue_test was freely available — an # inconclusive finish_test used to end the run as a failure, which # on a platform surface reads as the simulated user going silent # mid-conversation. Treat it as continue_test and let the # conversation play out; a FORCED judgment (last turn, an explicit # judgment_request, discovery exhaustion) keeps its terminal # behavior unchanged. if not verdict_forced and verdict == "inconclusive": logger.debug( "finish_test returned an inconclusive verdict without a " "forced judgment - continuing the conversation" ) return [] criteria_verdicts = args.get("criteria", {}) # LLMs sometimes serialise the criteria object as a JSON *string* # instead of an inline dict, especially with complex dynamic # schemas (issue #161). Re-parse one level if that happens. if isinstance(criteria_verdicts, str): try: criteria_verdicts = json.loads(criteria_verdicts) except (json.JSONDecodeError, ValueError): criteria_verdicts = None # unparseable — handled below # If the criteria payload is not a usable object, we cannot trust # any per-criterion verdict. Do NOT fall back to {}: an empty dict # makes failed_criteria empty and lets a "success" verdict slip # through having evaluated ZERO criteria, masking the real problem # (issue #161 follow-up). Surface it as an explicit, fail-closed # result instead of swallowing it. if not isinstance(criteria_verdicts, dict): raw = args.get("criteria") logger.warning( "JudgeAgent could not resolve criteria verdicts to an " "object (got %s); failing the judgment instead of " "reporting an unverified success.", type(raw).__name__, ) return ScenarioResult( success=False, messages=cast(Any, input_messages), reasoning=( "JudgeAgent could not parse the per-criterion verdicts " "returned by the LLM, so the judgment could not be " f"verified (raw criteria value was of type " f"{type(raw).__name__}). Original verdict was " f"{verdict!r}. Treating the judgment as failed." ), passed_criteria=[], failed_criteria=list(effective_criteria), ) # Map each verdict back to its criterion BY the schema key we # generated for it. Positional .values() mapping silently # mislabels partial / reordered payloads and IndexErrors on extra # keys; key-based lookup is robust. A criterion passes ONLY on an # explicit "true"; anything else (false, inconclusive, missing, or # a nested/unexpected value) is a failure, so an unevaluated # criterion can never slip through as success. # Map each verdict to its criterion by the schema key we generated # for it (single source of truth: _criteria_keys). Positional # .values() mapping silently mislabels partial / reordered / nested # payloads and IndexErrors on extra keys; key lookup is robust. # A criterion passes ONLY on an explicit "true"; anything else # (false, inconclusive, missing, or a nested/unexpected value) is a # failure, so an unevaluated criterion can never slip through as # success. JSON booleans are coerced — some LLMs emit `true`/`false` # instead of the enum strings. criteria_keys = _criteria_keys(effective_criteria) passed_criteria: List[str] = [] failed_criteria: List[str] = [] for criterion, key in zip(effective_criteria, criteria_keys): raw_verdict = criteria_verdicts.get(key) if isinstance(raw_verdict, bool): raw_verdict = "true" if raw_verdict else "false" bucket = passed_criteria if raw_verdict == "true" else failed_criteria bucket.append(criterion) return ScenarioResult( success=verdict == "success" and len(failed_criteria) == 0, messages=cast(Any, input_messages), reasoning=reasoning, passed_criteria=passed_criteria, failed_criteria=failed_criteria, ) if tool_call.function.name in _DISCOVERY_TOOL_NAMES: logger.warning( f"Discovery tool {tool_call.function.name} leaked past " "discovery loop without reaching a terminal verdict" ) return ScenarioResult( success=False, messages=cast(Any, input_messages), reasoning=( "JudgeAgent: trace discovery did not converge on a " "verdict within the step budget" ), passed_criteria=[], failed_criteria=list(effective_criteria), ) raise Exception( f"Invalid tool call from judge agent: {tool_call.function.name}" )Ancestors
- AgentAdapter
- abc.ABC
Class variables
var api_base : str | Nonevar api_key : str | Nonevar criteria : List[str]var max_tokens : int | Nonevar model : strvar role : ClassVar[AgentRole]var system_prompt : str | Nonevar temperature : float
Methods
async def call(self, input: AgentInput) ‑> str | openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam | openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam | openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam | openai.types.chat.chat_completion_assistant_message_param.ChatCompletionAssistantMessageParam | openai.types.chat.chat_completion_tool_message_param.ChatCompletionToolMessageParam | openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam | List[openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam | openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam | openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam | openai.types.chat.chat_completion_assistant_message_param.ChatCompletionAssistantMessageParam | openai.types.chat.chat_completion_tool_message_param.ChatCompletionToolMessageParam | openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam] | ScenarioResult-
Evaluate the current conversation state against the configured criteria.
The judge runs in two phases. A mid-conversation call first makes an argument-free decision between continuing the conversation and moving to the verdict (continue_test / make_verdict); only a make_verdict decision triggers the verdict call, which settle-waits for remote traces when enabled and evaluates every criterion with finish_test. The last turn and an explicit judgment request skip the decision and go straight to the verdict.
Args
input- AgentInput containing conversation history and scenario context
Returns
AgentReturnTypes- Either an empty list (continue scenario) or a ScenarioResult (end scenario with verdict)
Raises
Exception- If the judge cannot make a valid decision or if there's an error in the evaluation process
Note
- Returns empty list [] to continue the scenario
- Returns ScenarioResult to end with success/failure
- Provides detailed reasoning for all decisions
- Evaluates each criterion independently
- Can end scenarios early if clear violation or success is detected
Expand source code
@scenario_cache() async def call( self, input: AgentInput, ) -> AgentReturnTypes: """ Evaluate the current conversation state against the configured criteria. The judge runs in two phases. A mid-conversation call first makes an argument-free decision between continuing the conversation and moving to the verdict (continue_test / make_verdict); only a make_verdict decision triggers the verdict call, which settle-waits for remote traces when enabled and evaluates every criterion with finish_test. The last turn and an explicit judgment request skip the decision and go straight to the verdict. Args: input: AgentInput containing conversation history and scenario context Returns: AgentReturnTypes: Either an empty list (continue scenario) or a ScenarioResult (end scenario with verdict) Raises: Exception: If the judge cannot make a valid decision or if there's an error in the evaluation process Note: - Returns empty list [] to continue the scenario - Returns ScenarioResult to end with success/failure - Provides detailed reasoning for all decisions - Evaluates each criterion independently - Can end scenarios early if clear violation or success is detected """ effective_criteria = ( input.judgment_request.criteria if input.judgment_request and input.judgment_request.criteria is not None else self.criteria ) max_turns = input.scenario_state.config.max_turns or 10 is_last_message = ( input.scenario_state.current_turn >= max_turns - 1 ) enforce_judgment = input.judgment_request is not None has_criteria = len(effective_criteria) > 0 if enforce_judgment and not has_criteria: return ScenarioResult( success=False, messages=[], reasoning="TestingAgent was called as a judge, but it has no criteria to judge against", ) # A judgment is required when the conversation cannot continue past # this call: the last turn, or an explicit judgment_request. Both go # straight to the verdict phase; only an unforced mid-conversation # call runs the decision phase first. judgment_required = is_last_message or enforce_judgment # min_turns floor (ADR-005): below the floor the decision is # predetermined (the conversation must continue), so nothing is spent # on it. The check runs before the conversation view is built, or a # gated voice turn pays for a transcription it discards. The judge # observes a 0-based current_turn: reset() overrides the initial # _new_turn() back to 0, so the call on turn N sees current_turn N-1. # The floor is unmet while current_turn < min_turns: with min_turns=4, # the first decision call happens on the turn-5 call. A required # judgment is never gated. min_turns = getattr(input.scenario_state.config, "min_turns", None) if ( not judgment_required and isinstance(min_turns, int) and input.scenario_state.current_turn < min_turns ): return [] fetch_remote_traces, trace_wait_timeout, trace_wait_extension = ( self._remote_trace_settings(input) ) view = await self._build_conversation_view(input) discovery_recap: List[dict] = [] if judgment_required: verdict_forced = True else: outcome = self._run_decision_phase( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, ) if outcome.decision == "continue": return [] # "verdict": the judge chose to end the conversation. Its verdict # stays voluntary so an inconclusive outcome continues the # conversation (#886). "exhausted": the decision loop burned its # discovery steps without deciding; the verdict is forced so the # run cannot churn through discovery again every turn. verdict_forced = outcome.decision == "exhausted" discovery_recap = outcome.discovery_recap return await self._run_judgment_phase( input=input, effective_criteria=effective_criteria, view=view, fetch_remote_traces=fetch_remote_traces, trace_wait_timeout=trace_wait_timeout, trace_wait_extension=trace_wait_extension, is_last_message=is_last_message, verdict_forced=verdict_forced, discovery_recap=discovery_recap, ) def effective_include_audio(self, conversation_has_audio: bool) ‑> bool-
Resolve include_audio: explicit wins, otherwise use modality resolver.
Intentional behavior change (Bundle 3 / AC3b): Before: gpt-4o → audio-capable (substring match). After: gpt-4o → text path (litellm advisory returns False). Before: gpt-audio-mini → NOT audio-capable (not in list). After: gpt-audio-mini → audio-capable (litellm advisory returns True). The old substring list was wrong; the resolver is the source of truth.
Expand source code
def effective_include_audio(self, conversation_has_audio: bool) -> bool: """Resolve include_audio: explicit wins, otherwise use modality resolver. Intentional behavior change (Bundle 3 / AC3b): Before: gpt-4o → audio-capable (substring match). After: gpt-4o → text path (litellm advisory returns False). Before: gpt-audio-mini → NOT audio-capable (not in list). After: gpt-audio-mini → audio-capable (litellm advisory returns True). The old substring list was wrong; the resolver is the source of truth. """ if self.include_audio is not None: # Explicit override always wins (AC3c) return self.include_audio and conversation_has_audio # Use resolver with per-role declaration (AC0, Bundle 6) tier, warnings = resolve_modality(declaration=self.modality, model_id=self.model or "") for w in warnings: logger.warning(w) return conversation_has_audio and (tier == ModalityTier.AUDIO_IN) def effective_include_timeline(self, conversation_has_audio: bool) ‑> bool-
Default timeline True for voice, False for text — unless explicitly set.
Expand source code
def effective_include_timeline(self, conversation_has_audio: bool) -> bool: """Default timeline True for voice, False for text — unless explicitly set.""" if self.include_timeline is not None: return self.include_timeline return conversation_has_audio def effective_include_traces(self, otel_configured: bool) ‑> bool-
Expand source code
def effective_include_traces(self, otel_configured: bool) -> bool: if self.include_traces is not None: return self.include_traces return otel_configured