LLM应用可观测性实战：Langfuse+Langchain+DeepSeek监控系统

如果你也在用 DeepSeek 这类大模型做线上服务，一定遇到过类似的场景：用户跑来问“为什么今天的回答质量变差了”，你打开日志一看，HTTP 200、响应正常、耗时也正常——可问题就是复现不出来。我上个月做 AI 客服时就卡在这个坑里，最后被逼着把对话监控这件事从零搭了一遍，整套方案就是题目里这套组合：Langfuse + Langchain + DeepSeek + FastAPI + WebSocket。

这套东西做完之后，我才算真正“看见”了每次对话的全过程：prompt 长什么样、模型返回了什么、Token 烧了多少、单个请求延迟多久、上下文窗口用了多少。出现异常时不用再靠猜，打开仪表盘就能定位是哪一段链路出了问题。这篇记录会从架构选型、后端实现、Langfuse 埋点、WebSocket 实时推送、Vue3 前端展示，一直写到上线后的踩坑复盘，代码都是可以直接复制运行的最小工程，适合正在用大模型做项目、想给服务加一层可观测性的读者。

1. 起因：AI客服上线第二周，我连“回答变差”的原因都查不到

事情是这样的。我们做的是一个面向 C 端的智能客服，底层对话能力用的 DeepSeek。上线第一周一切正常，第二周开始用户投诉变多，集中在“重复回答”“答非所问”这两类。我去翻后端日志，除了状态码和耗时，什么都看不到——没有完整的输入输出、没有提示词版本、没有 token 消耗统计，更没有一个统一的视角去看一次会话里到底发生了什么。

当时最崩溃的一点是：你根本不知道问题是出在模型本身、提示词、上下文拼接，还是用户输入太模糊。因为没有任何基线数据，你甚至没法验证“昨天和今天模型表现是否有差异”。后来我去社区逛了一圈，发现这不是我们一家的问题，做 LLM 应用的团队普遍存在这个盲区。传统的 APM 工具只能看到容器和接口层面的健康度，看不到 prompt 和模型输出这一层。

于是决定给整个对话链路加一套可观测系统。我调研了 Langfuse、LangSmith、Helicone 这几个方案，Langfuse 是开源的，可以自部署，能把 Trace、Span、Token 用量、延迟、成本全部记录下来，还提供了 Python SDK 的CallbackHandler，能和 Langchain 无缝对接。这就是整篇文章最核心的监控底座。

实际方案里还缺一个面向运营和开发团队的实时展示层。Langfuse 自带的 UI 偏调试和回溯，不适合直接丢给产品同事看。所以我用 FastAPI 做后端聚合服务，Vue3 写了一个精简仪表盘，WebSocket 负责把每次对话的关键指标实时推到前端。整体下来，这套组合的定位就是：Langfuse 负责记录一切，FastAPI 负责把记录变成接口，WebSocket 把新事件推到仪表盘，Langchain 负责串联 DeepSeek 调用。

如果你只是自己调试模型，Langfuse 单独就够用。但如果你要的是“团队可看、实时刷新、能定位线上问题”的监控面板，就值得照着这套架构自己搭一版。

2. 架构拆解：Langfuse记录、Langchain编排、FastAPI与WebSocket搬运

2.1 五个组件各自管什么

先看整体链路，我把它分成三层。

用户聊天页面发起提问，请求到 FastAPI 的/api/chat接口。FastAPI 调用 Langchain 构建的链，链的内部是提示词模板、模型调用的组合，最终请求打到 DeepSeek API。这个调用过程通过 Langfuse 的CallbackHandler自动上报，记录成一次 Trace。拿到模型返回后，FastAPI 把回答返回给聊天页面，同时把这次的 token 用量、延迟、模型名等指标打包成一条监控事件，通过 WebSocket 广播给所有在线的仪表盘客户端。

用一张表看各组件职责更清楚：

这套设计最关键的一点是：对话链路和监控链路是解耦的。对话链路走 HTTP，监控链路走 WebSocket，两者在 FastAPI 内部交汇。即使 WebSocket 断开了，聊天服务也不会挂；即使 Langfuse 暂时不可用，模型调用照样能完成。Langfuse 的回调失败默认不会阻断主流程，这很重要。

