Lab 13 · The Northwind evaluation harness

langchain==1.4.2
langchain-core==1.6.4
langchain-openai==1.6.2
python-dotenv==1.2.3
langgraph==1.2.11
agentevals==0.0.9
openevals==0.2.0
langsmith==0.14.0

# Azure OpenAI settings. Your instructor gives you these three values.
AZURE_OPENAI_ENDPOINT=https://<your-resource>.openai.azure.com/openai/v1
AZURE_OPENAI_API_KEY=<paste-your-key-here>
AZURE_OPENAI_DEPLOYMENT=<your-chat-deployment-name>

# Only used when your endpoint does NOT end in /openai/v1.
OPENAI_API_VERSION=2025-04-01-preview

"""Azure OpenAI model setup shared by every lab.

Import it in lab code as:  from model_setup import get_model, get_embeddings
"""

import asyncio
import os
import sys

from dotenv import load_dotenv
from langchain_core.embeddings import Embeddings
from langchain_core.language_models import BaseChatModel

# Read AZURE_OPENAI_* values from the .env file next to your lab code.
load_dotenv()

# Windows consoles default to cp1252 and crash on the curly quotes models love to print.
for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, "reconfigure"):
        _stream.reconfigure(encoding="utf-8", errors="replace")

def _setting(name: str) -> str:
    """Return a required setting from .env, or stop with a readable message."""
    value = os.getenv(name, "").strip()
    if not value:
        sys.exit(f"Missing {name} in your .env file. Open .env and fill it in.")
    return value


def _uses_v1_api(endpoint: str) -> bool:
    """True for the OpenAI-compatible Azure endpoint that ends in /openai/v1."""
    return endpoint.rstrip("/").endswith("/openai/v1")

def get_model(deployment: str | None = None, **kwargs) -> BaseChatModel:
    """Return a chat model for your Azure OpenAI deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import ChatOpenAI

        return ChatOpenAI(model=name, base_url=endpoint, api_key=api_key, **kwargs)

    from langchain_openai import AzureChatOpenAI

    return AzureChatOpenAI(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
        **kwargs,
    )

def get_embeddings(deployment: str | None = None) -> Embeddings:
    """Return an embeddings model for your Azure OpenAI embeddings deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_EMBEDDING_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import OpenAIEmbeddings

        return OpenAIEmbeddings(model=name, base_url=endpoint, api_key=api_key)

    from langchain_openai import AzureOpenAIEmbeddings

    return AzureOpenAIEmbeddings(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
    )

def use_stable_event_loop() -> None:
    """On Windows, use the Selector event loop so async MCP code exits cleanly."""
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

"""A scripted fake chat model, so an agent can run offline and deterministically.

Usage:
    model = scripted_model([tool_call("get_order_status", {"order_id": "A-100"}), "It shipped."])
    agent = create_agent(model, tools=[get_order_status])
"""

import itertools
from collections.abc import Iterable, Sequence
from typing import Any

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, ToolCall
from langchain_core.runnables import Runnable

# Every scripted tool call needs a unique id, so the tool result can be matched to it.
_ids = itertools.count(1)

class ScriptedChatModel(GenericFakeChatModel):
    """A fake chat model that replays scripted messages and tolerates tool binding.

    `create_agent` calls `model.bind_tools(...)` before every model call. The base fake model
    raises NotImplementedError there, so we ignore the tools (the script already decides which
    tool calls happen) and return the model itself.
    """

    def bind_tools(self, tools: Sequence[Any], **kwargs: Any) -> Runnable:  # type: ignore[override]
        """Accept the tools and return self: the script, not the tools, drives the run."""
        return self

    @property
    def _llm_type(self) -> str:
        """Name used in traces and callbacks."""
        return "scripted-chat-model"

def tool_call(name: str, args: dict[str, Any] | None = None) -> AIMessage:
    """Build an AIMessage whose only content is a single tool call."""
    return AIMessage(content="", tool_calls=[ToolCall(name=name, args=args or {}, id=f"call_{next(_ids)}")])


def scripted_model(responses: Iterable[AIMessage | str]) -> ScriptedChatModel:
    """A fake model that replays `responses` in order, one per model call (strings become answers)."""
    messages = [AIMessage(content=r) if isinstance(r, str) else r for r in responses]
    return ScriptedChatModel(messages=iter(messages))

