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

severities:
  - name: minor
    weight: -1
  - name: major
    weight: -5
  - name: critical
    weight: -10

annotation_task_name: "Agent Trajectory Error Localization"
task_dir: "."
 
data_files:
  - "data/traces.jsonl"
 
item_properties:
  id_key: "trace_id"
  text_key: "task"
 
instance_display:
  fields:
    - key: "task"
      type: text
      label: "Task"
    - key: "trace"
      type: agent_trace
      label: "Agent trace"
      display_options:
        show_step_numbers: true
        show_timestamps: false
        step_type_colors:
          thought: "#E8F0FE"
          action: "#FFF3E0"
          observation: "#F1F8E9"
          code: "#F3E5F5"
 
annotation_instructions: |
  Walk the trace one step at a time. Mark a step incorrect only when it is
  wrong on the information available at that step, not with hindsight. When
  several categories apply, choose the most specific one.
 
annotation_schemes:
  - annotation_type: trajectory_eval
    name: step_errors
    description: "Mark each step and classify any error"
    steps_key: trace
    step_text_key: content
    correctness_options:
      - correct
      - incorrect
      - partially_correct
    error_types:
      - name: reasoning
        subtypes:
          - logical_error
          - incorrect_assumption
          - over_generalization
          - circular_reasoning
          - incorrect_calculation
      - name: perception
        subtypes:
          - missed_element
          - misidentified_element
          - hallucinated_element
          - outdated_reference
      - name: action
        subtypes:
          - wrong_tool
          - wrong_arguments
          - premature_termination
          - unnecessary_action
          - destructive_action
      - name: communication
        subtypes:
          - unclear_explanation
          - missing_context
          - incorrect_summary
          - overconfident_claim
    severities:
      - name: minor
        weight: -1
      - name: major
        weight: -5
      - name: critical
        weight: -10
    show_score: true
    max_score: 100
 
  - annotation_type: radio
    name: overall_success
    description: "Did the run succeed overall?"
    labels:
      - success
      - partial
      - failure
    sequential_key_binding: true
 
output_annotation_dir: "annotation_output/"
export_annotation_format: jsonl
 
user_config:
  allow_all_users: true

{
  "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 calculate_discount to find the bug."
    },
    {
      "type": "action",
      "content": "search('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": "It returns the discount amount rather than the discounted price."
    },
    {
      "type": "action",
      "content": "edit_file('pricing.py:45', 'return price - (price * discount_pct / 100)')"
    },
    {
      "type": "observation",
      "content": "File edited successfully."
    }
  ]
}

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

error_types:
  - name: code_quality
    subtypes:
      - syntax_error
      - runtime_error
      - logic_bug
      - style_violation

potato start config.yaml -p 8000

{
  "steps": [
    {"step_index": 0, "correctness": "correct"},
    {
      "step_index": 1,
      "correctness": "incorrect",
      "error_type": "action",
      "error_subtype": "wrong_tool",
      "severity": "major",
      "rationale": "Should have used a more specific selector"
    }
  ],
  "score": 95
}

import json
import pandas as pd
from pathlib import Path
 
SCHEME = "step_errors"
 
traces = []
for f in Path("annotation_output/").rglob("*.jsonl"):
    with open(f) as fh:
        for line in fh:
            rec = json.loads(line)
            payload = rec.get("labels", rec).get(SCHEME)
            if not payload:
                continue
            traces.append({
                "instance_id": rec.get("instance_id"),
                "annotator": rec.get("user_id"),
                "steps": payload.get("steps", []),
                "score": payload.get("score"),
            })
 
print(f"Loaded {len(traces)} annotated traces")

errors = []
for t in traces:
    for step in t["steps"]:
        if step.get("correctness") == "correct":
            continue
        errors.append({
            "instance_id": t["instance_id"],
            "step_index": step.get("step_index"),
            "category": step.get("error_type"),
            "subtype": step.get("error_subtype"),
            "severity": step.get("severity"),
        })
 
error_df = pd.DataFrame(errors)
print(f"Total errors: {len(error_df)}\n")
print("By category:\n", error_df["category"].value_counts(), "\n")
print("Top subtypes:\n", error_df["subtype"].value_counts().head(10), "\n")
print("By severity:\n", error_df["severity"].value_counts())

import matplotlib.pyplot as plt
 
positions = []
for t in traces:
    n = len(t["steps"])
    if n < 2:
        continue
    for step in t["steps"]:
        if step.get("correctness") != "correct":
            positions.append(step["step_index"] / (n - 1))
 
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)

score_df = pd.DataFrame([
    {"instance_id": t["instance_id"], "score": t["score"]}
    for t in traces if t["score"] is not None
])
 
print(score_df["score"].describe())

WEIGHT = {"minor": 1, "major": 5, "critical": 10}
 
modes = (
    error_df.assign(w=error_df["severity"].map(WEIGHT))
    .groupby(["category", "subtype"])
    .agg(count=("w", "size"), avg_weight=("w", "mean"))
)
modes["impact"] = modes["count"] * modes["avg_weight"]
 
print("By frequency:\n", modes.sort_values("count", ascending=False).head(10).to_string())
print("\nBy impact:\n", modes.sort_values("impact", ascending=False).head(10).to_string())