Trajectory Judge

Watch a live agent run with a second model, and steer it back on course mid-run.

Source

While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.

The problem

Long-horizon runs drift. Instructions fade, unsupported claims compound, and the agent wanders away from the goal. Guardrails catch policy violations at fixed boundaries, and evals score the run after it ends, when recovery costs a whole re-run. Neither watches the trajectory as it unfolds and intervenes while a correction is still cheap.

The solution

TrajectoryJudge reviews the run’s recent trajectory with a second model on a cadence: every every model requests, the most recent window tokens of the conversation (user messages, assistant messages, tool calls, and tool results, rendered as a transcript) go to the judge model for evaluation. The evaluation runs concurrently with the agent, so the run is never blocked waiting on a judge.

The judge delivers exactly one verdict per evaluation, as its output type:

• AllGood — the run is on track; nothing happens.
• Steer(message=...) — the run needs correction; the message is enqueued into the running conversation (RunContext.enqueue, 'asap' priority) and delivered on the next model request, attributed to the judge: Steering from trajectory judge 'hallucination-check': ....

Steering is the judge’s output, not a tool it calls: the judge knows it gets one final verdict per evaluation, rather than being tempted to steer repeatedly mid-thought.

from pydantic_ai import Agent
from pydantic_ai_harness import TrajectoryJudge

agent = Agent(
    'anthropic:claude-sonnet-5',
    capabilities=[
        TrajectoryJudge(
            model='anthropic:claude-haiku-4-5',
            instructions='Flag claims that lack evidence from files the agent actually read.',
            every=20,
        )
    ],
)

result = agent.run_sync('Fix the flaky checkout test and add a regression test.')
print(result.output)

Cadence and window

• every counts model requests within the run; the judge evaluates on each multiple.
• window bounds what each evaluation sees: the transcript is clamped to its most recent window tokens (estimated at ~4 characters per token), so per-evaluation cost stays bounded no matter how long the run gets.
• At most one evaluation per judge is in flight at a time. A cadence tick that finds the previous evaluation still running is skipped, so a slow judge falls behind rather than piling up concurrent calls.
• An evaluation still in flight when the run ends is cancelled: its steering would have nowhere to go.

Several judges

Each judge is its own capability instance; add one per concern. They schedule and evaluate independently, and each steering message carries its own attribution (name, the judge agent’s name, or 'trajectory-judge').

from pydantic_ai import Agent
from pydantic_ai_harness import TrajectoryJudge

agent = Agent(
    'anthropic:claude-sonnet-5',
    capabilities=[
        TrajectoryJudge(
            model='anthropic:claude-haiku-4-5',
            name='hallucination-check',
            instructions='Flag claims that lack evidence from files the agent actually read.',
            every=20,
        ),
        TrajectoryJudge(
            model='google:gemini-3.7-flash',
            name='scope-creep',
            instructions='Flag work that was not asked for in the original request.',
            every=10,
        ),
    ],
)

Advanced: bring your own judge agent

For anything beyond a model and a review focus (model settings, toolsets, fallback models, custom instructions), pass a full Agent instead of piling knobs onto the capability. Every evaluation runs it with output_type=[AllGood, Steer], so the verdict contract is enforced at the run boundary whatever output type the agent was configured with, and an existing agent can be reused as-is only when it is dependency-free; judge dependencies are not passed to evaluations. The one constraint: the judge agent must not have output validators, which are incompatible with a per-run output_type.

from pydantic_ai import Agent
from pydantic_ai_harness import TrajectoryJudge
from pydantic_ai_harness.trajectory_judge import AllGood, Steer

judge = Agent(
    'openai:gpt-5.6-luna',
    name='security-risk',
    instructions=(
        'You review an AI agent trajectory for security risks: exposed secrets, unsafe '
        'file access, and unexpected egress. Return all-good, or steer with a specific warning.'
    ),
    output_type=[AllGood, Steer],
)

agent = Agent(
    'anthropic:claude-sonnet-5',
    capabilities=[TrajectoryJudge(agent=judge, every=10)],
)

agent is mutually exclusive with model/instructions: the passed agent owns its own instructions.

Cost and failure semantics

• The judge’s model usage is threaded onto the run’s usage and respects the run’s usage_limits: each launch claims one request on the shared usage before the evaluation starts, so the parent’s next request and concurrent judges account for in-flight evaluations and the shared request limit cannot be exceeded. A launch the request budget cannot fit skips the tick, like one that finds an evaluation still in flight.
• An evaluation failure is raised on the run at the next cadence tick or at run end; judge failures are never silently dropped. If you need a judge to degrade instead, give it a fallback model through agent (for example a FallbackModel): resilience policy belongs to the judge agent, not to fields on the capability.
• A judged run inside a durable execution workflow or flow (Temporal, DBOS, Prefect) is rejected with UserError before the first model request: the evaluation is launched from a capability hook in orchestration context, so its model calls would not be checkpointed and could repeat on replay. A durable-capable agent run outside its workflow or flow is unaffected. Run judged work outside durable execution.

Observability

on_verdict is called with each verdict after it is processed (after any steering has been enqueued):

from pydantic_ai_harness import TrajectoryJudge

TrajectoryJudge(
    model='anthropic:claude-haiku-4-5',
    every=20,
    on_verdict=lambda verdict: print(f'judge verdict: {verdict}'),
)

Composition

• System Reminders is the rule-based sibling: cadence or condition-triggered reminders with no extra model call. Reach for it first; a trajectory judge earns its cost when the condition requires actually understanding the trajectory.
• Guardrails remain the enforcement boundary for inputs, tool calls, and outputs. A judge observes and steers; it does not block or rewrite anything.

