1. 为什么我要手搓一个英语情景教学Agent

去年年底我在帮几个朋友准备雅思口语，发现一个很尴尬的事：市面上练口语的产品要么是固定脚本的对话机器人，翻来覆去就那几句"How are you"、"I'm fine thank you"；要么是通用大模型套壳，聊着聊着就跑题，你问它餐厅点餐，它跟你扯到量子力学。真正能"进入一个情景、扮演一个角色、围绕一个任务把对话推进下去"的产品，少得可怜。

这就是我动手做这个英语情景教学Agent的起点。它的核心目标很明确：用户选一个场景（比如"机场值机"、"酒店投诉"、"面试自我介绍"），Agent扮演场景里的NPC（地勤、前台、面试官），用英语跟用户完成一段有目标、有反馈、有纠错的对话。对话结束后，Agent还要给出语法、用词、流利度三个维度的点评。

技术栈我选了FastAPI + WebSocket + React这套组合。后端FastAPI负责Agent编排、大模型调用、会话状态管理；WebSocket负责把Agent的流式回复实时推到前端；前端React负责场景选择、对话界面、实时字幕和纠错高亮。为什么不用轮询？因为对话场景对延迟极其敏感，用户说完一句话等三秒才看到回复，体验直接崩掉。WebSocket的全双工特性天然适合这种"边说边推"的场景。

这篇文章我会把整个开发过程拆开讲：架构怎么设计、WebSocket心跳和断线重连怎么处理、Agent的Prompt怎么编排、并发怎么扛、前端React怎么组织状态。适合有Python和React基础、想做一个真实可用的AI Agent项目的同学。没有基础也能看，我会把关键概念用生活化的例子讲清楚。

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

2.1 三层架构：前端、网关、Agent核心

整个系统我分成三层，这个分层不是拍脑袋定的，是被实际需求逼出来的。

第一层是React前端，负责交互和展示。它要处理的东西比想象中多：场景列表渲染、麦克风录音、WebSocket连接管理、流式文字逐字显示、纠错标注高亮。这些状态如果全塞在一个组件里，代码会烂成一锅粥。所以我用Zustand做全局状态管理，把"连接状态"、"当前会话"、"消息列表"、"纠错结果"拆成独立的store。

第二层是FastAPI网关，它是整个系统的中枢。所有WebSocket连接在这里建立，所有HTTP接口（场景列表、历史记录、用户配置）也在这里暴露。网关层不直接调大模型，而是把请求转发给Agent核心，同时负责鉴权、限流、会话路由。

第三层是Agent核心，包含Prompt编排、对话状态机、大模型调用、纠错分析。这一层我单独抽成一个模块，不跟WebSocket耦合，原因是：Agent逻辑未来可能要复用到其他入口（比如小程序、命令行工具），如果跟WebSocket绑死，复用成本极高。

三层之间用明确的接口通信：前端通过WebSocket发JSON消息给网关，网关通过内部函数调用Agent核心，Agent核心返回异步生成器（async generator），网关再把生成器的每个chunk推回前端。这个设计的关键点是Agent核心返回的是异步生成器而不是完整字符串，这样才能实现流式输出。

2.2 为什么是FastAPI而不是Django

热词里有人问"python django websocket实现后台有数据前端推送"，我实际对比过两个方案。

Django本身是同步框架，WebSocket要靠Channels这个额外组件，而Channels又依赖ASGI服务器（Daphne或Uvicorn），配置链路长，调试起来一层套一层。更麻烦的是Django的ORM是同步的，在异步的WebSocket消费者里调用同步ORM，要么用sync_to_async包一层，要么忍受性能损耗。

FastAPI原生就是ASGI框架，async def直接写，WebSocket支持是内置的，不需要额外装东西。而且FastAPI的依赖注入系统（Depends）在WebSocket场景下也能用，鉴权、数据库会话注入都很自然。我实测下来，同样一个"接收消息-调用模型-流式返回"的链路，FastAPI的代码量大概是Django Channels的一半。

当然Django也有优势，比如自带Admin、ORM成熟、生态庞大。如果你的项目已经有Django存量代码，硬切FastAPI不划算。但这是从零开始的新项目，FastAPI是更顺手的选择。

2.3 为什么不用Gradio

