1. 为什么 AI 应用必须补上“可观测”这一课

先说一个真实的场景。我接手过一个基于 LangChain 做的智能客服项目，功能看着都正常，用户偶尔反馈“回答变慢了”“有时候答非所问”，但打开日志一看，只有一行行的 JSON 输出，根本看不出是哪一步出了问题。是 LLM 调用超时？是检索到的文档不对？还是 Prompt 被上下文撑爆了？全凭猜。

后来我才意识到，传统后端那套“打日志 + 看监控”的思路，放到 AI 应用里完全不够用。AI 应用和普通 Web 服务最大的不同在于：它不是一个“请求进来、结果出去”的简单链路，而是一串由模型调用、工具调用、检索、记忆、路由组成的复杂流水线。任何一个环节的异常，都会被下游放大，最终表现为用户感知到的“变傻”或者“变贵”。

这就是我为什么在项目里引入 LangFuse 和 LangChain 组合的原因。LangFuse 是专门做 LLM 可观测性和评估优化的开源平台，LangChain 则是目前生态最成熟的 AI 应用编排框架。两者配合，可以在不改动业务逻辑的前提下，把一次完整请求的轨迹、Token 消耗、耗时、成本、质量指标全部记录到可视化面板里。

这篇文章不聊空泛的架构，就讲我从零开始，怎么用这套组合把项目从“盲人摸象”变成“开卷考试”。内容包括环境搭建、埋点改造、自定义事件上报、成本核算到问题排查，全部是实际操作过的方案，适合正在做 LangChain 项目的工程师，也适合想给现有 AI 服务补上监控能力的团队参考。

2. 整体设计思路：可观测到底要观测什么

2.1 先分清遥测数据的四个层次

开始动工之前，我花了不少时间想清楚一个问题：可观测性不是“多打点日志”，而是“数据能回答什么问题”。对于 AI 应用，我认为至少要覆盖四个层面的数据：

第一层是调用轨迹（Trace）。也就是一次用户请求从进入系统开始，经过哪些节点，每个节点的输入输出是什么，谁调用了谁，父子关系如何。这一层解决的是“链路是否通、卡在哪一步”的问题。

第二层是模型调用明细（Span）。每个 LLM 调用的模型名、Prompt 内容、回复内容、Token 用量、延迟、温度参数等。这一层解决的是“模型表现是否正常、是否被 Prompt 误导”的问题。

第三层是成本数据。按照模型单价和 Token 用量，实时计算出每一次请求花费了多少钱，并可以按用户、按功能模块、按时间维度聚合。这一层解决的是“AI 功能到底烧了多少预算”的问题。

第四层是质量与评估数据。包括用户反馈、人工标注、自动评估器的打分等。没有这一层，你就只知道系统“能跑”，但不知道它“跑得好不好”。

LangFuse 的模型设计恰好和这个分层一一对应。它把数据组织成 Observation、Trace、Span、Generation、Event 五种基础类型，外部系统通过 SDK 或 API 上报。我一开始也觉得概念有点多，但用顺了之后会发现，这套设计就是为了覆盖上面四层需求而生的。

提示：如果你此前用过传统的 APM 工具，可以把 Trace 理解成一次分布式请求的完整链路，Span 理解成链路里的每一个独立调用单元，Generation 则是专门为 LLM 调用设计的 Span 子类型。

2.2 为什么选择 LangFuse 而不是自己写日志表

在选型阶段，我也认真考虑过“自己建表、自己打日志、自己画面板”的野路子。毕竟团队里不缺后端工程师，写几个接口记录 Token 成本也不是难事。但我很快否定了这个方案，原因有三个：

第一，AI 应用的调用结构太复杂。一次带 Agent 的请求可能触发四五轮模型调用，每轮还可能调用多个工具，手动去维护 parent-child 关系的日志结构，工作量远超预期。LangChain 的 Callback 机制天然可以拿到完整的嵌套调用结构，LangFuse 又是为此设计的，填平这个模型沟壑的成本最低。

