企业官网建设流程全解析

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

去年帮朋友的孩子找英语口语练习工具，翻了一圈市面上的产品，发现一个挺尴尬的现状：要么是纯聊天式的AI对话，孩子聊两句就腻了；要么是录播课，根本没有实时反馈。我当时就想，能不能做一个真正能“演”出场景的Agent——比如模拟机场值机、餐厅点餐、酒店入住，让用户在具体情境里用英语完成任务，而不是干巴巴地背单词。

这个想法落地下来，就是今天要聊的英语情景教学Agent。它的核心逻辑不复杂：前端用React做一个对话界面，后端用FastAPI承载Agent的推理和状态管理，中间通过WebSocket做双向实时通信。用户说一句话，Agent不仅回复文字，还会根据当前情景推进剧情、纠正表达、给出提示。整套东西从零到一跑通，我大概花了三周左右的业余时间，踩了不少坑，也积累了一些可以直接复用的经验。

这篇文章适合谁看？如果你有Python基础、了解一点前端，想做一个带实时交互的AI Agent项目，那这篇内容基本可以当施工图来用。如果你只是想了解Agent开发的整体思路，里面关于状态管理、WebSocket心跳、并发处理的部分也值得扫一眼。我不会只讲“怎么做”，更会讲“为什么这么选”——比如为什么用WebSocket而不是SSE，为什么Agent的状态要放在服务端而不是前端，这些决策背后的取舍才是真正值钱的东西。

2. 整体架构设计与技术选型思路

2.1 为什么是React + FastAPI + WebSocket这套组合

先说说技术选型的逻辑。市面上做AI对话应用，常见方案有几种：Gradio快速搭原型、Streamlit做数据类Demo、或者前后端分离自己写。Gradio确实快，但它的交互模式偏“表单提交”，做情景教学需要Agent主动推进剧情、实时纠正发音文本，这种双向流式交互Gradio很别扭。Streamlit同理，它的刷新机制不适合长连接场景。

所以我选了前后端分离：React负责交互层，FastAPI负责Agent逻辑层，WebSocket负责通信层。这个组合的好处是每一层职责清晰，前端可以精细控制消息的渲染时机（比如Agent的“思考中”状态、逐字输出效果），后端可以专注做Agent的状态机和推理调度。

FastAPI在这里的优势特别明显。它是原生异步框架，配合websockets库或者starlette自带的WebSocket支持，处理长连接非常自然。而且FastAPI的依赖注入系统很适合做Agent的会话管理——每个WebSocket连接对应一个会话实例，通过依赖注入把会话状态传给处理函数，代码结构很干净。

React这边，我用的是函数组件加Hooks。核心状态包括：当前情景ID、对话历史、Agent状态（思考中/等待输入/已完成）、连接状态。这些状态用useState和useReducer管理，WebSocket实例放在useRef里避免重复创建。这里有个细节：WebSocket对象不要放在useState里，因为它的变化不需要触发重渲染，放useRef更合适，否则容易引起不必要的组件更新。

2.2 Agent状态放服务端还是前端

这是我在设计阶段纠结最久的问题。最初想的是前端维护对话历史，每次发消息把完整历史传给后端。但很快发现两个问题：一是情景教学需要Agent记住“当前进行到哪个环节”，比如点餐场景里已经点了主菜，接下来该推荐甜点了，这个状态如果放前端，每次都要传回去，逻辑很冗余；二是安全考虑，情景的“标准答案”和评分逻辑不应该暴露在前端。

所以最终方案是：Agent的完整状态（当前情景、已完成步骤、用户表现记录）全部放在服务端，前端只保留渲染需要的对话消息列表。WebSocket连接建立时，服务端创建一个会话对象，用连接ID做索引。每次收到用户消息，服务端更新会话状态，把Agent的回复推回去。这样前端很轻，后端很稳。

