Score by Component

View as Markdown

The quickstart scored one thing: whether the final answer contained the right keyword. But an agent can reach the right answer the wrong way — guessing instead of looking something up, or looping on a tool. Agent evaluation lets you score how the agent worked, not just the outcome, and combine several signals into one reported view.

This guide extends the quickstart. You’ll add a trajectory metric that reads the agent’s tool calls, keep the quickstart’s outcome metric, and combine them into a quality view. It stays zero-dependency — one local process, no API keys.

This builds on the quickstart’s KeywordMatchMetric and tasks. The full runnable script is at the end.

1. The outcome metric

Reuse the quickstart’s KeywordMatchMetric — it scores the final answer, reading it from input.candidate.output_text and the grader-only truth from input.row.data:

1from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
2
3class KeywordMatchMetric:
4 @property
5 def type(self) -> str:
6 return "keyword_match"
7
8 def output_spec(self) -> list[MetricOutputSpec]:
9 return [MetricOutputSpec.continuous_score("score")]
10
11 async def compute_scores(self, input: MetricInput) -> MetricResult:
12 expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
13 answer = (input.candidate.output_text or "").lower()
14 return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])

2. A trajectory metric

A metric can read more than the final output. input.candidate.evidence exposes the trial’s evidence. This guide produces an ATIF trace, whose handle returns modeled tool calls in order. Each ToolCall has a function_name and arguments. This metric scores 1.0 when the agent used the tool you expected and 0.0 otherwise:

1from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
2from nemo_evaluator_sdk.values.evidence import EVIDENCE_TRACE
3
4class UsedExpectedToolMetric:
5 """Score whether the agent's trajectory used a given tool."""
6
7 def __init__(self, expected_tool: str) -> None:
8 self._expected = expected_tool
9
10 @property
11 def type(self) -> str:
12 return "used_expected_tool"
13
14 def output_spec(self) -> list[MetricOutputSpec]:
15 return [MetricOutputSpec.continuous_score("tool_use")]
16
17 async def compute_scores(self, input: MetricInput) -> MetricResult:
18 evidence = input.candidate.evidence
19 if evidence is None:
20 return MetricResult(outputs=[MetricOutput(name="tool_use", value=0.0)])
21 try:
22 handle = await evidence.trace(EVIDENCE_TRACE, format="atif")
23 except KeyError:
24 return MetricResult(outputs=[MetricOutput(name="tool_use", value=0.0)])
25 calls = await handle.tool_calls()
26 used = any(call.function_name == self._expected for call in calls)
27 return MetricResult(outputs=[MetricOutput(name="tool_use", value=1.0 if used else 0.0)])

Asking for format="atif" raises KeyError when the trial recorded no ATIF trace, which this metric scores the same as missing evidence: 0.0. Harbor makes OTLP primary, so a trial that recorded only an OTLP trace scores 0.0 here rather than failing — read that score as “this metric could not see a trajectory,” not as proof the tool went unused. An OTLP metric asks for format="otlp" and inspects resource_spans() with exporter-specific logic.

3. Produce a trajectory

Runners that execute the agent capture this evidence for you — a deployed HTTP agent returns a trajectory when you set trajectory_path, and container/harness backends record one as the agent runs. (A Harbor run scores a verifier reward instead.) Here the agent is a local callable, so it returns a TrialDraft — its final output plus, when it used a tool, a small ATIF trace built from the SDK’s trajectory models. One task’s agent uses the search tool; the other skips it and just guesses, returning no trace at all:

