계약 테스트와 운영

capture_observability()를 사용하면 운영 자격 증명이나 Langfuse 서버 없이 type, tags, metadata, 부모 관계, score 대상을 검증할 수 있습니다.

CLI 선택 예제로 시작

새 Flow를 만들 때 관측·평가 예제까지 포함하려면 다음 옵션을 추가합니다.

uv run llamon flow my-flow \
  --template flow-seq \
  --example observability \
  --yes

이 옵션은 app/nodes.py의 업무 로직을 수정하지 않습니다. app/graph.py의 종료 노드에 evaluators=[output_completeness, output_ready]를 연결하고, app/evaluation.py에는 짧은 커스텀 evaluator와 SDK 기본 evaluator를 생성합니다. 전체 계약은 Runtime Evaluator Framework를 참고하세요.

기록	Langfuse 대상
evaluation:output_completeness	평가 대상 Flow 노드의 하위 observation
evaluation:output_ready	평가 대상 Flow 노드의 하위 observation
output_completeness, output_ready	평가 대상 Flow 노드의 Numeric score

실제 Flow를 반복 호출하려면 서버를 실행한 뒤 다음 명령을 사용합니다. 고객지원 케이스 3개를 각각 세 번 호출하며, 요청마다 새 contextId를 사용합니다. 한 호출이 실패해도 남은 케이스를 계속 진행하고 마지막에 성공/실패 요약을 출력합니다(실패가 있으면 종료 코드 1). 이 실행기는 케이스별 비교를 위해 선택 요청 metadata인 evaluationSuite와 evaluationCase를 추가합니다. 일반 호출이나 Runtime Evaluator 자체에는 두 값이 필요하지 않습니다.

uv run python scripts/run_evaluation_cases.py --repeat 3

Langfuse에서 eval:response-quality tag로 실행을 좁힙니다. trace의 evaluationSuite·evaluationCase·workflowVariant와 evaluator observation의 policyVariant를 기준으로 output_completeness 값을 비교합니다. 같은 비교를 터미널에서 보려면 생성된 scripts/compare_scores.py를 Langfuse Public API 자격 증명(LANGFUSE_PUBLIC_KEY· LANGFUSE_SECRET_KEY·LANGFUSE_BASE_URL)이 있는 환경 변수 파일로 실행합니다. evaluationSuite·케이스·워크플로우 구성별 runs/mean/min/max 표가 출력됩니다.

uv run --env-file .env python scripts/compare_scores.py --suite customer-support-v1

기본 점수는 텍스트·설명·구조화 데이터의 존재 여부를 조합한 0~1 값입니다. 정확성이나 근거성을 평가하려면 output_completeness()를 도메인 규칙으로 바꾸거나 RubricEvaluator·GroundednessEvaluator를 연결하세요. 커스텀 기준을 바꾸면 decorator의 version도 함께 올립니다. 생성된 계약 테스트는 네트워크 없이 두 응답의 NUMERIC 점수를 비교합니다.

uv run --with pytest pytest -q

네트워크 없는 계약 테스트

아래 예제는 observe()와 record_score()로 직접 기록하는 저수준 계약을 검증합니다. 앞 절의 생성 예제가 사용하는 Runtime Evaluator 계약과는 observation·score 위치가 다릅니다.

from app.config import build_observability
from llamon_agent.observability import observe, record_score
from llamon_agent.testing import capture_observability


def test_document_verification_observability_contract():
    with capture_observability(
        observability=build_observability(),
    ) as capture:
        with observe(
            "normalize-document",
            observation_type="chain",
            metadata={"stage": "normalization"},
        ):
            normalize_document()

        with observe(
            "contract-check",
            observation_type="evaluator",
            metadata={"stage": "validation", "policyVariant": "contract-v3"},
        ):
            validate_contract()
            record_score(
                "policy_compliance",
                0.91,  # 테스트 값입니다. 실제 실행에서는 평가 결과를 사용합니다.
                data_type="NUMERIC",
                target="observation",
            )

    normalization = capture.one_observation(name="normalize-document")
    assert normalization.observation_type == "chain"
    assert normalization.metadata["stage"] == "normalization"

    validation = capture.one_observation(name="contract-check")
    assert validation.observation_type == "evaluator"

    score = capture.one_score(
        name="policy_compliance",
        observation_id=validation.observation_id,
    )
    assert score.value == 0.91

