Golden Trajectories: Curating Human-Approved Agent Runs

The gap between a working agent and a reliable agent is data. You need examples of agents succeeding — not just once, but consistently. Golden trajectories are human-approved agent runs that serve as ground truth: they define what “good” looks like, provide training data for fine-tuning, and form the backbone of regression test suites.

What Are Golden Trajectories

A golden trajectory is a complete record of an agent’s execution that has been reviewed and approved by a human expert. It captures:

• The initial task and context
• Every decision the agent made (tool calls, routing, reasoning)
• The intermediate states and outputs
• The final result
• A human approval signal with optional corrections

Think of it as a “verified solution” for agent tasks — similar to how textbooks provide worked examples for students.

Why They Matter

For Evaluation

Golden trajectories give you deterministic benchmarks. Instead of asking “did the agent produce a good output?” you ask “did the agent follow the same path as the approved trajectory?”

For Fine-Tuning

Approved trajectories become training data. You can fine-tune models to replicate successful patterns without reinventing them from scratch.

For Debugging

When an agent fails, comparing its trajectory to a golden one shows exactly where it diverged — which decision was wrong, which tool call was unnecessary, where the reasoning broke down.

Building the Curation Pipeline

Step 1: Collect Agent Runs

Record every agent execution with full trace data:

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class AgentStep:
    step_id: int
    action: str  # "tool_call", "reasoning", "routing"
    tool_name: str | None
    input_data: dict
    output_data: Any
    timestamp: datetime
    duration_ms: float

@dataclass
class AgentRun:
    run_id: str
    task: str
    steps: list[AgentStep]
    final_output: str
    total_duration_ms: float
    token_usage: dict
    status: str  # "completed", "failed", "timeout"

class TrajectoryCollector:
    def __init__(self):
        self.runs: list[AgentRun] = []
    
    def record_run(self, run: AgentRun):
        self.runs.append(run)
        
        # Store in database for curation
        self._persist(run)
    
    def _persist(self, run: AgentRun):
        # Save to database
        pass

Step 2: Human Review Interface

Build an interface for humans to review and approve trajectories:

@dataclass
class TrajectoryReview:
    run_id: str
    reviewer: str
    approved: bool
    rating: int  # 1-5 quality score
    corrections: list[dict]  # Manual fixes applied
    notes: str
    reviewed_at: datetime

class CurationPipeline:
    def __init__(self):
        self.pending_reviews: list[AgentRun] = []
        self.approved_trajectories: list[dict] = []
    
    def queue_for_review(self, run: AgentRun):
        self.pending_reviews.append(run)
    
    def submit_review(self, review: TrajectoryReview):
        run = self._get_run(review.run_id)
        
        if review.approved:
            trajectory = self._build_trajectory(run, review)
            self.approved_trajectories.append(trajectory)
            
            # Store for fine-tuning
            self._store_golden(trajectory)
        
        self._update_status(review.run_id, "reviewed")
    
    def _build_trajectory(self, run: AgentRun, review: TrajectoryReview) -> dict:
        return {
            "task": run.task,
            "trajectory": [
                {
                    "step": step.step_id,
                    "action": step.action,
                    "tool": step.tool_name,
                    "input": step.input_data,
                    "output": step.output_data,
                }
                for step in run.steps
            ],
            "final_output": run.final_output,
            "rating": review.rating,
            "corrections": review.corrections,
            "reviewer": review.reviewer,
        }

Step 3: Automatic Quality Filtering

Pre-filter runs before human review to save time:

class QualityFilter:
    def __init__(self):
        self.min_tool_success_rate = 0.8
        self.max_retries = 3
        self.min_output_quality = 0.6
    
    def should_review(self, run: AgentRun) -> tuple[bool, str]:
        # Check tool success rate
        tool_steps = [s for s in run.steps if s.action == "tool_call"]
        if tool_steps:
            success_rate = sum(
                1 for s in tool_steps 
                if not s.output_data.get("error")
            ) / len(tool_steps)
            
            if success_rate < self.min_tool_success_rate:
                return False, f"Low tool success rate: {success_rate:.2f}"
        
        # Check for excessive retries
        retry_count = sum(
            1 for s in run.steps 
            if s.action == "retry"
        )
        if retry_count > self.max_retries:
            return False, f"Too many retries: {retry_count}"
        
        # Check output quality (heuristic)
        output_length = len(run.final_output)
        if output_length < 50:
            return False, "Output too short"
        
        return True, "Passed quality filter"

Using Golden Trajectories

