最近在调 LLM 应用时我特别强烈地感觉到一件事对话能跑通只是万里长征第一步。真正让人头大的是模型为什么突然变慢、为什么同一个问题两次回答的风格差那么多、以及一次调用到底烧了多少 token。为了把这些事从“靠感觉”变成“看得见”我把 Langfuse、LangChain、DeepSeek、FastAPI、WebSocket 这条链路完整搭了一遍做了一个 AI 对话监控仪表盘。这篇文章就把从零搭建的过程、核心源码、以及我在联调里踩过的坑一次性拆出来。不管你是刚接触 LangChain 的初学者还是已经在做 agent 应用、需要一个轻量可观测方案的后端工程师这套架构的思路和代码都能直接拿去改。核心是Langfuse 负责记录和追踪FastAPI 负责业务接口WebSocket 负责把监控数据实时推到仪表盘DeepSeek 负责真正的对话能力。四者各司其职组合起来就是一套可以持续观察线上对话质量的基础设施。1. 为什么需要一套对话监控仪表盘从光是“能跑”到“追得了根”1.1 传统日志体系在 LLM 场景下的失效如果只是做一个普通的 CRUD 接口打印一下请求参数和返回状态码就够了。但 LLM 应用完全是另一回事。模型输出没有确定性同样的 prompt 在不同温度下可能给出截然不同的答案调用延迟波动很大首 token 延迟从几百毫秒到几十秒都可能上下文窗口会被多轮对话慢慢撑满token 消耗也会悄悄膨胀。更麻烦的是当业务链路里出现工具调用、RAG 检索、多步推理时单条日志根本表达不了“这次回答是怎么一步步得出来的”。我在项目初期就是简单地把用户提问和模型回答打到日志文件里结果排查一个问答质量问题时根本看不清是哪一次检索给错了材料、哪一层 prompt 写串了。这时候需要的不是日志而是 trace——一次完整请求内部所有步骤的关联记录。1.2 技术选型为什么是这四个组件选 Langfuse 做可观测底座因为它就是专门为 LLM 调用的 trace、span、token 统计和评测场景设计的自带 UI不用自己从零写一套数据模型和前端页面。LangChain 的价值在于抽象层模型、提示词、工具、输出解析器都可以以统一方式组合以后换模型或加检索都不会把业务代码推翻重来。DeepSeek 的接入成本极低接口是 OpenAI 兼容格式只要把 base_url 和 model 名改掉就能跑通。FastAPI 的异步特性配合 WebSocket 天然适合做实时推送而且写起来很轻不需要额外引入重量级框架。这套组合最舒服的地方是Langfuse 管“事后查”WebSocket 管“实时看”LangChain 管“链路编排”DeepSeek 管“智能生成”。每层的职责非常清晰扩展起来也很方便。2. Langfuse 自托管部署、初始化与接入 LangChain2.1 Docker 部署步骤Langfuse 官方提供了一键部署方式我使用的是 Docker Compose 自托管方案。先把官方部署目录拉下来里面包含 Langfuse 主服务以及依赖的数据库组件然后设置好环境变量再启动。需要关注的环境变量主要有这么几个变量作用说明LANGFUSE_INIT_USER_EMAIL初始管理员邮箱首次部署时用来登录后台LANGFUSE_INIT_USER_PASSWORD初始管理员密码部署后建议尽快修改DATABASE_URL数据库连接串自托管时指向自己的数据存储ENCRYPTION_KEY数据加密密钥用于敏感字段加密SALT哈希盐值配合加密密钥使用启动命令很简单进入部署目录后执行docker compose up -d等镜像拉完、服务起来后访问http://localhost:3000就能打开 Langfuse 后台。首次登录时会让你设置项目创建好项目后在项目设置里拿到public_key和secret_key这两个 key 是后面 SDK 接入要用的凭证。提示ENCRYPTION_KEY 和 SALT 一旦定下来就不要随便改否则已有的加密数据会解不开。2.2 SDK 初始化与 LangChain 版本适配服务端跑起来之后Python 端只需要安装几个包langfuse、langchain、langchain-openai、fastapi、uvicorn、websockets。这里要先提醒一句新版 LangChain 已经把模型提供商拆分成了独立包老的langchain.llms.OpenAI这种导入方式在 0.3 之后不再推荐要使用langchain_openai.ChatOpenAI。DeepSeek 走的是 OpenAI 兼容协议所以ChatOpenAI完全够用。先初始化 Langfuse 客户端from langfuse import Langfuse langfuse Langfuse( public_keypk-xxxxx, secret_keysk-xxxxx, hosthttp://localhost:3000, debugFalse )接入 LangChain 时langfuse包提供了一个现成的回调处理器CallbackHandler把它挂到 chain 的调用配置里就行。这也是我最推荐的方式——不需要侵入业务逻辑去手动创建 traceLangfuse 会自动按 chain 的执行结构生成层级分明的记录。from langfuse.callback import CallbackHandler handler CallbackHandler( public_keypk-xxxxx, secret_keysk-xxxxx, hosthttp://localhost:3000 )2.3 最小可用的回调埋点代码下面是一段最简化的链路代码把 LangChain 的 chain 和 Langfuse 回调串起来from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI llm ChatOpenAI( modeldeepseek-chat, api_keysk-deepseek-xxxxx, base_urlhttps://api.deepseek.com, temperature0.7 ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个精炼的助手用中文回答问题。), (human, {question}) ]) chain prompt | llm | StrOutputParser() # 调用 chain 时挂上 Langfuse 回调 result chain.invoke( {question: 帮我解释一下什么是 Langfuse}, config{callbacks: [handler]} ) print(result)跑完这次调用后打开 Langfuse 后台在 Trace 列表里就能看到一条记录。点进去可以展开结构最外层是 Trace下面包含一次 Chain Span、一次 LLM Generation每个节点都有输入、输出、耗时、token 数量。这基本就是监控仪表盘的数据源头。2.4 数据到什么地步才算“被记录”Langfuse 后台默认展示的 Trace 列表包含名称、时间、用户 ID、session ID、总耗时、总 token 数等。在 UI 上展开一条 trace能看到完整的调用链。如果你只接了 LLM那记录会比较简单如果以后加了检索器、工具调用Langfuse 会自动把这些步骤变成 Trace 下面的子节点每一层都能单独看耗时和输入输出。我实际使用中比较注意的一个点是Langfuse 上报是异步的接口返回后数据不会立刻出现在后台通常有几百毫秒到几秒的延迟。这个延迟不影响事后排查但对“实时仪表盘”来说就有点不够了所以我在架构里让 FastAPI 业务层承担实时广播Langfuse 承担最终数据存储和查询两个各干各的。3. 接 DeepSeek 并建立可观测的调用链路3.1 DeepSeek 的 OpenAI 兼容方式DeepSeek 的接口设计成 OpenAI 兼容格式这意味着所有基于 OpenAI SDK 的封装都能直接复用。只要把base_url改成https://api.deepseek.com模型名换成deepseek-chat再把 API key 配上就完成了模型接入。在 LangChain 生态里用ChatOpenAI这个类就行不需要额外装 DeepSeek 专用包。这一点很省事因为项目里如果之前已经用了 LangChain 的抽象层切换模型几乎就是改一个字符串的事。3.2 用 LangChain 封装 DeepSeek以最简单的问答链路为例可以这样封装from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI deepseek_llm ChatOpenAI( modeldeepseek-chat, api_keysk-deepseek-xxxxx, base_urlhttps://api.deepseek.com, temperature0.5, max_tokens1024 ) prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深技术博主回答要通俗、准确。), (human, {question}) ]) qa_chain prompt | deepseek_llm | StrOutputParser()如果你在业务里还需要流式输出把调用方式换成chain.stream(...)也可以Langfuse 的回调处理器同样能捕获流式调用过程中的事件。流式场景下仪表盘推送最好等完整输出结束后再统一广播避免前端高频重绘导致卡顿。3.3 给追踪加上业务维度监控仪表盘如果只有模型名和耗时价值会单薄很多。接入真实业务后我会在每次调用时给 Langfuse 补上业务维度信息这样做有两个直接好处一是后期可以按用户、按功能模块筛选 trace二是定位问题时能快速锁定“这是哪个用户、哪个入口、哪个 session 出的问题”。from langfuse.callback import CallbackHandler handler CallbackHandler( public_keypk-xxxxx, secret_keysk-xxxxx, hosthttp://localhost:3000, user_iduser_12345, session_idsession_67890, metadata{ 来源: web, 功能模块: 智能客服, 版本: v1.2.0 }, tags[production, customer-service] )在 Langfuse 后台这些字段会直接显示在 trace 列表的对应列里。session_id 特别适合多轮对话场景——同一个 session 的多次请求会被串联在一起你可以完整地看到用户在这个会话里的上下文是怎么变化的。3.4 快速验证一次带监控的对话接入完上面的代码用这个简单的 chain 跑一次真实请求然后去 Langfuse 后台刷新列表。你会看到类似这样的层级Tracetrace_name 可自定义Chain SpanLangChain 执行GenerationDeepSeek 模型调用每个节点都自带 input、output、model、usage 等信息。如果第一次跑完没在后台看到数据不要慌先看两点一是host是否指向 Langfuse 服务地址二是 SDK 的debug参数是否打开。把debugTrue加上后终端会输出上报请求的日志能帮你快速定位问题。注意CallbackHandler复用时要小心。如果在一个循环里反复用同一个 handlerLangfuse 可能会按照默认逻辑把多次调用归到同一个 trace。我的做法是每个独立请求新建一个 handler或者通过trace_id参数显式控制分组。4. FastAPI WebSocket把 Langfuse 里的数据实时推到仪表盘4.1 FastAPI 的 WebSocket 连接管理如果仪表盘要实时刷新轮询 Langfuse 的 API 是个办法但延迟高、压力也不小。更优雅的方案是 FastAPI 和仪表盘之间建立 WebSocket 长连接服务端一旦检测到新的 LLM 调用完成就主动把数据推送到所有在线客户端。先写一个连接管理器维护所有活跃连接from fastapi import WebSocket, WebSocketDisconnect 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): if websocket in self.active_connections: self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections[:]: try: await connection.send_json(message) except Exception: # 连接可能已经断开清理掉 self.disconnect(connection) manager ConnectionManager()这里用send_json直接传字典FastAPI 会自动序列化成 JSON非常方便。广播时遍历的是副本[:]避免在迭代过程中修改列表导致异常。4.2 让 LLM 调用结束后自动广播WebSocket 建立后业务接口逻辑很简单用户通过 POST /chat 提交问题FastAPI 调用 LangChain 链模型回复完成后立即通过 manager 广播一条监控事件给所有前端。这一层要特别注意一个坑FastAPI 的 async 接口里如果直接同步调用 LangChainLLM 执行会阻塞事件循环导致 WebSocket 通道上的其他请求被卡住。我在联调时就遇到过一次 WebSocket 连接在请求中途被断开报stream disconnected before completion的错误怀疑就是事件循环被 LLM 调用占用导致的。解决办法是把耗时的同步调用丢到线程池里跑用asyncio.to_thread包一层。import asyncio import time from datetime import datetime, timezone from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): question: str app.post(/chat) async def chat(req: ChatRequest): question req.question start time.perf_counter() # 新建 handler确保 trace 独立 handler CallbackHandler( public_keypk-xxxxx, secret_keysk-xxxxx, hosthttp://localhost:3000, metadata{来源: web, 功能模块: monitor-demo} ) # 同步的 chain.invoke 放到线程池避免阻塞 WebSocket 事件循环 result await asyncio.to_thread( qa_chain.invoke, {question: question}, {callbacks: [handler]} ) duration_ms round((time.perf_counter() - start) * 1000, 2) payload { type: new_trace, data: { question: question, answer: result, model: deepseek-chat, latency_ms: duration_ms, ts: datetime.now(timezone.utc).isoformat() } } # 实时推送到仪表盘 await manager.broadcast(payload) return payload[data]广播的 payload 我取的是业务层最关心、最即时的一批字段问题、答案、模型名、耗时、时间戳。至于更完整的数据如 token 数、内部 span 结构交给 Langfuse 后台展示。这样设计既能保证仪表盘秒开秒刷又不丢失持久化的可观测数据。4.3 前端仪表盘怎么接前端用原生 HTML JavaScript 就能演示整套流程避免引入前端工程化的复杂度。核心逻辑是建立 WebSocket 连接然后在onmessage里把服务端推过来的数据渲染成一个监控卡片列表。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleAI 对话监控仪表盘/title /head body h1AI 对话监控/h1 div idtrace-list stylefont-family: monospace;/div script const listEl document.getElementById(trace-list); let ws null; function renderTrace(msg) { const item document.createElement(div); item.style.borderBottom 1px solid #eee; item.style.padding 10px 0; item.innerHTML divstrong时间/strong${msg.data.ts}/div divstrong模型/strong${msg.data.model}/div divstrong耗时/strong${msg.data.latency_ms} ms/div divstrong问题/strong${msg.data.question}/div divstrong回答/strong${msg.data.answer.slice(0, 200)}/div ; listEl.prepend(item); } function connect() { ws new WebSocket(ws://${location.host}/ws/monitor); ws.onopen () console.log(WebSocket connected); ws.onmessage (event) { const msg JSON.parse(event.data); if (msg.type new_trace) { renderTrace(msg); } }; ws.onclose (ev) { console.log(WebSocket closed, code:, ev.code, reason:, ev.reason); setTimeout(connect, 2000); }; } connect(); /script /body /htmlonclose里的自动重连逻辑非常关键。我在实际用的时候发现浏览器切后台、网络抖动、服务器重启都会导致 WebSocket 静默断开。如果不做重连仪表盘就会悄悄变成“死屏”看起来还在线其实数据早断了。FastAPI 这边提供一个返回 HTML 的接口和 WebSocket 接口from fastapi.responses import HTMLResponse HTML_PAGE 把上面的 HTML 内容放进来 app.get(/) async def index(): return HTMLResponse(HTML_PAGE) app.websocket(/ws/monitor) async def websocket_monitor(ws: WebSocket): await manager.connect(ws) try: while True: # 保持连接等待客户端消息也可用作心跳 text await ws.receive_text() if text ping: await ws.send_text(pong) except WebSocketDisconnect: manager.disconnect(ws)WebSocket 接口里我用receive_text()挂住连接顺便实现了心跳回包。前端如果每隔 15 秒发一条 ping服务端回 pong就能有效延长连接生命周期减少被中间层断开的情况。4.4 多客户端与断线重连的处理如果有多个打开仪表盘页面上面写的broadcast会自动把消息推给所有在线连接。我在测试时开了两个浏览器窗口一个手机浏览器模拟一个桌面浏览器两边都能在几百毫秒内看到新数据。断线重连还有个细节WebSocket接口不能在connect和broadcast之间同时被同一个连接重复注册。也就是说前端如果在上一次连接还没完全关闭时就执行new WebSocket可能造成服务端出现重复连接、后续广播的时候同一个前端收到多条重复数据。稳妥的做法是在onclose里先ws null再setTimeout(connect, 2000)。5. 端到端联调从用户提问到仪表盘刷新5.1 准备一个真实用例为了验证这套架构不只是“玩具 demo”我给它加了点业务味道。在 prompt 里让模型输出固定格式的结论然后把整个链路跑通。用户提问后FastAPI 收到请求LangChain 链调 DeepSeekDeepSeek 返回文本Langfuse 记录 trace同时 WebSocket 把概要信息推到前端仪表盘。这个流程里每一个环节都有单独验证的点FastAPI 接口层面curl POST /chat确认返回 200 和完整文本。Langfuse 后台出现对应 trace展开有 LLM 节点和 token 数。WebSocket 层浏览器控制台打印WebSocket connected页面上出现新卡片。我把qa_chain换成带一点 RAG 感觉的 prompt输入“介绍 Langfuse 的核心概念”输出按要点排布。这样跑出来的 trace 比较有分析价值也能看到 token 消耗不是小数目。5.2 联调中发现的数据时序问题联调时发现一个很有意思的现象业务接口已经返回了Langfuse 后台的 trace 列表却还是空的等过几秒刷新它才出现。这就是异步上报的时序差。如果仪表盘的数据完全依赖 Langfuse API那前端就得做“不确定延迟的轮询”实时性会大打折扣。所以我最终的方案是双轨制实时通道FastAPI 在 LLM 调用结束后立刻广播业务快照。持久化通道Langfuse 完整记录 trace供历史查询和深度分析。这两条通道的数据结构不完全一致实时通道的data是精简字段Langfuse 后台看的是完整 trace。这样取舍是有意为之仪表盘要的是“秒级可见”Langfuse 要的是“准确完整”。另外如果你想在实时通道里也展示 token 数和完整调用链可以在广播前调用 Langfuse SDK 的fetch_traces()去查一次最新 trace把详情取回来再推给前端。不过我不推荐这么做因为 Langfuse SDK 的查询本身有同步网络开销会把“实时”拖成“准实时”。除非你的调用量很小否则性价比不高。5.3 性能开销怎么看监控本身也是有成本的。Langfuse 回调是异步线程上报对 LangChain 主链路的影响很小但依然有网络 IO。如果业务量上来建议开启采样只记录一部分流量避免监控存储膨胀。Langfuse 的CallbackHandler支持传sample_ratio参数比如sample_ratio0.1就表示只采样 10% 的调用。我在测试环境是全量记录生产环境会酌情开启采样尤其是对话频率高的入口。WebSocket 广播的频率也需要控制。如果每次都把完整回答文本推过去高并发下前端渲染压力不小。通常我只推问题摘要、回答截断版、延迟和模型名想看完整内容再去 Langfuse 后台查。6. 踩坑记录与后续优化方向6.1 最容易踩的三个坑第一个坑是 LangChain 包结构变化。老教程里from langchain.llms import OpenAI的写法在新环境里直接报导入错误。现在的正确打开方式是from langchain_openai import ChatOpenAI。如果你从网上抄了一段旧代码跑不起来非常正常先用pip show langchain看清版本再确认装没装langchain-openai。第二个坑是 Langfuse 回调的 trace 归组。我一开始在同一个 FastAPI 服务里复用一个全局 handler结果发现所有用户的对话被串到了同一个 trace 下面排查问题时根本分不清谁是谁。后来改成每次请求创建新的 handler并显式传入user_id、session_id和metadata数据才干净。第三个坑是 WebSocket 连接在长任务执行中被断开。主要原因是在异步接口里同步调用模型阻塞了事件循环导致 WebSocket 没机会处理心跳包被网关判定为死连接。解决方式就是前文提到的asyncio.to_thread把 LLM 调用放到线程池。这个坑不踩一次真的很难意识到。6.2 真实调试技巧把 Langfuse SDK 的debug参数打开。它会在终端输出每次上报的 HTTP 状态码能帮你确认数据到底有没有发出去、是不是被服务端拒了。如果 WebSocket 连接在手机上很容易断检查是不是没有做心跳。我在前端加了定时ping服务端回pong后连接稳定性明显提升不再频繁触发code: 1006那种异常关闭。本地开发时建议用uvicorn app.main:app --reload --port 8000启动 FastAPIWebSocket 在重载时可能会断前端有重连逻辑就不影响调试。6.3 这个架构可以怎么扩展这套底座搭完之后我准备在它上面继续加三个能力一是把 Langfuse 后台的评分功能接到仪表盘前端不止看延迟还能看到每轮对话的质量评分二是接入 LangGraph 编排更复杂的 agent 流程让 trace 里出现工具调用、多步推理的完整路径三是根据延迟和 token 消耗设置告警阈值当某个入口的平均延迟超过预设值时通过 WebSocket 主动推送一条告警事件到监控大屏。从实际使用体验来说这套系统最大的价值不是“炫技”而是把以前靠猜、靠用户反馈才能发现的对话问题变成了打开页面就能看到的事实。我始终觉得LLM 应用的可观测性应该和业务功能同时建设而不是上线之后出了问题再补。先把监控仪表盘搭起来后面无论迭代 prompt 还是调整模型参数心里都有底。