尧图网站建设 
1. 项目缘起与整体架构设计1.1 为什么我要做这么一个东西事情的起因很简单我有个朋友在做英语培训他跟我吐槽说现在市面上的英语学习产品要么是纯背单词的要么是录播课学生开口说英语的机会少得可怜。真人外教一对一效果确实好但成本摆在那里一节课动辄两三百普通家庭很难长期负担。他问我能不能用技术手段做一个“能跟人对话的英语陪练”最好还能根据场景来教学比如点餐、问路、面试这些具体情境。这个需求其实很明确需要一个能理解上下文、能扮演角色、能纠正错误的对话系统。大语言模型天然适合干这个事但光有模型还不够你得把它包装成一个完整的产品——前端要有聊天界面后端要管理会话状态模型调用要稳定可靠还得考虑流式输出让对话体验更自然。这就是我决定从零到一开发一个英语情景教学Agent的完整动机。这个项目适合谁看如果你已经会一点Python和JavaScript想了解怎么把大模型能力落地成一个真实可用的产品那这篇文章就是写给你的。我会把整个开发过程中的技术选型、踩过的坑、关键代码实现都摊开来讲尽量做到你照着做就能跑起来。1.2 技术栈选型与背后的考量先说说我为什么选这套技术栈。前端用React后端用FastAPI通信层用WebSocket模型调用走Python的异步生态。这个组合不是拍脑袋定的每一个选择都有具体的理由。React做前端核心原因是它的组件化模型非常适合聊天界面这种“消息列表输入框状态管理”的结构。每条消息是一个组件消息列表是一个容器组件输入框是受控组件整个数据流非常清晰。而且React生态里有大量现成的UI库和工具开发效率高。2026年的React生态已经非常成熟Hooks API让状态管理变得直观不需要引入额外的状态管理库就能搞定大部分场景。FastAPI做后端主要是看中它的异步支持和WebSocket原生集成。FastAPI基于Starlette对WebSocket的支持是一等公民级别的写起来非常顺手。另外它的依赖注入系统和Pydantic模型验证让代码组织很干净接口文档自动生成调试起来方便。相比DjangoFastAPI更轻量更适合这种以API和实时通信为主的项目。WebSocket做通信这是关键决策。英语对话教学的核心体验是“实时感”如果用传统的HTTP轮询每次发消息都要重新建立连接延迟高不说服务端也没法主动推送消息。WebSocket建立一次连接后可以双向通信模型生成的内容可以逐字推送到前端用户看到的就是打字机效果体验好很多。而且WebSocket的心跳机制可以保持连接活跃避免长时间不操作后断连。Python异步生态因为大模型调用本身是IO密集型的用异步可以同时处理多个用户的请求不会因为一个用户的模型调用阻塞其他用户。FastAPI的async/await语法和Python的asyncio库配合得很好整个后端可以做到高并发。1.3 整体架构长什么样整个系统的架构可以分成四层第一层是前端展示层React应用负责渲染聊天界面、管理用户输入、维护消息列表状态。用户打开页面后前端会尝试建立WebSocket连接连接成功后进入对话状态。第二层是通信层WebSocket连接承载所有实时消息。前端发送用户消息后端推送模型生成的回复。心跳机制保证连接不会因为空闲而断开。第三层是业务逻辑层FastAPI应用处理WebSocket消息管理会话上下文调用大模型API处理流式响应。这一层还负责场景切换、错误处理、会话超时清理等逻辑。第四层是模型服务层实际的大模型调用。我选择的是通过API调用的方式而不是本地部署模型因为API调用更稳定不需要维护GPU资源成本也更可控。这四层之间的数据流是这样的用户在React界面输入一句话前端通过WebSocket发送一个JSON消息到后端后端解析消息后从会话存储中取出历史上下文拼接成Prompt发给大模型大模型流式返回结果后端每收到一个token就通过WebSocket推送给前端前端逐字渲染出来。2. 核心细节解析与实操要点2.1 WebSocket连接的生命周期管理WebSocket连接不是建立后就一劳永逸的它有自己的生命周期需要仔细管理。我把整个生命周期分成四个阶段连接建立、消息收发、心跳保活、连接关闭。连接建立阶段前端在组件挂载时发起WebSocket连接。这里有个细节需要注意React的StrictMode在开发环境下会故意重复挂载组件导致WebSocket被创建两次。我的处理方式是在useEffect的清理函数中关闭连接并且用一个ref来标记连接是否已经建立避免重复创建。const wsRef useRef(null); const isConnectedRef useRef(false); useEffect(() { if (isConnectedRef.current) return; const ws new WebSocket(ws://localhost:8000/ws/chat); wsRef.current ws; ws.onopen () { isConnectedRef.current true; console.log(WebSocket connected); }; ws.onmessage (event) { const data JSON.parse(event.data); // 处理消息 }; ws.onclose () { isConnectedRef.current false; console.log(WebSocket closed); }; return () { if (wsRef.current) { wsRef.current.close(); isConnectedRef.current false; } }; }, []);消息收发阶段前端发送的消息格式我定义成JSON包含type和payload两个字段。type标识消息类型比如user_message表示用户发送的对话内容scene_change表示切换教学场景。payload携带具体数据。后端返回的消息也是类似结构type为assistant_message时payload里是模型生成的内容片段。心跳保活阶段这是很多人容易忽略的地方。WebSocket连接如果长时间没有数据传输中间的网络设备比如负载均衡器、代理服务器可能会主动断开连接。我设置的心跳间隔是30秒前端每隔30秒发送一个ping消息后端收到后回复pong。如果前端连续三次没有收到pong就认为连接已断开触发重连逻辑。const heartbeatInterval setInterval(() { if (wsRef.current wsRef.current.readyState WebSocket.OPEN) { wsRef.current.send(JSON.stringify({ type: ping })); } }, 30000);连接关闭阶段需要区分正常关闭和异常关闭。正常关闭时清理定时器、重置状态即可。异常关闭时需要触发重连我设置的重连策略是指数退避第一次1秒后重连第二次2秒第三次4秒最多重试5次。如果5次都失败就在界面上提示用户检查网络。2.2 会话上下文的管理策略英语情景教学Agent的核心能力是“记住对话历史”这样才能进行连贯的对话。但大模型的上下文窗口是有限的不能无限追加历史消息。我的策略是滑动窗口加摘要压缩。具体来说每个会话维护一个消息列表包含用户和助手的对话记录。当消息数量超过20条时把最早的10条拿出来调用模型生成一段摘要然后把摘要作为系统消息放在最前面后面保留最近的10条消息。这样既保留了长期记忆又控制了上下文长度。会话数据存在哪里我用的是内存字典加定期持久化。内存字典的key是session_idvalue是会话对象。每隔5分钟把活跃会话写入SQLite数据库服务重启后可以从数据库恢复。对于生产环境建议用Redis做会话存储性能和可靠性都更好。class SessionManager: def __init__(self): self.sessions {} self.lock asyncio.Lock() async def get_session(self, session_id: str) - Session: async with self.lock: if session_id not in self.sessions: self.sessions[session_id] Session(session_id) return self.sessions[session_id] async def add_message(self, session_id: str, role: str, content: str): session await self.get_session(session_id) session.messages.append({role: role, content: content}) if len(session.messages) 20: await self._compress_history(session)2.3 流式输出的实现细节流式输出是提升对话体验的关键。用户发送消息后如果等模型生成完整回复再显示可能要等好几秒体验很差。流式输出让用户看到文字一个个蹦出来感觉就像真人在打字。后端实现流式输出的核心是异步生成器。调用大模型API时开启stream模式API会返回一个异步迭代器每产生一个token就yield出来。FastAPI的WebSocket send_text方法可以逐个发送这些token。async def stream_response(session_id: str, user_message: str): session await session_manager.get_session(session_id) messages session.get_context_messages() messages.append({role: user, content: user_message}) async for chunk in llm_client.stream_chat(messages): yield chunk前端接收流式数据时需要维护一个“当前正在生成的消息”状态。每收到一个chunk就追加到这个消息的内容后面触发React重新渲染。这里有个性能优化点不要每个chunk都setState可以用一个buffer累积每隔50毫秒批量更新一次减少渲染次数。const bufferRef useRef(); const flushTimerRef useRef(null); ws.onmessage (event) { const data JSON.parse(event.data); if (data.type assistant_message) { bufferRef.current data.payload.content; if (!flushTimerRef.current) { flushTimerRef.current setTimeout(() { setMessages(prev { const last prev[prev.length - 1]; if (last last.role assistant last.streaming) { return [...prev.slice(0, -1), { ...last, content: bufferRef.current }]; } return [...prev, { role: assistant, content: bufferRef.current, streaming: true }]; }); bufferRef.current ; flushTimerRef.current null; }, 50); } } };2.4 教学场景的Prompt工程设计这个Agent跟普通聊天机器人的区别在于“教学”属性。它不只是跟用户对话还要在对话中纠正错误、引导表达、解释语法。这些行为都通过Prompt来定义。我为每个场景设计了一个系统Prompt模板包含角色设定、教学目标、行为规则三个部分。以“餐厅点餐”场景为例你是一位英语口语教练正在模拟餐厅点餐场景。 你的角色是餐厅服务员用户是顾客。 教学目标 1. 引导用户使用完整的句子点餐 2. 在用户表达错误时先自然回应再指出错误并给出正确说法 3. 适当引入新词汇比如appetizer、recommendation、medium rare 行为规则 - 始终用英语对话除非用户明确要求用中文解释 - 每次回复不超过3句话保持对话节奏 - 如果用户沉默超过两轮主动提问引导这个Prompt的关键在于“先自然回应再指出错误”。如果用户说“I want eat pizza”直接纠正会打断对话流畅感。更好的方式是先回应“Sure, one pizza coming up. By the way, we usually say I want to eat pizza or Id like to eat pizza.”这样既完成了对话又完成了教学。3. 实操过程与核心环节实现3.1 后端项目结构与核心代码FastAPI项目的目录结构我习惯这样组织backend/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── config.py # 配置管理 │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py # Pydantic模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── session.py # 会话管理 │ │ ├── llm_client.py # 大模型调用 │ │ └── prompt.py # Prompt模板 │ ├── api/ │ │ ├── __init__.py │ │ └── websocket.py # WebSocket路由 │ └── utils/ │ ├── __init__.py │ └── logger.py # 日志工具 ├── requirements.txt └── run.pymain.py是应用入口负责创建FastAPI实例、注册路由、配置CORSfrom fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.websocket import router as ws_router app FastAPI(titleEnglish Teaching Agent) app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.include_router(ws_router, prefix/ws) app.get(/health) async def health_check(): return {status: ok}WebSocket路由是核心处理连接建立、消息接收、消息发送、连接关闭from fastapi import APIRouter, WebSocket, WebSocketDisconnect import json import asyncio from app.services.session import session_manager from app.services.llm_client import llm_client from app.services.prompt import get_system_prompt router APIRouter() router.websocket(/chat) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() session_id None try: while True: raw await websocket.receive_text() data json.loads(raw) if data[type] ping: await websocket.send_text(json.dumps({type: pong})) continue if data[type] init: session_id data[payload][session_id] scene data[payload].get(scene, restaurant) await session_manager.init_session(session_id, scene) await websocket.send_text(json.dumps({ type: init_ack, payload: {session_id: session_id} })) continue if data[type] user_message: user_text data[payload][content] await session_manager.add_message(session_id, user, user_text) # 发送开始标记 await websocket.send_text(json.dumps({ type: assistant_start, payload: {} })) # 流式生成 full_response async for chunk in llm_client.stream_chat( session_manager.get_context(session_id) ): full_response chunk await websocket.send_text(json.dumps({ type: assistant_chunk, payload: {content: chunk} })) await session_manager.add_message(session_id, assistant, full_response) # 发送结束标记 await websocket.send_text(json.dumps({ type: assistant_end, payload: {} })) except WebSocketDisconnect: if session_id: await session_manager.cleanup(session_id) except Exception as e: await websocket.send_text(json.dumps({ type: error, payload: {message: str(e)} })) await websocket.close()3.2 大模型调用的封装大模型调用我封装成一个独立的客户端类支持流式和非流式两种模式。这里以OpenAI兼容接口为例import httpx import json from typing import AsyncGenerator class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.api_key api_key self.base_url base_url self.model model self.client httpx.AsyncClient(timeout60.0) async def stream_chat(self, messages: list) - AsyncGenerator[str, None]: headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: messages, stream: True, temperature: 0.7, max_tokens: 500 } async with self.client.stream( POST, f{self.base_url}/chat/completions, headersheaders, jsonpayload ) as response: async for line in response.aiter_lines(): if line.startswith(data: ): data line[6:] if data [DONE]: break try: chunk json.loads(data) delta chunk[choices][0][delta] if content in delta: yield delta[content] except (json.JSONDecodeError, KeyError): continue这里有几个细节值得注意。第一timeout设置成60秒因为流式响应可能持续较长时间。第二用httpx的stream方法而不是普通post这样才能逐行读取。第三异常处理要细致因为流式数据可能不完整JSON解析可能失败这些都要catch住不能让整个连接崩掉。3.3 前端React组件的组织前端我拆成三个主要组件App、ChatWindow、MessageInput。App负责WebSocket连接管理和全局状态ChatWindow负责消息列表渲染MessageInput负责用户输入。App组件的核心逻辑function App() { const [messages, setMessages] useState([]); const [connected, setConnected] useState(false); const [scene, setScene] useState(restaurant); const wsRef useRef(null); const sessionIdRef useRef(generateSessionId()); useEffect(() { const ws new WebSocket(ws://localhost:8000/ws/chat); wsRef.current ws; ws.onopen () { setConnected(true); ws.send(JSON.stringify({ type: init, payload: { session_id: sessionIdRef.current, scene } })); }; ws.onmessage (event) { const data JSON.parse(event.data); handleMessage(data); }; ws.onclose () setConnected(false); return () ws.close(); }, []); const handleMessage (data) { switch (data.type) { case assistant_start: setMessages(prev [...prev, { role: assistant, content: , streaming: true }]); break; case assistant_chunk: setMessages(prev { const last prev[prev.length - 1]; if (last last.streaming) { return [...prev.slice(0, -1), { ...last, content: last.content data.payload.content }]; } return prev; }); break; case assistant_end: setMessages(prev { const last prev[prev.length - 1]; if (last last.streaming) { return [...prev.slice(0, -1), { ...last, streaming: false }]; } return prev; }); break; case error: console.error(Server error:, data.payload.message); break; } }; const sendMessage (text) { if (!wsRef.current || wsRef.current.readyState ! WebSocket.OPEN) return; setMessages(prev [...prev, { role: user, content: text }]); wsRef.current.send(JSON.stringify({ type: user_message, payload: { content: text } })); }; return ( div classNameapp SceneSelector scene{scene} onChange{setScene} / ChatWindow messages{messages} / MessageInput onSend{sendMessage} disabled{!connected} / /div ); }MessageInput组件需要处理回车发送、Shift回车换行、发送后清空输入框这些细节function MessageInput({ onSend, disabled }) { const [text, setText] useState(); const handleKeyDown (e) { if (e.key Enter !e.shiftKey) { e.preventDefault(); if (text.trim() !disabled) { onSend(text.trim()); setText(); } } }; return ( div classNameinput-area textarea value{text} onChange{(e) setText(e.target.value)} onKeyDown{handleKeyDown} placeholderType your message... disabled{disabled} rows{2} / button onClick{() { if (text.trim() !disabled) { onSend(text.trim()); setText(); } }} disabled{disabled || !text.trim()} Send /button /div ); }3.4 场景切换的实现场景切换需要通知后端更新系统Prompt。我在前端加了一个下拉选择器用户切换场景时发送scene_change消息const changeScene (newScene) { setScene(newScene); setMessages([]); wsRef.current.send(JSON.stringify({ type: scene_change, payload: { scene: newScene } })); };后端收到scene_change后清空当前会话的消息历史用新场景的Prompt重新初始化if data[type] scene_change: new_scene data[payload][scene] await session_manager.reset_session(session_id, new_scene) await websocket.send_text(json.dumps({ type: scene_changed, payload: {scene: new_scene} }))这里有个体验细节切换场景后最好让Agent主动说一句开场白比如餐厅场景说“Welcome to our restaurant! What can I get for you today?”这样用户知道场景已经切换了也知道该怎么接话。4. 常见问题与排查技巧实录4.1 WebSocket连接问题排查问题一连接建立后立即断开。这个最常见的原因是CORS配置不对。FastAPI的CORSMiddleware默认不允许WebSocket跨域需要显式配置allow_origins。另外注意如果前端用的是localhost:3000后端配置的origin必须完全匹配包括协议和端口。问题二连接建立成功但收不到消息。排查思路是先在浏览器开发者工具的Network面板看WebSocket帧。如果能看到发送的帧但收不到接收帧说明后端没有正确发送。检查后端代码中websocket.send_text是否在正确的协程中调用有没有被异常吞掉。我遇到过一次是因为在异步生成器里抛了异常但异常被外层catch住了导致消息没发出去。问题三心跳包不生效。检查setInterval是否在组件卸载时清理了否则会内存泄漏。另外注意如果页面在后台标签页浏览器可能会节流定时器心跳间隔可能变长。这种情况可以监听visibilitychange事件页面回到前台时立即发一次心跳。问题四重连后消息重复。这是因为重连时没有重置消息列表。我的做法是在重连成功后发送一个sync请求后端返回当前会话的完整消息列表前端直接替换本地状态。4.2 流式输出卡顿的优化流式输出最常见的体验问题是“卡顿”表现为文字不是均匀蹦出来而是一阵一阵的。原因通常有两个一是前端setState太频繁二是网络传输有缓冲。前端优化前面提过了用buffer加定时flush。后端优化则是确保每次send_text后不要做耗时操作让事件循环尽快处理下一个chunk。另外httpx的stream默认有缓冲可以设置limitshttpx.Limits(max_keepalive_connections5)来减少连接开销。还有一个容易被忽略的点如果用了Nginx做反向代理需要配置proxy_buffering off否则Nginx会缓冲WebSocket消息导致流式效果消失。4.3 模型回复不符合教学预期有时候模型会忘记自己的“教师”身份回复得过于随意或者不纠正用户的错误。这是Prompt工程的问题。我的经验是第一在系统Prompt里用明确的指令比如“你必须在每次回复中至少指出一个语言错误”而不是“你可以指出错误”。模型对“必须”的遵循度远高于“可以”。第二给模型提供few-shot示例。在Prompt里放两三轮对话示例展示期望的回复风格。这比单纯用文字描述有效得多。第三如果模型仍然不听话可以在后端加一层后处理。比如检测用户消息中是否有语法错误如果有但模型没纠正就追加一条系统消息提醒模型。4.4 常见问题速查表问题现象可能原因排查方法解决方案连接立即断开CORS配置错误看浏览器控制台报错配置allow_origins收不到消息后端异常被吞加日志打印send前后检查异常处理逻辑心跳不生效定时器未清理检查useEffect返回值清理setInterval流式卡顿setState太频繁React DevTools看渲染次数buffer定时flush模型不纠错Prompt不够明确打印实际Prompt用“必须”few-shot重连消息重复未重置状态看重连后消息列表重连后sync会话场景切换无效后端未重置会话看后端日志reset_session清空历史长对话变慢上下文过长看token数量滑动窗口摘要压缩4.5 几个踩过的坑坑一WebSocket的readyState判断。发送消息前一定要检查ws.readyState WebSocket.OPEN否则在连接关闭后发送会抛异常。我一开始没加这个判断用户快速连续发送消息时偶尔会报错。坑二异步生成器的异常处理。在async for循环中如果生成器内部抛异常外层try/except能捕获但WebSocket连接可能已经处于半关闭状态。我的做法是在生成器内部catch异常yield一个错误提示文本而不是让异常冒泡。坑三React的闭包陷阱。在WebSocket的onmessage回调中访问state拿到的是建立连接时的旧值。解决方案是用useRef保存最新值或者用函数式setState。坑四会话ID的生成。不要用Math.random()碰撞概率虽然低但存在。我用的是crypto.randomUUID()浏览器原生支持生成的是标准UUID。坑五模型API的速率限制。免费额度的API通常有QPS限制如果多个用户同时对话可能触发限流。我的做法是加一个简单的令牌桶限流器每个会话每秒最多发一次请求超出的请求排队等待。5. 部署与性能优化建议5.1 本地开发环境搭建后端需要Python 3.10以上推荐用venv创建虚拟环境python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn httpx pydantic python-dotenv前端用Vite创建React项目npm create vitelatest frontend -- --template react cd frontend npm install npm run dev启动后端uvicorn app.main:app --reload --host 0.0.0.0 --port 8000注意--host 0.0.0.0是必须的否则只能本机访问。如果前端和后端不在同一台机器还需要配置防火墙放行端口。5.2 生产环境部署要点生产环境我建议用Docker Compose编排前端用Nginx做静态文件服务后端用Gunicorn加Uvicorn Worker。Nginx配置WebSocket代理的关键几行location /ws/ { proxy_pass http://backend:8000; 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_buffering off; }proxy_read_timeout要设长一点因为WebSocket连接可能保持很久。proxy_buffering off是流式输出的关键。后端用Gunicorn启动gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000Worker数量建议是CPU核数的2倍加1。但注意WebSocket连接是有状态的如果用了多Worker同一个会话的多次请求可能落到不同Worker上导致会话状态不一致。解决方案是把会话状态存到Redis所有Worker共享。5.3 性能监控与日志生产环境一定要加日志和监控。我在关键路径上都打了日志WebSocket连接建立/关闭、消息收发、模型调用耗时、异常堆栈。日志用结构化格式方便后续用ELK或Loki查询。监控指标我关注这几个当前活跃WebSocket连接数、平均消息响应时间、模型调用成功率、会话平均轮次。这些指标可以反映系统的健康状态和用户的使用情况。如果发现响应时间变长优先排查模型API的延迟。可以在日志里记录每次模型调用的开始和结束时间算出差值。如果模型API本身慢考虑换更快的模型或者加缓存。6. 后续扩展方向这个项目的基础版本跑通后可以往几个方向扩展。语音输入输出集成浏览器的Web Speech API让用户可以直接说话Agent的回复也可以用语音合成播报。这样更接近真实的对话体验。多模态教学在对话中插入图片比如餐厅场景显示菜单图片用户指着图片点餐。这需要前端支持图片渲染后端支持图片消息类型。学习进度追踪记录用户常犯的错误类型生成学习报告。这需要在后端加一个错误分析模块定期汇总用户的语法错误、词汇使用情况。多人对话场景支持多个用户同时进入一个场景比如模拟小组讨论。这需要WebSocket支持广播会话管理要支持多用户。离线模式用WebLLM或类似方案在浏览器本地跑小模型不依赖后端API。适合网络不稳定或者对隐私要求高的场景。我个人在实际操作中的体会是做Agent类产品技术实现只占三成七成精力要花在Prompt调优和用户体验打磨上。模型能力再强如果Prompt写得不好回复也会很生硬。反过来即使模型一般精心设计的Prompt和交互流程也能做出不错的效果。另外WebSocket的稳定性比想象中脆弱网络切换、休眠唤醒、代理超时都可能导致断连重连逻辑一定要做扎实否则用户聊到一半突然没反应体验会非常差。 从零构建英语情景教学Agent：WebSocket流式对话与Prompt工程实战
1. 项目缘起与整体架构设计1.1 为什么我要做这么一个东西事情的起因很简单我有个朋友在做英语培训他跟我吐槽说现在市面上的英语学习产品要么是纯背单词的要么是录播课学生开口说英语的机会少得可怜。真人外教一对一效果确实好但成本摆在那里一节课动辄两三百普通家庭很难长期负担。他问我能不能用技术手段做一个“能跟人对话的英语陪练”最好还能根据场景来教学比如点餐、问路、面试这些具体情境。这个需求其实很明确需要一个能理解上下文、能扮演角色、能纠正错误的对话系统。大语言模型天然适合干这个事但光有模型还不够你得把它包装成一个完整的产品——前端要有聊天界面后端要管理会话状态模型调用要稳定可靠还得考虑流式输出让对话体验更自然。这就是我决定从零到一开发一个英语情景教学Agent的完整动机。这个项目适合谁看如果你已经会一点Python和JavaScript想了解怎么把大模型能力落地成一个真实可用的产品那这篇文章就是写给你的。我会把整个开发过程中的技术选型、踩过的坑、关键代码实现都摊开来讲尽量做到你照着做就能跑起来。1.2 技术栈选型与背后的考量先说说我为什么选这套技术栈。前端用React后端用FastAPI通信层用WebSocket模型调用走Python的异步生态。这个组合不是拍脑袋定的每一个选择都有具体的理由。React做前端核心原因是它的组件化模型非常适合聊天界面这种“消息列表输入框状态管理”的结构。每条消息是一个组件消息列表是一个容器组件输入框是受控组件整个数据流非常清晰。而且React生态里有大量现成的UI库和工具开发效率高。2026年的React生态已经非常成熟Hooks API让状态管理变得直观不需要引入额外的状态管理库就能搞定大部分场景。FastAPI做后端主要是看中它的异步支持和WebSocket原生集成。FastAPI基于Starlette对WebSocket的支持是一等公民级别的写起来非常顺手。另外它的依赖注入系统和Pydantic模型验证让代码组织很干净接口文档自动生成调试起来方便。相比DjangoFastAPI更轻量更适合这种以API和实时通信为主的项目。WebSocket做通信这是关键决策。英语对话教学的核心体验是“实时感”如果用传统的HTTP轮询每次发消息都要重新建立连接延迟高不说服务端也没法主动推送消息。WebSocket建立一次连接后可以双向通信模型生成的内容可以逐字推送到前端用户看到的就是打字机效果体验好很多。而且WebSocket的心跳机制可以保持连接活跃避免长时间不操作后断连。Python异步生态因为大模型调用本身是IO密集型的用异步可以同时处理多个用户的请求不会因为一个用户的模型调用阻塞其他用户。FastAPI的async/await语法和Python的asyncio库配合得很好整个后端可以做到高并发。1.3 整体架构长什么样整个系统的架构可以分成四层第一层是前端展示层React应用负责渲染聊天界面、管理用户输入、维护消息列表状态。用户打开页面后前端会尝试建立WebSocket连接连接成功后进入对话状态。第二层是通信层WebSocket连接承载所有实时消息。前端发送用户消息后端推送模型生成的回复。心跳机制保证连接不会因为空闲而断开。第三层是业务逻辑层FastAPI应用处理WebSocket消息管理会话上下文调用大模型API处理流式响应。这一层还负责场景切换、错误处理、会话超时清理等逻辑。第四层是模型服务层实际的大模型调用。我选择的是通过API调用的方式而不是本地部署模型因为API调用更稳定不需要维护GPU资源成本也更可控。这四层之间的数据流是这样的用户在React界面输入一句话前端通过WebSocket发送一个JSON消息到后端后端解析消息后从会话存储中取出历史上下文拼接成Prompt发给大模型大模型流式返回结果后端每收到一个token就通过WebSocket推送给前端前端逐字渲染出来。2. 核心细节解析与实操要点2.1 WebSocket连接的生命周期管理WebSocket连接不是建立后就一劳永逸的它有自己的生命周期需要仔细管理。我把整个生命周期分成四个阶段连接建立、消息收发、心跳保活、连接关闭。连接建立阶段前端在组件挂载时发起WebSocket连接。这里有个细节需要注意React的StrictMode在开发环境下会故意重复挂载组件导致WebSocket被创建两次。我的处理方式是在useEffect的清理函数中关闭连接并且用一个ref来标记连接是否已经建立避免重复创建。const wsRef useRef(null); const isConnectedRef useRef(false); useEffect(() { if (isConnectedRef.current) return; const ws new WebSocket(ws://localhost:8000/ws/chat); wsRef.current ws; ws.onopen () { isConnectedRef.current true; console.log(WebSocket connected); }; ws.onmessage (event) { const data JSON.parse(event.data); // 处理消息 }; ws.onclose () { isConnectedRef.current false; console.log(WebSocket closed); }; return () { if (wsRef.current) { wsRef.current.close(); isConnectedRef.current false; } }; }, []);消息收发阶段前端发送的消息格式我定义成JSON包含type和payload两个字段。type标识消息类型比如user_message表示用户发送的对话内容scene_change表示切换教学场景。payload携带具体数据。后端返回的消息也是类似结构type为assistant_message时payload里是模型生成的内容片段。心跳保活阶段这是很多人容易忽略的地方。WebSocket连接如果长时间没有数据传输中间的网络设备比如负载均衡器、代理服务器可能会主动断开连接。我设置的心跳间隔是30秒前端每隔30秒发送一个ping消息后端收到后回复pong。如果前端连续三次没有收到pong就认为连接已断开触发重连逻辑。const heartbeatInterval setInterval(() { if (wsRef.current wsRef.current.readyState WebSocket.OPEN) { wsRef.current.send(JSON.stringify({ type: ping })); } }, 30000);连接关闭阶段需要区分正常关闭和异常关闭。正常关闭时清理定时器、重置状态即可。异常关闭时需要触发重连我设置的重连策略是指数退避第一次1秒后重连第二次2秒第三次4秒最多重试5次。如果5次都失败就在界面上提示用户检查网络。2.2 会话上下文的管理策略英语情景教学Agent的核心能力是“记住对话历史”这样才能进行连贯的对话。但大模型的上下文窗口是有限的不能无限追加历史消息。我的策略是滑动窗口加摘要压缩。具体来说每个会话维护一个消息列表包含用户和助手的对话记录。当消息数量超过20条时把最早的10条拿出来调用模型生成一段摘要然后把摘要作为系统消息放在最前面后面保留最近的10条消息。这样既保留了长期记忆又控制了上下文长度。会话数据存在哪里我用的是内存字典加定期持久化。内存字典的key是session_idvalue是会话对象。每隔5分钟把活跃会话写入SQLite数据库服务重启后可以从数据库恢复。对于生产环境建议用Redis做会话存储性能和可靠性都更好。class SessionManager: def __init__(self): self.sessions {} self.lock asyncio.Lock() async def get_session(self, session_id: str) - Session: async with self.lock: if session_id not in self.sessions: self.sessions[session_id] Session(session_id) return self.sessions[session_id] async def add_message(self, session_id: str, role: str, content: str): session await self.get_session(session_id) session.messages.append({role: role, content: content}) if len(session.messages) 20: await self._compress_history(session)2.3 流式输出的实现细节流式输出是提升对话体验的关键。用户发送消息后如果等模型生成完整回复再显示可能要等好几秒体验很差。流式输出让用户看到文字一个个蹦出来感觉就像真人在打字。后端实现流式输出的核心是异步生成器。调用大模型API时开启stream模式API会返回一个异步迭代器每产生一个token就yield出来。FastAPI的WebSocket send_text方法可以逐个发送这些token。async def stream_response(session_id: str, user_message: str): session await session_manager.get_session(session_id) messages session.get_context_messages() messages.append({role: user, content: user_message}) async for chunk in llm_client.stream_chat(messages): yield chunk前端接收流式数据时需要维护一个“当前正在生成的消息”状态。每收到一个chunk就追加到这个消息的内容后面触发React重新渲染。这里有个性能优化点不要每个chunk都setState可以用一个buffer累积每隔50毫秒批量更新一次减少渲染次数。const bufferRef useRef(); const flushTimerRef useRef(null); ws.onmessage (event) { const data JSON.parse(event.data); if (data.type assistant_message) { bufferRef.current data.payload.content; if (!flushTimerRef.current) { flushTimerRef.current setTimeout(() { setMessages(prev { const last prev[prev.length - 1]; if (last last.role assistant last.streaming) { return [...prev.slice(0, -1), { ...last, content: bufferRef.current }]; } return [...prev, { role: assistant, content: bufferRef.current, streaming: true }]; }); bufferRef.current ; flushTimerRef.current null; }, 50); } } };2.4 教学场景的Prompt工程设计这个Agent跟普通聊天机器人的区别在于“教学”属性。它不只是跟用户对话还要在对话中纠正错误、引导表达、解释语法。这些行为都通过Prompt来定义。我为每个场景设计了一个系统Prompt模板包含角色设定、教学目标、行为规则三个部分。以“餐厅点餐”场景为例你是一位英语口语教练正在模拟餐厅点餐场景。 你的角色是餐厅服务员用户是顾客。 教学目标 1. 引导用户使用完整的句子点餐 2. 在用户表达错误时先自然回应再指出错误并给出正确说法 3. 适当引入新词汇比如appetizer、recommendation、medium rare 行为规则 - 始终用英语对话除非用户明确要求用中文解释 - 每次回复不超过3句话保持对话节奏 - 如果用户沉默超过两轮主动提问引导这个Prompt的关键在于“先自然回应再指出错误”。如果用户说“I want eat pizza”直接纠正会打断对话流畅感。更好的方式是先回应“Sure, one pizza coming up. By the way, we usually say I want to eat pizza or Id like to eat pizza.”这样既完成了对话又完成了教学。3. 实操过程与核心环节实现3.1 后端项目结构与核心代码FastAPI项目的目录结构我习惯这样组织backend/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── config.py # 配置管理 │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py # Pydantic模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── session.py # 会话管理 │ │ ├── llm_client.py # 大模型调用 │ │ └── prompt.py # Prompt模板 │ ├── api/ │ │ ├── __init__.py │ │ └── websocket.py # WebSocket路由 │ └── utils/ │ ├── __init__.py │ └── logger.py # 日志工具 ├── requirements.txt └── run.pymain.py是应用入口负责创建FastAPI实例、注册路由、配置CORSfrom fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.websocket import router as ws_router app FastAPI(titleEnglish Teaching Agent) app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.include_router(ws_router, prefix/ws) app.get(/health) async def health_check(): return {status: ok}WebSocket路由是核心处理连接建立、消息接收、消息发送、连接关闭from fastapi import APIRouter, WebSocket, WebSocketDisconnect import json import asyncio from app.services.session import session_manager from app.services.llm_client import llm_client from app.services.prompt import get_system_prompt router APIRouter() router.websocket(/chat) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() session_id None try: while True: raw await websocket.receive_text() data json.loads(raw) if data[type] ping: await websocket.send_text(json.dumps({type: pong})) continue if data[type] init: session_id data[payload][session_id] scene data[payload].get(scene, restaurant) await session_manager.init_session(session_id, scene) await websocket.send_text(json.dumps({ type: init_ack, payload: {session_id: session_id} })) continue if data[type] user_message: user_text data[payload][content] await session_manager.add_message(session_id, user, user_text) # 发送开始标记 await websocket.send_text(json.dumps({ type: assistant_start, payload: {} })) # 流式生成 full_response async for chunk in llm_client.stream_chat( session_manager.get_context(session_id) ): full_response chunk await websocket.send_text(json.dumps({ type: assistant_chunk, payload: {content: chunk} })) await session_manager.add_message(session_id, assistant, full_response) # 发送结束标记 await websocket.send_text(json.dumps({ type: assistant_end, payload: {} })) except WebSocketDisconnect: if session_id: await session_manager.cleanup(session_id) except Exception as e: await websocket.send_text(json.dumps({ type: error, payload: {message: str(e)} })) await websocket.close()3.2 大模型调用的封装大模型调用我封装成一个独立的客户端类支持流式和非流式两种模式。这里以OpenAI兼容接口为例import httpx import json from typing import AsyncGenerator class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.api_key api_key self.base_url base_url self.model model self.client httpx.AsyncClient(timeout60.0) async def stream_chat(self, messages: list) - AsyncGenerator[str, None]: headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: messages, stream: True, temperature: 0.7, max_tokens: 500 } async with self.client.stream( POST, f{self.base_url}/chat/completions, headersheaders, jsonpayload ) as response: async for line in response.aiter_lines(): if line.startswith(data: ): data line[6:] if data [DONE]: break try: chunk json.loads(data) delta chunk[choices][0][delta] if content in delta: yield delta[content] except (json.JSONDecodeError, KeyError): continue这里有几个细节值得注意。第一timeout设置成60秒因为流式响应可能持续较长时间。第二用httpx的stream方法而不是普通post这样才能逐行读取。第三异常处理要细致因为流式数据可能不完整JSON解析可能失败这些都要catch住不能让整个连接崩掉。3.3 前端React组件的组织前端我拆成三个主要组件App、ChatWindow、MessageInput。App负责WebSocket连接管理和全局状态ChatWindow负责消息列表渲染MessageInput负责用户输入。App组件的核心逻辑function App() { const [messages, setMessages] useState([]); const [connected, setConnected] useState(false); const [scene, setScene] useState(restaurant); const wsRef useRef(null); const sessionIdRef useRef(generateSessionId()); useEffect(() { const ws new WebSocket(ws://localhost:8000/ws/chat); wsRef.current ws; ws.onopen () { setConnected(true); ws.send(JSON.stringify({ type: init, payload: { session_id: sessionIdRef.current, scene } })); }; ws.onmessage (event) { const data JSON.parse(event.data); handleMessage(data); }; ws.onclose () setConnected(false); return () ws.close(); }, []); const handleMessage (data) { switch (data.type) { case assistant_start: setMessages(prev [...prev, { role: assistant, content: , streaming: true }]); break; case assistant_chunk: setMessages(prev { const last prev[prev.length - 1]; if (last last.streaming) { return [...prev.slice(0, -1), { ...last, content: last.content data.payload.content }]; } return prev; }); break; case assistant_end: setMessages(prev { const last prev[prev.length - 1]; if (last last.streaming) { return [...prev.slice(0, -1), { ...last, streaming: false }]; } return prev; }); break; case error: console.error(Server error:, data.payload.message); break; } }; const sendMessage (text) { if (!wsRef.current || wsRef.current.readyState ! WebSocket.OPEN) return; setMessages(prev [...prev, { role: user, content: text }]); wsRef.current.send(JSON.stringify({ type: user_message, payload: { content: text } })); }; return ( div classNameapp SceneSelector scene{scene} onChange{setScene} / ChatWindow messages{messages} / MessageInput onSend{sendMessage} disabled{!connected} / /div ); }MessageInput组件需要处理回车发送、Shift回车换行、发送后清空输入框这些细节function MessageInput({ onSend, disabled }) { const [text, setText] useState(); const handleKeyDown (e) { if (e.key Enter !e.shiftKey) { e.preventDefault(); if (text.trim() !disabled) { onSend(text.trim()); setText(); } } }; return ( div classNameinput-area textarea value{text} onChange{(e) setText(e.target.value)} onKeyDown{handleKeyDown} placeholderType your message... disabled{disabled} rows{2} / button onClick{() { if (text.trim() !disabled) { onSend(text.trim()); setText(); } }} disabled{disabled || !text.trim()} Send /button /div ); }3.4 场景切换的实现场景切换需要通知后端更新系统Prompt。我在前端加了一个下拉选择器用户切换场景时发送scene_change消息const changeScene (newScene) { setScene(newScene); setMessages([]); wsRef.current.send(JSON.stringify({ type: scene_change, payload: { scene: newScene } })); };后端收到scene_change后清空当前会话的消息历史用新场景的Prompt重新初始化if data[type] scene_change: new_scene data[payload][scene] await session_manager.reset_session(session_id, new_scene) await websocket.send_text(json.dumps({ type: scene_changed, payload: {scene: new_scene} }))这里有个体验细节切换场景后最好让Agent主动说一句开场白比如餐厅场景说“Welcome to our restaurant! What can I get for you today?”这样用户知道场景已经切换了也知道该怎么接话。4. 常见问题与排查技巧实录4.1 WebSocket连接问题排查问题一连接建立后立即断开。这个最常见的原因是CORS配置不对。FastAPI的CORSMiddleware默认不允许WebSocket跨域需要显式配置allow_origins。另外注意如果前端用的是localhost:3000后端配置的origin必须完全匹配包括协议和端口。问题二连接建立成功但收不到消息。排查思路是先在浏览器开发者工具的Network面板看WebSocket帧。如果能看到发送的帧但收不到接收帧说明后端没有正确发送。检查后端代码中websocket.send_text是否在正确的协程中调用有没有被异常吞掉。我遇到过一次是因为在异步生成器里抛了异常但异常被外层catch住了导致消息没发出去。问题三心跳包不生效。检查setInterval是否在组件卸载时清理了否则会内存泄漏。另外注意如果页面在后台标签页浏览器可能会节流定时器心跳间隔可能变长。这种情况可以监听visibilitychange事件页面回到前台时立即发一次心跳。问题四重连后消息重复。这是因为重连时没有重置消息列表。我的做法是在重连成功后发送一个sync请求后端返回当前会话的完整消息列表前端直接替换本地状态。4.2 流式输出卡顿的优化流式输出最常见的体验问题是“卡顿”表现为文字不是均匀蹦出来而是一阵一阵的。原因通常有两个一是前端setState太频繁二是网络传输有缓冲。前端优化前面提过了用buffer加定时flush。后端优化则是确保每次send_text后不要做耗时操作让事件循环尽快处理下一个chunk。另外httpx的stream默认有缓冲可以设置limitshttpx.Limits(max_keepalive_connections5)来减少连接开销。还有一个容易被忽略的点如果用了Nginx做反向代理需要配置proxy_buffering off否则Nginx会缓冲WebSocket消息导致流式效果消失。4.3 模型回复不符合教学预期有时候模型会忘记自己的“教师”身份回复得过于随意或者不纠正用户的错误。这是Prompt工程的问题。我的经验是第一在系统Prompt里用明确的指令比如“你必须在每次回复中至少指出一个语言错误”而不是“你可以指出错误”。模型对“必须”的遵循度远高于“可以”。第二给模型提供few-shot示例。在Prompt里放两三轮对话示例展示期望的回复风格。这比单纯用文字描述有效得多。第三如果模型仍然不听话可以在后端加一层后处理。比如检测用户消息中是否有语法错误如果有但模型没纠正就追加一条系统消息提醒模型。4.4 常见问题速查表问题现象可能原因排查方法解决方案连接立即断开CORS配置错误看浏览器控制台报错配置allow_origins收不到消息后端异常被吞加日志打印send前后检查异常处理逻辑心跳不生效定时器未清理检查useEffect返回值清理setInterval流式卡顿setState太频繁React DevTools看渲染次数buffer定时flush模型不纠错Prompt不够明确打印实际Prompt用“必须”few-shot重连消息重复未重置状态看重连后消息列表重连后sync会话场景切换无效后端未重置会话看后端日志reset_session清空历史长对话变慢上下文过长看token数量滑动窗口摘要压缩4.5 几个踩过的坑坑一WebSocket的readyState判断。发送消息前一定要检查ws.readyState WebSocket.OPEN否则在连接关闭后发送会抛异常。我一开始没加这个判断用户快速连续发送消息时偶尔会报错。坑二异步生成器的异常处理。在async for循环中如果生成器内部抛异常外层try/except能捕获但WebSocket连接可能已经处于半关闭状态。我的做法是在生成器内部catch异常yield一个错误提示文本而不是让异常冒泡。坑三React的闭包陷阱。在WebSocket的onmessage回调中访问state拿到的是建立连接时的旧值。解决方案是用useRef保存最新值或者用函数式setState。坑四会话ID的生成。不要用Math.random()碰撞概率虽然低但存在。我用的是crypto.randomUUID()浏览器原生支持生成的是标准UUID。坑五模型API的速率限制。免费额度的API通常有QPS限制如果多个用户同时对话可能触发限流。我的做法是加一个简单的令牌桶限流器每个会话每秒最多发一次请求超出的请求排队等待。5. 部署与性能优化建议5.1 本地开发环境搭建后端需要Python 3.10以上推荐用venv创建虚拟环境python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn httpx pydantic python-dotenv前端用Vite创建React项目npm create vitelatest frontend -- --template react cd frontend npm install npm run dev启动后端uvicorn app.main:app --reload --host 0.0.0.0 --port 8000注意--host 0.0.0.0是必须的否则只能本机访问。如果前端和后端不在同一台机器还需要配置防火墙放行端口。5.2 生产环境部署要点生产环境我建议用Docker Compose编排前端用Nginx做静态文件服务后端用Gunicorn加Uvicorn Worker。Nginx配置WebSocket代理的关键几行location /ws/ { proxy_pass http://backend:8000; 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_buffering off; }proxy_read_timeout要设长一点因为WebSocket连接可能保持很久。proxy_buffering off是流式输出的关键。后端用Gunicorn启动gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000Worker数量建议是CPU核数的2倍加1。但注意WebSocket连接是有状态的如果用了多Worker同一个会话的多次请求可能落到不同Worker上导致会话状态不一致。解决方案是把会话状态存到Redis所有Worker共享。5.3 性能监控与日志生产环境一定要加日志和监控。我在关键路径上都打了日志WebSocket连接建立/关闭、消息收发、模型调用耗时、异常堆栈。日志用结构化格式方便后续用ELK或Loki查询。监控指标我关注这几个当前活跃WebSocket连接数、平均消息响应时间、模型调用成功率、会话平均轮次。这些指标可以反映系统的健康状态和用户的使用情况。如果发现响应时间变长优先排查模型API的延迟。可以在日志里记录每次模型调用的开始和结束时间算出差值。如果模型API本身慢考虑换更快的模型或者加缓存。6. 后续扩展方向这个项目的基础版本跑通后可以往几个方向扩展。语音输入输出集成浏览器的Web Speech API让用户可以直接说话Agent的回复也可以用语音合成播报。这样更接近真实的对话体验。多模态教学在对话中插入图片比如餐厅场景显示菜单图片用户指着图片点餐。这需要前端支持图片渲染后端支持图片消息类型。学习进度追踪记录用户常犯的错误类型生成学习报告。这需要在后端加一个错误分析模块定期汇总用户的语法错误、词汇使用情况。多人对话场景支持多个用户同时进入一个场景比如模拟小组讨论。这需要WebSocket支持广播会话管理要支持多用户。离线模式用WebLLM或类似方案在浏览器本地跑小模型不依赖后端API。适合网络不稳定或者对隐私要求高的场景。我个人在实际操作中的体会是做Agent类产品技术实现只占三成七成精力要花在Prompt调优和用户体验打磨上。模型能力再强如果Prompt写得不好回复也会很生硬。反过来即使模型一般精心设计的Prompt和交互流程也能做出不错的效果。另外WebSocket的稳定性比想象中脆弱网络切换、休眠唤醒、代理超时都可能导致断连重连逻辑一定要做扎实否则用户聊到一半突然没反应体验会非常差。