日志和可观测性

应用上线之后，用户在你看不见的地方使用它。某天有人反馈"你们的助手回答了一个完全错误的东西"，或者月底账单比预期多了三倍，你打算怎么查？

如果手里只有用户的问题和最终的回答，你几乎无从下手：是检索没找到对的文档？是模型读错了？是某个工具报错了？还是它在某一步走偏了，接下来越错越远？

可观测性（observability）说的就是：系统运行时留下足够的记录，让你事后能还原出它到底做了什么。对 AI 应用来说，最重要的记录是每一次模型调用和每一次工具调用。

要记什么

第 03 模块第 4 课的 call_llm 已经在记账了：每次调用写一行 JSON，包括时间、模型、耗时、词元和费用。那是一个好的开始，但对智能体来说还不够。智能体回答一个问题要调用好几次模型、好几次工具，这些记录要能串起来。

所以每条记录还要有：

• trace_id：同一个问题的所有记录，共用一个编号。凭它能把一个问题的全部过程找出来。
• span_id 和 parent_id：每条记录自己的编号，以及它的上级是谁。比如"回答问题"是上级，下面有三次模型调用和三次工具调用。
• kind 和 name：记录的类型（任务、模型调用、工具调用）和具体名字（哪个模型、哪个工具）。
• 输入和输出的摘要：工具的参数、返回内容的前几十个字、模型要求调用了哪些工具。
• 耗时、词元、费用、状态：成功、报错，还是工具返回了错误信息。

这套叫法（trace、span）来自分布式系统的追踪技术。现成的开源工具，比如 Langfuse（可以自己部署），以及 OpenTelemetry 这一套标准，都用同样的概念。先自己写一个最简单的，理解它在做什么。

一个几十行的追踪器

import json
import threading
import time
import uuid
from contextlib import contextmanager
from pathlib import Path

_lock = threading.Lock()
_local = threading.local()  # 记录当前线程正在进行的 span，用来自动找到上级


