日誌和可觀測性

應用上線之後，使用者在你看不見的地方使用它。某天有人反饋"你們的助手回答了一個完全錯誤的東西"，或者月底賬單比預期多了三倍，你打算怎麼查？

如果手裡只有使用者的問題和最終的回答，你幾乎無從下手：是檢索沒找到對的文件？是模型讀錯了？是某個工具報錯了？還是它在某一步走偏了，接下來越錯越遠？

可觀測性（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. 為什麼日誌裡不要記錄完整的使用者問題和回答？

使用者的輸入裡可能有個人資訊、密碼、公司機密，日誌一旦洩露或者被不該看的人看到，就會造成隱私問題。只記排查問題所需的部分，寫入前過濾敏感資訊，並設定保留期限和訪問許可權。