Trajectory Scoring: Rewarding Good Plans, Not Just Answers

The right answer from the wrong process is a accident waiting to happen. Trajectory scoring evaluates the reasoning path an agent takes—not just whether it arrived at the correct answer, but whether the plan was sound, efficient, and generalizable.

Why Reward the Journey?

Consider two agents answering “What’s the population of France divided by Germany’s GDP?”

• Agent A calls get_population("France") then get_gdp("Germany") and divides. Correct, efficient.
• Agent B calls search_web("population of France divided by Germany GDP"), gets a wrong answer from a blog post. Correct result by coincidence.

Traditional evals score both as correct. Trajectory scoring catches Agent B’s fragile process.

Defining Trajectory Quality

Plan Quality Dimensions

from dataclasses import dataclass
from enum import Enum

class StepType(Enum):
    REASONING = "reasoning"
    TOOL_CALL = "tool_call"
    OBSERVATION = "observation"
    CONCLUSION = "conclusion"

@dataclass
class TrajectoryStep:
    step_type: StepType
    content: str
    tool_name: str | None = None
    tool_args: dict | None = None
    duration_ms: float = 0
    cost: float = 0

@dataclass
class TrajectoryScore:
    plan_soundness: float     # Was the overall strategy correct?
    step_efficiency: float    # Minimal unnecessary steps?
    tool_selection: float     # Right tools at the right time?
    reasoning_quality: float  # Logical, well-grounded reasoning?
    error_recovery: float     # Handled failures gracefully?
    overall: float            # Weighted composite

Implementation with LlamaIndex

LlamaIndex’s agent framework exposes the full trajectory for evaluation:

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.core.evaluation import (
    RelevancyEvaluator,
    FaithfulnessEvaluator,
)

# Define tools
def get_population(country: str) -> int:
    """Get the current population of a country."""
    # Implementation...
    pass

def get_gdp(country: str) -> float:
    """Get the GDP in trillions of a country."""
    # Implementation...
    pass

population_tool = FunctionTool.from_defaults(fn=get_population)
gdp_tool = FunctionTool.from_defaults(fn=get_gdp)

# Create agent
agent = ReActAgent.from_tools(
    tools=[population_tool, gdp_tool],
    llm=llm,
    verbose=True
)

# Capture full trajectory
response = agent.chat("What's the ratio of France's population to Germany's GDP?")

# Extract trajectory from response
trajectory = response.chat_history

Trajectory Scorer

import re

def score_trajectory(trajectory: list, task: str) -> TrajectoryScore:
    steps = parse_trajectory(trajectory)
    
    # Plan soundness: Did the agent decompose the task correctly?
    plan_score = evaluate_plan(steps, task)
    
    # Step efficiency: Were there redundant tool calls?
    tool_calls = [s for s in steps if s.step_type == StepType.TOOL_CALL]
    efficiency_score = 1.0 - (max(0, len(tool_calls) - expected_tool_calls(task)) * 0.15)
    efficiency_score = max(0, efficiency_score)
    
    # Tool selection: Were the right tools called?
    tool_score = evaluate_tool_selection(steps, task)
    
    # Reasoning quality: Was the reasoning logical?
    reasoning_steps = [s for s in steps if s.step_type == StepType.REASONING]
    reasoning_score = evaluate_reasoning(reasoning_steps)
    
    # Error recovery: Did the agent handle failures?
    error_steps = [s for s in steps if "error" in s.content.lower() or "failed" in s.content.lower()]
    recovery_score = evaluate_error_recovery(steps, error_steps)
    
    overall = (
        plan_score * 0.3 +
        efficiency_score * 0.2 +
        tool_score * 0.25 +
        reasoning_score * 0.15 +
        recovery_score * 0.1
    )
    
    return TrajectoryScore(
        plan_soundness=plan_score,
        step_efficiency=efficiency_score,
        tool_selection=tool_score,
        reasoning_quality=reasoning_score,
        error_recovery=recovery_score,
        overall=overall
    )