class Tracer:
    def __init__(self, path):
        self.path = Path(path)

    def _write(self, record):
        with _lock, self.path.open("a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    @contextmanager
    def span(self, kind, name, **attrs):
        """用法：with tracer.span("llm", "chat") as s: ...; s["tokens"] = 100"""
        parent = getattr(_local, "current", None)
        record = {
            "trace_id": parent["trace_id"] if parent else uuid.uuid4().hex[:12],
            "span_id": uuid.uuid4().hex[:8],
            "parent_id": parent["span_id"] if parent else None,
            "kind": kind,
            "name": name,
            "start": time.strftime("%Y-%m-%d %H:%M:%S"),
            **attrs,
        }
        _local.current = record
        start = time.time()
        try:
            yield record
            record.setdefault("status", "ok")
        except Exception as e:
            record["status"] = "error"
            record["error"] = f"{type(e).__name__}: {e}"
            raise
        finally:
            record["ms"] = round((time.time() - start) * 1000)
            _local.current = parent
            self._write(record)

用法是一个 with 语句：

with tracer.span("tool", "grep_docs", args={"keyword": "timeout"}) as s:
    result = grep_docs("timeout")
    s["result_chars"] = len(result)

with 块结束时，追踪器自动算出耗时，写一行 JSON。块里出了异常，也会记下来，状态是 error，然后照常把异常抛出去。

parent_id 是自动找的：追踪器用 threading.local() 记住"当前正在进行的是哪个 span"。在"回答问题"这个 span 里面开始的模型调用，它的上级自然就是"回答问题"。每个线程有自己的记录，所以多线程并发处理几个问题也不会串。

接到智能体上

不用改第 05 模块第 2 课的智能体代码，只要在外面包一层。模型调用：

class TracedModel(agent_loop.RealModel):
    """每次调用模型，记一个 llm 类型的 span。"""

    def __call__(self, messages, tools):
        with tracer.span("llm", self.model, input_messages=len(messages)) as s:
            message, usage = super().__call__(messages, tools)
            s.update(prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens,
                     cost=round(cost(usage), 6), tool_calls=[c.function.name for c in message.tool_calls or []])
            return message, usage

工具调用：

def traced(name, fn):
    """包装一个工具函数，每次调用记一个 tool 类型的 span。"""
    def wrapper(**kwargs):
        with tracer.span("tool", name, args=kwargs) as s:
            result = fn(**kwargs)
            s["result_chars"] = len(result)
            s["result_preview"] = result[:80]
            if result.startswith("错误"):
                s["status"] = "tool_error"
            return result
    return wrapper


for name, t in agent_loop.TOOLS.items():
    t["fn"] = traced(name, t["fn"])

工具返回"错误：……"时，函数本身并没有抛异常（第 05 模块讲过，错误要变成观察结果交给模型），所以要单独标记成 tool_error，否则日志里看起来一切正常。

最外层，每个问题开一个 task 类型的 span：

    for q in QUESTIONS:
        with tracer.span("task", "answer_question", question=q) as s:
            answer, stats = agent_loop.run_agent(model, q, verbose=False)
            s["answer_preview"] = (answer or "")[:60]

跑三个问题，然后只看日志

三个问题，最后一个的文件名是故意写错的（正确的是 timeouts.md）。跑完之后，程序只读日志文件做分析，就像事后排查问题时那样（完整代码在 code/06-production/traced_agent.py）：

日志共 25 条记录，最后一条：
{"trace_id": "173ee49bde3e", "span_id": "f9e2c3a5", "parent_id": null, "kind": "task", "name": "answer_question", "start": "2026-09-14 23:07:41", "question": "advanced/timeout.md 里写了什么？", "answer_preview": "`advanced/timeouts.md` 的内容如下（文件共 71 行；注意文件名是复数 timeouts，没有 `", "status": "ok", "ms": 5729}

「httpx 的超时分成哪几种？」 4006 毫秒，3 次模型调用，3 次工具调用，0.00104 美元
「httpx 怎么上传文件？」 5352 毫秒，3 次模型调用，6 次工具调用，0.00210 美元
「advanced/timeout.md 里写了什么？」 5729 毫秒，4 次模型调用，3 次工具调用，0.00135 美元

最慢的一步：llm deepseek-flash，3120 毫秒
工具报错次数：无

三个问题一共 25 条记录。按 trace_id 分组，每个问题的耗时、调用次数、费用一目了然。"怎么上传文件"调用了 6 次工具，比别的问题多一倍，费用也高一倍，打开它的记录就能看到它先后搜了 files=、multipart、upload 三个关键词，又读了三个文件。最慢的一步是一次模型调用，3 秒多，工具调用都只要几毫秒。

我原本以为第三个问题会留下一条"文件不存在"的工具报错，结果没有：智能体先调用了 list_docs 看了看有哪些文件，发现正确的名字是 timeouts.md，然后直接读了它，回答里还提醒了用户"注意文件名是复数"。这也是日志的价值：它记下的是真实发生的事，而不是你以为会发生的事。

用日志排查问题

有了这样的日志，常见问题的排查方式：

JSONL 文件简单、可靠，用 Python 几行代码就能分析，适合个人项目和早期的产品。量大了之后，可以把这些记录发到专门的工具里（比如 Langfuse），它会提供界面来浏览每一条 trace、按条件搜索、画统计图。概念是一样的。

注意隐私

日志里会记下用户的问题、工具的参数、回答的开头。如果用户在问题里写了自己的手机号、密码、公司内部信息，这些都会进入日志。

• 只记必要的内容。上面的代码只记了回答的前 60 个字、工具结果的前 80 个字，不记完整内容。
• 过滤敏感信息。在写日志之前，把看起来像密钥、手机号、身份证号的内容替换掉。下下一课"护栏"会写一个简单的检测函数。
• 设定保留期限。日志不要永远留着，比如只保留 30 天。
• 控制访问权限。能看日志的人，就能看到用户问了什么。

练习

1. 运行 traced_agent.py，然后自己写几行代码读取 traces.jsonl，找出每个问题里输入词元最多的那次模型调用。
2. 给 TracedModel 加一个字段，记录每次调用时输入的总字符数，看看智能体每一步的上下文是怎么增长的。
3. 故意制造一次工具报错（比如临时修改 read_doc，让它对某个文件总是返回"错误：……"），确认日志的统计能发现它。

自测

1. trace_id、span_id、parent_id 分别有什么用？

trace_id 把同一个任务（比如回答一个问题）的所有记录串在一起；span_id 是每一条记录自己的编号；parent_id 指向它的上级记录。有了这三个字段，就能从日志里还原出一个任务的完整结构：先做了什么，里面又调用了什么。

2. 工具返回"错误：没有这个文件"时，函数并没有抛出异常。日志里怎么才能发现它？

要在包装工具的代码里检查返回值，发现是错误信息时，单独把状态标记成 tool_error 之类的值。否则日志里这次调用的状态是正常的，统计错误时就会漏掉。

3. 为什么日志里不要记录完整的用户问题和回答？

用户的输入里可能有个人信息、密码、公司机密，日志一旦泄露或者被不该看的人看到，就会造成隐私问题。只记排查问题所需的部分，写入前过滤敏感信息，并设定保留期限和访问权限。