AI Agent 可观测性实战：Langfuse 全链路追踪与埋点指南

1. 从“能跑通”到“看得见”：AI Agent 工程化的分水岭

我最早做 AI Agent 项目的时候，和大多数人一样，关注点全在“能不能跑通”上。一个 Prompt 调通、工具函数接上、模型能返回结果，就觉得大功告成。直到有一次线上环境出了个诡异问题：用户反馈 Agent 给出的答案时好时坏，但我在本地怎么复现都正常。翻日志翻了整整一个下午，只看到一堆零散的print输出，根本拼不出完整的调用链路——模型输入是什么、检索到了哪些文档、工具调用了几次、每次耗时多少、Token 消耗了多少，全是黑盒。

那次之后我才真正意识到，AI Agent 和传统后端服务有一个本质区别：它的执行路径是不确定的。传统接口的调用链是代码写死的，你读一遍代码就知道数据怎么流转；但 Agent 的每一步决策都依赖模型输出，同一个输入可能走完全不同的分支。这种不确定性带来的调试成本，靠print和日志文件是扛不住的。

这就是Langfuse这类可观测性工具切入的痛点。它做的事情，简单说就是给 AI Agent 装上一套“行车记录仪”：每一次模型调用、每一次工具执行、每一轮对话，都被记录成结构化的 Trace，你可以在一个界面里看到完整的调用树、每步的输入输出、耗时和成本。这篇文章我想聊的，就是怎么把 Langfuse 真正落到一个 AI Agent 系统的工程实践里，从最基础的模型调用埋点，到全链路可观测体系的搭建，包括我踩过的坑和总结出来的配置经验。

适合谁看？如果你正在做 AI Agent 开发，不管是基于 LangChain、LangGraph 还是自己手写的编排逻辑，只要你的系统开始变复杂、开始上生产、开始有人问你“为什么这次回答这么慢”，那这套东西你就绕不开。哪怕你现在还在本地跑 Demo，提前把可观测性埋进去，后面会省掉大量返工。

2. 为什么 AI Agent 比普通服务更需要可观测性

2.1 Agent 的“不确定性”到底体现在哪

先把这个核心问题讲透，不然后面所有的工程决策你都会觉得是“多此一举”。

普通 Web 服务的请求处理路径是确定的：请求进来，经过中间件、路由、业务逻辑、数据库查询、返回响应，每一步都是代码显式定义的。你加个日志，基本就能还原整个流程。

AI Agent 完全不是这个逻辑。一个典型的 Agent 执行流程是这样的：用户输入 → 模型判断意图 → 决定是否调用工具 → 工具返回结果 → 模型再次判断 → 可能继续调用工具 → 最终生成回答。这里面有几个关键的不确定点：

• 分支不确定：模型可能选择调用工具，也可能直接回答，还可能连续调用多个工具。你无法在代码层面穷举所有路径。
• 输入不确定：每次传给模型的 Prompt 可能因为上下文拼接、检索结果不同而完全不同。
• 输出不确定：同样的输入，模型可能给出不同格式的输出，甚至偶尔“跑偏”。
• 性能不确定：模型响应时间波动极大，工具调用可能超时，检索可能返回空结果。

我遇到过最典型的一个案例：一个客服 Agent 在处理退款问题时，正常情况下应该调用订单查询工具，但某次模型“自作主张”直接根据上下文编了一个订单号返回给用户。如果没有完整的 Trace 记录，你根本不知道它是从哪一步开始跑偏的。

2.2 没有可观测性，你会损失什么

我把这个问题拆成三个层面，都是我实际踩过的：

调试层面：线上出问题，你只能看到最终输出是错的，但不知道错在哪一步。是检索召回了错误文档？是工具调用参数传错了？还是模型本身理解偏了？没有 Trace，你只能靠猜。

成本层面：AI Agent 的 Token 消耗是实打实的钱。一个复杂的 Agent 一次对话可能调用模型五六次，每次的输入长度都不一样。如果你不知道哪一步消耗最大、哪些调用是冗余的，成本优化就无从下手。我见过一个项目，光是重复的意图识别调用就占了总 Token 的 40%。

质量层面：你想做评测、想做 A/B 测试、想对比不同 Prompt 的效果，前提是你得有一份结构化的、可回溯的执行数据。否则你连“这次改动到底有没有变好”都说不清楚。