热词里有"gradio和fastapi"的对比。Gradio确实能几分钟搭一个AI Demo，拖几个组件、绑一个函数就完事。但它的问题在于定制能力太弱。我想要的是：对话气泡带纠错高亮、场景卡片有封面图、录音按钮有波形动画、连接断了要自动重连。这些Gradio要么做不了，要么要写一堆hack。

Gradio适合做内部演示和快速验证，不适合做面向用户的产品。我的判断标准很简单：如果这个界面未来要上线给真实用户用，就别用Gradio，老老实实写React。

2.4 目录结构：一个能长期维护的FastAPI项目长什么样

热词里"fastapi项目目录结构"被搜了很多次，我把自己用的结构贴出来，这个结构是踩过坑之后定下来的：

english-scenario-agent/ ├── app/ │ ├── main.py # FastAPI入口，注册路由和WebSocket │ ├── config.py # 配置管理，读环境变量 │ ├── api/ │ │ ├── routes_scenario.py # 场景相关HTTP接口 │ │ └── routes_history.py # 历史记录接口 │ ├── ws/ │ │ ├── connection.py # WebSocket连接管理器 │ │ └── handler.py # WebSocket消息处理逻辑 │ ├── agent/ │ │ ├── orchestrator.py # Agent编排主逻辑 │ │ ├── prompts.py # Prompt模板 │ │ ├── state.py # 对话状态机 │ │ └── evaluator.py # 纠错与评分 │ ├── models/ │ │ └── schemas.py # Pydantic模型 │ └── services/ │ └── llm_client.py # 大模型客户端封装 ├── frontend/ # React项目 ├── tests/ └── requirements.txt

这个结构的关键是按职责分层而不是按技术分层。很多教程喜欢建controllers/、services/、utils/这种目录，结果一个业务逻辑散落在四五个文件夹里。我按api、ws、agent分，每个目录内部自洽，改一个功能基本只动一个目录。

3. WebSocket实时通信的核心细节

3.1 连接管理器：为什么不能只存一个连接

新手最容易犯的错是全局存一个WebSocket对象。单用户测试没问题，一上多用户就串线——A用户的消息推给了B用户。正确做法是用一个连接管理器，用字典维护用户ID -> 连接的映射。

# app/ws/connection.py from fastapi import WebSocket from typing import Dict import asyncio class ConnectionManager: def __init__(self): self.active: Dict[str, WebSocket] = {} self.lock = asyncio.Lock() async def connect(self, user_id: str, ws: WebSocket): await ws.accept() async with self.lock: self.active[user_id] = ws async def disconnect(self, user_id: str): async with self.lock: self.active.pop(user_id, None) async def send(self, user_id: str, message: dict): ws = self.active.get(user_id) if ws: await ws.send_json(message) manager = ConnectionManager()

这里加asyncio.Lock是有原因的。FastAPI的WebSocket处理是并发的，两个协程同时读写self.active字典，在极端情况下会出问题。加锁虽然有一点点性能开销，但换来的是数据一致性，值得。

3.2 心跳机制：为什么连接会"假死"

热词里"websocket心跳机制实现"和"websocket连接但不接受信息"这两个问题，本质是同一件事：TCP连接看起来还在，但实际上已经断了。

这种情况在移动网络下特别常见。用户手机从WiFi切到4G，运营商的NAT网关会把旧连接悄悄回收，但客户端和服务端都不知道，双方还傻等着对方发消息。结果就是用户发了一句话，界面一直转圈，永远等不到回复。

解决办法是心跳。服务端每隔30秒发一个ping，客户端收到后回pong。如果服务端连续两次没收到pong，就判定连接已死，主动关闭并通知前端重连。

# app/ws/handler.py import asyncio from fastapi import WebSocket, WebSocketDisconnect async def heartbeat(ws: WebSocket, interval: int = 30): try: while True: await asyncio.sleep(interval) await ws.send_json({"type": "ping"}) except Exception: pass async def ws_endpoint(ws: WebSocket, user_id: str): await manager.connect(user_id, ws) hb_task = asyncio.create_task(heartbeat(ws)) try: while True: data = await ws.receive_json() if data.get("type") == "pong": continue # 处理正常业务消息 await handle_message(user_id, data) except WebSocketDisconnect: pass finally: hb_task.cancel() await manager.disconnect(user_id)

