📌 前置知识：已完成第一课至第十一课
🎯 本课目标：构建运行时可观测性系统，让 Agent 的每个操作都有据可查
💡 核心概念：结构化日志 / 跨度（Span）/ 追踪（Trace）/ 指标（Metrics）

---

前言

前十一课，我们搭出了一个功能丰富的 Agent：

agent = Agent(model="qwen2.5:7b")

# 第八课：会规划
plan = agent.create_plan("写一篇博客")

# 第九课：会拆解
atomic = agent.create_atomic_action("写初稿")

# 第十课：会编排
graph = agent.create_aot_plan("写博客")
results = agent.execute_aot_plan(graph)

# 第十一课：会自测
agent.run_evals()

看起来很完整了，对不对？

但有一个致命问题：

你在部署运行期间，完全不知道它在干什么。

你调用 agent.chat("帮我订一张机票")，它返回了一个错误。是因为 LLM 调用超时了？是工具调用参数错了？还是记忆里存了错误的信息？

没有日志，你只能猜。

第十一课（评测）告诉你 Agent 在部署前是否工作。第十二课（遥测）告诉你部署期间到底发生了什么。

评测 = 发货前的质检。
遥测 = 飞行中的黑匣子。

---

一、为什么需要遥测？

1.1 没有遥测的调试是"盲猜"

想象你收到用户报告：“Agent 给了一个奇怪的答案。”

没有遥测，你的调试流程是：

1. 猜测可能是哪里出了问题
2. 加几个 print 语句
3. 重新跑一次
4. 如果复现不了 → 回到步骤 1
5. 如果复现了 → 看 print 输出
6. 修 bug，部署
7. 下次出问题 → 回到步骤 1

这就是盲猜。每一步都在靠直觉。

有了遥测，调试流程变成：

1. 让用户提供追踪 ID（trace_id）
2. 从日志里找到那个追踪的所有记录
3. 看 LLM 调用了什么、返回了什么、工具执行了什么
4. 定位问题
5. 修 bug，部署

区别：从"猜测"变成了"查记录"。

1.2 没有遥测你不知道系统健康度

你的 Agent 在生产环境中跑了一周。你如何回答这些问题：

• JSON 解析成功率是多少？
• 平均 LLM 调用延迟是多少？
• 工具调用失败率是多少？
• 重试次数有没有上升？

没有指标，这些都是未知数。你不知道系统是在平稳运行还是正在逐渐恶化。

1.3 什么是遥测？

遥测 = 结构化日志 + 追踪 + 指标。

结构化日志：每个事件按固定 schema 记录 → 机器可读
追踪：相关事件通过 trace_id 链接在一起 → 可调试
指标：聚合计数和统计 → 一眼看健康

这三个东西组合起来，就是"运行时可观测性"。

---

二、核心概念

2.1 结构化日志

结构化日志意味着 JSON 日志，而不是 print 语句。

# ❌ 非结构化的 print
print(f"LLM call took {duration}ms, success={success}")

# ✅ 结构化的 JSON 日志
{"event_type": "llm_call", "timestamp": "2024-01-15T10:30:00", "duration_ms": 1523, "success": true}

为什么 JSON 更好？

特性	print 日志	JSON 日志
可搜索	❌ grep “1523” 一个数字可能出现在任何地方	✅ grep “duration_ms” 精准定位
可解析	❌ 每行格式不同	✅ 统一 schema，json.loads 直接解析
可聚合	❌ 需要写正则提取数据	✅ 字段名固定，直接求和/平均
机器可读	❌ 人工看还行，程序处理难	✅ 天然结构，程序友好

2.2 跨度（Span）和追踪（Trace）

**跨度（Span）**是一个操作——一次 LLM 调用、一次工具执行、一次记忆访问。

**追踪（Trace）**是完整的智能体交互——由追踪 ID 链接在一起的多个跨度。

