【从0搭建AI智能体·10】Agent 可观测性：日志、链路追踪与成本监控

📚 本文是《从 0 搭建你的 AI 智能体》专栏第 10 篇。
上一篇：《多智能体协作实战》

标签：可观测性 日志 链路追踪 成本监控 AI Agent LLM运维

📌 前言：你的 Agent 是个「黑盒」，直到月底账单吓你一跳

前面九篇，我们把 Agent 越搭越强。但有个残酷的现实：上线之后，你其实不知道它在干什么。

• 用户说「它答错了」，你复现不出来——不知道当时的输入、检索到了什么、调了哪个工具；
• 月底收到 API 账单，金额远超预期，但你说不清钱花在哪——哪个用户、哪个功能烧的？
• 有时候响应慢得要命，你不知道卡在哪——是模型慢、检索慢、还是工具慢？

这些问题的根源，是 Agent 天然是个「黑盒」：一个请求进去，中间经过模型、检索、工具、多轮调用，最后吐个结果出来，中间过程全看不见。可观测性（Observability） 就是给这个黑盒装上「仪表盘」——把每一步都记录、可查、可统计。

这一篇讲怎么给 Agent 加上日志、链路追踪、成本监控三件套，让黑盒变白盒。这也是本专栏「从能跑到能运维」的关键一环——这里我们会把第 11 篇 Linux 那套监控思路接回来。

💡 本文适合谁：Agent 已经上线或即将上线、需要排障和控成本的开发者。以 Python 为例，从零实现，也介绍现成工具。阅读约 13 分钟。

目录

1. 可观测性的三大支柱
2. 支柱一：结构化日志（最基础也最重要）
3. 支柱二：链路追踪（Trace 一次完整请求）
4. 支柱三：成本监控（Token 就是钱）
5. 用装饰器优雅地埋点
6. 实时告警：花超了/错太多就通知你
7. 现成工具：LangSmith / Langfuse
8. 完整可运行 Demo：一个自带仪表盘的 Agent
9. 常见坑与建议
10. FAQ
11. 总结

---

① 可观测性的三大支柱

业界把可观测性拆成三块，套到 Agent 上就是：

支柱	回答什么问题	对应 Agent
Logs（日志）	发生了什么？	每次调用的输入、输出、错误
Traces（追踪）	一次请求的完整链路？	模型→检索→工具→多轮的全过程
Metrics（指标）	整体数值趋势？	Token 消耗、延迟、错误率、QPS

这三块层层递进：日志是原料，追踪把一次请求的日志串成链路，指标是统计出来的趋势。下面逐个落地。

---

② 支柱一：结构化日志（最基础也最重要）

普通 print 是可观测性的敌人——没法检索、没法统计。要用结构化日志（JSON 格式），每条日志是一个带字段的对象，机器可解析。

import json, logging

logging.basicConfig(filename="agent.log", level=logging.INFO)


def log_event(event, **fields):
    """记录一条结构化日志（JSON 一行）"""
    # 注意：不记录敏感信息（密钥、完整对话可按需脱敏）
    record = {"event": event, **fields}
    logging.info(json.dumps(record, ensure_ascii=False))


# 用法：每个关键节点打点
def call_llm(messages, user_id):
    log_event("llm_request", user_id=user_id, msg_count=len(messages))
    try:
        resp = client.chat.completions.create(model=MODEL, messages=messages)
        reply = resp.choices[0].message.content
        log_event("llm_response", user_id=user_id,
                  prompt_tokens=resp.usage.prompt_tokens,
                  completion_tokens=resp.usage.completion_tokens)
        return reply
    except Exception as e:
        log_event("llm_error", user_id=user_id, error=str(e))   # 错误必记
        raise

日志长这样（每行一个 JSON，可被 ELK、Loki 等直接采集分析）：

{"event": "llm_request", "user_id": "u123", "msg_count": 4}
{"event": "llm_response", "user_id": "u123", "prompt_tokens": 320, "completion_tokens": 85}

🎯 结构化 vs 纯文本的天壤之别：结构化日志能直接查询「user_id=u123 今天的所有请求」「所有 llm_error」「completion_tokens > 1000 的请求」。纯文本 print 只能靠 grep 硬翻，出了事根本查不动。

🚨 安全红线：日志里别记密钥、别记完整敏感对话。记 user_id、token 数、状态就够排障了。要记内容也先脱敏。（呼应第 1 篇的安全规范）

