基于FastAPI+React+WebSocket的英语情景教学Agent架构设计与实现

1. 项目缘起与整体设计思路

1.1 为什么我要做这个英语情景教学Agent

做英语学习类产品的朋友应该都有体会：市面上的口语练习工具，要么是固定脚本的对话机器人，翻来覆去就那几句；要么是纯聊天式的AI对话，聊得挺热闹但完全没有教学结构，用户练了半个月也不知道自己进步在哪。我在带几个朋友练口语的过程中发现，真正有效的练习需要一个有场景、有角色、有反馈闭环的环境——就像把你扔到咖啡店点单、机场值机、酒店投诉这些真实情境里，逼着你用英语把事办成。

这就是我做这个英语情景教学Agent的出发点。它不是又一个聊天框，而是一个能动态生成情景、扮演角色、实时纠错、给出阶段性反馈的教学系统。核心关键词围绕Agent架构、英语情景教学、WebSocket实时通信、React前端、FastAPI后端展开。适合谁看？如果你正在做AI应用开发、想了解Agent怎么落地到具体教学场景、或者单纯想搞明白WebSocket在实时交互系统里怎么用，这篇内容应该能给你一些可直接抄作业的东西。

整个项目从零到一大概花了我三周多的业余时间，踩了不少坑，也推翻过两次架构。下面我把设计思路、技术选型、实操细节和排查经验完整拆开讲。

1.2 整体架构：为什么是FastAPI + React + WebSocket这套组合

先说我最终定下来的架构，再解释为什么这么选。

后端用FastAPI做主体服务，负责Agent编排、会话管理、教学逻辑；前端用React做交互界面；前后端之间用WebSocket做双向实时通信；Agent的对话生成和纠错逻辑通过调用大模型API完成，中间用一层编排逻辑管理情景状态。

为什么后端选FastAPI而不是Django或者Flask？这里有几个实际考量。第一，这个项目的核心是Agent的异步编排——一次用户输入可能触发"理解意图→更新情景状态→生成角色回复→检查语法错误→决定是否给提示"这一串操作，全是IO密集型的异步任务。FastAPI原生基于async/await，配合Python的asyncio，处理这种并发场景非常自然。Django虽然也能做async，但它的ORM和整体设计还是偏同步思维，改造成本高。第二，FastAPI的Pydantic模型让Agent之间传递的结构化数据（比如情景状态、纠错结果）定义得非常清晰，类型校验自动完成，调试时省心很多。第三，它自带OpenAPI文档，前后端联调时前端直接看/docs就能知道接口长什么样。

前端选React的理由更直接：这个教学界面有大量状态需要管理——当前情景、对话历史、角色状态、纠错提示、进度面板，这些状态之间还有联动。React的组件化和状态管理机制（我用的是useState + useReducer + Context，没上Redux，后面会讲为什么）能把这些状态理清楚。而且WebSocket的消息处理需要和UI状态紧密绑定，React的响应式更新天然适合。

WebSocket是整个实时体验的关键。为什么不用HTTP轮询或者SSE？轮询的延迟和资源浪费在对话场景里很致命——用户说完一句话，等两三秒才看到回复，体验直接崩。SSE只能服务端单向推送，但教学场景里用户要不断发消息，双向通信是刚需。WebSocket建立一次连接后全双工通信，用户消息和Agent回复都能即时到达，这是唯一合理的选择。

提示：如果你的场景只是服务端单向推送数据（比如股票行情），SSE更简单；但只要涉及双向交互，WebSocket就是标准答案。

1.3 Agent的核心设计：不是聊天机器人，是教学编排器

很多人一听"Agent"就以为是套个大模型API加个prompt。这个项目里Agent的职责远不止生成回复，它要维护一整套教学状态机。

我把Agent拆成四个协作模块：情景管理器负责维护当前场景的所有状态（地点、角色、任务目标、已完成步骤）；角色扮演器负责以特定角色身份生成符合情景的英语回复；纠错引擎负责分析用户输入的语法、用词、表达地道性；教学调度器负责决定当前该推进剧情、该给提示、还是该做阶段性总结。