心跳间隔设30秒是个经验值。设太短（比如5秒）会增加无谓的网络流量和电量消耗；设太长（比如120秒）则断线检测不及时，用户要等两分钟才发现连不上。30秒是大多数即时通讯产品的默认值。

3.3 断线重连：前端要做的三件事

后端有心跳还不够，前端必须配合做重连。React这边我用一个自定义Hook封装，核心逻辑是三件事：

第一，监听onclose事件，触发重连。第二，重连要用指数退避，第一次等1秒，第二次等2秒，第三次等4秒，最多等到30秒。为什么？因为如果服务端挂了，所有客户端同时疯狂重连，会把刚恢复的服务端再次打垮。第三，重连成功后要补拉断线期间的消息，否则用户会发现自己漏了几句话。

// frontend/src/hooks/useWebSocket.js import { useEffect, useRef, useCallback } from 'react'; export function useWebSocket(url, onMessage) { const wsRef = useRef(null); const retryRef = useRef(0); const timerRef = useRef(null); const connect = useCallback(() => { const ws = new WebSocket(url); wsRef.current = ws; ws.onopen = () => { retryRef.current = 0; ws.send(JSON.stringify({ type: 'resume', lastId: getLastMsgId() })); }; ws.onmessage = (e) => { const msg = JSON.parse(e.data); if (msg.type === 'ping') { ws.send(JSON.stringify({ type: 'pong' })); return; } onMessage(msg); }; ws.onclose = () => { const delay = Math.min(1000 * 2 ** retryRef.current, 30000); retryRef.current += 1; timerRef.current = setTimeout(connect, delay); }; }, [url, onMessage]); useEffect(() => { connect(); return () => { clearTimeout(timerRef.current); wsRef.current?.close(); }; }, [connect]); return wsRef; }

这里有个细节：onclose里不要直接调connect，要用setTimeout。因为如果连接建立失败，onclose会立刻触发，直接调connect会形成同步递归，把浏览器栈打爆。

3.4 消息协议设计：别用裸字符串

WebSocket传的是文本，但千万别直接传用户说的话。要设计一套消息协议，用JSON包一层，带上type字段区分消息种类。我的协议长这样：

用type区分的好处是前端可以用一个switch统一分发，新增消息类型不用改通信层。而且未来要加"用户打断Agent"、"Agent请求澄清"这类功能，直接加新type就行，协议是向后兼容的。

4. Agent核心：Prompt编排与对话状态机

4.1 情景教学Agent的Prompt该怎么写

这是整个项目最核心的部分。通用聊天机器人的Prompt很简单："你是一个有用的助手"。但情景教学Agent不行，它必须同时满足四个约束：扮演特定角色、维持场景设定、推进对话目标、纠正用户错误。

我的Prompt模板分四块：

# app/agent/prompts.py SCENARIO_PROMPT = """You are role-playing as {role} in a scenario: {scenario}. ## Your Character - Role: {role} - Personality: {personality} - Goal: {role_goal} ## Conversation Rules 1. Always respond in English, keep sentences natural and conversational. 2. Stay in character. Never break the scenario unless the user explicitly asks. 3. If the user makes a grammar or word-choice error, DO NOT interrupt the flow. Instead, note it silently and continue the conversation naturally. 4. Gently steer the conversation toward the scenario goal if the user drifts. 5. Keep each response under 60 words to maintain a natural dialogue pace. ## Scenario Context {context} ## Current Progress Turn: {turn_count} Goal achieved: {goal_achieved} """

这里有几个设计决策值得说。

为什么要求回复不超过60词？因为这是口语练习，不是阅读理解。Agent说一大段，用户根本没机会开口。60词大概是3-4句话，既能推进对话，又给用户留出回应空间。

为什么纠错要"silently note"而不是当场打断？这是教学法的考量。真实对话中，你说话时对方不停纠正你，你会紧张到说不出话。正确做法是让对话自然进行，结束后统一反馈。这跟人类外教的做法一致。

为什么要传turn_count和goal_achieved？因为大模型没有记忆，每次调用都是无状态的。如果不告诉它"已经聊了5轮、目标还没达成"，它可能会在第2轮就草草结束对话。把状态显式传进去，Agent才能有节奏感。

4.2 对话状态机：让Agent知道"现在该干嘛"

