Haystack evaluation pipeline tutorial: answer correctness

Haystack’s evaluation framework lets you measure answer correctness programmatically instead of eyeballing outputs. This tutorial walks through building a complete evaluation pipeline: defining ground truth, configuring the AnswerCorrectnessEvaluator, running batch evaluations, and wiring results into CI. You’ll end up with a reproducible script that fails fast when regression creeps in.

Prerequisites

• Python 3.10+
• Haystack 2.10+ (pip install haystack-ai)
• An OpenAI-compatible endpoint for the judge model (GPT-4o-mini works well and keeps costs low)
• A small labeled dataset — 20 to 50 QA pairs with reference answers

Install the dependencies:

pip install haystack-ai datasets pandas

If you’re using a non-OpenAI provider, set the base URL and key accordingly. The evaluator only needs a chat completion endpoint that accepts the standard messages format.

export OPENAI_API_KEY="sk-..."
export OPENAI_BASE_URL="https://api.openai.com/v1"  # or your gateway endpoint

Define your ground truth dataset

Evaluation starts with data. Create a JSONL file where each line contains a question, the reference answer, and optionally the generated answer you want to score. If you omit generated_answer, the pipeline will run your RAG system first — useful for end-to-end runs.

{"question": "What is the capital of France?", "reference_answer": "Paris is the capital city of France.", "generated_answer": "Paris is the capital of France."}
{"question": "Who wrote '1984'?", "reference_answer": "George Orwell wrote the novel 1984.", "generated_answer": "The author of 1984 is George Orwell."}
{"question": "What is the boiling point of water?", "reference_answer": "Water boils at 100 degrees Celsius at standard atmospheric pressure.", "generated_answer": "Water boils at 100°C at sea level."}
{"question": "Explain photosynthesis.", "reference_answer": "Photosynthesis is the process by which green plants convert light energy into chemical energy, producing glucose and oxygen from carbon dioxide and water.", "generated_answer": "Plants use sunlight to turn CO2 and water into sugar and oxygen."}
{"question": "What is the largest planet?", "reference_answer": "Jupiter is the largest planet in our solar system.", "generated_answer": "Saturn is the largest planet."}

Save this as eval_data.jsonl. The last entry is intentionally wrong — we want the evaluator to catch it.

Build the evaluation pipeline

Haystack 2.x uses a Pipeline object to wire components. For answer correctness, the core component is AnswerCorrectnessEvaluator, which uses an LLM judge to compare generated_answer against reference_answer.

# eval_pipeline.py
import os
import json
from pathlib import Path
from haystack import Pipeline
from haystack.components.evaluators import AnswerCorrectnessEvaluator
from haystack.components.generators import OpenAIGenerator
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret

def build_pipeline(judge_model: str = "gpt-4o-mini") -> Pipeline:
    """
    Build a pipeline that scores answer correctness on a 0-1 scale.
    """
    pipe = Pipeline()

    # The judge generator — swap base_url for your gateway if needed
    judge = OpenAIGenerator(
        model=judge_model,
        api_key=Secret.from_env_var("OPENAI_API_KEY"),
        api_base_url=os.getenv("OPENAI_BASE_URL"),
        generation_kwargs={"temperature": 0.0, "max_tokens": 512},
    )
    pipe.add_component("judge", judge)

    # The evaluator wraps the judge and parses its structured output
    evaluator = AnswerCorrectnessEvaluator(judge=judge)
    pipe.add_component("evaluator", evaluator)

    # Connect: evaluator expects `questions`, `contexts`, `generated_answers`, `reference_answers`
    # We'll feed these directly from our dataset loader
    return pipe

The AnswerCorrectnessEvaluator returns a list of EvaluationResult objects with score (float 0-1), feedback (string), and meta (dict). A score above 0.7 generally indicates acceptable correctness; tune the threshold to your domain.

Load data and run evaluation

Now write a script that reads the JSONL, feeds the pipeline, and prints a summary. We’ll also persist raw results for later analysis.