这四个模块不是简单的串行调用，而是根据情景状态动态编排。举个例子：用户在"餐厅点餐"情景里说了一句"I want eat pizza"，角色扮演器会正常以服务员身份回应（保持对话流畅），同时纠错引擎在后台标记出"want eat"应该是"want to eat"，教学调度器判断这是本情景第三次出现动词搭配错误，于是决定在角色回复后追加一条温和的提示。整个过程用户感知到的是一条连贯的回复，但背后是多个模块协作的结果。

这种设计的好处是教学逻辑和对话生成解耦了。我想调整纠错策略，只改纠错引擎；想加新情景，只改情景管理器。如果全塞进一个大prompt里，改一处崩一片。

2. 核心细节解析与实操要点

2.1 FastAPI项目目录结构怎么组织才不乱

FastAPI项目最容易犯的错就是所有代码堆在main.py里。项目稍微一大就变成几千行的面条代码。我最终采用的目录结构是这样的：

app/ ├── main.py # 应用入口，注册路由和WebSocket ├── core/ │ ├── config.py # 配置管理（API密钥、模型参数） │ └── logging.py # 日志配置 ├── api/ │ ├── routes/ │ │ ├── scenarios.py # 情景相关HTTP接口 │ │ └── sessions.py # 会话管理接口 │ └── websocket/ │ └── chat.py # WebSocket连接处理 ├── agents/ │ ├── scenario_manager.py # 情景管理器 │ ├── role_player.py # 角色扮演器 │ ├── grammar_checker.py # 纠错引擎 │ └── orchestrator.py # 教学调度器 ├── models/ │ ├── schemas.py # Pydantic数据模型 │ └── session.py # 会话状态模型 ├── services/ │ ├── llm_client.py # 大模型调用封装 │ └── session_store.py # 会话存储（内存/Redis） └── utils/ └── helpers.py

这个结构的关键在于按职责分层：api层只管请求进出，agents层是纯业务逻辑，services层封装外部依赖。这样测试的时候可以单独测agents层，不用起整个服务。

llm_client.py这层封装特别重要。我一开始直接在Agent里调大模型API，后来想换模型、想加缓存、想加重试，改得到处都是。抽出一层client后，所有模型调用都走统一入口，换模型只改一个文件。

2.2 WebSocket连接管理：一个连接对应一个会话

WebSocket在FastAPI里的基础写法很简单，但生产环境要考虑的东西不少。核心问题是：连接和会话怎么绑定。

我的做法是每个WebSocket连接建立时，前端先发一条初始化消息带上session_id（或者由后端生成返回），后端用这个id在session_store里创建或恢复会话状态。之后这个连接的所有消息都关联到这个会话。

from fastapi import WebSocket, WebSocketDisconnect @app.websocket("/ws/chat/{session_id}") async def chat_endpoint(websocket: WebSocket, session_id: str): await websocket.accept() session = await session_store.get_or_create(session_id) try: while True: data = await websocket.receive_json() # 交给orchestrator处理 response = await orchestrator.handle(session, data) await websocket.send_json(response) except WebSocketDisconnect: await session_store.persist(session)

看起来简单，但这里有几个坑。第一，receive_json是阻塞等待的，如果Agent处理耗时较长，用户连续发消息会堆积。我的处理是给每条消息加序号，前端在收到上一条回复前禁用发送按钮，从交互层面避免堆积。第二，连接断开时要持久化会话状态，否则用户刷新页面进度就丢了。第三，要考虑同一用户多标签页打开的情况，我用session_id做隔离，不同标签页是不同会话。

2.3 心跳机制：为什么必须有，怎么实现

WebSocket连接不是建立后就永远活着的。中间的代理、负载均衡、网络设备都可能在空闲一段时间后悄悄断开连接，而两端可能都不知道。这就是为什么心跳机制是必须的。

