先聊一个我最近被问得最多的问题大模型对话应用做完 Demo 之后怎么知道它在线上到底跑得好不好模型有没有在乱答响应慢是因为网络还是模型本身一次会话到底烧了多少 Token用户卡在哪个环节放弃的。这些问题如果你靠事后翻日志去猜基本等于瞎忙。所以我干脆把一个完整的实时监控链路搭了出来用 Langfuse 做全链路追踪LangChain 编排 AgentDeepSeek 做推理FastAPI 提供服务层WebSocket 把监控数据实时推到浏览器仪表盘上。这篇文章把整个从零到一的实现过程、踩坑记录和核心代码都整理出来适合已经有 Python 基础、想把自己的 LLM 应用做成可交付系统的开发者。1. 为什么是这五个技术栈系统架构与选型思路1.1 五个组件各司其职把监控从“事后查日志”变成“实时看板”这套系统的核心目标是回答三个问题第一用户和 AI 的每一轮对话发生了什么第二每轮对话花了多少钱、多少时间、多少 Token第三当对话出问题时能不能在秒级内定位到是应用层、模型层还是网络层的问题。要回答这三个问题单靠一个框架做不到所以我把系统拆成了五个角色。DeepSeek 负责推理它是我选的模型提供商性价比高而且对中文场景友好LangChain 负责编排把模型调用、工具调用和提示词模板统一管理FastAPI 是后端服务的骨架提供 REST 接口和 WebSocket 端点WebSocket 负责把监控数据实时推到前端Langfuse 则是最关键的一环它把每一次模型调用的输入、输出、Token 消耗、延迟、成本全部记录下来并且自带可视化看板。如果你把整套系统想象成一个外卖平台DeepSeek 是后厨炒菜的LangChain 是帮你按菜单配菜的FastAPI 是前厅接待的WebSocket 是传菜员Langfuse 就是后厨里的摄像头——它能把每一道菜从下单到出餐的全过程录下来出了问题回放一下就知道谁慢了、谁洒了。1.2 选型取舍为什么不自研监控为什么不是消息队列有人会问为什么监控不自己写用一张表记录请求日志不行吗我自己一开始也是这么干的但很快就放弃了。实时监控的难点不在于“记录”而在于“关联”。一次用户对话可能会触发多轮模型调用、多次工具调用甚至会有并行调用。如果只记一条日志你根本看不清一条完整链路上每个环节的耗时和消耗。Langfuse 这种专门做 LLM 可观测性的平台天然把 trace、observation、generation 的关系建好了省掉我大半张表结构设计的工作。另一个问题是数据通路。监控数据要不要走消息队列我的答案是人一多再加。目前的架构下FastAPI 服务直接把监控事件上报给 Langfuse再通过 WebSocket 推送关键指标到前端数据量级在个人项目和中小团队内部工具这个范围内完全够用。消息队列适合的是每天百万级请求、需要异步削峰的场景我把它写在扩展方案里而不是第一版架构里因为过度设计才是这种项目最常见的失败原因。1.3 LangChain 与 LangGraph 的选择边界标题里写的是 LangChain但我必须提一句 LangGraph。LangChain 是个大而全的生态有模型封装、提示词管理、输出解析、记忆、检索、Agent 工具等一堆模块。LangGraph 更聚焦在 Agent 的状态机编排上适合流程复杂、需要条件分支和人类介入的场景。热词里大家都在对比这两个框架我的理解是如果你的 Agent 是“工具调用”类型LangChain 的 AgentExecutor 足够如果你的 Agent 要处理多轮人机协同、并行分支、循环终止这类逻辑尽早切到 LangGraph。我做这套监控系统时用的是 LangChain 的经典 Agent 写法因为业务逻辑不复杂而 LangGraph 留作后续演进方向后面我会贴在扩展部分讲清楚怎么迁移。2. 环境准备与后端服务搭建先把地基夯实2.1 Langfuse 自托管Docker Compose 一条命令拉起Langfuse 有云服务和自托管两种模式我建议你自己部署一套原因有二一是对话数据很可能涉及业务隐私不要轻易发到第三方平台二是自托管版本完全免费你还能顺便观察它的数据库表结构对理解 trace 数据模型有帮助。我用的是 Docker Compose 方式部署。Langfuse v4 版本自带了 PostgreSQL 和 Redis 依赖docker-compose.yml 里只需要定义 langfuse、db、redis 三个服务然后执行docker compose up -d首次启动后访问http://localhost:3000默认管理员账号密码是 admin/admin登录后强制改密。在项目设置里创建一个新项目拿到公钥Public Key和私钥Secret Key这两个 Key 是 SDK 上报数据的凭证后面写配置文件和环境变量要用它。注意如果你用的是旧版本升级到 v4数据库迁移会有一段初始化时间。我第一次升级时因为直接删掉了 volume 导致历史 trace 全丢所以升级前一定要备份pgdata目录。2.2 FastAPI 项目结构与配置管理FastAPI 负责两件事对外提供 HTTP 接口给前端调用对内维护 WebSocket 长连接并广播监控指标。我建议项目目录按照“配置、服务、路由、模型”四层拆分不要把所有代码堆在一个 main.py 里。我常用的结构是app/ ├── main.py # FastAPI 实例、路由注册、启动事件 ├── core/ │ └── config.py # 环境变量读取和统一配置 ├── agents/ │ └── chat_agent.py # LangChain Agent 构建 ├── schemas/ │ └── chat.py # Pydantic 请求响应模型 ├── services/ │ ├── llm_service.py # 调用 Agent 的业务封装 │ └── monitor.py # Langfuse 回调和 WebSocket 广播逻辑 └── routers/ └── chat.py # HTTP 和 WebSocket 路由依赖管理我用的是uv创建虚拟环境这比 pip 快很多热词里也有人问这个我顺手说下uv venv .venv source .venv/bin/activate uv pip install fastapi[standard] uvicorn langchain langchain-openai langfuse openai python-dotenv这一步的选型理由是FastAPI 对 WebSocket 有原生的支持WebSocketEndpoint和app.websocket装饰器都封装得很顺不需要额外引入 aiohttp 之类的东西。而且 FastAPI 的异步特性能够和异步 OpenAI 客户端配合得很好不会因为模型响应慢而阻塞整个事件循环。2.3 DeepSeek 接入 LangChain兼容 OpenAI 协议的真香现场DeepSeek 提供了一个 OpenAI 兼容的 API 接口这意味着 LangChain 里不需要任何自定义封装直接用langchain-openai的ChatOpenAI类改一下base_url和 API Key 就能工作。在config.py里我通常这样组织配置import os from dotenv import load_dotenv load_dotenv() DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) DEEPSEEK_BASE_URL os.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com) DEEPSEEK_MODEL os.getenv(DEEPSEEK_MODEL, deepseek-chat) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY) LANGFUSE_HOST os.getenv(LANGFUSE_HOST, http://localhost:3000)这里面有个细节DeepSeek 的 API Key 和 OpenAI 的 Key 不通用但ChatOpenAI类只管协议格式不校验 Key 属于哪家所以你在填api_key参数时填 DeepSeek 的 Key 就行。base_url一定要写成https://api.deepseek.com注意不要加/v1DeepSeek 官方说他们的兼容端点不需要带 v1 路径。环境变量放在.env文件里所有密钥不进代码仓库这是我最坚持的一条习惯。3. 核心实现Agent 编排、全链路追踪与 WebSocket 实时推送3.1 构建带工具的 Agent让 DeepSeek 不仅能聊天还能查数据纯对话的场景不需要 LangChain直接调 API 就完事了。我在这套系统里引入 LangChain 的原因是为了让模型具备调用工具的能力。我给 Agent 加了一个“获取当前时间”的工具虽然看起来简单但它是后续接业务 API、查数据库、发通知这类能力的起点。工具的定义用tool装饰器from langchain_core.tools import tool from datetime import datetime tool def get_current_time(format: str %Y-%m-%d %H:%M:%S) - str: 获取当前时间用户问时间时使用。format 参数为时间格式默认按年月日时分秒返回。 return datetime.now().strftime(format)然后是 Agent 的构建。我用的是create_openai_tools_agent加AgentExecutor的方式。DeepSeek 模型对 OpenAI 的 function calling 协议支持得不错所以这条路走得通from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI( modeldeepseek-chat, api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, temperature0.7, streamingTrue, ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助手可以调用工具来获取实时信息。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_openai_tools_agent(llm, [get_current_time], prompt) agent_executor AgentExecutor( agentagent, tools[get_current_time], verboseTrue, return_intermediate_stepsTrue, )几个值得注意的细节必须传agent_scratchpad占位符否则 Agent 无法记录中间步骤return_intermediate_stepsTrue能让我们拿到工具调用的中间结果这个后面在 Langfuse 里追踪很有用streamingTrue是为了实现 token 级别的实时输出否则 WebSocket 推送就只能等整段结果返回后才动一下。3.2 Langfuse 回调埋点一次注入全链路可见Langfuse 的集成方式很巧妙LangChain 官方给它做了回调接口你只需要把CallbackHandler实例传给链或者 Agent 的callbacks参数它就会自动捕获模型调用、Token 统计、延迟等数据。我封装了一个 monitor.py 用来统一管理from langfuse.callback import CallbackHandler from app.core.config import LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST def get_langfuse_handler(session_id: str): return CallbackHandler( public_keyLANGFUSE_PUBLIC_KEY, secret_keyLANGFUSE_SECRET_KEY, hostLANGFUSE_HOST, session_idsession_id, )在实际调用 Agent 时把 handler 塞进去langfuse_handler get_langfuse_handler(user_session_id) result await agent_executor.ainvoke( {input: user_message}, config{callbacks: [langfuse_handler]}, )这里有个很关键的体验点Langfuse 里的 trace 是按 session 聚合的。如果你希望一次用户会话的多轮对话在同一段时间轴里展示就要保证每次调用传入相同的session_id。前端的每个会话页面生成一个 UUID一路往下传就行。Langfuse 的 UI 会展示这些指标每次调用的输入输出、耗时、总 Token 数、费用估算需要在项目设置里配置模型单价、工具调用链。你还可以在 trace 列表页按 session 筛选逐条点开看整条链路的耗时分布这个能力比自己写日志收集系统强太多了。3.3 WebSocket 通道从“轮询”到“实时流”监控数据要实时上屏HTTP 轮询虽然实现简单但一秒一问的体验太差了而且对后端压力很大。WebSocket 的优势在于建立一次连接服务端可以持续主动推送数据。我用 FastAPI 写了一个轻量级的广播中心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): for conn in self.active_connections[:]: try: await conn.send_json(message) except Exception: self.disconnect(conn) manager ConnectionManager()广播端点和普通的 HTTP 路由放在同一个 FastAPI 实例里这样整个服务只需要一个端口就能同时提供 REST 和 WebSocket 能力部署省心。我在routers/chat.py里加了 WebSocket 路由from fastapi import APIRouter, WebSocket, WebSocketDisconnect router APIRouter() router.websocket(/ws/metrics) async def websocket_metrics(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(websocket)这里有个细节客户端如果只收不发有些防火墙和代理层会在空闲一段时间后掐断连接。所以我在前端加了一个心跳定时器每 30 秒发一个ping字符串上面的receive_text()就是用来接收这个心搏信号的。服务端也可以跑一个后台任务定时检查连接存活状态超过 90 秒没有收到任何消息的客户端就主动关闭避免僵尸连接堆积。3.4 前端仪表盘先把数据亮出来如果项目刚起步前端我建议先用一个原生 HTML JavaScript 页面把核心指标展示出来就够了。关键是 WebSocket 的连接和断线重连逻辑。我做了一个简化版本const ws new WebSocket(ws://${location.host}/ws/metrics); ws.onmessage (event) { const metric JSON.parse(event.data); updatePanel(metric); // 更新延迟、Token、耗时等面板 }; ws.onclose (e) { if (e.code ! 1000) { setTimeout(() reconnect(), 3000); } }; function reconnect() { const newWs new WebSocket(ws://${location.host}/ws/metrics); newWs.onmessage ws.onmessage; newWs.onclose ws.onclose; }收到的metric数据可以包含这些字段会话 ID、消息内容摘要、模型名称、延迟毫秒数、Token 消耗、成本、工具调用次数。页面上用卡片展示最近一条请求的耗时和 Token 变化趋势即可。如果前端用的是 Vue 3WebSocket 连接建议放在一个独立的 composable 里统一管理避免每个组件各自建连导致连接数爆炸。仪表盘刚开始不需要做得很花哨先把数据链路打通后面想加图表、加筛选器都容易。我见过太多人一上来就想做全功能的监控大屏结果卡在图形库配置上反而忽略了最核心的数据流这是本末倒置。4. 上线后最常踩的坑问题排查与修复实录4.1 Langfuse 面板空白trace 死活不出数据Langfuse 部署好了、调用也传了 callback但面板上就是一条记录都没有这是我最常被问到的问题。排查路径按照顺序来先确认环境变量LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY、LANGFUSE_HOST有没有被正确加载很多情况下是.env文件没被读取再确认你的CallbackHandler是不是放在了config的callbacks参数中而不是把它塞进了prompt或者llm里最后确认 Langfuse 服务本身的网络连通性在服务器上执行curl http://localhost:3000/api/public/health看看是否返回正常。我踩过的一个隐蔽坑是CallbackHandler实例不能重复用在不同 trace 上如果你在一段代码里创建了 handler 并复用到多个请求Langfuse 会把它们关联到同一个 trace 里导致 UI 上看到的是一条大杂烩记录。每个请求必须新建 handler。4.2 连接秒断WebSocket 出现 code 1006 和 io 层报错WebSocket 的 code 1006 表示连接异常关闭没有收到正常的 close frame。这个问题的来源很多我遇到过的有Nginx 代理没开 Upgrade 头、服务端在启动时崩溃导致连接被系统回收、心跳机制没有实现导致空闲连接被中间层断开。如果你用 Nginx 做了反向代理必须在 location 配置里显式加上proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade;否则 WebSocket 的握手请求根本到不了 FastAPI。如果已经加了还是 1006下一个排查重点就是服务端有没有在处理器里正确捕获异常。我在流式推送时遇到过stream disconnected before completion: failed to send websocket request这类错误多半是模型端响应中断后服务端没有及时把连接清理掉客户端还在傻等数据。解决方式是完善WebSocketDisconnect异常处理并在连接发送失败时立刻从活跃列表移除。4.3 流式输出中断模型返回一半连接就断了DeepSeek 在处理长文本时耗时可能超过默认的 HTTP 客户端超时时间。LangChain 的ChatOpenAI底层走的是 OpenAI SDK默认超时可能在 60 秒左右如果你用流式模式就没有这个问题因为 token 会持续到达。但如果关闭了流式或者用的是非流式接口长文本生成就容易触发超时。这种问题的修复方式是显式设置请求超时参数llm ChatOpenAI( modeldeepseek-chat, api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, request_timeout120, )另外一个被人忽略的因素是本地代理环境变量如果你机器上设置了HTTP_PROXY或HTTPS_PROXYOpenAI SDK 会默认走代理代理一断就可能出现 stream 中断。开发环境下建议把这些变量清掉或者用http_client参数自定义一个不走代理的客户端。4.4 上下文超限与“对话达到长度上限请开启新对话”跑了一阵子之后你会发现多轮对话的总 Token 数会慢慢逼近模型的上下文窗口。DeepSeek 的上下文窗口比较大但也不是无限大。当对话太长时API 会报错相当于热词里大家提到的“对话达到长度上限请开启新对话”。这个问题的解法不是在出错后提醒用户而是主动做上下文管理。我在这套系统里做了两步第一步每次请求前估算当前会话的历史 Token 数超过阈值的自动丢弃最早的消息第二步在错误响应里返回一个特殊状态码前端捕获后弹窗引导用户开启新会话。LangChain 的ConversationTokenBufferMemory或者trim_messages都封装了类似逻辑我建议直接用现成的别自己手写截断逻辑因为要同时考虑 system prompt 和工具调用历史的保留很容易搞错。4.5 Langfuse 从旧版本升到 v4初始化与数据迁移Langfuse v4 版本调整了部分数据库结构和初始化流程如果你是从 v3 之类旧版本升级的容器启动后可能会出现初始化失败或者面板 502。我的做法是备份旧数据库卷然后让新版本容器自己执行迁移命令。如果你舍不得 Docker 卷里的历史 trace 数据不要贸然用docker compose down -v清数据先备份pgdata再升。如果只是从零安装 v4那就简单很多照着官方文档的 docker-compose 来就行。表结构不用你手动建容器启动时会自动执行迁移脚本。5. 扩展思路从监控到可观测再到成本治理5.1 把 LangGraph 作为高复杂度 Agent 的演进方向如果你发现 Agent 的流程开始变得复杂比如需要在多个工具之间做条件判断、需要人工审批环节、或者要支持并行调用多个模型那 LangChain 经典的 AgentExecutor 会显得不够灵活。LangGraph 把 Agent 定义成一张状态机图每个节点是一个处理步骤每条边是一个流转条件调试和扩展都更直观。迁移路径不复杂原来的工具函数可以直接复用ChatOpenAI的模型实例也能继续用主要工作是重新组织控制流。LangGraph 也有自己的回调机制Langfuse 同样支持接入所以监控体系不用重做。5.2 把 Langfuse 的数据用起来成本与质量复盘Langfuse 不只是出问题的时候看的。我每周会拉一次统计数据重点看两个指标平均每次请求的 Token 消耗和单次对话的累计成本。如果某段时间平均 Token 突然涨了多半是上下文管理策略失效或者 prompt 模板被改成了啰嗦版本。Langfuse 提供了 Python SDK 查询接口你可以用get_generations之类的 API 把数据拉下来做成周报也可以在面板里设置告警规则。监控系统只有被持续使用才有价值否则就是一套昂贵的摆设。5.3 延迟优化把“模型耗时的透明度”作为默认能力最后分享一个我实际使用中的体会监控仪表盘最有价值的功能不是“出了错才看”而是让每一次请求的耗时都变得透明。用户觉得系统卡的时候你能立刻区分是网络延迟、模型推理还是工具调用慢这比靠猜靠谱得多。我在这个项目里建议你在每一轮 WebSocket 推送的 metric 中都带上stage字段区分llm_start、tool_call、llm_end、full_response。前端根据 stage 渲染不同的时间节点这样用户看到的不再是一个冰冷的“正在输入”而是可以感知到系统当前处于哪个阶段。这种透明感对内部工具的可信度提升非常明显。6. 最后再分享一个我在实际项目中的小技巧如果你准备照着这套架构落地先不要急着把 Langfuse、WebSocket、Agent 全部一次接好。我自己的习惯是分三步走第一步先用 FastAPI 写一个最简单的聊天接口直接调 DeepSeek确认 Key 和网络没问题第二步把 LangChain 接上先不加工具再加一个简单工具跑通 AgentExecutor第三步再接 Langfuse每次调用后去面板里看 trace最后才加 WebSocket 和前端仪表盘。这样每走一步都有可验证的成果排查问题时也只需要在一个层面找原因。我见过不少人在第一步就卡住跑去查 Langfuse 面板为什么没数据结果发现是 DeepSeek API 压根没通。先把地基打好再往上盖楼这个顺序在 AI 应用开发里尤其重要。再补一个代码层面的细节FastAPI 开发调试时记得启动加--reload参数uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload热更新能省掉你大量手动重启的时间。不过注意WebSocket 连接在 reload 时会断开这属于正常现象前端断线重连逻辑会自动恢复不需要额外处理。