三个函数依次调用不叫多智能体。真正值得做的部分是让每个 Agent 有独立收件箱能把任务发给别人、收到修改意见后继续工作并让调用方看到完整通信轨迹。本文用 FastAPI 和asyncio.Queue搭一个可直接运行的最小版本。网上不少“多智能体教程”大概长这样planplanner(task)draftresearcher(plan)resultreviewer(draft)把三个函数改名为 Agent并不会让它们成为一个团队。这段代码仍是一条写死的流水线研究员不能主动找审核员审核员也不能把意见退回去。这次我们做点真的。最终通信路线如下用户 → 规划 Agent → 研究 Agent → 审核 Agent ↑ │ └── 驳回 ──┘ │ 协调 Agent → 返回结果审核 Agent 第一次会故意驳回初稿。研究 Agent 收到意见后修改再次送审。接口响应中还能看到每一条消息经过了谁。为什么选择 FastAPI asyncio.Queue这个示例只需要两个东西FastAPI 提供 HTTP 服务和生命周期管理asyncio.Queue给每个 Agent 一个异步收件箱。Python 官方文档将asyncio.Queue定义为面向async/await代码的异步队列当队列为空时await queue.get()会等待新消息不会用死循环占住 CPU。Python asyncio Queue 文档FastAPI 推荐使用lifespan管理服务启动和关闭所以 Agent 的后台任务也在这里统一启动、取消。FastAPI lifespan 文档先别上消息队列中间件。一个进程内的 Agent 团队用标准库队列就够了。第一步定义 Agent 之间的消息fromdataclassesimportdataclassdataclass(slotsTrue)classMessage:run_id:strsender:strreceiver:strkind:strcontent:strround:int0这里有两个不能省的字段。run_id用来隔离并发请求。两个用户同时提交任务时消息不能串台。receiver明确指定接收者避免多个 Agent 从同一个队列抢消息。第二步做一个最小消息总线classMessageBus:def__init__(self):self.queues{}self.futures{}self.traces{}defregister(self,name):self.queues[name]asyncio.Queue()asyncdefsend(self,message):ifmessage.receivernotinself.queues:raiseValueError(funknown receiver:{message.receiver})self.traces.setdefault(message.run_id,[]).append(asdict(message))awaitself.queues[message.receiver].put(message)每个 Agent 注册一个独立队列。发送消息时总线先记录轨迹再把消息投入接收方队列。这比“共享一个队列让所有 Agent 自己判断消息是不是给我的”更稳。共享队列可能被错误消费者先取走还得重新投递。第三步让每个 Agent 独立运行classAgent:def__init__(self,name,bus):self.namename self.busbus self.taskNonebus.register(name)defstart(self):self.taskasyncio.create_task(self._loop())asyncdef_loop(self):queueself.bus.queues[self.name]whileTrue:messageawaitqueue.get()try:awaitself.handle(message)finally:queue.task_done()asyncdefhandle(self,message):raiseNotImplementedError每个 Agent 都是一个后台协程。没有消息时它停在queue.get()收到消息后调用自己的handle()。Agent 之间没有直接调用彼此的方法。它们只认识消息总线。这样才能保留通信记录也方便以后把进程内队列替换成 Redis Streams、RabbitMQ 或 Kafka。第四步实现“驳回后修改”研究 Agent 接受两类消息第一次研究以及审核员退回的修改意见。classResearcherAgent(Agent):asyncdefhandle(self,message):ifmessage.kindresearch:draftf初稿{message.content}方案采用 FastAPI asyncio.Queue。round_number1else:draft(f修订稿{message.content}补充启动命令、消息轨迹、超时处理和安全边界。)round_numbermessage.roundawaitself.bus.send(Message(message.run_id,self.name,reviewer,review,draft,round_number,))审核 Agent 第一轮不放行而是把意见发回研究 AgentclassReviewerAgent(Agent):asyncdefhandle(self,message):ifmessage.round1:awaitself.bus.send(Message(message.run_id,self.name,researcher,revise,初稿缺少可运行入口和异常边界请修改,2,))returnawaitself.bus.send(Message(message.run_id,self.name,coordinator,final,message.content))到这里双向通信已经发生researcher → reviewer → researcher → reviewer这才是本文和普通串行调用的区别。第五步接入 FastAPI服务启动时创建 Agent 后台任务关闭时统一取消teamAgentTeam()asynccontextmanagerasyncdeflifespan(_:FastAPI):team.start()yieldawaitteam.stop()appFastAPI(title多智能体通信示例,lifespanlifespan)接口只做三件事创建run_id、发送第一条消息、等待协调 Agent 返回最终结果。classTaskRequest(BaseModel):task:strField(min_length1,max_length500)app.post(/tasks)asyncdefcreate_task(request:TaskRequest):try:returnawaitteam.run(request.task)exceptRuntimeErrorasexc:raiseHTTPException(status_code504,detailstr(exc))fromexc输入长度必须有限制Agent 团队也必须设置超时。否则一条异常请求就可能长期占用内存和后台任务。运行项目目录里只有三个文件Python多智能体通信实战/ ├── app.py ├── requirements.txt └── README.md安装依赖python-mpipinstall-rrequirements.txt先运行自检python app.py正常输出自检通过 planner - researcher - reviewer - researcher - reviewer - coordinator再启动服务python-muvicorn app:app--reload打开http://127.0.0.1:8000/docs调用POST /tasks{task:为一个企业知识库设计最小可行方案}响应里除了结果还有完整消息轨迹{run_id:1dd34c...,result:修订稿初稿缺少可运行入口和异常边界请修改……,messages:[{sender:user,receiver:planner,kind:task},{sender:planner,receiver:researcher,kind:research},{sender:researcher,receiver:reviewer,kind:review},{sender:reviewer,receiver:researcher,kind:revise},{sender:researcher,receiver:reviewer,kind:review},{sender:reviewer,receiver:coordinator,kind:final}]}怎么换成真正的大模型 Agent当前示例故意用字符串生成结果因为重点是通信不是绑定某家模型。接入大模型时只改各个 Agent 的handle()把draft ...换成一次模型调用返回结果仍通过bus.send()发给下一个 Agent。消息总线、超时、轨迹和 HTTP 接口都不用动。需要注意不要把 API Key 放进消息不要让外部输入直接变成 Shell 命令给审核轮次设上限涉及文件修改、数据库写入和外部发送时增加人工确认。这个版本故意没做什么它没有数据库所以服务重启后轨迹会消失没有跨进程队列所以不能直接开多个 Uvicorn worker也没有自动发现 Agent。这些不是遗漏。单进程原型先证明消息路线和职责拆分有效再决定是否换 Redis 和持久化。否则很容易搭出一套复杂基础设施最后发现三个 Agent 仍然只是在轮流调用同一个提示词。完整源码已随文章提供。先跑通再把研究员和审核员替换成你实际使用的模型。