我踩过的坑：本地测试一切正常，部署到有反向代理的环境后，用户放着不动两三分钟，再发消息就没反应了。排查半天发现是代理层把空闲连接断了，但前端还以为连着。

心跳的实现分两端。后端定时（我用的30秒）向客户端发ping，前端收到后回pong；同时前端也定时发ping，后端回pong。任意一端在超时时间内（我设的60秒）没收到对方的心跳，就主动关闭连接并触发重连。

import asyncio async def heartbeat(websocket: WebSocket): while True: await asyncio.sleep(30) try: await websocket.send_json({"type": "ping"}) except Exception: break

前端对应处理：

useEffect(() => { const timer = setInterval(() => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'ping' })); } }, 30000); return () => clearInterval(timer); }, []);

注意：心跳间隔要小于中间设备的空闲超时时间。常见代理默认60秒，所以心跳设30秒比较稳妥。设太长没意义，设太短浪费资源。

2.4 React状态管理：为什么我没上Redux

这个项目的状态确实不少：对话消息列表、当前情景信息、角色状态、纠错记录、连接状态、加载状态。但我最终没用Redux，用的是useReducer + Context的组合。

理由是：这些状态虽然多，但更新逻辑集中——几乎所有的状态变化都源于WebSocket消息的到达。也就是说，我只需要一个reducer处理各种类型的消息，就能驱动所有状态更新。Redux解决的"状态分散在多个组件、更新逻辑散落各处"的问题，在这个项目里并不突出。

function chatReducer(state, action) { switch (action.type) { case 'MESSAGE_RECEIVED': return { ...state, messages: [...state.messages, action.payload], loading: false, }; case 'CORRECTION_RECEIVED': return { ...state, corrections: [...state.corrections, action.payload], }; case 'SCENARIO_UPDATED': return { ...state, scenario: action.payload }; default: return state; } }

用Context把state和dispatch传下去，组件里用useContext取。这个方案对这个规模的项目刚刚好，既没有Redux的样板代码负担，又保证了状态更新的可预测性。如果项目再大、状态再复杂，我会考虑上Zustand或者Jotai这类更轻量的方案，而不是Redux。

3. 实操过程与核心环节实现

3.1 情景管理器的数据结构设计

情景管理器是整个Agent的骨架，它的数据结构设计直接决定了后续所有逻辑的复杂度。我设计的情景状态包含这几块：

class ScenarioState(BaseModel): scenario_id: str scenario_type: str # 如 "restaurant", "airport" location: str # 场景地点描述 user_role: str # 用户扮演的角色 ai_role: str # AI扮演的角色 goal: str # 本情景的任务目标 steps: list[ScenarioStep] # 任务步骤列表 current_step: int # 当前进行到第几步 completed: bool turn_count: int # 对话轮次 error_history: list[str] # 历史错误记录

steps是关键。以"餐厅点餐"为例，步骤可能是：进店问候→看菜单→点主菜→点饮料→确认订单→结账。每个步骤有完成条件，教学调度器根据用户输入判断当前步骤是否完成，完成了就推进到下一步。

为什么要把步骤显式建模，而不是让大模型自己判断进度？因为大模型判断进度不稳定，同样的对话它可能这次说完成了下次说没完成。显式建模后，进度推进由代码逻辑控制，大模型只负责生成符合当前步骤的回复，可控性强得多。

3.2 角色扮演器的prompt工程

角色扮演器的核心是构造一个能让大模型稳定扮演特定角色的prompt。我试过很多版本，最终的结构是这样的：

def build_role_prompt(scenario: ScenarioState, user_input: str): return f"""你正在扮演一个英语情景对话中的角色。 场景：{scenario.location} 你的角色：{scenario.ai_role} 对方的角色：{scenario.user_role} 当前任务阶段：{scenario.steps[scenario.current_step].description} 要求： 1. 始终用英语回复，语言难度控制在CEFR B1水平 2. 回复要符合你的角色身份和当前场景 3. 回复长度控制在1-3句话，不要长篇大论 4. 如果对方表达不清，用自然的追问引导，不要直接纠正语法 5. 不要跳出角色，不要提及你是AI 对方说：{user_input} 你的回复："""