Evaluation Suite

Compare new agent runs against golden trajectories:

class TrajectoryEvaluator:
    def __init__(self, golden_trajectories: list[dict]):
        self.golden = {t["task"]: t for t in golden_trajectories}
    
    def evaluate(self, agent_run: AgentRun) -> dict:
        golden = self.golden.get(agent_run.task)
        if not golden:
            return {"status": "no_golden_available"}
        
        # Compare tool call sequences
        tool_sequence_score = self._compare_tool_sequences(
            agent_run.steps,
            golden["trajectory"]
        )
        
        # Compare final output
        output_similarity = self._compare_outputs(
            agent_run.final_output,
            golden["final_output"]
        )
        
        # Check for unnecessary steps
        efficiency_score = self._calculate_efficiency(
            len(agent_run.steps),
            len(golden["trajectory"])
        )
        
        return {
            "tool_sequence_match": tool_sequence_score,
            "output_similarity": output_similarity,
            "efficiency": efficiency_score,
            "overall": (tool_sequence_score + output_similarity + efficiency_score) / 3,
        }
    
    def _compare_tool_sequences(self, actual_steps, golden_steps):
        actual_tools = [s.tool_name for s in actual_steps if s.action == "tool_call"]
        golden_tools = [s["tool"] for s in golden_steps if s["action"] == "tool_call"]
        
        # Longest common subsequence
        lcs_length = self._lcs_length(actual_tools, golden_tools)
        max_length = max(len(actual_tools), len(golden_tools))
        
        return lcs_length / max_length if max_length > 0 else 1.0

Fine-Tuning Data Generation

Convert golden trajectories into training data:

def trajectory_to_training_data(trajectory: dict) -> list[dict]:
    """Convert a golden trajectory into fine-tuning examples."""
    training_examples = []
    
    messages = [{"role": "system", "content": "You are a helpful assistant."}]
    
    for step in trajectory["trajectory"]:
        if step["action"] == "tool_call":
            # User message (context)
            messages.append({
                "role": "user",
                "content": f"Task: {trajectory['task']}"
            })
            
            # Assistant response (tool call)
            messages.append({
                "role": "assistant",
                "content": None,
                "tool_calls": [{
                    "function": {
                        "name": step["tool"],
                        "arguments": json.dumps(step["input"])
                    }
                }]
            })
            
            # Tool result
            messages.append({
                "role": "tool",
                "content": json.dumps(step["output"])
            })
    
    # Final response
    messages.append({
        "role": "assistant",
        "content": trajectory["final_output"]
    })
    
    training_examples.append({
        "messages": messages,
        "quality_score": trajectory["rating"],
    })
    
    return training_examples

Regression Testing

Run golden trajectories as regression tests:

import pytest

@pytest.fixture
def golden_suite():
    return load_golden_trajectories("trajectories/")

def test_trajectory_regression(golden_suite):
    evaluator = TrajectoryEvaluator(golden_suite)
    
    for task, golden in golden_suite.items():
        # Run agent on the same task
        agent_run = agent.execute(task)
        
        # Evaluate against golden
        result = evaluator.evaluate(agent_run)
        
        # Assert minimum quality
        assert result["overall"] >= 0.8, \
            f"Agent degraded on task '{task}': {result}"

Curation Best Practices

Diverse Coverage

Ensure golden trajectories cover:

• Common tasks (80% of volume)
• Edge cases (unusual inputs, ambiguous requests)
• Error recovery (tool failures, timeouts)
• Multi-step workflows (complex task chains)

Regular Refresh

Agent capabilities evolve. Refresh golden trajectories:

• Monthly for fast-moving applications
• When models or prompts change significantly
• After major tool or API updates

Reviewer Calibration

Multiple reviewers should agree on trajectory quality:

def calculate_inter_reviewer_agreement(reviews: list[TrajectoryReview]) -> float:
    agreements = 0
    total = 0
    
    for i in range(len(reviews)):
        for j in range(i + 1, len(reviews)):
            if reviews[i].run_id == reviews[j].run_id:
                total += 1
                if reviews[i].approved == reviews[j].approved:
                    agreements += 1
    
    return agreements / total if total > 0 else 0.0

Conclusion

Golden trajectories bridge the gap between “it works” and “it works reliably.” By systematically collecting, reviewing, and storing approved agent runs, you build a foundation for evaluation, fine-tuning, and regression testing. Start with the most common tasks, invest in a clean review interface, and treat your golden trajectory dataset as a living asset that evolves with your agent.