2.3 Langfuse 在技术选型中的位置

市面上可观测性工具不少，为什么我最终选了 Langfuse？这里说几个我实际对比后的判断：

最关键的一点是，Langfuse 的数据模型天然贴合 LLM 应用的执行结构：Trace 代表一次完整请求，Span 代表一个执行步骤，Generation 专门记录模型调用。这种分层设计和 Agent 的执行逻辑是一一对应的，不需要你去做额外的抽象映射。

3. Langfuse 核心概念拆解与埋点设计

3.1 Trace、Span、Generation 三层结构怎么理解

很多人第一次看 Langfuse 文档会被这几个概念绕晕。我用一个生活化的类比来解释：

把一次完整的 Agent 对话想象成一次“外卖下单”：

• Trace就是这一整单外卖，从你打开 App 到收到餐，是一个完整的闭环。
• Span是这单里的每个环节：浏览商家、选菜、下单、支付、配送。每个环节有开始和结束时间。
• Generation是其中专门涉及“模型思考”的环节，比如 Agent 判断“用户想要什么”、决定“调用哪个工具”，这些都需要模型参与，所以单独标记。

这样设计的好处是，你在 Langfuse 界面里看到的是一棵树：根节点是 Trace，下面挂着各个 Span，模型调用作为 Generation 嵌在对应的 Span 里。哪一步慢、哪一步贵、哪一步出错，一目了然。

3.2 埋点位置的选择逻辑

埋点不是越多越好，关键是要覆盖“决策点”和“边界点”。我在实践中总结了几条原则：

• 模型调用必埋：每一次 LLM 调用都要记录，包括输入 Prompt、输出内容、Token 数、耗时、模型名称。这是最核心的数据。
• 工具调用必埋：工具名称、入参、返回值、执行耗时、是否成功。工具是 Agent 和外部世界交互的边界，出问题概率最高。
• 检索环节必埋：如果 Agent 有 RAG 能力，检索的 query、召回文档、相似度分数都要记录。检索质量直接决定回答质量。
• 关键分支必埋：Agent 的意图判断、路由决策这些逻辑节点，即使不涉及模型调用，也建议用 Span 标记，方便还原决策路径。

注意：不要在每个函数入口都无脑加埋点，那样 Trace 会变得极其臃肿，反而看不清主线。埋点的目的是还原“有意义的执行路径”，不是做代码覆盖率统计。

3.3 数据模型设计的一个关键决策

这里有个容易被忽略但很重要的点：Trace 的粒度怎么定。

我的建议是：一次用户请求对应一个 Trace。哪怕这次请求触发了多轮 Agent 循环、调用了十几次模型，也全部挂在同一个 Trace 下。原因是，用户关心的是“我这一次提问的完整结果”，而不是中间调用了多少次模型。把一次请求拆成多个 Trace，反而割裂了上下文。

但如果是批处理任务或者定时任务，那就要按“一个任务实例一个 Trace”来设计。这个粒度选择直接影响你后续查询和分析的便利性，一开始就要想清楚。

4. 实操：从零搭建 Langfuse 可观测链路

4.1 环境准备与部署方式选择

Langfuse 提供两种使用方式：云托管版和自托管版。我的建议是：

• 个人项目、快速验证：直接用云托管版，注册就能用，省去部署成本。
• 企业项目、数据敏感：自托管，用 Docker Compose 拉起完整服务栈。

自托管的部署命令大致是这样（基于官方 Compose 配置）：

# 拉取官方 compose 配置 git clone https://github.com/langfuse/langfuse.git cd langfuse # 启动服务，包含 web、worker、postgres、clickhouse、redis docker compose up -d

启动后默认在 3000 端口提供 Web 界面。第一次登录需要创建组织和项目，然后拿到public key和secret key，这两个是后续 SDK 接入的凭证。

提示：自托管版本依赖 ClickHouse 做分析存储，对机器配置有一定要求。如果只是小规模使用，2 核 4G 起步，但建议给到 4 核 8G 以上，否则 Trace 数据量上来后查询会明显变慢。

4.2 SDK 接入与基础埋点

以 Python 为例，最基础的接入方式是这样：

