Trajectory Judge

이 문서에서는 TrajectoryJudge capability를 소개해요. 두 번째 모델로 실시간 에이전트 실행을 살펴보고 실행 중간에 다시 궤도로 이끌 수 있어요. 판사(judge)가 실행의 최근 궤적을 주기로 평가하고, 실행과 동시에 판단하므로 실행이 판사를 기다리며 차단되지 않아요.

출처: 문서

본문

두 번째 모델로 실시간 에이전트 실행을 살펴보고, 실행 중간에 다시 궤도로 이끌어요.

소스

Pydantic AI Harness가 0.x 릴리스인 동안에는 minor 릴리스 사이에 API가 바뀔 수 있어요. 그럴 때는 deprecation 경고와 릴리스 노트 마이그레이션 안내가 (당신이나 당신의 에이전트에게) 정확히 업그레이드 방법을 알려줘요. 버전 정책을 참고하세요.

The problem

장기간 실행은 표류해요. 지침이 퇴색하고, 근거 없는 주장이 누적되며, 에이전트가 목표에서 벗어나요. Guardrails는 고정 경계에서 정책 위반을 잡고, evals는 실행이 끝난 뒤(복구에 재실행 전체가 들 때) 점수를 매겨요. 어느 쪽도 궤적이 펼쳐지는 대로 보고 교정이 여전히 저렴할 때 개입하지 않아요.

The solution

TrajectoryJudge는 주기로 두 번째 모델로 실행의 최근 궤적을 검토해요. every개의 모델 요청마다 대화의 가장 최근 window 토큰(사용자 메시지, 어시스턴트 메시지, 도구 호출, 도구 결과를 트랜스크립트로 렌더링)이 판사 모델로 평가를 위해 전송돼요. 평가는 에이전트와 동시에 실행되므로 실행이 판사를 기다리며 차단되지 않아요.

판사는 출력 타입으로 정확히 평가당 하나의 판정을 전달해요:

AllGood — 실행이 궤도에 있음. 아무 일도 일어나지 않아요.
Steer(message=...) — 실행에 교정이 필요함. 메시지가 진행 중인 대화로 큐에 들어가고(RunContext.enqueue, 'asap' 우선순위) 다음 모델 요청에 판사 귀속으로 전달돼요: Steering from trajectory judge 'hallucination-check': ....

스티어링은 판사의 출력 이지, 그것이 호출하는 도구가 아니에요. 판사는 평가당 최종 판정 하나를 받는다는 것을 알므로, 사고 중에 반복적으로 스티어링하려는 유혹을 받지 않아요.

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는 실행 내의 모델 요청을 세고, 판사는 각 배수마다 평가해요.
window는 각 평가가 보는 것을 제한해요. 트랜스크립트가 가장 최근 window 토큰(토큰당 ~4자 추정)으로 클램프되어, 실행이 아무리 길어져도 평가당 비용이 제한돼요.
판사당 한 번에 최대 한 평가만 진행 돼요. 이전 평가가 여전히 진행 중임을 발견한 주기 tick은 건너뛰어져서, 느린 판사는 동시 호출이 쌓이는 대신 뒤처져요.
실행이 끝났을 때 아직 진행 중인 평가는 취소돼요. 그것의 스티어링이 갈 곳이 없으니까요.

Several judges