1from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import TrialDraft
2from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask
3from nemo_evaluator_sdk.agent_eval.trials import AgentOutput
4from nemo_evaluator_sdk.values.atif import Step, ToolCall, Trajectory
5from nemo_evaluator_sdk.values.evidence import (
6 EVIDENCE_FORMAT_ATIF,
7 EVIDENCE_TRACE,
8 CandidateEvidence,
9 EvidenceDescriptor,
10)
11
12def _trace(tool_name: str, query: str) -> CandidateEvidence:
13 trajectory = Trajectory(
14 schema_version="ATIF-v1.7",
15 steps=[Step(source="agent", tool_calls=[ToolCall(function_name=tool_name, arguments={"query": query})])],
16 )
17 return CandidateEvidence(
18 descriptors={
19 EVIDENCE_TRACE: EvidenceDescriptor(
20 kind="trace", format=EVIDENCE_FORMAT_ATIF, data=trajectory.model_dump(mode="json")
21 )
22 }
23 )
24
25async def my_agent(task: AgentEvalTask) -> TrialDraft:
26 answers = {
27 "capital-france": ("The capital of France is Paris.", "search"), # used the expected tool
28 "capital-japan": ("The capital of Japan is Tokyo.", None), # right answer, but skipped the tool
29 }
30 text, tool = answers[task.id]
31 evidence = _trace(tool, task.inputs["instruction"]) if tool else None
32 return TrialDraft(output=AgentOutput(output_text=text), evidence=evidence)

4. Combine the signals into a view

Attach both metrics to each task, then define a view — a named roll-up of this task’s metric outputs. SemanticView reduces its signals (each a metric.output) into one score; MEAN averages them. The view is reported per task and aggregated across the run as view.<name>.

1from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask, SemanticReducer, SemanticView, ViewSignal
2
3def _quality_view() -> SemanticView:
4 return SemanticView(
5 reducer=SemanticReducer.MEAN,
6 signals=[
7 ViewSignal(metric="keyword_match", output="score"),
8 ViewSignal(metric="used_expected_tool", output="tool_use"),
9 ],
10 )
11
12def make_tasks() -> list[AgentEvalTask]:
13 metrics = [KeywordMatchMetric(), UsedExpectedToolMetric("search")]
14 return [
15 AgentEvalTask(
16 id="capital-france", intent="Answer the geography question.",
17 inputs={"instruction": "What is the capital of France?"}, reference={"expected": "Paris"},
18 metrics=metrics, views={"quality": _quality_view()},
19 ),
20 AgentEvalTask(
21 id="capital-japan", intent="Answer the geography question.",
22 inputs={"instruction": "What is the capital of Japan?"}, reference={"expected": "Tokyo"},
23 metrics=metrics, views={"quality": _quality_view()},
24 ),
25 ]

5. Run it and read the components

1from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
2from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import CallableAgentTaskRunner
3from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig
4
5async def run_evaluation() -> None:
6 result = await AgentEvaluator().run(
7 tasks=make_tasks(),
8 target=CallableAgentTaskRunner(my_agent),
9 config=AgentEvalRunConfig(parallelism=2),
10 )
11
12 for aggregate in result.summary.scores.scores:
13 print(f"{aggregate.name}: {aggregate.mean}")

Output:

keyword_match.score: 1.0
used_expected_tool.tool_use: 0.5
view.quality: 0.75

This is the whole point of scoring by component. The outcome looks perfect — every answer was correct (1.0). But the trajectory metric shows only half the agents actually used the expected tool (0.5); the other got the right answer by guessing. The view combines the two into a single quality score (0.75) you can track over time. Scoring only the answer would have hidden the shortcut entirely.

Full script

