FastAPI+WebSocket+React实战：手搓英语情景教学Agent

1. 为什么我要手搓一个英语情景教学Agent去年年底我在帮几个朋友准备雅思口语发现一个很尴尬的事市面上练口语的产品要么是固定脚本的对话机器人翻来覆去就那几句How are you、Im fine thank you要么是通用大模型套壳聊着聊着就跑题你问它餐厅点餐它跟你扯到量子力学。真正能进入一个情景、扮演一个角色、围绕一个任务把对话推进下去的产品少得可怜。这就是我动手做这个英语情景教学Agent的起点。它的核心目标很明确用户选一个场景比如机场值机、酒店投诉、面试自我介绍Agent扮演场景里的NPC地勤、前台、面试官用英语跟用户完成一段有目标、有反馈、有纠错的对话。对话结束后Agent还要给出语法、用词、流利度三个维度的点评。技术栈我选了FastAPI WebSocket React这套组合。后端FastAPI负责Agent编排、大模型调用、会话状态管理WebSocket负责把Agent的流式回复实时推到前端前端React负责场景选择、对话界面、实时字幕和纠错高亮。为什么不用轮询因为对话场景对延迟极其敏感用户说完一句话等三秒才看到回复体验直接崩掉。WebSocket的全双工特性天然适合这种边说边推的场景。这篇文章我会把整个开发过程拆开讲架构怎么设计、WebSocket心跳和断线重连怎么处理、Agent的Prompt怎么编排、并发怎么扛、前端React怎么组织状态。适合有Python和React基础、想做一个真实可用的AI Agent项目的同学。没有基础也能看我会把关键概念用生活化的例子讲清楚。2. 整体架构设计与技术选型思路2.1 三层架构前端、网关、Agent核心整个系统我分成三层这个分层不是拍脑袋定的是被实际需求逼出来的。第一层是React前端负责交互和展示。它要处理的东西比想象中多场景列表渲染、麦克风录音、WebSocket连接管理、流式文字逐字显示、纠错标注高亮。这些状态如果全塞在一个组件里代码会烂成一锅粥。所以我用Zustand做全局状态管理把连接状态、当前会话、消息列表、纠错结果拆成独立的store。第二层是FastAPI网关它是整个系统的中枢。所有WebSocket连接在这里建立所有HTTP接口场景列表、历史记录、用户配置也在这里暴露。网关层不直接调大模型而是把请求转发给Agent核心同时负责鉴权、限流、会话路由。第三层是Agent核心包含Prompt编排、对话状态机、大模型调用、纠错分析。这一层我单独抽成一个模块不跟WebSocket耦合原因是Agent逻辑未来可能要复用到其他入口比如小程序、命令行工具如果跟WebSocket绑死复用成本极高。三层之间用明确的接口通信前端通过WebSocket发JSON消息给网关网关通过内部函数调用Agent核心Agent核心返回异步生成器async generator网关再把生成器的每个chunk推回前端。这个设计的关键点是Agent核心返回的是异步生成器而不是完整字符串这样才能实现流式输出。2.2 为什么是FastAPI而不是Django热词里有人问python django websocket实现后台有数据前端推送我实际对比过两个方案。Django本身是同步框架WebSocket要靠Channels这个额外组件而Channels又依赖ASGI服务器Daphne或Uvicorn配置链路长调试起来一层套一层。更麻烦的是Django的ORM是同步的在异步的WebSocket消费者里调用同步ORM要么用sync_to_async包一层要么忍受性能损耗。FastAPI原生就是ASGI框架async def直接写WebSocket支持是内置的不需要额外装东西。而且FastAPI的依赖注入系统Depends在WebSocket场景下也能用鉴权、数据库会话注入都很自然。我实测下来同样一个接收消息-调用模型-流式返回的链路FastAPI的代码量大概是Django Channels的一半。当然Django也有优势比如自带Admin、ORM成熟、生态庞大。如果你的项目已经有Django存量代码硬切FastAPI不划算。但这是从零开始的新项目FastAPI是更顺手的选择。2.3 为什么不用Gradio热词里有gradio和fastapi的对比。Gradio确实能几分钟搭一个AI Demo拖几个组件、绑一个函数就完事。但它的问题在于定制能力太弱。我想要的是对话气泡带纠错高亮、场景卡片有封面图、录音按钮有波形动画、连接断了要自动重连。这些Gradio要么做不了要么要写一堆hack。Gradio适合做内部演示和快速验证不适合做面向用户的产品。我的判断标准很简单如果这个界面未来要上线给真实用户用就别用Gradio老老实实写React。2.4 目录结构一个能长期维护的FastAPI项目长什么样热词里fastapi项目目录结构被搜了很多次我把自己用的结构贴出来这个结构是踩过坑之后定下来的english-scenario-agent/ ├── app/ │ ├── main.py # FastAPI入口注册路由和WebSocket │ ├── config.py # 配置管理读环境变量 │ ├── api/ │ │ ├── routes_scenario.py # 场景相关HTTP接口 │ │ └── routes_history.py # 历史记录接口 │ ├── ws/ │ │ ├── connection.py # WebSocket连接管理器 │ │ └── handler.py # WebSocket消息处理逻辑 │ ├── agent/ │ │ ├── orchestrator.py # Agent编排主逻辑 │ │ ├── prompts.py # Prompt模板 │ │ ├── state.py # 对话状态机 │ │ └── evaluator.py # 纠错与评分 │ ├── models/ │ │ └── schemas.py # Pydantic模型 │ └── services/ │ └── llm_client.py # 大模型客户端封装 ├── frontend/ # React项目 ├── tests/ └── requirements.txt这个结构的关键是按职责分层而不是按技术分层。很多教程喜欢建controllers/、services/、utils/这种目录结果一个业务逻辑散落在四五个文件夹里。我按api、ws、agent分每个目录内部自洽改一个功能基本只动一个目录。3. WebSocket实时通信的核心细节3.1 连接管理器为什么不能只存一个连接新手最容易犯的错是全局存一个WebSocket对象。单用户测试没问题一上多用户就串线——A用户的消息推给了B用户。正确做法是用一个连接管理器用字典维护用户ID - 连接的映射。# app/ws/connection.py from fastapi import WebSocket from typing import Dict import asyncio class ConnectionManager: def __init__(self): self.active: Dict[str, WebSocket] {} self.lock asyncio.Lock() async def connect(self, user_id: str, ws: WebSocket): await ws.accept() async with self.lock: self.active[user_id] ws async def disconnect(self, user_id: str): async with self.lock: self.active.pop(user_id, None) async def send(self, user_id: str, message: dict): ws self.active.get(user_id) if ws: await ws.send_json(message) manager ConnectionManager()这里加asyncio.Lock是有原因的。FastAPI的WebSocket处理是并发的两个协程同时读写self.active字典在极端情况下会出问题。加锁虽然有一点点性能开销但换来的是数据一致性值得。3.2 心跳机制为什么连接会假死热词里websocket心跳机制实现和websocket连接但不接受信息这两个问题本质是同一件事TCP连接看起来还在但实际上已经断了。这种情况在移动网络下特别常见。用户手机从WiFi切到4G运营商的NAT网关会把旧连接悄悄回收但客户端和服务端都不知道双方还傻等着对方发消息。结果就是用户发了一句话界面一直转圈永远等不到回复。解决办法是心跳。服务端每隔30秒发一个ping客户端收到后回pong。如果服务端连续两次没收到pong就判定连接已死主动关闭并通知前端重连。# app/ws/handler.py import asyncio from fastapi import WebSocket, WebSocketDisconnect async def heartbeat(ws: WebSocket, interval: int 30): try: while True: await asyncio.sleep(interval) await ws.send_json({type: ping}) except Exception: pass async def ws_endpoint(ws: WebSocket, user_id: str): await manager.connect(user_id, ws) hb_task asyncio.create_task(heartbeat(ws)) try: while True: data await ws.receive_json() if data.get(type) pong: continue # 处理正常业务消息 await handle_message(user_id, data) except WebSocketDisconnect: pass finally: hb_task.cancel() await manager.disconnect(user_id)心跳间隔设30秒是个经验值。设太短比如5秒会增加无谓的网络流量和电量消耗设太长比如120秒则断线检测不及时用户要等两分钟才发现连不上。30秒是大多数即时通讯产品的默认值。3.3 断线重连前端要做的三件事后端有心跳还不够前端必须配合做重连。React这边我用一个自定义Hook封装核心逻辑是三件事第一监听onclose事件触发重连。第二重连要用指数退避第一次等1秒第二次等2秒第三次等4秒最多等到30秒。为什么因为如果服务端挂了所有客户端同时疯狂重连会把刚恢复的服务端再次打垮。第三重连成功后要补拉断线期间的消息否则用户会发现自己漏了几句话。// frontend/src/hooks/useWebSocket.js import { useEffect, useRef, useCallback } from react; export function useWebSocket(url, onMessage) { const wsRef useRef(null); const retryRef useRef(0); const timerRef useRef(null); const connect useCallback(() { const ws new WebSocket(url); wsRef.current ws; ws.onopen () { retryRef.current 0; ws.send(JSON.stringify({ type: resume, lastId: getLastMsgId() })); }; ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type ping) { ws.send(JSON.stringify({ type: pong })); return; } onMessage(msg); }; ws.onclose () { const delay Math.min(1000 * 2 ** retryRef.current, 30000); retryRef.current 1; timerRef.current setTimeout(connect, delay); }; }, [url, onMessage]); useEffect(() { connect(); return () { clearTimeout(timerRef.current); wsRef.current?.close(); }; }, [connect]); return wsRef; }这里有个细节onclose里不要直接调connect要用setTimeout。因为如果连接建立失败onclose会立刻触发直接调connect会形成同步递归把浏览器栈打爆。3.4 消息协议设计别用裸字符串WebSocket传的是文本但千万别直接传用户说的话。要设计一套消息协议用JSON包一层带上type字段区分消息种类。我的协议长这样type方向说明user_message前端→后端用户输入的文本或语音转写结果agent_chunk后端→前端Agent流式回复的单个片段agent_done后端→前端Agent回复结束附带纠错结果ping/pong双向心跳error后端→前端错误通知resume前端→后端重连后请求补发消息用type区分的好处是前端可以用一个switch统一分发新增消息类型不用改通信层。而且未来要加用户打断Agent、Agent请求澄清这类功能直接加新type就行协议是向后兼容的。4. Agent核心Prompt编排与对话状态机4.1 情景教学Agent的Prompt该怎么写这是整个项目最核心的部分。通用聊天机器人的Prompt很简单你是一个有用的助手。但情景教学Agent不行它必须同时满足四个约束扮演特定角色、维持场景设定、推进对话目标、纠正用户错误。我的Prompt模板分四块# app/agent/prompts.py SCENARIO_PROMPT You are role-playing as {role} in a scenario: {scenario}. ## Your Character - Role: {role} - Personality: {personality} - Goal: {role_goal} ## Conversation Rules 1. Always respond in English, keep sentences natural and conversational. 2. Stay in character. Never break the scenario unless the user explicitly asks. 3. If the user makes a grammar or word-choice error, DO NOT interrupt the flow. Instead, note it silently and continue the conversation naturally. 4. Gently steer the conversation toward the scenario goal if the user drifts. 5. Keep each response under 60 words to maintain a natural dialogue pace. ## Scenario Context {context} ## Current Progress Turn: {turn_count} Goal achieved: {goal_achieved} 这里有几个设计决策值得说。为什么要求回复不超过60词因为这是口语练习不是阅读理解。Agent说一大段用户根本没机会开口。60词大概是3-4句话既能推进对话又给用户留出回应空间。为什么纠错要silently note而不是当场打断这是教学法的考量。真实对话中你说话时对方不停纠正你你会紧张到说不出话。正确做法是让对话自然进行结束后统一反馈。这跟人类外教的做法一致。为什么要传turn_count和goal_achieved因为大模型没有记忆每次调用都是无状态的。如果不告诉它已经聊了5轮、目标还没达成它可能会在第2轮就草草结束对话。把状态显式传进去Agent才能有节奏感。4.2 对话状态机让Agent知道现在该干嘛光有Prompt还不够Agent需要一个状态机来管理对话流程。我把一个情景对话分成四个阶段开场阶段openingAgent主动打招呼设定场景。比如Good morning, welcome to Heathrow Airport. May I see your passport?推进阶段progressing用户和Agent来回对话Agent根据用户回答推进剧情。收尾阶段closing目标达成Agent自然结束对话。评估阶段evaluating对话结束后Agent切换到评估模式输出纠错和评分。# app/agent/state.py from enum import Enum from pydantic import BaseModel from typing import List class Stage(str, Enum): OPENING opening PROGRESSING progressing CLOSING closing EVALUATING evaluating class DialogueState(BaseModel): scenario_id: str stage: Stage Stage.OPENING turn_count: int 0 goal_achieved: bool False history: List[dict] [] errors: List[dict] [] def advance(self): self.turn_count 1 if self.turn_count 3 and self.goal_achieved: self.stage Stage.CLOSING elif self.turn_count 8: self.stage Stage.CLOSING else: self.stage Stage.PROGRESSINGturn_count 8这个兜底很重要。有些用户会一直闲聊不推进目标如果不设上限对话永远结束不了用户会觉得这个Agent没有边界感。4.3 流式输出async generator是关键Agent调用大模型时如果用普通的await等完整回复用户要等3-5秒才能看到第一个字。体验很差。正确做法是用流式API模型每生成一个token就推给前端。# app/agent/orchestrator.py from typing import AsyncGenerator async def run_agent(state: DialogueState, user_input: str) - AsyncGenerator[dict, None]: state.history.append({role: user, content: user_input}) prompt build_prompt(state) full_response async for chunk in llm_client.stream(prompt, state.history): full_response chunk yield {type: agent_chunk, content: chunk} state.history.append({role: assistant, content: full_response}) state.advance() if state.stage Stage.CLOSING: evaluation await evaluate_dialogue(state) yield {type: agent_done, evaluation: evaluation} else: yield {type: agent_done}这个async for配合yield的写法是FastAPI WebSocket流式推送的标准模式。WebSocket handler里直接async for chunk in run_agent(...)每拿到一个chunk就await ws.send_json(chunk)。整条链路全异步没有阻塞。4.4 纠错评估怎么让大模型给出靠谱的反馈对话结束后的评估是教学价值的核心。我的做法是让大模型返回结构化JSON而不是自由文本。自由文本没法在前端做高亮和分类展示。EVAL_PROMPT Analyze the following English dialogue between a learner and an NPC. Return a JSON object with this exact structure: { grammar_errors: [{original: ..., corrected: ..., explanation: ...}], vocabulary_suggestions: [{used: ..., better: ..., reason: ...}], fluency_score: 0-100, grammar_score: 0-100, vocabulary_score: 0-100, overall_comment: ... } Dialogue: {dialogue} 用JSON Schema约束输出配合大模型的结构化输出能力比如OpenAI的response_format能拿到稳定的JSON。如果模型偶尔返回了带markdown代码块的JSON解析前先strip掉json和。5. 并发处理与性能优化5.1 AI Agent怎么扛并发热词里ai agent 怎么扛并发是个真问题。Agent的瓶颈不在WebSocket而在大模型调用。一次对话请求要占用一个模型连接几秒钟100个用户同时说话就是100个并发模型请求。我的策略是分层限流 连接池。第一层在WebSocket入口做用户级限流。同一个用户1秒内最多发1条消息防止有人狂点发送按钮。第二层在Agent核心做全局并发控制。用一个asyncio.Semaphore限制同时进行的模型调用数量比如设成50。超过的请求排队等待而不是直接拒绝。# app/agent/orchestrator.py import asyncio LLM_SEMAPHORE asyncio.Semaphore(50) async def run_agent(state, user_input): async with LLM_SEMAPHORE: # 模型调用逻辑 ...Semaphore的好处是它不丢请求只是让请求排队。用户感觉到的只是回复慢了一点而不是发送失败。第三层模型客户端本身要复用HTTP连接。用httpx.AsyncClient而不是每次请求新建client能省掉大量TCP握手开销。5.2 会话状态存哪里对话状态DialogueState不能只放内存。原因有两个一是服务重启状态就丢了二是未来要水平扩展多台服务器要共享状态。我的方案是内存 Redis双层。热数据当前活跃会话放内存读写快同时异步写一份到Redis用于持久化和跨实例共享。用户重连时先从内存找找不到再从Redis恢复。# app/services/session_store.py import json import redis.asyncio as redis class SessionStore: def __init__(self): self.memory {} self.redis redis.from_url(redis://localhost:6379) async def save(self, user_id: str, state: DialogueState): self.memory[user_id] state await self.redis.setex( fsession:{user_id}, 3600, state.model_dump_json() ) async def load(self, user_id: str) - DialogueState | None: if user_id in self.memory: return self.memory[user_id] raw await self.redis.get(fsession:{user_id}) if raw: state DialogueState.model_validate_json(raw) self.memory[user_id] state return state return Nonesetex的3600是过期时间一小时。用户一小时内没继续对话会话自动清理不会无限占内存。5.3 前端React状态管理别把所有东西塞进useStateReact这边我踩过一个坑一开始把所有状态都放在顶层组件的useState里结果每次Agent推一个chunk整个组件树都重渲染界面卡得不行。后来改用Zustand把状态拆成几个独立的store每个store只订阅自己关心的部分。Agent推chunk时只有消息列表那个store更新场景列表、用户配置这些不受影响。// frontend/src/store/dialogueStore.js import { create } from zustand; export const useDialogueStore create((set) ({ messages: [], streaming: false, currentStreamId: null, appendChunk: (chunk) set((state) { const msgs [...state.messages]; const last msgs[msgs.length - 1]; if (last last.id state.currentStreamId) { last.content chunk; } else { msgs.push({ id: state.currentStreamId, role: agent, content: chunk }); } return { messages: msgs }; }), startStream: (id) set({ streaming: true, currentStreamId: id }), endStream: () set({ streaming: false, currentStreamId: null }), }));这里appendChunk的逻辑是如果最后一条消息就是当前流式消息就追加内容否则新建一条。这样前端就能实现逐字打字的效果。5.4 流式渲染的性能陷阱逐字追加内容时如果每条消息都触发一次完整列表重渲染消息多了会卡。优化手段有两个一是用React.memo包裹消息气泡组件只有内容变化的那个气泡重渲染。二是对chunk做批量合并。模型每秒可能推几十个chunk如果每个chunk都setStateReact的调度器会被打爆。我的做法是用一个buffer每50毫秒flush一次。// frontend/src/hooks/useStreamBuffer.js import { useRef, useEffect } from react; export function useStreamBuffer(onFlush, interval 50) { const bufferRef useRef(); const timerRef useRef(null); useEffect(() { timerRef.current setInterval(() { if (bufferRef.current) { onFlush(bufferRef.current); bufferRef.current ; } }, interval); return () clearInterval(timerRef.current); }, [onFlush, interval]); return (chunk) { bufferRef.current chunk; }; }50毫秒是人眼感知的阈值。低于这个间隔的更新人眼分辨不出来纯属浪费性能。6. 常见问题与排查技巧实录6.1 WebSocket连接上了但收不到消息这是热词里websocket连接但不接受信息的典型症状。我排查过几次原因基本是这三类第一类消息被中间层吞了。如果前面挂了Nginx默认配置下WebSocket的升级请求可能被拦截。要在Nginx配置里加location /ws/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_read_timeout 3600s; }proxy_read_timeout默认是60秒不改的话连接空闲60秒就被Nginx掐断表现就是连上了但过一会儿收不到消息。第二类后端异常被静默吞掉。WebSocket handler里如果抛异常FastAPI默认会关闭连接但不一定发错误消息。我的做法是在handler外层包一层try/except捕获后主动发一个{type: error}给前端前端弹个提示用户至少知道出问题了。第三类前端onmessage没绑对。这个最蠢但最常见。ws.onmessage要在new WebSocket()之后立刻绑如果放在onopen里绑会漏掉onopen和onmessage之间到达的消息。6.2 Agent回复跑题或出戏大模型偶尔会出戏比如你让它扮演机场地勤它突然开始跟你聊天气。这是Prompt约束不够强导致的。我的解决办法是在Prompt里加负面约束明确列出不能做的事## Strict Rules - NEVER mention that you are an AI or a language model. - NEVER switch to Chinese unless the user explicitly asks for translation. - NEVER discuss topics unrelated to the scenario. - If the user asks something off-topic, respond in character and redirect: Sorry, I can only help with check-in today. Shall we continue?负面约束比正面约束更有效因为大模型对不要做什么的遵循度往往高于要做什么。6.3 流式输出时前端闪烁逐字追加内容时如果每次都重新计算整个消息列表的高度界面会抖动。解决办法是给消息容器设固定高度加overflow-y: auto并且用scrollIntoView时加behavior: smooth让滚动平滑而不是跳变。另一个技巧是给正在流式的消息加一个光标动画用一个CSS伪元素实现不占用实际DOM节点性能开销几乎为零。6.4 常见问题速查表症状可能原因排查方向连接建立后立即断开鉴权失败或路径不匹配检查WebSocket URL和token连接正常但收不到消息Nginx超时或后端异常查Nginx日志和后端异常栈消息延迟高模型调用慢或并发排队看Semaphore等待时间前端卡顿频繁setState触发重渲染用buffer批量更新重连后消息丢失没做消息补拉实现resume协议Agent出戏Prompt约束不足加负面约束和角色锚定内存持续增长会话状态没清理检查Redis过期和内存字典6.5 几个我踩过的坑坑一不要在WebSocket handler里做耗时同步操作。我一开始在handler里直接读文件加载场景配置结果一个用户读文件时其他用户的WebSocket消息全被阻塞。后来改成启动时预加载到内存handler里只读内存。坑二asyncio.create_task创建的任务要保存引用。心跳任务如果只create_task不保存Python的垃圾回收可能会把它回收掉心跳就停了。要么保存到变量要么用asyncio.TaskGroup管理。坑三前端录音要用MediaRecorder而不是Web Audio API裸写。MediaRecorder是浏览器原生封装几行代码就能录音并拿到Blob。用Web Audio API自己处理PCM数据代码量翻十倍还容易出兼容性问题。坑四语音转文字不要在前端做。前端做STT要么用浏览器自带的兼容性差要么调第三方API暴露密钥。正确做法是把音频Blob通过WebSocket传给后端后端调STT服务转写结果再走Agent流程。7. 前端React的关键实现细节7.1 场景选择页卡片式布局与状态预加载场景选择页看起来简单但有个细节要注意用户点击场景卡片后不要等进入对话页才开始建立WebSocket连接。应该在点击的瞬间就发起连接同时并行加载场景配置。这样用户进入对话页时连接已经就绪Agent的开场白可以立刻推送。// frontend/src/pages/ScenarioSelect.jsx const handleSelect async (scenario) { const wsUrl ${WS_BASE}/ws/chat?scenario${scenario.id}; const ws new WebSocket(wsUrl); // 把ws实例存到store对话页直接复用 useConnectionStore.getState().setSocket(ws); navigate(/chat/${scenario.id}); };这个提前建连的技巧能把用户感知的首屏延迟从1-2秒降到几乎为零。7.2 对话页消息气泡与纠错高亮对话页的核心是消息列表。用户消息靠右Agent消息靠左这是即时通讯的通用范式用户不需要学习成本。纠错高亮是教学产品的特色。对话结束后Agent返回的grammar_errors里每条都有original字段。前端在渲染用户消息时用字符串匹配找到这些片段包一层mark标签。function highlightErrors(text, errors) { let result text; errors.forEach((err) { const regex new RegExp(\\b${escapeRegex(err.original)}\\b, gi); result result.replace(regex, mark classerror$/mark); }); return result; }注意escapeRegex用户输入里可能有正则特殊字符不转义会导致正则报错。7.3 录音与语音输入录音用MediaRecorder核心代码就几行const startRecording async () { const stream await navigator.mediaDevices.getUserMedia({ audio: true }); const recorder new MediaRecorder(stream, { mimeType: audio/webm }); const chunks []; recorder.ondataavailable (e) chunks.push(e.data); recorder.onstop () { const blob new Blob(chunks, { type: audio/webm }); sendAudio(blob); }; recorder.start(); return recorder; };sendAudio把Blob转成ArrayBuffer通过WebSocket的二进制帧发出去。后端收到后调STT转写结果再走正常的文本消息流程。这里有个兼容性坑Safari不支持audio/webm要用audio/mp4。所以mimeType要做特性检测不能写死。7.4 连接状态指示器用户需要知道当前连接是否正常。我在页面顶部放了一个小圆点绿色表示已连接黄色表示重连中红色表示断开。这个指示器直接订阅connectionStore的状态不需要额外逻辑。const status useConnectionStore((s) s.status); const color { connected: #22c55e, reconnecting: #eab308, disconnected: #ef4444 }[status];别小看这个指示器。用户遇到问题时第一反应是是不是网断了有了指示器他们能自己判断减少大量客服咨询。8. 部署与扩展的一些实际考虑8.1 单机部署的最小配置这个项目单机跑起来2核4G的云服务器足够。FastAPI用Uvicorn起多worker但要注意WebSocket连接是有状态的多worker之间不共享连接。如果起4个worker用户A连到worker1用户B连到worker2他们之间的消息无法互通。解决办法是用Redis做pub/sub所有worker订阅同一个channel消息通过Redis广播。或者更简单WebSocket服务单独起一个workerHTTP接口起多个worker两者分开部署。8.2 水平扩展时的会话粘性如果要多台服务器必须解决会话粘性问题。同一个用户的WebSocket连接必须始终落在同一台服务器上否则状态会丢。方案有两种一是用Nginx的ip_hash做粘性路由二是把会话状态完全外置到Redis任何服务器都能处理任何请求。我推荐第二种虽然每次请求都要读Redis但扩展性最好服务器可以随意增减。8.3 监控要看哪些指标上线后我重点盯三个指标WebSocket连接数、模型调用P95延迟、会话完成率。连接数突然下降说明可能有断线风暴模型延迟飙升说明并发排队严重要考虑扩容或换更快的模型会话完成率低说明Agent的对话设计有问题用户聊到一半就跑了。这三个指标我用Prometheus Grafana做可视化FastAPI这边用prometheus-fastapi-instrumentator自动埋点WebSocket的自定义指标手动埋。8.4 成本控制大模型调用是主要成本。我的控制手段有三个一是限制单次对话的最大轮数8轮防止用户无限闲聊二是对简单场景用小模型复杂场景才用大模型三是缓存常见场景的开场白开场白是固定的不需要每次调模型生成。开场白缓存这个技巧很实用。一个场景的开场白就那几句预生成好存起来用户进入场景直接推缓存省掉一次模型调用。1000个用户进入同一个场景就省了1000次调用。9. 我个人的一些体会这个项目从想法到能跑起来大概花了我三个周末。最大的感受是AI Agent项目的难点不在模型而在工程。模型能力现在都很强但要把模型能力稳定、低延迟、可扩展地交付给用户中间全是工程细节。WebSocket的心跳、重连、消息补拉会话状态的持久化流式输出的性能优化这些才是决定产品能不能用的关键。另一个体会是Prompt要当代码来管理。我一开始把Prompt硬编码在Python文件里改一次要重启服务。后来抽成独立的模板文件支持热加载调Prompt的效率高了很多。而且Prompt要版本化每次改动记录一下改了什么、为什么改不然改着改着就忘了当初为什么那么写。最后分享一个小技巧给Agent加一个调试模式。在URL上加个?debug1前端就会显示每次模型调用的完整Prompt、token消耗、耗时。这个功能在排查Agent为什么这么回复时极其有用强烈建议一开始就加上。这个项目后续还可以扩展的方向不少比如加多模态让Agent能看到用户上传的图片、加多人对话多个用户和Agent一起练、加自适应难度根据用户水平动态调整Agent的用词复杂度。但那是下一步的事了先把核心对话链路打磨稳定比什么都重要。