追踪（trace_id: "a1b2c3d4"）
├── 跨度 1: 🤖 LLM 调用 "什么是 Python？"  耗时 1523ms
├── 跨度 2: 🔧 工具请求 calculator(42, 7)    耗时 45ms
├── 跨度 3: ⚙️ 工具执行 → 294              耗时 3ms
├── 跨度 4: 🧠 记忆操作 "存入用户偏好"       耗时 10ms
└── 跨度 5: 🎯 决策 → "analyze"             耗时 312ms

当某个交互出问题时，你找到追踪 ID，就能看到完整的操作链——每个步骤、每次调用、每个参数、每次失败。

2.3 指标

指标是聚合的数字——不是单个事件，而是批量事件的统计。

┌──────────────────────────────────┐
│        指标面板                    │
├──────────────────────────────────┤
│ LLM 调用：      100 次             │
│   成功率：        97%              │
│   平均延迟：      1345ms           │
│   重试次数：      3               │
│ 工具调用：      42 次              │
│   成功率：        95.24%           │
│ 记忆操作：      28 次              │
└──────────────────────────────────┘

看到这个面板，你一眼就知道系统是否健康：

• 成功率下降 → 检查 Prompt 是否被修改了
• 延迟升高 → 检查模型或硬件是否有问题
• 重试增多 → 检查 JSON 解析逻辑是否需要加强

---

三、代码实现

3.1 数据模型：agent/telemetry.py

Span 数据类

@dataclass
class Span:
    """追踪中的单个操作。"""
    span_id: str
    trace_id: str
    event_type: str        # llm_call / tool_request / tool_execution / memory_op / decision
    timestamp: str
    duration_ms: float = None
    data: dict = None
    error: str = None

每个 Span 都包含：

• span_id：自己的唯一标识
• trace_id：所属追踪的标识（多个 Span 共享）
• event_type：事件类型，决定如何处理这个 Span
• timestamp：ISO 格式时间戳
• duration_ms：操作耗时（毫秒）
• data：事件相关的业务数据
• error：如果失败，记录错误信息

Metrics 数据类

@dataclass
class Metrics:
    """智能体聚合指标。"""
    llm_calls: int = 0
    llm_failures: int = 0
    llm_retries: int = 0
    tool_calls: int = 0
    tool_failures: int = 0
    memory_ops: int = 0
    total_latency_ms: float = 0.0

    @property
    def avg_latency_ms(self) -> float:
        return self.total_latency_ms / self.llm_calls if self.llm_calls > 0 else 0.0

    @property
    def llm_success_rate(self) -> float:
        return 1 - (self.llm_failures / self.llm_calls) if self.llm_calls > 0 else 0.0

    @property
    def tool_success_rate(self) -> float:
        return 1 - (self.tool_failures / self.tool_calls) if self.tool_calls > 0 else 0.0

注意三个 computed property——它们不是手动更新的，而是由 llm_calls / llm_failures / total_latency_ms 等基础计数自动计算出来的。 这确保指标永远一致。

Telemetry 类

class Telemetry:
    def __init__(self, log_file: str = "agent_telemetry.jsonl"):
        self.log_file = log_file
        self.current_trace_id = None
        self.metrics = Metrics()
        self._spans: list[Span] = []   # 内存缓存

    def start_trace(self) -> str:
        """开始新追踪。"""
        self.current_trace_id = str(uuid4())[:8]
        return self.current_trace_id

    def log_llm_call(self, prompt_length, response_length,
                     duration_ms, success=True, retries=0, error=None):
        """记录 LLM 调用。"""
        span = self._make_span(
            event_type="llm_call",
            duration_ms=duration_ms,
            data={
                "prompt_length": prompt_length,
                "response_length": response_length,
                "success": success,
                "retries": retries,
            },
            error=error,
        )
        self._write_span(span)

        # 更新指标
        self.metrics.llm_calls += 1
        self.metrics.total_latency_ms += duration_ms
        self.metrics.llm_retries += retries
        if not success:
            self.metrics.llm_failures += 1

设计要点：

① _make_span() 统一创建 Span

