Automated Evaluation Pipelines
Why Automate Evaluation
Section titled “Why Automate Evaluation”Manual evaluation doesn’t scale. Running a benchmark by hand after every prompt change is slow, inconsistent, and easy to skip under deadline pressure. Automated pipelines enforce quality gates the same way unit tests do: every change either passes or it doesn’t.
The goal is to have evaluation run automatically so that:
- A prompt regression is caught before it ships, not after users complain
- Model upgrade decisions are data-driven (“Claude 4 scores 8% higher on our benchmark”)
- Quality trends are visible over time without anyone manually tracking them
The Evaluation Pipeline Architecture
Section titled “The Evaluation Pipeline Architecture”Developer pushes code │ ▼GitHub Actions triggered │ ▼[Eval job]├── Load golden test dataset├── Run test cases against current model/prompt├── Compute metrics (faithfulness, relevancy, etc.)├── Compare scores to baseline│├── Score improved or unchanged? ──► ✅ Pass, deploy allowed└── Score regressed beyond threshold? ──► ❌ Fail, block deploy │ ▼Score report posted to PR as comment │ ▼Scores logged to LangSmith / dashboardGitHub Actions Evaluation Workflow
Section titled “GitHub Actions Evaluation Workflow”name: AI Evaluation
on: pull_request: paths: - "src/prompts/**" - "src/rag/**" - "src/agents/**"
jobs: evaluate: runs-on: ubuntu-latest
steps: - uses: actions/checkout@v4
- uses: actions/setup-python@v5 with: python-version: "3.12" cache: "pip"
- name: Install dependencies run: pip install deepeval ragas langsmith openai
- name: Run evaluation suite env: OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }} LANGCHAIN_TRACING_V2: "true" LANGCHAIN_PROJECT: "ci-evals" run: | python scripts/run_evals.py \ --dataset golden_dataset.jsonl \ --output eval_results.json \ --threshold 0.75
- name: Check regression threshold run: python scripts/check_regression.py eval_results.json
- name: Post results to PR if: always() uses: actions/github-script@v7 with: script: | const fs = require('fs'); const results = JSON.parse(fs.readFileSync('eval_results.json')); const body = `## 🧪 AI Evaluation Results\n\n` + `| Metric | Score | Status |\n|--------|-------|--------|\n` + Object.entries(results.scores).map(([metric, score]) => `| ${metric} | ${score.toFixed(3)} | ${score >= results.thresholds[metric] ? '✅' : '❌'} |` ).join('\n'); github.rest.issues.createComment({ issue_number: context.issue.number, owner: context.repo.owner, repo: context.repo.repo, body });Evaluation Runner Script
Section titled “Evaluation Runner Script”import argparseimport jsonimport sysfrom pathlib import Pathfrom deepeval import evaluate as deepeval_evaluatefrom deepeval.metrics import AnswerRelevancyMetric, FaithfulnessMetricfrom deepeval.test_case import LLMTestCase
from my_app.rag_pipeline import run_rag # your application code
def load_dataset(path: str) -> list[dict]: return [json.loads(line) for line in Path(path).read_text().strip().splitlines()]
def run_evaluation(dataset: list[dict]) -> dict: test_cases = [] for item in dataset: result = run_rag(item["input"]) test_cases.append(LLMTestCase( input=item["input"], actual_output=result["answer"], expected_output=item["expected_output"], retrieval_context=result.get("context", []), ))
metrics = [ AnswerRelevancyMetric(threshold=0.7), FaithfulnessMetric(threshold=0.75), ]
results = deepeval_evaluate(test_cases, metrics)
scores = {} for metric in metrics: metric_name = type(metric).__name__ passed = [r for r in results if r.success] scores[metric_name] = len(passed) / len(results)
return scores
def main(): parser = argparse.ArgumentParser() parser.add_argument("--dataset", required=True) parser.add_argument("--output", required=True) parser.add_argument("--threshold", type=float, default=0.75) args = parser.parse_args()
dataset = load_dataset(args.dataset) scores = run_evaluation(dataset)
thresholds = {metric: args.threshold for metric in scores} output = {"scores": scores, "thresholds": thresholds}
Path(args.output).write_text(json.dumps(output, indent=2)) print(f"Evaluation complete: {scores}")
if __name__ == "__main__": main()Regression Check Script
Section titled “Regression Check Script”import jsonimport sysfrom pathlib import Path
def check_regression(results_file: str) -> bool: results = json.loads(Path(results_file).read_text()) scores = results["scores"] thresholds = results["thresholds"]
failures = [] for metric, score in scores.items(): threshold = thresholds.get(metric, 0.75) if score < threshold: failures.append(f"{metric}: {score:.3f} < threshold {threshold:.3f}")
if failures: print("❌ Evaluation failed:") for f in failures: print(f" - {f}") return False
print("✅ All metrics passed thresholds") return True
if __name__ == "__main__": results_file = sys.argv[1] passed = check_regression(results_file) sys.exit(0 if passed else 1)LangSmith Integration
Section titled “LangSmith Integration”LangSmith provides tracing and dataset-based evaluation as a hosted service. It integrates natively with LangChain but works with any LLM app via the SDK.
from langsmith import Clientfrom langsmith.evaluation import evaluate
client = Client()
# Create or update your dataset in LangSmithdataset = client.create_dataset( dataset_name="support-bot-golden-set", description="100 production support queries with gold answers")
# Add examplesclient.create_examples( inputs=[{"question": "How do I reset my password?"}], outputs=[{"answer": "Click 'Forgot password' on the login page."}], dataset_id=dataset.id,)
# Define your evaluatordef relevance_evaluator(run, example) -> dict: from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o") prompt = ChatPromptTemplate.from_template(""" Question: {question} Answer: {answer} Expected: {expected}
Score 0-1: Is the answer relevant and correct? Return only a number. """) score = float((prompt | llm).invoke({ "question": example.inputs["question"], "answer": run.outputs["answer"], "expected": example.outputs["answer"] }).content.strip()) return {"key": "relevance", "score": score}
# Run evaluationresults = evaluate( lambda inputs: my_rag_chain.invoke(inputs), data=dataset.name, evaluators=[relevance_evaluator], experiment_prefix="ci-run",)
print(results.to_pandas())Scheduling Regular Evaluations
Section titled “Scheduling Regular Evaluations”Beyond PR-triggered evaluations, run scheduled evaluations to catch model drift:
name: Weekly Model Health Check
on: schedule: - cron: "0 9 * * 1" # Every Monday at 9am UTC workflow_dispatch: # Also allow manual trigger
jobs: health-check: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Run full benchmark suite env: OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} run: python scripts/full_benchmark.py --output weekly_report.json
- name: Send report run: | python scripts/send_eval_report.py \ --results weekly_report.json \ --channel "#ai-alerts"Model providers update their models silently. GPT-4o “0513” behaves differently from GPT-4o “1120”. Scheduled evaluations catch these changes before users notice.
Evaluation Performance Tips
Section titled “Evaluation Performance Tips”Running 100 test cases through an LLM is slow and expensive. Strategies to keep CI fast:
| Strategy | How | Tradeoff |
|---|---|---|
| Parallel execution | asyncio.gather() for concurrent LLM calls | Faster, hits rate limits |
| Caching | Cache responses for unchanged inputs | Misses real regressions if cached |
| Tiered datasets | Run 20-case “smoke test” on PR, full 200-case suite nightly | Faster CI, slower regression detection |
| Sampling | Run 30% random sample per PR, full set on merge to main | Probabilistic coverage |
| Cheaper judge | Use GPT-3.5 as judge instead of GPT-4o | Faster/cheaper, less accurate scoring |
# Parallel evaluation with asyncioimport asynciofrom langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
async def evaluate_case(case: dict) -> dict: response = await llm.ainvoke(case["input"]) return {"input": case["input"], "output": response.content}
async def run_parallel_eval(test_cases: list[dict]) -> list[dict]: tasks = [evaluate_case(case) for case in test_cases] return await asyncio.gather(*tasks)
results = asyncio.run(run_parallel_eval(test_cases))