Not spec-serializable

TrajectoryJudge.get_serialization_name() returns None: the capability may hold a live Agent instance and a callback, which cannot be serialized to an agent spec.

Further reading

• Pydantic AI capabilities
• Hooks — after_model_request drives the cadence; wrap_run settles the judge at run end
• System Reminders — rule-based mid-run steering without a second model

API reference

TrajectoryJudge

Bases: AbstractCapability[AgentDepsT]

Review a live run with a second model on a cadence, and steer it mid-run.

Long-horizon runs drift: instructions fade, unsupported claims compound, and the agent wanders off the goal. A TrajectoryJudge evaluates the most recent window tokens of the run’s trajectory every every model requests, concurrently with the run, and delivers exactly one verdict per evaluation: AllGood, or Steer with a corrective message. Steering is enqueued into the run (RunContext.enqueue, 'asap' priority) with attribution to the judge, so the running agent course-corrects while recovery is still cheap.

At most one evaluation per judge is in flight at a time; a cadence tick that finds one still running is skipped. An evaluation still in flight when the run ends is cancelled. The judge’s model and tool usage are threaded onto the run’s usage and respect its usage_limits: each launch claims one request on the shared usage before the evaluation starts, so the parent’s next preflight and sibling launches account for the in-flight call, and a launch the request budget cannot fit is skipped. An evaluation failure is raised on the run at the next cadence tick or at run end; give the judge a fallback model (via agent) if you need it to degrade instead.

from pydantic_ai import Agent
from pydantic_ai_harness import TrajectoryJudge

agent = Agent(
    'anthropic:claude-sonnet-5',
    capabilities=[
        TrajectoryJudge(
            model='anthropic:claude-haiku-4-5',
            instructions='Flag claims that lack evidence from files the agent actually read.',
            every=20,
        )
    ],
)

Several judges can watch one run: add one TrajectoryJudge per concern to capabilities. Each schedules and evaluates independently.

A judged run inside a durable workflow or flow (Temporal, DBOS, Prefect) is rejected with UserError before the first model request: the evaluation is launched from a capability hook in orchestration context, so its model calls would not be checkpointed and could repeat on replay. Run judged work outside durable execution.

Attributes

model

The model that evaluates the trajectory. Provide this (with optional instructions) or a full agent, not both.

Type: Model | KnownModelName | str | None Default: None

instructions

The judge’s review focus, appended to the built-in judge instructions. Only valid together with model; a passed agent owns its own instructions.

Type: str | None Default: None

agent

A full judge agent, for advanced customization (own instructions, model settings, toolsets, fallback models). Every evaluation runs it with output_type=[AllGood, Steer] regardless of its own configured output type, so an existing agent can be reused as-is; it must not have output validators, which are incompatible with a per-run output_type. Mutually exclusive with model/instructions.

Type: Agent[None, object] | None Default: None

every

Evaluate every N model requests within the run.

Type: int Default: 10

window

Sliding token window: each evaluation sees at most this many tokens of the most recent trajectory (estimated at ~4 characters per token), rendered as a transcript.

Type: int Default: 20000

name

Name used to attribute steering messages. Defaults to the judge agent’s name when one is passed, then to 'trajectory-judge'.

Type: str | None Default: None

on_verdict

Optional observability callback invoked with each verdict after it is processed (after any steering has been enqueued).

Type: Callable[[TrajectoryVerdict], None] | None Default: None

Methods

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> TrajectoryJudge[AgentDepsT]

Return a fresh per-run instance so step counts and in-flight evaluations are not shared.

replace re-runs __init__ and __post_init__, resetting the init=False fields: _steps to 0, _task to None, and _judge rebuilt from the same config.

Returns

TrajectoryJudge[AgentDepsT]

before_run

@async

def before_run(ctx: RunContext[AgentDepsT]) -> None

Reject a judged run inside a durable workflow or flow, before any budget is spent.

The evaluation is launched from a capability hook, so it would run in orchestration context: its model calls would not be checkpointed (and could repeat on replay, billing included) and enqueued steering would not persist across replay. A durable-capable agent run outside its workflow or flow is unaffected, matching how core’s durability capabilities scope their own before_run rejections.

Returns

None

after_model_request

@async

def after_model_request(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    response: ModelResponse,
) -> ModelResponse

Count the model request and launch an evaluation when the cadence is due.

A finished evaluation is reaped first, so a failure surfaces here rather than being silently dropped. The evaluation itself runs as a background task: the trajectory is rendered synchronously (no race with later mutation), the judge call and any steering enqueue happen concurrently with the run, and the steering is delivered when the run next drains its pending messages.

Returns

ModelResponse

wrap_run

@async

def wrap_run(
    ctx: RunContext[AgentDepsT],
    *,
    handler: WrapRunHandler,
) -> AgentRunResult[Any]

Run the agent, then settle the judge: surface a finished failure, cancel the rest.

An evaluation still in flight when the run ends is cancelled rather than awaited; its steering has nowhere to go. One that already finished with an error is re-raised so a judge failure is never silently dropped. When the run itself is failing, the evaluation’s outcome is discarded entirely so it cannot mask the run’s own error.

Returns

AgentRunResult[Any]

get_serialization_name

@classmethod

def get_serialization_name(cls) -> str | None

Not spec-serializable: the capability may hold a live Agent and a callback.

Returns

str | None

AllGood

The run is on track. No intervention is needed.

Steer

The run needs correction; the message is delivered to the running agent.

Attributes

message

The corrective guidance to deliver: short, specific, and actionable.

Type: str