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).

Interactive forms (e.g. WhatsApp Flows)

Some bots reply with a structured form rather than text — a WhatsApp Flow, say. The simulator can fill those in. When your adapter's send() returns a Delivered carrying one or more Forms, the simulator sees the fields as a human would, and on its next turn replies with a submission the harness parses and hands back to send() as a FormSubmission:

from sonar_eval import BaseBotAdapter, Delivered, Form, FormField


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

    async def send(self, session, user_msg, media=None, submission=None) -> Delivered:
        if submission is not None:
            # The user filled in a form. In real fidelity, turn this into the Flow
            # `nfm_reply` webhook WhatsApp would send; in mocked fidelity, call the
            # handler directly.
            await session.handle.submit_flow(submission.form, submission.values)
            return Delivered(messages=("Thanks — all set.",))
        # ...otherwise deliver a form to be filled in:
        form = Form(
            name="book_appointment",
            title="Book an appointment",
            fields=(
                FormField(name="date", type="date", required=True),
                FormField(name="time", type="choice", options=("9am", "10am"), required=True),
            ),
        )
        return Delivered(messages=("Pick a slot:",), forms=(form,))

The WhatsApp specifics live entirely in the adapter — the simulator only needs to see the form and produce field values. Keep those values in the scenario's user_instructions / persona / seed so submissions stay deterministic across the k trials (an LLM inventing values freely makes pass^k flaky). Adapters for bots that never send forms can omit the submission parameter.

Scripted user turns (bypassing the simulator LLM)

Sometimes you want an exact, reproducible conversation rather than an LLM improvising one — a regression test for a specific phrasing, or a scenario where the user turns must be identical across the k trials. Set scripted_messages on the scenario and the user simulator sends those strings verbatim, in order, without ever calling the user LLM:

scenario = ScenarioConfig(
    name="exact-refund-request",
    user_instructions="",  # ignored while scripted
    scripted_messages=[
        "I want a refund for order 1234",
        "Yes, the whole order",
    ],
)

• The messages are sent one per turn; the conversation ends (terminated=True) once the list is exhausted, so you don't need a termination signal.
• max_turns still caps the run, and a scripted message containing the termination_signal ends it early.
• Form submissions still work: a scripted message that is the submission JSON (see Interactive forms) is parsed and handed to send() as a FormSubmission, just like a simulated turn.
• user_instructions, persona, and user_llm are unused while scripted.

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.