---

③ 支柱二：链路追踪（Trace 一次完整请求）

一次 Agent 请求，可能经过：判断→检索→调工具→再调模型……好几步。链路追踪就是给每次请求一个唯一的 trace_id，把这一路的所有日志串起来，让你能还原「这一个请求」的完整过程。

import uuid, time


def traced_agent(user_input, user_id):
    trace_id = str(uuid.uuid4())[:8]        # 本次请求的唯一标识
    t0 = time.time()
    log_event("trace_start", trace_id=trace_id, user_id=user_id, input=user_input[:100])

    # 每一步都带上同一个 trace_id
    log_event("step", trace_id=trace_id, name="retrieve", detail="检索知识库")
    # ... 检索 ...

    log_event("step", trace_id=trace_id, name="llm_call", detail="调用模型")
    # ... 调模型 ...

    log_event("trace_end", trace_id=trace_id,
              duration_ms=int((time.time() - t0) * 1000))    # 总耗时

之后按 trace_id 一查，这次请求的每一步、每步耗时、在哪报错，一目了然：

{"event":"trace_start","trace_id":"a3f2c1e0","name":...}
{"event":"step","trace_id":"a3f2c1e0","name":"retrieve",...}
{"event":"step","trace_id":"a3f2c1e0","name":"llm_call",...}
{"event":"trace_end","trace_id":"a3f2c1e0","duration_ms":2840}

🎯 trace_id 是排障神器：用户报「刚才那次答错了」，你让他给个请求 ID（或按时间+user_id 定位 trace_id），一查就能还原当时的完整现场——检索到了啥、调了啥工具、模型看到的上下文是什么。没有 trace，全靠猜。

---

④ 支柱三：成本监控（Token 就是钱）

大模型按 Token 计费，Token 就是真金白银。不监控成本，月底账单能让你心梗。核心是记录每次调用的 Token 并累计。

# 各模型的价格（每 1K token 的价格，按你的实际计费改）
PRICE = {"standard-agent-v1": {"input": 0.001, "output": 0.002}}


def track_cost(model, usage, user_id):
    """记录单次调用成本"""
    p = PRICE.get(model, {"input": 0, "output": 0})
    cost = (usage.prompt_tokens / 1000 * p["input"] +
            usage.completion_tokens / 1000 * p["output"])
    log_event("cost", user_id=user_id, model=model,
              prompt_tokens=usage.prompt_tokens,
              completion_tokens=usage.completion_tokens,
              cost_yuan=round(cost, 6))
    return cost

有了这些日志，你就能统计出关键成本指标：

指标	怎么算	用途
单请求成本	一个 trace 内所有 cost 之和	发现「烧钱」的功能
单用户成本	按 user_id 聚合	发现异常高消耗用户（薅羊毛？）
日/月总成本	按时间聚合	对账、预算控制
平均每轮 Token	completion_tokens 均值	优化 Prompt、发现啰嗦

💡 最容易烧钱的三个地方：① 对话历史无限增长（每轮重发全部，越聊越贵）→ 用第1/6篇的裁剪；② 多 Agent 无节制调用（第9篇）；③ 检索结果整坨塞入（第2/7篇）。监控能帮你精准定位是哪个在烧钱。

---

⑤ 用装饰器优雅地埋点

到处手写 log_event 很啰嗦。用装饰器把日志、计时、成本统一封装，业务代码保持干净：

import time, functools