from langfuse import Langfuse langfuse = Langfuse( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="http://localhost:3000" ) # 创建一个 Trace trace = langfuse.trace(name="agent-request", user_id="user_123") # 记录一次模型调用 generation = trace.generation( name="intent-classification", model="gpt-4", input=[{"role": "user", "content": "帮我查一下订单状态"}], ) # 模型调用完成后，更新输出 generation.end( output="查询订单", usage={"input": 120, "output": 8} )

这段代码看起来简单，但有几个细节值得说：

• trace的user_id字段非常有用，后续可以按用户维度分析行为。
• generation的input建议传结构化的消息数组，而不是拼接好的字符串，这样在界面上展示更清晰。
• usage字段一定要填，这是成本统计的基础。如果你用的是 OpenAI SDK，可以直接从 response 里取usage字段。

4.3 与 LangChain/LangGraph 的集成

如果你的 Agent 是基于 LangChain 或 LangGraph 构建的，Langfuse 提供了现成的 Callback Handler，接入成本极低：

from langfuse.callback import CallbackHandler langfuse_handler = CallbackHandler( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="http://localhost:3000" ) # 在调用 chain 或 agent 时传入 result = agent.invoke( {"input": "帮我查一下订单状态"}, config={"callbacks": [langfuse_handler]} )

这个 Handler 会自动帮你完成几件事：每次 LLM 调用自动创建 Generation、每次工具调用自动创建 Span、整个链路自动串成一个 Trace。基本上你不需要写额外的埋点代码，就能拿到完整的可观测数据。

注意：Callback Handler 的自动埋点虽然方便，但有些自定义逻辑它覆盖不到。比如你自己写的路由判断、自定义的检索逻辑，还是需要手动加 Span。我的做法是“自动为主、手动补漏”，关键的自定义节点手动补上。

4.4 一个完整的 Agent 埋点示例

下面这个例子展示了一个带工具调用和检索的 Agent 完整埋点结构：

def handle_user_query(query: str, user_id: str): trace = langfuse.trace(name="agent-query", user_id=user_id, input=query) # 第一步：意图识别 intent_span = trace.span(name="intent-recognition") intent_gen = intent_span.generation( name="classify", model="gpt-4", input=[{"role": "user", "content": query}] ) intent = classify_intent(query) intent_gen.end(output=intent) intent_span.end() # 第二步：检索（如果需要） if intent == "knowledge_query": retrieval_span = trace.span(name="retrieval") docs = retrieve_documents(query) retrieval_span.end(output={"doc_count": len(docs), "docs": docs}) # 第三步：工具调用 tool_span = trace.span(name="tool-call", input={"tool": "order_query"}) tool_result = call_tool("order_query", {"order_id": "12345"}) tool_span.end(output=tool_result) # 第四步：生成最终回答 final_gen = trace.generation( name="final-answer", model="gpt-4", input=build_prompt(query, docs, tool_result) ) answer = generate_answer(query, docs, tool_result) final_gen.end(output=answer, usage={"input": 800, "output": 150}) trace.end(output=answer) return answer

这个结构跑起来后，你在 Langfuse 界面里看到的是一棵清晰的树：根节点是agent-query，下面依次挂着意图识别、检索、工具调用、最终生成。每一步的输入输出、耗时、Token 都清清楚楚。

5. 全链路可观测的进阶玩法

5.1 用 Session 串联多轮对话

单次 Trace 只能看到一轮对话，但真实场景下用户是连续提问的。Langfuse 的 Session 概念就是解决这个问题的：

trace = langfuse.trace( name="agent-query", user_id="user_123", session_id="conversation_abc" )

只要session_id相同，多轮对话就会在界面上聚合成一个会话视图。这对于分析“用户在第几轮开始不满意”“上下文累积到多少轮后模型开始跑偏”这类问题特别有用。

我实际用下来，Session 视图最大的价值是能直观看到上下文膨胀的过程。有时候一个会话聊到第十轮，Token 消耗已经是最初的五六倍，这时候你就该考虑做上下文压缩或者摘要了。

5.2 成本追踪与 Token 优化

Langfuse 会自动根据模型名称和 Token 数计算费用，但前提是你要正确填写model字段和usage。我建议在项目里统一封装一个模型调用函数，把埋点逻辑收口，避免每个地方都手写。