def _make_span(self, event_type, duration_ms, data=None, error=None) -> Span:
    return Span(
        span_id=str(uuid4())[:8],
        trace_id=self.current_trace_id or "no-trace",
        event_type=event_type,
        timestamp=datetime.now().isoformat(),
        duration_ms=round(duration_ms, 2),
        data=data,
        error=error,
    )

所有记录方法通过 _make_span() 统一创建 Span，确保：

• 字段格式一致（duration_ms 统一 round(..., 2)）
• trace_id 统一处理（没有活跃追踪时标记为 "no-trace"）
• 时间戳格式一致

② _write_span() 写 JSONL

def _write_span(self, span: Span) -> None:
    self._spans.append(span)                     # 内存缓存
    with open(self.log_file, "a", encoding="utf-8") as f:
        f.write(json.dumps(asdict(span), ensure_ascii=False) + "\n")  # 追加

JSONL = JSON Lines。 每行一个 JSON 对象，用换行分隔。比 JSON 数组好的是——可以逐行追加，不需要读整个文件。

③ 记录 + 指标同步

每个 log_* 方法做两件事：

1. 写 Span 到日志
2. 更新 Metrics 计数

这是一个事务——不会出现"有日志没指标"或"有指标没日志"的情况。

3.2 追踪管理器：trace 上下文管理器

class trace:
    """追踪上下文管理器，确保作用域内所有操作共享同一 trace_id。"""
    def __init__(self, telemetry: Telemetry):
        self.telemetry = telemetry
        self._previous_trace_id = None

    def __enter__(self) -> str:
        self._previous_trace_id = self.telemetry.current_trace_id
        return self.telemetry.start_trace()   # 开始新追踪

    def __exit__(self, *args):
        self.telemetry.current_trace_id = self._previous_trace_id  # 恢复上级追踪

用法：

telemetry = Telemetry()

with trace(telemetry) as trace_id:
    # 这里的所有操作自动关联到 trace_id
    telemetry.log_llm_call(...)
    telemetry.log_tool_request(...)

使用 with 块的好处：

• 作用域清晰：进入块开始追踪，离开块自动恢复
• 支持嵌套：内层块的追踪结束后，自动回到外层块的追踪
• 避免遗忘：不会忘记 reset trace_id

3.3 Agent 仪表化

在 agent/agent.py 中，核心方法都注入了遥测记录：

class Agent:
    def __init__(self, ..., log_file="agent_telemetry.jsonl"):
        from agent.telemetry import Telemetry
        self.telemetry = Telemetry(log_file=log_file)
        self.memory = AgentMemory(telemetry=self.telemetry)

    def generate_structured(self, user_input, schema):
        self._ensure_trace()         # 自动开始追踪
        # ... 调用 LLM ...
        self._record_llm_call(       # 自动记录耗时
            start, prompt, text, success=parsed is not None, retries=retries
        )
        return parsed

    def request_tool(self, user_input):
        # ... LLM 调用 ...
        self.telemetry.log_tool_request(tool_name, arguments, duration)
        return result

    def execute_tool_call(self, tool_call):
        # ... 执行工具 ...
        self.telemetry.log_tool_execution(tool_name, result, duration, success)
        return result

关键辅助方法：

def _ensure_trace(self) -> None:
    """确保当前有活跃的追踪。"""
    if self.telemetry and not self.telemetry.current_trace_id:
        self.telemetry.start_trace()

def _record_llm_call(self, start_time, prompt_text, response_text,
                     success=True, retries=0, error=None) -> None:
    """统一记录 LLM 调用耗时。"""
    duration = (time.time() - start_time) * 1000
    self.telemetry.log_llm_call(
        prompt_length=len(prompt_text),
        response_length=len(response_text),
        duration_ms=duration,
        success=success,
        retries=retries,
        error=error,
    )

设计模式：

每个方法调用前 → _ensure_trace()     保证有追踪 ID
每个方法记录 start → time.time()
调用 LLM / 工具 / 记忆 → 实际业务逻辑
结束后 → _record_*() / log_*()        记录耗时和结果

