Langchain Eval Harness

# evals/golden_set/v2026.04.jsonl
{"id": "gs-0001", "input": "Refund policy for SKU ABC-42?", "expected": "30 days with receipt", "contexts": ["policy_v3.md"], "tags": ["refund"], "difficulty": "easy", "dataset_version": "2026.04"}
{"id": "gs-0002", "input": "Return policy for opened software?", "expected": "No, opened software is final sale", "contexts": ["policy_v3.md#returns"], "tags": ["refund"], "difficulty": "medium", "dataset_version": "2026.04"}

from langsmith import Client
from langsmith.evaluation import evaluate, EvaluationResult
from langchain_anthropic import ChatAnthropic

client = Client()
DATASET_VERSION = "2026.04"

# One-time: upload golden set as a versioned dataset
def upload_golden_set(jsonl_path, dataset_name):
    examples = [json.loads(line) for line in open(jsonl_path)]
    client.create_dataset(dataset_name)
    client.create_examples(
        inputs=[{"input": e["input"]} for e in examples],
        outputs=[{"expected": e["expected"]} for e in examples],
        metadata=[{"id": e["id"], "tags": e["tags"]} for e in examples],
        dataset_name=dataset_name,
    )

chain = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, timeout=30)

def target(inputs):
    return {"answer": chain.invoke(inputs["input"]).content}

def correctness(outputs, reference_outputs):
    """Deterministic exact-match floor — baseline, not ceiling."""
    match = outputs["answer"].strip().lower() == reference_outputs["expected"].strip().lower()
    return EvaluationResult(key="exact_match", score=float(match))

results = evaluate(
    target,
    data=f"golden-set-v{DATASET_VERSION}",
    evaluators=[correctness],
    experiment_prefix="refund-bot-v3",
    max_concurrency=10,   # Avoid 429s on judge LLM (P22)
)

from ragas import evaluate as ragas_evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from datasets import Dataset

judge = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embed = OpenAIEmbeddings(model="text-embedding-3-small")

# Prepare rows — ragas wants HuggingFace Dataset shape
rows = []
for ex in golden_examples:
    result = rag_chain.invoke({"question": ex["input"]})
    rows.append({
        "question": ex["input"],
        "answer": result["answer"],
        "contexts": [d.page_content for d in result["source_documents"]],
        "ground_truth": ex["expected"],
    })

ragas_results = ragas_evaluate(
    Dataset.from_list(rows),
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=judge,
    embeddings=embed,
)
# ragas_results is a dict of per-metric means; call .to_pandas() for per-row

import statistics
from deepeval.metrics import GEval
from deepeval.test_case import LLMTestCase, LLMTestCaseParams

def eval_with_quorum(test_case, metric, n=3):
    scores = []
    for _ in range(n):
        metric.measure(test_case)
        scores.append(metric.score)
    return statistics.median(scores), statistics.stdev(scores) if n > 1 else 0.0

correctness = GEval(
    name="Correctness",
    criteria="Does the actual output match the expected output in meaning?",
    evaluation_params=[LLMTestCaseParams.INPUT, LLMTestCaseParams.ACTUAL_OUTPUT, LLMTestCaseParams.EXPECTED_OUTPUT],
    model="gpt-4o-mini",
)

for ex in golden_examples:
    result = chain.invoke({"input": ex["input"]})
    case = LLMTestCase(input=ex["input"], actual_output=result, expected_output=ex["expected"])
    median, sd = eval_with_quorum(case, correctness, n=3)
    if sd > 0.2:  # judge disagreeing with itself — flag, don't gate
        flag_for_review(ex["id"], median, sd)

from langchain_core.messages import AIMessage

def extract_trajectory(final_state: dict) -> list[dict]:
    return [
        {"tool": tc["name"], "args": tc["args"]}
        for msg in final_state["messages"] if isinstance(msg, AIMessage)
        for tc in (msg.tool_calls or [])
    ]

def trajectory_score(expected: list[str], actual: list[str]) -> dict:
    e_set, a_set = set(expected), set(actual)
    coverage = len(e_set & a_set) / len(e_set) if e_set else 1.0
    precision = len(e_set & a_set) / len(a_set) if a_set else 0.0
    shared = [t for t in actual if t in e_set]
    order = _kendall_tau(expected, shared) if len(shared) >= 2 else 1.0
    return {"coverage": coverage, "precision": precision, "order": order}

# Composite: 0.5 * coverage + 0.3 * precision + 0.2 * order

from scipy.stats import wilcoxon

def paired_regression_check(baseline, candidate, alpha=0.05):
    """Wilcoxon — right test when metric distribution is non-normal (most LLM metrics)."""
    n = len(baseline)
    if n < 50:
        return {"verdict": "too_small_n", "n": n}
    diffs = [c - b for b, c in zip(baseline, candidate)]
    _, p = wilcoxon(diffs, alternative="less")
    return {"n": n, "mean_delta": sum(diffs) / n, "p_value": float(p),
            "regression": p < alpha and sum(diffs) < 0}