FastAPI+React+WebSocket：从零搭建英语情景教学AI Agent实战

英语情景教学这个方向我断断续续折腾了大半年。最开始的想法特别朴素——能不能做一个能跟人用英语聊天、还能根据场景自动切换角色和话题的陪练工具。市面上现成的口语练习产品不少但要么是固定脚本的对话树要么是纯聊天机器人没有教学结构用起来总觉得差点意思。后来我决定自己从零搭一个技术栈选了 FastAPI 做后端、React 做前端、WebSocket 做实时通信、Python 做 Agent 逻辑编排。这套组合不是拍脑袋定的下面我会把每个选型背后的理由、踩过的坑、以及最终跑通的完整方案都摊开讲。这篇文章适合两类人看一类是想动手做一个 AI Agent 项目但不知道从哪下手的开发者另一类是对 WebSocket 实时通信、FastAPI 项目结构、React 状态管理这些具体技术点有需求的前端或后端工程师。我会尽量把每个环节讲透包括为什么这么做、不这么做会怎样、以及实际写代码时那些文档里不会告诉你的细节。1. 为什么英语情景教学需要一个 Agent 而不是普通聊天机器人1.1 普通对话机器人在教学场景下的三个硬伤我最初用最简单的方案试过——直接调一个大模型的对话接口把用户输入拼上系统提示词发过去拿回回复展示。跑通只花了不到一小时但真正用起来问题很快就暴露了。第一个硬伤是没有状态管理。英语情景教学的核心是情景比如在咖啡店点单这个场景它应该有一个明确的开始、推进和结束。用户说了一句I want a latte机器人回复了然后呢下一轮用户可能突然跳到今天天气怎么样整个教学场景就散了。普通聊天机器人没有场景状态的概念它不知道当前进行到哪一步了也不知道接下来该引导用户练习什么。第二个硬伤是角色一致性无法保证。情景教学里我可能让 AI 扮演咖啡店店员、机场值机柜台工作人员、或者面试官。这些角色有不同的说话风格、词汇范围和对话目标。如果只是把角色描述塞进系统提示词多轮对话之后模型很容易出戏开始用通用助手的口吻说话。我实测下来超过七八轮之后角色漂移的概率明显上升。第三个硬伤是缺乏教学反馈。用户说了一句语法有问题的英语普通聊天机器人可能直接忽略继续对话但教学场景需要指出问题、给出纠正、甚至记录用户的常见错误类型。这些都需要在对话流程之外有一套独立的评估和反馈机制。1.2 Agent 架构带来的核心能力差异Agent 和普通聊天机器人的本质区别在于Agent 有自主决策能力。它不只是收到输入→生成输出这么一条直线而是在中间多了思考下一步该做什么的环节。放到英语情景教学这个场景里Agent 需要做的决策包括当前场景进行到哪个阶段了用户这句话有没有语法或用词问题需要纠正下一步是继续推进场景还是切换话题用户的英语水平大概在什么级别我该用多复杂的句子这些决策如果全部塞进一个提示词里让模型一次性完成效果很差。我的做法是把它们拆成独立的步骤用一个编排层来管理。这就是 Agent 框架的价值——它让你能把复杂的对话逻辑拆解成可管理、可测试、可替换的模块。具体来说我设计的 Agent 包含四个核心模块场景管理器负责维护当前场景的状态和进度对话生成器负责根据场景状态和用户输入生成回复语言评估器负责分析用户输入中的语法和用词问题难度调节器负责根据用户表现动态调整对话的复杂度。这四个模块通过一个编排器串联起来每个模块可以独立迭代和优化。1.3 从能聊到能教的关键设计决策做出一个能聊的英语对话工具不难难的是能教。这两者之间的差距我总结为三个关键设计决策。第一个决策是场景要有明确的教学目标。每个场景不是随便聊聊而是围绕特定的词汇集和句型来设计的。比如在餐厅点餐这个场景教学目标包括掌握食物名称的英文表达、学会用Would you like...和Id like...句型、理解服务员的常见问法。这些目标决定了 Agent 在对话中应该引导用户练习什么。第二个决策是反馈要即时但不打断。用户说错了你不能每句话都打断纠正那样对话根本进行不下去。我的做法是把反馈分成两类一类是即时反馈只在关键错误时触发用温和的方式嵌入对话中比如Just to clarify, youd say I would like instead of I want in this context另一类是延迟反馈在场景结束后统一给出详细的语法和用词分析报告。第三个决策是难度要动态适配。同一个场景对初级用户和高级用户的对话策略应该完全不同。初级用户需要短句、慢速、多重复高级用户需要长句、地道表达、甚至俚语。我在 Agent 里维护了一个用户能力模型根据用户的历史表现动态调整对话生成时的提示词参数。2. FastAPI 后端项目结构从目录设计到 WebSocket 路由2.1 为什么选 FastAPI 而不是 Django 或 Flask后端框架的选择我对比了三个Django、Flask 和 FastAPI。最终选 FastAPI 的原因很具体。Django 太重了。它自带 ORM、Admin、模板引擎、认证系统这些对于一个 Agent 后端来说大部分用不上。而且 Django 的 WebSocket 支持需要额外装 Channels配置起来比较繁琐。Flask 够轻但原生不支持异步WebSocket 需要靠 Flask-SocketIO 这类扩展性能和开发体验都不如原生异步方案。FastAPI 的优势在于原生异步支持基于 Starlette、内置 WebSocket 支持、自动生成 API 文档、Pydantic 做数据校验。对于 Agent 这种需要处理大量并发 WebSocket 连接、同时要跟大模型 API 做异步交互的场景FastAPI 几乎是天然匹配的。如果你之前只用过 Django转 FastAPI 最大的思维转变是一切都要考虑异步。数据库操作要用异步驱动HTTP 请求要用 httpx 的异步客户端文件 IO 要用 aiofiles。混用同步代码会阻塞事件循环在高并发下性能急剧下降。2.2 项目目录结构的实战设计我试过好几种目录组织方式最后稳定下来的结构是这样的english-agent/ ├── app/ │ ├── main.py # 应用入口注册路由和中间件 │ ├── config.py # 配置管理读取环境变量 │ ├── models/ # Pydantic 数据模型 │ │ ├── conversation.py │ │ ├── scenario.py │ │ └── user.py │ ├── agents/ # Agent 核心逻辑 │ │ ├── orchestrator.py # 编排器 │ │ ├── scenario_manager.py │ │ ├── dialogue_generator.py │ │ ├── language_evaluator.py │ │ └── difficulty_adapter.py │ ├── services/ # 外部服务封装 │ │ ├── llm_client.py # 大模型 API 调用 │ │ └── session_store.py # 会话存储 │ ├── api/ # 路由层 │ │ ├── websocket.py # WebSocket 端点 │ │ └── rest.py # REST 端点 │ └── utils/ # 工具函数 │ ├── prompt_templates.py │ └── text_processing.py ├── tests/ ├── requirements.txt └── .env这个结构的关键设计思路是按职责分层而不是按技术类型分层。agents 目录下放的是纯业务逻辑不依赖任何 Web 框架services 目录封装外部依赖方便替换和 mockapi 目录只负责协议转换把 WebSocket 消息转成 Agent 能处理的输入再把 Agent 的输出转成 WebSocket 消息发回去。这样分层的实际好处是测试 Agent 逻辑时完全不需要启动 Web 服务器直接实例化 orchestrator 传参调用就行。我在开发过程中大量使用了这种方式做快速迭代比每次都要起服务发请求效率高得多。2.3 WebSocket 端点的完整实现与心跳机制WebSocket 是前端和后端之间的实时通道。英语情景教学场景下用户说完一句话Agent 可能需要几百毫秒到几秒的时间来生成回复取决于大模型的响应速度如果用 HTTP 轮询要么延迟高要么请求浪费。WebSocket 建立一次连接后可以双向推送非常适合这种场景。FastAPI 的 WebSocket 端点写法很直观from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.agents.orchestrator import AgentOrchestrator app FastAPI() app.websocket(/ws/conversation/{session_id}) async def conversation_websocket(websocket: WebSocket, session_id: str): await websocket.accept() orchestrator AgentOrchestrator(session_id) try: while True: data await websocket.receive_json() if data[type] ping: await websocket.send_json({type: pong}) continue if data[type] user_message: # 先发送一个正在思考的状态 await websocket.send_json({ type: agent_thinking, data: {status: processing} }) # 异步处理用户消息 response await orchestrator.process(data[content]) await websocket.send_json({ type: agent_message, data: response }) except WebSocketDisconnect: await orchestrator.cleanup()这段代码里有几个细节值得展开说。心跳机制是必须的。WebSocket 连接在长时间没有数据传输时中间的网络设备负载均衡、反向代理可能会主动断开。我设置的是客户端每 30 秒发一次 ping服务端回 pong。如果服务端连续两次没收到 ping就主动关闭连接并清理资源。这个逻辑我在前端和后端都做了实现后面讲前端的时候会详细说。正在思考状态这个设计是我踩坑之后加的。最开始用户发完消息后界面上没有任何反馈要等 Agent 完全生成完回复才一次性显示。大模型生成一段回复可能需要两三秒用户在这段时间里会以为程序卡死了。加上这个中间状态后前端可以立即显示一个对方正在输入的动画体验好很多。会话清理也很关键。每个 WebSocket 连接对应一个会话会话里保存了对话历史、场景状态、用户能力模型等数据。连接断开时必须清理这些数据否则内存会持续增长。我的做法是在 orchestrator 的 cleanup 方法里把会话数据持久化到存储层我用的是 Redis然后从内存中移除。2.4 会话状态管理与并发处理Agent 的会话状态包括当前场景 ID、场景进度、对话历史、用户能力评估、错误记录。这些数据需要在多次 WebSocket 消息之间保持。我最初把状态存在内存字典里key 是 session_id。单机跑没问题但一旦要考虑多进程部署就完蛋了——用户的第二条消息可能被路由到另一个进程那个进程里没有这个会话的状态。解决方案是把会话状态外置到 Redis。每次处理消息时从 Redis 读取状态处理完写回。Redis 的读写速度足够快不会成为瓶颈。而且 Redis 天然支持过期时间可以给会话设置一个合理的 TTL我设的是 2 小时避免僵尸会话占用内存。并发方面FastAPI 的异步特性让单个进程可以同时处理大量 WebSocket 连接。但要注意的是Agent 处理逻辑里如果有 CPU 密集型的操作比如文本分析会阻塞事件循环。我的做法是把这类操作放到run_in_executor里执行或者用asyncio.to_thread包装。import asyncio async def process_with_evaluation(self, user_input: str): # 对话生成是 IO 密集型直接 await dialogue_task asyncio.create_task( self.dialogue_generator.generate(user_input) ) # 语言评估如果涉及复杂计算放到线程池 evaluation_task asyncio.create_task( asyncio.to_thread(self.language_evaluator.analyze, user_input) ) dialogue, evaluation await asyncio.gather( dialogue_task, evaluation_task ) return self._merge_response(dialogue, evaluation)这段代码展示了并行执行对话生成和语言评估的思路。两个任务互不依赖可以同时跑总耗时取决于较慢的那个而不是两者之和。在实际测试中这种并行方式比串行快了大约 40%。3. React 前端的实时对话界面与状态管理3.1 WebSocket 连接的生命周期管理前端这边WebSocket 连接的管理是整个应用的核心。我把它封装成了一个自定义 Hook叫useAgentSocket。这个 Hook 负责建立连接、维护心跳、处理断线重连、以及分发消息。import { useEffect, useRef, useCallback, useState } from react; function useAgentSocket(sessionId) { const wsRef useRef(null); const heartbeatRef useRef(null); const reconnectRef useRef(null); const [status, setStatus] useState(disconnected); const [messages, setMessages] useState([]); const connect useCallback(() { const ws new WebSocket(ws://localhost:8000/ws/conversation/${sessionId}); wsRef.current ws; ws.onopen () { setStatus(connected); // 启动心跳 heartbeatRef.current setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping })); } }, 30000); }; ws.onmessage (event) { const data JSON.parse(event.data); if (data.type pong) return; if (data.type agent_thinking) { setStatus(thinking); return; } if (data.type agent_message) { setStatus(connected); setMessages(prev [...prev, { role: agent, content: data.data.text, feedback: data.data.feedback }]); } }; ws.onclose () { setStatus(disconnected); clearInterval(heartbeatRef.current); // 指数退避重连 reconnectRef.current setTimeout(() { connect(); }, 2000); }; ws.onerror () { ws.close(); }; }, [sessionId]); useEffect(() { connect(); return () { clearInterval(heartbeatRef.current); clearTimeout(reconnectRef.current); wsRef.current?.close(); }; }, [connect]); const sendMessage useCallback((content) { if (wsRef.current?.readyState WebSocket.OPEN) { wsRef.current.send(JSON.stringify({ type: user_message, content })); setMessages(prev [...prev, { role: user, content }]); } }, []); return { status, messages, sendMessage }; }这个 Hook 里有几个我在实际开发中反复调整过的地方。心跳间隔的选择。30 秒是我测试下来比较稳妥的值。太短了浪费带宽和电量太长了可能被中间设备判定为空闲连接而断开。如果你的部署环境有负载均衡器建议查一下它的空闲超时设置心跳间隔要小于那个值。重连策略。我最初用的是固定间隔重连每 2 秒试一次后来改成了指数退避——第一次 1 秒第二次 2 秒第三次 4 秒最多到 30 秒。这样在服务端短暂不可用时不会疯狂重连打爆服务器同时恢复后也能较快重新连上。消息去重。网络不稳定时可能出现消息重复推送的情况。我在消息对象里加了一个服务端生成的唯一 ID前端收到消息后先检查这个 ID 是否已经处理过避免重复显示。3.2 对话界面的组件拆分与渲染优化对话界面看起来简单——不就是消息列表加输入框吗但实际做起来性能优化和用户体验的细节非常多。组件拆分上我分成了四个主要组件MessageList负责渲染消息列表MessageBubble负责单条消息的展示InputArea负责输入和发送ScenarioPanel负责显示当前场景信息和进度。性能优化的关键是避免不必要的重渲染。消息列表在每次新消息到来时都会更新如果MessageBubble没有做 memo 优化所有历史消息都会重新渲染。消息多了之后超过 50 条明显能感觉到卡顿。import React, { memo } from react; const MessageBubble memo(function MessageBubble({ message }) { const isUser message.role user; return ( div className{bubble ${isUser ? user : agent}} div classNamecontent{message.content}/div {message.feedback ( div classNamefeedback span classNamelabel语言提示/span p{message.feedback}/p /div )} /div ); });用memo包裹之后只有 message 对象的引用发生变化时才会重新渲染。配合在消息列表更新时使用不可变数据模式每次创建新数组而不是修改原数组可以确保只有新增的消息会触发渲染。另一个优化点是自动滚动到底部。新消息到来时消息列表应该自动滚动到最新位置。但如果用户正在往上翻看历史消息就不应该强制滚动。我的做法是监听滚动事件判断用户当前是否在底部附近距离底部小于 100px只有在底部时才自动滚动。3.3 场景选择与进度展示的交互设计场景选择界面是用户进入对话前的第一步。我设计了六个初始场景咖啡店点单、机场值机、酒店入住、餐厅用餐、面试对话、日常闲聊。每个场景卡片上显示场景名称、难度等级、预计对话轮数、以及核心学习目标。用户选择场景后前端通过 REST 接口创建一个新会话拿到 session_id然后用这个 ID 建立 WebSocket 连接。这个流程的顺序很重要——必须先有 session_id 才能连 WebSocket因为服务端需要根据 session_id 来初始化对应的场景状态。进度展示方面我在对话界面右侧放了一个侧边栏显示当前场景的完成进度。进度是根据 Agent 内部维护的场景状态来计算的比如咖啡店点单场景有五个阶段问候、点单、询问细节、确认订单、结账。每完成一个阶段进度条就前进一格。这个进度信息是通过 WebSocket 从服务端推送到前端的。每次 Agent 处理完用户消息后除了返回对话回复还会附带当前的场景状态。前端根据这个状态更新进度展示。这里有个容易忽略的细节场景进度的推进不应该完全由对话轮数决定而应该由 Agent 判断用户是否完成了当前阶段的目标。比如点单阶段用户可能说了三句话才成功点了一杯咖啡也可能一句话就搞定了。用轮数来推进进度会导致体验很僵硬。4. Agent 核心逻辑场景编排、语言评估与难度适配4.1 场景状态机的设计与实现场景状态机是整个 Agent 的骨架。每个场景定义为一个状态机包含若干状态和状态之间的转移条件。以咖啡店点单为例我定义的状态包括GREETING问候、ORDERING点单、CUSTOMIZING定制需求比如加奶、加糖、CONFIRMING确认订单、PAYMENT付款、COMPLETED完成。状态转移由 Agent 根据用户输入来判断。比如从GREETING到ORDERING的条件是用户表达了点单意图从ORDERING到CUSTOMIZING的条件是用户点了某种咖啡但未指定细节。from enum import Enum from pydantic import BaseModel class ScenarioState(str, Enum): GREETING greeting ORDERING ordering CUSTOMIZING customizing CONFIRMING confirming PAYMENT payment COMPLETED completed class ScenarioContext(BaseModel): scenario_id: str current_state: ScenarioState state_history: list[ScenarioState] turn_count: int collected_info: dict # 已收集的信息如咖啡类型、大小、是否加奶 class ScenarioManager: def __init__(self, scenario_id: str): self.context ScenarioContext( scenario_idscenario_id, current_stateScenarioState.GREETING, state_history[], turn_count0, collected_info{} ) def evaluate_transition(self, user_input: str, llm_analysis: dict) - ScenarioState: current self.context.current_state if current ScenarioState.GREETING: if llm_analysis.get(intent) ordering: return ScenarioState.ORDERING if current ScenarioState.ORDERING: if llm_analysis.get(has_incomplete_order): return ScenarioState.CUSTOMIZING if llm_analysis.get(order_complete): return ScenarioState.CONFIRMING # ... 其他状态转移逻辑 return current这里的关键设计是状态转移的判断交给大模型来做但转移规则由代码控制。大模型负责理解用户输入的意图比如判断用户是不是在点单但具体从哪个状态转到哪个状态是由代码里定义的规则决定的。这样做的好处是状态机不会跑飞同时又能处理自然语言的多样性。4.2 语言评估器的实现细节语言评估器负责分析用户输入中的语法、用词、表达地道程度。我把它设计成一个独立的模块输入是用户的一句话输出是一个结构化的评估结果。评估维度包括语法正确性是否有主谓一致、时态、冠词等错误、用词恰当性是否用了不合适的词、表达地道性是否有更自然的说法、以及复杂度句子的词汇和语法复杂度。实现上我用了一个专门的提示词来让大模型做评估要求它输出 JSON 格式的结果EVALUATION_PROMPT You are an English language evaluator. Analyze the following user input in the context of a {scenario} scenario. User input: {user_input} Provide your evaluation in the following JSON format: {{ grammar_errors: [ {{original: ..., corrected: ..., explanation: ...}} ], word_choice_suggestions: [ {{original: ..., suggested: ..., reason: ...}} ], naturalness_score: 0-100, complexity_level: beginner|intermediate|advanced, overall_feedback: ... }} Only include errors that are relevant to the current scenario. Do not over-correct minor issues that would interrupt the conversation flow. 这里有个重要的经验不要过度纠正。我最初让模型把所有语法问题都指出来结果用户每说一句话都被挑出三四个错误体验非常糟糕。后来我在提示词里加了只关注与当前场景相关的错误和不要过度纠正的约束效果好多了。另外评估结果不是每次都展示给用户。我的策略是如果用户输入没有明显错误评估结果只记录不展示如果有错误根据错误的严重程度决定是即时反馈还是延迟反馈。严重错误比如时态完全用错导致意思变了即时反馈轻微错误比如冠词用错但不影响理解延迟到场景结束后统一反馈。4.3 难度动态调节的策略与参数难度调节器是我觉得最有意思的模块。它的目标是根据用户的表现动态调整 Agent 回复的复杂度。我维护了一个用户能力模型包含几个维度词汇量估计、语法准确率、平均句长、对话流畅度。每次用户输入后这些指标都会更新。然后根据这些指标计算一个综合能力分数映射到三个难度等级。难度等级影响的是对话生成时的提示词参数难度等级句长词汇范围语速暗示反馈频率初级短句为主5-10词基础1000词慢多停顿高每轮都可能有提示中级中等句长10-20词常用3000词正常中关键错误才提示高级长句和复合句丰富词汇含习语正常偏快低仅严重错误提示这个表格不是拍脑袋定的是我根据教学经验和实际测试调整出来的。初级用户如果收到太长的句子会直接懵掉高级用户如果一直收到简单句会觉得无聊。难度调节的粒度是每个场景重新评估一次而不是每轮对话都变。频繁变化会让用户感到困惑。如果一个场景内用户表现明显超出或低于当前难度Agent 会在场景结束后调整下一个场景的难度。5. 部署与性能让 Agent 扛住真实并发5.1 大模型调用的延迟优化Agent 的响应延迟主要来自大模型调用。一次完整的对话处理可能需要调用大模型两到三次一次做意图分析一次生成对话一次做语言评估。如果串行调用总延迟可能达到 5-8 秒用户体验很差。我的优化策略是能并行的并行能缓存的缓存。并行方面意图分析和语言评估可以同时进行因为它们都只依赖用户输入互不依赖。对话生成需要等意图分析的结果因为要知道当前处于哪个场景状态所以它必须在意图分析之后。缓存方面场景的静态信息比如场景描述、教学目标、角色设定可以预先处理好放在提示词模板里不需要每次调用都重新生成。另外一些常见的用户输入比如hello、thank you的评估结果可以缓存避免重复调用。import hashlib import json class LLMClient: def __init__(self, redis_client): self.redis redis_client self.cache_ttl 3600 # 1 hour async def evaluate(self, user_input: str, scenario: str) - dict: cache_key feval:{hashlib.md5(f{user_input}:{scenario}.encode()).hexdigest()} cached await self.redis.get(cache_key) if cached: return json.loads(cached) result await self._call_llm_evaluation(user_input, scenario) await self.redis.setex(cache_key, self.cache_ttl, json.dumps(result)) return result这个缓存策略在实际测试中把重复输入的评估延迟从平均 1.5 秒降到了 20 毫秒以内。对于教学场景来说很多用户的输入是相似的尤其是初级用户缓存命中率相当可观。5.2 WebSocket 连接数的压力测试与调优我用 Locust 做了一轮压力测试模拟多个用户同时建立 WebSocket 连接并进行对话。测试环境是 4 核 8G 的云服务器单进程 FastAPI。第一轮测试结果不太理想50 个并发连接时响应还正常到 100 个连接时开始出现明显的延迟增加到 200 个连接时部分连接开始超时断开。排查后发现两个瓶颈。第一个是大模型 API 的并发限制。我的大模型服务商对并发请求数有限制超过之后请求会排队。解决方案是加了一个信号量来控制并发调用数超出的请求排队等待而不是直接失败。第二个瓶颈是Python 的 GIL。虽然 FastAPI 是异步的但一些同步的库调用比如 JSON 序列化、文本处理仍然会占用 GIL。解决方案是把这些操作尽量用异步库替代或者放到线程池里执行。调优之后同样的硬件配置下稳定支持 150 个并发连接平均响应延迟在 2 秒以内。对于一个小规模的教学应用来说这个容量已经足够了。如果要支持更多用户最直接的方式是水平扩展——部署多个实例用负载均衡分发 WebSocket 连接。WebSocket 的负载均衡需要注意普通的轮询负载均衡会导致同一个会话的消息被分发到不同实例。解决方案是使用一致性哈希或者粘性会话sticky session确保同一个 session_id 的连接始终路由到同一个实例。5.3 从开发环境到生产环境的配置差异开发环境和生产环境的配置差异比我想象的大。我列一下主要的不同点配置项开发环境生产环境大模型快速但质量一般的模型质量更好的模型日志级别DEBUGINFO错误处理显示完整堆栈只返回错误码和简要信息WebSocket 地址ws://localhost:8000wss://api.example.com会话存储内存Redis静态资源开发服务器CDN并发控制无限制信号量限流生产环境必须用 wssWebSocket Secure否则浏览器会阻止混合内容。这需要在反向代理我用的是 Nginx上配置 SSL 证书和 WebSocket 升级。Nginx 的 WebSocket 代理配置有几个关键参数location /ws/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }proxy_read_timeout和proxy_send_timeout必须设得足够大否则 Nginx 会在默认的 60 秒后主动断开 WebSocket 连接。我设的是 3600 秒配合前端 30 秒的心跳可以保持连接稳定。6. 开发过程中踩过的坑与解决方案6.1 WebSocket 连接建立但不接收消息的问题这个问题困扰了我整整一个下午。现象是前端 WebSocket 的onopen回调正常触发连接状态显示已连接但发送消息后服务端没有任何反应onmessage也从来不触发。排查过程是这样的先确认前端确实发送了消息在ws.send后面加了 console.log确认执行了然后检查服务端的 WebSocket 端点是否收到了消息在receive_json后面加日志发现根本没走到这行最后用浏览器的开发者工具看 WebSocket 帧发现消息确实发出去了但服务端没收到。根因是Nginx 配置缺少 WebSocket 升级头。开发时前端直接连的是 FastAPI 的端口没有经过 Nginx所以没问题。部署到生产环境后请求经过 Nginx 转发但 Nginx 默认不支持 WebSocket 升级导致连接建立了但数据帧无法正常传输。解决方案就是上面提到的那段 Nginx 配置关键是proxy_set_header Upgrade $http_upgrade和proxy_set_header Connection upgrade这两行。加上之后问题立刻解决。这个坑的教训是开发环境和生产环境的网络拓扑差异一定要尽早验证。如果我在开发初期就把 Nginx 加上这个问题在第一天就会暴露而不是等到部署时才发现。6.2 React 状态更新导致的消息乱序另一个让我头疼的问题是消息乱序。用户快速连续发送两条消息时前端的显示顺序偶尔会颠倒。根因是 React 的状态更新是异步的。我在sendMessage里调用了setMessages但这个更新不会立即生效。如果用户紧接着又发了一条消息第二次setMessages可能基于旧的 state 计算导致消息顺序错乱。解决方案是使用函数式更新setMessages(prev [...prev, newMessage]);而不是setMessages([...messages, newMessage]);函数式更新确保每次都是基于最新的 state 来计算新值避免了闭包捕获旧 state 的问题。这个改动很小但解决了一个很隐蔽的 bug。另外服务端推送消息时也可能出现乱序。我的解决方案是在每条消息上附加一个递增的序列号前端收到消息后按序列号排序再显示。如果发现序列号有跳跃说明有消息丢失前端会主动请求补发。6.3 大模型输出格式不稳定的处理让大模型输出 JSON 格式的评估结果时偶尔会遇到格式不正确的情况——比如多了 markdown 代码块标记、少了引号、或者字段名拼写错误。这会导致json.loads直接抛异常。我的处理策略是三层防护。第一层是在提示词里明确要求只输出 JSON不要包含任何其他文本并给出格式示例。第二层是在解析前做预处理去掉可能的 markdown 标记和多余空白。第三层是如果解析仍然失败降级到一个简化的评估结果而不是让整个对话流程崩溃。import json import re def parse_evaluation(raw_output: str) - dict: # 去掉 markdown 代码块标记 cleaned re.sub(r^json\s*, , raw_output.strip()) cleaned re.sub(r\s*$, , cleaned) try: return json.loads(cleaned) except json.JSONDecodeError: # 尝试提取第一个完整的 JSON 对象 match re.search(r\{.*\}, cleaned, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass # 降级方案 return { grammar_errors: [], word_choice_suggestions: [], naturalness_score: 70, complexity_level: intermediate, overall_feedback: }这个降级方案看起来简单但在实际运行中救了很多次场。大模型偶尔抽风是不可避免的关键是不能让它的不稳定影响到整个系统的可用性。6.4 会话数据在重连后的恢复用户网络波动导致 WebSocket 断开重连后之前的对话历史和场景状态应该恢复。这个功能看起来简单实现起来有几个细节要注意。首先是重连时的身份识别。前端重连时用的 session_id 必须和之前一致否则服务端找不到对应的会话数据。我的做法是把 session_id 存在 sessionStorage 里页面刷新或重连时从里面读取。其次是消息的补发。重连后前端需要知道断开期间有没有遗漏的消息。我的做法是前端记录最后收到的消息序列号重连时把这个序列号发给服务端服务端把之后的消息补发过来。最后是场景状态的恢复。Agent 的场景状态机状态需要从 Redis 里恢复。这里要注意的是恢复后的状态应该是继续而不是重新开始。比如用户之前在确认订单阶段断开了重连后应该继续在这个阶段而不是回到问候阶段。7. 后续可以继续打磨的方向这个项目从最初的想法到跑通完整流程大概花了三个月左右的业余时间。目前的状态是可用的但离好用还有距离。短期最想优化的是语音输入和输出。现在用户需要打字输入这对口语练习来说不够自然。接入语音识别和语音合成之后整个体验会更接近真实的对话练习。技术方案上语音识别可以用 Whisper 这类模型语音合成可以用一些开源的 TTS 方案。主要的挑战是延迟——语音识别加语音合成会额外增加一到两秒的处理时间需要和对话生成做并行优化。中期想加的是多用户对战模式。两个用户同时进入一个场景一个扮演顾客一个扮演服务员Agent 在旁边做裁判和提示。这个模式对练口语的互动性更强但技术上需要处理多路 WebSocket 的同步问题复杂度会高不少。长期来看个性化学习路径是更有价值的方向。根据用户的历史表现自动推荐下一个应该练习的场景针对薄弱环节生成专项练习。这需要积累足够的用户数据以及一套更精细的能力评估模型。如果你也在做类似的项目我的建议是先把最小可用的对话流程跑通不要一上来就追求完美的架构。我在最开始花了很多时间设计完美的 Agent 框架结果发现很多设计在实际使用中根本不重要而一些没考虑到的小细节反而成了瓶颈。快速迭代、根据实际使用反馈来调整比前期过度设计有效得多。