2.2 为什么选 WebSocket 而不是 HTTP 轮询

最初我也考虑过最省事的方案：前端每 5 秒调一次/api/metrics拉取最新数据。但实际做下来发现问题很多。

LLM 调用本身是不定时的，用户可能连续问 5 句，也可能 10 分钟不开口。轮询模式下，没有新事件时每次请求都在空转，白白耗费服务端资源；有新事件时又最多延迟 5 秒展示，监控面板的“实时性”就打了折扣。更麻烦的是，轮询接口要维护“上次拉取到哪条消息”的游标，多客户端在线时消息会重复或者丢失。

WebSocket 解决的是“服务端主动推送”的问题。FastAPI 原生支持 WebSocket，不需要额外装库，连接建立后双向通信，聊天事件产生的一瞬间就能推到前端。对监控仪表盘这种场景，它就是比轮询更合适的传输层。

2.3 关于 LangGraph：这轮先不上

选型时团队里有人提议直接用 LangGraph 代替 Langchain，因为 LangGraph 支持复杂的状态流转，未来做 Agent 更方便。我的判断是：这个监控仪表盘项目的核心诉求是“在现有 LLM 调用链路里插入可观测性”，不是构建 Agent 状态机。Langchain 用prompt | llm这种 LCEL 表达式就能把链路组织好，代码量小，回调机制也是现成的。

LangGraph 更擅长的场景是有多步骤、条件分支、循环、跨多轮保持状态的编排。如果后面要上“规划-调用工具-再规划”这种 Agent 架构，再把 LangGraph 引进来不迟。监控层不必提前为业务层的复杂度买单。

3. 后端落地：从空目录到 Langchain + DeepSeek 跑通对话

3.1 初始化环境和依赖

假设你是在一台干净的 Linux 服务器或者自己电脑上操作。我用的是uv这个 Python 包管理器，比 pip 快很多，而且创建虚拟环境非常省事。如果你没装过，先装它：

curl -LsSf https://astral.sh/uv/install.sh | sh

然后建项目目录和环境：

mkdir ai-dialog-monitor && cd ai-dialog-monitor uv venv .venv source .venv/bin/activate

依赖清单如下，我直接全部装好：

uv pip install fastapi uvicorn[standard] langchain langchain-openai langfuse python-dotenv

这里有几个版本上的体会。langchain-openai是独立包，ChatOpenAI从这里面导入，不再用老版本的langchain.llms.OpenAI；langfuse的 SDK 版本我用的是 v3 以上，对应 Langfuse v4 服务端。FastAPI 本身对新版本 Python 支持很好，建议直接用 Python 3.10+，避免在老版本上遇到类型语法兼容问题。

3.2 DeepSeek 接入 Langchain 的正确姿势

DeepSeek 的 API 兼容 OpenAI 格式，所以不需要任何定制封装，直接借助ChatOpenAI把base_url指向 DeepSeek 的地址即可。这就是为什么市面上各种工具能轻松接入 DeepSeek——它遵循的是标准协议。

# llm.py import os from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1", temperature=0.7, )

注意base_url写完整路径，带不带/v1我都试过，带上的兼容性最稳。Langchain 底层走的是 OpenAI SDK，有时候版本差异会导致补路径的行为不一致，干脆直接写全。

然后是构建链。为了能拿到 token 用量和响应元数据，我这里没有接StrOutputParser，而是让链直接返回AIMessage：

