Evaluating chains and RAG

Datasets and evaluators, correctness and groundedness scoring, regression suites, and testing a prompt change before you release it.

Building an evaluation dataset

from langsmith import Client

client = Client()

examples = [
    {
        "inputs": {"question": "How long is the refund window?"},
        "outputs": {"answer": "30 days from delivery."},
        "metadata": {"split": "billing", "difficulty": "easy"},
    },
    {
        "inputs": {"question": "Can I refund a gift card?"},
        "outputs": {"answer": "Gift cards are non-refundable."},
        "metadata": {"split": "billing", "difficulty": "easy"},
    },
    {
        "inputs": {"question": "What is the enterprise SLA?"},
        "outputs": {"answer": "A one-hour response target."},
        "metadata": {"split": "sla", "difficulty": "medium"},
    },
]

dataset = client.create_dataset("support-qa", description="Labelled support questions")
client.create_examples(dataset_id=dataset.id, examples=examples)
print(dataset.id, dataset.name)

# local equivalent with no hosted service: a JSONL file and a runner
import json
with open("eval/qa.jsonl", "w", encoding="utf-8") as fh:
    for example in examples:
        fh.write(json.dumps(example, ensure_ascii=False) + "\n")

• Build the dataset from real user questions, not from documents. Questions generated from documents contain the document's vocabulary and make retrieval look easy.
• Include a slice of unanswerable questions. A system that never abstains scores well on answerable ones and fails exactly where trust matters.
• Tag every example with a category so you can report per-slice results. An aggregate score hides the slice that regressed.
• Freeze the dataset. Adding examples in response to a failure is how a benchmark drifts into a training set.

Evaluators

from langsmith.evaluation import evaluate, LangChainStringEvaluator

# deterministic evaluators first: cheap, fast, no model needed
def exact_match(run, example):
    got = (run.outputs or {}).get("answer", "").strip().lower()
    want = (example.outputs or {}).get("answer", "").strip().lower()
    return {"key": "exact_match", "score": float(got == want)}

def contains_expected(run, example):
    got = (run.outputs or {}).get("answer", "").lower()
    want = (example.outputs or {}).get("answer", "").lower()
    return {"key": "contains", "score": float(want in got)}

# a groundedness judge: every claim must be supported by the retrieved context
GROUNDEDNESS = """You are grading a support answer.
Question: {question}
Retrieved context: {context}
Answer: {answer}

Score 1 if every factual claim in the answer is supported by the context,
0 if any claim is unsupported or the answer contradicts the context.
Reply with only 0 or 1."""

judge = LangChainStringEvaluator(
    "cot_qa",
    config={"llm": model},
    prepare_data=lambda run, example: {
        "query": example.inputs["question"],
        "prediction": run.outputs["answer"],
        "input": example.inputs["question"],
    },
)

def grounded(run, example):
    prompt = (GROUNDEDNESS
              .replace("{question}", example.inputs["question"])
              .replace("{context}", (run.outputs or {}).get("context", "")[:4000])
              .replace("{answer}", (run.outputs or {}).get("answer", "")))
    verdict = model.invoke(prompt).content.strip()
    return {"key": "groundedness", "score": float(verdict.startswith("1"))}

results = evaluate(
    lambda inputs: {"answer": chain.invoke(inputs).content},
    data="support-qa",
    evaluators=[exact_match, contains_expected, grounded],
    experiment_prefix="baseline",
)
print(results.to_pandas()[["feedback.contains", "feedback.groundedness"]].mean())

Regression suites and release gates

CASES = [
    {"question": "How long is the refund window?", "must_contain": ["30 days"]},
    {"question": "Can I refund a gift card?", "must_contain": ["non-refundable"]},
    {"question": "What is the enterprise SLA?", "must_contain": ["one-hour", "1 hour"]},
]

def run_suite(chain, cases=CASES):
    failures = []
    for case in cases:
        out = chain.invoke({"question": case["question"]})
        answer = out.get("answer", "").lower()
        if not any(term.lower() in answer for term in case["must_contain"]):
            failures.append({"question": case["question"],
                             "answer": answer[:160],
                             "expected_any": case["must_contain"]})
    return failures

failures = run_suite(candidate_chain)
print(f"{len(CASES) - len(failures)}/{len(CASES)} passed")
for failure in failures:
    print(failure)

# a release gate in CI: fail the pipeline when the pass rate drops
def gate(pass_rate, threshold=0.9):
    if pass_rate < threshold:
        raise SystemExit(f"evaluation gate failed: {pass_rate:.2f} < {threshold:.2f}")
    print(f"gate passed: {pass_rate:.2f}")

gate((len(CASES) - len(failures)) / len(CASES))

• Run the full dataset before a release and a small smoke set on every pull request. The full set is too slow for a per-commit gate.
• Compare against the previous version, not against an absolute threshold alone: the useful signal is whether this change made things worse.
• Record the prompt version, model version and retrieval configuration as part of the experiment. A score without the configuration that produced it cannot be reproduced.
• Improvements on one slice with a regression on another are the common case. Decide the trade-off explicitly, in writing, rather than averaging it away.

FAQ

Related

Observability with callbacks and LangSmith Retrievers in depth

Last refreshed 2026-09-18.