把监控能力直接嵌入开发链路，这是一套我在实战中打磨出来的 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 | 一次完整的请求链路 | 用户发起的一次对话 |
| Span | Trace 内部的子阶段 | 检索、工具调用、模型推理 |
| Generation | 一次 LLM 调用记录 | DeepSeek 的 request 和 response |
| Score | 对回答质量打分的指标 | 手动或自动评分的出口 |
| 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 能让前端实时收到每个阶段的事件（比如"正在调用模型""模型返回了第一个字""本次调用完成"）。
那为什么不用 SSE（Server-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:langfuse@db:5432/langfuse NEXTAUTH_URL: http://localhost:3000 NEXTAUTH_SECRET: your-secret SALT: your-salt ENCRYPTION_KEY: your-encryption-key LANGFUSE_INIT_USER_EMAIL: admin@example.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_URL：Langfuse 读写 Postgres 的连接串，这里用的容器网络内部的地址ENCRYPTION_KEY：用于加密敏感数据（比如你后面可能配置的 API Key），必须是一个 32 字节的 Base64 字符串，可以用openssl rand -base64 32生成NEXTAUTH_SECRET和SALT：用于登录认证，随便生成两串足够长的随机字符串就行LANGFUSE_INIT_USER_EMAIL和LANGFUSE_INIT_USER_PASSWORD：Langfuse v3 首次启动时会自动创建管理员账号
启动命令就一句：
docker compose up -d启动后等一两分钟，浏览器打开http://localhost:3000，用刚才配置的管理员账号登录，第一步就算完成了。
3.2 密钥体系与 Lanchain 的连通性验证
在 Langfuse 后台，你需要创建项目并获取两把钥匙：公钥（Public Key）和私钥（Secret Key）。这个概念和很多人的直觉相反——公钥反而要放在前端或环境里用于上报数据，私钥要保密。为了让你后端调用时能读写，我建议在 FastAPI 服务的环境变量里这样配置：
LANGFUSE_SECRET_KEY=sk-lf-xxxx LANGFUSE_PUBLIC_KEY=pk-lf-xxxx LANGFUSE_HOST=http://localhost:3000这三个变量，Langfuse 的 SDK 和 Langchain 的 CallbackHandler 会自动读取，你不需要在代码里硬编码。
配置完成后，验证连通性的最快方式是写一段极简的测试代码：
from langfuse import Langfuse langfuse = Langfuse() # 如果配置正确，后台会看到这个 trace langfuse.trace(name="connection-test").update(input="hello", output="world") langfuse.flush()跑完去 Langfuse 后台看，如果connection-test出现在 Trace 列表里，说明部署和密钥都通了。这一步不要跳过，我见过太多人代码全对，结果卡在密钥配置上，浪费几个小时。
3.3 版本匹配：Langfuse 和 Langchain 的兼容问题
这是最容易被忽略的问题，也是让我踩坑最深的点。Langfuse 对 Langchain 的集成高度依赖回调接口的具体实现，Langchain 升级后回调签名变了，Langfuse 旧版根本识别不了。
我的建议是，统一使用较新且互相兼容的版本组合：
| 软件包 | 我使用的版本 |
|---|---|
| langfuse | 2.41.x 或以上 |
| langchain | 0.2.x 或以上 |
| langchain-openai | 0.1.x 或以上 |
| fastapi | 0.115.x |
| uvicorn | 0.30.x |
安装命令一并给你：
pip install "langfuse>=2.41" "langchain>=0.2" "langchain-openai>=0.1" "fastapi>=0.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_url：
from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com", temperature=0.7, )第二种方式，如果你的 Langchain 版本较新，也可以用init_chat_model统一入口：
from langchain.chat_models import init_chat_model llm = init_chat_model( "deepseek-chat", model_provider="openai", base_url="https://api.deepseek.com", api_key=os.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。
先初始化 handler：
from 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或astream：
async 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 页面加原生 WebSocket：
const 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 打一个 score：
from langfuse import Langfuse langfuse_client = Langfuse() @app.post("/score/{trace_id}") async def score_trace(trace_id: str, score: int): langfuse_client.score( trace_id=trace_id, name="user-feedback", value=score, ) return {"status": "ok"}这个功能看似简单，价值却很大。它能让你把不同提示词、不同参数下模型的回答质量量化下来，长期积累的数据对优化 prompt 非常有用。
6. 实测踩坑记录：这套架构最容易翻车的地方
最后一部分，我把我实际运行这套架构时遇到的坑集中列出来。每个坑都花了我不少时间排查，希望你能直接避开。
6.1 回调不生效：Langchain 版本和回调传入方式的坑
最典型的报错是：代码不报错，Langfuse 后台也看得见项目，但就是没有 trace 进来。排查思路先看 Langchain 版本是不是太老或太新。我踩过一次很深的坑，当时 Langchain 升级到 0.3.x，CallbackHandler的接口有了变化，老版本的 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 小节稍微提过，这里再强调一下。多用户并发时，如果不为每次请求新建CallbackHandler，trace 会互相穿插，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 的报表接口搬到自己的内网看板里，也可以把评分数据接入告警系统。动手跑通第一版，比继续读十篇架构分析更有价值。
