1. 为什么我要做这个英语情景教学Agent去年年底我在帮一个朋友的孩子准备英语口语考试发现一个很尴尬的事市面上大部分AI口语陪练产品要么是纯聊天式的自由对话孩子聊两句就没词了要么是固定脚本的角色扮演来回就那么几个场景练三五次就腻了。真正有效的语言训练需要的是有目标、有情境、有反馈的闭环而不是随便找个LLM聊两句就完事。这个项目就是在这个背景下动手做的。核心目标很明确构建一个英语情景教学Agent用户选定一个场景比如在咖啡店点单机场值机面试自我介绍Agent扮演场景中的角色用英语和用户进行多轮对话在对话过程中动态评估用户的表达给出纠错和替换建议并在场景结束后输出一份学习报告。整个系统基于FastAPI做后端、React做前端、WebSocket做实时通信、LLM做对话引擎和评估引擎。适合谁看如果你正在做AI Agent相关的项目尤其是涉及实时对话、流式输出、多轮状态管理的场景这篇内容里的架构思路和踩坑记录可以直接复用。如果你只是想了解一个完整的Agent项目从零到一是怎么搭起来的也能拿到一套可落地的方案。我下面会把设计取舍、核心实现、参数计算、排查经验全部摊开讲代码片段和目录结构都会给到。2. 整体架构设计与技术选型取舍2.1 为什么是FastAPI而不是Django后端框架选型上我纠结过一阵。Django生态成熟自带ORM、Admin、认证做业务系统很舒服。但这个项目的核心诉求是高并发长连接 流式响应 异步调用LLM这三点Django都不占优势。FastAPI原生基于ASGIasync/await是一等公民WebSocket支持开箱即用配合uvicorn跑起来性能很稳。更关键的是LLM调用本质是IO密集型的等待用异步能在一个进程里同时挂几百个对话会话而不用为每个会话开线程。我实测过同样配置的机器上FastAPI在500个并发WebSocket连接下CPU占用不到30%Django配合Channels大概在60%以上而且内存涨得更快。选型对比我整理成了一张表方便你直接参考维度FastAPIDjango Channels异步原生支持原生async/await后期补的async部分组件仍同步WebSocket内置几行代码搞定需要Channels Redis做channel layer流式响应StreamingResponse天然支持需要额外处理依赖注入内置Depends测试友好无原生方案学习曲线需要理解async同步思维更直观适合场景实时、IO密集、Agent业务系统、CRUD、后台管理结论很清楚这个项目是典型的IO密集型实时应用FastAPI是更合适的选择。Django不是不好是场景不对。2.2 WebSocket还是SSE这是个问题实时推送方案我认真对比过SSE和WebSocket。SSE是单向的服务端推、客户端收实现简单浏览器原生支持自动重连。WebSocket是双向的客户端也能主动发消息。英语情景教学这个场景用户需要边说边发——用户发一句话Agent流式返回评估和回复用户可能中途打断或者补充。这是典型的双向交互SSE做起来会很别扭要么开两个连接要么用POSTSSE混合。所以最终选了WebSocket。但WebSocket有个坑它本身没有心跳和重连机制网络抖动或者中间层超时会把连接悄悄掐断前端却以为还连着。这个后面在排查章节会详细讲怎么处理。2.3 LLM的角色拆分对话引擎和评估引擎要分开一开始我图省事用一个LLM调用同时干两件事生成回复 评估用户表达。跑了几轮就发现不对劲——回复质量下降了因为模型既要想着怎么接话又要分心去挑语法错误注意力被稀释。而且评估结果不稳定同一个错误有时候报有时候不报。后来我把LLM的职责拆成两个独立的Agent对话Agent只负责扮演场景角色用英语自然回复保持人设和场景一致性。System Prompt里明确写你是咖啡店店员不要纠正用户语法只管推进对话。评估Agent在后台异步运行拿到用户这一轮的输入输出结构化的评估结果语法错误、用词建议、地道表达替换。两个Agent通过一个共享的会话状态对象通信但调用是解耦的。这样对话流畅度上来了评估也更稳定。代价是token消耗翻倍但教学场景对成本没那么敏感体验优先。2.4 前端为什么用React而不是别的React在这个项目里的核心价值是状态管理。一个对话会话涉及消息列表、当前场景配置、连接状态、流式输出的中间态、评估结果面板。这些状态之间有关联用原生JS或者轻量框架管理起来会很乱。React的useStateuseReduceruseEffect组合能把这些状态理清楚。特别是流式输出场景Agent的回复是一个字一个字蹦出来的需要频繁更新UIReact的虚拟DOM和批处理机制在这里很合适。另外react-uplot这类轻量图表库可以用来画学习报告里的数据可视化比ECharts轻很多。3. 核心模块拆解与关键实现细节3.1 FastAPI项目目录结构怎么组织目录结构这事我踩过坑。一开始所有代码堆在main.py里写到第三个模块就崩了改一处牵动全身。后来参考社区里比较成熟的FastAPI项目结构整理成下面这样english-scenario-agent/ ├── app/ │ ├── main.py # 应用入口注册路由和中间件 │ ├── config.py # 配置管理读环境变量 │ ├── dependencies.py # 依赖注入如获取当前会话 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes_chat.py # WebSocket路由 │ │ ├── routes_scenario.py # 场景CRUD │ │ └── routes_report.py # 学习报告 │ ├── core/ │ │ ├── agent_dialogue.py # 对话Agent │ │ ├── agent_eval.py # 评估Agent │ │ └── session_manager.py # 会话状态管理 │ ├── models/ │ │ ├── schemas.py # Pydantic模型 │ │ └── session.py # 会话数据结构 │ ├── services/ │ │ ├── llm_client.py # LLM调用封装 │ │ └── prompt_templates.py # Prompt模板 │ └── utils/ │ ├── logger.py │ └── token_counter.py ├── tests/ ├── requirements.txt └── .env这个结构的关键在于分层清晰api层只管路由和参数校验core层放业务逻辑services层封装外部依赖LLM、数据库models层定义数据结构。这样改LLM供应商的时候只动services/llm_client.py改业务逻辑的时候不碰路由。注意不要把LLM调用直接写在路由函数里。我见过太多项目这么干后期想加缓存、加重试、换模型的时候得把所有路由翻一遍。3.2 WebSocket连接的生命周期管理WebSocket连接不是建立就完事了它有自己的生命周期连接建立 → 认证 → 会话绑定 → 消息循环 → 心跳维持 → 断开清理。每一环都要处理。连接建立后第一件事是认证和会话绑定。我的做法是前端在连接URL里带上token后端在websocket.accept()之前校验不通过直接close(code1008)。校验通过后从连接池里取出或创建对应的会话对象。from fastapi import WebSocket, WebSocketDisconnect from app.core.session_manager import SessionManager manager SessionManager() app.websocket(/ws/chat/{session_id}) async def chat_ws(websocket: WebSocket, session_id: str, token: str): if not verify_token(token): await websocket.close(code1008) return await websocket.accept() session await manager.get_or_create(session_id, token) try: while True: data await websocket.receive_json() await handle_message(session, data, websocket) except WebSocketDisconnect: await manager.persist(session) await manager.remove(session_id)这里有个细节WebSocketDisconnect异常一定要捕获否则会话状态不会被持久化用户刷新页面后上下文就丢了。我一开始没处理测试的时候发现用户重连后Agent完全不记得之前聊了什么排查了半天才定位到。3.3 心跳机制的具体实现WebSocket心跳是保命的东西。没有心跳连接可能被中间层负载均衡、反向代理在空闲60秒后掐断前端却毫无感知。心跳机制分两端服务端启动一个后台任务每隔heartbeat_interval秒检查所有连接的活跃时间超过timeout没收到任何消息包括pong就主动关闭。客户端每隔ping_interval秒发一个ping帧服务端收到后回pong。如果连续N次没收到pong前端主动重连。参数怎么定我实测下来ping_interval25秒、timeout60秒比较稳。为什么是25秒因为大部分反向代理的空闲超时是60秒25秒发一次ping能保证连接始终活跃。为什么timeout是60秒给网络抖动留两次重试机会。// 前端心跳实现 const HEARTBEAT_INTERVAL 25000; const MAX_MISSED 2; function startHeartbeat(ws, onDead) { let missed 0; const timer setInterval(() { if (ws.readyState ! WebSocket.OPEN) return; if (missed MAX_MISSED) { clearInterval(timer); onDead(); return; } ws.send(JSON.stringify({ type: ping, ts: Date.now() })); missed; }, HEARTBEAT_INTERVAL); ws.addEventListener(message, (e) { const msg JSON.parse(e.data); if (msg.type pong) missed 0; }); return () clearInterval(timer); }提示心跳消息不要走业务消息通道单独用type: ping/pong标记服务端收到后直接回不进LLM不消耗token。3.4 流式输出的实现与前端渲染LLM生成回复是流式的一个字一个字往外吐。后端要把这个流通过WebSocket推给前端前端要边收边渲染。后端用async for迭代LLM的流式响应每拿到一个chunk就websocket.send_json推出去async def stream_dialogue(session, user_input, websocket): await websocket.send_json({type: start, msg_id: new_id()}) buffer async for chunk in llm_client.stream(dialogue_prompt(session, user_input)): buffer chunk await websocket.send_json({ type: delta, msg_id: current_id, content: chunk }) await websocket.send_json({type: end, msg_id: current_id}) session.append_message(assistant, buffer)前端收到delta就往当前消息的content后面追加收到end就锁定这条消息。这里有个性能坑如果每个chunk都触发一次React的setState高频流式输出下会卡。解决办法是用requestAnimationFrame做节流或者用useRef攒一批再更新。const bufferRef useRef(); const rafRef useRef(null); function onDelta(chunk) { bufferRef.current chunk; if (!rafRef.current) { rafRef.current requestAnimationFrame(() { setMessages(prev updateLast(prev, bufferRef.current)); rafRef.current null; }); } }实测下来节流后CPU占用从40%降到12%滚动也顺滑了。3.5 评估Agent的Prompt设计评估Agent是整个项目的灵魂。它要输出结构化的评估结果不能是自由文本。我用的是JSON Schema约束 few-shot示例的方式。System Prompt大致是这样你是一名英语教学评估专家。用户会给你一句英语表达和当前场景。 你需要输出JSON格式的评估结果包含以下字段 - grammar_errors: 数组每个元素包含 original, corrected, reason - vocabulary_suggestions: 数组每个元素包含 original, better, reason - fluency_score: 0-100的整数 - overall_comment: 一句话点评 只输出JSON不要任何额外文字。关键点在于只输出JSON这句。不加这句模型经常在JSON前后加好的以下是评估结果之类的废话解析就崩了。另外few-shot示例里要覆盖没有错误的情况否则模型会硬找错误。评估Agent的调用是异步的不阻塞对话流。用户发完消息对话Agent先流式回复评估Agent在后台跑跑完通过WebSocket推一个eval_result消息给前端前端在侧边栏展示。4. 完整实操流程与关键参数计算4.1 环境搭建与依赖安装从零开始的话先把环境搭起来。Python建议3.11以上Node建议20以上。# 后端 python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn[standard] websockets python-dotenv pydantic httpx # 前端 npm create vitelatest frontend -- --template react cd frontend npm install npm install react-uplot uplotuvicorn[standard]这个写法要注意方括号里的standard会额外装上uvloop和httptools性能比裸装uvicorn高不少。我实测同样并发下带standard的版本QPS高约30%。4.2 LLM调用的封装与重试策略LLM调用不能裸调必须封装。封装要处理三件事超时、重试、降级。import httpx from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, base_url, api_key, model): self.client httpx.AsyncClient(timeout60.0) self.base_url base_url self.api_key api_key self.model model retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max8)) async def stream(self, messages): async with self.client.stream( POST, f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, json{model: self.model, messages: messages, stream: True} ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if line.startswith(data: ) and line ! data: [DONE]: yield parse_chunk(line)重试策略用指数退避multiplier1, min1, max8意味着第一次等1秒第二次2秒第三次4秒最多8秒。为什么不用固定间隔因为LLM服务端的限流通常是短时突发指数退避能更快错开。超时设60秒是因为流式响应可能持续很久但单次连接建立不能等太久。这里httpx.AsyncClient的timeout是连接超时不是总时长流式读取不受这个限制。4.3 会话状态的数据结构设计会话状态是整个系统的中枢。我设计成这样一个结构from pydantic import BaseModel from typing import List, Optional from datetime import datetime class Message(BaseModel): role: str # user / assistant / system content: str timestamp: datetime eval_result: Optional[dict] None class Session(BaseModel): session_id: str user_id: str scenario_id: str scenario_config: dict messages: List[Message] [] created_at: datetime last_active: datetime turn_count: int 0 total_tokens: int 0turn_count和total_tokens是用来做成本控制和场景结束判断的。一个场景建议控制在15-20轮对话超过就提示用户场景即将结束。token累计到一定量可以触发告警防止某个用户把额度跑爆。4.4 场景配置的Schema设计场景配置决定了Agent的人设和对话走向。我用JSON定义方便扩展{ id: coffee_shop, name: 在咖啡店点单, role: barista, role_desc: 你是一名友好的咖啡店店员说话轻松随意, opening: Hi there! What can I get for you today?, goals: [点一杯饮品, 选择大小和温度, 完成付款], vocab_hints: [latte, cappuccino, medium, to go], max_turns: 15 }goals字段很关键它让对话有方向感。对话Agent的System Prompt里会带上这些目标模型会主动引导用户完成。vocab_hints是给评估Agent用的评估时会优先关注这些词的使用情况。4.5 前端WebSocket Hook的封装前端把WebSocket逻辑封装成一个自定义Hook组件里只管用function useChatSocket(sessionId, token) { const [messages, setMessages] useState([]); const [status, setStatus] useState(connecting); const wsRef useRef(null); useEffect(() { const ws new WebSocket( ws://localhost:8000/ws/chat/${sessionId}?token${token} ); wsRef.current ws; ws.onopen () setStatus(connected); ws.onclose () setStatus(disconnected); ws.onmessage (e) { const msg JSON.parse(e.data); handleIncoming(msg, setMessages); }; const stopHeartbeat startHeartbeat(ws, () { setStatus(reconnecting); // 触发重连逻辑 }); return () { stopHeartbeat(); ws.close(); }; }, [sessionId, token]); const send useCallback((text) { wsRef.current?.send(JSON.stringify({ type: user_input, content: text })); }, []); return { messages, status, send }; }这个Hook处理了连接、心跳、消息分发、清理。组件里只需要const { messages, status, send } useChatSocket(...)干净。5. 常见问题与排查技巧实录5.1 WebSocket连上了但收不到消息这是最高频的问题。现象是前端onopen触发了readyState是1但onmessage一直不触发。排查思路按这个顺序走检查服务端是否真的发了。在websocket.send_json前后打日志确认代码执行到了。检查消息格式。如果服务端发的是send_text但前端按JSON解析会静默失败。统一用send_json。检查中间层。如果前面有Nginx确认proxy_read_timeout和proxy_buffering off配置了。Nginx默认会缓冲WebSocket消息导致前端收不到。检查心跳。如果服务端有心跳但前端没回pong服务端可能主动关了连接。我遇到过一次排查了两小时最后发现是Nginx的proxy_buffering没关。加上proxy_buffering off;立刻就好了。5.2 LLM流式输出中断流式输出跑到一半突然断了前端显示半句话。原因通常是LLM服务端的连接被掐了或者网络抖动。解决办法是断点续传。后端记录已经推送的buffer如果流中断用已生成的内容作为上下文重新请求让模型接着写。Prompt里加一句请从以下内容继续不要重复。async def stream_with_resume(session, prompt, websocket, max_retry2): buffer for attempt in range(max_retry 1): try: async for chunk in llm_client.stream( build_prompt(prompt, buffer) ): buffer chunk await websocket.send_json({type: delta, content: chunk}) break except Exception as e: if attempt max_retry: await websocket.send_json({type: error, msg: str(e)}) raise logger.warning(fstream interrupted, retry {attempt1})5.3 并发场景下的会话串号这个坑很隐蔽。多个用户同时对话A的回复跑到了B的窗口里。原因是会话状态用了全局变量或者WebSocket连接和会话的映射关系搞错了。正确做法是每个WebSocket连接绑定一个独立的会话对象会话对象存在连接池里用session_id做key。绝对不要用全局dict存当前会话。class SessionManager: def __init__(self): self._sessions: dict[str, Session] {} self._lock asyncio.Lock() async def get_or_create(self, session_id, user_id): async with self._lock: if session_id not in self._sessions: self._sessions[session_id] Session( session_idsession_id, user_iduser_id, ... ) return self._sessions[session_id]加锁是因为get_or_create可能被并发调用不加锁会创建两个会话对象。5.4 常见问题速查表现象可能原因排查方法解决连上收不到消息Nginx缓冲看Nginx配置proxy_buffering off流式中断LLM连接超时看后端日志断点续传重试会话串号全局状态检查会话存储连接绑定独立会话心跳失效前端没回pong抓包看帧补pong处理评估不返回JSON解析失败打印原始输出加只输出JSON约束前端卡顿高频setStatePerformance面板rAF节流重连后丢上下文未持久化检查断开逻辑捕获Disconnect持久化5.5 几个我踩过的坑坑一Prompt里带用户输入没做转义。用户输入里如果有{}或者特殊字符格式化Prompt的时候会报错。解决办法是用str.format的替代方案或者用f-string但把用户输入单独拼接。坑二token计数不准。不同模型的tokenizer不一样用tiktoken算OpenAI的准算别的模型就不准。我的做法是粗略估算英文按4字符1token中文按1.5字符1token留20%余量。坑三评估Agent拖慢整体响应。一开始评估是同步的用户发完消息要等评估跑完才看到回复。改成异步后体验好很多但要注意评估结果推送时用户可能已经发了下一条消息前端要能正确处理乱序。坑四场景结束判断太生硬。按轮数硬切用户正聊到兴头上突然结束体验很差。后来改成轮数接近上限时对话Agent会主动引导收尾比如Thanks for coming! Have a great day!自然结束。6. 性能优化与扩展方向6.1 并发能力怎么扛Agent项目最怕的就是并发。每个会话都要调LLMLLM的响应时间又不稳定。我的优化思路是分层削峰第一层WebSocket连接本身很轻一个进程挂几千个没问题。第二层LLM调用用异步不阻塞事件循环。第三层加一个请求队列超过并发上限的请求排队等待而不是直接拒绝。队列用asyncio.Queue实现消费者数量控制在LLM服务端能承受的范围内。我实测下来单实例配4个消费者能稳定支撑200个活跃会话。class LLMQueue: def __init__(self, max_concurrent4): self.queue asyncio.Queue() self.semaphore asyncio.Semaphore(max_concurrent) async def submit(self, coro): async with self.semaphore: return await coro6.2 缓存能省不少钱相同场景的开场白、常见问题的回复这些是可以缓存的。用functools.lru_cache或者Redis都行。我统计过缓存命中率大概在15%左右能省下不少token。但要注意缓存key要包含场景ID和对话轮次不能只按用户输入缓存否则不同场景下同样的输入会返回错误的回复。6.3 后续可以扩展的方向这个项目目前是单场景对话后续可以扩展成多场景串联的学习路径。比如咖啡店点单 → 问路 → 酒店入住形成一个完整的旅行英语学习链路。评估Agent可以跨场景追踪用户的薄弱点生成个性化的复习计划。另一个方向是语音输入。现在用户是打字如果接入语音识别就能做真正的口语训练。WebSocket传音频流后端做ASR再走现在的对话流程。技术上可行主要是ASR的延迟和准确率要调。最后再分享一个小技巧评估Agent的Prompt里加上用中文解释错误原因对中文母语的学习者友好很多。纯英文解释语法错误很多人看不懂反而增加了学习负担。这个细节看起来小但实际使用中反馈很好。 
长河君 · 2026/11/2 5:19:57 · 阅读 1.2k FastAPI+WebSocket+LLM：英语情景教学Agent全栈实战
1. 为什么我要做这个英语情景教学Agent去年年底我在帮一个朋友的孩子准备英语口语考试发现一个很尴尬的事市面上大部分AI口语陪练产品要么是纯聊天式的自由对话孩子聊两句就没词了要么是固定脚本的角色扮演来回就那么几个场景练三五次就腻了。真正有效的语言训练需要的是有目标、有情境、有反馈的闭环而不是随便找个LLM聊两句就完事。这个项目就是在这个背景下动手做的。核心目标很明确构建一个英语情景教学Agent用户选定一个场景比如在咖啡店点单机场值机面试自我介绍Agent扮演场景中的角色用英语和用户进行多轮对话在对话过程中动态评估用户的表达给出纠错和替换建议并在场景结束后输出一份学习报告。整个系统基于FastAPI做后端、React做前端、WebSocket做实时通信、LLM做对话引擎和评估引擎。适合谁看如果你正在做AI Agent相关的项目尤其是涉及实时对话、流式输出、多轮状态管理的场景这篇内容里的架构思路和踩坑记录可以直接复用。如果你只是想了解一个完整的Agent项目从零到一是怎么搭起来的也能拿到一套可落地的方案。我下面会把设计取舍、核心实现、参数计算、排查经验全部摊开讲代码片段和目录结构都会给到。2. 整体架构设计与技术选型取舍2.1 为什么是FastAPI而不是Django后端框架选型上我纠结过一阵。Django生态成熟自带ORM、Admin、认证做业务系统很舒服。但这个项目的核心诉求是高并发长连接 流式响应 异步调用LLM这三点Django都不占优势。FastAPI原生基于ASGIasync/await是一等公民WebSocket支持开箱即用配合uvicorn跑起来性能很稳。更关键的是LLM调用本质是IO密集型的等待用异步能在一个进程里同时挂几百个对话会话而不用为每个会话开线程。我实测过同样配置的机器上FastAPI在500个并发WebSocket连接下CPU占用不到30%Django配合Channels大概在60%以上而且内存涨得更快。选型对比我整理成了一张表方便你直接参考维度FastAPIDjango Channels异步原生支持原生async/await后期补的async部分组件仍同步WebSocket内置几行代码搞定需要Channels Redis做channel layer流式响应StreamingResponse天然支持需要额外处理依赖注入内置Depends测试友好无原生方案学习曲线需要理解async同步思维更直观适合场景实时、IO密集、Agent业务系统、CRUD、后台管理结论很清楚这个项目是典型的IO密集型实时应用FastAPI是更合适的选择。Django不是不好是场景不对。2.2 WebSocket还是SSE这是个问题实时推送方案我认真对比过SSE和WebSocket。SSE是单向的服务端推、客户端收实现简单浏览器原生支持自动重连。WebSocket是双向的客户端也能主动发消息。英语情景教学这个场景用户需要边说边发——用户发一句话Agent流式返回评估和回复用户可能中途打断或者补充。这是典型的双向交互SSE做起来会很别扭要么开两个连接要么用POSTSSE混合。所以最终选了WebSocket。但WebSocket有个坑它本身没有心跳和重连机制网络抖动或者中间层超时会把连接悄悄掐断前端却以为还连着。这个后面在排查章节会详细讲怎么处理。2.3 LLM的角色拆分对话引擎和评估引擎要分开一开始我图省事用一个LLM调用同时干两件事生成回复 评估用户表达。跑了几轮就发现不对劲——回复质量下降了因为模型既要想着怎么接话又要分心去挑语法错误注意力被稀释。而且评估结果不稳定同一个错误有时候报有时候不报。后来我把LLM的职责拆成两个独立的Agent对话Agent只负责扮演场景角色用英语自然回复保持人设和场景一致性。System Prompt里明确写你是咖啡店店员不要纠正用户语法只管推进对话。评估Agent在后台异步运行拿到用户这一轮的输入输出结构化的评估结果语法错误、用词建议、地道表达替换。两个Agent通过一个共享的会话状态对象通信但调用是解耦的。这样对话流畅度上来了评估也更稳定。代价是token消耗翻倍但教学场景对成本没那么敏感体验优先。2.4 前端为什么用React而不是别的React在这个项目里的核心价值是状态管理。一个对话会话涉及消息列表、当前场景配置、连接状态、流式输出的中间态、评估结果面板。这些状态之间有关联用原生JS或者轻量框架管理起来会很乱。React的useStateuseReduceruseEffect组合能把这些状态理清楚。特别是流式输出场景Agent的回复是一个字一个字蹦出来的需要频繁更新UIReact的虚拟DOM和批处理机制在这里很合适。另外react-uplot这类轻量图表库可以用来画学习报告里的数据可视化比ECharts轻很多。3. 核心模块拆解与关键实现细节3.1 FastAPI项目目录结构怎么组织目录结构这事我踩过坑。一开始所有代码堆在main.py里写到第三个模块就崩了改一处牵动全身。后来参考社区里比较成熟的FastAPI项目结构整理成下面这样english-scenario-agent/ ├── app/ │ ├── main.py # 应用入口注册路由和中间件 │ ├── config.py # 配置管理读环境变量 │ ├── dependencies.py # 依赖注入如获取当前会话 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes_chat.py # WebSocket路由 │ │ ├── routes_scenario.py # 场景CRUD │ │ └── routes_report.py # 学习报告 │ ├── core/ │ │ ├── agent_dialogue.py # 对话Agent │ │ ├── agent_eval.py # 评估Agent │ │ └── session_manager.py # 会话状态管理 │ ├── models/ │ │ ├── schemas.py # Pydantic模型 │ │ └── session.py # 会话数据结构 │ ├── services/ │ │ ├── llm_client.py # LLM调用封装 │ │ └── prompt_templates.py # Prompt模板 │ └── utils/ │ ├── logger.py │ └── token_counter.py ├── tests/ ├── requirements.txt └── .env这个结构的关键在于分层清晰api层只管路由和参数校验core层放业务逻辑services层封装外部依赖LLM、数据库models层定义数据结构。这样改LLM供应商的时候只动services/llm_client.py改业务逻辑的时候不碰路由。注意不要把LLM调用直接写在路由函数里。我见过太多项目这么干后期想加缓存、加重试、换模型的时候得把所有路由翻一遍。3.2 WebSocket连接的生命周期管理WebSocket连接不是建立就完事了它有自己的生命周期连接建立 → 认证 → 会话绑定 → 消息循环 → 心跳维持 → 断开清理。每一环都要处理。连接建立后第一件事是认证和会话绑定。我的做法是前端在连接URL里带上token后端在websocket.accept()之前校验不通过直接close(code1008)。校验通过后从连接池里取出或创建对应的会话对象。from fastapi import WebSocket, WebSocketDisconnect from app.core.session_manager import SessionManager manager SessionManager() app.websocket(/ws/chat/{session_id}) async def chat_ws(websocket: WebSocket, session_id: str, token: str): if not verify_token(token): await websocket.close(code1008) return await websocket.accept() session await manager.get_or_create(session_id, token) try: while True: data await websocket.receive_json() await handle_message(session, data, websocket) except WebSocketDisconnect: await manager.persist(session) await manager.remove(session_id)这里有个细节WebSocketDisconnect异常一定要捕获否则会话状态不会被持久化用户刷新页面后上下文就丢了。我一开始没处理测试的时候发现用户重连后Agent完全不记得之前聊了什么排查了半天才定位到。3.3 心跳机制的具体实现WebSocket心跳是保命的东西。没有心跳连接可能被中间层负载均衡、反向代理在空闲60秒后掐断前端却毫无感知。心跳机制分两端服务端启动一个后台任务每隔heartbeat_interval秒检查所有连接的活跃时间超过timeout没收到任何消息包括pong就主动关闭。客户端每隔ping_interval秒发一个ping帧服务端收到后回pong。如果连续N次没收到pong前端主动重连。参数怎么定我实测下来ping_interval25秒、timeout60秒比较稳。为什么是25秒因为大部分反向代理的空闲超时是60秒25秒发一次ping能保证连接始终活跃。为什么timeout是60秒给网络抖动留两次重试机会。// 前端心跳实现 const HEARTBEAT_INTERVAL 25000; const MAX_MISSED 2; function startHeartbeat(ws, onDead) { let missed 0; const timer setInterval(() { if (ws.readyState ! WebSocket.OPEN) return; if (missed MAX_MISSED) { clearInterval(timer); onDead(); return; } ws.send(JSON.stringify({ type: ping, ts: Date.now() })); missed; }, HEARTBEAT_INTERVAL); ws.addEventListener(message, (e) { const msg JSON.parse(e.data); if (msg.type pong) missed 0; }); return () clearInterval(timer); }提示心跳消息不要走业务消息通道单独用type: ping/pong标记服务端收到后直接回不进LLM不消耗token。3.4 流式输出的实现与前端渲染LLM生成回复是流式的一个字一个字往外吐。后端要把这个流通过WebSocket推给前端前端要边收边渲染。后端用async for迭代LLM的流式响应每拿到一个chunk就websocket.send_json推出去async def stream_dialogue(session, user_input, websocket): await websocket.send_json({type: start, msg_id: new_id()}) buffer async for chunk in llm_client.stream(dialogue_prompt(session, user_input)): buffer chunk await websocket.send_json({ type: delta, msg_id: current_id, content: chunk }) await websocket.send_json({type: end, msg_id: current_id}) session.append_message(assistant, buffer)前端收到delta就往当前消息的content后面追加收到end就锁定这条消息。这里有个性能坑如果每个chunk都触发一次React的setState高频流式输出下会卡。解决办法是用requestAnimationFrame做节流或者用useRef攒一批再更新。const bufferRef useRef(); const rafRef useRef(null); function onDelta(chunk) { bufferRef.current chunk; if (!rafRef.current) { rafRef.current requestAnimationFrame(() { setMessages(prev updateLast(prev, bufferRef.current)); rafRef.current null; }); } }实测下来节流后CPU占用从40%降到12%滚动也顺滑了。3.5 评估Agent的Prompt设计评估Agent是整个项目的灵魂。它要输出结构化的评估结果不能是自由文本。我用的是JSON Schema约束 few-shot示例的方式。System Prompt大致是这样你是一名英语教学评估专家。用户会给你一句英语表达和当前场景。 你需要输出JSON格式的评估结果包含以下字段 - grammar_errors: 数组每个元素包含 original, corrected, reason - vocabulary_suggestions: 数组每个元素包含 original, better, reason - fluency_score: 0-100的整数 - overall_comment: 一句话点评 只输出JSON不要任何额外文字。关键点在于只输出JSON这句。不加这句模型经常在JSON前后加好的以下是评估结果之类的废话解析就崩了。另外few-shot示例里要覆盖没有错误的情况否则模型会硬找错误。评估Agent的调用是异步的不阻塞对话流。用户发完消息对话Agent先流式回复评估Agent在后台跑跑完通过WebSocket推一个eval_result消息给前端前端在侧边栏展示。4. 完整实操流程与关键参数计算4.1 环境搭建与依赖安装从零开始的话先把环境搭起来。Python建议3.11以上Node建议20以上。# 后端 python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn[standard] websockets python-dotenv pydantic httpx # 前端 npm create vitelatest frontend -- --template react cd frontend npm install npm install react-uplot uplotuvicorn[standard]这个写法要注意方括号里的standard会额外装上uvloop和httptools性能比裸装uvicorn高不少。我实测同样并发下带standard的版本QPS高约30%。4.2 LLM调用的封装与重试策略LLM调用不能裸调必须封装。封装要处理三件事超时、重试、降级。import httpx from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, base_url, api_key, model): self.client httpx.AsyncClient(timeout60.0) self.base_url base_url self.api_key api_key self.model model retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max8)) async def stream(self, messages): async with self.client.stream( POST, f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, json{model: self.model, messages: messages, stream: True} ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if line.startswith(data: ) and line ! data: [DONE]: yield parse_chunk(line)重试策略用指数退避multiplier1, min1, max8意味着第一次等1秒第二次2秒第三次4秒最多8秒。为什么不用固定间隔因为LLM服务端的限流通常是短时突发指数退避能更快错开。超时设60秒是因为流式响应可能持续很久但单次连接建立不能等太久。这里httpx.AsyncClient的timeout是连接超时不是总时长流式读取不受这个限制。4.3 会话状态的数据结构设计会话状态是整个系统的中枢。我设计成这样一个结构from pydantic import BaseModel from typing import List, Optional from datetime import datetime class Message(BaseModel): role: str # user / assistant / system content: str timestamp: datetime eval_result: Optional[dict] None class Session(BaseModel): session_id: str user_id: str scenario_id: str scenario_config: dict messages: List[Message] [] created_at: datetime last_active: datetime turn_count: int 0 total_tokens: int 0turn_count和total_tokens是用来做成本控制和场景结束判断的。一个场景建议控制在15-20轮对话超过就提示用户场景即将结束。token累计到一定量可以触发告警防止某个用户把额度跑爆。4.4 场景配置的Schema设计场景配置决定了Agent的人设和对话走向。我用JSON定义方便扩展{ id: coffee_shop, name: 在咖啡店点单, role: barista, role_desc: 你是一名友好的咖啡店店员说话轻松随意, opening: Hi there! What can I get for you today?, goals: [点一杯饮品, 选择大小和温度, 完成付款], vocab_hints: [latte, cappuccino, medium, to go], max_turns: 15 }goals字段很关键它让对话有方向感。对话Agent的System Prompt里会带上这些目标模型会主动引导用户完成。vocab_hints是给评估Agent用的评估时会优先关注这些词的使用情况。4.5 前端WebSocket Hook的封装前端把WebSocket逻辑封装成一个自定义Hook组件里只管用function useChatSocket(sessionId, token) { const [messages, setMessages] useState([]); const [status, setStatus] useState(connecting); const wsRef useRef(null); useEffect(() { const ws new WebSocket( ws://localhost:8000/ws/chat/${sessionId}?token${token} ); wsRef.current ws; ws.onopen () setStatus(connected); ws.onclose () setStatus(disconnected); ws.onmessage (e) { const msg JSON.parse(e.data); handleIncoming(msg, setMessages); }; const stopHeartbeat startHeartbeat(ws, () { setStatus(reconnecting); // 触发重连逻辑 }); return () { stopHeartbeat(); ws.close(); }; }, [sessionId, token]); const send useCallback((text) { wsRef.current?.send(JSON.stringify({ type: user_input, content: text })); }, []); return { messages, status, send }; }这个Hook处理了连接、心跳、消息分发、清理。组件里只需要const { messages, status, send } useChatSocket(...)干净。5. 常见问题与排查技巧实录5.1 WebSocket连上了但收不到消息这是最高频的问题。现象是前端onopen触发了readyState是1但onmessage一直不触发。排查思路按这个顺序走检查服务端是否真的发了。在websocket.send_json前后打日志确认代码执行到了。检查消息格式。如果服务端发的是send_text但前端按JSON解析会静默失败。统一用send_json。检查中间层。如果前面有Nginx确认proxy_read_timeout和proxy_buffering off配置了。Nginx默认会缓冲WebSocket消息导致前端收不到。检查心跳。如果服务端有心跳但前端没回pong服务端可能主动关了连接。我遇到过一次排查了两小时最后发现是Nginx的proxy_buffering没关。加上proxy_buffering off;立刻就好了。5.2 LLM流式输出中断流式输出跑到一半突然断了前端显示半句话。原因通常是LLM服务端的连接被掐了或者网络抖动。解决办法是断点续传。后端记录已经推送的buffer如果流中断用已生成的内容作为上下文重新请求让模型接着写。Prompt里加一句请从以下内容继续不要重复。async def stream_with_resume(session, prompt, websocket, max_retry2): buffer for attempt in range(max_retry 1): try: async for chunk in llm_client.stream( build_prompt(prompt, buffer) ): buffer chunk await websocket.send_json({type: delta, content: chunk}) break except Exception as e: if attempt max_retry: await websocket.send_json({type: error, msg: str(e)}) raise logger.warning(fstream interrupted, retry {attempt1})5.3 并发场景下的会话串号这个坑很隐蔽。多个用户同时对话A的回复跑到了B的窗口里。原因是会话状态用了全局变量或者WebSocket连接和会话的映射关系搞错了。正确做法是每个WebSocket连接绑定一个独立的会话对象会话对象存在连接池里用session_id做key。绝对不要用全局dict存当前会话。class SessionManager: def __init__(self): self._sessions: dict[str, Session] {} self._lock asyncio.Lock() async def get_or_create(self, session_id, user_id): async with self._lock: if session_id not in self._sessions: self._sessions[session_id] Session( session_idsession_id, user_iduser_id, ... ) return self._sessions[session_id]加锁是因为get_or_create可能被并发调用不加锁会创建两个会话对象。5.4 常见问题速查表现象可能原因排查方法解决连上收不到消息Nginx缓冲看Nginx配置proxy_buffering off流式中断LLM连接超时看后端日志断点续传重试会话串号全局状态检查会话存储连接绑定独立会话心跳失效前端没回pong抓包看帧补pong处理评估不返回JSON解析失败打印原始输出加只输出JSON约束前端卡顿高频setStatePerformance面板rAF节流重连后丢上下文未持久化检查断开逻辑捕获Disconnect持久化5.5 几个我踩过的坑坑一Prompt里带用户输入没做转义。用户输入里如果有{}或者特殊字符格式化Prompt的时候会报错。解决办法是用str.format的替代方案或者用f-string但把用户输入单独拼接。坑二token计数不准。不同模型的tokenizer不一样用tiktoken算OpenAI的准算别的模型就不准。我的做法是粗略估算英文按4字符1token中文按1.5字符1token留20%余量。坑三评估Agent拖慢整体响应。一开始评估是同步的用户发完消息要等评估跑完才看到回复。改成异步后体验好很多但要注意评估结果推送时用户可能已经发了下一条消息前端要能正确处理乱序。坑四场景结束判断太生硬。按轮数硬切用户正聊到兴头上突然结束体验很差。后来改成轮数接近上限时对话Agent会主动引导收尾比如Thanks for coming! Have a great day!自然结束。6. 性能优化与扩展方向6.1 并发能力怎么扛Agent项目最怕的就是并发。每个会话都要调LLMLLM的响应时间又不稳定。我的优化思路是分层削峰第一层WebSocket连接本身很轻一个进程挂几千个没问题。第二层LLM调用用异步不阻塞事件循环。第三层加一个请求队列超过并发上限的请求排队等待而不是直接拒绝。队列用asyncio.Queue实现消费者数量控制在LLM服务端能承受的范围内。我实测下来单实例配4个消费者能稳定支撑200个活跃会话。class LLMQueue: def __init__(self, max_concurrent4): self.queue asyncio.Queue() self.semaphore asyncio.Semaphore(max_concurrent) async def submit(self, coro): async with self.semaphore: return await coro6.2 缓存能省不少钱相同场景的开场白、常见问题的回复这些是可以缓存的。用functools.lru_cache或者Redis都行。我统计过缓存命中率大概在15%左右能省下不少token。但要注意缓存key要包含场景ID和对话轮次不能只按用户输入缓存否则不同场景下同样的输入会返回错误的回复。6.3 后续可以扩展的方向这个项目目前是单场景对话后续可以扩展成多场景串联的学习路径。比如咖啡店点单 → 问路 → 酒店入住形成一个完整的旅行英语学习链路。评估Agent可以跨场景追踪用户的薄弱点生成个性化的复习计划。另一个方向是语音输入。现在用户是打字如果接入语音识别就能做真正的口语训练。WebSocket传音频流后端做ASR再走现在的对话流程。技术上可行主要是ASR的延迟和准确率要调。最后再分享一个小技巧评估Agent的Prompt里加上用中文解释错误原因对中文母语的学习者友好很多。纯英文解释语法错误很多人看不懂反而增加了学习负担。这个细节看起来小但实际使用中反馈很好。