"""Lab 13: the Northwind evaluation harness and its regression gate.

Run:
    python eval_harness.py --offline
    python eval_harness.py --offline --save-baseline baseline.json
    python eval_harness.py --offline --variant candidate --baseline baseline.json
    python eval_harness.py --live 5
"""

import argparse
import json
import os
from collections import defaultdict
from collections.abc import Callable
from pathlib import Path
from typing import Any

from agentevals.trajectory.match import create_trajectory_match_evaluator

from fakes import scripted_model, tool_call
from model_setup import get_model
from support_agent import (
    as_trajectory,
    build_support_agent,
    called_tool_calls,
    final_text,
    find_pii,
    reference_trajectory,
)

MODULE_DIR = Path(__file__).resolve().parent
DATASET_PATH = MODULE_DIR / "datasets" / "northwind_support.jsonl"
DATASET_NAME = "northwind-support-v1"

def load_examples() -> list[dict[str, Any]]:
    """Read the JSONL dataset. A silently skipped row is a silently missing test."""
    rows = [json.loads(line) for line in DATASET_PATH.read_text(encoding="utf-8").splitlines() if line]
    for index, row in enumerate(rows):
        missing = {"inputs", "outputs", "metadata"} - row.keys()
        if missing:
            raise ValueError(f"{DATASET_PATH.name} line {index + 1} is missing {sorted(missing)}")
    return rows

def make_target(agent: Any) -> Callable[[dict[str, Any]], dict[str, Any]]:
    """Wrap an agent in the `(inputs: dict) -> dict` contract evaluators expect.

    Return everything an evaluator could need. What the target omits cannot be scored later
    without re-running the agent.
    """

    def target(inputs: dict[str, Any]) -> dict[str, Any]:
        """Run the agent on one question and collect answer, tool calls and trajectory."""
        result = agent.invoke({"messages": [{"role": "user", "content": inputs["question"]}]})
        return {
            "answer": final_text(result),
            "tool_calls": called_tool_calls(result),
            "messages": as_trajectory(result),
        }

    return target

def structural_match(inputs: dict, outputs: dict, reference_outputs: dict) -> dict[str, Any]:
    """Every required token is in the answer and no forbidden one is."""
    answer = outputs["answer"].lower()
    missing = [t for t in reference_outputs["must_contain"] if t.lower() not in answer]
    forbidden = [t for t in reference_outputs["must_not_contain"] if t.lower() in answer]
    ok = not missing and not forbidden
    return {"key": "structural_match", "score": 1.0 if ok else 0.0,
            "comment": "ok" if ok else f"missing={missing} forbidden={forbidden}"}

# `search_kb` may paraphrase its query and `escalate_to_human` may pick its own priority, but an
# order id must be exact. Overrides are how you say "strict here, loose there".
TRAJECTORY_EVALUATOR = create_trajectory_match_evaluator(
    trajectory_match_mode="unordered",
    tool_args_match_mode="exact",
    tool_args_match_overrides={"search_kb": "ignore", "escalate_to_human": ["reason"]},
)


def tool_trajectory(inputs: dict, outputs: dict, reference_outputs: dict) -> dict[str, Any]:
    """Did the agent call the right tools with the right arguments? Catches a lucky guess."""
    reference = reference_trajectory(
        inputs["question"], reference_outputs["expected_tool_calls"], reference_outputs["answer"]
    )
    result = TRAJECTORY_EVALUATOR(outputs=outputs["messages"], reference_outputs=reference)
    expected = [c["name"] for c in reference_outputs["expected_tool_calls"]]
    actual = [c["name"] for c in outputs["tool_calls"]]
    return {"key": "tool_trajectory", "score": 1.0 if result["score"] else 0.0,
            "comment": f"expected={expected} actual={actual}"}

def safety_pii(inputs: dict, outputs: dict, reference_outputs: dict) -> dict[str, Any]:
    """The agent must never echo personal data back to the customer."""
    leaked = find_pii(outputs["answer"])
    return {"key": "safety_pii", "score": 0.0 if leaked else 1.0,
            "comment": f"leaked {sorted(leaked)}" if leaked else "clean"}


def escalation_policy(inputs: dict, outputs: dict, reference_outputs: dict) -> dict[str, Any]:
    """Escalate exactly when policy says so. Scoring only the misses rewards escalating everything."""
    should = reference_outputs["should_escalate"]
    did = any(c["name"] == "escalate_to_human" for c in outputs["tool_calls"])
    return {"key": "escalation_policy", "score": 1.0 if should == did else 0.0,
            "comment": f"should_escalate={should} escalated={did}"}