这个模式确保了每个方法都自动记录，不需要手动管理追踪或计时。

3.4 查询与调试

def get_trace(self, trace_id: str) -> list[Span]:
    """按追踪 ID 获取所有跨度。"""
    # 先查内存缓存，再去日志文件读取

def get_failed_spans(self) -> list[Span]:
    """获取所有失败的跨度。"""

def print_trace_detail(self, trace_id: str) -> None:
    """打印指定追踪的完整信息（供调试用）。"""

def print_summary(self) -> None:
    """打印指标摘要。"""

---

四、运行示例

4.1 场景 1：手动遥测

from agent.telemetry import Telemetry

telemetry = Telemetry("demo.jsonl")
trace_id = telemetry.start_trace()

# 记录 LLM 调用
telemetry.log_llm_call(
    prompt_length=85, response_length=230,
    duration_ms=1523, success=True
)

# 记录工具调用
telemetry.log_tool_request("calculator", {"a": 42, "b": 7, "operation": "multiply"})
telemetry.log_tool_execution("calculator", 294, duration_ms=3, success=True)

# 记录记忆操作
telemetry.log_memory_op("add", "用户偏好：Python 开发者")

# 查看摘要
telemetry.print_summary()

输出：

==================================================
  TELEMETRY SUMMARY
==================================================

  LLM Calls:      1
    Success Rate:  100.00%
    Avg Latency:   1523ms
    Retries:       0
  Tool Calls:     1
    Success Rate:  100.00%
  Memory Ops:     1

  Log File:       demo.jsonl (512 B)
  Active Traces:  1
==================================================

4.2 场景 2：仪表化 Agent

from agent.agent import Agent, calculator

agent = Agent(model="qwen2.5:7b", log_file="agent_telemetry.jsonl")

# 注册工具
agent.register_tool(
    name="calculator", func=calculator,
    description="数学计算器",
    parameters={"a": {...}, "b": {...}, "operation": {...}}
)

# 自动记录遥测
result1 = agent.generate_structured("什么是 Python？", '{"answer": "string"}')
# → 自动记录 LLM 调用耗时

tool_call = agent.request_tool("请计算 42 * 7")
# → 自动记录工具请求

result2 = agent.execute_tool_call(tool_call)
# → 自动记录工具执行

# 查看汇总
agent.show_telemetry()

输出：

==================================================
  TELEMETRY SUMMARY
==================================================

  LLM Calls:      3
    Success Rate:  100.00%
    Avg Latency:   1245ms
    Retries:       0
  Tool Calls:     2
    Success Rate:  100.00%
  Memory Ops:     1

  Log File:       agent_telemetry.jsonl (1.5 KB)
  Active Traces:  1
==================================================

4.3 场景 3：追踪调试

from agent.telemetry import Telemetry, trace

telemetry = Telemetry()

# 一个正常的交互
with trace(telemetry) as tid1:
    telemetry.log_llm_call(100, 200, 800, success=True)
    telemetry.log_tool_execution("calculator", 15, duration_ms=2, success=True)

# 一个有故障的交互
with trace(telemetry) as tid2:
    telemetry.log_llm_call(200, 0, 3000, success=False,
                           error="JSON 解析失败：输出不是有效 JSON")
    telemetry.log_tool_execution("calculator", None, duration_ms=1, success=False,
                                 error="division by zero")

# 调试失败的追踪
telemetry.print_trace_detail(tid2)

输出（故障追踪）：

=======================================================
  TRACE: 3b81de54
=======================================================
  Spans: 2

  1. 🤖 llm_call ❌
     Span:  69968da2
     Time:  2026-05-03T13:09:57
     Dur:   3000ms
     Error: JSON 解析失败：输出不是有效 JSON

  2. ⚙️ tool_execution ❌
     Span:  36737726
     Time:  2026-05-03T13:09:57
     Dur:   1ms
     Error: division by zero

