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 HandlerLangChain 生态对 LangFuse 的支持做得相当好。官方提供了一个名为 LangfuseCallbackHandler 的集成可以直接挂在 LangChain 的调用链上自动捕获大部分遥测数据。如果你用的是 LangChain Python 版本安装依赖非常简单pip install langfuse langchain然后设置环境变量把前一步拿到的三个凭据填进去export LANGFUSE_PUBLIC_KEYyour-public-key export LANGFUSE_SECRET_KEYyour-secret-key export LANGFUSE_HOSThttp://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(modelgpt-4o-mini) agent_executor AgentExecutor(agent..., tools..., verboseTrue) 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(namecustomer-service, user_iduser_123) trace.event( nameuser-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 方法支持传入 configconfig 里可以带 metadata会透传到回调里。所以最优雅的方式是在入口统一加 metadataimport 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}, configconfig)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, headersheaders, paramsparams) 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(), keylambda x: x[1], reverseTrue): 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 应用的维护成本会肉眼可见地下降。