这里有几个经验点。第一，语言难度控制很重要，不控制的话大模型容易用高级词汇，学习者跟不上。第二，回复长度限制，不限制的话大模型喜欢写小作文，破坏对话节奏。第三，明确禁止直接纠正语法，因为纠错是纠错引擎的职责，角色扮演器要保持角色沉浸感。如果让角色扮演器顺便纠错，它会时不时跳出角色说"顺便提醒你..."，体验很割裂。

3.3 纠错引擎的实现策略

纠错引擎我做了两版。第一版是纯prompt让大模型分析语法错误，效果不稳定，有时候把正确的句子也标成错的。第二版改成规则+模型混合的方式。

规则层处理高频、明确的错误：主谓一致、冠词缺失、常见动词搭配、时态错误。这些用正则和简单规则就能覆盖相当一部分。模型层处理规则覆盖不到的、需要语义理解的错误，比如用词不地道、表达不自然。

async def check_grammar(text: str, context: ScenarioState): # 规则层先跑 rule_errors = rule_based_check(text) # 模型层补充 model_errors = await llm_check(text, context) # 合并去重 return merge_errors(rule_errors, model_errors)

纠错的输出不是简单标错，而是包含错误位置、错误类型、正确表达、简短解释。前端拿到后在对话气泡下方用下划线标出，鼠标悬停显示解释。这样既不打断对话流，又能让用户事后回顾。

实操心得：纠错提示的措辞很关键。用"这里更地道的说法是..."比"你错了，应该是..."友好得多。教学场景里，挫败感是用户流失的头号原因。

3.4 前后端WebSocket消息协议设计

消息协议设计得好，前后端联调能省一半时间。我定义的消息格式统一是：

{ "type": "message_type", "payload": { ... }, "timestamp": 1234567890 }

消息类型主要有这几类：

统一格式的好处是前端可以用一个switch处理所有消息，后端也可以用统一的封装函数发送。我一开始图省事，不同消息用不同格式，结果前端处理逻辑写得乱七八糟，后来重构统一了格式才清爽。

3.5 教学调度器的决策逻辑

教学调度器是Agent的"大脑"，它决定每次用户输入后该做什么。决策逻辑我简化成一个优先级判断：

async def decide_action(session, user_input): # 1. 先检查是否触发纠错 errors = await check_grammar(user_input, session.scenario) # 2. 判断当前步骤是否完成 step_done = await check_step_completion(user_input, session.scenario) # 3. 判断是否需要给提示（用户卡住时） need_hint = session.turn_count > 3 and not step_done # 4. 组装动作序列 actions = ["generate_reply"] if errors: actions.append("show_correction") if step_done: actions.append("advance_step") if need_hint: actions.append("give_hint") return actions

这个逻辑看起来简单，但实际调优花了不少时间。比如"什么时候给提示"这个判断，太早给用户觉得被小看，太晚给用户已经放弃了。我最后用的是"同一步骤停留超过3轮且没有进展"作为触发条件，实测下来比较合适。

4. 常见问题与排查技巧实录

4.1 WebSocket连接上了但收不到消息

这是我最开始遇到的最头疼的问题。现象是前端显示连接成功（onopen触发了），但发消息后收不到任何回复。

排查思路是这样的：先确认消息到底发出去没有。在浏览器开发者工具的Network面板里看WebSocket帧，如果能看到发送的帧，说明前端发送没问题。然后看后端日志，如果后端没打印收到消息，说明消息在传输层丢了或者后端没在监听。

