# Build a Reproducible CrewAI Multi-Agent Eval Harness with Braintrust

> You will build a three-agent CrewAI crew (researcher, writer, critic) and wire it to a Braintrust eval harness that scores every run on factual accuracy and latency. The result is a reproducible report you can drop into CI and track across commits.

- Canonical URL: https://agentry.press/tutorial/build-a-reproducible-crewai-multi-agent-eval-harness-with-braintrust/
- Type: Tutorial
- Published: 2026-06-15
- By: agentry
- Tags: crewai, braintrust, multi-agent, evals, llmops, ci

---

## 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.

```bash
uv pip install crewai braintrust autoevals
```

Export your OpenAI key. The blocks that actually call the model are marked `skip_execution_reason`; all structural blocks run without this variable.

```bash
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.

```python
# 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 result
```

## Step 2: Build the scorer functions

The eval harness needs two scorers:

1. **Factual accuracy** — checks whether the critic's verdict is `pass` and its score is above a threshold.
2. **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.

```python
# 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.

```python
# 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.

```python
# 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.

```python
# 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.

```bash
EVAL_DRY_RUN=1 python /workspace/eval_harness.py
```

Then run the CI check against the report the harness just wrote:

```bash
python /workspace/ci_check.py
```

Verify the report file was written and contains the expected keys:

```python
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:

```bash
EVAL_DRY_RUN=0 EVAL_LATENCY_BUDGET_SECONDS=120 python /workspace/eval_harness.py
```

This block is skipped in the sandbox because it requires a live OpenAI key and will make real API calls.

> [!PULLQUOTE]
> 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):

```python
# 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_json` with an `autoevals.LLMClassifier` that uses a separate judge model to evaluate factual consistency independently of the critic agent.
- **Parallelize the eval loop** — Wrap `run_crew` calls in `concurrent.futures.ThreadPoolExecutor` to run multiple examples simultaneously and cut total eval wall time.
- **Track score drift in CI** — Store `eval_report.json` as a GitHub Actions artifact and use the Braintrust SDK's experiment comparison API to alert when `avg_factual_accuracy` drops 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.

## References

1. https://www.runtm.com/