调试 Agent：当你不知道它为什么做了那件事

任务：帮我把 src/utils.py 里的 parse_date 函数改成支持 ISO 8601 格式

Agent 实际行为：
  Step 1: read_file("src/utils.py")          ← 正确
  Step 2: 分析代码，找到 parse_date 函数      ← 正确
  Step 3: write_file("src/utils.py", ...)    ← 写了修改后的代码
  Step 4: run_command("python -m pytest")     ← 测试失败
  Step 5: read_file("src/utils.py")          ← 重新读取
  Step 6: write_file("src/utils.py", ...)    ← 又写了一版
  Step 7: run_command("python -m pytest")     ← 还是失败
  Step 8: ...反复尝试...
  Step 15: 达到最大迭代次数，任务失败

import time
import uuid
from dataclasses import dataclass, field


@dataclass
class Span:
    """一个执行单元——可以是 LLM 调用、工具执行或其他操作。"""
    span_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    name: str = ""
    span_type: str = ""       # llm_call, tool_call, guardrail_check
    parent_id: str | None = None
    start_time: float = 0
    end_time: float = 0
    status: str = "running"   # running, success, error
    attributes: dict = field(default_factory=dict)
    events: list[dict] = field(default_factory=list)

    @property
    def duration_ms(self) -> float:
        if self.end_time and self.start_time:
            return (self.end_time - self.start_time) * 1000
        return 0


class Tracer:
    """
    Agent 执行追踪器。
    记录一次完整执行的所有 Span，支持嵌套结构。
    """

    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())[:12]
        self.spans: list[Span] = []
        self._active_span: Span | None = None

    def start_span(
        self,
        name: str,
        span_type: str,
        attributes: dict | None = None,
    ) -> Span:
        """开始一个新的 Span。"""
        span = Span(
            name=name,
            span_type=span_type,
            parent_id=self._active_span.span_id if self._active_span else None,
            start_time=time.time(),
            attributes=attributes or {},
        )
        self.spans.append(span)
        self._active_span = span
        return span

    def end_span(self, status: str = "success", attributes: dict | None = None):
        """结束当前活跃的 Span。"""
        if self._active_span:
            self._active_span.end_time = time.time()
            self._active_span.status = status
            if attributes:
                self._active_span.attributes.update(attributes)
            # 回到父 Span
            parent_id = self._active_span.parent_id
            self._active_span = next(
                (s for s in self.spans if s.span_id == parent_id), None
            )

    def add_event(self, name: str, attributes: dict | None = None):
        """在当前 Span 上添加一个事件。"""
        if self._active_span:
            self._active_span.events.append({
                "name": name,
                "timestamp": time.time(),
                "attributes": attributes or {},
            })

    def get_summary(self) -> dict:
        """生成 Trace 的摘要统计。"""
        total_llm_time = sum(
            s.duration_ms for s in self.spans if s.span_type == "llm_call"
        )
        total_tool_time = sum(
            s.duration_ms for s in self.spans if s.span_type == "tool_call"
        )
        total_tokens = sum(
            s.attributes.get("total_tokens", 0)
            for s in self.spans if s.span_type == "llm_call"
        )
        error_spans = [s for s in self.spans if s.status == "error"]

        return {
            "trace_id": self.trace_id,
            "total_spans": len(self.spans),
            "llm_calls": len([s for s in self.spans if s.span_type == "llm_call"]),
            "tool_calls": len([s for s in self.spans if s.span_type == "tool_call"]),
            "total_llm_time_ms": round(total_llm_time, 1),
            "total_tool_time_ms": round(total_tool_time, 1),
            "total_tokens": total_tokens,
            "errors": len(error_spans),
            "error_details": [
                {"span": s.name, "error": s.attributes.get("error")}
                for s in error_spans
            ],
        }

async def run_agent_with_tracing(
    user_message: str,
    tools: dict,
    tracer: Tracer,
) -> str:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.append({"role": "user", "content": user_message})

    for iteration in range(MAX_ITERATIONS):
        # ===== LLM 调用 Span =====
        tracer.start_span(
            name=f"llm_call_{iteration + 1}",
            span_type="llm_call",
            attributes={"iteration": iteration + 1},
        )

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOL_SCHEMAS,
        )

        msg = response.choices[0].message
        usage = response.usage

        tracer.end_span(
            status="success",
            attributes={
                "input_tokens": usage.prompt_tokens,
                "output_tokens": usage.completion_tokens,
                "total_tokens": usage.total_tokens,
                "thought": msg.content[:200] if msg.content else None,
                "finish_reason": response.choices[0].finish_reason,
            },
        )

        # 处理工具调用
        if msg.tool_calls:
            for tc in msg.tool_calls:
                tool_name = tc.function.name
                tool_args = json.loads(tc.function.arguments)

                # ===== 工具调用 Span =====
                tracer.start_span(
                    name=f"tool_{tool_name}",
                    span_type="tool_call",
                    attributes={
                        "tool_name": tool_name,
                        "tool_args": tool_args,
                    },
                )

                try:
                    result = await tools[tool_name](**tool_args)
                    tracer.end_span(
                        status="success",
                        attributes={"result_preview": str(result)[:200]},
                    )
                except Exception as e:
                    tracer.end_span(
                        status="error",
                        attributes={"error": str(e)},
                    )
                    result = {"error": str(e)}

                # ... 把结果追加到 messages ...

        if response.choices[0].finish_reason == "stop":
            break

    return msg.content