from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import os llm = ChatOpenAI( model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat"), api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1", temperature=0.7, ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个友好的在线客服助手，请用简洁的中文回答用户问题。"), ("human", "{question}"), ]) chain = prompt | llm

调用时这样拿结果：

result = await chain.ainvoke( {"question": "你们的退货政策是什么？"}, config={"callbacks": [langfuse_handler]} ) answer = result.content metadata = result.response_metadata

result.response_metadata里就有 token 用量和模型名，Langchain 对 OpenAI 兼容接口会自动填充：

{ "token_usage": {"prompt_tokens": 120, "completion_tokens": 46, "total_tokens": 166}, "model_name": "deepseek-chat", "finish_reason": "stop", }

3.3 后端的三个接口设计

这个后端服务只做三件事：接收聊天请求、广播监控事件、对外提供 Trace 查询。三个接口分别对应这三种职责。

# main.py import uuid, time from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel app = FastAPI(title="AI Dialog Monitor") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) class ChatRequest(BaseModel): session_id: str | None = None question: str @app.post("/api/chat") async def chat(req: ChatRequest): session_id = req.session_id or str(uuid.uuid4()) start = time.time() result = await chain.ainvoke( {"question": req.question}, config={"callbacks": [langfuse_handler]} ) meta = result.response_metadata usage = meta.get("token_usage", {}) event = { "type": "chat.completed", "session_id": session_id, "question": req.question, "answer": result.content, "model": meta.get("model_name", "deepseek-chat"), "prompt_tokens": usage.get("prompt_tokens", 0), "completion_tokens": usage.get("completion_tokens", 0), "total_tokens": usage.get("total_tokens", 0), "latency_ms": int((time.time() - start) * 1000), "timestamp": int(time.time() * 1000), } await manager.broadcast(event) return {"session_id": session_id, "answer": result.content}

心跳和 WebSocket 放一起：

@app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: data = await websocket.receive_text() if data == "ping": await websocket.send_text("pong") except WebSocketDisconnect: manager.disconnect(websocket)

Trace 查询接口我会在第 4 章详细说，这里先按下不表。

3.4 启动服务与热更新

启动命令是：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

--reload开启热更新，修改代码后自动重启服务。如果你在 PyCharm 里用运行按钮启动，注意在 Configuration 的 Parameters 里写上--reload，否则改代码不会生效。这个坑我在第 7 章细讲。

4. Langfuse 接入：回调不是“配置一下”就完事

4.1 Langfuse v4 自部署的环境变量

Langfuse v4 和之前的 v3 在自部署上差别很大。v3 依赖 ClickHouse，资源要求高；v4 改成了 PostgreSQL + S3 对象存储，用 Docker Compose 一套三个容器就能拉起来，对中小团队来说部署成本低了很多。我用的是 MinIO 作为 S3 兼容存储。

关键环境变量如下：

启动后在管理后台创建一个项目，拿到 Public Key 和 Secret Key，这两个 Key 是后端 SDK 初始化用的。

4.2 把 CallbackHandler 挂到 Langchain 链上

Langfuse 官方提供了两个对象，容易搞混。Langfuse类负责主动查询 Trace，CallbackHandler负责在 Langchain 调用链路上做埋点。两个对象用同一组 Key 初始化，但用途不同。

# langfuse_client.py import os from langfuse import Langfuse from langfuse.callbacks import CallbackHandler LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "http://localhost:3000") LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY") LANGFUSE_SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY") langfuse = Langfuse( public_key=LANGFUSE_PUBLIC_KEY, secret_key=LANGFUSE_SECRET_KEY, host=LANGFUSE_HOST, ) langfuse_handler = CallbackHandler( public_key=LANGFUSE_PUBLIC_KEY, secret_key=LANGFUSE_SECRET_KEY, host=LANGFUSE_HOST, )

重点来了。我在 Langchain 链上挂回调时，没有把 handler 放进ChatOpenAI的参数里，而是选择在每次ainvoke的config里传递。原因是：同一个 LLM 实例可能被多条链复用，把回调挂在链调用级别，可以控制“哪一次调用需要被追踪”，粒度更灵活。

await chain.ainvoke( {"question": req.question}, config={"callbacks": [langfuse_handler]} )

第一次跑通之后，去 Langfuse 后台刷新，能看到每次对话生成了一条 Trace，里面自动拆出了 LLM 调用的 Span，输入输出、token 用量、延迟都记录得很清晰。Langchain 的 LCEL 链每执行一个组件，CallbackHandler 就会生成对应的 Span，所以你能看到“提示词模板执行”和“ChatOpenAI 调用”两个节点。

4.3 Trace 详情的拉取与索引延迟

Langfuse 的写入是异步的，刚写入的 Trace 不会立刻能被查询到，经常要等个一两秒。仪表盘点击某条对话想看 Trace 详情时，直接用fetch_trace会偶发查不到。

我最终的做法是加一个重试机制，最多试 5 次，间隔 1 秒：

import asyncio from fastapi import HTTPException @app.get("/api/trace/{trace_id}") async def get_trace(trace_id: str): for _ in range(5): try: trace = langfuse.fetch_trace(trace_id) if trace is not None: return trace.data except Exception: pass await asyncio.sleep(1) raise HTTPException(status_code=404, detail="trace not found")

这里有一个小细节：fetch_trace返回的 Trace 对象结构很深，前端要的是可读性更好的树形结构。我会在后端做一层裁剪，提取每个节点的名称、类型、输入、输出、token 用量和耗时，再抛给前端。这样前端不需要理解 Langfuse 的数据模型。

4.4 除了 LLM 调用，还能记录什么

Langchain 的回调会自动记录 LLM 调用，但一次完整的客服对话往往还有别的环节，比如检索、日志、人工标记。Langfuse 的Langfuse对象也支持手动创建 Generation 或 Span，用来补充记录非 Langchain 环节。

generation = langfuse.generation( name="manual-search", input={"query": "退货政策"}, output={"hits": 3}, metadata={"source": "knowledge_base"}, model="bm25", ) generation.end()

这样就做到了一次会话全链路都能追踪。对排查问题来说，能看到“模型回答前有没有正确命中知识库”往往比看模型输出本身更重要。

但也要提醒一句：不要为了埋点而埋点。监控的核心目标是快速定位异常，埋点字段过多反而会让 Trace 可读性变差。我一般只在关键节点记录输入、输出、耗时和错误信息，业务元数据放进metadata，按需查看。

5. WebSocket 实时通道：连接管理、心跳和反代缺一不可

5.1 ConnectionManager 的写法

FastAPI 的原生 WebSocket 接口很简单，但你要自己管理连接集合。我写了一个简单的 ConnectionManager，支持连接、断开、广播三个操作。

# connection_manager.py 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): dead = [] for conn in self.active_connections: try: await conn.send_json(message) except Exception: dead.append(conn) for conn in dead: self.disconnect(conn) manager = ConnectionManager()

广播时给每个连接套一层 try/except，是为了防止某个客户端异常断开导致广播中断。如果不处理，一个连接断开会让send_json抛RuntimeError，后面所有客户端都收不到消息了。实际生产环境里，前端断网、用户直接关闭标签页这种场景非常频繁，这个保护是必备的。

主路由里用await manager.broadcast(event)就能把监控事件推给所有在线仪表盘。聊天接口和 WebSocket 通道都指向同一个manager单例对象，模块之间循环导入的问题要留意，我建议把manager实例放在单独文件里。

5.2 心跳为什么由客户端发起

WebSocket 连接如果长时间没有数据交互，中间的网络设备（尤其是云厂商的负载均衡器、NAT 网关）会在超时后静默断开连接。对仪表盘来说，用户打开页面挂一个上午不刷新是常态，心跳保活必须有。

我的做法是前端每 30 秒发送一次"ping"，后端收到后回复"pong"。后端代码里 while 循环已经实现了这个逻辑，任何非"ping"的文本消息也可以忽略——或者你后续扩展成前端给后端发指令，比如“订阅某个 session 的监控事件”。

有一个经验之谈：保活尽量让客户端主动发，不要依赖服务端定时推。经过 Nginx 和云负载均衡时，客户端主动发包能有效“刷新”连接状态，服务端主动推很容易被误判为闲置流量。

5.3 Nginx 反代和超时

FastAPI 服务通常不直接暴露公网，前面挂一层 Nginx。WebSocket 和普通 HTTP 在反代配置上有个关键区别：必须显式设置 Upgrade 和 Connection 两个头，否则浏览器端口上的连接会被当普通 HTTP 处理，WebSocket 握手直接失败。

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; }