1import asyncio
2
3from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
4from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import CallableAgentTaskRunner, TrialDraft
5from nemo_evaluator_sdk.agent_eval.tasks import (
6 AgentEvalRunConfig,
7 AgentEvalTask,
8 SemanticReducer,
9 SemanticView,
10 ViewSignal,
11)
12from nemo_evaluator_sdk.agent_eval.trials import AgentOutput
13from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
14from nemo_evaluator_sdk.values.atif import Step, ToolCall, Trajectory
15from nemo_evaluator_sdk.values.evidence import (
16 EVIDENCE_FORMAT_ATIF,
17 EVIDENCE_TRACE,
18 CandidateEvidence,
19 EvidenceDescriptor,
20)
21
22class KeywordMatchMetric:
23 @property
24 def type(self) -> str:
25 return "keyword_match"
26
27 def output_spec(self) -> list[MetricOutputSpec]:
28 return [MetricOutputSpec.continuous_score("score")]
29
30 async def compute_scores(self, input: MetricInput) -> MetricResult:
31 expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
32 answer = (input.candidate.output_text or "").lower()
33 return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])
34
35class UsedExpectedToolMetric:
36 """Score whether the agent's trajectory used a given tool."""
37
38 def __init__(self, expected_tool: str) -> None:
39 self._expected = expected_tool
40
41 @property
42 def type(self) -> str:
43 return "used_expected_tool"
44
45 def output_spec(self) -> list[MetricOutputSpec]:
46 return [MetricOutputSpec.continuous_score("tool_use")]
47
48 async def compute_scores(self, input: MetricInput) -> MetricResult:
49 evidence = input.candidate.evidence
50 if evidence is None:
51 return MetricResult(outputs=[MetricOutput(name="tool_use", value=0.0)])
52 try:
53 handle = await evidence.trace(EVIDENCE_TRACE, format="atif")
54 except KeyError:
55 return MetricResult(outputs=[MetricOutput(name="tool_use", value=0.0)])
56 calls = await handle.tool_calls()
57 used = any(call.function_name == self._expected for call in calls)
58 return MetricResult(outputs=[MetricOutput(name="tool_use", value=1.0 if used else 0.0)])
59
60def _trace(tool_name: str, query: str) -> CandidateEvidence:
61 trajectory = Trajectory(
62 schema_version="ATIF-v1.7",
63 steps=[Step(source="agent", tool_calls=[ToolCall(function_name=tool_name, arguments={"query": query})])],
64 )
65 return CandidateEvidence(
66 descriptors={
67 EVIDENCE_TRACE: EvidenceDescriptor(
68 kind="trace", format=EVIDENCE_FORMAT_ATIF, data=trajectory.model_dump(mode="json")
69 )
70 }
71 )
72
73def _quality_view() -> SemanticView:
74 return SemanticView(
75 reducer=SemanticReducer.MEAN,
76 signals=[
77 ViewSignal(metric="keyword_match", output="score"),
78 ViewSignal(metric="used_expected_tool", output="tool_use"),
79 ],
80 )
81
82def make_tasks() -> list[AgentEvalTask]:
83 metrics = [KeywordMatchMetric(), UsedExpectedToolMetric("search")]
84 return [
85 AgentEvalTask(
86 id="capital-france", intent="Answer the geography question.",
87 inputs={"instruction": "What is the capital of France?"}, reference={"expected": "Paris"},
88 metrics=metrics, views={"quality": _quality_view()},
89 ),
90 AgentEvalTask(
91 id="capital-japan", intent="Answer the geography question.",
92 inputs={"instruction": "What is the capital of Japan?"}, reference={"expected": "Tokyo"},
93 metrics=metrics, views={"quality": _quality_view()},
94 ),
95 ]
96
97async def my_agent(task: AgentEvalTask) -> TrialDraft:
98 answers = {
99 "capital-france": ("The capital of France is Paris.", "search"), # used the expected tool
100 "capital-japan": ("The capital of Japan is Tokyo.", None), # right answer, but skipped the tool
101 }
102 text, tool = answers[task.id]
103 evidence = _trace(tool, task.inputs["instruction"]) if tool else None
104 return TrialDraft(output=AgentOutput(output_text=text), evidence=evidence)
105
106async def main() -> None:
107 result = await AgentEvaluator().run(
108 tasks=make_tasks(),
109 target=CallableAgentTaskRunner(my_agent),
110 config=AgentEvalRunConfig(parallelism=2),
111 )
112 for aggregate in result.summary.scores.scores:
113 print(f"{aggregate.name}: {aggregate.mean}")
114
115asyncio.run(main())

Next steps