逐步错误定位：用轨迹评估找出智能体究竟在哪一步出错

severity_levels:
  - name: minor
    weight: -1
    description: "Small issue, does not derail the overall trace"
  - name: major
    weight: -5
    description: "Significant error that wastes effort or produces wrong intermediate results"
  - name: critical
    weight: -10
    description: "Fundamental failure that breaks the trace or causes harm"

annotation_task_name: "Agent Trajectory Error Localization"
 
data_files:
  - "data/traces.jsonl"
 
item_properties:
  id_key: "trace_id"
  text_key: "task"
 
# Display agent traces with step-by-step rendering
display:
  type: "agent_trace"
  trace_key: "trace"
  step_display:
    thought: { label: "Thought", color: "#E8F0FE" }
    action: { label: "Action", color: "#FFF3E0" }
    observation: { label: "Observation", color: "#F1F8E9" }
    code: { label: "Code", color: "#F3E5F5" }
 
annotation_schemes:
  - annotation_type: "trajectory_eval"
 
    # Per-step correctness check
 
    # Hierarchical error taxonomy (shown when step is marked incorrect)
 
      - category: "perception"
        label: "Perception Error"
        types:
          - name: "missed_element"
            label: "Missed Element"
            description: "Fails to notice relevant information in observations"
          - name: "misidentified_element"
            label: "Misidentified Element"
            description: "Misinterprets what it observes"
          - name: "hallucinated_element"
            label: "Hallucinated Element"
            description: "Refers to something not present in the context"
          - name: "outdated_reference"
            label: "Outdated Reference"
            description: "Uses stale information from a previous step"
 
      - category: "action"
        label: "Action Error"
        types:
          - name: "wrong_tool"
            label: "Wrong Tool"
            description: "Selects an inappropriate tool for the task"
          - name: "wrong_arguments"
            label: "Wrong Arguments"
            description: "Correct tool but incorrect parameters"
          - name: "premature_termination"
            label: "Premature Termination"
            description: "Stops before the task is complete"
          - name: "unnecessary_action"
            label: "Unnecessary Action"
            description: "Takes an action that adds no value"
          - name: "destructive_action"
            label: "Destructive Action"
            description: "Takes an action that causes harm or data loss"
 
      - category: "communication"
        label: "Communication Error"
        types:
          - name: "unclear_explanation"
            label: "Unclear Explanation"
            description: "Explanation is confusing or ambiguous"
          - name: "missing_context"
            label: "Missing Context"
            description: "Omits critical context from the response"
          - name: "incorrect_summary"
            label: "Incorrect Summary"
            description: "Summary does not match the actual actions taken"
          - name: "overconfident_claim"
            label: "Overconfident Claim"
            description: "States uncertain outcomes as certainties"
 
    # Severity levels with score weights
 
    # Running score configuration
 
    # Optional rationale field
 
  # Overall trace quality (optional, shown at the end)
  - annotation_type: "radio"
 
# Annotator settings
annotator_config:
  allow_back_navigation: true
  require_all_steps: true
 
# Output settings
output:
  path: "output/"
  format: "jsonl"

{
  "trace_id": "trace_042",
  "task": "Find the bug in the calculate_discount function and fix it",
  "trace": [
    {
      "type": "thought",
      "content": "I need to look at the calculate_discount function to find the bug. Let me search for it."
    },
    {
      "type": "action",
      "content": "search",
      "action_input": "def calculate_discount"
    },
    {
      "type": "observation",
      "content": "Found in pricing.py line 45:\ndef calculate_discount(price, discount_pct):\n    return price * discount_pct / 100"
    },
    {
      "type": "thought",
      "content": "The function calculates the discount amount but doesn't subtract it from the price. It should return price - (price * discount_pct / 100)."
    },
    {
      "type": "action",
      "content": "edit_file",
      "action_input": "pricing.py:45: return price - (price * discount_pct / 100)"
    },
    {
      "type": "observation",
      "content": "File edited successfully."
    },
    {
      "type": "action",
      "content": "finish",
      "action_input": "Fixed the calculate_discount function. It was returning the discount amount instead of the discounted price."
    }
  ]
}

python -m potato.trace_converter \
  --input raw_traces/ \
  --output data/traces.jsonl \
  --input-format react