proxy_read_timeout我一开始没配，用的默认 60 秒，结果前端控制台定期报 1006 异常断开，排查了很久才发现是 Nginx 服务端掐断的。设置为 3600 秒后，长连接一整天不掉线。

5.4 前端断线重连

网络不可能永远稳定。前端必须做自动重连，否则仪表盘开着一个下午，中途断网一次，恢复后数据就再也不刷新了，监控等于废了。

我用指数退避策略重连，避免服务端重启时前端疯狂发请求打满连接数：

let retryCount = 0; const maxRetry = 10; function connectMonitorSocket() { const protocol = location.protocol === "https:" ? "wss" : "ws"; const ws = new WebSocket(`${protocol}://${location.host}/ws`); ws.onopen = () => { retryCount = 0; heartbeatTimer = setInterval(() => { if (ws.readyState === WebSocket.OPEN) ws.send("ping"); }, 30000); }; ws.onmessage = (event) => { if (event.data === "pong") return; const msg = JSON.parse(event.data); handleMonitorEvent(msg); }; ws.onclose = () => { clearInterval(heartbeatTimer); if (retryCount < maxRetry) { const delay = Math.min(1000 * 2 ** retryCount, 30000); retryCount++; setTimeout(connectMonitorSocket, delay); } }; ws.onerror = () => ws.close(); }

