Runtime Evaluator Framework

Runtime Evaluator Framework v1은 한 번의 completed 출력을 실행 중에 평가하고 0~1 Numeric score를 연결합니다. 기본 evaluator와 짧은 Python 함수를 같은 목록에서 조합할 수 있습니다.

평가 케이스(행), 워크플로우 구성(열), evaluator 정책과 score의 위치는 Observability 개요와 공통 설정에서 한눈에 확인할 수 있습니다.

이 기능은 정답 응답, 실행 궤적, 다중 턴 시뮬레이션, 오프라인 데이터셋 실행기를 포함하는 종합 평가 시스템이 아닙니다. 그런 평가 케이스를 만들 때도 여기의 evaluator를 재사용할 수 있지만, v1이 제공하는 범위는 단일 출력의 런타임 평가입니다.

Scaffold starter

새 프로젝트에서 바로 시작하려면 Agent·Flow·Orchestrator 생성 명령에 --example을 붙입니다. Evaluation은 별도 프로젝트 타입이 아니며 기본값은 starter 없음입니다.

starter	기본 evaluator	추가 동작	필요한 설정
observability	output_completeness, output_ready	없음. 네트워크 없는 baseline	없음
rubric-judge	baseline 2개	answer_quality 정확성·명확성 rubric	Registry JUDGE_MODEL_ID 또는 local JUDGE_MODEL
groundedness	baseline 2개	입력 documents를 쓰는 groundedness	Registry JUDGE_MODEL_ID 또는 local JUDGE_MODEL

uv run llamon agent support --template agent-general --example rubric-judge --yes
uv run llamon flow retrieval --template flow-seq --example groundedness --yes
uv run llamon orch supervisor --workflow deterministic \
  --agent primary=8101 --example observability --yes

생성물은 app/evaluation.py, 반복 A2A 호출 스크립트, Langfuse score 비교 스크립트와 네트워크 없는 계약 테스트를 포함합니다. 의미 평가 starter의 rubric·evidence adapter는 업무 계약에 맞게 수정하세요. ORCH verification 전용 outcome-evaluator Agent에는 이 starter를 중첩하지 않습니다.

가장 짧은 사용법

app/evaluation.py

from llamon_agent.evaluation import EvaluationContext, OutputReadyEvaluator, evaluator


@evaluator("conciseness", threshold=0.7, version="v1")
def conciseness(context: EvaluationContext) -> float:
    chars = len(context.output_text)
    return 1.0 if 1 <= chars <= 500 else 0.0


output_ready = OutputReadyEvaluator()

Agent에서는 evaluator를 create_server()에 연결합니다.

main.py

from app.evaluation import conciseness, output_ready

app = await create_server(
    card=card,
    agent=agent,
    evaluators=[conciseness, output_ready],
    evaluator_timeout_seconds=10.0,
)

Flow 노드에는 함수 호출이나 래퍼 대신 evaluator 객체를 연결합니다.

app/graph.py

from app.evaluation import conciseness, output_ready

builder.node(
    "answer",
    answer_node,
    evaluators=[conciseness, output_ready],
    evaluator_timeout_seconds=10.0,
)

Agent에서는 최종 completed 응답을 평가하고, Flow에서는 선택한 노드 출력을 평가합니다. 업무 함수는 평가를 모르며 원래 출력만 반환합니다. Flow에서는 SDK가 대상 노드 observation 안에서 evaluation:conciseness와 evaluation:output_ready evaluator observation을 실행하고, 두 Numeric score를 대상 노드에 연결합니다. Agent·Orchestrator의 최종 응답 score는 trace에 연결됩니다.

기록 구조 한눈에 보기

Runtime Evaluator는 평가 과정을 evaluation:<evaluator-name> evaluator observation으로 남기되, score는 평가받은 원래 대상에 연결합니다. evaluator 이름과 Numeric score 이름은 같습니다. 아래는 대상이 열려 있는 Flow 노드에서 실행한 경우입니다.

평가 대상 observation  business-validate-document, metadata={stage: validation}
        ├─ 평가 과정    Evaluator evaluation:policy_compliance
        │                metadata={policyVariant: contract-v3, threshold: 0.8}
        │                output={passStatus: true, reason: ...}
        └─ score        policy_compliance=0.96  ← 원래 평가 대상에 연결
                         metadata={policyVariant: contract-v3, threshold: 0.8, passStatus: true}