- category: "code_quality"
  label: "Code Quality Error"
  types:
    - name: "syntax_error"
      label: "Syntax Error"
      description: "Generated code has syntax errors"
    - name: "runtime_error"
      label: "Runtime Error"
      description: "Code runs but produces an error"
    - name: "logic_bug"
      label: "Logic Bug"
      description: "Code runs without errors but produces wrong output"
    - name: "style_violation"
      label: "Style Violation"
      description: "Code works but violates project conventions"

potato start config.yaml -p 8000

import json
import pandas as pd
from collections import Counter
from pathlib import Path
 
# Load all annotation files
annotations = []
output_dir = Path("output/")
for f in output_dir.glob("*.jsonl"):
    with open(f) as fh:
        for line in fh:
            annotations.append(json.loads(line))
 
print(f"Loaded {len(annotations)} annotated traces")

# Extract all errors across all traces
errors = []
for ann in annotations:
    for step_ann in ann.get("error_localization", []):
        if step_ann["correctness"] == "incorrect":
            errors.append({
                "trace_id": ann["trace_id"],
                "step_index": step_ann["step_index"],
                "category": step_ann["error_category"],
                "error_type": step_ann["error_type"],
                "severity": step_ann["severity"],
                "rationale": step_ann.get("rationale", ""),
            })
 
error_df = pd.DataFrame(errors)
print(f"Total errors found: {len(error_df)}")
print()
 
# Error distribution by category
print("Errors by category:")
print(error_df["category"].value_counts())
print()
 
# Most common specific error types
print("Top 10 error types:")
print(error_df["error_type"].value_counts().head(10))
print()
 
# Severity distribution
print("Severity distribution:")
print(error_df["severity"].value_counts())

import matplotlib.pyplot as plt
import numpy as np
 
# Normalize step positions to [0, 1] range
for ann in annotations:
    trace_length = len(ann.get("error_localization", []))
    for step_ann in ann["error_localization"]:
        if step_ann["correctness"] == "incorrect":
            step_ann["normalized_position"] = step_ann["step_index"] / max(trace_length - 1, 1)
 
# Collect normalized positions
positions = [
    step_ann["normalized_position"]
    for ann in annotations
    for step_ann in ann.get("error_localization", [])
    if step_ann["correctness"] == "incorrect"
    and "normalized_position" in step_ann
]
 
plt.figure(figsize=(10, 4))
plt.hist(positions, bins=20, edgecolor="black", alpha=0.7)
plt.xlabel("Normalized Position in Trace (0 = start, 1 = end)")
plt.ylabel("Error Count")
plt.title("Where Do Agent Errors Occur?")
plt.tight_layout()
plt.savefig("error_position_distribution.png", dpi=150)
print("Saved error_position_distribution.png")

# Extract final running scores
final_scores = []
for ann in annotations:
    score = 100
    severity_weights = {"minor": -1, "major": -5, "critical": -10}
    for step_ann in ann.get("error_localization", []):
        if step_ann["correctness"] == "incorrect":
            score += severity_weights.get(step_ann["severity"], 0)
    score = max(score, 0)
    final_scores.append({
        "trace_id": ann["trace_id"],
        "final_score": score,
        "overall_success": ann.get("overall_success", "unknown"),
    })
 
score_df = pd.DataFrame(final_scores)
 
print("Score statistics:")
print(score_df["final_score"].describe())
print()
 
# Score distribution by overall success
for label in ["success", "partial", "failure"]:
    subset = score_df[score_df["overall_success"] == label]
    if len(subset) > 0:
        print(f"{label}: mean={subset['final_score'].mean():.1f}, "
              f"median={subset['final_score'].median():.1f}, "
              f"n={len(subset)}")

# Group errors by category + type for a failure mode analysis
failure_modes = (
    error_df.groupby(["category", "error_type"])
    .agg(
        count=("severity", "size"),
        avg_severity_weight=("severity", lambda x: x.map(
            {"minor": 1, "major": 5, "critical": 10}
        ).mean()),
    )
    .sort_values("count", ascending=False)
)
 
print("Top failure modes (by frequency):")
print(failure_modes.head(15).to_string())
print()
 
# Impact-weighted failure modes (frequency x average severity)
failure_modes["impact"] = failure_modes["count"] * failure_modes["avg_severity_weight"]
print("Top failure modes (by impact):")
print(failure_modes.sort_values("impact", ascending=False).head(10).to_string())