光有Prompt还不够，Agent需要一个状态机来管理对话流程。我把一个情景对话分成四个阶段：

开场阶段（opening）：Agent主动打招呼，设定场景。比如"Good morning, welcome to Heathrow Airport. May I see your passport?"

推进阶段（progressing）：用户和Agent来回对话，Agent根据用户回答推进剧情。

收尾阶段（closing）：目标达成，Agent自然结束对话。

评估阶段（evaluating）：对话结束后，Agent切换到评估模式，输出纠错和评分。

# app/agent/state.py from enum import Enum from pydantic import BaseModel from typing import List class Stage(str, Enum): OPENING = "opening" PROGRESSING = "progressing" CLOSING = "closing" EVALUATING = "evaluating" class DialogueState(BaseModel): scenario_id: str stage: Stage = Stage.OPENING turn_count: int = 0 goal_achieved: bool = False history: List[dict] = [] errors: List[dict] = [] def advance(self): self.turn_count += 1 if self.turn_count >= 3 and self.goal_achieved: self.stage = Stage.CLOSING elif self.turn_count >= 8: self.stage = Stage.CLOSING else: self.stage = Stage.PROGRESSING

turn_count >= 8这个兜底很重要。有些用户会一直闲聊不推进目标，如果不设上限，对话永远结束不了，用户会觉得这个Agent"没有边界感"。

4.3 流式输出：async generator是关键

Agent调用大模型时，如果用普通的await等完整回复，用户要等3-5秒才能看到第一个字。体验很差。正确做法是用流式API，模型每生成一个token就推给前端。

# app/agent/orchestrator.py from typing import AsyncGenerator async def run_agent(state: DialogueState, user_input: str) -> AsyncGenerator[dict, None]: state.history.append({"role": "user", "content": user_input}) prompt = build_prompt(state) full_response = "" async for chunk in llm_client.stream(prompt, state.history): full_response += chunk yield {"type": "agent_chunk", "content": chunk} state.history.append({"role": "assistant", "content": full_response}) state.advance() if state.stage == Stage.CLOSING: evaluation = await evaluate_dialogue(state) yield {"type": "agent_done", "evaluation": evaluation} else: yield {"type": "agent_done"}

这个async for配合yield的写法，是FastAPI + WebSocket流式推送的标准模式。WebSocket handler里直接async for chunk in run_agent(...)，每拿到一个chunk就await ws.send_json(chunk)。整条链路全异步，没有阻塞。

4.4 纠错评估：怎么让大模型给出靠谱的反馈

对话结束后的评估是教学价值的核心。我的做法是让大模型返回结构化JSON，而不是自由文本。自由文本没法在前端做高亮和分类展示。

EVAL_PROMPT = """Analyze the following English dialogue between a learner and an NPC. Return a JSON object with this exact structure: { "grammar_errors": [{"original": "...", "corrected": "...", "explanation": "..."}], "vocabulary_suggestions": [{"used": "...", "better": "...", "reason": "..."}], "fluency_score": 0-100, "grammar_score": 0-100, "vocabulary_score": 0-100, "overall_comment": "..." } Dialogue: {dialogue} """

用JSON Schema约束输出，配合大模型的结构化输出能力（比如OpenAI的response_format），能拿到稳定的JSON。如果模型偶尔返回了带markdown代码块的JSON，解析前先strip掉json和。

5. 并发处理与性能优化

5.1 AI Agent怎么扛并发

热词里"ai agent 怎么扛并发"是个真问题。Agent的瓶颈不在WebSocket，而在大模型调用。一次对话请求要占用一个模型连接几秒钟，100个用户同时说话，就是100个并发模型请求。

我的策略是分层限流 + 连接池。

第一层，在WebSocket入口做用户级限流。同一个用户1秒内最多发1条消息，防止有人狂点发送按钮。

第二层，在Agent核心做全局并发控制。用一个asyncio.Semaphore限制同时进行的模型调用数量，比如设成50。超过的请求排队等待，而不是直接拒绝。

# app/agent/orchestrator.py import asyncio LLM_SEMAPHORE = asyncio.Semaphore(50) async def run_agent(state, user_input): async with LLM_SEMAPHORE: # 模型调用逻辑 ...

