从零构建 AI 对话监控仪表盘Langfuse Langchain DeepSeek FastAPI WebSocket 实时架构实践我接手第一个 AI 对话项目时被用户问得最多的一句话不是“模型效果怎么样”而是“刚才那次回答为什么这么慢”。当时项目已经上线我只知道 Langchain 调 DeepSeek 的整条链路大概跑得通但用户报告的问题我根本没法定位——是模型推理慢是 Prompt 构造异常是接口超时还是网络波动我手上连一次请求的完整轨迹都拿不出来。后来我抽了整整一个周末基于 Langfuse Langchain DeepSeek FastAPI WebSocket 做了一套 AI 对话监控仪表盘。这套东西的核心价值不是“多一个后台页面”而是_把一次聊天的完整过程拆成了可观测的指标、日志和链路数据_从用户点击发送到 DeepSeek 返回内容每一步的耗时、Token 消耗、Prompt 内容、模型输出全部有据可查并且通过 WebSocket 实时推到前端页面不用刷新就能看到线上正在发生的每一轮对话。这篇文章适合两类人一类是正在用 Langchain 或者其他编排框架做大模型应用、但对线上运行状态完全没有掌控感的开发者另一类是想了解 Langfuse 到底怎么接入、WebSocket 在这种场景下怎么用的人。我会把选型逻辑、架构拆分、核心代码、以及我实际踩过的坑全部写出来尽量做到你照着能复现。1. 为什么需要给 AI 对话应用装上“行车记录仪”1.1 上线当天遇到的三个真实问题先说我在没有监控体系时遇到的三个具体问题这些问题属于“不踩不知道踩了才明白什么叫痛”。第一个问题是定位慢。用户说“AI 回答得不对”我连用户发的那条消息长什么样都没法快速调出来。日志里只有 FastAPI 的访问记录根本看不到 Langchain 内部到底执行了什么步骤、是调用了工具还是直接走了模型推理。第二个问题是成本失控。DeepSeek 按 Token 收费但 Token 消耗是一个随时间累积的隐藏成本。没有监控的时候我根本不知道哪个 Prompt 模板最“烧钱”也不知道哪些用户或者哪些会话在持续产生巨额调用。直到月底账单出来才傻眼。第三个问题是多会话上下文状态不透明。业务里有一部分逻辑依赖多轮对话的历史消息但历史消息到底有没有正确传进 Langchain 的 Message 列表只能靠肉眼猜。一旦上下文拼错用户得到的就是毫无关联的回答你还无从追溯。1.2 可观测性要观测什么从指标、日志、Trace 说起传统的后端可观测性一般分三个维度Metrics 指标、Logs 日志、Traces 链路。到了 LLM 应用里这三个维度的内涵会有所扩展。Metrics 指标层面我们最关心的不是普通的 QPS而是每次对话请求的端到端延迟、模型推理延迟、Input Token 数量、Output Token 数量、单次调用成本、错误率、上下文长度。这些指标可以直接反映模型服务的健康度和成本趋势。Logs 日志层面除了应用日志还必须记录Prompt 原始内容、模型输出内容、使用的模型名称、Temperature 这类采样参数、命中的 Prompt 模板版本。因为 LLM 的输出具有不确定性同样的输入不一定得到同样的输出没有日志就无法复盘“为什么它当时这么说”。Traces 链路层面LLM 应用通常不是“一次请求 一次模型调用”它可能是“一次请求 多次模型调用 多次工具调用 多条链路分支”。比如 Langchain 里的 Agent 有可能先调用搜索工具再根据搜索结果调一次模型做总结。这种情况下必须把一次用户请求内部的所有子调用串成一个 Trace才能看到时间都花在哪个环节。Langfuse 在这个体系里做的事情就是把上述三项统一收进一个平台。它原生的 Trace 模型和 LLM 应用非常匹配接入成本也比自己拿数据库和时序组件拼一套低得多。1.3 现有 LLM 应用监控方案的横向对比市面上常见的方案无非是以下几种完全自研、Langfuse、LangSmith、以及一些商业 APM 工具的 LLM 监控模块。方案核心能力接入成本自托管能力适用场景自研监控完全可控灵活度高极高需要自建 Trace 数据模型完全可控团队有专门的可观测性人力LangfuseLLM 专用 Trace / 指标 / 评估低提供 SDK 和 REST API支持 Docker 自托管中小团队、个人项目、对数据合规有要求LangSmith与 Langchain 深度整合低但数据在云端不支持自托管已经完全依赖 Langchain 生态的团队通用 APM 扩展兼顾传统服务和 LLM中需要额外配置视产品而定已有成熟 APM 体系的大团队我最后选择 Langfuse一个核心原因是自托管自由度和数据隐私。对话内容涉及业务数据我不想把 Prompt 和用户消息全部发到第三方闭源平台。Langfuse 用 Docker 就能本地部署数据完全在自己手里接入方式又兼容 OpenTelemetry后面就算业务场景切换也有退路。另一个原因是它有开源版对个人项目和小团队的成本压力几乎为零。2. 技术选型五个组件分别解决哪一环2.1 Langfuse 在监控体系里的定位Langfuse 可以理解成一个“LLM 应用专用可观测性后端”。它提供了两个关键入口一个是 SDK 客户端可以直接在 Python 代码里初始化一个 handler 传给 Langchain 的 callbacks另一个是 REST API可以手动上报事件。这两个入口解决的是不同场景的采集需求——前者适合标准 Langchain 链路后者适合自己拼 prompt 再裸调模型的情况。Langfuse 对 Langchain 的支持做得相当好。因为 Langchain 的 Runnable 对象天然支持 callbacks 参数只要把 Langfuse 的 callback handler 传进去一次调用的输入、输出、耗时、Token 用量都会被自动记录。这意味着你几乎不用在业务代码里埋太多的点只需要在关键 Runnable 链路上把 handler 透传进去。2.2 Langchain 是否必要编排层的取舍在选型 Langchain 之前我也犹豫过就调一个 DeepSeek 对话接口直接写 requests.post 不就行了吗为什么非要引一个编排框架后来我发现只要项目涉及“工具调用”“多步推理”“多模型切换”“Prompt 模板复用”这些需求裸写 requests 的代码会快速腐化。以我当前这个项目为例部分对话需要先判断意图再决定是直接回复还是调用检索工具最后再让模型基于检索结果生成回答。这种流程用 Langchain 的 LangGraph 或者 chain 组合来做代码的可读性和扩展性会好很多。当然Langchain 有自己的学习曲线尤其是Langchain 和 LangGraph 的区别经常让人困惑。简单说Langchain 偏重“把多个调用串成链”LangGraph 偏重“有分支、有循环、有条件跳转的图状流程”。如果只是线性的 prompt → model → parser用 Langchain 的 LCELLangChain Expression Language就够了如果要做 ReAct Agent 那种动态决定“下一步调用谁”的逻辑LangGraph 更合适。这个项目我两种都用到了主体对话走 LCEL带工具检索的对话走 LangGraph。2.3 DeepSeek API 接入的基本路径DeepSeek 接入 Langchain 有两种常见方式。第一种是直接用 langchain-openai 里的 ChatOpenAI把 base_url 指到 DeepSeek 的接口地址把 model 换成 deepseek-chat。第二种是用 langchain-deepseek 这个社区包它做了更直接的适配。两种方式本质都是走 OpenAI 兼容协议只是封装程度不同。实际开发中我更推荐第一种。原因有两个一是 langchain-openai 是 Langchain 生态官方维护的包更新及时二是如果你后续要接其他兼容 OpenAI 协议的服务比如各类代理网关只需要改 base_url 就行代码模型类都不用换。2.4 FastAPI 作为应用服务的理由FastAPI 在这个项目里的角色是应用服务入口。它负责接收前端对话请求、调用 Langchain 链路、把结果返回给前端同时把 Langfuse 生成的事件通过 WebSocket 推送给仪表盘。选 FastAPI 而不选 Flask核心原因是异步支持。当前端需要通过流式接口拿模型输出时FastAPI 的 async 机制 SSEServer-Sent Events或 WebSocket 支持都非常顺滑。Flask 做同步接口没问题但要做高并发的长连接或流式响应还得多配一层异步网关折腾。另外一个很重要的点是 FastAPI 的自动文档。调试这种多接口系统时Swagger 页面能直接触发请求和查看 WebSocket 测试入口省了我不少事。2.5 WebSocket 为什么比轮询更适合监控仪表盘仪表盘需要实时展示线上对话如果每隔几秒轮询一次一方面延迟高另一方面请求太密容易增加后端压力。这种场景本质是服务端向客户端持续推送事件WebSocket 的“全双工长连接”天然匹配。有人可能会问用 SSE 行不行SSE 是单向的只支持服务端推给客户端如果仪表盘页面还需要给后端发控制指令比如“停掉某次请求的监控”SSE 就不够方便。WebSocket 双向通信前端还可以主动控制订阅的会话范围整体更灵活。所以我最终选了 WebSocket 作为实时通道。3. 整体架构与数据流向3.1 一次对话请求的完整生命周期先描述一个最简单的请求链路方便后续对代码的理解。用户在前端输入问题浏览器把消息发给 FastAPI 的 /chat 接口。FastAPI 拿到消息后构造 Langchain 的 Runnable 链路把用户消息转换成 HumanMessage然后调用 DeepSeek 模型。这一步发生的同时Langfuse 的 callback handler 会记录该次调用的 Trace、Span、输入输出、Token 消耗。模型返回后FastAPI 把回答返回给前端。前端拿到完整回答后还需要把“这次对话的响应时间、Token 消耗、模型名、Trace ID”这些元信息展示在仪表盘上。因此 FastAPI 在处理完对话后会通过一个 WebSocket 管理器把监控事件广播给所有连接的仪表盘客户端。如果对话请求走的是流式输出那么 Langchain 会边生成边返回 Token前端对话区逐字显示回答等生成结束后FastAPI 再把监控事件推送到仪表盘。这两条路径互不干扰。3.2 核心模块划分整个项目我按功能拆成了这么几个模块模块职责api_serverFastAPI 应用入口路由注册、中间件、WebSocket 端点llm_chainLangchain 链路定义包括 Prompt 模板、模型调用、输出解析langfuse_clientLangfuse 初始化与 handler 封装ws_managerWebSocket 连接管理与事件广播dashboard简单的前端静态页面用原生 JavaScript 接收 WebSocket 消息这样拆的好处是后面如果要换模型、换监控平台、换前端框架都只需要动对应模块不用把整条链路推倒重来。3.3 数据模型监控数据的结构直接复用 Langfuse 的语义一次用户请求是一个 TraceTrace 内部由多个 Span 组成。对于最简单的对话链路一个 Trace 下面只有一个 LLM Span对于带工具调用的链路一个 Trace 下面会挂多个 Span包括 Tool Span 和 LLM Span。我通过 WebSocket 推送的数据结构是自己定义的一个轻量 JSON不直接转发 Langfuse 的原始事件因为原始事件字段太多前端用不上还会增加带宽。{ trace_id: xxxxx, session_id: sess_001, model: deepseek-chat, input_tokens: 128, output_tokens: 256, total_tokens: 384, latency_ms: 1245, timestamp: 2025-01-01T12:00:00Z, prompt: 用户输入的前 200 字, response: 模型输出的前 200 字, status: success }字段不用多关键信息足够就行。真正要查完整 Prompt 和完整输出时再通过 Trace ID 去 Langfuse 控制台里看原始数据。4. 从零搭建后端FastAPI Langchain DeepSeek4.1 环境准备我建议用 uv 来管理 Python 环境和依赖比 pip venv 组合省心很多尤其在 FastAPI 这类项目里依赖解析速度优势很明显。uv init llm-monitor cd llm-monitor uv add fastapi uvicorn[standard] langchain langchain-openai langchain-core langfuse websockets python-dotenv如果你不想用 uv也可以用 pip 创建虚拟环境之后安装同等依赖。需要注意一个坑安装 FastAPI 时有些环境下 uvicorn 的 extras 装不全会导致 WebSocket 协议支持出问题。安全起见直接用uvicorn[standard]它会把websockets库一并带上。安装完成后在项目根目录建一个.env文件写入 DeepSeek 和 Langfuse 的配置DEEPSEEK_API_KEYsk-xxxxxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com LANGFUSE_SECRET_KEYsk-lf-xxxx LANGFUSE_PUBLIC_KEYpk-lf-xxxx LANGFUSE_HOSThttp://localhost:30004.2 定义对话接口FastAPI 的入口文件我命名为api_server.py。先声明一个 FastAPI 实例再注册一个普通的 POST 接口/chat。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleLLM Monitor Demo) class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): trace_id: str answer: str latency_ms: int app.post(/chat) async def chat(req: ChatRequest): # 核心逻辑在 llm_chain 模块 trace_id, answer, latency await run_chain(req.session_id, req.message) return ChatResponse(trace_idtrace_id, answeranswer, latency_mslatency)这里有一个容易被忽略的细节Pydantic 的 BaseModel 在 FastAPI 里承担了请求参数的校验所以前端传参不规范时FastAPI 会直接返回 422而不是进入业务逻辑。4.3 Langchain 接入 DeepSeek核心链路放在llm_chain.py。Langchain 接入 DeepSeek 的关键就是 base_url 替换。具体看代码from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableConfig from langfuse.callback import CallbackHandler llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), temperature0.7, ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助手请用简洁、准确的语言回答问题。), (human, {input}), ]) chain prompt | llm | StrOutputParser()再构建一个 Langfuse handler并把这个 handler 传入RunnableConfiglangfuse_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) async def run_chain(session_id: str, user_message: str): config RunnableConfig( callbacks[langfuse_handler], metadata{session_id: session_id} ) response await chain.ainvoke( {input: user_message}, configconfig ) return response到这里一次简单的对话就通了。每次调用链路的输入、输出和 Token 消耗都会被 Langfuse 记录。需要注意的是Langfuse 的CallbackHandler是同步方式创建的在异步链路里也能传给 Langchain 使用Langchain 内部会适配。4.4 流式输出实现如果希望用户体验到“打字机效果”就不能用ainvoke要改用astream。FastAPI 端用StreamingResponse返回。from fastapi.responses import StreamingResponse import json app.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in chain.astream( {input: req.message}, configRunnableConfig(callbacks[langfuse_handler]) ): yield fdata: {json.dumps({delta: chunk}, ensure_asciiFalse)}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream )前端通过 EventSource 或者 fetch ReadableStream 来消费这个 SSE 流。不过要注意流式输出时 Langfuse 的 token 统计不是一次到位的它会按照生成过程分批上报最终在 Langfuse 控制台里会合并成一次完整的 Span。5. Langfuse 集成让每一次调用都有据可查5.1 安装与初始化Langfuse 服务端的部署方式官方推荐 Docker Compose。这是我本地docker-compose.yml的关键配置version: 3.9 services: langfuse: image: langfuse/langfuse:latest ports: - 3000:3000 environment: DATABASE_URL: postgresql://postgres:postgresdb:5432/langfuse NEXTAUTH_URL: http://localhost:3000 NEXTAUTH_SECRET: mysecret SALT: mysalt ENCRYPTION_KEY: myencryptionkey depends_on: - db db: image: postgres:15 environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: langfuse volumes: - postgres_data:/var/lib/postgresql/data volumes: postgres_data:启动后访问http://localhost:3000注册时创建一个项目拿到项目的 Public Key、Secret Key 和 Host 地址即可。提示Langfuse v4 之后接口地址和密钥获取入口略有变化但环境变量名没有变化。如果用的是旧版本教程里的/api/public路径要注意比对版本。5.2 给 Langchain 挂上 Langfuse Handler前面在llm_chain.py里已经用到了CallbackHandler。给 Langchain 挂 handler 有两种方式一种是在每次调用时通过RunnableConfig传入另一种是在创建 model 实例或者 chain 时直接在构造函数里声明。我推荐前一种因为可以在 metadata 里动态带上 session_id、user_id 等业务字段方便在 Langfuse 里做筛选。Langfuse 官方文档里还提到可以用observe()装饰器手动埋点。这个方式适合那些不走 Langchain 编排的裸模型调用。但我建议能用 Langchain 统一管理就用 Langchain埋点代码越少业务侵入越小。5.3 Traces / Spans / Observations 的概念与指标解读Langfuse 的数据模型有三个核心概念Trace一次完整的用户请求所有内部执行都归属于某一个 Trace。SpanTrace 内部的子步骤。每次 LLM 调用、每次工具调用都应该是一个 Span。Observation最底层的事件节点可以理解为一个具体的耗时记录或日志片段。实际使用中Langchain 的CallbackHandler会自动帮你把这些关系建好。你不需要手动去创建 Span只需要把 trace 的名字和 metadata 指定清楚。这样在 Langfuse 控制台里你可以按 Trace 看整体耗时也可以点进某个 Span 看具体某一轮 Prompt 和模型输出。关于指标解读我有几个从实战中总结的直接经验指标健康范围参考异常说明TTFT首 Token 时间300ms ~ 1500ms 取决于模型负载如果超过 3s大概率是网络或模型服务问题单次调用总延迟1s ~ 10s过长需要检查 Prompt 是否太复杂Input Token 占比视业务而定过高说明上下文塞了太多历史消息Output Token 波动同上突增可能说明模型进入了重复生成循环这些指标没有绝对标准但有了数据之后至少你能在一个时间周期内看到趋势变化而不是凭感觉做判断。6. WebSocket 实时推送仪表盘的“直播通道”6.1 WebSocket 连接管理与生命周期WebSocket 端点的职责是维护一个活动连接集合。当 FastAPI 的/chat接口处理完一次请求后就把监控事件广播给所有订阅的仪表盘客户端。我在ws_manager.py里封装了一个简单的连接管理器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): 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()这里有一个关键的细节广播时不能直接 for 所有 connection 然后同步 send。如果某个客户端已经断开send 会抛异常。所以我在 send_json 外面包了 try/except并在异常时调用 disconnect 清理连接。否则时间一长活跃连接列表里会堆一堆死连接。FastAPI 端的 WebSocket 路由from fastapi import WebSocket, WebSocketDisconnect app.websocket(/ws/monitor) async def websocket_monitor(websocket: WebSocket): await manager.connect(websocket) try: while True: # 保持连接等待客户端消息 data await websocket.receive_text() # 可以在这里做订阅控制暂时忽略内容 except WebSocketDisconnect: manager.disconnect(websocket)6.2 将 Langfuse 回调转发到仪表盘这里要解决一个问题Langfuse 的 callback 是在 Langchain 内部触发的我如何在对话结束后拿到 Trace ID 和耗时数据然后广播出去我的做法是在业务函数里手动记录开始时间调用链结束后计算耗时再把关键指标拼成 JSON 广播。Trace ID 需要从 Langfuse 的 callback 里取出来可以通过在run_chain里定义 callback 的on_llm_end事件来捕获。简化实现如下from langfuse.callback import CallbackHandler from langfuse.types import LangfuseTrace class MonitorCallbackHandler(CallbackHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.current_trace_id None def on_chain_start(self, serialized, inputs, **kwargs): super().on_chain_start(serialized, inputs, **kwargs) trace self.get_trace() if trace is not None: self.current_trace_id trace.id然后在run_chain里对话完成后构造指标信息async def run_chain(session_id, user_message): start time.time() langfuse_handler MonitorCallbackHandler(...) response await chain.ainvoke({input: user_message}, config...) latency_ms int((time.time() - start) * 1000) event { trace_id: langfuse_handler.current_trace_id, session_id: session_id, prompt: user_message[:200], response: str(response)[:200], latency_ms: latency_ms, timestamp: datetime.utcnow().isoformat() } await manager.broadcast(event) return response这样仪表盘就能在对话结束后实时看到这条记录。想要看完整的 Token 消耗可以从 Langfuse 的 Trace 里再拉一次数据但为了降低延迟仪表盘上先用 latency 和文本预览做主展示即可。6.3 前端接收与展示前端我为了减少依赖没有上框架直接用原生 HTML JavaScript。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleAI 对话监控仪表盘/title /head body div idevents/div script const ws new WebSocket(ws://localhost:8000/ws/monitor); ws.onmessage function (event) { const data JSON.parse(event.data); const div document.createElement(div); div.innerHTML strong${data.session_id}/strong | ${data.model} | ${data.latency_ms}ms | ${data.timestamp}; document.getElementById(events).prepend(div); }; ws.onclose function (e) { // 重点WebSocket 1006 异常关闭后要有重连策略 console.log(连接关闭准备重连, e.code); setTimeout(() { location.reload(); }, 3000); }; /script /body /html这里有个我踩过的坑WebSocket 1006 异常关闭。如果你用 Nginx 反代 WebSocket需要配置合适的超时时间否则空闲连接会被服务端切断前端会不停重连。FastAPI 和 uvicorn 的标准配置下本地测试通常不会出现 1006但一旦上了反向代理问题就来了。7. 上线前必须处理好的四个问题7.1 流式中断与连接重置1006 问题排查过程我第一次做联调时仪表盘连接经常在空闲一会儿后自动断开浏览器控制台里出现[websocket] onclose, code: 1006。排查链路大概是这样的第一步我先确认是前端还是后端断开。我在 FastAPI 的disconnect里打日志发现服务端没收到断开事件说明是服务端或中间层主动切断了连接。第二步检查本地直连关闭 uvicorn 之外的代理层发现本地直连稳定。问题锁定在 Nginx 层。第三步定位到 Nginx 默认的proxy_read_timeout是 60s。WebSocket 长连接超过 60s 没有消息Nginx 就会掐断连接。解决办法是给 WebSocket 的 location 单独加上location /ws/ { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }这类问题表面上看着是 1006 代码本质上是代理层的超时配置问题。以后只要遇到 1006建议先排查是不是有中间代理设置了一分钟级别的超时。7.2 Langfuse 异步上报与业务延迟的隔离Langfuse 的 callback handler 上报事件时如果走同步 HTTP POST会拖慢业务接口的响应。我在压测时发现并发请求一上来/chat接口的延迟波动很厉害。解决办法有两个配置 Langfuse 客户端的批处理模式让它后台异步批量上报。确保业务侧的关键延迟指标只统计模型推理本身不要把 Langfuse 上报时间算进去。Langfuse SDK 默认是异步批量上传但在某些自定义 handler 写法下容易不小心改成同步。如果你发现每次请求都会多出几百毫秒的固定耗时检查一下是否在 Langchain 外手动调用了langfuse.flush()之类的同步方法。7.3 FastAPI 热更新不生效的真相开发期间我遇到过 FastAPI 改代码后--reload不生效的情况尤其是新增了websocket路由之后。后来发现不是热更新失效而是 uvicorn 监控的目录不对。用uvicorn api_server:app --reload时默认监控当前目录但如果你的代码在 src 目录下需要显式指定--reload-dir src。另外一个细节是WebSocket 连接会阻止 uvicorn 正常退出。开发时如果开着 WebSocket 页面CtrlC 杀掉服务后端口可能还被占用。这时候需要手动清理进程否则Address already in use会让人一头雾水。7.4 Langchain 与 LangGraph 的编排边界这个项目里有一部分会话需要通过工具调用增强回答能力我用了 LangGraph 来构建。最初我用 Langchain 的 Agent 概念把工具列表塞给模型让它自己决定调用哪个工具。结果发现一旦工具数量变多模型经常选错工具或者在没有必要的时候也调用工具。后来我改成 LangGraph 写了一个带条件分支的流程第一步用一个小模型做意图分类决定是否需要工具。如果需要进入工具调用节点拿到工具结果。最后再进入主模型生成回答。这样编排的好处是每个阶段的输入输出都会成为 Langfuse 里的独立 Span监控数据比一个黑盒 Agent 清晰得多。如果你在 Langchain 里开发 Agent 类功能建议优先考虑 LangGraph 的显式图结构而不是把所有逻辑塞给 ReAct 循环。8. 项目跑起来之后如何扩展到生产级8.1 多会话订阅隔离当前 WebSocket 广播是所有客户端共享的也就是每个打开的仪表盘都会看到所有会话的监控数据。如果团队里有多个业务线都接入同一个监控后端那就需要引入订阅隔离机制。做法比较简单WebSocket 连接建立后客户端先发送一个 JSON 订阅消息比如{type: subscribe, session_id: sess_001}服务端维护一个“连接 - 订阅条件”的映射广播时根据 event 里的 session_id 做过滤。这个改造不到一百行代码但能让系统从“演示 demo”变成“可多人协同使用的内部平台”。8.2 用 Langfuse 评估面板做回归分析Langfuse 除了 Trace 和指标还有一个比较容易被忽略的能力数据集与评估。你可以把一批固定的测试问题定期跑一遍链路记录每次回答的质量评分。这样模型升级、Prompt 修改之后不用依赖人工逐个验证直接在面板上对比历史实验。我在项目里的用法是把常用的业务问答场景整理成一个 CSV定时脚本通过 Langchain 调用 DeepSeek得到结果后把评分打到 Langfuse 的数据集里。这个方法对 Prompt 版本的回归测试非常有效。8.3 Token 成本预测告警当监控数据积累到一定量之后可以做成本预测。DeepSeek 的定价虽然便宜但调用量放大后成本同样是不可忽视的指标。Langfuse 的对外 API 可以按时间范围拉取 Token 用量我写了一个定时任务每小时拉一次对比昨日同时段的用量如果增长率超过阈值就发告警到群里。还可以进一步拆分哪个 Prompt 模板消耗最多、哪个用户会话消耗最多、哪个模型版本成本最高。这些分析出来后可以直接反哺业务决策例如调整免费用户的调用频率或者给高消耗的 Prompt 模板写缓存层。整个项目从零跑通核心代码量不大但每一步选型背后的逻辑才是真正值得沉淀的东西。如果你也在做大模型应用建议别等系统出问题再去补监控先花一天把这套最小闭环搭出来。等到某天用户抱怨回答质量差你能在五分钟内定位到是 Prompt 问题还是模型问题那种掌控感绝对值得。