每个失败的事件都标红 ❌，一目了然。这就是"飞行记录仪"的价值——坠机了，你有黑匣子可以查。

4.4 场景 4：指标监控

agent = Agent(model="qwen2.5:7b")

# 运行 20 次结构化输出
for i in range(20):
    result = agent.generate_structured(
        f"生成数字 {i} 的 JSON",
        '{"number": "int", "squared": "int"}'
    )

# 看指标
m = agent.telemetry.metrics
print(f"LLM 调用次数：{m.llm_calls}")
print(f"LLM 成功率：{m.llm_success_rate * 100:.2f}%")
print(f"平均延迟：{m.avg_latency_ms:.0f}ms")
print(f"总重试次数：{m.llm_retries}")

输出：

LLM 调用次数：20
LLM 成功率：95.00%
平均延迟：1567ms
总重试次数：1

看一眼就知道：20 次调用里 1 次失败（95% 成功率），1 次重试。如果这个数字逐渐恶化（90% → 80% → 70%），说明系统在退化，需要介入。

---

五、与前课的关系

5.1 第十一课 vs 第十二课

	第十一课（评测）	第十二课（遥测）
运行时机	部署前	部署期间
输入	已知的黄金数据集	未知的用户真实输入
输出	通过/失败二元结果	结构化日志 + 聚合指标
用途	预防回归	支持调试
面向	开发者（开发期）	运维 / 开发者（运行期）
数据量	小（几十个用例）	大（持续增长）

它们是互补关系：

评测防止坏代码发货。遥测帮助你理解已发货代码在做什么。

发货前 ✅ → 评测通过 → 发货后 → 遥测监工

5.2 本课并未引入新工具

回顾本课的所有实现：

Telemetry.log_llm_call()   = time.time() + json.dumps() + file.write()
Telemetry.start_trace()    = uuid4() + 字符串拼接
Metrics.llm_success_rate   = 1 - (failures / calls)     ← 小学数学

没有任何外部依赖。 没有 OpenTelemetry、没有 Prometheus、没有 Grafana。

遥测的核心就是结构化日志 + 聚合计数。技术上极其简单，价值上极其巨大。

---

六、关键洞察

6.1 遥测只是结构化日志

没有魔法。你在写入 JSON 到文件。

力量在于：

• 一致的 schema：每个事件都知道怎么解析
• 追踪 ID 链接：相关事件可以跨时间、跨模块关联
• 聚合指标：从海量事件中提取关键数字

6.2 追踪是你的调试超能力

当用户报告"Agent 给了一个奇怪的答案"时，你：

1. 获取追踪 ID（从用户消息或应用日志）
2. 查该追踪的所有跨度（filter by trace_id）
3. 逐步看发生了什么（LLM 说了什么 → 工具返回了什么 → 决策选了什么）

没有追踪，你只能猜测。有追踪，你就有证据。

6.3 指标告诉你系统健康

看一眼指标就知道是否有问题：

成功率从 98% 降到 85%？ → 检查 Prompt 是否被改了
延迟从 1s 涨到 3s？     → 检查模型或硬件
重试从 0 涨到 10？       → 检查 JSON 解析逻辑

指标是"体温计"。不需要深入看每个 Span，看一眼数字就知道是不是发烧了。

6.4 从简单开始

本课的实现记录到文件。开始这就够了。

以后你可以随时升级：

• 文件 → 数据库（SQLite / PostgreSQL）
• 手动查看 → 实时仪表板（Grafana）
• 无告警 → 阈值告警（当 success_rate < 90% 时自动通知）

但从文件开始。够用就行。

---

七、常见问题

Q：日志文件越来越大了怎么办？

A：轮转日志。每天或每小时一个新文件，用日期命名，保留最近 7 天。或者只记录失败事件（成功事件在指标里已经有统计了）。

# 只用日期分文件
log_file = f"agent_telemetry_{datetime.now().strftime('%Y%m%d')}.jsonl"

Q：我找不到需要的追踪怎么办？