각 판사는 자체 capability 인스턴스예요. 관심사당 하나를 추가하세요. 독립적으로 스케줄링하고 평가하며, 각 스티어링 메시지는 자체 귀속(name, 판사 에이전트의 name, 또는 '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

모델과 검토 초점 너머의 것(모델 설정, 툴셋, 폴백 모델, 커스텀 지침)이 필요하면 capability에 손잡이를 쌓는 대신 완전한 Agent를 전달하세요. 각 평가는 그것을 output_type=[AllGood, Steer]로 실행하므로, 에이전트가 어떤 출력 타입으로 구성됐든 판정 계약이 실행 경계에서 강제돼요. 기존 에이전트는 의존성이 없을 때만 그대로 재사용될 수 있어요. 판사 의존성은 평가에 전달되지 않아요. 한 가지 제약: 판사 에이전트에는 output validator가 없어야 해요. 실행별 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는 model/instructions와 상호 배타적이에요. 전달된 에이전트는 자체 지침을 소유해요.

Cost and failure semantics

판사의 모델 사용량은 실행의 usage에 이어붙여지고 실행의 usage_limits를 존중해요. 각 실행은 평가가 시작되기 전에 공유 usage에서 요청 한 개를 청구하므로, 부모의 다음 요청과 동시 판사들이 진행 중인 평가를 계산하고 공유 요청 한도를 초과할 수 없어요. 요청 예산이 맞지 않는 실행은 tick을 건너뛰어요. 진행 중인 평가를 발견한 것처럼요.
평가 실패는 다음 주기 tick이나 실행 종료 시 실행에 발생돼요. 판사 실패는 절대 조용히 버려지지 않아요. 판사가 대신 저하되길 원한다면 agent를 통해 폴백 모델을 주세요(예: FallbackModel). 복원력 정책은 capability의 필드가 아니라 판사 에이전트의 몫이에요.
durable execution 워크플로우나 플로우(Temporal, DBOS, Prefect) 안의 판단된 실행은 첫 모델 요청 전에 UserError로 거부돼요. 평가가 오케스트레이션 컨텍스트의 capability 훅에서 실행되므로, 그 모델 호출은 체크포인트되지 않고 재생 시 반복될 수 있어요. 워크플로우나 플로우 밖의 durable-capable 에이전트 실행은 영향받지 않아요. 판단된 작업은 durable execution 밖에서 실행하세요.

Observability

on_verdict는 각 판정이 처리된 후(스티어링이 큐에 들어간 후) 호출돼요:

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는 규칙 기반의 형제예요. 추가 모델 호출 없는 주기 또는 조건 트리거 리마인더. 먼저 그것을 쓰세요. 조건이 궤적을 실제로 이해할 것을 요구할 때 판사가 그 비용을 정당화해요.
Guardrails는 입력, 도구 호출, 출력의 강제 경계로 남아요. 판사는 관찰하고 스티어링할 뿐, 아무것도 차단하거나 다시 쓰지 않아요.

Not spec-serializable

TrajectoryJudge.get_serialization_name()은 None을 반환해요. capability가 라이브 Agent 인스턴스와 콜백을 지닐 수 있으며, agent spec으로 직렬화할 수 없어요.

API reference

TrajectoryJudge

Bases: AbstractCapability[AgentDepsT]

주기로 두 번째 모델로 라이브 실행을 검토하고 실행 중간에 스티어링.

장기간 실행은 표류해요. 지침이 퇴색하고, 근거 없는 주장이 누적되며, 에이전트가 목표에서 벗어나요. TrajectoryJudge는 실행과 동시에 every개의 모델 요청마다 실행 궤적의 가장 최근 window 토큰을 평가하고, 평가당 정확히 하나의 판정을 전달해요. AllGood, 또는 교정 메시지를 담은 Steer. 스티어링은 실행에 큐에 들어가고(RunContext.enqueue, 'asap' 우선순위) 판사에 귀속되므로, 실행 에이전트는 복구가 여전히 저렴할 때 코스수정해요.

판사당 한 번에 최대 한 평가만 진행 돼요. 진행 중인 것을 발견한 주기 tick은 건너뛰어져요. 실행이 끝났을 때 아직 진행 중인 평가는 취소돼요. 판사의 모델·도구 사용량은 실행의 usage에 이어붙여지고 그 usage_limits를 존중해요. 각 실행은 평가가 시작되기 전에 공유 usage에서 요청 한 개를 청구하므로, 부모의 다음 preflight와 형제 실행들은 진행 중인 호출을 계산하고, 요청 예산이 맞지 않는 실행은 건너뛰어져요. 평가 실패는 다음 주기 tick이나 실행 종료 시 실행에 발생돼요. 대신 저하시키려면 판사에게 폴백 모델을 주세요(agent로).

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,
        )
    ],
)

여러 판사가 한 실행을 관찰할 수 있어요. 관심사당 하나의 TrajectoryJudge를 capabilities에 추가하세요. 각각 독립적으로 스케줄링하고 평가해요.

durable 워크플로우나 플로우(Temporal, DBOS, Prefect) 안의 판단된 실행은 첫 모델 요청 전에 UserError로 거부돼요. 평가가 capability 훅에서 실행되므로 그 모델 호출은 체크포인트되지 않고 재생 시 반복될 수 있어요. 판단된 작업은 durable execution 밖에서 실행하세요.

Attributes

model

궤적을 평가하는 모델. 이것(선택적 instructions 포함) 또는 완전한 agent, 둘 다는 아니에요.

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

instructions

