Evaluate a Deployed Agent over HTTP

1
from nemo_evaluator_sdk.enums import AgentFormat
2
from nemo_evaluator_sdk.values import GenericAgent
3
4
agent = GenericAgent(
5
    name="my-http-agent",
6
    url="http://127.0.0.1:8000/agent",
7
    format=AgentFormat.GENERIC,
8
    body={"message": "{{ instruction }}"},  # -> {"message": "What is the capital of France?"}
9
    response_path="$.answer",               # reads {"answer": "..."} from the response
10
    # api_key_secret="MY_AGENT_TOKEN",      # optional: env var (local run) or platform secret (job), sent as a bearer token
11
)

1
import json
2
import threading
3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
4
5
6
class _AgentHandler(BaseHTTPRequestHandler):
7
    def do_POST(self) -> None:
8
        payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
9
        text = str(payload.get("message", "")).lower()
10
        answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
11
        body = json.dumps({"answer": answer}).encode()
12
        self.send_response(200)
13
        self.send_header("Content-Type", "application/json")
14
        self.end_headers()
15
        self.wfile.write(body)
16
17
    def log_message(self, format: str, *args: object) -> None:
18
        return  # keep the demo output quiet

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

1
from functools import partial
2
3
from nemo_evaluator_sdk import GenericAgent, SecretRef
4
from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
5
from nemo_evaluator_sdk.agent_inference import make_agent_inference_fn
6
7
agent = GenericAgent(
8
    name="hermes-agent",
9
    url="http://127.0.0.1:8642/v1/responses",
10
    api_key_secret=SecretRef("HERMES_TOKEN"),
11
    body={
12
        "model": "hermes-agent",
13
        "input": "{{ instruction }}",
14
        "stream": True,
15
        "store": False,
16
    },
17
    response_path="$..text",
18
    stream=True,
19
)
20
21
evaluator = AgentEvaluator(
22
    agent_inference_fn_factory=partial(
23
        make_agent_inference_fn,
24
        # Copy HermesStreamTranslator from the complete example linked above.
25
        stream_translator=HermesStreamTranslator(),
26
        capture_evidence=True,
27
    ),
28
    default_headers={"Accept": "text/event-stream"},
29
)

1
import asyncio
2
import json
3
import threading
4
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
5
6
from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
7
from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask
8
from nemo_evaluator_sdk.enums import AgentFormat
9
from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
10
from nemo_evaluator_sdk.values import GenericAgent
11
12
13
class KeywordMatchMetric:
14
    @property
15
    def type(self) -> str:
16
        return "keyword_match"
17
18
    def output_spec(self) -> list[MetricOutputSpec]:
19
        return [MetricOutputSpec.continuous_score("score")]
20
21
    async def compute_scores(self, input: MetricInput) -> MetricResult:
22
        expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
23
        answer = (input.candidate.output_text or "").lower()
24
        return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])
25
26
27
def make_tasks() -> list[AgentEvalTask]:
28
    return [
29
        AgentEvalTask(id="capital-france", intent="Answer the geography question.",
30
                      inputs={"instruction": "What is the capital of France?"},
31
                      reference={"expected": "Paris"}, metrics=[KeywordMatchMetric()]),
32
        AgentEvalTask(id="capital-japan", intent="Answer the geography question.",
33
                      inputs={"instruction": "What is the capital of Japan?"},
34
                      reference={"expected": "Tokyo"}, metrics=[KeywordMatchMetric()]),
35
    ]
36
37
38
# A stand-in "deployed agent": a tiny local HTTP server. Replace with your real agent's URL.
39
class _AgentHandler(BaseHTTPRequestHandler):
40
    def do_POST(self) -> None:
41
        payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
42
        text = str(payload.get("message", "")).lower()
43
        answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
44
        body = json.dumps({"answer": answer}).encode()
45
        self.send_response(200)
46
        self.send_header("Content-Type", "application/json")
47
        self.end_headers()
48
        self.wfile.write(body)
49
50
    def log_message(self, format: str, *args: object) -> None:
51
        return
52
53
54
async def main() -> None:
55
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _AgentHandler)
56
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
57
    host, port = httpd.server_address[0], httpd.server_address[1]
58
59
    agent = GenericAgent(
60
        name="my-http-agent",
61
        url=f"http://{host}:{port}/agent",
62
        format=AgentFormat.GENERIC,
63
        body={"message": "{{ instruction }}"},
64
        response_path="$.answer",
65
    )
66
    try:
67
        result = await AgentEvaluator().run(
68
            tasks=make_tasks(),
69
            target=agent,
70
            config=AgentEvalRunConfig(parallelism=2),
71
        )
72
    finally:
73
        httpd.shutdown()
74
75
    for aggregate in result.summary.scores.scores:
76
        print(f"{aggregate.name}: {aggregate.mean}")
77
    for trial in result.trials:
78
        output_text = trial.output.output_text if trial.output else None
79
        print(f"  {trial.task_id}: {output_text!r}")
80
81
82
asyncio.run(main())