直接说个真实现状大模型应用做出来容易上线之后想搞清楚“它到底怎么答的”“为什么答错”“一次会话烧了多少钱”难度反而比写业务代码高得多。很多团队停留在看日志文件的阶段可 LLM 的输出是非结构化的prompt 里改一个词、上下文里多一段历史结果就完全变了传统日志根本没法回答这些问题。我这次做的就是一个典型的 LLM 可观测性项目用 Langfuse 做全链路追踪Langchain 编排调用逻辑DeepSeek 提供对话推理能力FastAPI 承载 HTTP 和 WebSocket 服务最后在前端实时渲染出每一次对话的 token 消耗、耗时、模型回复和 trace 链路。整套架构从零开始搭代码可复现适合正在做 LLM 应用、准备引入线上监控体系的开发者参考。1. 整体架构设计监控仪表盘到底在解决什么问题动手写代码之前先花点时间想清楚“对话监控”这个需求要覆盖哪些对象。市面上很多文章一上来就贴 Langfuse 配置但如果不理解监控的数据流后面排查问题会非常被动。1.1 对话监控需要管住哪些数据我把它拆成四个维度成本维度每次请求的 prompt tokens、completion tokens、总 tokens。DeepSeek 虽然便宜但线上量大了之后成本同样会失控。性能维度首 token 延迟、总耗时、网络错误、模型限流。用户说“卡了”你得能直接看到是不是模型侧超时。质量维度模型输出的内容、当时使用的 prompt 版本、检索到的上下文片段。出 bad case 时能完整回放现场。行为维度用户的 session 从哪里来、在哪个会话分支上反复追问、是否触发了安全兜底。这四个维度的数据分散在 API 网关日志、应用日志、模型服务端日志里如果不做统一采集和关联查一个问题要在四五个系统之间来回跳。Langfuse 就是用来把这些数据串成一条 trace 的。尤其要强调session_id和trace_id的关联设计Langfuse 里一条 trace 对应一次完整对话链多个 trace 通过session_id聚合为一次用户会话。这样从“用户级别”下钻到“请求级别”才是一条清晰路径。1.2 技术选型为什么是 Langfuse Langchain DeepSeek 这条链选 Langfuse 而不是自研或者 LangSmith主要基于三个判断第一Langfuse 是开源的可以 Docker 自托管。数据不出内网对很多有合规要求的企业是硬条件。LangSmith 确实好用但数据要上云且付费之后才解锁完整功能。自研方案我直接否了一套 trace 系统从数据模型设计到 UI 展示没有两个月做不出来周期太长。第二Langfuse 对 Langchain 的集成度极高。只要在 Langchain 的 invoke / stream 调用里传入langfuse_handlertoken 用量、耗时、模型名、提示词内容、输出内容都会自动上报不需要手写插桩代码。这一点非常关键意味着能最大程度保持业务代码的干净。第三DeepSeek 是 OpenAI 兼容协议。也就是说我可以用langchain-openai这个包只改base_url和api_key就能接入 DeepSeek完全复用 OpenAI 生态的工具链。Langchain 这一层主要解决的是“调用链编排”问题比如后续要加多轮记忆、要接检索、要加路由都能在 Langchain 框架内完成而不是写一堆胶水函数。1.3 WebSocket 而不是轮询或 SSE实时监控场景第一反应是轮询每 2 秒拉一次 trace 列表。但轮询有两个问题一是延迟不可控拉取间隔内的监控事件不能立刻展示二是大量请求打在 FastAPI 上明明没有新数据也要空转。SSE 适合服务端单向推送但我们的仪表盘前端需要能主动发指令比如“查看某条 trace 的详细 span”“停止当前展示”这是双向通信需求SSE 实现不了。所以 WebSocket 是最合适的传输层方案。整条链路是浏览器通过 WebSocket 连到 FastAPIFastAPI 在调用 Langchain 时实时收到 token 级回调然后把这些监控事件通过 WebSocket 转发给前端。这样监控端看到的效果是对话还没结束token 消耗曲线已经在涨了。协议选择上我做了一个简单的对比最终确定 WebSocket方案延迟双向通信实现复杂度适用场景轮询高支持伪低低频数据拉取SSE低不支持低服务端单向推送WebSocket低支持中实时双向互动2. 环境准备先把地基打好这套系统的依赖面比较广有 Langfuse 服务端、Python 侧 SDK、前端展示三块。我先说目录结构和依赖安装再逐个服务配置。2.1 工程目录与依赖清单项目结构我采用单仓库方式前端用一个静态 HTML 文件承载不单独拆前端工程降低复现门槛ai-monitor-dashboard/ ├── backend/ │ ├── main.py # FastAPI 入口包含 WebSocket 路由 │ ├── llm_chain.py # Langchain 链模型配置 CallbackHandler │ ├── schemas.py # Pydantic 请求/响应模型 │ ├── requirements.txt │ └── .env # API Key、Langfuse 地址等敏感信息 ├── frontend/ │ └── index.html # 仪表盘页面 └── docker-compose.yml # Langfuse 服务端requirements.txt里的依赖版本需要注意Langchain 迭代非常快版本不锁定的话有些 API 会变。以下是我验证过的组合fastapi0.110.0 uvicorn[standard]0.29.0 websockets12.0 langchain0.1.16 langchain-openai0.1.3 langfuse2.19.0 openai1.30.0 pydantic2.7.0 python-dotenv1.0.1建议使用 Python 3.10 以上的版本主要原因是 Langchain 0.1.x 系列对 Pydantic v2 的适配在 3.10 下更稳定。我用 Python 3.11 实测下来没有兼容性问题。安装时先把虚拟环境建好直接用python -m venv .venv然后激活即可不建议把依赖装进全局环境。2.2 部署 Langfuse 并配置项目密钥Langfuse 的部署方式官方提供 Docker Compose。我在项目根目录放了一份docker-compose.yml核心服务包含web、worker、dbPostgreSQL、redis以及一个用于初始化数据库的migrate服务。直接执行docker-compose up -d第一次启动会自动初始化数据库之后访问http://localhost:3000用默认账号或注册页创建管理员账号。登录后进入项目设置创建新项目拿到两串关键密钥public_key和secret_key这两个值配置在 Python 服务端的.env里。这里有一个容易踩的坑Langfuse 的 API host 配置。如果你把 Langfuse 跑在 Docker 里而 FastAPI 服务在宿主机上直接运行host应该填http://localhost:3000。如果 FastAPI 也容器化那就要填 Docker 网络内的服务名比如http://web:3000。我一开始就是因为 host 填错导致 Python SDK 上报失败查了半天才发现是网络互通的问题。2.3 DeepSeek 接入参数与认证配置DeepSeek 官方提供了 OpenAI 兼容的 API 端点。在.env里这样配置DEEPSEEK_API_KEYsk-xxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com/v1 DEEPSEEK_MODELdeepseek-chat模型名这里提一下DeepSeek 有两个可用模型deepseek-chat和deepseek-reasoner。reasoner 是推理模型会输出推理链适合复杂逻辑任务chat 是通用对话模型响应快、成本低仪表盘演示用 chat 就够了。如果你要接入 Langfuse 的 token 统计DeepSeek 官方 API 会在响应里返回 usage 字段Langfuse 会自动读取不需要额外处理。3. 核心链路实现让一次对话从请求到追踪完整串联依赖装好、密钥配好之后进入正题。这一节是最关键的部分涉及三条主线Langchain 怎么调 DeepSeek、Langfuse 怎么把调用链记下来、FastAPI 怎么把这套逻辑暴露成服务。3.1 用 langchain-openai 对接 DeepSeek因为 DeepSeek 兼容 OpenAI 协议直接用ChatOpenAI类覆盖base_url就是完整的接入方式。看代码from langchain_openai import ChatOpenAI llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), temperature0.7, streamingTrue, max_tokens1024, )streamingTrue这行要特别留意。如果不开流式模型会等完整响应生成后再返回首 token 延迟高用户体感差开了之后Langchain 会以迭代器方式产出增量内容配合 WebSocket 就能做打字机效果。而且 Langfuse 对流的支持很完整流式调用结束后token 统计会自动汇总。3.2 通过 CallbackHandler 打通 Langfuse 追踪Langfuse 接入 Langchain 不复杂核心是初始化一个全局的CallbackHandler然后把它传给chain.invoke()或者chain.stream()from langfuse.callback import CallbackHandler langfuse_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), # 例如 http://localhost:3000 trace_nameai_dashboard_trace, session_idsession_id, user_iduser_id, ) response chain.stream( {input: user_message}, config{callbacks: [langfuse_handler]}, )这段代码是整套监控体系的数据入口。session_id和user_id由业务侧传入前者用来把多次对话聚合成一个会话后者标记具体用户。建议在入口函数中每次请求都新建 handler而不是复用同一个全局实例因为trace_name、session_id、user_id都需要随请求变化。3.3 设计 trace 结构让日志后续能查询、能对比Langfuse 的 trace 结构是一个树顶层 trace 下有 spanspan 下还可以有子 span最底层的叶子节点是 generation也就是一次模型调用。默认情况下Langchain 的CallbackHandler会自动创建一条 trace并把链中每一步都挂到这条 trace 下面包括模型调用、检索器调用、工具调用等。但自动创建的 trace 对业务可读性不够好。我的做法是在 handler 初始化时传入trace_name然后在invoke之前手动开启一个 span把业务上下文比如“用户来自哪个页面”“当前是第几轮对话”加到 span 的 metadata 里from langfuse import Langfuse langfuse Langfuse() with langfuse.start_span(namebusiness-context, input{session_id: session_id}) as span: span.update( metadata{ page_source: page_source, turn_index: turn_index, prompt_version: prompt_version, } ) response chain.stream(...) span.end(output{reply: response_text})这样做的好处是Langfuse 界面上筛 trace 的时候可以直接用 metadata 里的字段做过滤。比如“查出所有来自首页的坏样本”“这个 session 的第几轮触发了兜底”这类查询是否高效完全取决于你埋的 metadata 是否完整。3.4 用 FastAPI 把链路包成正式服务FastAPI 侧我提供了两个核心接口一个是 HTTP 接口/chat兼容普通 API 调用一个是 WebSocket 接口/ws用于仪表盘实时推送。先看 HTTP 接口的实现from fastapi import FastAPI, WebSocket, WebSocketDisconnect from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str session_id: str default user_id: str anonymous app.post(/chat) async def chat(req: ChatRequest): handler build_langfuse_handler(req.session_id, req.user_id) response await llm_chain.ainvoke( {input: req.message}, config{callbacks: [handler]}, ) return {reply: response.content}这里有个细节ainvoke是异步方法FastAPI 中必须用async def定义路由才能正确等待协程执行否则会阻塞事件循环。回调链config{callbacks: [handler]}是 Langchain 的标准传法确保这一步模型调用的所有事件都上报到 Langfuse。4. WebSocket 实时推送与前端仪表盘HTTP 接口只能做请求-响应式的监控用户发一条消息返回一条完整回复。但真正的实时监控仪表盘需要的是“对话还没结束监控数据就开始流动”的效果。WebSocket 就是为此服务的。4.1 连接管理器与消息协议设计先定义一个连接管理器负责维护所有活跃的 WebSocket 连接from typing import List 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): self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections.copy(): await connection.send_json(message) manager ConnectionManager()WebSocket 路由接收客户端的监控事件请求比如“请求建立连接、设置过滤条件”app.websocket(/ws) async def ws_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: data await websocket.receive_text() # data 是一个 JSON 字符串可以包含过滤条件、查询参数等 # 服务端可以按条件订阅简化场景下直接广播 except WebSocketDisconnect: manager.disconnect(websocket)消息协议这里建议提前定义好避免前后端各写各的。我用的是一组 JSON 消息类型通过type字段区分type方向内容chat_request客户端 - 服务端用户输入的消息token_delta服务端 - 客户端模型生成的增量文本trace_update服务端 - 客户端trace 汇总数据用量、耗时、状态trace_detail服务端 - 客户端指定 trace 的完整 span 结构error服务端 - 客户端异常信息4.2 流式响应过程中的监控事件下发核心业务逻辑在 WebSocket 路由内接收到用户消息后先创建 Langfuse handler然后调用 Langchain 的astream方法逐块获取模型输出同时按照约定向前端推送消息app.websocket(/ws) async def ws_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: data await websocket.receive_text() payload json.loads(data) if payload[type] chat_request: message payload[message] session_id payload.get(session_id, default) user_id payload.get(user_id, anonymous) handler build_langfuse_handler(session_id, user_id) try: async for chunk in llm_chain.astream( {input: message}, config{callbacks: [handler]}, ): if chunk.content: await websocket.send_json({ type: token_delta, content: chunk.content, }) except Exception as e: await websocket.send_json({ type: error, message: str(e), }) except WebSocketDisconnect: manager.disconnect(websocket)这里有一个 Langchain 流式输出结构的问题。astream产出的每个 chunk 是一个AIMessageChunk直接取chunk.content就是增量文本。但要特别注意流式结束时最后一个 chunk 可能不含 content却携带了 usage 信息比如累计 tokens。如果想在流式结束后从代码里拿到总 tokens需要自己累加或者等 Langfuse 汇总后通过 API 查询。为了实时性我在前端主要展示的是 token_delta 的字数和时间戳等 trace 汇总完成后再通过 Langfuse 接口拉一次精确的 usage 数据。4.3 前端页面与断线重连处理前端我用一个 HTML 文件搞定核心是WebSocket对象与图表的联动。简单来说页面底部是对话输入区左侧是实时指标卡总 token、当前耗时、模型名右侧是对话历史和 trace 树展示区。WebSocket 连上之后页面监听trace_update类型的消息来刷新指标。关键的断线重连逻辑也放在这段let socket null; let reconnectAttempts 0; function connectWebSocket() { socket new WebSocket(ws://${location.host}/ws); socket.onopen () { reconnectAttempts 0; console.log(WebSocket connected); }; socket.onmessage (event) { const msg JSON.parse(event.data); handleServerMessage(msg); }; socket.onclose (event) { if (event.code 1006) { // 1006 表示异常断开需要重连 const delay Math.min(3000 * (reconnectAttempts 1), 15000); reconnectAttempts; setTimeout(connectWebSocket, delay); } }; socket.onerror (error) { console.error(WebSocket error:, error); }; } function handleServerMessage(msg) { switch (msg.type) { case token_delta: appendTokenToContent(msg.content); break; case trace_update: updateMetricCards(msg); break; case trace_detail: renderTraceTree(msg); break; case error: showError(msg.message); break; default: console.warn(Unknown message type:, msg.type); } }重连逻辑里用了指数退避策略第一次断线后延迟 3 秒第二次 6 秒最多延迟 15 秒。这个策略对生产环境非常关键因为网络抖动、服务重启都会导致 WebSocket 断开没有重连机制仪表盘就得手动刷新。5. 常见问题排查与踩坑实录整套系统跑通之后我把遇到过的典型问题整理成了排查清单。这些问题在 Langfuse、WebSocket、FastAPI 的日常使用中很常见先看速查表再看详细踩坑过程。现象涉及环节排查要点Langfuse 部署后登录白屏Langfuse检查 Docker 日志确认迁移服务是否完成模型能回复但 Langfuse 没有 trace上报链路确认 host 是否为宿主机可达地址检查 public_key/secret_keyWebSocket 自动断开code 1006前后端连接看服务端日志是否被异常中断检查代理层 idle timeoutDeepSeek 返回超时模型调用看 Langchain 是否有重试机制确认 max_tokens 是否合理FastAPI 改了代码不生效开发体验确认 uvicorn 是否开启--reload且代码目录正确5.1 你可能也会遇到的几个高频问题第一个问题Trace 汇总数据在流式结束后延迟才到。Langfuse 的 token usage 并不是实时写入的流式调用结束后handler 需要异步上报数据到服务端再由后台 worker 写入数据库。这个过程通常有几秒延迟。仪表盘如果要求“秒级精确”建议在流式结束时前端先显示临时估算值比如按字符数估算等 Langfuse 数据落库后再修正。第二个问题是 Langchain 版本差异导致CallbackHandler的导入路径变化。老版本是from langfuse.callback import CallbackHandler新版本在 langfuse v3/v4 里改成了from langfuse.callback import CallbackHandler没有变但 Langfuse 2.x 与 3.x 的 Python SDK 有破坏性变更。我最初用的 langfuse 2.x 配 langchain 0.1.x必须锁定版本后来升级到 3.x 又调整了依赖。建议统一使用我在 requirements 里锁定的版本组合。第三个问题是 DeepSeek 的base_url。有人习惯在 URL 末尾加/chat/completions这是不对的ChatOpenAI类自己会拼接路径base_url只应该配到 API 根地址也就是https://api.deepseek.com/v1。配错之后模型会报 404 或路由错误。5.2 三个印象最深的坑第一个坑是 Langfuse 自托管时数据库没初始化成功。docker-compose 起完服务后我访问 Langfuse 首页发现是白屏看容器日志才发现migrate服务启动失败原因是 PostgreSQL 数据卷权限问题。解决方法是删除数据卷重新创建或者给数据卷挂载目录授权后再启动。第二个坑是 WebSocket 断线重连。前后端都在本地时不会暴露问题但一旦通过 Nginx 反向代理WebSocket 默认 60 秒没有数据传输就会被断开表现为前端收到[websocket] onclose, code: 1006。解决办法是在 Nginx 配置里设置更长的 proxy_read_timeout同时前端配合指数退避重连。这个坑最容易在生产环境出现建议一开始就写进部署文档。第三个坑是 FastAPI 打包成 APP 后 WebSocket 连接失败。标题场景里提到“打包为 app 连接不了”我也遇到过类似情况。桌面端 APP 或移动 APP 里WebSocket 地址拼接要特别注意如果是文件协议打开 HTMLlocation.host可能是空的要么写死服务端地址要么加一层配置文件。另外 APP 连接 WebSocket 时协议要写ws://或wss://如果服务端上了 HTTPSWebSocket 必须用wss否则浏览器层面直接拦截。5.3 一些基于实战的调优建议第一Langfuse 的上报策略在生产环境要做好采样。默认全量上报QPS 高的时候 Langfuse 服务端会成为瓶颈。可以在CallbackHandler里加sample_rate参数比如线上保留 30% 的 trace线下全量。这个参数 Langfuse SDK 原生支持改起来成本最低。第二WebSocket 消息体控制在合理范围。不要直接把完整的 Langchain 内部事件一股脑推送前端处理不过来也没有必要。我推荐的方案是服务端做一层精简只推送前端需要的字段消息 ID、角色、内容增量、token 数、耗时。第三前端图表别复杂化。一开始我试图在页面上画完整调用链的树状图后面发现堆叠太深反而看不清关键信息。后来改成“列表 下钻”的模式默认显示 trace 汇总卡片点击某条 trace 再展示完整的 span 树。这个交互对监控场景更实用。写在最后一点真实的实践体会整套系统跑通之后最大的体会是模型能力和基础设施往往不是瓶颈真正决定一个 LLM 应用能否长期维护下去的是可观测性做得好不好。刚开始做的时候我建议不要一上来就追求复杂的 Langgraph 编排、多级 span 嵌套先把“一次对话 - Langfuse 一条 trace - 仪表盘一张卡片”这条主线跑通数据积累起来之后你自然会知道下一步该在哪些环节加深埋点。另外一个体会是Langfuse 的 metadata 字段一定要在设计阶段就规划好。后续做用户行为分析、bad case 归因、prompt 版本对比全靠这些字段做筛选。如果埋点不规范数据积累越多越难用回填成本非常高。这个小项目后续还能扩展的方向不少接一个 Redis 做会话级缓存、接入更多的模型供应商统一对比成本与效果、把 Langfuse 的 trace 数据通过定时任务同步到数仓做更细粒度的分析。但底子就是这个监控架构先把地基打稳往上面加功能会很顺手。如果你正在做类似的方向希望这套实践能帮你少踩几个坑。