CODE_EVALUATORS = [structural_match, tool_trajectory, safety_pii, escalation_policy]

def scripted_agent_for(example: dict[str, Any], variant: str) -> Any:
    """Real tools, real loop, fake model: deterministic and still representative.

    `baseline` replays the tool calls the dataset expects; `candidate` skips them and answers
    from memory, which is the regression the gate has to catch.
    """
    calls = example["outputs"]["expected_tool_calls"] if variant == "baseline" else []
    script = [tool_call(c["name"], c["args"]) for c in calls] + [example["outputs"]["answer"]]
    return build_support_agent(scripted_model(script))

def score_examples(
    examples: list[dict[str, Any]],
    agent_for: Callable[[dict[str, Any]], Any],
    variant: str,
    verbose: bool = False,
) -> dict[str, Any]:
    """Run every example through its agent and score it with every code evaluator."""
    scores: dict[str, list[float]] = defaultdict(list)
    failures: list[str] = []
    for number, example in enumerate(examples, start=1):
        outputs = make_target(agent_for(example))(example["inputs"])
        if verbose:
            tools = ", ".join(c["name"] for c in outputs["tool_calls"]) or "none"
            print(f"[{number}/{len(examples)}] {example['inputs']['question']}")
            print(f"      tools:  {tools}")
            print(f"      answer: {outputs['answer']}")
        for evaluator in CODE_EVALUATORS:
            result = evaluator(example["inputs"], outputs, example["outputs"])
            scores[result["key"]].append(result["score"])
            if result["score"] < 1.0:
                failures.append(f"{example['metadata']['intent']:<26}{result['key']:<18}"
                                f"{result['comment']}")
    return {
        "variant": variant,
        "n": len(examples),
        "scores": {key: round(sum(v) / len(v), 3) for key, v in scores.items()},
        "failures": failures,
    }


def run_offline(variant: str = "baseline") -> dict[str, Any]:
    """Score every dataset row with a scripted agent. No keys, no network, deterministic."""
    return score_examples(load_examples(), lambda example: scripted_agent_for(example, variant), variant)


def run_live(limit: int) -> dict[str, Any]:
    """Score the first `limit` rows with the real Azure OpenAI agent. Costs a few model calls."""
    agent = build_support_agent(get_model())
    return score_examples(load_examples()[:limit], lambda example: agent, "live", verbose=True)

def print_summary(report: dict[str, Any]) -> None:
    """Print the per-evaluator means and the first few failing rows."""
    print(f"\nvariant={report['variant']}  n={report['n']}")
    for key, score in sorted(report["scores"].items()):
        print(f"  {key:<20}{score:>7.3f}")
    for line in report["failures"][:5]:
        print(f"  ! {line}")
    if len(report["failures"]) > 5:
        print(f"  ! ... and {len(report['failures']) - 5} more")

def check_regression(report: dict[str, Any], baseline: dict[str, Any],
                     tolerance: float = 0.02) -> list[str]:
    """Name every metric that dropped by more than the tolerance.

    The tolerance is not sloppiness: judge scores wobble a couple of points between runs, and a
    gate that fires on noise gets switched off within a week.
    """
    return [
        f"{key}: {before:.3f} -> {report['scores'].get(key, 0.0):.3f}"
        for key, before in baseline["scores"].items()
        if report["scores"].get(key, 0.0) < before - tolerance
    ]

def make_correctness_judge() -> Callable[..., dict[str, Any]]:
    """The LLM judge, built lazily so importing this module never needs an API key."""
    from openevals.llm import create_llm_as_judge
    from openevals.prompts import CORRECTNESS_PROMPT

    judge = create_llm_as_judge(prompt=CORRECTNESS_PROMPT, judge=get_model(), feedback_key="correctness")

    def correctness(inputs: dict, outputs: dict, reference_outputs: dict) -> dict[str, Any]:
        """Is the answer consistent with the reference answer?"""
        return judge(inputs=inputs["question"], outputs=outputs["answer"],
                     reference_outputs=reference_outputs["answer"])

    return correctness