6. 仪表盘长什么样：Vue3 实时展示 Token、延迟与对话轨迹

6.1 页面结构

前端用的 Vite + Vue3 + TypeScript，没有引入重型 UI 库，组件全部自己写，包体积小、加载快。页面分成三个区域：

• 顶部一排指标卡：今天对话总数、总 Token 数、平均延迟、估算成本
• 左侧对话列表：每次对话的摘要，包括问题、模型、Token 用量、耗时
• 主区域对话详情：点击列表项后展示 Langfuse Trace 的完整调用链

实时数据流是这样：/api/chat返回后，FastAPI 广播chat.completed事件，前端收到后做三件事——更新指标卡数字、把新对话插入列表头部、如果当前处于“只看未读”模式就提示用户。

6.2 WebSocket 客户端封装

我在websocket.ts里封装了一套带重连、心跳、事件分发的客户端，第 5 章的代码可以塞进这个模块。收到不同类型的事件后，通过回调分发到对应的状态更新函数。

function handleMonitorEvent(msg: any) { if (msg.type === "chat.completed") { chatList.value.unshift({ sessionId: msg.session_id, question: msg.question, answer: msg.answer, model: msg.model, totalTokens: msg.total_tokens, latencyMs: msg.latency_ms, timestamp: msg.timestamp, }); } }

有一个细节要注意：如果用户正在查看某条会话的详情，新事件到来时不要强制把视图切走，只更新数字和列表就够了。抢焦点的设计会让人很烦躁。

6.3 指标卡与成本估算

指标卡的数据不必重新请求接口，直接基于 WebSocket 推送的事件计算即可。前端的全局状态用一个reactive对象维护。

const metrics = reactive({ totalCalls: 0, totalTokens: 0, totalPromptTokens: 0, totalCompletionTokens: 0, avgLatencyMs: 0, estimatedCost: 0, }); function updateMetrics(msg: any) { metrics.totalCalls++; metrics.totalTokens += msg.total_tokens; metrics.totalPromptTokens += msg.prompt_tokens; metrics.totalCompletionTokens += msg.completion_tokens; metrics.avgLatencyMs = Math.round( (metrics.avgLatencyMs * (metrics.totalCalls - 1) + msg.latency_ms) / metrics.totalCalls ); }

成本估算要按模型的定价规则算。DeepSeek deepseek-chat 的计费特点可以总结成两行公式：

const cost = (promptTokens / 1000000) * inputPricePerM + (completionTokens / 1000000) * outputPricePerM;

每次后端推送事件时算一次，前端实时累加。这样运营同学可以看到“今天又烧了多少钱”，比起月底看账单，监控起来要直观太多。

