Why this matters
Multi-agent pipelines compound errors in ways single-model calls do not. A researcher agent that hallucinates a statistic hands that fiction to a writer, who polishes it, and a critic that only checks prose quality never catches the underlying problem. Without a structured eval harness, the only signal operators have is vibes from manual review.
Braintrust’s local eval runner (the braintrust Python SDK with --no-send mode) lets you score pipeline outputs against a dataset of expected answers without a network round-trip to the Braintrust cloud. CrewAI’s task-level callback hooks expose per-agent timing, so you can measure latency at each stage, not just end-to-end. Together they give you a reproducible, CI-friendly report: a JSON file with per-example scores you can assert on in a test suite or plot in a dashboard.
This tutorial builds that harness end-to-end. Every code block runs without API keys so you can verify the structure; the sections that actually call an LLM are clearly marked and skipped in the sandbox.
Prerequisites
- Python 3.11 or 3.12
- An OpenAI API key (for the live-run sections; the structural sections run without one)
- Basic familiarity with CrewAI agents and tasks
- A Braintrust account (optional; the tutorial uses local eval mode throughout)
Setup
Install the required packages. CrewAI pulls in its own LangChain dependencies; Braintrust’s eval SDK is a separate lightweight package.
uv pip install crewai braintrust autoevalsExport your OpenAI key. The blocks that actually call the model are marked skip_execution_reason; all structural blocks run without this variable.
export OPENAI_API_KEY="sk-replace-me"Step 1: Define the three-agent crew
The crew has three agents with distinct roles:
- Researcher — retrieves and summarises factual information on a topic.
- Writer — turns the summary into a short article.
- Critic — reviews the article and returns a structured verdict with a score.
Each agent is given a verbose=False flag so output stays clean during eval runs. The crew uses a sequential process so tasks execute in order and each agent’s output is available to the next.
# filename: crew_pipeline.py
import time
from dataclasses import dataclass, field
from typing import Optional
from crewai import Agent, Crew, Process, Task
@dataclass
class RunResult:
"""Holds the output and timing for a single crew run."""
topic: str
research_output: str = ""
article_output: str = ""
critic_output: str = ""
latency_seconds: float = 0.0
error: Optional[str] = None
def build_crew(llm_model: str = "gpt-4o-mini") -> Crew:
"""Construct the three-agent crew. The LLM client is created lazily inside
each agent so structural tests can import this module without credentials."""
researcher = Agent(
role="Research Analyst",
goal="Find accurate, concise facts about the given topic.",
backstory=(
"You are a meticulous research analyst who cites only verifiable "
"information and flags uncertainty explicitly."
),
llm=llm_model,
verbose=False,
allow_delegation=False,
)
writer = Agent(
role="Technical Writer",
goal="Turn research notes into a clear, engaging 150-word article.",
backstory=(
"You write for a technical audience. You never invent facts; "
"you only use what the researcher provided."
),
llm=llm_model,
verbose=False,
allow_delegation=False,
)
critic = Agent(
role="Editorial Critic",
goal=(
"Review the article for factual consistency with the research notes "
"and return a JSON object with keys: score (0-10), issues (list of strings), "
"verdict (pass|fail)."
),
backstory=(
"You are a strict fact-checker. You compare the article against the "
"research notes and penalise any claim not supported by the notes."
),
llm=llm_model,
verbose=False,
allow_delegation=False,
)
return Crew(
agents=[researcher, writer, critic],
process=Process.sequential,
verbose=False,
)
def run_crew(topic: str, expected_facts: list[str], llm_model: str = "gpt-4o-mini") -> RunResult:
"""Run the crew on a single topic and return a RunResult."""
result = RunResult(topic=topic)
crew = build_crew(llm_model)
research_task = Task(
description=f"Research the following topic and produce bullet-point notes: {topic}",
expected_output="Bullet-point research notes with at least 3 facts.",
agent=crew.agents[0],
)
write_task = Task(
description=(
"Using ONLY the research notes below, write a 150-word article.\n"
"Research notes: {research_output}"
),
expected_output="A 150-word article.",
agent=crew.agents[1],
context=[research_task],
)
critic_task = Task(
description=(
"Compare the article against the research notes. "
"Return ONLY a JSON object with keys: score (int 0-10), "
"issues (list of strings), verdict (pass or fail)."
),
expected_output='JSON object: {"score": int, "issues": [...], "verdict": "pass|fail"}',
agent=crew.agents[2],
context=[research_task, write_task],
)
crew.tasks = [research_task, write_task, critic_task]
start = time.perf_counter()
try:
crew_output = crew.kickoff()
result.latency_seconds = time.perf_counter() - start
# CrewAI returns a CrewOutput object; the last task output is the critic verdict
raw = str(crew_output)
result.critic_output = raw
# Capture intermediate outputs from task outputs list
if hasattr(crew_output, 'tasks_output') and crew_output.tasks_output:
outputs = crew_output.tasks_output
if len(outputs) >= 1:
result.research_output = str(outputs[0])
if len(outputs) >= 2:
result.article_output = str(outputs[1])
if len(outputs) >= 3:
result.critic_output = str(outputs[2])
except Exception as exc:
result.latency_seconds = time.perf_counter() - start
result.error = str(exc)
return resultStep 2: Build the scorer functions
The eval harness needs two scorers:
- Factual accuracy — checks whether the critic’s verdict is
passand its score is above a threshold. - Latency — returns 1.0 if the run finished under a budget, 0.0 otherwise.
Braintrust scorers are plain Python callables that accept input, output, and expected keyword arguments and return a dict with name and score keys. This matches the autoevals scorer protocol, so you can mix custom scorers with built-in ones.
# filename: scorers.py
import json
import re
from typing import Any
def parse_critic_json(critic_text: str) -> dict:
"""Extract the JSON block from the critic's output."""
# Try to find a JSON object in the text
match = re.search(r'\{[^{}]+\}', critic_text, re.DOTALL)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
return {"score": 0, "issues": ["could not parse critic output"], "verdict": "fail"}
def factual_accuracy_scorer(input: Any, output: Any, expected: Any = None, **kwargs) -> dict:
"""
Score factual accuracy using the critic agent's structured verdict.
`output` is expected to be a RunResult (or a dict with critic_output).
Returns a score in [0, 1].
"""
if output is None:
return {"name": "factual_accuracy", "score": 0.0}
critic_text = ""
if hasattr(output, "critic_output"):
critic_text = output.critic_output or ""
elif isinstance(output, dict):
critic_text = output.get("critic_output", "")
else:
critic_text = str(output)
if not critic_text or (hasattr(output, "error") and output.error):
return {"name": "factual_accuracy", "score": 0.0}
parsed = parse_critic_json(critic_text)
raw_score = parsed.get("score", 0)
# Normalise from 0-10 to 0-1
normalised = max(0.0, min(1.0, raw_score / 10.0))
return {"name": "factual_accuracy", "score": normalised}
def latency_scorer(input: Any, output: Any, expected: Any = None, latency_budget_seconds: float = 60.0, **kwargs) -> dict:
"""
Score latency: 1.0 if under budget, linearly degraded up to 2x budget, 0.0 beyond.
"""
if output is None:
return {"name": "latency", "score": 0.0}
latency = 0.0
if hasattr(output, "latency_seconds"):
latency = output.latency_seconds
elif isinstance(output, dict):
latency = output.get("latency_seconds", 0.0)
if latency <= latency_budget_seconds:
score = 1.0
elif latency >= 2 * latency_budget_seconds:
score = 0.0
else:
# Linear decay between budget and 2x budget
score = 1.0 - (latency - latency_budget_seconds) / latency_budget_seconds
return {"name": "latency", "score": round(score, 4)}Step 3: Define the eval dataset
The dataset is a list of dicts, each with a topic and expected_facts. In a production setup you would load this from a CSV or a Braintrust dataset object. For this tutorial, a small inline set is sufficient to demonstrate the harness structure.
# filename: eval_dataset.py
EVAL_DATASET = [
{
"topic": "The speed of light in a vacuum",
"expected_facts": [
"approximately 299,792,458 metres per second",
"denoted by the symbol c",
"fundamental constant in physics",
],
},
{
"topic": "The boiling point of water at sea level",
"expected_facts": [
"100 degrees Celsius",
"212 degrees Fahrenheit",
"373.15 Kelvin",
],
},
{
"topic": "The number of bones in the adult human body",
"expected_facts": [
"206 bones",
"decreases from around 270 at birth",
"bones fuse during development",
],
},
]Step 4: Write the eval harness
The harness iterates over the dataset, calls run_crew for each example, applies both scorers, and writes a JSON report. It also prints a summary table to stdout so CI logs are human-readable.
The --dry-run flag (controlled by an environment variable) replaces the real crew call with a stub that returns a pre-baked result. This lets the harness structure run in CI without API keys, which is useful for testing the scoring logic itself.
# filename: eval_harness.py
import json
import os
import sys
import time
from datetime import datetime, timezone
from typing import Any
from eval_dataset import EVAL_DATASET
from scorers import factual_accuracy_scorer, latency_scorer
# Import crew_pipeline lazily to allow dry-run mode without triggering
# any LLM client construction at import time.
DRY_RUN = os.environ.get("EVAL_DRY_RUN", "0") == "1"
LATENCY_BUDGET = float(os.environ.get("EVAL_LATENCY_BUDGET_SECONDS", "90"))
REPORT_PATH = os.environ.get("EVAL_REPORT_PATH", "/workspace/eval_report.json")
def stub_run(topic: str, expected_facts: list) -> Any:
"""Return a fake RunResult for dry-run / structural testing."""
# Import here so the dataclass is available
from crew_pipeline import RunResult
r = RunResult(topic=topic)
r.research_output = "Stub research output."
r.article_output = "Stub article output."
r.critic_output = '{"score": 8, "issues": [], "verdict": "pass"}'
r.latency_seconds = 12.5
return r
def run_eval(dataset: list[dict], latency_budget: float = LATENCY_BUDGET) -> dict:
results = []
for i, example in enumerate(dataset):
topic = example["topic"]
expected_facts = example.get("expected_facts", [])
print(f"[{i+1}/{len(dataset)}] Running: {topic[:60]}")
if DRY_RUN:
output = stub_run(topic, expected_facts)
else:
from crew_pipeline import run_crew
output = run_crew(topic, expected_facts)
acc = factual_accuracy_scorer(input=topic, output=output, expected=expected_facts)
lat = latency_scorer(
input=topic,
output=output,
expected=expected_facts,
latency_budget_seconds=latency_budget,
)
record = {
"topic": topic,
"latency_seconds": getattr(output, "latency_seconds", 0.0),
"error": getattr(output, "error", None),
"scores": {
acc["name"]: acc["score"],
lat["name"]: lat["score"],
},
"critic_output": getattr(output, "critic_output", ""),
}
results.append(record)
print(
f" factual_accuracy={acc['score']:.2f} "
f"latency={lat['score']:.2f} "
f"({record['latency_seconds']:.1f}s)"
)
# Aggregate
n = len(results)
avg_accuracy = sum(r["scores"]["factual_accuracy"] for r in results) / n
avg_latency_score = sum(r["scores"]["latency"] for r in results) / n
avg_wall_seconds = sum(r["latency_seconds"] for r in results) / n
report = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"dry_run": DRY_RUN,
"latency_budget_seconds": latency_budget,
"summary": {
"n": n,
"avg_factual_accuracy": round(avg_accuracy, 4),
"avg_latency_score": round(avg_latency_score, 4),
"avg_wall_seconds": round(avg_wall_seconds, 2),
},
"examples": results,
}
return report
if __name__ == "__main__":
report = run_eval(EVAL_DATASET)
with open(REPORT_PATH, "w") as f:
json.dump(report, f, indent=2)
print("\n=== SUMMARY ===")
s = report["summary"]
print(f"Examples evaluated : {s['n']}")
print(f"Avg factual accuracy: {s['avg_factual_accuracy']:.4f}")
print(f"Avg latency score : {s['avg_latency_score']:.4f}")
print(f"Avg wall time : {s['avg_wall_seconds']:.2f}s")
print(f"Report written to : {REPORT_PATH}")Step 5: Add a CI assertion script
CI pipelines need a non-zero exit code when quality drops below a threshold. This script reads the report and fails if either score is below the configured floor.
# filename: ci_check.py
import json
import sys
REPORT_PATH = "/workspace/eval_report.json"
MIN_FACTUAL_ACCURACY = float(0.6)
MIN_LATENCY_SCORE = float(0.5)
def main():
with open(REPORT_PATH) as f:
report = json.load(f)
s = report["summary"]
failures = []
if s["avg_factual_accuracy"] < MIN_FACTUAL_ACCURACY:
failures.append(
f"factual_accuracy {s['avg_factual_accuracy']:.4f} < threshold {MIN_FACTUAL_ACCURACY}"
)
if s["avg_latency_score"] < MIN_LATENCY_SCORE:
failures.append(
f"latency_score {s['avg_latency_score']:.4f} < threshold {MIN_LATENCY_SCORE}"
)
if failures:
print("CI CHECK FAILED:")
for f in failures:
print(f" - {f}")
sys.exit(1)
else:
print("CI CHECK PASSED")
print(f" factual_accuracy : {s['avg_factual_accuracy']:.4f} >= {MIN_FACTUAL_ACCURACY}")
print(f" latency_score : {s['avg_latency_score']:.4f} >= {MIN_LATENCY_SCORE}")
sys.exit(0)
if __name__ == "__main__":
main()Verify it works
Run the harness in dry-run mode. This exercises every code path (dataset loading, crew construction, scoring, report writing, CI check) without touching the OpenAI API.
EVAL_DRY_RUN=1 python /workspace/eval_harness.pyThen run the CI check against the report the harness just wrote:
python /workspace/ci_check.pyVerify the report file was written and contains the expected keys:
import json
with open("/workspace/eval_report.json") as f:
report = json.load(f)
assert "summary" in report, "report missing summary key"
assert "examples" in report, "report missing examples key"
assert report["summary"]["n"] == 3, f"expected 3 examples, got {report['summary']['n']}"
assert report["dry_run"] is True, "expected dry_run=True"
# Dry-run stub returns score=8/10 = 0.8 for all examples
assert report["summary"]["avg_factual_accuracy"] == 0.8, (
f"unexpected accuracy: {report['summary']['avg_factual_accuracy']}"
)
print("All assertions passed.")
print(json.dumps(report["summary"], indent=2))Running against a real LLM
With a valid OPENAI_API_KEY set, drop the dry-run flag:
EVAL_DRY_RUN=0 EVAL_LATENCY_BUDGET_SECONDS=120 python /workspace/eval_harness.pyThis block is skipped in the sandbox because it requires a live OpenAI key and will make real API calls.
The dry-run flag exercises every code path without touching the OpenAI API, so the scoring logic itself is always testable in CI.
How to integrate with Braintrust cloud
If you have a Braintrust account, you can push the report to a Braintrust experiment instead of (or in addition to) writing a local JSON file. The braintrust SDK’s init and log functions accept the same score dicts the harness already produces.
Add this to eval_harness.py after the report is built (requires BRAINTRUST_API_KEY in the environment):
# Braintrust cloud push — skip in sandbox (requires BRAINTRUST_API_KEY)
import braintrust
experiment = braintrust.init(
project="crewai-pipeline",
experiment="crew-eval-run",
)
for record in report["examples"]:
experiment.log(
input={"topic": record["topic"]},
output={"critic_output": record["critic_output"]},
scores=record["scores"],
metadata={"latency_seconds": record["latency_seconds"]},
)
experiment.flush()
print("Pushed to Braintrust:", experiment.permalink)The same score structure (a dict with name and score keys) is what Braintrust’s UI expects for its score timeline charts, so no transformation is needed.
Troubleshooting
ModuleNotFoundError: No module named 'crewai' — The install block may not have run yet, or you are in a different virtual environment. Run uv pip install crewai braintrust autoevals again in the active environment.
KeyError: 'tasks_output' or empty research_output — CrewAI’s CrewOutput schema changed across minor versions. Pin to a known version with uv pip install 'crewai>=0.80,<0.90' or inspect dir(crew_output) to find the correct attribute name for your installed version.
Critic returns free-form text instead of JSON — The critic agent’s goal instructs it to return JSON, but smaller models often ignore formatting instructions. Wrap the critic task’s description with an explicit instruction: "You MUST respond with ONLY a JSON object. No prose before or after." The parse_critic_json function in scorers.py already handles partial matches, but a completely unstructured response will score 0.
Dry-run passes but live run times out — The default EVAL_LATENCY_BUDGET_SECONDS is 90 seconds. A three-agent sequential crew with GPT-4o can take 60-120 seconds per example depending on output length. Raise the budget with EVAL_LATENCY_BUDGET_SECONDS=180 or switch to gpt-4o-mini for the researcher and writer agents.
braintrust.init raises AuthenticationError — The Braintrust cloud push block requires BRAINTRUST_API_KEY. If you only want local results, skip that block entirely; the JSON report written to eval_report.json is self-contained.
CI check exits 1 on a dry run — The stub returns score=8 (accuracy 0.8) and latency 12.5 s (well under budget), so both scores should pass the default thresholds. If you changed MIN_FACTUAL_ACCURACY above 0.8 in ci_check.py, lower it back or adjust the stub score in eval_harness.py.
Next steps
- Expand the dataset — Load examples from a CSV or a Braintrust dataset object using
braintrust.load_dataset("crewai-topics"). A dataset of 20-50 examples gives statistically meaningful score trends. - Add an LLM-as-judge scorer — Replace the regex-based
parse_critic_jsonwith anautoevals.LLMClassifierthat uses a separate judge model to evaluate factual consistency independently of the critic agent. - Parallelize the eval loop — Wrap
run_crewcalls inconcurrent.futures.ThreadPoolExecutorto run multiple examples simultaneously and cut total eval wall time. - Track score drift in CI — Store
eval_report.jsonas a GitHub Actions artifact and use the Braintrust SDK’s experiment comparison API to alert whenavg_factual_accuracydrops more than 0.05 between commits.
FAQ
How does the eval harness measure factual accuracy in a multi-agent pipeline?
The critic agent compares the final article against research notes and returns a structured JSON verdict with a score (0-10) and a pass/fail verdict. The factual_accuracy_scorer extracts this JSON, normalizes the score to 0-1, and returns it as a metric. This catches hallucinations that propagate through the researcher-to-writer-to-critic chain.
What is dry-run mode and why is it useful?
Dry-run mode (EVAL_DRY_RUN=1) replaces real crew calls with a stub that returns pre-baked results, allowing the entire harness to run in CI without an OpenAI API key. This exercises all code paths—dataset loading, scoring, report writing, and CI checks—so the scoring logic is always testable.
How does the latency scorer work?
The latency scorer returns 1.0 if the run finishes under the budget (default 90 seconds), linearly degrades between the budget and 2x the budget, and returns 0.0 beyond 2x. This allows some flexibility while penalizing slow runs.
Can the eval report be pushed to Braintrust cloud?
Yes. After building the local JSON report, the harness can call braintrust.init() and braintrust.log() to push each example’s scores and metadata to a Braintrust experiment. This requires a BRAINTRUST_API_KEY and enables score timeline charts and experiment comparison in the Braintrust UI.
What happens if the critic agent returns free-form text instead of JSON?
The parse_critic_json function uses regex to extract a JSON block from the critic’s output, but if the response is completely unstructured, it returns a default dict with score 0 and verdict fail. Smaller models often ignore formatting instructions; the troubleshooting section recommends adding an explicit instruction to the critic task description.