第二，评估功能不是简单的日志。除了记录，LangFuse 还提供了 Prompt 版本管理、数据集管理、在线评估、评分标注等功能。这些功能自己实现的话，等于从零做一个内部的 LLMOps 平台，周期至少按月算。

第三，LangFuse 是开源可自托管的。这就解决了数据合规的问题。模型调用内容、Prompt、业务数据可以留在自己的服务器上，不必传到第三方 SaaS。我最终选择自托管部署，也是出于对数据出域的顾虑。

3. 环境搭建：LangFuse 自托管部署与初始化

3.1 用 Docker Compose 拉起全套服务

LangFuse 部署方式很多，官方有云服务，也可以 Docker 部署。我们因为要对接内部业务环境，选择了自托管。官方仓库提供了现成的 docker-compose.yaml，包含 Web 应用、Worker、PostgreSQL 数据库和一个用于异步任务的 Redis。如果你只需要本地体验，这套配置足够。

我建议先把 Docker 和 Docker Compose 装好，然后执行以下步骤：

git clone https://github.com/langfuse/langfuse.git cd langfuse cp .env.example .env # 修改 .env 中的数据库密码、密钥等配置 docker compose up -d

启动完成后，访问 http://localhost:3000 就能看到登录界面。第一次使用需要注册账号并创建组织（Organization）和项目（Project）。项目创建完成后，进入项目设置页面，可以看到三个关键凭据：Public Key、Secret Key 和 Host URL。这三个值就是后面埋点要用到的连接信息。

这里有一个非常容易踩的坑：.env 文件里的 ENCRYPTION_KEY 必须要设置。这是用于加密敏感数据的密钥，如果留空或其他节点随机生成，重启容器后会因为密钥变动导致历史数据无法解密。我当时第一次部署就是因为没注意这个，重启后所有会话记录都变成了乱码。

3.2 初始化数据库与账号体系

如果你是第一次部署，还需要运行数据库迁移命令。新版本的 LangFuse 在容器启动时会自动执行迁移，但如果你用的是旧版本镜像，可能需要手动执行。保险起见，建议在 docker-compose 启动后观察一下 Web 容器的日志，确认没有报数据库相关的错误。

另外提一下账号体系。LangFuse 的企业版支持 SSO、RBAC 等能力，但我们用社区版也够。社区版的账号体系比较简单，每个用户属于一个组织，组织下可以创建多个项目。如果你的团队有多条产品线，建议按产品线拆分项目，这样成本统计和评估数据天然隔离，不会混在一起。

初始化完成之后，我习惯先在界面上手动创建几个 API Key，一个给开发环境，一个给生产环境。这样后续如果某个环境的 Key 泄漏或者需要轮换，可以直接在界面上吊销，不影响其他环境。

4. LangChain 埋点实操：从自动追踪到自定义事件

4.1 最省事的集成方式：LangChain Callback Handler

LangChain 生态对 LangFuse 的支持做得相当好。官方提供了一个名为 LangfuseCallbackHandler 的集成，可以直接挂在 LangChain 的调用链上，自动捕获大部分遥测数据。

如果你用的是 LangChain Python 版本，安装依赖非常简单：

pip install langfuse langchain

然后设置环境变量，把前一步拿到的三个凭据填进去：

export LANGFUSE_PUBLIC_KEY=your-public-key export LANGFUSE_SECRET_KEY=your-secret-key export LANGFUSE_HOST=http://localhost:3000

在代码中创建 handler，并传给 LangChain 的 callbacks 参数：

from langfuse.callback import CallbackHandler from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor langfuse_handler = CallbackHandler() llm = ChatOpenAI(model="gpt-4o-mini") agent_executor = AgentExecutor(agent=..., tools=..., verbose=True) response = agent_executor.invoke( {"input": "帮我查一下上周的销售数据"}, config={"callbacks": [langfuse_handler]} )