代价是服务端需要管理会话生命周期。我用了一个简单的内存字典做会话存储，配合定时清理任务。生产环境肯定要换Redis，但原型阶段内存足够。

2.3 通信协议选WebSocket而不是SSE的理由

热搜词里有人问“react + sse/websocket 轮询文件变化”，这个问题我也遇到过。SSE是单向的，服务端推客户端收，做聊天够用。但情景教学Agent有个特殊需求：用户可能在说话过程中被打断，或者Agent需要主动发起一轮对话（比如“时间到了，我们进入下一个环节”）。这种双向主动推送，SSE做不了，必须WebSocket。

另外WebSocket的延迟更低。实测下来，同样的消息往返，WebSocket比“SSE + HTTP POST”组合快30%到50%，因为省去了每次发消息重新建立HTTP连接的开销。对于口语练习场景，延迟超过500毫秒用户就会觉得“卡”，所以这个优化很关键。

注意：WebSocket虽然好，但它的连接保持需要心跳机制。没有心跳的话，中间的网络设备（路由器、负载均衡）可能在几分钟后静默断开连接，而两端都不知道。这个坑我后面会详细讲怎么填。

3. 后端核心：FastAPI里的Agent状态机与WebSocket管理

3.1 FastAPI项目目录结构怎么搭

先看目录结构，这个是我试了几版之后觉得最顺手的：

english-scenario-agent/ ├── app/ │ ├── main.py # FastAPI入口，WebSocket路由 │ ├── agent/ │ │ ├── __init__.py │ │ ├── scenario.py # 情景定义与加载 │ │ ├── state.py # 会话状态机 │ │ └── engine.py # Agent推理引擎 │ ├── models/ │ │ └── schemas.py # Pydantic模型 │ ├── core/ │ │ ├── config.py # 配置 │ │ └── session.py # 会话管理 │ └── utils/ │ └── heartbeat.py # 心跳处理 ├── scenarios/ # 情景JSON文件 │ ├── airport.json │ ├── restaurant.json │ └── hotel.json ├── requirements.txt └── run.py

这个结构的好处是Agent逻辑和WebSocket传输层解耦。engine.py只负责“给定状态和输入，返回下一步动作”，不关心消息怎么发出去。main.py里的WebSocket处理函数负责收发，调用engine。这样测试的时候可以直接测engine，不用起WebSocket。

3.2 会话状态机的设计

情景教学的核心是“状态机”。每个情景定义一组状态节点，比如餐厅点餐场景：

# scenarios/restaurant.json 的简化结构 { "id": "restaurant", "name": "餐厅点餐", "initial_state": "greeting", "states": { "greeting": { "agent_prompt": "Welcome to our restaurant. Do you have a reservation?", "expected_keywords": ["yes", "no", "reservation"], "next_states": ["check_reservation", "ask_party_size"] }, "ask_party_size": { "agent_prompt": "How many people in your party?", "expected_keywords": ["people", "persons", "two", "three"], "next_states": ["seat_guest"] } } }

服务端的会话对象长这样：

# app/agent/state.py from dataclasses import dataclass, field from typing import List, Dict, Optional @dataclass class SessionState: session_id: str scenario_id: str current_state: str history: List[Dict] = field(default_factory=list) user_scores: List[float] = field(default_factory=list) turn_count: int = 0 is_active: bool = True

每次用户发消息，engine做三件事：判断用户输入是否匹配当前状态的预期、决定下一个状态、生成Agent回复。匹配逻辑我用的是关键词加简单语义相似度，没有上大模型做意图识别，因为情景教学的对话范围是收敛的，关键词足够覆盖80%的情况，而且响应快。

3.3 WebSocket连接的生命周期管理

FastAPI里处理WebSocket的基本写法：