import logging
import json
from datetime import datetime

# 使用结构化日志格式，便于后续查询和分析
logger = logging.getLogger("agent")


class AgentLogger:
    """
    Agent 专用日志器。
    按 Agent 的执行阶段记录不同粒度的信息。
    """

    def __init__(self, session_id: str):
        self.session_id = session_id

    def log_thought(self, iteration: int, thought: str):
        """记录模型的推理文本——这是调试推理问题的核心信息。"""
        logger.info(json.dumps({
            "event": "thought",
            "session_id": self.session_id,
            "iteration": iteration,
            "thought": thought,
            "timestamp": datetime.now().isoformat(),
        }, ensure_ascii=False))

    def log_tool_call(
        self,
        iteration: int,
        tool_name: str,
        tool_args: dict,
        result: dict,
        duration_ms: float,
    ):
        """记录工具调用的完整信息。"""
        # 对大结果做截断，避免日志膨胀
        result_str = json.dumps(result, ensure_ascii=False)
        if len(result_str) > 1000:
            result_preview = result_str[:1000] + "...[truncated]"
        else:
            result_preview = result_str

        logger.info(json.dumps({
            "event": "tool_call",
            "session_id": self.session_id,
            "iteration": iteration,
            "tool_name": tool_name,
            "tool_args": tool_args,
            "result_preview": result_preview,
            "result_success": result.get("success", True),
            "duration_ms": round(duration_ms, 1),
            "timestamp": datetime.now().isoformat(),
        }, ensure_ascii=False))

    def log_decision_point(
        self,
        iteration: int,
        description: str,
        chosen_action: str,
        alternatives: list[str] | None = None,
    ):
        """
        记录关键决策点——Agent 在多个选项中做了选择。
        这是调试"为什么走了那条路"的关键信息。
        """
        logger.info(json.dumps({
            "event": "decision",
            "session_id": self.session_id,
            "iteration": iteration,
            "description": description,
            "chosen": chosen_action,
            "alternatives": alternatives,
            "timestamp": datetime.now().isoformat(),
        }, ensure_ascii=False))

    def log_context_state(self, iteration: int, messages: list[dict]):
        """
        记录上下文状态——当前消息历史有多长、Token 预算还剩多少。
        这是调试"信息丢失"问题的关键信息。
        """
        total_chars = sum(len(str(m.get("content", ""))) for m in messages)
        logger.info(json.dumps({
            "event": "context_state",
            "session_id": self.session_id,
            "iteration": iteration,
            "message_count": len(messages),
            "estimated_chars": total_chars,
            "estimated_tokens": total_chars // 4,  # 粗略估算
            "timestamp": datetime.now().isoformat(),
        }, ensure_ascii=False))

    def log_error(self, iteration: int, error_type: str, error_msg: str):
        """记录错误。"""
        logger.error(json.dumps({
            "event": "error",
            "session_id": self.session_id,
            "iteration": iteration,
            "error_type": error_type,
            "error_msg": error_msg,
            "timestamp": datetime.now().isoformat(),
        }, ensure_ascii=False))

from dataclasses import dataclass, field
import time


@dataclass
class AgentMetrics:
    """
    Agent 运行指标收集器。
    在每次执行中更新，定期聚合查看趋势。
    """

    # 任务维度
    total_tasks: int = 0
    successful_tasks: int = 0
    failed_tasks: int = 0

    # 成本维度
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_cost_usd: float = 0

    # 效率维度
    total_iterations: int = 0
    total_tool_calls: int = 0
    total_duration_seconds: float = 0

    # 质量维度
    tool_errors: int = 0
    retry_count: int = 0
    max_iteration_hits: int = 0  # 触发最大迭代次数的次数

    @property
    def success_rate(self) -> float:
        return self.successful_tasks / max(self.total_tasks, 1)

    @property
    def avg_iterations(self) -> float:
        return self.total_iterations / max(self.total_tasks, 1)

    @property
    def avg_cost_per_task(self) -> float:
        return self.total_cost_usd / max(self.total_tasks, 1)

    @property
    def avg_duration_seconds(self) -> float:
        return self.total_duration_seconds / max(self.total_tasks, 1)

    @property
    def tool_error_rate(self) -> float:
        return self.tool_errors / max(self.total_tool_calls, 1)

    def report(self) -> dict:
        return {
            "tasks": {
                "total": self.total_tasks,
                "success_rate": f"{self.success_rate:.1%}",
            },
            "cost": {
                "total_usd": f"${self.total_cost_usd:.4f}",
                "avg_per_task": f"${self.avg_cost_per_task:.4f}",
                "total_tokens": self.total_input_tokens + self.total_output_tokens,
            },
            "efficiency": {
                "avg_iterations": f"{self.avg_iterations:.1f}",
                "avg_duration": f"{self.avg_duration_seconds:.1f}s",
                "tool_error_rate": f"{self.tool_error_rate:.1%}",
            },
            "warnings": {
                "max_iteration_hits": self.max_iteration_hits,
                "retry_count": self.retry_count,
            },
        }