6.4 对话详情：把 Langfuse 的 Trace 拉过来

对话列表里的每一项，点击后向后端请求/api/trace/{session_id}。等等，这里有个坑要说明：Langfuse 的 Trace ID 和我的业务 session_id 不是同一个东西。

Langfuse 会在每次ainvoke时自动生成一个新的 Trace ID，但我要在前端列表里按“一次会话”聚合，所以需要让业务 session_id 出现在 Trace 的元数据里。有两种办法：一是调用ainvoke时传入一个自定义run_id，二是结合 LangChain 的run_name和 Langfuse 的session_id参数。

我用的是第二种。在版本较新的 Langfuse SDK 中，可以直接在回调里传session_id，这样 Langfuse 后台的 Session 视图会自动把同一 session 的多轮对话聚合在一起。初始化CallbackHandler时加一个参数：

langfuse_handler = CallbackHandler( public_key=LANGFUSE_PUBLIC_KEY, secret_key=LANGFUSE_SECRET_KEY, host=LANGFUSE_HOST, session_id=session_id, )

不过前端列表想拿 Trace 详情时，还是得用 Trace ID。所以我让 Langchain 的ainvoke返回值带上本次的 Trace ID。改造一下chat接口：

from langfuse.callbacks import get_trace_id result = await chain.ainvoke(...) trace_id = get_trace_id()

在同一个上下文中调用get_trace_id()，能拿到 Langfuse 刚分配的 Trace ID，存到本地列表里，前端点详情时用这个 ID 请求 Langfuse 的 Trace 树。

7. 实测排查：热更新失效、1006断线、流式中断的全链路复盘

7.1 uvicorn --reload 不生效

现象：在 PyCharm 里直接点运行按钮，启动命令是uvicorn main:app --reload，但改代码后服务不自动重启，非要手动停掉再起。

排查后发现是 PyCharm 运行配置的问题。当你在 Run Configuration 里指定了“Script path”为main.py，并且内容只有uvicorn.run(app, ...)时，--reload参数不会自动带上。对话框中也没有可视化的 reload 开关。

解决办法有三个：一是运行配置里用“Module name”为uvicorn，Parameters 写main:app --reload；二是在代码里直接写uvicorn.run("main:app", reload=True)，注意这里的第一个参数要传字符串，不能传app对象；三是我现在最常用的——不用 PyCharm 的运行按钮，直接用终端启动，简单粗暴不出错。

7.2 WebSocket 1006 断线

1006 是 WebSocket 协议里的异常断开码，表示连接没有经过正常的关闭握手就断了。前端控制台最常见的就是这一段：

[websocket] onclose, code: 1006, reason: , reconnect: true

我把这个坑完整复盘一下，因为它牵扯到三个不同的原因。

第一次遇到是在本地联调，前端页面放着不动，几分钟后 1006。排查到是 Nginx 代理超时，proxy_read_timeout默认 60 秒，空闲连接被掐断。加上 3600 秒配置后解决。

第二次是后端服务重启导致的。本地调试时改了代码，--reload触发服务重启，进程断掉，所有 WebSocket 连接全部异常关闭。这个没法完全避免，只能在前端做自动重连，保证服务恢复后仪表盘自动恢复。

第三次是云服务器负载均衡导致的。云厂商的 SLB 默认空闲超时时间也比较短，需要去控制台调长 WebSocket 会话保活时间，后端心跳包 30 秒间隔基本能覆盖。

7.3 stream disconnected before completion

做流式输出时控制台冒出来一个错误：

stream disconnected before completion: failed to send websocket request: io

一开始我以为是后端 WebSocket 坏了，后来定位到是 DeepSeek 流式接口的正常现象：用户在前端还没等模型回答完就关闭了页面，或者切走了连接，客户端侧的 WebSocket 断开，服务端那边的流式读取还在继续，底层就抛出 io 异常。

这个问题有两个处理方向。一是健康检查时忽略这一类异常，它是业务层面的“用户取消”，不是系统故障；二是在前端组件销毁时主动调用ws.close()并通知后端取消当前 LLM 调用，避免资源白白占用。