def run_langsmith() -> None:
    """Optional: run a LangSmith experiment with the real agent, the code evaluators and a judge."""
    if not os.environ.get("LANGSMITH_API_KEY"):
        print("LANGSMITH_API_KEY is not set - skipping the LangSmith experiment.")
        print("Use --offline (scripted, free) or --live N (real model, no LangSmith needed).")
        return

    from langsmith import Client

    client = Client()
    if not client.has_dataset(dataset_name=DATASET_NAME):
        dataset = client.create_dataset(DATASET_NAME, description="Northwind support agent.")
        client.create_examples(dataset_id=dataset.id, examples=load_examples())
        print(f"Uploaded {DATASET_PATH.name} as LangSmith dataset '{DATASET_NAME}'.")

    results = client.evaluate(
        make_target(build_support_agent(get_model())),
        data=DATASET_NAME,
        evaluators=[*CODE_EVALUATORS, make_correctness_judge()],
        experiment_prefix="northwind-support",
        max_concurrency=4,
    )
    print(f"Experiment: {results.experiment_name}\n{results.url}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Northwind support evaluation harness")
    parser.add_argument("--offline", action="store_true", help="score locally with scripted models")
    parser.add_argument("--live", type=int, metavar="N", help="score the first N rows with the real model")
    parser.add_argument("--variant", choices=["baseline", "candidate"], default="baseline")
    parser.add_argument("--save-baseline", type=Path, help="write this run's scores to a file")
    parser.add_argument("--baseline", type=Path, help="fail when a score drops below this file's")
    args = parser.parse_args()

    if not args.offline and not args.live:
        run_langsmith()
        raise SystemExit(0)

    report = run_live(args.live) if args.live else run_offline(args.variant)
    print_summary(report)

    if args.save_baseline:
        args.save_baseline.write_text(json.dumps(report, indent=2), encoding="utf-8")
        print(f"\nbaseline written to {args.save_baseline}")

    if args.baseline:
        regressions = check_regression(report, json.loads(args.baseline.read_text(encoding="utf-8")))
        for line in regressions:
            print(f"REGRESSION  {line}")
        print("\ngate: FAILED" if regressions else "\ngate: passed")
        raise SystemExit(1 if regressions else 0)

variant=baseline  n=30
  escalation_policy     1.000
  safety_pii            1.000
  structural_match      1.000
  tool_trajectory       1.000

variant=baseline  n=30
  escalation_policy     1.000
  safety_pii            1.000
  structural_match      1.000
  tool_trajectory       1.000

gate: passed

variant=baseline  n=30
  escalation_policy     1.000
  safety_pii            1.000
  structural_match      1.000
  tool_trajectory       1.000

baseline written to baseline.json

variant=candidate  n=30
  escalation_policy     0.900
  safety_pii            1.000
  structural_match      1.000
  tool_trajectory       0.167
  ! order_status              tool_trajectory   expected=['get_order_status'] actual=[]
  ! order_status              tool_trajectory   expected=['get_order_status'] actual=[]
  ! order_status              tool_trajectory   expected=['get_order_status'] actual=[]
  ! order_status              tool_trajectory   expected=['get_order_status'] actual=[]
  ! order_status              tool_trajectory   expected=['get_order_status', 'get_order_status'] actual=[]
  ! ... and 23 more
REGRESSION  tool_trajectory: 1.000 -> 0.167
REGRESSION  escalation_policy: 1.000 -> 0.900

gate: FAILED

[1/5] Where is my order A-100?
      tools:  get_order_status
      answer: Your order A-100 has been shipped via DHL with an estimated delivery date of 2026-09-15. Would you like me to fetch the DHL tracking number or check the latest delivery updates?
[2/5] Has order A-101 shipped yet?
      tools:  get_order_status
      answer: Order A-101 hasn't shipped yet; it's currently processing. The estimated arrival date is 2026-09-18 — I can notify you when it ships or check carrier details if you'd like.
[3/5] did order a 102 ever arrive??
      tools:  get_order_status
      answer: Yes — order A-102 was delivered on 2026-08-20 via UPS. If you haven’t received it, please check with neighbors or your building/concierge and let me know. I can open an investigation with our claims team for you if you’d like.
[4/5] I never got anything for order #A-103, what happened?
      tools:  get_order_status
      answer: I checked your order: A-103 was cancelled and was never charged. That means nothing was shipped and no refund is required since no payment was taken. Would you like me to investigate why it was cancelled or help place a replacement order?
[5/5] Can you check both A-100 and A-104 for me?
      tools:  get_order_status, get_order_status
      answer: Order A-100 is shipped via DHL with an ETA of 2026-09-15. Order A-104 is shipped via UPS with an ETA of 2026-09-14. Would you like tracking links or help with anything else for these orders?

variant=live  n=5
  escalation_policy     1.000
  safety_pii            1.000
  structural_match      1.000
  tool_trajectory       1.000