先说个实际场景。你负责的AI对话服务最近总被吐槽产品经理甩过来一条用户反馈说“这回答明显跑偏了”你跑到服务器翻了半天日志只看到一堆时间戳和请求记录既不知道模型当时拿到了什么上下文也不知道为什么这次请求的token开销比平时翻了一倍。更难受的是这种问题不稳定复现等你连上环境现场早就没了。我前阵子也被这个折磨得够呛后来用Langfuse Langchain DeepSeek FastAPI WebSocket搭了一套实时对话监控仪表盘把每次对话的完整链路全部拉出来晒在面板上定位问题从小时级降到了分钟级。这篇文章就是这套架构从零到一的完整实践核心代码全部贴出来照着跑就能复现。1. 先聊聊为什么要给AI对话装上“监控仪表盘”1.1 大模型应用的黑盒困境传统后端服务出问题链路上每个服务都会打日志排查起来层层递进就行。但LLM应用不太一样你调一次模型输入是一个带系统提示词和对话历史的完整消息数组输出是一段文本加一堆token统计中间过程对开发方来说完全是个黑盒。同一个问题模型在不同温度、不同历史长度、不同系统提示词下可能给出完全不同的答案。我踩过最典型的一次坑线上知识库问答突然出现大量“答非所问”本地复现怎么都正常。后来才查出来是因为某轮对话的上下文越积越长把系统提示词的约束效果给稀释了。这种情况靠传统日志根本看不出来你需要的是能观察“模型实际看到了什么”的工具。1.2 Langfuse在可观测性体系里的定位Langfuse就是一个专门给LLM应用做的可观测性平台它把每次对话抽象成一条TraceTrace下面挂着不同层级的Observation比如LLM调用、检索器调用、Agent动作、自定义业务步骤等。每一条Observation都记录输入、输出、模型名、token用量、延迟时间、成本估算还能给Trace关联Session和User。对比传统日志Langfuse解决的不只是“有没有日志”的问题而是帮你把日志组织成结构化的追踪链路。传统日志是一堆行文本你需要自己拿request_id去串Langfuse天然就是树形结构打开一条Trace就能看到完整调用链。说实话用过一次就回不去了。1.3 实时监控和事后排查是两件事Langfuse本身是事后查看的工具但对话质量类问题往往需要实时发现。比如你在调试prompt想让每组修改立刻反映到运行效果上或者你想观察某个上线版本的模型回答是否有异常等日志落盘再聚合到Grafana中间延迟太高了。所以我在Langfuse之上加了一层实时推送后端处理完每次对话通过WebSocket把消息、token、trace链接推到仪表盘打开页面就能看到实时跳动的对话流同时还能点进Langfuse看完整调用链。这套组合拳用下来一个面板搞定实时观测和深度追踪两层诉求。2. 技术选型拆解四件套的职责边界与协作逻辑2.1 Langchain、DeepSeek、FastAPI、WebSocket各管什么这四个组件分工非常清晰我用一个表来说明组件职责类比Langchain对话编排、消息构造、模型调用的标准化入口总装车间DeepSeek实际生成回答的大模型通过API提供服务核心引擎FastAPIHTTP/WebSocket服务框架负责接口暴露和生命周期管理门卫调度室WebSocket服务端到浏览器的实时推送通道高速传送带Langchain的价值在于你不必手写模板字符串拼接各种消息它的ChatOpenAI接DeepSeek非常顺因为DeepSeek的API就是OpenAI兼容格式只需要改base_url和model名其余逻辑完全复用。Langchain还顺手帮你处理了重试、超时、token统计这些脏活。2.2 为什么不用轮询而要用WebSocket监控面板这类场景最朴素的实现是前端每隔1秒轮询一次后端接口。轮询的问题有两个一是空载时大量HTTP请求打到服务端纯粹浪费资源二是延迟不稳定顶多做到“准实时”。WebSocket建立一条长连接服务端有数据直接推过去延迟几乎为零对实时监控来说是更合理的选择。成本上一个WebSocket长连接的维护开销远小于高频HTTP轮询特别是在多客户端同时围观监控面板的场景下优势更明显。FastAPI原生支持WebSocket写一个/ws/chat接口配合ConnectionManager做连接池管理和广播也就几十行代码。2.3 Langchain和Langgraph的区别这个场景该怎么选经常看到有人问Langchain和Langgraph的关系通俗点说Langchain是基础工具库像一个大工具箱提供模型调用、消息封装、链式编排这些能力Langgraph是建立在Langchain之上的状态机编排框架专门跑那种有状态、有循环、有条件分支的复杂Agent流程。我们这个监控仪表盘场景中间没有复杂的状态流转也不需要循环执行Agent工具用Langchain的普通异步调用就足够。如果你后面要做的Agent需要“根据前一步结果决定下一步动作”再考虑引入Langgraph不迟现阶段不要引入它增加心智负担。2.4 DeepSeek在生态里的兼容性优势DeepSeek开放平台的接口格式和OpenAI保持一致这意味着Langchain生态里凡是支持OpenAI格式的组件几乎都能无缝切到DeepSeek。设置base_url指向https://api.deepseek.com模型名填deepseek-chatAPI key填DeepSeek平台生成的key就能直接跑起来。这个兼容性给了架构很大的弹性今天用DeepSeek跑对话明天想换其他模型改一行配置就行监控层和后端服务完全不用动。3. 环境准备Langfuse自托管与项目骨架搭建3.1 Langfuse部署方式选择Langfuse有云托管版和自托管版。自己搭监控面板做技术验证我建议自托管原因很直接数据不出内网调试期随便造数据不用心疼额度还能顺便学习它的部署依赖。最简单的自托管方式是直接拉官方仓库跑docker composegit clone https://github.com/langfuse/langfuse.git cd langfuse docker compose up -d官方仓库根目录自带完整的compose编排包含Postgres、Redis、Web服务、Worker服务四个容器。等容器状态都healthy之后浏览器打开http://localhost:3000首次访问会让你创建管理员账号然后登录进去创建一个Project拿到一组Public Key和Secret Key。注意Langfuse的Secret Key和DeepSeek的API Key虽然都叫sk开头但完全是两码事。前者用来往Langfuse服务写入追踪数据后者用来调用模型别搞混。3.2 项目目录结构后端和前端放在同一个仓库里目录结构如下ai-monitor-dashboard/ ├── .env # 环境变量 ├── requirements.txt # Python依赖 ├── backend/ │ ├── ai_client.py # Langchain DeepSeek Langfuse追踪 │ ├── ws_manager.py # WebSocket连接管理 │ └── main.py # FastAPI应用入口 └── frontend/ └── index.html # 监控仪表盘页面后端单独开一个虚拟环境我用的是python -m venv venv创建也可以直接用uv venv速度更快。激活环境后安装依赖pip install -r requirements.txtrequirements.txt内容如下版本号不需要完全一致按这个范围装就行fastapi0.115 uvicorn[standard]0.30 langchain0.3 langchain-openai0.2 langfuse2.50 python-dotenv1.03.3 环境变量配置在项目根目录建.env文件DEEPSEEK_API_KEYsk-你的deepseek_key DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat LANGFUSE_PUBLIC_KEYpk-你的langfuse公钥 LANGFUSE_SECRET_KEYsk-你的langfuse私钥 LANGFUSE_HOSThttp://localhost:3000后端启动时通过load_dotenv()加载这些配置。Langfuse的SDK会自动读取LANGFUSE_*这三个环境变量这意味着只要配好了你甚至不用在代码里显式初始化Langfuse客户端装饰器就能自动连接服务。这里有个细节值得提一下如果你后面把Langfuse服务挂到了非3000端口务必把LANGFUSE_HOST改成对应地址不然SDK默认往http://localhost:3000写数据服务对不上号数据就丢了。4. 对话链路实现Langchain DeepSeek 接入与Langfuse追踪埋点4.1 DeepSeek接入Langchain的正确姿势Langchain生态里对接OpenAI兼容接口是langchain-openai包里的ChatOpenAI类。DeepSeek只需要换三个参数from langchain_openai import ChatOpenAI llm ChatOpenAI( modeldeepseek-chat, api_keysk-xxx, base_urlhttps://api.deepseek.com, temperature0.7, timeout60, max_retries2, )ChatOpenAI内部会把用户消息转成OpenAI协议格式发给base_url对应的服务所以DeepSeek平台能直接识别。消息构造用的是Langchain的HumanMessage、SystemMessage、AIMessage三个类from langchain_core.messages import HumanMessage, SystemMessage, AIMessage messages [ SystemMessage(content你是一个乐于助人的AI助手。), HumanMessage(content你好), ]对话历史我建议只保留最近几轮全量塞给模型既贵又慢通常20轮以内的上下文足够保持连贯性。4.2 追踪埋点的两种方式装饰器与CallbackHandlerLangfuse接入Langchain最常用的两种方式。第一种是observe装饰器直接标注你关心的业务函数装饰器会自动把函数内部所有被Langfuse支持的调用串成一条Tracefrom langfuse.decorators import observe, langfuse_context observe(as_typegeneration) async def generate(self, user_message, historyNone): response await self.llm.ainvoke(messages) return response第二种是CallbackHandler适合你想精确控制哪一段Langchain调用被追踪的场景from langfuse.callback import CallbackHandler handler CallbackHandler() response llm.ainvoke(messages, config{callbacks: [handler]}) trace_id handler.get_trace_id()两者选哪个我的建议是优先用observe装饰器理由有三个代码侵入小、自动追踪嵌套调用、不需要手动管理handler的生命周期。如果你用的是最新Langfuse版本装饰器底层已经做好了和Langchain回调机制的集成装饰器套在调用Langchain的函数上就能完整记录到模型的输入输出和token统计。4.3 拿到trace_id并拼出可跳转的追踪链接监控仪表盘上最有价值的交互就是每条AI回复后面跟一个“查看Trace”的链接点过去直接落到Langfuse的对应调用链详情页。在装饰器函数内部langfuse_context.get_current_trace_id()可以拿到当前上下文里的Trace ID再拼上Langfuse的页面地址就得到完整链接trace_id langfuse_context.get_current_trace_id() trace_url fhttp://localhost:3000/trace/{trace_id}生产环境里这个URL前缀别硬编码建议从环境变量读因为Langfuse部署地址变更后这里记得同步。4.4 手动补充业务元数据Langfuse自动采集的信息包括模型名、输入输出、token用量和耗时但有些业务维度的信息它不知道。比如当前请求来自哪个渠道、用户ID是什么、这次对话属于哪个Session这些建议用langfuse_context.update_current_trace挂上去langfuse_context.update_current_trace( nameai_chat, session_idfsession_{uuid.uuid4().hex[:8]}, metadata{source: monitor_dashboard}, )挂完之后Langfuse的Session面板就能按会话维度过滤后续做用户级质量分析会方便很多。下面这一版ai_client.py完整代码把上面提到的内容全部揉在一起了import os import uuid from dotenv import load_dotenv from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from langchain_openai import ChatOpenAI from langfuse.decorators import observe, langfuse_context load_dotenv() class DeepSeekChat: def __init__(self, system_prompt: str 你是一个乐于助人的AI助手。): self.llm ChatOpenAI( modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com), temperature0.7, timeout60, max_retries2, ) self.system_prompt system_prompt observe(as_typegeneration) async def generate(self, user_message: str, history: list[dict] | None None): messages [SystemMessage(contentself.system_prompt)] if history: for turn in history: messages.append(HumanMessage(contentturn.get(user, ))) messages.append(AIMessage(contentturn.get(assistant, ))) messages.append(HumanMessage(contentuser_message)) langfuse_context.update_current_trace( nameai_chat, session_idfsession_{uuid.uuid4().hex[:8]}, metadata{source: monitor_dashboard}, ) response await self.llm.ainvoke(messages) trace_id langfuse_context.get_current_trace_id() usage getattr(response, usage_metadata, {}) or {} model_name response.response_metadata.get(model, deepseek-chat) return { content: response.content, trace_id: trace_id, model: model_name, usage: { input_tokens: usage.get(input_tokens, 0), output_tokens: usage.get(output_tokens, 0), total_tokens: usage.get(total_tokens, 0), }, }注意response.usage_metadata是langchain-openai新版提供的数据结构包含input_tokens、output_tokens、total_tokens三个字段比旧的response_metadata.token_usage更稳定。如果你的版本比较老取不到usage_metadata就到response_metadata里翻token_usage字段兜底。5. 实时通道实现FastAPI WebSocket 的服务端编排5.1 WebSocket连接池ConnectionManager服务端需要维护所有活跃的WebSocket连接客户端断开时要及时清理否则广播时会往死连接里写数据触发异常。from fastapi import WebSocket 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): stale [] for connection in self.active_connections: try: await connection.send_json(message) except Exception: stale.append(connection) for connection in stale: self.disconnect(connection)广播时我把发送异常但还残留在列表里的连接收集起来统一剔除。这个操作很关键不然跑一阵子就会遇到类似failed to send websocket request: io的报错本质都是往已断连的socket写数据导致的。5.2 WebSocket接口的完整编排逻辑FastAPI里写WebSocket接口非常直接app.websocket(/ws/chat)装饰器一挂函数签名接收WebSocket对象就行。核心流程在while True循环里接收前端发来的文本解析JSON按类型分发处理。import json from datetime import datetime from pathlib import Path from dotenv import load_dotenv from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from ai_client import DeepSeekChat from ws_manager import ConnectionManager load_dotenv() app FastAPI(titleAI对话监控仪表盘) app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) chat DeepSeekChat() manager ConnectionManager() def now_text(): return datetime.now().strftime(%H:%M:%S) app.websocket(/ws/chat) async def chat_endpoint(websocket: WebSocket): await manager.connect(websocket) history: list[dict] [] try: while True: raw await websocket.receive_text() payload json.loads(raw) if payload.get(type) ping: await websocket.send_json({type: pong, timestamp: now_text()}) continue user_msg payload.get(message, ).strip() if not user_msg: continue await manager.broadcast({ type: user, content: user_msg, timestamp: now_text(), }) try: result await chat.generate(user_msg, history) history.append({user: user_msg, assistant: result[content]}) if len(history) 20: history history[-20:] trace_url fhttp://localhost:3000/trace/{result[trace_id]} await manager.broadcast({ type: assistant, content: result[content], timestamp: now_text(), model: result[model], usage: result[usage], trace_url: trace_url, }) except Exception as e: await manager.broadcast({ type: error, content: str(e), timestamp: now_text(), }) except WebSocketDisconnect: manager.disconnect(websocket) except Exception: manager.disconnect(websocket) app.mount(/, StaticFiles(directory../frontend, htmlTrue), namefrontend)这段代码有几个点值得解释第一每个连接维护一个独立的history列表多个浏览器打开监控面板时每个面板自己的对话上下文互不干扰。如果你希望所有人的对话能共享上下文需要把history提升到全局或存到Redis里这个按需调整。第二type字段做了消息分发约定。前端发过来的消息有chat和ping两种后端广播出去的消息有user、assistant、error、pong四种。前端拿到后按类型渲染逻辑清晰。第三app.mount(/, StaticFiles(...))这段必须放在所有路由注册之后否则会抢先匹配掉/ws/chat。5.3 为什么AI调用和前端推送放同一个WebSocket接口有些设计会把AI调用和消息推送拆成两个独立接口前端先调HTTP接口等AI结果再通过WebSocket收推送。这个方案增加了前后端的协调成本我直接简化成前端把用户消息从WebSocket发上来后端收到后调用AI结果也通过同一个连接广播出去。好处很直接一次连接解决全部通信消息顺序天然一致前端代码也简单。坏处是如果后续要做复杂的权限控制可能得在消息体里加身份字段但现在监控面板是内部工具不需要这个复杂度。6. 仪表盘前端一个HTML文件里的实时监控面板6.1 页面布局与功能设计前端我刻意没上Vue或React就一个原生HTML文件零依赖、开箱即用你甚至不需要打包。布局分成左右两栏左侧是监控概览右侧是对话流面板。左侧展示三块核心指标WebSocket连接状态、累计对话轮数、累计Token消耗下面挂一个最近Trace链接列表。右侧是对话消息流用户消息靠右蓝色背景AI回复靠左深色背景每条消息附带时间、模型名、Token数和Trace链接。CSS方面用了简单的深色主题终端风格和监控面板的气质比较搭关键数字用大字号突出连接状态有绿黄红三色圆点标识。6.2 WebSocket客户端的连接与断线处理前端通过new WebSocket(ws:// location.host /ws/chat)建立连接。域名和端口跟页面保持一致开发环境下页面跑在http://localhost:8000WebSocket地址就是ws://localhost:8000/ws/chat。断线重连是WebSocket应用中必须要处理的问题。浏览器端WebSocket断开时onclose事件会带一个code参数我见过最多的是1006这个code表示连接异常断开通常不是客户端主动关闭而是网络中断或服务端进程崩溃。简单方案检测到onclose后提示用户连接断开3秒后location.reload()整页刷新重新走一遍建连逻辑。粗暴但有效监控面板这种内部工具完全够用。6.3 消息渲染与Trace链接跳转收到后端消息后按data.type做分支渲染。前端代码核心如下socket.onmessage (event) { const data JSON.parse(event.data); switch (data.type) { case user: appendMessage(user, data.content, { text: data.timestamp }); break; case assistant: msgCount; const total data.usage ? data.usage.total_tokens : 0; tokenCount total; document.getElementById(msgCount).textContent msgCount; document.getElementById(tokenCount).textContent tokenCount; let metaText ${data.timestamp} | 模型${data.model} | Token${total}; appendMessage( assistant, data.content, { text: metaText }, data.trace_url ? [{ url: data.trace_url, label: 查看Trace }] : [] ); break; case error: appendMessage(error, data.content, { text: data.timestamp }); break; } };appendMessage函数支持额外传入链接数组渲染时把Trace链接拼在meta信息后面点击新窗口打开Langfuse详情页。这样监控面板上就能做到“看到问题直接点链接看链路的完整追踪信息”。完整的前端代码比较长我在文章末尾的源代码清单里给出完整版本这里只贴关键片段。整体逻辑就是连接WebSocket、按类型渲染消息、维护统计数字、断线自动刷新。7. 全流程实测与踩坑记录7.1 启动顺序与验证步骤整套服务的启动顺序按依赖关系来# 1. 启动Langfuse如果容器没在跑 docker compose up -d # 2. 启动后端 cd backend uvicorn main:app --reload --port 8000 # 3. 打开浏览器 # 访问 http://localhost:8000/打开页面后确认左侧连接状态变成“已连接”和绿点。在输入框里输一句“你好”右侧会依次出现你的消息和AI回复AI回复的meta行里能看到模型名、Token数和一个“查看Trace”链接。点击链接浏览器会打开Langfuse详情页能看到完整的调用链。如果一切正常Langfuse的Trace详情页大概长这样最上层是一条ai_chat的Trace下面挂着一个generation类型的ObservationObservation里记录着DeepSeek的输入消息、输出内容、token统计和耗时。7.2 Langfuse看板上的数据长什么样跑个十几轮对话之后回到Langfuse首页左侧的Trace列表里能看到所有历史对话按时间倒序排列。点进任意一条顶部展示Trace的Session ID、输入输出摘要、延迟、总Token数和估算成本。往下滚动可以看到消息级别的详情。我特别关注的是模型输入里的完整消息数组系统提示词、用户问题、历史上下文都在里面。之前排查的“答非所问”问题在这里一眼就能看出来——系统提示词已经被20多轮的历史消息淹没了。这个视角是传统日志给不了的Langfuse把模型实际看到的原始上下文完整摊开所有“隐藏的prompt影响”都无处遁形。这也是我强烈建议任何做LLM应用的人都配一套Langfuse的原因。7.3 我踩过的坑与解决方式坑一observe装饰器导入报错某次新建环境pip install langfuse装出来的版本比较老from langfuse.decorators import observe, langfuse_context直接抛ImportError。解决方式是升级langfuse到2.30以上如果还有问题检查项目里有没有装多个版本的langfuse建议先pip uninstall langfuse再重装。坑二uvicorn热更新不生效--reload参数不起作用改代码后服务不自动重启。这个坑的根源是缺了watchfiles这个依赖uvicorn[standard]扩展包里才带它只装uvicorn的话reload功能会静默失效。解决方式就是安装uvicorn[standard]重装后先uvicorn main:app --reload --port 8000验证一下改代码是否自动重启。坑三WebSocket广播报Stream disconnected或code 1006这个问题我之前碰到过诱因是前端断线后服务端还挂着旧连接广播时往死连接写数据。解决方式就是ConnectionManager.broadcast里的异常剔除逻辑每次广播捕获发送异常并清理连接。另外如果后面你用Nginx代理WebSocket记得把代理的超时时间调长并在location配置里加上proxy_http_version 1.1和Upgrade相关的header否则空闲连接会被Nginx掐断前端就会频繁收到1006。坑四Langfuse的Trace链接打不开Trace链接直接拼http://localhost:3000/trace/{id}如果你在浏览器上看到404先确认Langfuse容器是不是真的在3000端口。另外如果前端页面通过别的域名访问后端Trace链接里的localhost指的是用户自己的浏览器机器不是后端服务器。这种情况最好把LANGFUSE_PUBLIC_URL单独配置从环境变量读出来再拼。7.4 进阶优化方向这套基础架构跑通之后有几个方向可以继续扩展。一是引入Langchain的RunnableParallel并行调用不同模型。比如同时请求deepseek-chat和deepseek-reasoner两个模型对比各自回答Langfuse会自动在一条Trace下面生成两条generation Observation做模型横向评测非常方便。二是把history存储从内存换成Redis。多实例部署时每个实例各存各的内存态会有会话丢失问题Redis能统一管理会话也方便做会话维度的聚合分析。三是对接告警。Langfuse支持Webhook你可以配置当某条Trace的响应延迟超过阈值或某类错误出现时自动推送到企业微信或钉钉群。这个功能对于线上服务的日常巡检很实用。我个人的体会是LLM应用的可观测性建设越早做越省心。前期不重视等线上问题多到排查不过来再想补历史数据已经全部丢失了。这套Langfuse Langchain DeepSeek FastAPI WebSocket的架构代码量不算大但把“实时监控”和“深度追踪”两个关键能力都补齐了后续不管是做模型评测还是prompt调优都能有一份清晰的数据底子可以依赖。