값	기록 위치
stage 같은 업무 단계	평가 대상 observation의 metadata
policyVariant, threshold, 평가 대상 ID	evaluation observation의 metadata
reason, details, passStatus	evaluation observation의 output
점수와 policyVariant, threshold, passStatus	원래 평가 대상에 연결된 Numeric score

evaluation observation 자체를 점수 대상으로 오해하거나, 대상의 stage가 그 metadata에 자동 복사된다고 가정하면 안 됩니다. 종료된 observation을 명시적으로 평가하면 evaluation observation의 부모가 대상과 다를 수 있습니다. 이때도 evaluatedObservationId가 관계를 보존하고 score는 지정한 대상에 연결됩니다. evaluation observation과 score를 직접 만들어야 한다면 Score 기록을 사용하세요.

커스텀 함수 반환 계약

@evaluator 함수는 동기와 비동기를 모두 지원합니다. 동기 함수는 이벤트 루프를 막지 않도록 워커 스레드에서 실행됩니다.

반환값	의미
float	0.0~1.0 점수
bool	True → 1.0, False → 0.0
None	정상적인 미평가. score를 기록하지 않음
EvaluationResult	score, reason, details를 함께 반환

반환값이 bool이면 숫자 범위를 검사하기 전에 변환합니다. int 반환값, bool 타입의 threshold, NaN, 무한대, 범위 밖 값은 거부합니다. 데코레이터가 만든 객체의 evaluate()는 언제나 EvaluationResult를 반환합니다.

from llamon_agent.evaluation import EvaluationResult, evaluator


@evaluator("citation_coverage", threshold=0.8, version="v2")
def citation_coverage(context):
    citations = context.output_data[0].get("citations", []) if context.output_data else []
    if not context.output_text:
        return None
    return EvaluationResult(
        score=min(len(citations) / 3, 1.0),
        reason="expected up to three citations",
        details={"citationCount": len(citations)},
    )

reason과 details는 observation에 저장하기 전에 JSON 직렬화가 가능하고 크기가 제한된 값으로 정규화됩니다. 객체의 표현 문자열이나 예외 메시지를 그대로 trace에 남기지 않습니다.

데코레이터 인자는 다음과 같습니다.

인자	기본값	계약
name	필수	evaluator와 Numeric score의 이름
threshold	0.5	score >= threshold이면 passStatus=true
version	"v1"	평가 정책이 바뀔 때 올리는 정책 식별자
judge_llm	None	커스텀 evaluator에 EvaluationJudge를 주입할 judge 모델

LLM을 사용하는 커스텀 evaluator

위의 citation_coverage처럼 기준을 코드로 계산할 때는 judge_llm과 judge 인자가 필요 없습니다. 동의어와 문맥까지 판단해야 할 때만 LLM judge를 선택하세요.

방식	데코레이터	함수
결정론적 규칙	judge_llm 생략	동기 또는 비동기, 결과를 직접 반환
LLM 판정	judge_llm=...	비동기, 키워드 전용 judge 사용

공유할 모델 설정은 app/config.py에 둡니다.

app/config.py

from llamon_agent.config import LLMConfig, resolve_env_override


JUDGE_MODEL_ID = "evaluation-judge"

judge_llm = LLMConfig(
    id=resolve_env_override("JUDGE_MODEL_ID", JUDGE_MODEL_ID, source_file=__file__),
    temperature=0.0,
)

평가 정책은 설정을 가져온 뒤 데코레이터에 주입합니다.

app/evaluation.py

from app.config import judge_llm
from llamon_agent.evaluation import (
    EvaluationContext,
    EvaluationJudge,
    EvaluationResult,
    evaluator,
)


@evaluator(
    "context_relevance",
    threshold=0.5,
    version="semantic-v1",
    judge_llm=judge_llm,
)
async def context_relevance(
    context: EvaluationContext,
    *,
    judge: EvaluationJudge,
) -> EvaluationResult:
    documents = list(context.output_data)
    if not documents:
        return EvaluationResult(
            score=0.0,
            reason="no documents retrieved",
            details={"documentCount": 0},
        )
    return await judge.score(
        criteria="Judge how directly and sufficiently the documents answer the question.",
        input={
            "question": context.input_text,
            "documents": documents,
        },
    )