这样改完之后，跑一个请求，再到 LangFuse 面板里刷新，就能看到一条完整的 Trace，里面包含 Agent 的思考过程、每一次 LLM 调用的 Prompt 和 Completion、检索工具的执行结果等。整个过程不需要手写埋点代码，是最快的接入方式。

注意：CallbackHandler 一定要在每次请求时都传入，不要做成全局单例复用。虽然 LangFuse SDK 内部有队列和线程池，但多线程环境下复用同一个 handler 还是容易串数据。我在压测时遇到过 Trace 错乱的问题，改成每次请求新建实例后解决。

4.2 自定义 Span 与事件上报

自动追踪能覆盖大部分场景，但有些业务语义是框架层拿不到的。比如用户身份、业务订单号、A/B 实验分组、用户反馈结果，这些数据对后续做成本归因和质量分析非常重要。

LangFuse SDK 提供了手动创建 Span 和 Event 的接口。比如我想记录一个“用户反馈”事件，可以这样做：

from langfuse import Langfuse langfuse = Langfuse() trace = langfuse.trace(name="customer-service", user_id="user_123") trace.event( name="user-feedback", input={"message_id": "msg_456", "rating": 5}, metadata={"order_id": "order_789", "campaign": "summer_sale"} ) trace.update( output={"final_answer": "已为用户完成退款"}, metadata={"cost_estimate": 0.023} )

如果是在一次 Agent 执行过程中插入自定义事件，更推荐的做法是利用回调机制，在某个工具执行前后埋点。这样可以把业务信息挂在同一个 Trace 下，而不是另起一个孤立的 Trace。

4.3 如何给每个请求关联业务上下文

做过传统可观测的人都知道，Trace 只有在能关联到业务上下文时才有价值。否则一堆乱糟糟的调用记录，根本没法定位到具体用户和订单。LangFuse 支持在创建 Trace 时传入 user_id 和 session_id，同时可以在 metadata 里塞任意结构化的业务数据。

我的做法是在入口处生成一个 request_id，把它同时写入 Trace 的 metadata 和业务日志。用户后续在面板里搜索这个 ID，就能看到完整链路。这里有一个细节：LangChain 的 invoke 方法支持传入 config，config 里可以带 metadata，会透传到回调里。所以最优雅的方式是在入口统一加 metadata：

import uuid from langchain_core.runnables import RunnableConfig request_id = str(uuid.uuid4()) config = RunnableConfig( callbacks=[langfuse_handler], metadata={"request_id": request_id, "user_id": user_id, "channel": "wechat"} ) result = chain.invoke({"question": question}, config=config)

5. 成本监控：让每一笔 Token 花费都有迹可循

5.1 Token 用量与成本核算是怎么自动完成的

LangFuse 默认会记录每次 LLM 调用的输入 Token 数、输出 Token 数、模型名称和提供商。它可以基于你在控制台配置的模型单价自动计算成本。这里的模型单价可以在项目的 Models 配置里维护，比如 gpt-4o-mini 的输入价格是 0.15 美元/百万 Token，输出价格是 0.6 美元/百万 Token。

你不需要在代码里手动计算成本，只需要确保 LangChain 调用时传入了正确的模型名称。LangFuse 会把 model 参数作为映射 key 去匹配单价。如果用 Azure OpenAI 部署，model 名称可能是自定义的 deployment name，这时候需要单独配置规则。

5.2 按用户、按功能模块做成本归因

单纯看总账单是远远不够的。我比较关注两个维度的成本拆分：按用户维度，看有没有人滥用导致成本激增；按功能模块维度，看哪个 Prompt 或 Agent 特别烧钱。

LangFuse 面板提供了 trace 列表和聚合统计，你可以按 user_id 或 metadata 中的字段做筛选。我自己写了一个简单的成本日报脚本，定时从 LangFuse API 拉取前一天的所有 Trace，按 metadata 里的功能模块字段聚合成本，推送到群里。

