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
3
4class KeywordMatchMetric:
5 @property
6 def type(self) -> str:
7 return "keyword_match"
8
9 def output_spec(self) -> list[MetricOutputSpec]:
10 return [MetricOutputSpec.continuous_score("score")]
11
12 async def compute_scores(self, input: MetricInput) -> MetricResult:
13 expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
14 answer = (input.candidate.output_text or "").lower()
15 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 — for the trajectory, an Agent Trajectory Interchange Format (ATIF) trace of the agent’s steps. evidence.trace(...) gives a handle whose tool_calls() returns the 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
4
5class UsedExpectedToolMetric:
6 """Score whether the agent's trajectory used a given tool."""
7
8 def __init__(self, expected_tool: str) -> None:
9 self._expected = expected_tool
10
11 @property
12 def type(self) -> str:
13 return "used_expected_tool"
14
15 def output_spec(self) -> list[MetricOutputSpec]:
16 return [MetricOutputSpec.continuous_score("tool_use")]
17
18 async def compute_scores(self, input: MetricInput) -> MetricResult:
19 used = False
20 evidence = input.candidate.evidence
21 if evidence is not None and evidence.get(EVIDENCE_TRACE) is not None:
22 calls = await (await evidence.trace(EVIDENCE_TRACE)).tool_calls()
23 used = any(call.function_name == self._expected for call in calls)
24 return MetricResult(outputs=[MetricOutput(name="tool_use", value=1.0 if used else 0.0)])

The metric checks for evidence first — a trial without a trace simply scores 0.0 rather than erroring.

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
12
13def _trace(tool_name: str, query: str) -> CandidateEvidence:
14 trajectory = Trajectory(
15 schema_version="ATIF-v1.7",
16 steps=[Step(source="agent", tool_calls=[ToolCall(function_name=tool_name, arguments={"query": query})])],
17 )
18 return CandidateEvidence(
19 descriptors={
20 EVIDENCE_TRACE: EvidenceDescriptor(
21 kind="trace", format=EVIDENCE_FORMAT_ATIF, data=trajectory.model_dump(mode="json")
22 )
23 }
24 )
25
26
27async def my_agent(task: AgentEvalTask) -> TrialDraft:
28 answers = {
29 "capital-france": ("The capital of France is Paris.", "search"), # used the expected tool
30 "capital-japan": ("The capital of Japan is Tokyo.", None), # right answer, but skipped the tool
31 }
32 text, tool = answers[task.id]
33 evidence = _trace(tool, task.inputs["instruction"]) if tool else None
34 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
3
4def _quality_view() -> SemanticView:
5 return SemanticView(
6 reducer=SemanticReducer.MEAN,
7 signals=[
8 ViewSignal(metric="keyword_match", output="score"),
9 ViewSignal(metric="used_expected_tool", output="tool_use"),
10 ],
11 )
12
13
14def make_tasks() -> list[AgentEvalTask]:
15 metrics = [KeywordMatchMetric(), UsedExpectedToolMetric("search")]
16 return [
17 AgentEvalTask(
18 id="capital-france", intent="Answer the geography question.",
19 inputs={"instruction": "What is the capital of France?"}, reference={"expected": "Paris"},
20 metrics=metrics, views={"quality": _quality_view()},
21 ),
22 AgentEvalTask(
23 id="capital-japan", intent="Answer the geography question.",
24 inputs={"instruction": "What is the capital of Japan?"}, reference={"expected": "Tokyo"},
25 metrics=metrics, views={"quality": _quality_view()},
26 ),
27 ]

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

Next steps