Semaphore的好处是它不丢请求，只是让请求排队。用户感觉到的只是"回复慢了一点"，而不是"发送失败"。

第三层，模型客户端本身要复用HTTP连接。用httpx.AsyncClient而不是每次请求新建client，能省掉大量TCP握手开销。

5.2 会话状态存哪里

对话状态（DialogueState）不能只放内存。原因有两个：一是服务重启状态就丢了；二是未来要水平扩展，多台服务器要共享状态。

我的方案是内存 + Redis双层。热数据（当前活跃会话）放内存，读写快；同时异步写一份到Redis，用于持久化和跨实例共享。用户重连时，先从内存找，找不到再从Redis恢复。

# app/services/session_store.py import json import redis.asyncio as redis class SessionStore: def __init__(self): self.memory = {} self.redis = redis.from_url("redis://localhost:6379") async def save(self, user_id: str, state: DialogueState): self.memory[user_id] = state await self.redis.setex( f"session:{user_id}", 3600, state.model_dump_json() ) async def load(self, user_id: str) -> DialogueState | None: if user_id in self.memory: return self.memory[user_id] raw = await self.redis.get(f"session:{user_id}") if raw: state = DialogueState.model_validate_json(raw) self.memory[user_id] = state return state return None

setex的3600是过期时间，一小时。用户一小时内没继续对话，会话自动清理，不会无限占内存。

5.3 前端React状态管理：别把所有东西塞进useState

React这边我踩过一个坑：一开始把所有状态都放在顶层组件的useState里，结果每次Agent推一个chunk，整个组件树都重渲染，界面卡得不行。

后来改用Zustand，把状态拆成几个独立的store，每个store只订阅自己关心的部分。Agent推chunk时，只有消息列表那个store更新，场景列表、用户配置这些不受影响。

// frontend/src/store/dialogueStore.js import { create } from 'zustand'; export const useDialogueStore = create((set) => ({ messages: [], streaming: false, currentStreamId: null, appendChunk: (chunk) => set((state) => { const msgs = [...state.messages]; const last = msgs[msgs.length - 1]; if (last && last.id === state.currentStreamId) { last.content += chunk; } else { msgs.push({ id: state.currentStreamId, role: 'agent', content: chunk }); } return { messages: msgs }; }), startStream: (id) => set({ streaming: true, currentStreamId: id }), endStream: () => set({ streaming: false, currentStreamId: null }), }));

这里appendChunk的逻辑是：如果最后一条消息就是当前流式消息，就追加内容；否则新建一条。这样前端就能实现"逐字打字"的效果。

5.4 流式渲染的性能陷阱

逐字追加内容时，如果每条消息都触发一次完整列表重渲染，消息多了会卡。优化手段有两个：

一是用React.memo包裹消息气泡组件，只有内容变化的那个气泡重渲染。

二是对chunk做批量合并。模型每秒可能推几十个chunk，如果每个chunk都setState，React的调度器会被打爆。我的做法是用一个buffer，每50毫秒flush一次。

// frontend/src/hooks/useStreamBuffer.js import { useRef, useEffect } from 'react'; export function useStreamBuffer(onFlush, interval = 50) { const bufferRef = useRef(''); const timerRef = useRef(null); useEffect(() => { timerRef.current = setInterval(() => { if (bufferRef.current) { onFlush(bufferRef.current); bufferRef.current = ''; } }, interval); return () => clearInterval(timerRef.current); }, [onFlush, interval]); return (chunk) => { bufferRef.current += chunk; }; }

50毫秒是人眼感知的阈值。低于这个间隔的更新，人眼分辨不出来，纯属浪费性能。

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

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

这是热词里"websocket连接但不接受信息"的典型症状。我排查过几次，原因基本是这三类：

第一类，消息被中间层吞了。如果前面挂了Nginx，默认配置下WebSocket的升级请求可能被拦截。要在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 3600s; }

proxy_read_timeout默认是60秒，不改的话，连接空闲60秒就被Nginx掐断，表现就是"连上了但过一会儿收不到消息"。

第二类，后端异常被静默吞掉。WebSocket handler里如果抛异常，FastAPI默认会关闭连接但不一定发错误消息。我的做法是在handler外层包一层try/except，捕获后主动发一个{"type": "error"}给前端，前端弹个提示，用户至少知道出问题了。