import requests from collections import defaultdict url = "http://localhost:3000/api/public/traces" params = { "limit": 100, "from": "2025-01-01T00:00:00Z", "to": "2025-01-02T00:00:00Z" } headers = {"Authorization": "Bearer YOUR_PK_KEY"} resp = requests.get(url, headers=headers, params=params) result = defaultdict(float) for trace in resp.json().get("data", []): module = trace.get("metadata", {}).get("module", "unknown") cost = sum( obs.get("cost", 0) for obs in trace.get("observations", []) if obs.get("type") == "GENERATION" ) result[module] += cost for module, cost in sorted(result.items(), key=lambda x: x[1], reverse=True): print(f"{module}: ${cost:.4f}")

这里的观察点在于：LangFuse 的 API 返回数据里，每个 Generation 类型的 Observation 都带有 cost 字段，前提是你在控制台配过模型单价。有了这个接口，你完全可以脱离自带面板，自己拼一套更贴合业务需求的可视化看板。

心得：成本监控要尽量做在事前而不是事后。我给项目加了一个简单的熔断逻辑，当单个用户单日成本超过阈值时，自动降级到更便宜的模型或者限制调用频率。阈值就来自 LangFuse 的成本接口返回的数据。

6. 进阶：LangGraph 项目的可观测性改造

6.1 LangGraph 和 LangChain 到底差在哪

很多刚入门的同学都会问 LangGraph 和 LangChain 的区别。两者的关系很简单：LangChain 偏重提供各类组件和工具集成，像一座大型工具箱；LangGraph 则是一个更底层的编排运行时，专门把 AI 应用定义成一张有状态的状态机图，节点和节点之间可以有条件跳转、循环、人工介入。

如果你的应用是简单的链式调用，LangChain 就够了。但一旦涉及多轮 Agent 决策、状态在多步之间传递、需要根据不同结果走不同分支，LangGraph 的表达能力会更强。LangGraph 和 LangChain 在生态上是兼容的，LangGraph 的节点内部依然可以使用 LangChain 的模型、工具和检索器。

6.2 LangGraph 场景下的埋点方式

LangGraph 的底层也是基于 LangChain 的回调机制，所以 CallbackHandler 依然能捕获到完整的节点流转信息。但有个现象值得注意：LangGraph 引入了一层新的执行框架，原生的 Trace 结构里可能会看到所有节点被记录在一个 graph 根节点之下，节点的输入输出都堆在一起，排序看起来有点乱。

我实际测试下来，LangFuse 的 LangChain 集成对 LangGraph 的 node 做了映射，每个 node 基本会变成一个 Span。但由于 LangGraph 的循环结构，Trace 的树形结构会比较深。如果你在写 LangGraph 应用，我的建议是在每个节点的函数内部，额外用 langfuse_context 创建 Span 来补充节点级的关键指标，比如相关性分数或工具返回结果摘要。

6.3 用观察数据反哺评估优化

LangFuse 除了观测，还有一个重要能力是评估。官方支持自定义评估函数、LLM-as-a-judge、人工评分等。我在项目中用它对每个回答做幻觉检测：把用户问题、检索到的上下文、模型回答一起交给另一个评估模型，要求输出是否有幻觉的判断和理由，然后作为 Event 上报到同一个 Trace。这样不仅能看到调用是否成功，还能看到回答质量的量化指标。

数据积累一段时间后，就可以用 LangFuse 的 Dataset 功能做回归测试。把线上暴露过问题的示例整理成数据集，每次修改 Prompt 或更换模型后，批量跑一遍评估，对比分数变化。这种“观测 + 评估 + 回归”的闭环，才算真正把可观测性用起来了。

7. 常见问题与排查技巧实录

7.1 数据不一致或 Trace 丢失