judge.score()는 score, reason, details가 담긴 JSON을 요구하고 필드 형식과 score 범위를 검증합니다. 함수는 passStatus를 만들지 않습니다. SDK가 threshold와 score를 비교해 evaluation observation과 Numeric score metadata에 같은 값을 기록합니다. score=None이나 평가 오류이면 passStatus=null이며 Numeric score는 만들지 않습니다.

criteria가 별도 눈금을 정하지 않으면 SDK는 0.0을 기준 미충족, 0.5를 부분 충족, 1.0을 완전 충족으로 해석하며 기준점 사이의 값도 허용합니다. 일반적인 criteria에는 정확성·관련성처럼 평가할 의미만 적으면 됩니다. 도메인 전용 눈금이 필요하면 점수별 의미를 criteria에 명시하세요. threshold는 통과 판정용이므로 criteria에 넣지 않습니다.

judge_llm이 있는 함수는 비동기여야 하며 judge를 키워드 전용 인자로 명시해야 합니다. Registry ID 조회, 동시 최초 호출 준비, 모델 정책 식별자 계산은 기본 LLM evaluator와 같은 경로를 사용합니다.

input의 프로퍼티 이름은 애플리케이션이 정합니다. 문자열 키와 JSON 직렬화가 가능한 값이면 중첩 매핑과 시퀀스도 전달할 수 있습니다. SDK는 매핑 전체를 JSON으로 직렬화해 공통 judge 프롬프트에 넣지만, 그 원문을 evaluation observation에 자동 저장하지는 않습니다. 추적에 남길 값은 EvaluationResult.details에 선별해서 반환하세요.

기본 눈금은 연속 점수를 허용하지만 정밀한 소수점이 곧 판정 정밀도를 뜻하지는 않습니다. 대표 평가 세트로 분포와 threshold를 함께 보정하세요.

통과 상태

통과 여부는 evaluator 결과와 threshold에서 계산합니다.

from llamon_agent.evaluation import pass_status

status = pass_status(result, threshold=0.8)  # True | False | None

score=None의 상태도 None입니다. 평가 실패나 미평가를 threshold 미달 False와 섞지 마세요.

EvaluationContext와 출력 투영

EvaluationContext는 원본을 보존하면서 공통 채널을 제공합니다.

필드	내용
input, output	커스텀 evaluator가 읽을 원본 객체
input_text, output_text	정규화된 텍스트. 공백뿐이면 빈 값
input_data, output_data	DataPart 성격의 매핑 튜플
input_files, output_files	FilePart 성격의 매핑 튜플

다음 출력은 같은 방식으로 투영됩니다.

• RuntimeOutput
• AgentCallResult의 text, data_parts, files
• ObservedAgentCallResult의 업무 결과와 observation 참조
• output_text·output_data·output_files를 가진 dict 또는 중첩 output dict

알 수 없는 객체를 str(value)로 바꾸지 않습니다. 투영 규칙이 없는 타입은 채널이 빈 상태로 남고 원본은 context.output에서만 접근할 수 있습니다.

기본 evaluator

OutputReadyEvaluator

text, data, files 중 하나라도 비어 있지 않으면 1.0입니다.

from llamon_agent.evaluation import OutputReadyEvaluator

output_ready = OutputReadyEvaluator()  # name="output_ready", threshold=1.0

RubricEvaluator

여러 rubric 항목을 judge가 각각 0~1로 평가한 뒤 같은 가중치로 평균냅니다.

from app.config import judge_llm
from llamon_agent.evaluation import RubricEvaluator

quality = RubricEvaluator(
    name="answer_quality",
    rubric={
        "correct": "The answer is supported by the supplied input.",
        "clear": "The answer is concise and actionable.",
    },
    judge_llm=judge_llm,
    threshold=0.8,
)

빈 rubric은 생성 시 거부합니다. judge 결과의 ID가 요청한 ID와 정확히 일치하지 않거나 score 스키마가 잘못되면 일부 항목만 채점하지 않고 전체 평가를 오류로 처리합니다.

GroundednessEvaluator

Groundedness는 정적 evidence와 동기·비동기 어댑터를 모두 받습니다.