第三类，前端onmessage没绑对。这个最蠢但最常见。ws.onmessage要在new WebSocket()之后立刻绑，如果放在onopen里绑，会漏掉onopen和onmessage之间到达的消息。

6.2 Agent回复跑题或出戏

大模型偶尔会"出戏"，比如你让它扮演机场地勤，它突然开始跟你聊天气。这是Prompt约束不够强导致的。

我的解决办法是在Prompt里加负面约束，明确列出不能做的事：

## Strict Rules - NEVER mention that you are an AI or a language model. - NEVER switch to Chinese unless the user explicitly asks for translation. - NEVER discuss topics unrelated to the scenario. - If the user asks something off-topic, respond in character and redirect: "Sorry, I can only help with check-in today. Shall we continue?"

负面约束比正面约束更有效，因为大模型对"不要做什么"的遵循度往往高于"要做什么"。

6.3 流式输出时前端闪烁

逐字追加内容时，如果每次都重新计算整个消息列表的高度，界面会抖动。解决办法是给消息容器设固定高度加overflow-y: auto，并且用scrollIntoView时加behavior: 'smooth'，让滚动平滑而不是跳变。

另一个技巧是给正在流式的消息加一个光标动画，用一个CSS伪元素实现，不占用实际DOM节点，性能开销几乎为零。

6.4 常见问题速查表

6.5 几个我踩过的坑

坑一：不要在WebSocket handler里做耗时同步操作。我一开始在handler里直接读文件加载场景配置，结果一个用户读文件时，其他用户的WebSocket消息全被阻塞。后来改成启动时预加载到内存，handler里只读内存。

坑二：asyncio.create_task创建的任务要保存引用。心跳任务如果只create_task不保存，Python的垃圾回收可能会把它回收掉，心跳就停了。要么保存到变量，要么用asyncio.TaskGroup管理。

坑三：前端录音要用MediaRecorder而不是Web Audio API裸写。MediaRecorder是浏览器原生封装，几行代码就能录音并拿到Blob。用Web Audio API自己处理PCM数据，代码量翻十倍，还容易出兼容性问题。

坑四：语音转文字不要在前端做。前端做STT要么用浏览器自带的（兼容性差），要么调第三方API（暴露密钥）。正确做法是把音频Blob通过WebSocket传给后端，后端调STT服务，转写结果再走Agent流程。

7. 前端React的关键实现细节

7.1 场景选择页：卡片式布局与状态预加载

场景选择页看起来简单，但有个细节要注意：用户点击场景卡片后，不要等进入对话页才开始建立WebSocket连接。应该在点击的瞬间就发起连接，同时并行加载场景配置。这样用户进入对话页时，连接已经就绪，Agent的开场白可以立刻推送。

// frontend/src/pages/ScenarioSelect.jsx const handleSelect = async (scenario) => { const wsUrl = `${WS_BASE}/ws/chat?scenario=${scenario.id}`; const ws = new WebSocket(wsUrl); // 把ws实例存到store，对话页直接复用 useConnectionStore.getState().setSocket(ws); navigate(`/chat/${scenario.id}`); };

这个"提前建连"的技巧，能把用户感知的首屏延迟从1-2秒降到几乎为零。

7.2 对话页：消息气泡与纠错高亮

对话页的核心是消息列表。用户消息靠右，Agent消息靠左，这是即时通讯的通用范式，用户不需要学习成本。

纠错高亮是教学产品的特色。对话结束后，Agent返回的grammar_errors里每条都有original字段。前端在渲染用户消息时，用字符串匹配找到这些片段，包一层<mark>标签。

function highlightErrors(text, errors) { let result = text; errors.forEach((err) => { const regex = new RegExp(`\\b${escapeRegex(err.original)}\\b`, 'gi'); result = result.replace(regex, `<mark class="error">$&</mark>`); }); return result; }

注意escapeRegex，用户输入里可能有正则特殊字符，不转义会导致正则报错。

7.3 录音与语音输入

录音用MediaRecorder，核心代码就几行：

const startRecording = async () => { const stream = await navigator.mediaDevices.getUserMedia({ audio: true }); const recorder = new MediaRecorder(stream, { mimeType: 'audio/webm' }); const chunks = []; recorder.ondataavailable = (e) => chunks.push(e.data); recorder.onstop = () => { const blob = new Blob(chunks, { type: 'audio/webm' }); sendAudio(blob); }; recorder.start(); return recorder; };