Capture 레코드

capture.observations, capture.scores, capture.events는 불변 레코드 튜플입니다. 각 속성은 내부 capture 상태와 분리된 페이로드 스냅샷을 반환합니다. 테스트 코드가 반환된 metadata, input, output을 수정해도 이후 검증 결과는 달라지지 않습니다.

CapturedObservation.tags는 각 observation에 실제로 전파될 tag를 담습니다. Langfuse v4는 observation에 tag를 적용하고, trace에는 하위 observation의 tag 집합을 모읍니다.

• find_observations()와 find_scores()는 여러 결과를 찾습니다.
• one_observation()과 one_score()는 정확히 하나의 결과를 기대할 때 사용합니다.
• snapshot(normalize_ids=True)는 synthetic ID를 안정화한 golden snapshot을 만듭니다.

Capture 상태는 작업별로 분리됩니다. pytest를 중첩하거나 병렬로 실행해도 운영 백엔드 싱글턴을 바꾸지 않습니다. Orchestrator 테스트에서는 MockOrchestratorContext.call_observed()를 함께 사용하면 A2A 요청의 metadata를 오염시키지 않고 하위 observation 참조와 score 연결을 검증합니다.

Prompt cache usage

code-first, Registry, A2A와 Studio의 invoke/stream은 같은 task-local cache usage scope를 사용합니다. 중첩 실행에서도 provider usage event는 모든 활성 scope에 집계됩니다.

prompt cache usage: scope=a2a.execute stats={'total_input': 4210, 'cache_read': 3072, 'cache_write': 0, 'status': 'read', ...}

OpenAI와 Anthropic usage는 total_input, cache_read, cache_write로 정규화됩니다. read, write, miss는 provider가 반환한 token만으로 판정하며 prefix fingerprint가 같다는 이유로 hit를 추정하지 않습니다. request 관측에는 provider, model, mode, message 수, prefix 길이와 fingerprint만 남고 prompt 원문, binding 값과 cache key 원문은 남지 않습니다. code-first root trace에는 같은 집계가 cache_stats metadata로 연결됩니다.

provider breakpoint, TTL과 canary 기준은 프롬프트 권한 분리와 provider cache를 참고하세요.

ObservabilityConfig 입력 제약

• tag와 전파 metadata의 키, 값은 각각 최대 200자입니다.
• tag는 앞뒤 공백과 중복을 제거한 뒤 보존합니다.
• metadata 키의 snake_case와 kebab-case는 camelCase로 정규화합니다. 정규화 뒤 같은 키가 생기면 거부합니다.
• 전파 metadata 값에는 str, int, float, bool 스칼라만 허용하며 문자열로 정규화합니다.
• list, object, None은 observation 로컬 metadata에서만 사용합니다.
• 요청 또는 런타임 식별자를 나타내는 SDK 예약 key는 루트 설정에서 거부합니다.

안전한 운영

tag와 metadata에는 자격 증명, 원문 문서, PII, 고카디널리티 요청 ID를 넣지 마세요. TRACE_BACKEND=none이거나 백엔드 기록에 실패해도 관측·score API는 비즈니스 결과를 바꾸지 않습니다.

AgentCardBuilder.add_skill(tags=...)는 A2A discovery에 쓰는 필드입니다. Langfuse 분석용 tags로 자동 복사되지 않습니다. 두 곳에 같은 분류가 필요하면 개발자가 각각 명시합니다.