from app.config import judge_llm
from llamon_agent.evaluation import EvidenceItem, GroundednessEvaluator


async def retrieved_evidence(context):
    documents = context.input.get("documents", [])
    return [
        EvidenceItem(id=doc["id"], text=doc["text"])
        for doc in documents
    ]


groundedness = GroundednessEvaluator(
    evidence=retrieved_evidence,
    judge_llm=judge_llm,
    threshold=0.8,
    reason_language="ko",
)

RubricEvaluator와 GroundednessEvaluator도 같은 judge_llm 이름을 사용합니다. 하나의 LLMConfig를 여러 evaluator가 공유할 수 있고, 정책별 모델이 필요하면 설정 객체를 나눕니다. 기존 코드의 llm=은 호환 별칭으로 계속 동작하지만 새 코드에서는 judge_llm=을 사용하세요.

reason_language에는 ko, en, en-US 같은 BCP 47 언어 태그를 지정합니다. 이 옵션은 judge가 생성하는 최상위 reason의 언어만 정하며 JSON 키와 enum 값은 바꾸지 않습니다. 생략하면 언어를 강제하지 않습니다. 값은 프롬프트 정책과 policyVariant에 반영됩니다.

Judge의 thinking 설정은 Registry provider 이름이 아니라 실제 엔드포인트 규약에 맞춥니다.

실제 엔드포인트	LLMConfig.provider_extra
Ollama OpenAI 호환 /v1 프록시	{"reasoning_effort": "none"}
Moonshot 공식 API의 Kimi K2.6	{"thinking": {"type": "disabled"}}

reasoning_content가 없더라도 프록시가 필드를 숨겼을 수 있습니다. 최종 content가 짧은데 output_tokens와 지연 시간이 크다면 thinking이 실제로 꺼졌는지 확인하세요. 자세한 내용은 reasoning 모드 제어를 참고하세요.

evidence 입력은 str, EvidenceItem, 이들의 시퀀스, 또는 Callable[[EvaluationContext], ...]입니다. 문자열 하나를 문자 시퀀스로 나누지 않습니다. 빈 evidence는 평가 불가 상태이므로 judge를 호출하지 않고 score=None, status="not_evaluated", reasonCode="evidence_empty"로 기록합니다. 어댑터 실행 실패, 지원하지 않는 형식, 빈 ID·본문, 중복 ID는 EvidenceUnavailableError로 남습니다.

judge는 응답 모드를 answered, abstained, non_answer, empty로 구분합니다.

• 사실 주장이 있으면 supported / factual claims
• 실제 답변 유보이고 evidence가 답하기에 불충분하면 1.0
• 답할 evidence가 있는데 답변을 유보하면 0.0
• 사실 주장이 없는 답변 회피는 0.0
• 쓸 수 있는 evidence가 있는데 출력이 비어 있으면 0.0
• 알 수 없는 evidence ID, 서로 충돌하는 응답 모드처럼 판정할 수 없는 judge 출력은 오류

인사말·제목·형식 문장은 사실 주장 분모에서 제외합니다. 모순되는 주장도 사실 주장에는 포함되지만 뒷받침되지 않으므로 점수를 낮춥니다.

직접 실행과 오류 정책

런타임 통합 없이도 프레임워크를 직접 호출할 수 있습니다.

from llamon_agent.evaluation import evaluate

results = await evaluate(
    output,
    input=request,
    using=[output_ready, quality],
    on_error="raise",
    timeout_seconds=120.0,
)

quality_result = results["answer_quality"]

반환형은 dict[str, EvaluationResult]이며 using 순서를 보존합니다. 중복 이름은 evaluator 실행과 observation 생성을 시작하기 전에 거부합니다.

on_error	동작
raise	단일 오류는 원래 예외, 복수 오류는 EvaluationBatchError
record	실패한 evaluator를 score=None 결과로 바꾸고 나머지 결과도 반환

EvaluationBatchError.failures는 (evaluator_name, exception) 튜플을 실행 목록 순서로 보존합니다. 병렬 evaluator는 모두 회수된 뒤 예외를 냅니다. 외부 취소는 record 결과로 바꾸지 않습니다. 함께 실행 중인 작업을 취소·회수한 뒤 취소 신호를 다시 전파합니다.