sendAudio把Blob转成ArrayBuffer，通过WebSocket的二进制帧发出去。后端收到后调STT，转写结果再走正常的文本消息流程。

这里有个兼容性坑：Safari不支持audio/webm，要用audio/mp4。所以mimeType要做特性检测，不能写死。

7.4 连接状态指示器

用户需要知道当前连接是否正常。我在页面顶部放了一个小圆点，绿色表示已连接，黄色表示重连中，红色表示断开。这个指示器直接订阅connectionStore的状态，不需要额外逻辑。

const status = useConnectionStore((s) => s.status); const color = { connected: '#22c55e', reconnecting: '#eab308', disconnected: '#ef4444' }[status];

别小看这个指示器。用户遇到问题时，第一反应是"是不是网断了"，有了指示器，他们能自己判断，减少大量客服咨询。

8. 部署与扩展的一些实际考虑

8.1 单机部署的最小配置

这个项目单机跑起来，2核4G的云服务器足够。FastAPI用Uvicorn起多worker，但要注意：WebSocket连接是有状态的，多worker之间不共享连接。如果起4个worker，用户A连到worker1，用户B连到worker2，他们之间的消息无法互通。

解决办法是用Redis做pub/sub，所有worker订阅同一个channel，消息通过Redis广播。或者更简单：WebSocket服务单独起一个worker，HTTP接口起多个worker，两者分开部署。

8.2 水平扩展时的会话粘性

如果要多台服务器，必须解决会话粘性问题。同一个用户的WebSocket连接必须始终落在同一台服务器上，否则状态会丢。

方案有两种：一是用Nginx的ip_hash做粘性路由；二是把会话状态完全外置到Redis，任何服务器都能处理任何请求。我推荐第二种，虽然每次请求都要读Redis，但扩展性最好，服务器可以随意增减。

8.3 监控要看哪些指标

上线后我重点盯三个指标：WebSocket连接数、模型调用P95延迟、会话完成率。

连接数突然下降，说明可能有断线风暴；模型延迟飙升，说明并发排队严重，要考虑扩容或换更快的模型；会话完成率低，说明Agent的对话设计有问题，用户聊到一半就跑了。

这三个指标我用Prometheus + Grafana做可视化，FastAPI这边用prometheus-fastapi-instrumentator自动埋点，WebSocket的自定义指标手动埋。

8.4 成本控制

大模型调用是主要成本。我的控制手段有三个：一是限制单次对话的最大轮数（8轮），防止用户无限闲聊；二是对简单场景用小模型，复杂场景才用大模型；三是缓存常见场景的开场白，开场白是固定的，不需要每次调模型生成。

开场白缓存这个技巧很实用。一个场景的开场白就那几句，预生成好存起来，用户进入场景直接推缓存，省掉一次模型调用。1000个用户进入同一个场景，就省了1000次调用。

9. 我个人的一些体会

这个项目从想法到能跑起来，大概花了我三个周末。最大的感受是：AI Agent项目的难点不在模型，而在工程。模型能力现在都很强，但要把模型能力稳定、低延迟、可扩展地交付给用户，中间全是工程细节。WebSocket的心跳、重连、消息补拉，会话状态的持久化，流式输出的性能优化，这些才是决定产品能不能用的关键。

另一个体会是Prompt要当代码来管理。我一开始把Prompt硬编码在Python文件里，改一次要重启服务。后来抽成独立的模板文件，支持热加载，调Prompt的效率高了很多。而且Prompt要版本化，每次改动记录一下改了什么、为什么改，不然改着改着就忘了当初为什么那么写。

最后分享一个小技巧：给Agent加一个"调试模式"。在URL上加个?debug=1，前端就会显示每次模型调用的完整Prompt、token消耗、耗时。这个功能在排查"Agent为什么这么回复"时极其有用，强烈建议一开始就加上。

这个项目后续还可以扩展的方向不少，比如加多模态（让Agent能"看到"用户上传的图片）、加多人对话（多个用户和Agent一起练）、加自适应难度（根据用户水平动态调整Agent的用词复杂度）。但那是下一步的事了，先把核心对话链路打磨稳定，比什么都重要。