# app/main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.core.session import SessionManager from app.agent.engine import AgentEngine app = FastAPI() session_manager = SessionManager() engine = AgentEngine() @app.websocket("/ws/{client_id}") async def websocket_endpoint(websocket: WebSocket, client_id: str): await websocket.accept() session = session_manager.create_session(client_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": response = await engine.process( session, data["content"] ) await websocket.send_json({ "type": "agent_message", "content": response.text, "state": response.next_state, "score": response.score }) except WebSocketDisconnect: session_manager.remove_session(client_id)

这里有几个关键点。第一，client_id由前端生成并放在URL里，服务端用它做会话索引。第二，心跳消息（ping/pong）在同一个循环里处理，不单独开协程，这样逻辑简单。第三，WebSocketDisconnect异常必须捕获，否则连接断开时会报错。

实操心得：receive_json()在连接断开时会抛异常，但有时候网络抖动不会立刻触发，所以心跳机制是必须的。我试过不加心跳，连接在闲置约90秒后被中间设备断开，前端还以为连着，发消息没反应，用户体验极差。

3.4 心跳机制的具体实现

心跳的核心是：客户端定时发ping，服务端回pong；如果客户端连续几次没收到pong，就认为连接断了，主动重连。

服务端这边，我在main.py里加了一个后台任务，定期检查所有会话的最后活跃时间：

# app/utils/heartbeat.py import asyncio from datetime import datetime, timedelta async def cleanup_stale_sessions(session_manager, timeout_seconds=60): while True: await asyncio.sleep(30) now = datetime.now() stale = [ sid for sid, s in session_manager.sessions.items() if now - s.last_active > timedelta(seconds=timeout_seconds) ] for sid in stale: session_manager.remove_session(sid)

前端那边，用setInterval每20秒发一次ping，同时记录最后一次收到pong的时间。如果超过45秒没收到，就关闭当前连接并重连。这个参数组合是我实测下来比较稳的：20秒间隔不会太频繁，45秒超时能容忍一两次丢包。

4. 前端实现：React里的WebSocket封装与对话渲染

4.1 一个可复用的WebSocket Hook

React里直接用new WebSocket()不是不行，但状态管理很乱。我封装了一个useWebSocketHook，把连接、重连、心跳、消息分发都包进去：

// hooks/useWebSocket.js import { useEffect, useRef, useState, useCallback } from 'react'; export function useWebSocket(url) { const wsRef = useRef(null); const [status, setStatus] = useState('connecting'); const [messages, setMessages] = useState([]); const heartbeatRef = useRef(null); const lastPongRef = useRef(Date.now()); const connect = useCallback(() => { const ws = new WebSocket(url); wsRef.current = ws; ws.onopen = () => { setStatus('connected'); heartbeatRef.current = setInterval(() => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'ping' })); } if (Date.now() - lastPongRef.current > 45000) { ws.close(); } }, 20000); }; ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'pong') { lastPongRef.current = Date.now(); return; } setMessages((prev) => [...prev, data]); }; ws.onclose = () => { setStatus('disconnected'); clearInterval(heartbeatRef.current); setTimeout(connect, 3000); }; }, [url]); useEffect(() => { connect(); return () => { clearInterval(heartbeatRef.current); wsRef.current?.close(); }; }, [connect]); const send = useCallback((msg) => { if (wsRef.current?.readyState === WebSocket.OPEN) { wsRef.current.send(JSON.stringify(msg)); } }, []); return { status, messages, send }; }

这个Hook解决了几个问题：自动重连、心跳保活、消息累积。messages数组直接给对话组件渲染，send函数给输入框用。

注意：useEffect的清理函数里一定要close()，否则React严格模式下组件挂载两次会创建两个连接。这个坑我在开发环境遇到好几次，表现为消息重复发送。

4.2 对话界面的渲染逻辑

对话列表的渲染不复杂，但有个细节：Agent的消息需要支持“逐字输出”效果，模拟打字感。我的做法是前端收到完整消息后，用一个定时器逐字显示，而不是让后端流式发送。原因很简单：后端流式发送需要处理分片、乱序、重传，复杂度高；前端做逐字显示只是视觉效果，实现成本低。