我遇到的具体原因是：后端在WebSocket处理函数里，receive_json之后调用了耗时的Agent处理，处理期间没有继续receive，导致后续消息堆积在缓冲区。更隐蔽的是，如果Agent处理抛异常了但没被捕获，整个连接会静默关闭，前端却因为没收到close帧还以为连着。

解决办法是给Agent处理加完整的异常捕获，任何异常都要给前端发一条error消息，而不是让连接断掉：

try: response = await orchestrator.handle(session, data) await websocket.send_json(response) except Exception as e: logger.exception("Agent处理失败") await websocket.send_json({ "type": "error", "payload": {"message": "处理出错了，请重试"} })

4.2 心跳导致的重连风暴

加了心跳机制后，我遇到一个新问题：网络抖动时，前后端可能同时判定对方超时，同时断开重连，然后重连后又因为某种原因再次超时，形成重连风暴，服务器压力骤增。

解决方法是加指数退避和抖动。重连间隔不是固定的，而是1秒、2秒、4秒、8秒这样递增，同时加一个随机抖动避免多个客户端同时重连。

function reconnect(attempt) { const baseDelay = Math.min(1000 * Math.pow(2, attempt), 30000); const jitter = Math.random() * 1000; setTimeout(() => connect(), baseDelay + jitter); }

另外，重连成功后要能恢复会话状态。我的做法是重连时带上session_id，后端从session_store里恢复之前的会话，用户感觉不到断线。

4.3 大模型响应慢导致的体验问题

大模型生成回复通常要1-3秒，复杂场景可能更久。这段时间用户盯着屏幕没反应，体验很差。

我用了两个手段缓解。第一是流式输出，让大模型边生成边推送，用户能看到文字一个个蹦出来，感知等待时间大幅缩短。WebSocket天然支持流式，后端每收到一个token就推一次。

async for chunk in llm_client.stream(prompt): await websocket.send_json({ "type": "ai_message_chunk", "payload": {"text": chunk} })

第二是前端加载态，在等待期间显示"对方正在输入..."的动画，让用户知道系统在工作。

4.4 常见问题速查表

4.5 几个我踩过的坑和独家技巧

第一个坑是Pydantic模型的默认值陷阱。我在ScenarioState里给steps设了默认空列表，结果多个会话共享了同一个列表对象，一个会话的修改影响了另一个。Python的可变默认参数问题在Pydantic里同样存在，要用Field(default_factory=list)。

第二个坑是React的useEffect依赖问题。WebSocket连接的useEffect如果依赖数组写错，会导致连接反复建立和销毁。我的做法是连接逻辑只在组件挂载时执行一次，依赖数组留空，所有需要最新state的回调通过useRef获取。

第三个技巧是给Agent加超时保护。大模型调用偶尔会卡住，如果不设超时，整个会话就挂在那了。我给所有模型调用加了15秒超时，超时后返回一个降级回复，保证对话能继续。

try: reply = await asyncio.wait_for(llm_client.generate(prompt), timeout=15) except asyncio.TimeoutError: reply = "Sorry, could you say that again?"

第四个技巧是日志要带session_id。排查问题时，没有session_id的日志就是一团乱麻。我在所有日志里都带上session_id，出问题时能快速过滤出某个会话的完整链路。

4.6 关于并发的一点经验

有人问AI Agent怎么扛并发。这个项目的并发瓶颈主要在两个地方：WebSocket连接数和模型调用频率。

WebSocket连接本身很轻，单机扛几千个连接没问题，关键是每个连接背后的会话状态要管理好。我用的是内存存储加定期持久化，如果要做分布式，得换成Redis。

模型调用是真正的瓶颈。每个用户消息都触发模型调用，并发高了要么被限流要么成本爆炸。我的优化是：简单判断用规则处理不走模型（比如步骤完成判断），模型调用加缓存（相同输入直接返回缓存结果），以及给每个会话做请求队列避免同一会话并发调用。

这些经验都是实际跑起来才发现的，文档里不会写。做Agent项目，技术选型只是开始，真正的功夫在把这些细节磨平。