先聊一个我最近被问得最多的问题：大模型对话应用做完 Demo 之后，怎么知道它在线上到底跑得好不好？模型有没有在乱答，响应慢是因为网络还是模型本身，一次会话到底烧了多少 Token，用户卡在哪个环节放弃的。这些问题如果你靠事后翻日志去猜，基本等于瞎忙。所以我干脆把一个完整的实时监控链路搭了出来，用 Langfuse 做全链路追踪，LangChain 编排 Agent，DeepSeek 做推理，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，注意不要加/v1，DeepSeek 官方说他们的兼容端点不需要带 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( model="deepseek-chat", api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL, temperature=0.7, streaming=True, ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个智能助手，可以调用工具来获取实时信息。"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_openai_tools_agent(llm, [get_current_time], prompt) agent_executor = AgentExecutor( agent=agent, tools=[get_current_time], verbose=True, return_intermediate_steps=True, )几个值得注意的细节：必须传agent_scratchpad占位符，否则 Agent 无法记录中间步骤；return_intermediate_steps=True能让我们拿到工具调用的中间结果，这个后面在 Langfuse 里追踪很有用；streaming=True是为了实现 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_key=LANGFUSE_PUBLIC_KEY, secret_key=LANGFUSE_SECRET_KEY, host=LANGFUSE_HOST, session_id=session_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 3，WebSocket 连接建议放在一个独立的 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( model="deepseek-chat", api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL, request_timeout=120, )另外一个被人忽略的因素是本地代理环境变量，如果你机器上设置了HTTP_PROXY或HTTPS_PROXY，OpenAI 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 时会断开，这属于正常现象，前端断线重连逻辑会自动恢复，不需要额外处理。