成本分析的一个实用技巧：按 Span 名称聚合 Token 消耗。你很快就能发现哪个环节最“烧钱”。我之前的项目里，意图识别这个看似简单的步骤，因为每轮对话都要调用一次，累计消耗居然排到了第二。后来改成用小模型做意图识别，成本直接降了 60%。

5.3 评测与数据集管理

Langfuse 内置了评测功能，可以把你线上收集到的 Trace 直接加入数据集，然后用不同的 Prompt 或模型重新跑一遍，对比效果。这个流程我总结成三步：

1. 收集：从线上 Trace 里挑选有代表性的样本，加入数据集。
2. 标注：人工给这些样本打上期望输出或评分。
3. 对比：用新版本的 Prompt 跑一遍数据集，看评分变化。

这套流程的价值在于，它把“Prompt 调优”从玄学变成了可量化的实验。你改了一版 Prompt，跑一下数据集，分数涨了还是跌了，一目了然。

5.4 告警与异常监控

Langfuse 本身不直接提供告警功能，但你可以通过它的 API 拉取数据，接入自己的监控体系。我常用的几个告警指标：

• 错误率：Generation 中标记为 error 的比例超过阈值。
• P95 延迟：模型调用或工具调用的 P95 耗时突增。
• Token 异常：单次 Trace 的 Token 消耗超过正常范围。
• 空结果率：检索环节返回空文档的比例异常升高。

这些指标通过定时任务拉取 Langfuse API，推送到告警平台，基本能覆盖大部分线上异常场景。

6. 踩坑实录与常见问题排查

6.1 埋点数据丢失或不完整

现象：Trace 创建了，但某些 Span 没有记录，或者 Generation 的 output 是空的。

排查思路：最常见的原因是异常路径没有正确end()。比如模型调用抛异常了，代码直接跳到 except 分支，Generation 的end()没执行。解决办法是用 try/finally 包住：

generation = trace.generation(name="call", model="gpt-4", input=prompt) try: result = call_model(prompt) generation.end(output=result) except Exception as e: generation.end(output={"error": str(e)}, level="ERROR") raise

6.2 自托管性能问题

现象：Trace 数据量上来后，Langfuse 界面加载变慢，查询超时。

排查思路：主要是 ClickHouse 的查询压力。几个优化方向：一是定期清理过期数据，二是给 ClickHouse 增加资源，三是调整查询的时间范围。我一般会设置数据保留策略，超过 30 天的 Trace 归档或删除。

6.3 常见问题速查表

6.4 几个我踩过的坑

坑一：在异步代码里用同步 SDK。Langfuse 的 Python SDK 有同步和异步两套接口，如果你在 async 函数里用了同步接口，会阻塞事件循环，导致整体性能下降。一定要用langfuse.async_trace()这类异步方法。

坑二：Trace 嵌套过深。有一次我把一个循环里的每次迭代都创建了 Span，结果一个 Trace 下面挂了上百个节点，界面卡得打不开。后来改成只在关键迭代节点记录，问题解决。

坑三：忽略采样。生产环境流量大的时候，全量记录 Trace 会带来不小的存储和性能开销。Langfuse 支持采样率配置，高流量场景下建议设置 10% 到 20% 的采样率，既能发现问题又不至于压垮系统。

7. 关于 AI Agent 工程实践的一点个人体会

做 AI Agent 这一年多，我最大的感受是：可观测性不是锦上添花，而是工程化的基础设施。你可以在 Demo 阶段不接 Langfuse，但一旦系统开始面对真实用户、真实流量、真实成本压力，没有可观测性就是在裸奔。

Langfuse 这套东西的价值，不在于它记录了多少数据，而在于它把 Agent 的“黑盒执行”变成了“白盒可见”。你能看到模型在想什么、工具在做什么、钱花在哪里、问题出在哪一步。这种可见性带来的调试效率提升，是任何日志方案都替代不了的。

如果让我给正在做 AI Agent 的朋友一个建议，那就是：从第一天就把 Langfuse 接进去。哪怕你现在只是跑个本地 Demo，提前把埋点习惯养好，后面系统复杂起来的时候，你会感谢当初的自己。至于部署方式，个人项目直接上云托管，企业项目自托管，别在这上面纠结太久，先把数据跑起来再说。