1. 为什么我要做这个英语情景教学Agent去年帮几个朋友准备雅思口语我发现一个特别尴尬的事市面上大部分AI口语陪练本质上就是个套了壳的聊天机器人。你问它点餐场景怎么练它给你列一堆句型你说模拟一下机场值机它倒是配合演两句但演着演着就跳戏了要么突然开始讲语法要么把上一轮的角色设定忘得一干二净。问题的根子不在模型能力而在架构设计。一个真正能用的情景教学Agent需要同时解决三件事角色一致性不能演着演着变成老师、场景状态管理对话进行到哪一步了、用户哪些词用错了、实时反馈发音、语法、用词建议要即时推给前端。这三件事用传统的请求-响应模式做体验会非常割裂。所以我决定从零搭一个。技术选型上后端用FastAPI Python前端用React通信层用WebSocket做双向实时推送。这套组合不是拍脑袋定的——FastAPI原生支持异步和WebSocketPython生态里做Agent编排的库最成熟React的状态管理配合WebSocket能做出对话流式输出实时纠错的效果。这篇文章我会把整个开发过程拆开讲包括架构决策的逻辑、WebSocket心跳和重连的坑、Agent状态机的设计、以及前后端联调时踩过的那些雷。适合有一定Python和React基础、想动手做一个完整Agent项目的朋友。代码我会给关键片段但更重要的是讲清楚为什么这么设计。2. 整体架构设计与技术选型逻辑2.1 为什么是FastAPI而不是Django或Flask先说后端框架的选择。Django自带ORM、Admin、认证做CRUD项目确实快但它的WebSocket支持依赖Channels而Channels本质上是在Django的同步体系上打补丁配置起来层级多、调试麻烦。Flask更轻但异步支持一直是个短板做流式输出和并发连接管理会很难受。FastAPI的优势在这个场景里特别明显原生async/awaitAgent调用大模型API是IO密集型操作异步能大幅提升并发能力。我实测过同样配置下FastAPI处理50个并发WebSocket连接响应延迟比Flaskgevent方案低40%左右。WebSocket是一等公民app.websocket(/ws)装饰器直接支持不需要额外中间件。Pydantic做数据校验Agent的消息格式、场景配置、用户状态都可以用Pydantic模型定义前后端契约清晰。项目目录结构我参考了FastAPI官方推荐的分层方式但做了简化english-agent/ ├── app/ │ ├── main.py # 应用入口挂载路由和WebSocket │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── agent_engine.py # Agent核心逻辑 │ ├── models/ │ │ ├── scene.py # 场景数据模型 │ │ └── message.py # 消息模型 │ ├── services/ │ │ ├── llm_service.py # 大模型调用封装 │ │ └── feedback.py # 纠错反馈逻辑 │ └── ws/ │ └── connection.py # WebSocket连接管理 ├── frontend/ # React前端 └── requirements.txt这个结构的好处是职责分离Agent引擎不关心WebSocket怎么传WebSocket层不关心Agent怎么决策两边通过消息模型解耦。后面要换通信方式比如加SSE支持或者换模型改动范围可控。2.2 Agent状态机的核心设计思路这是整个项目最核心的部分。一个情景教学Agent和普通聊天机器人的区别在于它需要维护一个对话状态知道当前处于场景的哪个阶段、用户的表现如何、下一步该引导什么。我设计了一个轻量级状态机每个场景定义为一组状态节点# models/scene.py from pydantic import BaseModel from typing import List, Optional class SceneState(BaseModel): state_id: str prompt_template: str # 该状态下Agent的提示词模板 expected_keywords: List[str] # 期望用户使用的关键词 next_states: List[str] # 可能的下一状态 max_turns: int 5 # 该状态最多持续几轮 class Scene(BaseModel): scene_id: str name: str description: str initial_state: str states: dict[str, SceneState] system_prompt: str # 场景级系统提示词以机场值机场景为例状态流转大致是greeting打招呼→check_in办理值机→baggage行李托运→seat_selection选座→boarding登机口确认→end。每个状态节点有自己的提示词模板Agent在当前状态下只关注该阶段的任务。这样做的好处是避免角色漂移——Agent不会在值机阶段突然开始教语法因为它的提示词里明确限定了当前任务。状态转移的触发条件有两个一是用户完成了当前阶段的关键信息交换比如提供了航班号二是达到了max_turns上限。转移逻辑放在agent_engine.py里# core/agent_engine.py class AgentEngine: def __init__(self, scene: Scene, llm_service): self.scene scene self.llm llm_service self.current_state scene.initial_state self.turn_count 0 self.history [] async def process_message(self, user_input: str) - dict: state self.scene.states[self.current_state] # 构建当前状态的提示词 prompt self._build_prompt(state, user_input) # 调用LLM response await self.llm.generate(prompt, self.history) # 判断是否转移状态 self.turn_count 1 if self._should_transition(state, user_input): self.current_state self._pick_next_state(state) self.turn_count 0 # 生成纠错反馈 feedback self._analyze_feedback(user_input) return { reply: response, state: self.current_state, feedback: feedback }这里有个关键决策状态转移判断不依赖LLM。我试过让LLM自己判断当前阶段是否结束结果很不稳定有时候用户只说了一句Hello它就认为打招呼阶段完成了。后来改成基于关键词匹配轮次计数的规则判断虽然笨但可靠。LLM只负责生成回复内容不负责流程控制。2.3 WebSocket通信层的设计考量为什么不用HTTP轮询或者SSE三个原因第一双向实时性。情景教学需要Agent主动推送消息比如用户沉默太久Agent要主动引导HTTP轮询做不到真正的服务端推送。第二流式输出体验。大模型生成回复是逐token的用WebSocket可以做到打字机效果用户感知的响应速度快很多。SSE也能做流式但SSE是单向的用户消息还得走HTTP两套连接管理起来更复杂。第三状态同步。对话过程中Agent的状态变化、纠错反馈、场景进度都需要实时同步到前端。WebSocket天然适合这种多类型消息的推送。连接管理我写了一个简单的ConnectionManager# ws/connection.py from fastapi import WebSocket from typing import Dict class ConnectionManager: def __init__(self): self.active: Dict[str, WebSocket] {} self.engines: Dict[str, AgentEngine] {} async def connect(self, session_id: str, ws: WebSocket): await ws.accept() self.active[session_id] ws async def disconnect(self, session_id: str): self.active.pop(session_id, None) self.engines.pop(session_id, None) async def send(self, session_id: str, data: dict): ws self.active.get(session_id) if ws: await ws.send_json(data)每个会话对应一个session_id绑定一个独立的AgentEngine实例。这样多个用户可以同时练习不同场景互不干扰。3. WebSocket心跳机制与断线重连的实操细节3.1 心跳机制为什么必须做WebSocket连接在理想网络下可以保持很久但现实是移动网络切换、代理超时、浏览器标签页休眠都会导致连接悄无声息地断开。最坑的是服务端和客户端都不知道对方已经挂了——服务端以为连接还在继续往一个死连接发消息客户端以为还连着用户发消息石沉大海。心跳机制就是解决这个问题的。原理很简单客户端每隔N秒发一个ping服务端收到后回一个pong。如果连续M次没收到pong客户端就认为连接断了主动重连。N和M怎么定我试过几个组合心跳间隔超时次数效果30秒2次断线检测慢用户感知明显15秒3次平衡较好推荐5秒2次检测快但心跳包太频繁浪费带宽最终选了15秒间隔、3次超时。这个配置下最坏情况45秒能检测到断线对教学场景来说可以接受。前端实现// frontend/src/hooks/useWebSocket.js const HEARTBEAT_INTERVAL 15000; const MAX_MISSED 3; function useWebSocket(sessionId) { const wsRef useRef(null); const missedRef useRef(0); const timerRef useRef(null); const connect useCallback(() { const ws new WebSocket(ws://localhost:8000/ws/${sessionId}); wsRef.current ws; ws.onopen () { missedRef.current 0; startHeartbeat(); }; ws.onmessage (event) { const data JSON.parse(event.data); if (data.type pong) { missedRef.current 0; return; } // 处理正常业务消息 handleMessage(data); }; ws.onclose () { stopHeartbeat(); // 指数退避重连 setTimeout(connect, Math.min(1000 * 2 ** retryCount, 30000)); }; }, [sessionId]); const startHeartbeat () { timerRef.current setInterval(() { if (missedRef.current MAX_MISSED) { wsRef.current?.close(); return; } wsRef.current?.send(JSON.stringify({ type: ping })); missedRef.current 1; }, HEARTBEAT_INTERVAL); }; }服务端对应处理# main.py app.websocket(/ws/{session_id}) async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.connect(session_id, ws) try: while True: data await ws.receive_json() if data.get(type) ping: await ws.send_json({type: pong}) continue # 正常业务处理 result await engine.process_message(data[content]) await ws.send_json({type: reply, **result}) except WebSocketDisconnect: await manager.disconnect(session_id)注意心跳包不要走业务消息通道。我一开始图省事把ping混在普通消息里结果业务逻辑里到处要判断if type ping代码很脏。后来改成独立的消息类型业务层完全不用关心心跳。3.2 断线重连后的状态恢复重连本身不难难的是重连后怎么恢复对话状态。用户可能正在场景中间断线重连后不能从头开始。我的方案是服务端的AgentEngine实例不随连接断开而销毁而是保留一段时间比如5分钟。重连时用同一个session_id直接复用之前的引擎实例。async def disconnect(self, session_id: str): self.active.pop(session_id, None) # 不立即销毁engine延迟清理 asyncio.create_task(self._delayed_cleanup(session_id)) async def _delayed_cleanup(self, session_id: str): await asyncio.sleep(300) # 5分钟 if session_id not in self.active: self.engines.pop(session_id, None)前端重连成功后发一个sync消息服务端把当前状态、历史消息摘要推回去前端恢复UI。这里踩过一个坑重连时如果用户正在等待Agent回复重连后这条回复可能丢了。解决办法是服务端维护一个消息队列未确认送达的消息在重连后重发。我加了个简单的ack机制前端收到消息后回一个ack服务端收到ack才从队列移除。3.3 并发连接的性能考量AI Agent怎么扛并发是个高频问题。我的实测数据单台4核8G的机器FastAPI uvicorn4个worker每个WebSocket连接保持长连接但大部分时间空闲能稳定支撑800-1000个并发连接。瓶颈不在WebSocket本身而在LLM API的调用。如果每个用户每10秒发一条消息1000个用户就是100 QPS的LLM调用这个量级需要做限流和排队。我的做法是每个会话的LLM调用串行化同一个用户的消息按顺序处理全局用一个信号量限制同时进行的LLM调用数比如50个超出的请求进入队列前端显示思考中状态import asyncio llm_semaphore asyncio.Semaphore(50) async def generate_with_limit(prompt, history): async with llm_semaphore: return await llm_service.generate(prompt, history)这样即使瞬时请求很多也不会把LLM API打爆用户体验是稍微等一下而不是直接报错。4. React前端的核心实现与状态管理4.1 对话流式输出的实现流式输出是提升体验的关键。用户发完消息后Agent的回复逐字出现比等3秒突然蹦出一整段要自然得多。实现上服务端把LLM的流式响应通过WebSocket逐块推送async def stream_reply(self, session_id: str, prompt: str): async for chunk in self.llm.stream(prompt): await manager.send(session_id, { type: stream, content: chunk }) await manager.send(session_id, {type: stream_end})前端维护一个正在流式输出的消息对象每收到一个chunk就追加内容const [messages, setMessages] useState([]); const [streaming, setStreaming] useState(null); // 收到stream消息 case stream: setStreaming(prev (prev || ) data.content); break; case stream_end: setMessages(prev [...prev, { role: assistant, content: streaming }]); setStreaming(null); break;实操心得流式输出时不要每个chunk都触发React重渲染会卡。我用了一个buffer每50ms批量更新一次state。或者用useRef存内容配合requestAnimationFrame更新。实测下来50ms的批量更新在视觉上完全看不出延迟但渲染压力小很多。4.2 纠错反馈的UI呈现纠错反馈是情景教学Agent区别于普通聊天的核心功能。我的设计是Agent的回复正常显示在对话流里纠错反馈以侧边卡片的形式呈现不打断对话节奏。反馈分三类语法错误标红给出修正建议用词建议标黄给出更地道的表达亮点表达标绿正向激励数据结构{ type: feedback, items: [ { level: error, original: I go to airport yesterday, suggestion: I went to the airport yesterday, explanation: 过去时间应使用过去式 } ] }前端用不同颜色的卡片渲染用户可以选择查看详情展开解释。这里有个设计决策反馈不要实时弹出打断用户。我一开始做成实时弹窗结果用户说到一半就被打断体验很差。后来改成对话轮次结束后统一展示用户可以在侧边栏随时查看历史反馈。4.3 React状态管理的取舍这个项目的状态不算复杂但有几个特殊点WebSocket连接状态、对话消息列表、流式输出中的临时消息、纠错反馈列表、场景进度。用Redux有点重用Context又担心性能。最终方案是useReducer Context把WebSocket相关的状态集中管理const initialState { connected: false, messages: [], streaming: null, feedback: [], sceneState: null, }; function reducer(state, action) { switch (action.type) { case WS_CONNECTED: return { ...state, connected: true }; case ADD_MESSAGE: return { ...state, messages: [...state.messages, action.payload] }; case STREAM_CHUNK: return { ...state, streaming: (state.streaming || ) action.payload }; case STREAM_END: return { ...state, messages: [...state.messages, { role: assistant, content: state.streaming }], streaming: null, }; case ADD_FEEDBACK: return { ...state, feedback: [...state.feedback, ...action.payload] }; default: return state; } }这个方案的好处是状态变更可预测每个action对应一个明确的用户操作或服务端事件。调试时用Redux DevTools也能接入虽然没用Redux但reducer的逻辑是一样的。关于有没有通用React开发标准这个问题我的看法是没有银弹。小项目用useStateContext就够了中型项目useReducerContext大型项目才需要Redux或Zustand。关键是状态的作用域要清晰——WebSocket状态是全局的用Context单个消息组件的展开/收起状态是局部的用useState。5. Agent核心逻辑从提示词工程到状态流转5.1 场景提示词的分层设计提示词是Agent的灵魂。我试过一版把所有的场景规则、角色设定、纠错要求都塞进一个system prompt结果模型经常顾此失彼。后来改成三层结构第一层全局系统提示词。定义Agent的基本身份和行为准则。You are an English conversation partner for scenario-based learning. Your role is to play a specific character in the given scenario. Never break character. Never explain grammar unless explicitly asked. Keep responses concise (2-3 sentences) to encourage user participation.第二层场景级提示词。定义当前场景的背景、角色、目标。Scenario: Airport Check-in Your role: A check-in counter staff Users role: A passenger Goal: Help the user complete check-in, baggage drop, and seat selection.第三层状态级提示词。定义当前阶段的具体任务。Current stage: Baggage Drop Task: Ask about the users baggage, inform them of weight limits, and guide them to the oversized baggage counter if needed. Expected user keywords: luggage, suitcase, overweight, carry-on三层拼起来作为完整的prompt传给LLM。这样设计的好处是每一层的职责清晰改场景不用动全局设定改阶段不用动场景设定。5.2 状态转移的规则引擎前面提到状态转移不依赖LLM而是用规则判断。规则引擎的核心逻辑def _should_transition(self, state: SceneState, user_input: str) - bool: # 规则1达到最大轮次 if self.turn_count state.max_turns: return True # 规则2用户输入包含足够的期望关键词 matched sum(1 for kw in state.expected_keywords if kw.lower() in user_input.lower()) if matched len(state.expected_keywords) * 0.6: return True # 规则3用户明确表示进入下一阶段 transition_signals [next, whats next, okay, got it] if any(sig in user_input.lower() for sig in transition_signals): return True return False_pick_next_state的逻辑更简单如果当前状态只有一个下一状态直接去如果有多个根据用户输入的关键词匹配度选最合适的。踩坑记录一开始我用LLM来判断状态转移prompt里写如果用户完成了当前阶段的任务返回TRANSITION。结果模型经常在用户只说了一句OK的时候就返回TRANSITION导致场景推进太快。后来改成规则引擎虽然不够智能但稳定可控。教学场景里流程的可预测性比智能性更重要。5.3 纠错反馈的实现细节纠错反馈我用了一个独立的LLM调用和对话生成分开。为什么不分在一次调用里因为对话生成需要低延迟用户等着回复纠错分析可以稍微慢一点对话结束后展示。分开调用可以让对话生成用更快的模型纠错用更准的模型。纠错的prompt设计Analyze the following user input in an English learning context. Identify: 1. Grammar errors (tense, agreement, article usage) 2. Word choice issues (unnatural or non-idiomatic expressions) 3. Positive highlights (good expressions worth praising) Return JSON format: { errors: [{original: , suggestion: , explanation: }], highlights: [{expression: , reason: }] } User input: {user_input} Context: {scene_context}这里有个细节纠错要考虑场景上下文。比如在机场值机场景里用户说I want to check in这是完全正确的但如果用户说I want to check in my luggage在值机阶段可能就不太合适应该先办值机再托运行李。所以纠错prompt里要带上场景上下文。反馈的展示时机也很重要。我的做法是对话轮次结束后如果检测到错误在侧边栏显示一个提示徽章用户点击才展开详情。这样既不打断对话又保证了反馈的可及性。6. 常见问题与排查技巧实录6.1 WebSocket连接问题速查现象可能原因排查方法解决方案连接建立后立即断开服务端未accept或异常查看服务端日志确保await ws.accept()在try块内连接但不接收消息消息格式不匹配浏览器Network面板看WS帧统一JSON格式加type字段心跳正常但业务消息丢失消息处理阻塞检查是否有同步阻塞操作耗时操作放异步任务重连后状态丢失engine被提前销毁检查cleanup逻辑延迟销毁重连时复用多标签页互相干扰session_id冲突检查session生成逻辑每个标签页独立session_id6.2 LLM调用超时与降级LLM API偶尔会超时或限流必须有降级方案async def generate_with_fallback(prompt, history): try: return await asyncio.wait_for( llm_service.generate(prompt, history), timeout10.0 ) except asyncio.TimeoutError: # 降级返回预设的引导语 return Im sorry, could you repeat that? I didnt quite catch it. except RateLimitError: # 限流排队重试 await asyncio.sleep(2) return await generate_with_fallback(prompt, history)实操心得降级回复不要用系统繁忙请稍后重试这种话会破坏场景沉浸感。用场景内的自然语言比如值机场景里说Sorry, the system is a bit slow today. Could you say that again?用户完全感知不到是技术问题。6.3 前端白屏与启动问题React Native项目启动白屏是个高频问题虽然这个项目是Web React但类似的问题也存在。常见原因WebSocket连接在组件挂载前发起把连接逻辑放在useEffect里确保组件已挂载。初始状态未处理messages初始为null而不是[]渲染时报错。所有数组状态都要给空数组初始值。异步数据未做loading态场景配置从服务端拉取拉取完成前要有loading UI否则渲染undefined。6.4 场景配置的热更新开发过程中经常要调整场景的提示词和状态定义。如果每次都要重启服务效率太低。我的做法是场景配置存在JSON文件里加一个/reload接口调用后重新加载配置。app.post(/admin/reload-scenes) async def reload_scenes(): scene_loader.reload() return {status: ok, scenes: len(scene_loader.scenes)}前端加一个隐藏的调试按钮点击后调这个接口。这样调提示词的时候不用重启改完JSON直接reload几秒钟就能看到效果。6.5 关于Agent安全的几点考虑Agent安全是个容易被忽视的问题。这个项目里我做了几层防护输入长度限制单条消息不超过2000字符防止prompt注入攻击。场景边界限制Agent的system prompt里明确限定只讨论当前场景相关话题用户如果试图把话题引到其他领域Agent会礼貌地拉回来。输出过滤对LLM的输出做基本的内容检查过滤掉不适当的表达。会话隔离每个session_id对应独立的engine实例用户之间不能互相访问对话历史。这些措施不能保证100%安全但能挡住大部分常见问题。如果要做生产级应用还需要加更严格的审核和日志。7. 我在这套架构上踩过的坑和后续扩展方向先说几个印象深刻的坑。第一个坑是WebSocket的消息顺序。我一开始没做消息队列用户快速连发两条消息服务端的处理顺序可能和发送顺序不一致。后来在服务端给每个session加了一个asyncio.Queue消息先入队再按序处理问题解决。第二个坑是LLM的流式输出和WebSocket的缓冲。uvicorn默认会对WebSocket消息做缓冲导致流式输出的chunk不是实时到达前端的。需要在发送时加await ws.send_json(data)后跟一个await asyncio.sleep(0)让出控制权或者配置uvicorn的ws_ping_interval参数。第三个坑是React的闭包陷阱。WebSocket的onmessage回调里如果直接引用state拿到的永远是初始值。必须用useRef或者useReducer的dispatch来避免闭包问题。这个坑我调了大半天才定位到。后续扩展方向我目前想到几个多模态输入加语音输入用Whisper做语音识别这样更接近真实的口语练习场景。场景编辑器做一个可视化的场景配置界面非技术人员也能创建新场景。学习报告积累用户的对话数据生成学习报告指出高频错误和进步趋势。多Agent协作一个场景里可以有多个角色比如值机场景里有值机员和安检员用多个Agent实例分别扮演。这套架构的核心思路是分层解耦通信层、Agent逻辑层、场景配置层各自独立任何一层要替换或扩展都不会影响其他层。这也是我从这个项目里得到的最大的架构经验——不要把所有逻辑塞进一个地方哪怕一开始看起来更简单。 
长河君 · 2026/11/2 9:49:37 · 阅读 1.2k FastAPI+React+WebSocket：构建英语情景教学Agent的架构设计与实战
1. 为什么我要做这个英语情景教学Agent去年帮几个朋友准备雅思口语我发现一个特别尴尬的事市面上大部分AI口语陪练本质上就是个套了壳的聊天机器人。你问它点餐场景怎么练它给你列一堆句型你说模拟一下机场值机它倒是配合演两句但演着演着就跳戏了要么突然开始讲语法要么把上一轮的角色设定忘得一干二净。问题的根子不在模型能力而在架构设计。一个真正能用的情景教学Agent需要同时解决三件事角色一致性不能演着演着变成老师、场景状态管理对话进行到哪一步了、用户哪些词用错了、实时反馈发音、语法、用词建议要即时推给前端。这三件事用传统的请求-响应模式做体验会非常割裂。所以我决定从零搭一个。技术选型上后端用FastAPI Python前端用React通信层用WebSocket做双向实时推送。这套组合不是拍脑袋定的——FastAPI原生支持异步和WebSocketPython生态里做Agent编排的库最成熟React的状态管理配合WebSocket能做出对话流式输出实时纠错的效果。这篇文章我会把整个开发过程拆开讲包括架构决策的逻辑、WebSocket心跳和重连的坑、Agent状态机的设计、以及前后端联调时踩过的那些雷。适合有一定Python和React基础、想动手做一个完整Agent项目的朋友。代码我会给关键片段但更重要的是讲清楚为什么这么设计。2. 整体架构设计与技术选型逻辑2.1 为什么是FastAPI而不是Django或Flask先说后端框架的选择。Django自带ORM、Admin、认证做CRUD项目确实快但它的WebSocket支持依赖Channels而Channels本质上是在Django的同步体系上打补丁配置起来层级多、调试麻烦。Flask更轻但异步支持一直是个短板做流式输出和并发连接管理会很难受。FastAPI的优势在这个场景里特别明显原生async/awaitAgent调用大模型API是IO密集型操作异步能大幅提升并发能力。我实测过同样配置下FastAPI处理50个并发WebSocket连接响应延迟比Flaskgevent方案低40%左右。WebSocket是一等公民app.websocket(/ws)装饰器直接支持不需要额外中间件。Pydantic做数据校验Agent的消息格式、场景配置、用户状态都可以用Pydantic模型定义前后端契约清晰。项目目录结构我参考了FastAPI官方推荐的分层方式但做了简化english-agent/ ├── app/ │ ├── main.py # 应用入口挂载路由和WebSocket │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── agent_engine.py # Agent核心逻辑 │ ├── models/ │ │ ├── scene.py # 场景数据模型 │ │ └── message.py # 消息模型 │ ├── services/ │ │ ├── llm_service.py # 大模型调用封装 │ │ └── feedback.py # 纠错反馈逻辑 │ └── ws/ │ └── connection.py # WebSocket连接管理 ├── frontend/ # React前端 └── requirements.txt这个结构的好处是职责分离Agent引擎不关心WebSocket怎么传WebSocket层不关心Agent怎么决策两边通过消息模型解耦。后面要换通信方式比如加SSE支持或者换模型改动范围可控。2.2 Agent状态机的核心设计思路这是整个项目最核心的部分。一个情景教学Agent和普通聊天机器人的区别在于它需要维护一个对话状态知道当前处于场景的哪个阶段、用户的表现如何、下一步该引导什么。我设计了一个轻量级状态机每个场景定义为一组状态节点# models/scene.py from pydantic import BaseModel from typing import List, Optional class SceneState(BaseModel): state_id: str prompt_template: str # 该状态下Agent的提示词模板 expected_keywords: List[str] # 期望用户使用的关键词 next_states: List[str] # 可能的下一状态 max_turns: int 5 # 该状态最多持续几轮 class Scene(BaseModel): scene_id: str name: str description: str initial_state: str states: dict[str, SceneState] system_prompt: str # 场景级系统提示词以机场值机场景为例状态流转大致是greeting打招呼→check_in办理值机→baggage行李托运→seat_selection选座→boarding登机口确认→end。每个状态节点有自己的提示词模板Agent在当前状态下只关注该阶段的任务。这样做的好处是避免角色漂移——Agent不会在值机阶段突然开始教语法因为它的提示词里明确限定了当前任务。状态转移的触发条件有两个一是用户完成了当前阶段的关键信息交换比如提供了航班号二是达到了max_turns上限。转移逻辑放在agent_engine.py里# core/agent_engine.py class AgentEngine: def __init__(self, scene: Scene, llm_service): self.scene scene self.llm llm_service self.current_state scene.initial_state self.turn_count 0 self.history [] async def process_message(self, user_input: str) - dict: state self.scene.states[self.current_state] # 构建当前状态的提示词 prompt self._build_prompt(state, user_input) # 调用LLM response await self.llm.generate(prompt, self.history) # 判断是否转移状态 self.turn_count 1 if self._should_transition(state, user_input): self.current_state self._pick_next_state(state) self.turn_count 0 # 生成纠错反馈 feedback self._analyze_feedback(user_input) return { reply: response, state: self.current_state, feedback: feedback }这里有个关键决策状态转移判断不依赖LLM。我试过让LLM自己判断当前阶段是否结束结果很不稳定有时候用户只说了一句Hello它就认为打招呼阶段完成了。后来改成基于关键词匹配轮次计数的规则判断虽然笨但可靠。LLM只负责生成回复内容不负责流程控制。2.3 WebSocket通信层的设计考量为什么不用HTTP轮询或者SSE三个原因第一双向实时性。情景教学需要Agent主动推送消息比如用户沉默太久Agent要主动引导HTTP轮询做不到真正的服务端推送。第二流式输出体验。大模型生成回复是逐token的用WebSocket可以做到打字机效果用户感知的响应速度快很多。SSE也能做流式但SSE是单向的用户消息还得走HTTP两套连接管理起来更复杂。第三状态同步。对话过程中Agent的状态变化、纠错反馈、场景进度都需要实时同步到前端。WebSocket天然适合这种多类型消息的推送。连接管理我写了一个简单的ConnectionManager# ws/connection.py from fastapi import WebSocket from typing import Dict class ConnectionManager: def __init__(self): self.active: Dict[str, WebSocket] {} self.engines: Dict[str, AgentEngine] {} async def connect(self, session_id: str, ws: WebSocket): await ws.accept() self.active[session_id] ws async def disconnect(self, session_id: str): self.active.pop(session_id, None) self.engines.pop(session_id, None) async def send(self, session_id: str, data: dict): ws self.active.get(session_id) if ws: await ws.send_json(data)每个会话对应一个session_id绑定一个独立的AgentEngine实例。这样多个用户可以同时练习不同场景互不干扰。3. WebSocket心跳机制与断线重连的实操细节3.1 心跳机制为什么必须做WebSocket连接在理想网络下可以保持很久但现实是移动网络切换、代理超时、浏览器标签页休眠都会导致连接悄无声息地断开。最坑的是服务端和客户端都不知道对方已经挂了——服务端以为连接还在继续往一个死连接发消息客户端以为还连着用户发消息石沉大海。心跳机制就是解决这个问题的。原理很简单客户端每隔N秒发一个ping服务端收到后回一个pong。如果连续M次没收到pong客户端就认为连接断了主动重连。N和M怎么定我试过几个组合心跳间隔超时次数效果30秒2次断线检测慢用户感知明显15秒3次平衡较好推荐5秒2次检测快但心跳包太频繁浪费带宽最终选了15秒间隔、3次超时。这个配置下最坏情况45秒能检测到断线对教学场景来说可以接受。前端实现// frontend/src/hooks/useWebSocket.js const HEARTBEAT_INTERVAL 15000; const MAX_MISSED 3; function useWebSocket(sessionId) { const wsRef useRef(null); const missedRef useRef(0); const timerRef useRef(null); const connect useCallback(() { const ws new WebSocket(ws://localhost:8000/ws/${sessionId}); wsRef.current ws; ws.onopen () { missedRef.current 0; startHeartbeat(); }; ws.onmessage (event) { const data JSON.parse(event.data); if (data.type pong) { missedRef.current 0; return; } // 处理正常业务消息 handleMessage(data); }; ws.onclose () { stopHeartbeat(); // 指数退避重连 setTimeout(connect, Math.min(1000 * 2 ** retryCount, 30000)); }; }, [sessionId]); const startHeartbeat () { timerRef.current setInterval(() { if (missedRef.current MAX_MISSED) { wsRef.current?.close(); return; } wsRef.current?.send(JSON.stringify({ type: ping })); missedRef.current 1; }, HEARTBEAT_INTERVAL); }; }服务端对应处理# main.py app.websocket(/ws/{session_id}) async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.connect(session_id, ws) try: while True: data await ws.receive_json() if data.get(type) ping: await ws.send_json({type: pong}) continue # 正常业务处理 result await engine.process_message(data[content]) await ws.send_json({type: reply, **result}) except WebSocketDisconnect: await manager.disconnect(session_id)注意心跳包不要走业务消息通道。我一开始图省事把ping混在普通消息里结果业务逻辑里到处要判断if type ping代码很脏。后来改成独立的消息类型业务层完全不用关心心跳。3.2 断线重连后的状态恢复重连本身不难难的是重连后怎么恢复对话状态。用户可能正在场景中间断线重连后不能从头开始。我的方案是服务端的AgentEngine实例不随连接断开而销毁而是保留一段时间比如5分钟。重连时用同一个session_id直接复用之前的引擎实例。async def disconnect(self, session_id: str): self.active.pop(session_id, None) # 不立即销毁engine延迟清理 asyncio.create_task(self._delayed_cleanup(session_id)) async def _delayed_cleanup(self, session_id: str): await asyncio.sleep(300) # 5分钟 if session_id not in self.active: self.engines.pop(session_id, None)前端重连成功后发一个sync消息服务端把当前状态、历史消息摘要推回去前端恢复UI。这里踩过一个坑重连时如果用户正在等待Agent回复重连后这条回复可能丢了。解决办法是服务端维护一个消息队列未确认送达的消息在重连后重发。我加了个简单的ack机制前端收到消息后回一个ack服务端收到ack才从队列移除。3.3 并发连接的性能考量AI Agent怎么扛并发是个高频问题。我的实测数据单台4核8G的机器FastAPI uvicorn4个worker每个WebSocket连接保持长连接但大部分时间空闲能稳定支撑800-1000个并发连接。瓶颈不在WebSocket本身而在LLM API的调用。如果每个用户每10秒发一条消息1000个用户就是100 QPS的LLM调用这个量级需要做限流和排队。我的做法是每个会话的LLM调用串行化同一个用户的消息按顺序处理全局用一个信号量限制同时进行的LLM调用数比如50个超出的请求进入队列前端显示思考中状态import asyncio llm_semaphore asyncio.Semaphore(50) async def generate_with_limit(prompt, history): async with llm_semaphore: return await llm_service.generate(prompt, history)这样即使瞬时请求很多也不会把LLM API打爆用户体验是稍微等一下而不是直接报错。4. React前端的核心实现与状态管理4.1 对话流式输出的实现流式输出是提升体验的关键。用户发完消息后Agent的回复逐字出现比等3秒突然蹦出一整段要自然得多。实现上服务端把LLM的流式响应通过WebSocket逐块推送async def stream_reply(self, session_id: str, prompt: str): async for chunk in self.llm.stream(prompt): await manager.send(session_id, { type: stream, content: chunk }) await manager.send(session_id, {type: stream_end})前端维护一个正在流式输出的消息对象每收到一个chunk就追加内容const [messages, setMessages] useState([]); const [streaming, setStreaming] useState(null); // 收到stream消息 case stream: setStreaming(prev (prev || ) data.content); break; case stream_end: setMessages(prev [...prev, { role: assistant, content: streaming }]); setStreaming(null); break;实操心得流式输出时不要每个chunk都触发React重渲染会卡。我用了一个buffer每50ms批量更新一次state。或者用useRef存内容配合requestAnimationFrame更新。实测下来50ms的批量更新在视觉上完全看不出延迟但渲染压力小很多。4.2 纠错反馈的UI呈现纠错反馈是情景教学Agent区别于普通聊天的核心功能。我的设计是Agent的回复正常显示在对话流里纠错反馈以侧边卡片的形式呈现不打断对话节奏。反馈分三类语法错误标红给出修正建议用词建议标黄给出更地道的表达亮点表达标绿正向激励数据结构{ type: feedback, items: [ { level: error, original: I go to airport yesterday, suggestion: I went to the airport yesterday, explanation: 过去时间应使用过去式 } ] }前端用不同颜色的卡片渲染用户可以选择查看详情展开解释。这里有个设计决策反馈不要实时弹出打断用户。我一开始做成实时弹窗结果用户说到一半就被打断体验很差。后来改成对话轮次结束后统一展示用户可以在侧边栏随时查看历史反馈。4.3 React状态管理的取舍这个项目的状态不算复杂但有几个特殊点WebSocket连接状态、对话消息列表、流式输出中的临时消息、纠错反馈列表、场景进度。用Redux有点重用Context又担心性能。最终方案是useReducer Context把WebSocket相关的状态集中管理const initialState { connected: false, messages: [], streaming: null, feedback: [], sceneState: null, }; function reducer(state, action) { switch (action.type) { case WS_CONNECTED: return { ...state, connected: true }; case ADD_MESSAGE: return { ...state, messages: [...state.messages, action.payload] }; case STREAM_CHUNK: return { ...state, streaming: (state.streaming || ) action.payload }; case STREAM_END: return { ...state, messages: [...state.messages, { role: assistant, content: state.streaming }], streaming: null, }; case ADD_FEEDBACK: return { ...state, feedback: [...state.feedback, ...action.payload] }; default: return state; } }这个方案的好处是状态变更可预测每个action对应一个明确的用户操作或服务端事件。调试时用Redux DevTools也能接入虽然没用Redux但reducer的逻辑是一样的。关于有没有通用React开发标准这个问题我的看法是没有银弹。小项目用useStateContext就够了中型项目useReducerContext大型项目才需要Redux或Zustand。关键是状态的作用域要清晰——WebSocket状态是全局的用Context单个消息组件的展开/收起状态是局部的用useState。5. Agent核心逻辑从提示词工程到状态流转5.1 场景提示词的分层设计提示词是Agent的灵魂。我试过一版把所有的场景规则、角色设定、纠错要求都塞进一个system prompt结果模型经常顾此失彼。后来改成三层结构第一层全局系统提示词。定义Agent的基本身份和行为准则。You are an English conversation partner for scenario-based learning. Your role is to play a specific character in the given scenario. Never break character. Never explain grammar unless explicitly asked. Keep responses concise (2-3 sentences) to encourage user participation.第二层场景级提示词。定义当前场景的背景、角色、目标。Scenario: Airport Check-in Your role: A check-in counter staff Users role: A passenger Goal: Help the user complete check-in, baggage drop, and seat selection.第三层状态级提示词。定义当前阶段的具体任务。Current stage: Baggage Drop Task: Ask about the users baggage, inform them of weight limits, and guide them to the oversized baggage counter if needed. Expected user keywords: luggage, suitcase, overweight, carry-on三层拼起来作为完整的prompt传给LLM。这样设计的好处是每一层的职责清晰改场景不用动全局设定改阶段不用动场景设定。5.2 状态转移的规则引擎前面提到状态转移不依赖LLM而是用规则判断。规则引擎的核心逻辑def _should_transition(self, state: SceneState, user_input: str) - bool: # 规则1达到最大轮次 if self.turn_count state.max_turns: return True # 规则2用户输入包含足够的期望关键词 matched sum(1 for kw in state.expected_keywords if kw.lower() in user_input.lower()) if matched len(state.expected_keywords) * 0.6: return True # 规则3用户明确表示进入下一阶段 transition_signals [next, whats next, okay, got it] if any(sig in user_input.lower() for sig in transition_signals): return True return False_pick_next_state的逻辑更简单如果当前状态只有一个下一状态直接去如果有多个根据用户输入的关键词匹配度选最合适的。踩坑记录一开始我用LLM来判断状态转移prompt里写如果用户完成了当前阶段的任务返回TRANSITION。结果模型经常在用户只说了一句OK的时候就返回TRANSITION导致场景推进太快。后来改成规则引擎虽然不够智能但稳定可控。教学场景里流程的可预测性比智能性更重要。5.3 纠错反馈的实现细节纠错反馈我用了一个独立的LLM调用和对话生成分开。为什么不分在一次调用里因为对话生成需要低延迟用户等着回复纠错分析可以稍微慢一点对话结束后展示。分开调用可以让对话生成用更快的模型纠错用更准的模型。纠错的prompt设计Analyze the following user input in an English learning context. Identify: 1. Grammar errors (tense, agreement, article usage) 2. Word choice issues (unnatural or non-idiomatic expressions) 3. Positive highlights (good expressions worth praising) Return JSON format: { errors: [{original: , suggestion: , explanation: }], highlights: [{expression: , reason: }] } User input: {user_input} Context: {scene_context}这里有个细节纠错要考虑场景上下文。比如在机场值机场景里用户说I want to check in这是完全正确的但如果用户说I want to check in my luggage在值机阶段可能就不太合适应该先办值机再托运行李。所以纠错prompt里要带上场景上下文。反馈的展示时机也很重要。我的做法是对话轮次结束后如果检测到错误在侧边栏显示一个提示徽章用户点击才展开详情。这样既不打断对话又保证了反馈的可及性。6. 常见问题与排查技巧实录6.1 WebSocket连接问题速查现象可能原因排查方法解决方案连接建立后立即断开服务端未accept或异常查看服务端日志确保await ws.accept()在try块内连接但不接收消息消息格式不匹配浏览器Network面板看WS帧统一JSON格式加type字段心跳正常但业务消息丢失消息处理阻塞检查是否有同步阻塞操作耗时操作放异步任务重连后状态丢失engine被提前销毁检查cleanup逻辑延迟销毁重连时复用多标签页互相干扰session_id冲突检查session生成逻辑每个标签页独立session_id6.2 LLM调用超时与降级LLM API偶尔会超时或限流必须有降级方案async def generate_with_fallback(prompt, history): try: return await asyncio.wait_for( llm_service.generate(prompt, history), timeout10.0 ) except asyncio.TimeoutError: # 降级返回预设的引导语 return Im sorry, could you repeat that? I didnt quite catch it. except RateLimitError: # 限流排队重试 await asyncio.sleep(2) return await generate_with_fallback(prompt, history)实操心得降级回复不要用系统繁忙请稍后重试这种话会破坏场景沉浸感。用场景内的自然语言比如值机场景里说Sorry, the system is a bit slow today. Could you say that again?用户完全感知不到是技术问题。6.3 前端白屏与启动问题React Native项目启动白屏是个高频问题虽然这个项目是Web React但类似的问题也存在。常见原因WebSocket连接在组件挂载前发起把连接逻辑放在useEffect里确保组件已挂载。初始状态未处理messages初始为null而不是[]渲染时报错。所有数组状态都要给空数组初始值。异步数据未做loading态场景配置从服务端拉取拉取完成前要有loading UI否则渲染undefined。6.4 场景配置的热更新开发过程中经常要调整场景的提示词和状态定义。如果每次都要重启服务效率太低。我的做法是场景配置存在JSON文件里加一个/reload接口调用后重新加载配置。app.post(/admin/reload-scenes) async def reload_scenes(): scene_loader.reload() return {status: ok, scenes: len(scene_loader.scenes)}前端加一个隐藏的调试按钮点击后调这个接口。这样调提示词的时候不用重启改完JSON直接reload几秒钟就能看到效果。6.5 关于Agent安全的几点考虑Agent安全是个容易被忽视的问题。这个项目里我做了几层防护输入长度限制单条消息不超过2000字符防止prompt注入攻击。场景边界限制Agent的system prompt里明确限定只讨论当前场景相关话题用户如果试图把话题引到其他领域Agent会礼貌地拉回来。输出过滤对LLM的输出做基本的内容检查过滤掉不适当的表达。会话隔离每个session_id对应独立的engine实例用户之间不能互相访问对话历史。这些措施不能保证100%安全但能挡住大部分常见问题。如果要做生产级应用还需要加更严格的审核和日志。7. 我在这套架构上踩过的坑和后续扩展方向先说几个印象深刻的坑。第一个坑是WebSocket的消息顺序。我一开始没做消息队列用户快速连发两条消息服务端的处理顺序可能和发送顺序不一致。后来在服务端给每个session加了一个asyncio.Queue消息先入队再按序处理问题解决。第二个坑是LLM的流式输出和WebSocket的缓冲。uvicorn默认会对WebSocket消息做缓冲导致流式输出的chunk不是实时到达前端的。需要在发送时加await ws.send_json(data)后跟一个await asyncio.sleep(0)让出控制权或者配置uvicorn的ws_ping_interval参数。第三个坑是React的闭包陷阱。WebSocket的onmessage回调里如果直接引用state拿到的永远是初始值。必须用useRef或者useReducer的dispatch来避免闭包问题。这个坑我调了大半天才定位到。后续扩展方向我目前想到几个多模态输入加语音输入用Whisper做语音识别这样更接近真实的口语练习场景。场景编辑器做一个可视化的场景配置界面非技术人员也能创建新场景。学习报告积累用户的对话数据生成学习报告指出高频错误和进步趋势。多Agent协作一个场景里可以有多个角色比如值机场景里有值机员和安检员用多个Agent实例分别扮演。这套架构的核心思路是分层解耦通信层、Agent逻辑层、场景配置层各自独立任何一层要替换或扩展都不会影响其他层。这也是我从这个项目里得到的最大的架构经验——不要把所有逻辑塞进一个地方哪怕一开始看起来更简单。