evaluate()의 timeout_seconds와 자동 평가 API의 evaluator_timeout_seconds 기본값은 evaluator마다 90초입니다. 여러 evaluator가 병렬로 실행되므로 전체 목록에 한 번 적용되는 시간이 아니라 각 evaluator에 독립적으로 적용됩니다. None은 제한을 없애지만, 자동 평가는 응답을 내보내기 전에 실행되므로 completed event가 무기한 늦어질 수 있습니다.

Agent와 Orchestrator 최종 출력

Agent 서버의 모든 completed 성공 경로에 같은 evaluator를 적용할 수 있습니다.

from llamon_agent import create_server

app = await create_server(
    card=card,
    agent=agent,
    evaluators=[output_ready, quality],
    evaluator_timeout_seconds=120.0,
)

Orchestrator도 동일합니다.

from llamon_agent.orchestrator.server import build_orchestrator_app

app = await build_orchestrator_app(
    card=card,
    run_turn=run_turn,
    evaluators=[groundedness],
    evaluator_timeout_seconds=120.0,
)

자동 평가는 completed 결과에만 적용합니다. input_required와 failed 결과는 평가하지 않습니다. HITL 재개, 스트리밍, dispatch, invoke 경로도 최종 성공 출력이 확정된 한 지점을 통과한 뒤 정확히 한 번 평가됩니다. 자동 통합의 오류 정책은 record이므로 judge 장애가 업무 응답을 실패로 바꾸지 않습니다.

점수 대상과 observation 구조

score 대상은 evaluator 작업을 만들기 전에 다음 우선순위로 캡처합니다.

1. evaluate(..., observation=...)의 명시적 observation
2. ObservedAgentCallResult.observation
3. Flow 노드나 하위 래퍼가 캡처한 런타임 대상
4. 활성 trace
5. 관측 컨텍스트가 없으면 기록 없이 결과만 반환

따라서 evaluation observation이 실수로 자기 자신에게 score를 붙이지 않습니다. score=None일 때도 evaluation observation은 남지만 score는 만들지 않습니다.

Flow evaluator span은 대상 노드 span이 닫히기 전에 그 아래 생성됩니다. 이미 종료된 Orchestrator 하위 호출을 평가하면 백엔드 제약 때문에 evaluation span은 현재 Orchestrator의 형제 observation으로 남을 수 있습니다. 이때 score는 종료된 하위 호출에 정확히 연결됩니다. evaluation metadata의 evaluatedObservationId가 두 observation의 관계를 보존합니다.

Registry 모델과 policyVariant

Registry ID를 쓰는 judge evaluator만 있는 Flow·Agent도 Registry 의존성이 있는 것으로 인식됩니다. 첫 호출이 동시에 여러 번 들어와도 모델은 한 번만 조회해 캐시합니다.

내부 순서는 다음과 같습니다.

validate → resolve/prepare → policy variant 확정
         → evaluation observation → judge 실행

Registry 스냅샷 키의 범위는 evaluator 이름과 조회 전 설정의 digest로 나뉩니다. 최종 policyVariant는 조회된 모델의 안전한 허용 목록 필드, rubric, 알고리즘 버전으로 계산합니다. API 키, 임의 객체의 표현 문자열, URL 자격 증명은 digest에 넣지 않습니다.

Studio에서 편집

Studio에서는 Flow 캔버스의 노드를 선택한 뒤 오른쪽 노드 설정 패널 → Runtime 평가에서 evaluator를 연결합니다. 목록에서 evaluator를 추가하고 위·아래 버튼으로 결과·오류 정렬 순서를 바꾸며, evaluator별 제한 시간을 양의 초 단위로 입력하거나 제한 없음을 선택할 수 있습니다. 여러 evaluator는 병렬로 실행됩니다. 이 정렬은 부수 효과나 호출 제한을 제어하지 않으며 실행 순서를 보장하지 않습니다. 제한 시간 필드명은 evaluator_timeout_seconds이고 기본값은 90.0입니다. 저장하면 evaluator 목록, 순서, 제한 시간이 graph IR과 app/graph.py에 함께 반영됩니다.

