sonar-eval

Evaluation harness SDK for agentic AI chatbots, built on sonar-tracing.

The harness drives your bot through simulated conversations, optionally injects faults into its tool calls, and grades the transcript it pulls back from sonar-ingest. You write one small adapter for your bot; the trial loop, fault injection, assertions, LLM grading, and pass@k / pass^k rollups are generic.

pip install sonar-eval

The mental model

conversation_id is the spine of the whole design. For each trial the harness:

1. mints a conversation_id via sonar_tracing.new_conversation() — this is the trial id, the fault-injection scope key, and the join key all at once;
2. hands it to your BotAdapter, which makes the bot trace under that id (in a real service, by POSTing it to an app-internal endpoint the app threads into tracing.identity(conversation_id=...));
3. lets a user simulator (an LLM playing the human) drive turns until it emits the termination signal or hits max_turns. The simulator sees only the messages delivered back to it — never the bot's tools or spans;
4. pulls the whole conversation back out of ingest by that id (GET /v1/conversations/{id}) and reconstructs the tool-call transcript;
5. runs programmatic assertions (tier 1) and an optional LLM rubric grader (tier 2), then repeats k times and rolls the results up per assertion.

The grader scores what actually happened — the span tree in Postgres — not whatever the adapter chose to return. That is why send() returns only delivered messages and the transcript comes from ingest.

What you provide

Two things are project-specific: an LLMClient (the harness never imports a vendor SDK) and a BotAdapter for your bot.

from sonar_eval import BaseBotAdapter, Delivered, LLMConfig, Message, Session, TrialContext


class MyLLM:
    """Wrap whatever model client you use. Return the completion text."""

    async def complete(self, config: LLMConfig, system: str, messages: list[Message]) -> str:
        ...  # call your provider with `system` + `messages`, return the text


class MyBotAdapter(BaseBotAdapter):
    name = "support-bot"

    async def setup_trial(self, ctx: TrialContext) -> Session:
        # Isolate/seed state for this trial and make the bot trace under
        # ctx.conversation_id. For a real service: POST ctx.conversation_id
        # (and, for real fidelity, serialize_faults(ctx.faults)) to your bot's
        # internal endpoint so it wraps its turns in identity(conversation_id=...).
        handle = await open_session(ctx.conversation_id, seed=ctx.seed.data)
        return Session(conversation_id=ctx.conversation_id, handle=handle)

    async def send(self, session: Session, user_msg: str, media=None) -> Delivered:
        reply = await session.handle.send(user_msg)   # drive one user turn
        return Delivered(messages=(reply,))            # only what the user sees

    async def teardown_trial(self, session: Session) -> None:
        await session.handle.close()                   # safe after a failed setup

Running a scenario

import asyncio

from sonar_eval import (
    Assertion, Fidelity, LLMConfig, Orchestrator, ScenarioConfig, Severity,
    TaskConfig, no_tool_errors, tool_was_called,
)

task = TaskConfig(
    ingest_base_url="http://localhost:4319",
    tenant="acme",
    user_llm=LLMConfig(model="claude-sonnet-5"),
    grader_llm=LLMConfig(model="claude-opus-5"),
    # project defaults to "eval-fullcontent" so the grader sees unredacted tool I/O
)

scenario = ScenarioConfig(
    name="order-a-coffee",
    user_instructions="You want a large oat-milk latte. Order it, then stop.",
    k=5,
    assertions=[
        # capability: passes if it held on ANY of the k trials (pass@k)
        Assertion("looked-up-menu", Severity.CAPABILITY, tool_was_called("menu_lookup")),
        # safety-critical: passes only if it held on EVERY trial (pass^k == 1.0)
        Assertion("no-tool-errors", Severity.SAFETY_CRITICAL, no_tool_errors()),
    ],
    grader_rubric="Did the bot confirm the exact order before charging? Score 0..1.",
)

orch = Orchestrator(task, MyBotAdapter(), MyLLM())
report = asyncio.run(orch.run_scenario(scenario))

print(report.passed)                       # True only if every assertion passed
for a in report.assertions:
    print(a.name, a.severity, a.pass_at_k, a.pass_hat_k)
print(report.mean_grader_score)            # None unless a grader ran

Built-in checks: tool_was_called(name), no_tool_errors(), max_turns(n), max_total_tokens(n). A check is any Callable[[Transcript, ConversationLog], bool], so you can write your own over the pulled transcript.

Grading is opt-in: pass a grader_factory to the orchestrator and set grader_rubric on the scenario.

from sonar_eval import Grader

orch = Orchestrator(
    task, MyBotAdapter(), MyLLM(),
    grader_factory=lambda s: Grader(MyLLM(), s.grader_llm, s.grader_rubric),
    store=None,   # defaults to NullStore; use JSONFileStore("results/") to persist
)

Fault injection

Faults are declared once and applied by whichever mechanism the scenario's fidelity selects. Rules are declarative so trials stay reproducible (pass^k needs determinism).

from sonar_eval import Action, FaultRule, FaultSpec

faults = FaultSpec(rules=(
    # the 2nd call to payment_api raises, every trial
    FaultRule(target="payment_api", action=Action.ERROR, when_call_index=1),
    # a content-aware fault (mocked fidelity only): fail if it tried to overcharge
    FaultRule(
        target="payment_api", action=Action.ERROR,
        when=lambda call: call.args.get("amount", 0) > 100,
    ),
))
scenario = ScenarioConfig(name="payment-outage", user_instructions="...", faults=faults)

• Fidelity.MOCKED (default): the adapter routes its mocked dependencies through a harness-owned ProxyInterceptor built from the spec. Fully in-process; the when content predicate works here.
• Fidelity.REAL: the adapter posts serialize_faults(ctx.faults) to your app's internal endpoint and the app's own tool dispatch honours it. A when predicate cannot cross a process boundary, so serialize_faults() raises on one rather than silently dropping it — use when_call_index for real-fidelity runs.

Actions: ERROR, TIMEOUT, REPLACE_OUTPUT (returns payload), LATENCY (sleeps payload seconds).

Metrics

Results are rolled up per assertion, not just per scenario, so a scenario can legitimately pass its capability bar while failing a safety gate:

• capability → pass@k — passes if the assertion held on at least one of k trials;
• safety-critical → pass^k — passes only if it held on every trial.

ScenarioReport.passed is true only when every assertion passes under its own rule. Records are written through a pluggable ResultStore (JSONFileStore / NullStore ship in the box).

App-side integration

For the harness to drive your bot, the bot must accept a conversation_id per conversation and thread it straight into identity(conversation_id=...) — and, to be drivable, expose a way for the simulator to hand it that id. See the "Grouping runs into a conversation" section of docs/IMPLEMENTATION_STRANDS.md / docs/IMPLEMENTATION_LANGCHAIN.md, and DESIGN.md §10 for the full rationale.

License

MIT — see LICENSE.