import json
from dataclasses import dataclass


@dataclass
class ExecutionSnapshot:
    """一次完整执行的快照，包含复现所需的全部信息。"""
    session_id: str
    task: str
    system_prompt: str
    messages: list[dict]          # 完整的消息历史
    tool_results: list[dict]      # 每次工具调用的参数和返回值
    model: str
    temperature: float
    trace_summary: dict


class ReplayEngine:
    """
    执行重放引擎。
    两种模式：模拟重放（不调用真实 API）和真实重放（重跑 LLM）。
    """

    def save_snapshot(self, snapshot: ExecutionSnapshot, path: str):
        """保存执行快照到文件。"""
        data = {
            "session_id": snapshot.session_id,
            "task": snapshot.task,
            "system_prompt": snapshot.system_prompt,
            "messages": snapshot.messages,
            "tool_results": snapshot.tool_results,
            "model": snapshot.model,
            "temperature": snapshot.temperature,
            "trace_summary": snapshot.trace_summary,
        }
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load_snapshot(self, path: str) -> ExecutionSnapshot:
        """加载执行快照。"""
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        return ExecutionSnapshot(**data)

    def simulate_replay(self, snapshot: ExecutionSnapshot):
        """
        模拟重放：按时间顺序展示每一步的输入、推理和输出，
        不调用真实 API。用于分析问题发生的过程。
        """
        print(f"=== Replay: {snapshot.session_id} ===")
        print(f"Task: {snapshot.task}\n")

        tool_idx = 0
        for i, msg in enumerate(snapshot.messages):
            role = msg["role"]

            if role == "system":
                print(f"[SYSTEM] (prompt, {len(msg['content'])} chars)")

            elif role == "user":
                print(f"\n[USER] {msg['content'][:200]}")

            elif role == "assistant":
                if msg.get("content"):
                    print(f"\n[THOUGHT] {msg['content'][:300]}")
                if msg.get("tool_calls"):
                    for tc in msg["tool_calls"]:
                        fn = tc["function"]
                        print(f"[ACTION] {fn['name']}({fn['arguments'][:100]})")

            elif role == "tool":
                content = msg["content"]
                # 高亮错误
                if '"success": false' in content.lower():
                    print(f"[RESULT] ⚠️ ERROR: {content[:200]}")
                else:
                    print(f"[RESULT] ✓ {content[:200]}")

        print(f"\n=== Trace Summary ===")
        for k, v in snapshot.trace_summary.items():
            print(f"  {k}: {v}")

    async def live_replay(
        self,
        snapshot: ExecutionSnapshot,
        stop_at_iteration: int | None = None,
    ):
        """
        真实重放：用保存的工具返回值 mock 工具调用，
        但使用真实的 LLM API。用于验证 prompt 修改是否解决了问题。
        """
        # 使用保存的工具结果作为 mock
        tool_results_iter = iter(snapshot.tool_results)

        async def mock_tool(name: str, args: dict) -> dict:
            """用保存的结果替代真实工具调用。"""
            saved = next(tool_results_iter, None)
            if saved and saved["tool_name"] == name:
                return saved["result"]
            return {"error": f"No saved result for {name}"}

        # 用真实 LLM + mock 工具重跑
        # 这让你可以测试 prompt 修改的效果，
        # 而不需要重新执行真实的工具调用
        # ...

from langfuse import Langfuse

langfuse = Langfuse()

# 创建一个 Trace
trace = langfuse.trace(name="research_agent_run", input=user_task)

# 在 LLM 调用时记录 Span
span = trace.span(name="llm_call_1", input=messages)
response = await client.chat.completions.create(...)
span.end(output=response.choices[0].message.content)

# 在工具调用时记录 Span
tool_span = trace.span(name="web_search", input={"query": query})
result = await web_search(query)
tool_span.end(output=result)