把监控能力直接嵌入开发链路这是一套我在实战中打磨出来的 AI 对话监控仪表盘方案。技术栈是 Langfuse 做 LLM 可观测性、Langchain 做编排、DeepSeek 做模型底座、FastAPI 做后端服务、WebSocket 做实时推送。整套系统解决的核心问题只有一个当 AI 应用上线后你怎么知道模型每一次回答的质量、延迟、成本和异常如果你正在用 Langchain 接 DeepSeek又不想全靠看日志猜问题这篇实战笔记应该能帮你少走不少弯路从零开始搭出一个真实可用的监控仪表盘。1. 为什么 AI 对话应用需要独立的监控层先聊一个让我印象很深的场景。我第一次用 Langchain 接 DeepSeek 做问答机器人时本地调试一切正常一上线就出问题用户反馈回答变慢了有时候直接报错同样的题目两次答案完全不一样。我打开服务器日志看到的只有一长串请求记录根本无从判断是网络问题、模型问题、提示词问题还是上下文太长导致的截断。那一刻我意识到传统后端日志在 LLM 应用面前基本是瞎的。1.1 LLM 应用出问题传统日志根本帮不上忙传统 Web 服务出问题看状态码、看堆栈、看数据库慢查询基本能定位个七八成。但 LLM 应用的失败模式完全不同请求可能成功返回了但回答的内容是错的这叫静默失败模型推理本身没有报错但 token 消耗异常成本翻了好几倍同一套提示词模型换了个版本输出风格完全变了用户多轮对话时上下文没有被正确拼装导致回答质量直线下降这些问题靠看日志是看不出来的。你需要知道每一次请求里用户到底传了什么、模型返回了什么、中间有没有调用工具或检索器、每一步花了多长时间、烧了多少 token。这正是 Langfuse 这类 LLM 可观测性平台存在的意义。1.2 Langfuse 到底记了什么Langfuse 是开源的可观测性平台专门为 LLM 应用设计。我推荐它的核心理由是它和 Langchain 做了深度集成你不必手动埋点记录每一次请求的参数只要在调用链上挂一个回调Langfuse 就能自动把整个执行过程记下来。我实际用下来Langfuse 里最重要的几个概念概念作用对应我这套架构里的内容Trace一次完整的请求链路用户发起的一次对话SpanTrace 内部的子阶段检索、工具调用、模型推理Generation一次 LLM 调用记录DeepSeek 的 request 和 responseScore对回答质量打分的指标手动或自动评分的出口Observation上述所有记录的统一抽象可查询的最小单元这些都存下来之后你能在仪表盘上看到某次对话里DeepSeek 返回用了 4.2 秒、消耗 1567 个 token、花费约 xxx 元、prompt 当时长什么样。这类信息在生产环境里价值比任何日志都高。1.3 自托管还是用云服务Langfuse 官方提供了云服务注册就能用。但我选择自托管原因有三数据隐私对话内容可能包含用户敏感信息我一律不建议把生产对话直接传到第三方平台网络环境服务部署在内网时外部的 Langfuse 云服务会有额外的延迟和稳定性风险成本Langfuse 开源版功能已经够用自托管只需要一台能跑 Docker 的机器这里需要说明的是Langfuse 在 GitHub 上随版本迭代v2 和 v3 的部署方式略有不同。本文我使用的是 v3 系列的 docker-compose 部署方式因为 v3 将 web 和 worker 合并成了单服务更简单也更适合从零开始。2. 整体链路拆解一次对话从点击到刷新的完整路径搭这套架构时我第一步不是写代码而是先把数据流理清楚。整个链路一点也不神秘画成文字就是这个样子浏览器打开仪表盘页 → 建立 WebSocket 连接 → 用户输入问题 → FastAPI 收到消息 → 调用 Langchain 编排链 → DeepSeek 返回结果 → Langfuse 记录 Trace → 后端把事件推给前端 → 仪表盘实时刷新如果不用 Langfuse你会发现自己需要再造一个日志系统、数据看板、链路追踪系统三个东西拼起来才能达到差不多的效果。而 Langfuse 把这些全部合并成了一个。2.1 每个组件在这条链路里扮演的角色我列了一张分工表方便你对照理解。这张表基本就是整套系统的地图组件职责关键点FastAPI后端服务接收用户请求管理 WebSocket 连接异步运行不阻塞事件循环WebSocket实时双向通道推送监控事件前端能看到进行中的步骤Langchain编排业务流程负责拼接 prompt、调用模型、处理流式输出DeepSeek实际的大模型底座通过 OpenAI 兼容协议接入也可换成其他模型Langfuse记录并展示每一次调用的输入、输出、延迟、费用通过 callback 自动采集数据这里要特别强调一下 Langchain 在链路中的位置。很多人以为 Langchain 只是调用模型的一个封装其实它的价值是编排。比如你要做一个复杂的 AI 助手它可能要先去数据库查一条记录再调用一个函数计算最后才让模型总结。Langchain 能把这一整个流程编排起来而且把每一步的执行细节都暴露给回调机制Langfuse 才得以完整记录。2.2 为什么选 FastAPI WebSocket 而不是轮询监控仪表盘的关键需求是实时。如果你用传统的定时轮询每 5 秒刷新一次页面用户看到的延迟和资源消耗都不理想——因为一次大模型调用可能长达 30 秒甚至更久轮询的过程中频繁产生 HTTP 请求非常浪费。WebSocket 和轮询的本质区别在于轮询是前端主动问有结果了吗WebSocket 是后端主动说有结果了。AI 对话天然是不确定时延的可能 1 秒也可能 1 分钟用 WebSocket 能让前端实时收到每个阶段的事件比如正在调用模型模型返回了第一个字本次调用完成。那为什么不用 SSEServer-Sent Events呢SSE 是单向的服务端推送足够用但 WebSocket 支持双向通信客户端还可以随时发送取消生成切换模型获取当前会话详情等控制指令。后续做多用户场景时WebSocket 的可扩展性更好所以我最终选了它。2.3 实时监控的关键设计事件回传这套架构里前端仪表盘刷新的数据其实是分两类来源的Langfuse 后台的数据完整的 trace、token 费用、评分等通过 Langfuse 自己的 Web UI 查看WebSocket 推送的实时事件比如正在调用模型模型响应了前 10 个字这些是注入给前端仪表盘的我没有让前端直接去轮询 Langfuse 的 API而是让 FastAPI 在处理请求时边调用模型边把进度通过 WebSocket 推给浏览器。这样仪表盘可以同时展示实时状态和历史记录体验远好于等整个请求跑完再看 Langfuse。3. 环境准备Langfuse 自托管与三套密钥的正确配置业内流传一句话配置环境的时间永远比写代码长。放在这套架构里一点不夸张。Langfuse 部署本身不难但涉及到 Postgres、Redis、S3 存储、密钥配置、网络连通不提前搞清楚后面排查会非常痛苦。3.1 docker-compose 把 Langfuse 跑起来我采用的是 Langfuse v3 的 docker-compose 方案它把 web 和 worker 合并成了一个服务结构清爽了很多。最小化的 docker-compose.yml 大致是这样version: 3.8 services: langfuse: image: langfuse/langfuse:3 ports: - 3000:3000 depends_on: - db environment: DATABASE_URL: postgresql://langfuse:langfusedb:5432/langfuse NEXTAUTH_URL: http://localhost:3000 NEXTAUTH_SECRET: your-secret SALT: your-salt ENCRYPTION_KEY: your-encryption-key LANGFUSE_INIT_USER_EMAIL: adminexample.com LANGFUSE_INIT_USER_PASSWORD: admin123456 LANGFUSE_INIT_PROJECT_NAME: default-project db: image: postgres:16 environment: POSTGRES_USER: langfuse POSTGRES_PASSWORD: langfuse POSTGRES_DB: langfuse volumes: - langfuse-db-data:/var/lib/postgresql/data volumes: langfuse-db-data:几个关键参数的解释DATABASE_URLLangfuse 读写 Postgres 的连接串这里用的容器网络内部的地址ENCRYPTION_KEY用于加密敏感数据比如你后面可能配置的 API Key必须是一个 32 字节的 Base64 字符串可以用openssl rand -base64 32生成NEXTAUTH_SECRET和SALT用于登录认证随便生成两串足够长的随机字符串就行LANGFUSE_INIT_USER_EMAIL和LANGFUSE_INIT_USER_PASSWORDLangfuse v3 首次启动时会自动创建管理员账号启动命令就一句docker compose up -d启动后等一两分钟浏览器打开http://localhost:3000用刚才配置的管理员账号登录第一步就算完成了。3.2 密钥体系与 Lanchain 的连通性验证在 Langfuse 后台你需要创建项目并获取两把钥匙公钥Public Key和私钥Secret Key。这个概念和很多人的直觉相反——公钥反而要放在前端或环境里用于上报数据私钥要保密。为了让你后端调用时能读写我建议在 FastAPI 服务的环境变量里这样配置LANGFUSE_SECRET_KEYsk-lf-xxxx LANGFUSE_PUBLIC_KEYpk-lf-xxxx LANGFUSE_HOSThttp://localhost:3000这三个变量Langfuse 的 SDK 和 Langchain 的 CallbackHandler 会自动读取你不需要在代码里硬编码。配置完成后验证连通性的最快方式是写一段极简的测试代码from langfuse import Langfuse langfuse Langfuse() # 如果配置正确后台会看到这个 trace langfuse.trace(nameconnection-test).update(inputhello, outputworld) langfuse.flush()跑完去 Langfuse 后台看如果connection-test出现在 Trace 列表里说明部署和密钥都通了。这一步不要跳过我见过太多人代码全对结果卡在密钥配置上浪费几个小时。3.3 版本匹配Langfuse 和 Langchain 的兼容问题这是最容易被忽略的问题也是让我踩坑最深的点。Langfuse 对 Langchain 的集成高度依赖回调接口的具体实现Langchain 升级后回调签名变了Langfuse 旧版根本识别不了。我的建议是统一使用较新且互相兼容的版本组合软件包我使用的版本langfuse2.41.x 或以上langchain0.2.x 或以上langchain-openai0.1.x 或以上fastapi0.115.xuvicorn0.30.x安装命令一并给你pip install langfuse2.41 langchain0.2 langchain-openai0.1 fastapi0.115 uvicorn[standard]0.30注意新版 Langchain 把ChatOpenAI拆分到了langchain-openai包里如果你还在用from langchain.chat_models import ChatOpenAI大概率会收到弃用警告甚至直接报错。这个细节在后面写代码时也要留意。4. 核心代码Langchain 接 DeepSeek并用 Langfuse 全链路追踪主体代码分两块讲模型接入方式和追踪回调的挂载方式。这两块搞明白你就能把任意 Langchain 应用接入 Langfuse而不仅仅是 DeepSeek。4.1 DeepSeek 接入 Langchain 的两种方式DeepSeek 官方提供了兼容 OpenAI 格式的 API所以接入 Langchain 非常简单。第一种方式直接指定base_urlfrom langchain_openai import ChatOpenAI llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, temperature0.7, )第二种方式如果你的 Langchain 版本较新也可以用init_chat_model统一入口from langchain.chat_models import init_chat_model llm init_chat_model( deepseek-chat, model_provideropenai, base_urlhttps://api.deepseek.com, api_keyos.getenv(DEEPSEEK_API_KEY), )两种方式本质一样我习惯用第一种因为更直观出问题也好排查。DeepSeek 的base_url我用的是https://api.deepseek.com如果你在某个版本里遇到验证错误可以试试补全为https://api.deepseek.com/v1两者目前都可用。4.2 用 CallbackHandler 把每次模型调用记录到 Langfuse这一步是整套监控的核心。Langfuse 提供了CallbackHandler把它挂到 Langchain 的调用链上Langfuse 就能自动记录整个 trace。先初始化 handlerfrom langfuse.callback import CallbackHandler langfuse_handler CallbackHandler()然后在一个 LCEL 管道里使用from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的 AI 助手。), (human, {input}), ]) chain prompt | llm response chain.invoke( {input: 用一句话介绍 Langfuse 是什么}, config{callbacks: [langfuse_handler]}, )这里的关键是config{callbacks: [langfuse_handler]}。Langchain 会把执行链路中所有的 LLM 调用、工具调用、检索操作统统通过这个回调上报给 Langfuse。如果你用的是RunnableSequence或者更复杂的LangGraph也是在构造好的对象上通过with_config()传入chain_with_tracing chain.with_config( callbacks[langfuse_handler], )这个方法的通用性极强Langchain 社区里很多复杂的 Agent 应用也都是用这种方式接入的。你不需要去改业务代码的逻辑只需要在调用入口挂一个回调。4.3 流式输出时如何兼顾实时显示和完整追踪监控仪表盘里用户最直观的体验就是模型一个字一个字往外蹦。要支持流式输出Langchain 侧要改用stream或astreamasync for chunk in chain.astream( {input: 讲一个笑话}, config{callbacks: [langfuse_handler]}, ): # chunk 是流式返回的增量内容 await websocket.send_text(chunk)Langfuse 的 CallbackHandler 对流式输出的处理逻辑是它在流式开始时生成一个 Generation然后随着每个 chunk 增量更新记录。也就是说你不需要等流式结束才去上传Langfuse 会在整个流式过程结束后自动把完整的输入输出、token 用量、延迟都固定下来。这里有个实际问题流式输出是异步的WebSocket 推送也是异步的两个异步任务并存时最怕的事件循环阻塞。我建议所有模型调用统一使用astream异步流式而不是在异步视图里用同步的chain.invoke。如果迫不得已要调用同步代码记得用run_in_executor放到线程池否则你会看到前端卡住、WebSocket 断线、Langfuse 上报超时三个问题一起爆发。5. FastAPI WebSocket 服务端实现把监控事件推送到前端现在进入最让我兴奋的部分让仪表盘真正动起来。FastAPI 对 WebSocket 的原生支持做得非常好你不需要额外装第三方库核心代码量也就几十行。5.1 ConnectionManager管理 WebSocket 连接的基座WebSocket 和普通 HTTP 请求最大的区别是连接是长久的你不能让每个连接各自为战必须统一管理。我写了一个简单的连接管理器from fastapi import WebSocket from typing import List class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except RuntimeError: pass manager ConnectionManager()这里要注意send_json的异常处理。前端打开页面后可能直接关掉或者网络中断服务端在向这个连接发送数据时会抛异常。不加 try/except 的话一个连接断开会导致整个监控服务崩溃。5.2 接收用户消息驱动 Langchain 执行并推送事件WebSocket 端点的核心逻辑是收到用户消息后按阶段推送事件。我把流程拆为四步每个步骤对应用户能感知的一个阶段from fastapi import APIRouter, WebSocket, WebSocketDisconnect router APIRouter() router.websocket(/ws/monitor) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 1. 等待前端传来的用户输入 data await websocket.receive_json() user_input data.get(message, ) # 2. 推送开始处理事件 await manager.broadcast({ type: status, stage: processing, message: 正在调用 DeepSeek 模型, }) # 3. 调用 Langchain 链流式获取结果同时上报 Langfuse langfuse_handler CallbackHandler() full_response async for chunk in chain.astream( {input: user_input}, config{callbacks: [langfuse_handler]}, ): full_response chunk await manager.broadcast({ type: token, content: chunk, }) # 4. 推送完成事件 await manager.broadcast({ type: status, stage: done, message: 本次对话已完成, }) except WebSocketDisconnect: manager.disconnect(websocket)这段代码有一个细节值得注意我在每次请求时都新建了CallbackHandler()。为什么因为 Langfuse 的 handler 是绑定 trace 上下文的整个链路应该一次会话只创建一个 trace。如果你在全局共享一个 handler多用户并发时你会发现 trace 全都串到同一个会话里了。这是我在测试并发时踩过的一个大坑。对于 token 级别的推送前端拿到的是一串不断累积的字符串你可以在前端逐字渲染体验就像 ChatGPT 的流式回答。而对于 status 类型的事件前端可以用来更新状态栏或日志面板。5.3 前端仪表盘怎么消费这些事件前端我用了一个非常朴素的 HTML 页面加原生 WebSocketconst ws new WebSocket(ws://localhost:8000/ws/monitor); ws.onopen () { console.log(WebSocket connected); }; ws.onmessage (event) { const msg JSON.parse(event.data); if (msg.type token) { // 把内容追加到对话框 document.getElementById(answer).innerText msg.content; } if (msg.type status) { // 更新状态栏 document.getElementById(status).innerText msg.message; } }; function send() { const input document.getElementById(question).value; ws.send(JSON.stringify({ message: input })); document.getElementById(answer).innerText ; }这个页面虽然简单但已经能达到实时显示模型输出的效果。如果你还想让前端直接展示 Langfuse 的历史 trace 列表可以把 Langfuse 的公有 API 包装成 FastAPI 的 HTTP 接口前端定时拉取或者通过 WebSocket 推送过去。这里还有一个加分项把 Langfuse 的评分功能露出来。对话结束后前端展示赞和踩两个按钮点击后调用一个接口给这条 trace 打一个 scorefrom langfuse import Langfuse langfuse_client Langfuse() app.post(/score/{trace_id}) async def score_trace(trace_id: str, score: int): langfuse_client.score( trace_idtrace_id, nameuser-feedback, valuescore, ) return {status: ok}这个功能看似简单价值却很大。它能让你把不同提示词、不同参数下模型的回答质量量化下来长期积累的数据对优化 prompt 非常有用。6. 实测踩坑记录这套架构最容易翻车的地方最后一部分我把我实际运行这套架构时遇到的坑集中列出来。每个坑都花了我不少时间排查希望你能直接避开。6.1 回调不生效Langchain 版本和回调传入方式的坑最典型的报错是代码不报错Langfuse 后台也看得见项目但就是没有 trace 进来。排查思路先看 Langchain 版本是不是太老或太新。我踩过一次很深的坑当时 Langchain 升级到 0.3.xCallbackHandler的接口有了变化老版本的 langfuse 不兼容导致回调被静默丢弃。解决办法很粗暴升级 langfuse 到最新版同时保证 langchain-openai 和 langchain 的版本在兼容范围内。另一个容易忽略的问题有些 Langchain 的链会创建内部子链比如create_sql_query_chain它会自己生成新的 Runnable。这时候只在最外层invoke上传回调没用内部子链不会自动继承。解决办法是查一下这个子链对象是否暴露了with_config方法如果有给它单独挂一个回调。简单说回调的传播是逐层显式传的不是全局魔法。6.2 token 统计为零或者串线的排查如果你发现 Langfuse 上记录的 token 数为 0先看 Langchain 那边使用的是不是标准的ChatOpenAI。某些自定义模型封装不会上报 token 使用情况Langfuse 这边拿不到 usage 字段就只能显示 0。这种情况下你可以试试在 Langfuse 的 UI 中查看原始 trace 数据看看 usage 是不是确实没有返回。串线问题我在 5.2 小节稍微提过这里再强调一下。多用户并发时如果不为每次请求新建CallbackHandlertrace 会互相穿插A 用户的输入和 B 用户的输入可能出现在同一条 trace 里。这个现象在流量小的时候不明显一压测就彻底暴露。我的建议是把 handler 的生命周期和一次请求绑定在 WebSocket 端点的循环内部创建。6.3 生产环境扩展思路这套架构在开发环境跑通只是第一步生产环境还有很多事情要做任务队列化FastAPI 进程如果重启正在处理的 WebSocket 连接会全部断开。更稳的做法是把模型调用任务丢到 Redis/RabbitMQ 队列由单独的 worker 进程处理再把结果通过 WebSocket 网关推送回去多工作节点一台机器跑单进程QPS 上不去。用 Uvicorn 多 worker 部署时ConnectionManager 里的连接列表是每进程独立的WebSocket 广播会变成每进程局部广播。要解决全量广播得引入 Redis Pub/Sub 做跨进程事件分发日志保留与清理Langfuse 的数据是存在 Postgres 里的流量一大数据库体积增长很快。我建议在 docker-compose 里挂一个定时任务定期清理超过 N 天的 trace 数据避免磁盘写满鉴权本文的 WebSocket 端点是纯裸奔的生产环境必须在建立连接前做鉴权。常见做法是链接上带一个短时效 token或者连接成功后前端先发一条鉴权消息服务端校验通过前不处理任何业务消息这些扩展方向里最值得优先做的是任务队列化。因为 WebSocket 连接时长和模型调用时长几乎相等直接占满进程的并发能力不做异步化稍微来一波流量服务就扛不住了。结个尾说点实在的整套架构从拆解到落地给我最大的体会是Langfuse 这种可观测性工具不是锦上添花而是 LLM 应用的必需品。没有它你在生产环境面对一个回答质量变差的反馈时连排查的入口都没有有了它你可以直接回放那次调用看到当时的 prompt、模型参数、上下文长度甚至复现用户的完整对话。如果你按这篇文章从头到尾搭一遍应该已经拥有一个能实时显示模型输出、完整记录每次调用明细、支持用户反馈评分的 AI 对话监控仪表盘。这套代码本身就具备扩展性你可以继续把 Langfuse 的报表接口搬到自己的内网看板里也可以把评分数据接入告警系统。动手跑通第一版比继续读十篇架构分析更有价值。