평가 탭의 Evaluator 정책에서는 app/evaluation.py 원문을 직접 편집할 수 있습니다. 저장 전에 Python 문법을 검사하고, 불러온 파일의 checksum과 현재 파일을 비교해 Studio 밖의 변경을 덮어쓰지 않습니다. 저장은 원자적으로 수행되고 기존 파일은 Studio 변경 이력에 남으므로 복원할 수 있습니다. 문법 오류나 파일 충돌이 발생하면 디스크를 바꾸지 않고 편집 중인 초안을 유지합니다.

오른쪽의 저장된 카탈로그는 app/evaluation.py를 가져오지 않고 AST로 정적 분석해 만듭니다. 따라서 목록을 표시하거나 저장하는 과정에서 프로젝트 코드, 모델 클라이언트, 비밀값 로더가 실행되지 않습니다. 카탈로그에는 참조 이름, evaluator 종류, score 이름, threshold, version, judge 사용 여부, 선언 위치가 표시됩니다. 리터럴이 아닌 동적 메타데이터는 값을 추측하지 않고 정적 확인 불가로 표시합니다. 파일을 Studio 밖에서 수정했다면 평가 정책 다시 불러오기로 원문과 카탈로그를 갱신하세요.

Studio에서 편집할 evaluator 구현은 app.evaluation의 최상위 이름으로 선언합니다.

from app.evaluation import conciseness, output_ready

@evaluator가 붙은 함수와 OutputReadyEvaluator(...)·GroundednessEvaluator(...) 같은 최상위 *Evaluator 객체가 목록에 나타납니다. 기존 graph에 연결됐지만 현재 카탈로그에서 찾지 못한 이름은 코드에서 유지됨으로 표시해 자동 삭제하지 않습니다. 정책을 저장하면 새 카탈로그가 노드의 Runtime 평가 선택 목록에도 즉시 반영됩니다.

최근 Playground 점수는 로컬 Playground 실행 기록에서 evaluator별 표본 수, 평균·최근 점수, pass/fail/판정 없음 수, 최근 노드·소요 시간·사유를 요약합니다. 이 패널은 새 평가를 실행하거나 외부 관측 백엔드를 조회하지 않으며, 기록에 명시적 pass/fail 판정이 없으면 점수만 보고 임의로 합격 여부를 계산하지 않습니다.

Evaluator 정책 편집기는 app/evaluation.py 전체를 사용자가 작성한 그대로 저장합니다. rubric, judge 모델, evidence 어댑터도 이 파일 안에서 직접 편집하며, Studio가 의미를 추론해 코드를 생성하거나 재작성하지 않습니다. 노드 연결 저장은 별도의 안전한 AST 패치 경로를 사용합니다.

GroundednessEvaluator(...) 같은 인라인 생성자나 다른 모듈의 표현식을 app/graph.py에 직접 넣으면 저장을 막는 진단 오류가 발생하며, 이때 기존 graph.py를 덮어쓰지 않습니다.

처음부터 실행되는 커스텀 evaluator와 SDK 기본 evaluator가 필요하면 Agent 또는 Flow 스캐폴드의 선택 예제를 사용합니다.

uv run llamon agent my-agent --template agent-general --example observability --yes
uv run llamon flow my-flow --template flow-seq --example observability --yes

두 명령 모두 app/evaluation.py에 짧은 @evaluator 함수와 OutputReadyEvaluator를 만듭니다. Agent 예제는 create_server()에서 최종 응답을 평가하고, Flow 예제는 종료 노드에 evaluator를 연결합니다. Flow에서 만든 두 evaluator는 Studio의 Runtime 평가 목록에도 표시됩니다.

일반 Agent 예제는 별도의 runtime_adapter.py를 만들지 않으며 기존 어댑터도 교체하지 않습니다. evaluator는 최종 응답을 점수화할 뿐, 업무 응답의 변환 계약에는 관여하지 않습니다.

outcome-evaluator는 ORCH verification 요청을 pass·revise·review로 판정하는 전용 Agent이므로 --example observability를 적용하지 않습니다. 일반 Agent의 점수 평가와 ORCH verification 판정은 계약과 운영 목적이 다릅니다.

저수준 record_score()로 Categorical·TEXT나 사후 score를 직접 연결해야 한다면 Score 기록을 참고하세요. 반복 실행 필터와 비교 축은 Langfuse 필터·비교에 설명되어 있습니다.