def evaluate_plan(steps, task):
    """Check if the plan decomposes the task into correct sub-goals."""
    expected_steps = {
        "ratio": ["get_value_a", "get_value_b", "divide"],
        "comparison": ["get_value_a", "get_value_b", "compare"],
        "summary": ["gather_info", "synthesize"],
    }
    
    task_type = identify_task_type(task)
    expected = expected_steps.get(task_type, [])
    
    actual_plan = [identify_step_purpose(s) for s in steps if s.step_type == StepType.REASONING]
    
    coverage = len(set(expected) & set(actual_plan)) / max(len(expected), 1)
    return coverage

Reasoning Quality Evaluation

def evaluate_reasoning(reasoning_steps: list) -> float:
    scores = []
    for step in reasoning_steps:
        score = 0.0
        
        # Check for logical connectors
        has_evidence = any(w in step.content.lower() for w in ["because", "therefore", "since", "based on"])
        if has_evidence:
            score += 0.3
        
        # Check for quantified claims
        has_numbers = bool(re.search(r'\d+', step.content))
        if has_numbers:
            score += 0.2
        
        # Check for uncertainty acknowledgment
        has_hedging = any(w in step.content.lower() for w in ["might", "approximately", "roughly", "not sure"])
        if has_hedging:
            score += 0.2
        
        # Check for step-by-step structure
        has_structure = any(w in step.content.lower() for w in ["first", "next", "then", "finally", "step"])
        if has_structure:
            score += 0.3
        
        scores.append(min(1.0, score))
    
    return sum(scores) / len(scores) if scores else 0.0

Golden Trajectory Matching

from difflib import SequenceMatcher

def trajectory_similarity(actual: list[str], golden: list[str]) -> float:
    """Compare action sequences, not exact text."""
    # Normalize to action signatures
    actual_actions = [normalize_to_action(s) for s in actual]
    golden_actions = [normalize_to_action(s) for s in golden]
    
    # Use sequence matching
    matcher = SequenceMatcher(None, golden_actions, actual_actions)
    return matcher.ratio()

def normalize_to_action(step: str) -> str:
    """Extract the key action from a step description."""
    if "call" in step.lower():
        tool_match = re.search(r'call (\w+)', step.lower())
        return f"tool:{tool_match.group(1)}" if tool_match else "tool:unknown"
    if "search" in step.lower():
        return "search"
    if "conclude" in step.lower() or "answer" in step.lower():
        return "conclude"
    return "reason"

Building a Trajectory Eval Dataset

trajectory_dataset = [
    {
        "task": "Calculate the GDP per capita of Japan",
        "golden_trajectory": [
            {"action": "reason", "content": "I need Japan's GDP and population"},
            {"action": "tool: get_gdp", "args": {"country": "Japan"}},
            {"action": "tool: get_population", "args": {"country": "Japan"}},
            {"action": "reason", "content": "Now divide GDP by population"},
            {"action": "conclude", "content": "Japan's GDP per capita is approximately $34,000"}
        ],
        "anti_patterns": [
            {"action": "tool: search_web", "reason": "Should use structured tools, not web search"},
            {"action": "tool: get_gdp", "args": {"country": "Japan"}, "action": "tool: get_gdp", "args": {"country": "Japan"}},  # Duplicate call
        ]
    }
]

Monitoring Trajectory Quality in Production

class TrajectoryMonitor:
    def __init__(self):
        self.scores = []
    
    def log_trajectory(self, task: str, trajectory: list, score: TrajectoryScore):
        self.scores.append({
            "timestamp": time.time(),
            "task": task,
            "score": score,
            "num_steps": len(trajectory)
        })
        
        # Alert on bad trajectories
        if score.plan_soundness < 0.5:
            alert(f"Poor plan quality for task: {task}")
        if score.step_efficiency < 0.3:
            alert(f"Inefficient trajectory for task: {task}")
    
    def get_trends(self) -> dict:
        recent = self.scores[-100:]
        return {
            "avg_plan_score": sum(s["score"].plan_soundness for s in recent) / len(recent),
            "avg_efficiency": sum(s["score"].step_efficiency for s in recent) / len(recent),
            "avg_steps": sum(s["num_steps"] for s in recent) / len(recent),
        }

Trajectory scoring reveals whether your agent is getting lucky or getting good. An agent with high answer accuracy but low trajectory scores is a ticking time bomb—it’ll fail on novel tasks where its fragile reasoning doesn’t happen to work.