def observe(name):
    """一个装饰器，自动记录函数的调用、耗时、异常"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            t0 = time.time()
            log_event("call_start", name=name)
            try:
                result = func(*args, **kwargs)
                log_event("call_end", name=name,
                          duration_ms=int((time.time() - t0) * 1000), status="ok")
                return result
            except Exception as e:
                log_event("call_end", name=name,
                          duration_ms=int((time.time() - t0) * 1000),
                          status="error", error=str(e))
                raise
        return wrapper
    return decorator


# 用法：一行注解就给函数加上了可观测性
@observe("retrieve")
def retrieve(query):
    ...

@observe("llm_call")
def call_llm(messages):
    ...

🎯 装饰器的好处：埋点和业务解耦。想给哪个环节加监控，加一行 @observe(...) 即可，不污染业务逻辑，也方便统一改造。

---

⑥ 实时告警：花超了/错太多就通知你

光记录还不够，出问题得主动通知你，而不是等你去翻日志。这里就把第 11 篇 Linux 告警那套思路接回来了——只是监控对象从「CPU/内存」换成了「成本/错误率」。

# 简单的阈值告警（生产可结合定时任务周期性统计）
DAILY_BUDGET = 100.0          # 每日预算（元）
_today_cost = {"sum": 0.0}


def check_budget(cost):
    _today_cost["sum"] += cost
    if _today_cost["sum"] > DAILY_BUDGET * 0.8:      # 达 80% 预警
        send_alert(f"⚠️ 今日 AI 成本已达 {_today_cost['sum']:.2f} 元，接近预算 {DAILY_BUDGET} 元")


def send_alert(msg):
    """复用第11篇的告警通道：邮件 / 钉钉 / 企业微信 Webhook"""
    log_event("alert", message=msg)
    # requests.post(WEBHOOK_URL, json={"text": msg})   # 发到钉钉/企业微信
    print(msg)

常见的告警规则：

告警项	触发条件	意义
成本超支	日成本达预算 80%	及时止损/查异常
错误率飙升	错误率 > 5%	服务可能挂了
延迟异常	P95 延迟 > 阈值	体验劣化
单用户异常	某用户消耗 > 均值 N 倍	疑似薅羊毛/攻击

💡 这就是可观测性和监控的闭环：日志记录 → 指标统计 → 阈值告警 → 主动通知。和你给 Linux 服务器做告警是一模一样的思路，只是换了监控维度。

---

⑦ 现成工具：LangSmith / Langfuse

自己造轮子适合理解原理和轻量场景。规模上来后，专业 LLM 可观测性平台能省很多事：

工具	特点
Langfuse	开源、可自部署，追踪 + 成本 + 评估，性价比高
LangSmith	LangChain 官方，调试/追踪/评估一体，生态好
Helicone	一行代理接入，侧重成本和缓存
Phoenix (Arize)	开源，侧重评估和可视化

它们提供开箱即用的可视化链路追踪、Token/成本仪表盘、Prompt 版本管理、效果评估，比自己搭全面得多。

🎯 选型建议：想快速理解原理、场景简单 → 自己埋点（本文方案）；团队协作、规模较大、要可视化 → 上 Langfuse（开源可自部署，隐私可控）或 LangSmith。原理你已经懂了，用工具只是换个更好的仪表盘。

---

⑧ 完整可运行 Demo：一个自带仪表盘的 Agent

把日志、trace、成本、告警四件套整合成一个能直接跑的脚本（无需真实 API，用假的 usage 演示，聚焦可观测性本身）：

import json, time, uuid, logging

logging.basicConfig(filename="agent.log", level=logging.INFO, format="%(message)s")

PRICE = {"standard-agent-v1": {"input": 0.001, "output": 0.002}}   # 每 1K token 价格
DAILY_BUDGET = 0.05                                                # 演示用小预算
_stat = {"cost": 0.0, "calls": 0, "errors": 0}


def log_event(event, **fields):
    line = json.dumps({"event": event, **fields}, ensure_ascii=False)
    logging.info(line)
    print("LOG >", line)                       # 演示时顺便打到控制台


def track_cost(model, prompt_tokens, completion_tokens):
    p = PRICE.get(model, {"input": 0, "output": 0})
    cost = prompt_tokens / 1000 * p["input"] + completion_tokens / 1000 * p["output"]
    _stat["cost"] += cost
    if _stat["cost"] > DAILY_BUDGET * 0.8:      # 达 80% 预算 → 告警
        log_event("alert", message=f"成本达 {_stat['cost']:.4f} 元，接近预算 {DAILY_BUDGET}")
    return cost


def handle_request(user_id, question, fake_usage, fail=False):
    """模拟一次带全套可观测性的 Agent 请求"""
    trace_id = str(uuid.uuid4())[:8]
    t0 = time.time()
    _stat["calls"] += 1
    log_event("trace_start", trace_id=trace_id, user_id=user_id, q=question[:30])
    try:
        if fail:
            raise RuntimeError("上游模型超时")
        cost = track_cost("standard-agent-v1", *fake_usage)
        log_event("llm_response", trace_id=trace_id,
                  prompt_tokens=fake_usage[0], completion_tokens=fake_usage[1],
                  cost_yuan=round(cost, 5))
        return "（模型回复）"
    except Exception as e:
        _stat["errors"] += 1
        log_event("error", trace_id=trace_id, error=str(e))    # 错误路径必记
        return "服务繁忙，请稍后再试"
    finally:
        log_event("trace_end", trace_id=trace_id,
                  duration_ms=int((time.time() - t0) * 1000))


if __name__ == "__main__":
    handle_request("u1", "北京天气", fake_usage=(300, 120))
    handle_request("u2", "写首诗", fake_usage=(200, 400))       # 累计成本推高 → 触发告警
    handle_request("u1", "算个数", fake_usage=(0, 0), fail=True)  # 模拟失败
    print("\n汇总:", _stat)

运行输出（每次请求的完整链路、成本、告警、错误一目了然）：

LOG > {"event": "trace_start", "trace_id": "a3f2c1e0", "user_id": "u1", "q": "北京天气"}
LOG > {"event": "llm_response", "trace_id": "a3f2c1e0", "prompt_tokens": 300, "completion_tokens": 120, "cost_yuan": 0.00054}
LOG > {"event": "trace_end", "trace_id": "a3f2c1e0", "duration_ms": 0}
LOG > {"event": "trace_start", "trace_id": "b1d4...", "user_id": "u2", "q": "写首诗"}
LOG > {"event": "alert", "message": "成本达 0.0415 元，接近预算 0.05"}
LOG > {"event": "llm_response", "trace_id": "b1d4...", ...}
LOG > {"event": "error", "trace_id": "c9a1...", "error": "上游模型超时"}
...
汇总: {'cost': 0.0413..., 'calls': 3, 'errors': 1}

这就是一个「看得见」的 Agent：每一次请求花了多少钱、走了多久、在哪报错、什么时候该告警，全都有据可查。 换成真实调用时，把 fake_usage 替换成 API 返回的 response.usage 即可。

---

⑨ 常见坑与建议

坑	说明	建议
用 print 当日志	无法检索统计	结构化 JSON 日志
日志记了敏感信息	密钥/隐私泄露	脱敏，只记必要字段
只记成功不记失败	出错时反而没日志	错误路径必须记日志
没有 trace_id	无法还原单次请求	每请求一个 trace_id 贯穿
只记录不告警	出事没人知道	加阈值告警，主动通知
日志无限增长	撑爆磁盘	日志轮转（logrotate）、定期归档

🔍 一条铁律：错误路径的日志比成功路径更重要。系统正常时你不看日志，恰恰是出错时你才需要它——所以 except 里一定要记全上下文。

---

⑩ FAQ

Q1：可观测性会拖慢 Agent 吗？
本地写日志开销极小。若用远程平台，用异步上报避免阻塞主流程即可。

Q2：Token 数从哪来？
大多数 API 响应里带 usage 字段（prompt_tokens/completion_tokens）。流式输出可能需要额外获取或自己用 tiktoken 估算。

Q3：该记多详细？
最少记：trace_id、user_id、时间、token、状态、错误。要复现问题可加输入输出（注意脱敏和存储成本）。

Q4：一定要上 Langfuse 这类平台吗？
不一定。小项目自己埋点足够。当你需要团队协作、可视化、Prompt 评估时再上，收益才明显。

Q5：怎么防止某个用户恶意刷爆成本？
结合第6篇的限流 + 本篇的单用户成本监控 + 告警。超阈值自动限制或封禁。

---

⑪ 总结

这一篇给 Agent 装上了「仪表盘」，把黑盒变白盒：

支柱	落地	作用
日志	结构化 JSON 打点	记录发生了什么
追踪	trace_id 贯穿全链路	还原单次请求
成本	记 Token、算钱、按维度聚合	控制预算、抓异常消耗
告警	阈值触发主动通知	出事第一时间知道

核心认知：上线不是终点，是运维的起点。 一个「看得见」的 Agent，才谈得上持续优化和稳定运营。你会发现，这套「记录→统计→告警」的思路，和给 Linux 服务器做监控本质相同——可观测性是所有线上系统的通用功课。

下一篇是「上线」主题：把 Agent 部署到生产环境，Docker 容器化、限流、熔断、优雅降级，一套齐活。

---

🔜 下一篇预告：《把 Agent 部署上线：Docker + 限流 + 熔断》——从本地脚本到扛得住真实流量的线上服务。
👍 如果本文帮到你，点赞 / 收藏 / 关注，追更不迷路。