판사의 검토 초점. 내장 판사 지침에 추가돼요. model과 함께만 유효하며, 전달된 agent는 자체 지침을 소유해요.

Type: str | None Default: None

agent

고급 커스터마이제이션(자체 지침, 모델 설정, 툴셋, 폴백 모델)을 위한 완전한 판사 에이전트. 각 평가는 자체 구성된 출력 타입과 무관하게 output_type=[AllGood, Steer]로 실행하므로 기존 에이전트를 그대로 재사용할 수 있어요. output validator가 없어야 해요. 실행별 output_type과 호환되지 않으니까요. model/instructions와 상호 배타적.

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

every

실행 내의 N개 모델 요청마다 평가.

Type: int Default: 10

window

슬라이딩 토큰 창. 각 평가는 가장 최근 궤적의 최대 이 토큰 수(토큰당 ~4자 추정)를 트랜스크립트로 렌더링해 봐요.

Type: int Default: 20000

name

스티어링 메시지 귀속에 사용되는 이름. 판사 agent의 name이 전달되면 기본값은 그것, 그 다음 'trajectory-judge'.

Type: str | None Default: None

on_verdict

각 판정이 처리된 후(스티어링이 큐에 들어간 후) 호출되는 선택적 관측 콜백.

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

Methods

for_run

@async

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

단계 수와 진행 중인 평가가 공유되지 않도록 새 실행별 인스턴스를 반환.

replace는 __init__과 __post_init__을 다시 실행해 init=False 필드를 리셋해요. _steps를 0으로, _task를 None으로, _judge를 같은 구성에서 재구성.

Returns

TrajectoryJudge[AgentDepsT]

before_run

@async

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

예산이 소요되기 전에 durable 워크플로우나 플로우 안의 판단된 실행을 거부.

평가가 capability 훅에서 실행되므로 오케스트레이션 컨텍스트에서 실행돼요. 그 모델 호출은 체크포인트되지 않고(재생 시 반복될 수 있고 청구 포함) 큐에 넣은 스티어링이 재생을 넘어 영속되지 않아요. 워크플로우나 플로우 밖의 durable-capable 에이전트 실행은 core의 durability capability가 자체 before_run 거부를 범위 지정하는 방식과 일치해 영향받지 않아요.

Returns

None

after_model_request

@async

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

모델 요청을 세고 주기가 되면 평가를 실행.

완료된 평가가 먼저 수확되므로, 실패는 조용히 버려지는 대신 여기서 표면화돼요. 평가 자체는 백그라운드 태스크로 실행돼요. 궤적은 동기로 렌더링되고(이후 변경과 경쟁 없음), 판사 호출과 스티어링 큐는 실행과 동시에 일어나며, 스티어링은 실행이 다음에 보류 중 메시지를 배수할 때 전달돼요.

Returns

ModelResponse

wrap_run

@async

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

에이전트를 실행한 다음 판사를 정리해요. 완료된 실패를 표면화하고 나머지를 취소.

실행이 끝났을 때 아직 진행 중인 평가는 기다리는 대신 취소돼요. 그것의 스티어링이 갈 곳이 없으니까요. 오류로 이미 완료된 것은 재발생되어 판사 실패가 절대 조용히 버려지지 않아요. 실행 자체가 실패하고 있을 때는 평가의 결과가 완전히 버려져 실행 자체의 오류를 가릴 수 없게 해요.

Returns

AgentRunResult[Any]

get_serialization_name

@classmethod

def get_serialization_name(cls) -> str | None

spec 직렬화 불가: capability가 라이브 Agent와 콜백을 지닐 수 있음.

Returns

str | None

AllGood

실행이 궤도에 있음. 개입이 필요 없음.

Steer

실행에 교정이 필요하며, 메시지가 실행 에이전트에 전달됨.

Attributes

message

전달할 교정 안내: 짧고, 구체적이며, 실행 가능하게.

Type: str

더 알아보기 (Learn more)

pydantic-ai-harness 문서

Trajectory Judge

Trajectory Judge

본문

The problem

The solution

Cadence and window

Several judges

Advanced: bring your own judge agent

Cost and failure semantics

Observability

Composition

Not spec-serializable

Further reading

API reference

TrajectoryJudge

Attributes

model

instructions

agent

every

window

name

on_verdict

Methods

for_run

Returns

before_run

Returns

after_model_request

Returns

wrap_run

Returns

get_serialization_name

Returns

AllGood

Steer

Attributes

message

더 알아보기 (Learn more)