A：在面向用户的错误消息里包含追踪 ID。这样用户报告问题时，自动带着 trace_id，你直接查就行。

用户看到：❌ 出错了（错误代码：TID-a1b2c3d4）
你在日志：grep "a1b2c3d4" agent_telemetry.jsonl

Q：遥测会不会拖慢 Agent？

A：本课的实现是同步写文件，确实有 I/O 开销。对于原型和中小规模够用。如果要优化：

• 异步写入（用队列攒一批，再 flush）
• 减少数据量（截断长 response 字段）
• 采样（只记录 10% 的正常事件，100% 记录失败事件）

Q：JSONL 和 JSON 数组有什么区别？

A：JSONL 每行一个独立 JSON 对象。JSON 数组是整个文件一个 [...]。JSONL 的优点：

• 可以追加：open("file", "a") 直接写，不需要读整个文件
• 可以逐行读取：grep 能直接过滤，tail -f 能实时查看
• 部分损坏不丢全部：如果文件被截断了，只会丢最后一行

Q：trace 上下文管理器怎么保证不会忘记？

A：不用保证。让 _ensure_trace() 自动兜底——每个方法调用时检查有没有 trace_id，没有就自动创建。大部分情况下用户甚至不需要手动调用 start_trace()。

---

八、十二课演进线

从第一课到第十二课，一个 Agent 的完整构建路径：

第一课：启动 OLLama 跑通 LLM
第二课：有角色了 + 能多轮对话
第三课：能输出 JSON 了 + 验证 + 重试
第四课：能做选择了（意图理解 → 动作路由）
第五课：能调用工具了（选择 + 参数 + 安全执行）
第六课：能循环了（Agent Loop + 状态追踪）
第七课：能记住了（跨对话记忆）
第八课：能规划了（目标 → 步骤 → 执行）
第九课：能拆解了（模糊步骤 → 原子动作）
第十课：能编排了（依赖图 + 拓扑排序）
第十一课：能自测了（回归测试 + 黄金数据集）
第十二课：能自报了（结构化日志 + 追踪 + 指标）

从"裸模型"到一个完整的、可观测的、可测试的智能体系统。

---

九、系列总结

恭喜！你已经完成了全部核心课程。

你的 Agent 现在拥有：

能力	实现课程	核心机制
对话	第一、二课	LLM + System Prompt + 历史
结构化输出	第三课	JSON + extract_json + 重试
决策	第四课	意图分类 + 动作路由
工具调用	第五课	工具选择 + 参数 + 安全执行
循环执行	第六课	Agent Loop + 状态 + 终止条件
记忆	第七课	存储 + 检索 + 管理
规划	第八课	目标 → 步骤 → 验证 → 执行
原子动作	第九课	模糊步骤 → 带参数的动作
任务编排	第十课	依赖图 + 拓扑排序
回归测试	第十一课	黄金数据集 + 自动化评测
运行时可观测性	第十二课	结构化日志 + 追踪 + 指标

这个系列的核心哲学：

每个能力都是同样的模式——让 LLM 输出结构化数据，然后你的代码来验证和处理。

没有黑盒，没有魔法，每一行代码你都理解。

最后一句：评测防止坏代码发货。遥测帮助你理解已发货代码在做什么。

两者缺一不可。

---

完整代码获取

本课涉及的完整代码包括：

• agent/telemetry.py——遥测系统核心（Span + Metrics + Telemetry + trace 上下文管理器）
• agent/agent.py——Agent 类（新增 show_telemetry() 和 analyze_trace()，全部方法仪表化）
• complete_example.py——演示模式（4 个场景）+ 交互模式

关注公众号「开源情报局」，回复「Agent」获取。

也可以到 GitHub 查看完整源码：https://github.com/你的用户名/handcraft-ai-agent

---

标签

#Python #AI Agent #LLM #遥测 #可观测性 #结构化日志 #追踪 #Ollama #Qwen #大模型 #手搓Agent

---

本文为《手搓 AI Agent 从 0 到 1》系列教程第 12 课，也是完结篇。