后端的处理方式是在chat接口的协程里包一层 try/except，捕获流式中断后记录一条日志并清理引用，不要把整个 FastAPI 进程带崩。

7.4 H5能连，打包App连不上

我们把仪表盘做成了移动端应用，发现一个诡异的问题：网页端 H5 连 WebSocket 一切正常，打包成 App 安装到 Android 手机上就永远处于重连状态。

打开日志发现是明文流量限制。从 Android 9 开始，系统默认禁止应用使用未加密的明文网络流量。网页端用ws://可以连，App 端直接拦截，连接建立不了。

解决办法是把 WebSocket 地址从ws://换成wss://，并给域名配好 HTTPS 证书。如果你只是在本地联调，临时可以在 AndroidManifest 里加android:usesCleartextTraffic="true"，但正式包千万别这么干，安全审计过不去。

7.5 PyCharm 装 FastAPI 失败报错

有读者问过我 PyCharm 里安装 FastAPI 一直报错。大部分情况是 pip 源的问题，国外的默认源在国内网络环境下经常超时。换成清华源：

pip install fastapi -i https://pypi.tuna.tsinghua.edu.cn/simple

还有一种是 Python 解释器版本太旧。FastAPI 新版要求 Python 3.8 以上，有些新语法需要 3.10 才能跑。如果你用的是 PyCharm 自带的旧解释器，建议新建虚拟环境时选 3.10+。

8. 扩展与体会：上下文占用预警、Agent 监控和成本治理

8.1 上下文窗口占用预警

DeepSeek 对话模型有上下文窗口上限，长时间对话很容易触顶。触顶之后模型要么报错，要么开始“失忆”。我在仪表盘里加了一个“上下文占用条”，每一轮对话用prompt_tokens累计估算当前会话的占用比例。

后端把 total_tokens 推给前端后，前端按会话维度累加：

if (!contextUsage[sessionId]) contextUsage[sessionId] = 0; contextUsage[sessionId] += msg.total_tokens;

占用超过模型窗口的 80% 时，列表里出现黄色提示，提醒运营引导用户开启新会话。这个功能上线后，直接减少了“DeepSeek 达到对话长度上限，请开启新对话”这类用户投诉。

8.2 从 LLM 链路扩展到 Agent 追踪

如果你后续要上 Agent，Langfuse 同样能接。LangGraph 的执行节点建议画成一个单独的 Span，每个工具调用也是一个 Span。这样监控面板可以看到 Agent 的完整思考路径：调用了哪个工具、返回了什么、下一步决策是什么。

这里要注意：工具调用的输入输出可能包含敏感数据，传到 Langfuse 之前最好做脱敏。我的做法是在 callback 里加一个metadata过滤器，把手机号、身份证之类的字段打码后再上报。

8.3 成本估算实时化

DeepSeek 的价格是动态调整的，硬编码进前端不可取。我把单价配置放在后端环境变量里，前端通过一个/api/config接口去拉：

{ "deepseek-chat": { "input_price_per_m": 2.0, "output_price_per_m": 8.0 } }

价格单位用人民币“元 / 百万 tokens”。每次价格变动时，后端只改环境变量就能生效，前端不用重新构建发布。成本监控的最大价值不是“算得准”，而是建立“每次对话都在花钱”的意识，倒逼开发团队优化 prompt 长度、控制上下文累加。

8.4 一点个人体会

踩完这些坑之后，我的体会是：大模型应用的可观测性，核心不是“加日志”，而是把一次对话变成可以被回溯、被分析、被量化的事件流。Langfuse 解决了记录的问题，WebSocket 解决了实时性的问题，FastAPI 把两者粘起来，Vue3 让数据变得能看懂。这套架构不大，但每一个环节都有它存在的理由。

最后分享一个使用习惯：所有监控数据都该设置保留期。Langfuse 的 Trace 默认全量保留，时间长了存储成本会涨。我用定时任务把超过 30 天的 Trace 异步清理，只保留指标聚合结果。监控系统本身也需要运维，控制成本要从第一天开始想。