function TypewriterText({ text, speed = 30 }) { const [displayed, setDisplayed] = useState(''); useEffect(() => { let i = 0; const timer = setInterval(() => { if (i < text.length) { setDisplayed(text.slice(0, i + 1)); i++; } else { clearInterval(timer); } }, speed); return () => clearInterval(timer); }, [text, speed]); return <span>{displayed}</span>; }

每个Agent消息用这个组件渲染，用户消息直接显示。消息列表用map遍历，key用消息ID而不是索引，避免重渲染错位。

4.3 情景选择与状态同步

前端还需要一个情景选择界面。用户选好情景后，通过WebSocket发一条start_scenario消息，服务端初始化会话状态并返回第一个Agent提示。这里有个状态同步问题：前端要知道当前进行到哪个环节，才能显示进度条或者提示。

我的方案是服务端在每条agent_message里带上state字段，前端存到currentState里。进度条根据state在情景定义里的顺序计算百分比。情景定义在前端也存一份（只存状态顺序，不存答案），这样进度显示不需要额外请求。

5. 实操全流程：从环境搭建到跑通第一个情景

5.1 后端环境准备与依赖安装

Python环境我用的是3.11，太老的版本对asyncio支持不够好。安装依赖：

python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn websockets pydantic python-dotenv

uvicorn是ASGI服务器，跑FastAPI必备。websockets库其实FastAPI内置了支持，但显式装上避免版本问题。

启动命令：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

--reload开发时用，改代码自动重启。生产环境去掉，用--workers 4起多进程。

实操心得：Windows上如果遇到uvicorn启动报RuntimeError: asyncio.run() cannot be called from a running event loop，大概率是Jupyter环境冲突。直接在终端跑，别在Notebook里跑。

5.2 前端项目初始化

React项目用Vite创建，比create-react-app快很多：

npm create vite@latest english-agent-frontend -- --template react cd english-agent-frontend npm install npm run dev

Vite默认端口5173，后端8000，跨域问题在开发环境用Vite的proxy解决：

// vite.config.js export default defineConfig({ plugins: [react()], server: { proxy: { '/ws': { target: 'ws://localhost:8000', ws: true } } } });

这样前端连ws://localhost:5173/ws/xxx，Vite自动转发到后端。

5.3 定义第一个情景：机场值机

情景JSON我放在scenarios/airport.json：

{ "id": "airport", "name": "机场值机", "initial_state": "check_in", "states": { "check_in": { "agent_prompt": "Good morning. May I see your passport and ticket, please?", "expected_keywords": ["passport", "ticket", "here"], "hints": ["You can say: Here you are."], "next_states": ["ask_destination"] }, "ask_destination": { "agent_prompt": "Where are you flying to today?", "expected_keywords": ["flying", "to", "London", "New York"], "hints": ["You can say: I'm flying to London."], "next_states": ["ask_luggage"] }, "ask_luggage": { "agent_prompt": "How many bags are you checking in?", "expected_keywords": ["bag", "bags", "one", "two"], "hints": ["You can say: Just one bag."], "next_states": ["give_boarding_pass"] }, "give_boarding_pass": { "agent_prompt": "Here's your boarding pass. Gate 12, boarding at 10:30. Have a nice flight!", "expected_keywords": ["thank", "thanks"], "hints": ["You can say: Thank you!"], "next_states": [] } } }

next_states为空表示情景结束。hints是给用户的提示，当用户卡住时显示。

5.4 跑通完整对话流程

启动后端和前端，打开浏览器，选“机场值机”，连接建立后服务端推第一条消息。用户输入“Here you are”，engine匹配到passport和here关键词，状态推进到ask_destination，Agent回复“Where are you flying to today?”。整个过程延迟在200毫秒以内，体验很流畅。