# run_eval.py
import json
import statistics
from pathlib import Path
from eval_pipeline import build_pipeline

DATA_PATH = Path("eval_data.jsonl")
RESULTS_PATH = Path("eval_results.jsonl")
THRESHOLD = 0.7

def load_dataset(path: Path):
    questions = []
    references = []
    generated = []
    contexts = []  # optional; leave empty if not using retrieval context

    with path.open() as f:
        for line in f:
            row = json.loads(line)
            questions.append(row["question"])
            references.append(row["reference_answer"])
            generated.append(row.get("generated_answer", ""))
            contexts.append(row.get("context", []))  # list of strings or empty

    return questions, contexts, generated, references

def main():
    questions, contexts, generated, references = load_dataset(DATA_PATH)

    pipe = build_pipeline()
    result = pipe.run({
        "evaluator": {
            "questions": questions,
            "contexts": contexts,
            "generated_answers": generated,
            "reference_answers": references,
        }
    })

    eval_results = result["evaluator"]["results"]

    # Persist raw results
    with RESULTS_PATH.open("w") as f:
        for i, er in enumerate(eval_results):
            record = {
                "question": questions[i],
                "reference": references[i],
                "generated": generated[i],
                "score": er.score,
                "feedback": er.feedback,
                "meta": er.meta,
            }
            f.write(json.dumps(record) + "\n")

    # Summary stats
    scores = [er.score for er in eval_results]
    passed = sum(1 for s in scores if s >= THRESHOLD)
    print(f"Evaluated {len(scores)} samples")
    print(f"Mean score: {statistics.mean(scores):.3f}")
    print(f"Median score: {statistics.median(scores):.3f}")
    print(f"Pass rate (>= {THRESHOLD}): {passed}/{len(scores)} ({100*passed/len(scores):.1f}%)")

    # Flag failures
    for i, er in enumerate(eval_results):
        if er.score < THRESHOLD:
            print(f"\n⚠️  FAIL [{er.score:.2f}] Q: {questions[i][:80]}...")
            print(f"   Ref: {references[i][:100]}")
            print(f"   Gen: {generated[i][:100]}")
            print(f"   Feedback: {er.feedback}")

if __name__ == "__main__":
    main()

Run it:

python run_eval.py

Expected output (scores will vary slightly by judge model):

Evaluated 5 samples
Mean score: 0.720
Median score: 0.850
Pass rate (>= 0.7): 4/5 (80.0%)

⚠️  FAIL [0.15] Q: What is the largest planet?...
   Ref: Jupiter is the largest planet in our solar system.
   Gen: Saturn is the largest planet.
   Feedback: The generated answer incorrectly identifies Saturn as the largest planet. The reference answer correctly states Jupiter is the largest planet. The generated answer contains a factual error.

The evaluator caught the hallucinated answer and gave it a low score with actionable feedback.

Understand the scoring rubric

The default AnswerCorrectnessEvaluator prompt asks the judge to score on a 0-1 scale where:

• 1.0: Generated answer is factually correct and complete relative to the reference
• 0.5: Partially correct — missing key details or contains minor inaccuracies
• 0.0: Factually incorrect, hallucinated, or irrelevant

You can customize the rubric by passing a system_prompt or instruction to the evaluator. For stricter grading (e.g., medical or legal), raise the bar:

from haystack.components.evaluators import AnswerCorrectnessEvaluator

STRICT_RUBRIC = """
You are a strict evaluator. Score 1.0 only if the generated answer matches the reference
in all factual claims, numbers, and entities. Any omission or deviation scores 0.5 or lower.
Score 0.0 for any hallucination or contradiction.
"""

evaluator = AnswerCorrectnessEvaluator(
    judge=judge,
    instruction=STRICT_RUBRIC,
)

Add retrieval context evaluation (optional)

If your RAG system returns context passages, you can evaluate faithfulness (does the answer stick to the retrieved context?) alongside correctness. Add FaithfulnessEvaluator to the same pipeline:

from haystack.components.evaluators import FaithfulnessEvaluator

pipe = Pipeline()
pipe.add_component("judge", judge)
pipe.add_component("correctness", AnswerCorrectnessEvaluator(judge=judge))
pipe.add_component("faithfulness", FaithfulnessEvaluator(judge=judge))

# Run both evaluators on the same inputs
result = pipe.run({
    "correctness": {
        "questions": questions,
        "contexts": contexts,
        "generated_answers": generated,
        "reference_answers": references,
    },
    "faithfulness": {
        "questions": questions,
        "contexts": contexts,
        "generated_answers": generated,
    },
})

This gives you two orthogonal signals: correctness vs. ground truth, and faithfulness vs. retrieved context. A high correctness but low faithfulness score means the model knew the answer but ignored your retrieval — worth investigating.

Wire into CI

Fail the build when pass rate drops below threshold. Here’s a minimal GitHub Actions job:

# .github/workflows/eval.yml
name: Haystack Evaluation
on:
  push:
    branches: [main]
  pull_request:

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install deps
        run: pip install haystack-ai datasets pandas
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          OPENAI_BASE_URL: ${{ secrets.OPENAI_BASE_URL }}
        run: python run_eval.py
      - name: Upload results
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: eval_results.jsonl

The script exits 0 on success. To make it fail on regression, add a sys.exit in run_eval.py:

import sys

# ... after summary stats ...
if passed / len(scores) < 0.8:  # require 80% pass rate
    print("❌ Evaluation gate failed: pass rate below 80%")
    sys.exit(1)

Common pitfalls

Judge model variance: The same answer can score 0.72 on one run and 0.68 on another. Mitigate by:

• Setting temperature: 0.0 on the judge
• Running each evaluation 3 times and taking the median (costs 3x)
• Using a larger judge model (GPT-4o) for final gates, smaller for PR checks

Reference answer quality: Garbage in, garbage out. If your reference answers are vague (“Paris is a city in France”), the evaluator will penalize precise generated answers (“Paris is the capital and most populous city of France”). Invest in high-quality references.

Context format: contexts expects a List[List[str]] — one list of context strings per question. If you pass a flat list, the evaluator will misalign and produce nonsense scores.

# Correct: list of lists
contexts = [
    ["Paris is the capital of France.", "Population: 2.1M"],
    ["George Orwell, pen name of Eric Blair..."],
]

# Wrong: flat list
contexts = ["Paris is the capital...", "George Orwell..."]

Cost control: At ~$0.15/1M tokens for GPT-4o-mini, a 500-sample eval costs pennies. But if you run on every commit, it adds up. Gate the full suite to nightly; run a 20-sample smoke test on PRs.

Extending: custom evaluators

For domain-specific criteria (citation format, tone, length), write a custom evaluator by subclassing Evaluator or using LLMEvaluator with a tailored prompt:

from haystack.components.evaluators import LLMEvaluator

CITATION_PROMPT = """
Check if the generated answer includes inline citations in [doc_id] format
for every factual claim. Score 1.0 if all claims are cited, 0.5 if some,
0.0 if none.
Question: {{question}}
Generated answer: {{generated_answer}}
Context documents: {{contexts}}
"""

citation_evaluator = LLMEvaluator(
    judge=judge,
    instruction=CITATION_PROMPT,
    inputs=["question", "generated_answer", "contexts"],
    outputs=["score", "feedback"],
)

Add it to the pipeline like the built-in evaluators. This keeps all your quality signals in one run.

Next steps

• Track scores over time with a lightweight dashboard (SQLite + Streamlit, or push to a metrics DB)
• Correlate evaluation scores with user feedback signals (thumbs up/down, regeneration rate)
• A/B judge prompts to reduce false positives/negatives on your specific domain
• Consider SASSEvaluator for semantic similarity when exact factual match is too strict

The pipeline you built today is production-ready. It runs in seconds, costs pennies, and catches regressions before users do. Ship it.