最常见的问题是 Trace 偶发丢失或者数据不全。排查方向基本是这几个：第一，检查 SDK 的异步上报是否被进程退出打断，如果是脚本类任务，记得在末尾调用 langfuse.flush()；第二，确认 Web 应用和 Worker 服务的队列配置正确，如果数据量大而 Worker 消费慢，Trace 会出现延迟；第三，自托管版本要关注 PostgreSQL 的连接数和磁盘空间，数据库满了之后写入会静默失败。

这里再说一个我在生产环境遇到的坑：如果应用运行在 Docker 容器里，而 LangFuse 部署在宿主机上，环境变量中的 Host 一定不能用 localhost，要用宿主机在 Docker 网络中的地址，或者直接用公网/内网域名。否则应用能启动，但数据永远上报不上去，面板里一片空白。

7.2 Prompt 内容过长或敏感信息记录

LangFuse 默认会记录完整的 Prompt 和 Completion，这既是它的优点也是风险点。生产环境如果涉及用户隐私内容，必须做脱敏处理。SDK 提供了 masking 回调函数，可以在数据上报前对字段做替换。我实现了一个简单的敏感信息掩码逻辑，把手机号、邮箱、身份证号统一替换成占位符。

另一个容易忽略的是上下文太长导致数据库变大。AI 应用一次调用可能包含很长的检索上下文，每条 Trace 都是几 KB 甚至几十 KB，日积月累很占空间。我的做法是在生产环境用 LangFuse 的采样功能，按比例记录部分 Trace，同时把完整 Trace 只保留在问题定位专用的 debug 环境。

7.3 模型单价配置错误导致成本虚高或虚低

如果你发现成本数字明显不对，十有八九是模型单价配置的锅。LangFuse 是按 model 名称精确匹配单价的，如果你的代码里模型名带了奇怪的版本后缀，或者不同部署分别用了相同的模型名，成本统计就会串。建议在配置模型单价时写清楚口径，并且在项目环境里统一模型命名规范，不要一个模型叫 gpt-4o-mini，另一个叫 gpt-4o-mini-v2。

另外，LangFuse 的 cost 是按单价的输入/输出价格分别计算的，如果只配了输入价格没配输出价格，输出的成本会按 0 计算，总成本看起来就会偏低。

7.4 排查实录：一次 Agent 超时问题的定位过程

举一个完整的排查案例。某个 LangChain Agent 功能最近经常超时，用户反馈响应要 30 秒以上。我先在 LangFuse 面板里筛选出耗时最长的 Trace，点进去看到时间主要消耗在某个工具调用的等待上，而那个工具本身是一个内部 HTTP API。确认不是模型调用慢之后，我们转而去查那个内部 API 的响应，发现是它出现了偶发的大延迟。

如果没有 Trace 数据，这种问题很难定位。因为从应用日志看，所有日志都是正常输出的，只是整体很慢。但 Trace 能直接告诉你时间消耗的分布，是卡在模型、卡在工具、还是卡在检索，一目了然。这也是我强烈建议所有 LangChain 项目上线第一天就接可观测性的原因。

8. 给新手的几个实操建议

如果你正准备给项目接这套东西，我的建议是按照“自动追踪 → 数据观测 → 成本分析 → 评估优化”四步去推进，不要一上来就想把所有的 Event 和评估机制都搭好。先把自动追踪跑通，看到 Trace 数据，再逐步叠加自定义事件。

数据采样策略可以从全量开始，但生产环境建议至少按用户进行分母采样，保证数据代表性又不至于太占存储。

LangFuse 和 LangChain 都在快速迭代，版本兼容性问题偶有发生。如果升级 LangChain 后发现 Trace 结构异常，先检查自己用的 langfuse 包是否还是最新版本，官方集成代码对 LangChain 内部结构依赖比较强，版本落后会导致部分回调失效。

最后分享一个我个人的体会：可观测性不是一次性的工具接入，它更像一种开发习惯。每当你在代码里新加一个工具函数，就要想一想“这个函数的输入输出要不要上报”“如果这一步变慢了，我要不要能看出来”。把这种思维方式固化下来，AI 应用的维护成本会肉眼可见地下降。