这里有个细节：关键词匹配要做大小写归一化和简单词形还原。用户可能说“Here's my passport”，passport能匹配，但“Here are my passports”里的passports需要还原成passport才能匹配。我用了一个简单的规则：去掉末尾的s再匹配，覆盖大部分情况。

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

6.1 WebSocket连接成功但收不到消息

这是热搜词里出现的问题，我也遇到过。现象是onopen触发了，但onmessage一直不执行。排查下来通常是三个原因：

最常见的是第三个：在组件里多次给ws.onmessage赋值，后面的覆盖前面的。用addEventListener代替直接赋值可以避免。

6.2 心跳机制导致连接频繁断开

我最初的心跳实现是客户端每5秒发ping，服务端回pong。结果发现连接反而更容易断。原因是ping太频繁，加上服务端处理其他消息的延迟，导致pong偶尔超时，客户端误判为断线。

调整后：心跳间隔20秒，超时阈值45秒。这个比例是3:1左右，能容忍两次丢包。另外服务端处理ping要优先，放在消息处理逻辑的最前面，不要被其他耗时操作阻塞。

6.3 Agent状态错乱：用户重复发消息导致状态跳跃

情景教学里，用户可能连续发两条消息。如果服务端处理第一条时状态已经推进，处理第二条时就会基于新状态匹配，导致跳步。我的解决方法是加一个处理锁：每个会话同时只处理一条消息，后续消息排队。

import asyncio class SessionState: def __init__(self): self.lock = asyncio.Lock() async def process(self, session, content): async with session.lock: # 处理逻辑 ...

这个锁很轻量，不会影响性能，但能保证状态推进的原子性。

6.4 前端重连后消息重复

自动重连后，如果服务端还保留着旧会话，新连接可能收到旧消息。我的做法是重连时生成新的client_id，服务端把旧会话标记为失效，新连接创建全新会话。代价是丢失对话历史，但情景教学通常不需要跨连接保留历史，用户重连后重新开始情景反而更合理。

避坑技巧：如果确实需要保留历史，可以在前端用localStorage存消息列表，重连后先渲染本地历史，再接收新消息。但要注意去重，用消息ID判断。

6.5 并发场景下的性能问题

热搜词里有人问“ai agent怎么扛并发”。我这个项目原型阶段用单进程内存会话，实测50个并发连接时CPU占用约30%，延迟还在可接受范围。但如果要上生产，几个优化方向：

• 会话存储换Redis：内存字典在多进程下不共享，换Redis后可以水平扩展。
• Agent推理异步化：如果Agent调用外部模型API，用asyncio.gather并发处理多个会话的推理请求。
• WebSocket连接数限制：单机WebSocket连接数有上限，超过后需要负载均衡。Nginx配置proxy_read_timeout要大于心跳间隔，否则会被Nginx断开。

location /ws/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 300s; }

proxy_read_timeout设300秒，远大于心跳间隔20秒，避免Nginx主动断连。

7. 后续可以扩展的方向

这套框架跑通后，扩展空间挺大。我目前想到几个方向：一是接入语音识别，让用户直接说话而不是打字，WebSocket传音频流，后端做ASR；二是加评分系统，根据用户回复的完整度、语法正确性给分，存到数据库做学习记录；三是多情景串联，比如“机场值机”完成后自动进入“安检对话”，形成完整出行场景。

代码层面，Agent的推理引擎可以换成更复杂的LangGraph状态图，支持分支剧情和条件跳转。但核心的WebSocket通信层和会话管理不用动，这也是当初分层设计的好处——换引擎不影响传输层。

我在实际开发中最大的体会是：先把通信层做稳，再堆业务逻辑。WebSocket的心跳、重连、状态同步这些基础设施如果没做好，后面加再多功能都是空中楼阁。我前两周大部分时间都花在调通信上，但后面加情景、加评分都很顺，因为底层稳了。