简介这是一套基于Python自研的智能体框架项目源码核心集成Live2D模型交互与动态渲染适用于需要可视化形象与实时对话能力的代理、虚拟助手类应用开发。包内包含完整代码、配置文件及说明文档整体结构围绕异步任务调度、事件驱动架构与WebSocket实时通信展开适合具备一定Python基础的开发者学习或二次扩展。压缩包共570个文件约53.96MB主要类型包括py源码、json配置文件、Live2D模型资源moc/moc3/physics等、mp3语音及png图像素材并附带说明文件与附赠文档便于快速上手。已有52人学习下载资源中Agent-main目录划分清晰涵盖交互、渲染、通信等模块可帮助使用者理解智能体框架的工程实现并在此基础上构建具备动态形象和流畅交互体验的对话代理产品。1. 自研智能体框架可视化形象和对话逻辑为什么值得拆成两层你大概率遇到过这种场景对话后端已经能跑通 LLM 接口回答也有模有样但界面上只有一个干巴巴的文本框。想做一个带 Live2D 形象的对话代理张嘴、眨眼、呼吸都能跟语音或文字联动结果一动手才发现把 Live2D 模型和对话逻辑拼在一起远比想象中麻烦。这个项目标题其实说得很清楚一个自研智能体框架把 Python 的异步任务调度、事件驱动架构、WebSocket 实时通信和 Live2D 动态渲染串成一条完整链路最终建出一个有可视化形象的对话代理或虚拟助手。适合谁给对话机器人、桌面助手、直播间数字人做界面层的 Python 开发者尤其是想把 LLM 回复和形象表现解耦、不想被某个商业产品锁死的团队。2. 事件驱动与异步任务调度先画清消息流再写第一行代码2.1 为什么必须用事件驱动而不是按“用户问一句、模型答一句”的线性流程写对话代理的实时链路里事件来源远不止用户文本。语音识别回调、大模型流式返回的 token、TTS 合成状态、Live2D 前端上报的动画完成通知这些事件的发生时刻互相独立根本没有固定先后关系。如果你用同步函数一层层调用下去一条 LLM 慢请求就能卡住整个形象动画模型嘴巴张到一半就僵住了。这不是网络问题是架构问题。事件驱动的核心思路很简单谁产生了消息谁就把消息扔进一个公共通道调度器根据事件类型决定交给哪个处理器处理器之间不互相调用只通过事件通道通信。把这一层做好之后LLM 卡三秒Live2D 的呼吸眨眼照常跑用户按下了打断按钮调度器可以立刻取消正在执行的对话任务动画进入正在聆听的状态。这种解耦是后面所有功能的地基。2.2 调度器的骨架两个有界队列加一张路由表我一般会先搭一个最小可用的调度器只有三个部分输入队列、输出队列、事件处理器注册表。输入队列接收所有原始事件处理器处理后把渲染指令放进输出队列WebSocket 层只负责把输出队列的内容推给前端。import asyncio import time import uuid from enum import Enum class EventType(str, Enum): USER_SAY user.say # 用户发来一句话 USER_CANCEL user.cancel # 用户打断 LLM_TOKEN llm.token # 大模型流式返回了一个片段 LLM_DONE llm.done # 大模型回复结束 LIVE2D_FINISH live2d.finish # 前端动画播完后的回执 IDLE_BREATH live2d.idle # 空闲呼吸指令 class AgentDispatcher: def __init__(self, max_input128, max_output64): self.input_q asyncio.Queue(maxsizemax_input) self.output_q asyncio.Queue(maxsizemax_output) self.handlers: dict[str, callable] {} self._tasks: set[asyncio.Task] set() def on(self, event_type: str): 注册某个事件类型的处理器装饰器用法 def deco(func): self.handlers[event_type] func return func return deco async def push(self, event: dict): 外部入口把任意来源的事件丢进输入队列 try: self.input_q.put_nowait(event) except asyncio.QueueFull: # 背压处理事件进不来时丢弃并记录不要阻塞调用方 print(f[dispatcher] input queue full, drop {event[type]}) return True async def _run_safe(self, event: dict): handler self.handlers.get(event.get(type)) if handler is None: return try: await handler(event) except asyncio.CancelledError: raise except Exception: # 处理器崩了不能影响调度器主循环 import traceback traceback.print_exc() async def run(self): 主循环从输入队列取事件分发给对应处理器 while True: event await self.input_q.get() task asyncio.create_task(self._run_safe(event)) self._tasks.add(task) task.add_done_callback(self._tasks.discard) async def emit(self, event_type: str, payload: dict): 处理器内部发送新事件用这个避免绕过输入队列 event {type: event_type, id: str(uuid.uuid4()), ts: time.time(), payload: payload} await self.push(event) async def send_to_client(self, item: dict): 输出指令走独立队列方便 WebSocket 层单独消费 try: self.output_q.put_nowait(item) except asyncio.QueueFull: # 输出队列满了说明客户端消费不过来了此时丢弃动画指令比堆内存重要 print(f[dispatcher] output queue full, drop live2d cmd)这段代码有两个细节是容易踩坑的。第一_tasks这个 set 必须存在。用asyncio.create_task创建的任务如果没有强引用Python 的垃圾回收会直接把它收掉事件处理器跑一半就无声无息地消失这是协程任务最常见的黑匣子问题。第二两个队列都设了上限。输入队列满的时候丢弃新事件输出队列满的时候丢弃 Live2D 动画指令宁可少动一下嘴也不能让内存无限涨下去。2.3 慢消费者拖垮一切用信号量和超时控制堵住最贵的资源事件驱动架构跑起来之后最常出现的问题是“什么都想做结果什么都没做完”。LLM 请求是慢操作Live2D 动画播放也是慢操作这条链路里真正稀缺的资源不是 CPU是事件循环线程。一个耗时 10 秒的 LLM 流式请求如果直接挂在 handler 里 await那么这 10 秒内所有 Live2D 指令都被堵住了这就是翻车位。两个手段必须同时上。第一给 LLM 请求加并发限制信号量同一时间最多发出去两三个请求超过的排队或者直接失败。第二给 handler 执行加超时超时后任务取消不能让它永远占着调度循环。# 在 AgentDispatcher 基础上增加并发控制和超时 class AgentDispatcherV2(AgentDispatcher): def __init__(self, *args, llm_concurrency3, **kwargs): super().__init__(*args, **kwargs) self._llm_sem asyncio.Semaphore(llm_concurrency) async def run(self): while True: event await self.input_q.get() task asyncio.create_task(self._run_with_timeout(event, 15.0)) self._tasks.add(task) task.add_done_callback(self._tasks.discard) async def _run_with_timeout(self, event: dict, timeout: float): try: async with asyncio.timeout(timeout): handler self.handlers.get(event.get(type)) if handler: await handler(event) except TimeoutError: print(f[dispatcher] event {event[type]} timeout after {timeout}s, cancelled)asyncio.timeout是 Python 3.11 引入的之前我用asyncio.wait_for实现同样效果。注意超时取消的不是整个任务链它只会取消当前等待的那一段如果 handler 内部还挂着子任务子任务要用asyncio.shield包一层才能在超时后继续跑否则会出现“打断了 LLM 请求但 TTS 音频还在播”的奇怪场景。这种半途取消的语义需要专门测一遍我吃过一次亏打断重说后系统还在播上一轮的音频Live2D 的嘴巴和当前文本完全对不上。2.4 常见误用把 ThreadPoolExecutor 当万能药很多人第一次遇到“LLM 很慢”这个问题会直接把 handler 丢进ThreadPoolExecutor。这个方案在单机脚本里没问题但在事件驱动框架里会埋下两个雷。第一个雷是线程爆炸高峰期一个 LLM 请求都不舍得拒绝线程池撑到几十上百上下文切换开销直接把 CPU 打满。第二个雷是状态共享多线程下访问全局会话状态要加锁加锁一多代码逻辑混乱到后面根本不敢改。协程方案天然规避了这两个问题。单个事件循环里所有并发都是协作式切换没有锁的生存空间慢请求通过异步 HTTP 客户端发出后事件循环会自动去处理其他事件不用你手动切线程。需要线程池的情况只有一个处理 CPU 密集型任务比如音频编解码、图像缩放。这时候用asyncio.to_thread单独包一下其余逻辑全部保持协程。3. WebSocket 实时通信连接生命周期、心跳与流式消息推送的落地配置3.1 连接管理不只是 accept建立连接后先握手再鉴权WebSocket 是连接级协议但智能体框架需要的是“会话级”语义。前端打开页面建立 WebSocket 连接后服务端并不知道这个连接对应哪个用户、哪个会话。我见过不少项目直接把用户 ID 塞进握手路径里比如/ws/room/42这能工作但扩展性和安全性都很差。常见做法是连接建立后客户端先发一条hello消息带上 token 和会话上下文服务端校验通过后回复hello_ack这条连接才算真正可用。import asyncio import json import time from websockets.asyncio.server import serve CLIENTS {} # sid - WebSocket 对象 SESSIONS {} # sid - 会话上下文 async def handle_ws(ws): sid str(uuid.uuid4()) CLIENTS[sid] ws # 阶段 1等待 hello 握手超时 5 秒不发言就断开 try: async with asyncio.timeout(5): hello await ws.recv() hello_data json.loads(hello) token hello_data.get(token, ) # 这里做 token 校验通常查 Redis 或 JWT 解码 await ws.send(json.dumps({type: hello_ack, sid: sid})) except TimeoutError: await ws.close(code4001, reasonhello timeout) return try: async for raw in ws: msg json.loads(raw) if msg.get(type) user.say: # 把用户消息推进调度器输入队列 await dispatcher.push({ type: EventType.USER_SAY, payload: {sid: sid, text: msg.get(text, )} }) elif msg.get(type) user.cancel: await dispatcher.push({type: EventType.USER_CANCEL, payload: {sid: sid}}) except Exception: pass finally: # 连接关闭时清掉客户端同时通知调度器清理会话 CLIENTS.pop(sid, None) await dispatcher.push({type: session.close, payload: {sid: sid}}) async def main(): async with serve(handle_ws, 0.0.0.0, 8765, ping_interval20, ping_timeout20): await asyncio.get_running_loop().create_future()这个握手流程的时序很关键。先建立 TCP/WebSocket 连接再应用层握手意味着中间的 5 秒超时窗口是必须的否则客户端连上不发消息服务端会一直挂着空连接。token 校验放在握手阶段而不是 URL 参数里的原因是URL 会被代理服务记入访问日志token 容易泄露虽然 query 里的 token 也能用但我个人习惯只把 sid 放在 URLtoken 放消息体。3.2 心跳机制参数化服务端超时、客户端间隔、代理空闲断开三级配置WebSocket 心跳是绕不开的坑。浏览器和服务端之间的链路隔着公网、网关、负载均衡任何一个环节都可能静默断开一条空闲连接。很多人的第一反应是“客户端每 10 秒发一条 ping 不就行了”实际运行起来发现收到了服务端TimeoutError。原因是服务端的空闲断连时间可能更短或者协议层 ping 和应用层消息没有区分开。我建议做两层心跳。第一层是协议层用websockets库内置的ping_interval和ping_timeout服务端每 20 秒发一个 WebSocket 协议 ping20 秒内没收到 pong 就判定连接死亡。第二层是应用层客户端每 15 秒发一条{type: ping, ts: ...}的业务心跳服务端收到后原样返回pong同时服务端记录每个连接的最后心跳时间超过 30 秒没有业务心跳的连接即使协议层还活着也主动判定为失活。# 服务端维护心跳状态30 秒无业务心跳则主动断开 HEARTBEAT_TIMEOUT 30.0 last_beat {} async def heartbeat_monitor(): while True: await asyncio.sleep(5) now time.time() dead [] for sid, last in last_beat.items(): if now - last HEARTBEAT_TIMEOUT: dead.append(sid) for sid in dead: ws CLIENTS.get(sid) if ws: await ws.close(code4002, reasonheartbeat timeout) last_beat.pop(sid, None)注意 ping_interval 不能设得太激进。有些云负载均衡器对 WebSocket 有 60 秒的空闲连接回收策略如果服务端 20 秒一个 ping它不会被视为空闲这条链路是安全的。但如果你把 ping_interval 设成 90 秒就可能撞上中间设备的回收时间出现“莫名其妙断线”却抓不到任何后端报错的情况。三级参数关系如下表层级参数推荐值作用客户端业务心跳间隔15 s保活、检测半开连接服务端协议 pingping_interval20 s维持协议层活跃中间设备空闲回收负载均衡空闲超时通常 60 s不触发的前提是上层心跳间隔小于它服务端业务失活heartbeat_timeout30 s清理假活连接释放资源这几组参数要按倍数拉开不能每层都设 10 秒。都设在 10 秒的时候网络抖动一次就可能触发连锁重连。3.3 流式推送LLM token 流怎么拼Live2D 指令怎么插对话代理里最影响体验的是流式返回。用户看到的不应该是“沉默三秒然后整段答案突然出现”而是“文字一个个蹦出来嘴巴跟着在动”。这里的关键是消息协议设计。一个 JSON 消息里如果包含多帧数据解析时容易粘包如果一帧一条 WebSocket 消息频率太高容易把浏览器渲染线程打崩。我通常用“行分帧 JSON”方案# 服务端推送流式 token 的示例在 LLM handler 里调用 async def push_tokens(sid: str, tokens: list[str]): ws CLIENTS.get(sid) if not ws: return # 每 6 个 token 拼一条消息减少 WebSocket 帧数 for i in range(0, len(tokens), 6): chunk tokens[i:i 6] await ws.send(json.dumps({ type: agent.reply_delta, sid: sid, chunk: .join(chunk), ts: time.time() })) await asyncio.sleep(0.01) # 给前端渲染留出喘息空间asyncio.sleep(0.01)看着像玄学实际上是在做背压。本地 WebSocket 发送速度极快如果你把一千个 token 一次性以每条 3 个 token 发出去浏览器在几百毫秒内收到几百条消息Live2D 的口型驱动和文本渲染会一起卡死。稍微压一下帧率反而整体更流畅。文本流之间穿插 Live2D 指令时也建议统一走dispatcher.output_q由同一个 writer 协程消费避免多个协程同时写一个 WebSocket 导致乱序。4. Live2D 模型交互与动态渲染让前端按指令动起来的关键协作4.1 动态渲染的本质不是视频流是状态指令流一开始接触 Live2D 集成的人最容易把“动态渲染”理解成实时画面传输试图把 Live2D 渲染成视频帧再通过 WebSocket 推送。这个方向基本是死路帧率、码率、延迟三个指标互相掐架。Live2D 模型的正确工作方式是模型文件在前端加载后端通过消息推送“表情 ID”“动作 ID”“参数名和值”这类指令前端 SDK 负责把指令实时应用到模型上。后端说的是“让嘴张开 0.6”前端自己把张嘴动画渲染出来渲染压力全在客户端 GPU 上服务端只承担逻辑和调度。模型文件通常包含四类资源model3.json是入口清单、.moc3是模型数据、贴图是纹理资源、.motion3.json和.exp3.json分别是动作和表情。网上有不少免费 Live2D 模型可以拿来做干跑测试但注意检查是否包含全部文件很多模型包只有贴图和 moc3缺动作和表情最后只能干瞪眼。4.2 后端指令协议把 Live2D 的 API 翻译成事件消息要把 Live2D 控制 API 和后端解耦我会在事件总线上定义三类指令状态切换、动作播放、参数直控。状态切换是最高层的语义比如“进入思考状态”“进入聆听状态”动作播放对应模型的一个动画参数直控则是直接改ParamMouthOpenY这种底层参数用于口型同步。# 后端给前端下发的 Live2D 指令消息结构 { type: live2d.state, state: speaking, # idle / listening / thinking / speaking priority: 3 # 优先级低优先级的指令可以被高优先级打断 } { type: live2d.play_motion, motion: tap_head, # 对应模型里的动作组 index: 0, # 同一动作组可能有多个变体 loop: False } { type: live2d.set_param, param: ParamMouthOpenY, value: 0.7, duration_ms: 80 # 前端做插值过渡用的直接跳变会显得很机械 }优先级字段很重要。想象一个场景用户正在说话Live2D 正在点头这时 LLM 返回了一个重要信息需要模型做一个吃惊的表情。如果所有指令一视同仁排队执行等点头动画播完再切表情实时性就崩了。我在每个指令里都带 priority前端 SDK 收到高优先级指令时会打断当前动作直接切换。这里有一个前端决策的细节是否允许打断需要前端自己判断后端只下发指令不做强制这样前端可以根据当前动画的进度决定是立刻切换还是等半秒。4.3 前端集成用 pixi-live2d-display 跑通最小渲染链路前端这一侧常见的做法是用支持 Live2D Cubism 4 的渲染库加载model3.json然后监听 WebSocket 消息把指令映射到模型 API 上。这里给出一个最小前端代码跑通“后端发指令、前端动起来”的闭环import { Live2DModel } from pixi-live2d-display; import { Application } from pixi.js; // 初始化 Pixi 渲染器 const app new Application({ view: document.getElementById(canvas), autoStart: true, resizeTo: window, transparent: true }); // 加载 Live2D 模型model3.json 是入口文件 const model await Live2DModel.from(/assets/avatar/model3.json); app.stage.addChild(model); model.scale.set(0.4); model.y app.screen.height - model.height; // 建立 WebSocket协议层心跳交给库自动处理业务心跳定时器自己发 const ws new WebSocket(ws://localhost:8765/ws); ws.onopen () { ws.send(JSON.stringify({ type: hello, token: your-token })); }; ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type agent.reply_delta) { // 文本流更新 简单口型驱动 updateChatText(msg.chunk); model.internalModel.coreModel.setParameterValueById(ParamMouthOpenY, 0.6); clearTimeout(mouthTimer); mouthTimer setTimeout(() { model.internalModel.coreModel.setParameterValueById(ParamMouthOpenY, 0.0); }, 120); } else if (msg.type live2d.play_motion) { model.motion(msg.motion, msg.index ?? 0); } else if (msg.type live2d.set_param) { model.internalModel.coreModel.setParameterValueById(msg.param, msg.value); } else if (msg.type live2d.state) { // 状态切换时可以联动 UI 和背景色不只是模型动作 handleAgentState(msg.state); } };口型同步这一段setParameterValueById(ParamMouthOpenY, 0.6)是直接张嘴120 毫秒后闭嘴。这算是最原始的口型驱动。想做得更精细可以按文本字数估算整段话的时长然后按字符节奏逐个触发张嘴想做得更真实就需要 TTS 音频驱动的口型数据这是后话了。前端部分有两点值得注意第一模型加载失败要把错误捕捉出来很多模型文件缺少贴图资源时框架不会报明确错误只会在控制台打一条 warning第二Pixi 的 ticker 和 WebSocket 消息都在浏览器主线程上跑消息帧率过高时主线程会忙到掉帧这也是上面说为什么要做服务端节流。4.4 呼吸与待机不要让你的虚拟助手像一座雕塑Live2D 模型如果只在收到指令时才动那静态时看起来就像个死物。真正自然的虚拟助手静态时也应该有呼吸、微转头的动作。这个逻辑我放在后端一个低优先级循环任务里每 3 到 5 秒向前端推一条live2d.play_motion随机播放一个“呼吸/待机”动作。低优先级意味着用户说话或 LLM 回复时这些待机动作会被打断不会干扰主体的表情表演。async def idle_loop(dispatcher, interval4.0): 每 4 秒随机触发一次待机动作只在无高优任务时生效 motions [breath_01, blink_01, look_around] while True: await asyncio.sleep(interval) await dispatcher.send_to_client({ type: live2d.play_motion, motion: random.choice(motions), priority: 0 # 最低优先级任何其他指令都能打断它 })前端收到 priority 为 0 的待机指令时如果当前模型正在执行 priority 更高的动作就忽略它。这个逻辑前端要做不能靠后端保证因为 WebSocket 消息到达前端的时间点不可控可能存在“先发的高优指令比后发的低优指令晚到”这种极端情况。前端对指令做一层优先级过滤等于给整个系统加了保险丝。5. 高频问题与排查5 个让整条链路断掉的血泪坑5.1 坑一WebSocket 空闲 60 秒准时断线日志全是 TimeoutError现象浏览器端 Live2D 好好的后端日志里websocket.exceptions.ConnectionClosedError每隔一段时间就出现一次时间间隔非常规律大概 60 秒左右。原因链路中间有一层负载均衡或网关空闲连接超过 60 秒就被静默回收。协议层 ping 虽然存在但你没开ping_interval连接在闲时没有任何数据包经过中间设备中间设备就把它当死连接清掉了。解决给websockets.serve加上ping_interval20, ping_timeout20。服务端每 20 秒发协议层 ping中间设备看到有数据流动就不会回收连接。同时客户端也要有兜底心跳因为协议层 ping 可能在某些代理下被过滤。两层心跳都设上这条坑才算真正填完。5.2 坑二LLM 一回复整个事件循环卡死好几秒现象用户发一句话后Live2D 动画直接冻结文本流也是顿一顿再蹦一个字整个过程像老电影卡帧。打开 CPU 监控发现只有一个核在忙。原因某一个 handler 里用了同步的 HTTP 请求库比如requests去调 LLM 接口。同步调用会阻塞事件循环线程线程卡住其他所有事件都排队等待。解决换用aiohttp或httpx.AsyncClient做 LLM 请求保持整个调用链都是协程。检查代码里还有没有其他同步点数据库驱动是不是同步的、文件读取有没有阻塞、time.sleep是不是真的应该用asyncio.sleep。这一段排查起来很枯燥但值得做彻底。可以在事件循环里挂一个每秒运行的定时任务记录两次执行的时间差一旦差值大于 500 毫秒就说明有同步操作卡住循环了。5.3 坑三create_task 出来的任务跑着跑着就没了现象调度器主循环正常队列里有事件处理器也注册了但某些事件始终没有被处理也没有任何异常日志。原因asyncio.create_task创建的任务对象没有保持强引用。Python 的事件循环只保存了任务对象的弱引用如果你的自定义代码里没有把 task 存下来垃圾回收会在下一个 GC 周期把它收掉任务还没执行完就被取消了。解决所有 create_task 出来的任务都加到一个 set 里任务完成后再用add_done_callback把自己从 set 里移除。这个 set 的生命周期要跟调度器一样长不能是每次调用函数的局部变量。5.4 坑四张嘴闭嘴很生硬像木头人在配音现象文本流过来了嘴巴也动了但动作只有“开”和“合”两种状态切换瞬间完成毫无过渡整个形象显得非常机械。原因后端发的是set_param指令前端直接用了setParameterValueById做硬赋值没有做插值。而且口型驱动只依赖文本到达事件没有考虑字数节奏和停顿。解决前端为每个 param 变更做一个短时间的线性插值时长 80 到 150 毫秒。口型状态机至少要有三个状态闭嘴、半开、全开根据当前播放到的字符位置切换。如果用 TTS 音频可以进一步分析音频振幅根据响度映射嘴型开合度这种效果最自然但成本也高。如果是纯文本驱动手动做插值和停顿效果比硬赋值好三倍。5.5 坑五输入队列无限增长内存顶到两个 G现象每天早上来看内存发现服务端进程占了几个 G重启后两小时又涨回去。队列长度监控显示持续堆积但单个事件处理时间并不长。原因消费者速度跟不上生产者速度时无界队列会无限吸收事件。而我的输入队列初始版本设了maxsize128你以为有界了但如果put_nowait满了之后没有抛异常处理逻辑写入方又把事件吞掉了队列就会变成实际无界。解决构造函数里明确asyncio.Queue(maxsizeN)写入端用put_nowait捕获QueueFull时根据事件类型执行丢弃或合并策略。比如用户连续说了三句话LLM 还在处理第一句后面的user.say可以丢弃但user.cancel不能丢丢了用户就打断不了当前回复。6. 进阶多客户端压测、事件循环滞后观测与个人调试习惯6.1 用 Python 脚本模拟 50 个客户端压测连接稳定性和 RTT手工测试只能验证单条链路通不通压测要看的是系统性风险。我常用一个 Python 脚本模拟多客户端同时连接同时发消息统计消息往返时延import asyncio import json import time import statistics import websockets async def fake_client(cid: int, uri: str, results: list): async with websockets.connect(uri, ping_interval20, ping_timeout20) as ws: # 业务握手 await ws.send(json.dumps({type: hello, token: test-token})) await ws.recv() # hello_ack latencies [] for i in range(30): t0 time.perf_counter() await ws.send(json.dumps({ type: user.say, text: fclient-{cid}-msg-{i} })) # 等服务端回一条 hello_ack / reply_delta 之外的消息 while True: raw await asyncio.wait_for(ws.recv(), timeout5) msg json.loads(raw) if ack in msg.get(type, ): break latencies.append((time.perf_counter() - t0) * 1000) avg statistics.mean(latencies) p95 sorted(latencies)[int(len(latencies) * 0.95)] results.append((cid, avg, p95)) print(fclient {cid}: avg {avg:.1f}ms, p95 {p95:.1f}ms) async def main(): uri ws://localhost:8765/ws results [] # 分批建连避免 50 个客户端同时握手造成服务端瞬时压力失真 async with asyncio.TaskGroup() as tg: for i in range(10): for _ in range(5): tg.create_task(fake_client(i * 5 _, uri, results)) await asyncio.sleep(0.5) # 汇总打印整体 all_avg [r[1] for r in results] print(ftotal clients: {len(results)}, total avg: {statistics.mean(all_avg):.1f}ms) if __name__ __main__: asyncio.run(main())压测时观察四个指标而不是只看 P95平均 RTT、P95 RTT、连接建立成功率、事件循环滞后时间。RTT 稳定但事件循环滞后持续增长通常说明 CPU 忙不过来连接建立成功率低说明握手阶段超时设置得太保守。6.2 事件循环滞后观测给 asyncio 加一个心跳监控asyncio最大的问题就是“看起来活着实际上卡了”。网上有不少现成的 lag monitor 思路我自己实现的方法是创建一个只负责计时的协程每隔 100 毫秒记录当前事件的循环时间如果上次运行到下次运行的时间差超过 500 毫秒就打一条 warningasync def loop_lag_monitor(threshold0.5): last time.perf_counter() while True: await asyncio.sleep(0.1) now time.perf_counter() lag now - last - 0.1 if lag threshold: print(f[EVENT LOOP LAG] {lag*1000:.0f}ms over {threshold*1000:.0f}ms) last now这个监控在压测期间、上线前跑半小时能暴露绝大多数阻塞问题。如果 lag 稳定在阈值以下再配合上面的多客户端压测系统的并发边界就心里有数了。6.3 个人调试习惯保留完整事件日志而不是只记错误最后说一个习惯也是我踩了无数次坑之后养成的事件驱动的框架里“当时发生了什么”比“现在发生了什么”重要得多。每个事件进队列前后都打一条日志带上event_id、sid、ts三个字段。这样线上出问题时可以用一个简单的脚本过滤同一个event_id把这条事件从进入到处理完成的完整路径拉出来。有一次用户反馈“Live2D 动作和声音完全不匹配”我把事件日志回放了一遍发现是 LLM 返回的 token 顺序乱了——两条并发流式请求写到了同一个会话里。没有完整事件日志这种问题几乎不可能定位。日志别只记错误和警告事件入队和出队各记一条 debug 级别的就够用量不大但价值极高。希望这些踩坑记录和调试习惯能帮你少走一点弯路把智能体框架的这条链路真正稳定跑起来。本文还有配套的精品资源点击获取