使用 Python 从零手写一个多代理协作 (Multi-Agent) 框架
💡 本教程技术要点
不依赖外部框架，用纯 Python + OpenAI API 编写一个支持团队协同的 Agent 系统，实现任务的规划与分发。
一、为什么我们要丢掉 CrewAI 和 AutoGen？
目前市面上有大量的 Multi-Agent（多智能体）框架，但它们都存在严重的过度封装（Over-engineering）问题。很多初学者连大模型的 API 是怎么往返调用的都还没搞懂，就套用了重重类库，结果一旦遇到 Agent 在中间环节陷入无限死循环或者工具参数解析报错，根本无从排查。
要真正搞明白智能体协作的精髓，最好的办法就是用纯 Python，不依赖任何第三方 Agent 库，只基于底层 OpenAI API，手写一个包含任务分发、长短期记忆流转和角色 handoff（交接）的微型多智能体框架。
二、极简 Multi-Agent 架构设计
我们设计一个简单的三角色协作流：
- 编排协调器（Orchestrator）： 接收用户的宏观任务，将其拆解为细分的 Task，并派发给对应角色的 Agent。
- 研究员智能体（Researcher）： 拥有搜索工具，负责收集和筛选事实资料。
- 写作者智能体（Writer）： 接收研究员的草稿，进行内容扩写与排版。
三、核心代码实现
我们用 Pydantic 来规范输入输出，保证智能体返回的是可解析的结构化数据：
import json
from typing import List
from pydantic import BaseModel
from openai import OpenAI
client = OpenAI(api_key="YOUR_KEY")
class Task(BaseModel):
id: int
assignee: str
description: str
dependencies: List[int]
status: str = "pending"
result: str = ""
# 定义智能体的基本状态与系统提示词
AGENT_PROMPTS = {
"Researcher": "你是一位专注于事实挖掘的研究员。你的任务是针对提供的主题收集最核心的数据与事实，过滤废话，直接给出带有数据源的事实提纲。",
"Writer": "你是一位金牌文案主笔。请接收研究员提供的事实提纲，扩写为一篇通顺、有吸引力且无AI腔的文章。不要捏造事实。"
}
class Orchestrator:
def __init__(self, goal: str):
self.goal = goal
self.tasks: List[Task] = []
self.memory = {}
def plan_tasks(self):
# 让大模型扮演规划脑，将总目标拆分为任务列表
prompt = f"""请将以下总目标拆分为具体的子任务列表，并指定分配角色（Researcher 或 Writer）。
目标：{self.goal}
任务拆解必须按逻辑先后进行。请返回 JSON 格式的列表，格式如下：
[{{\"id\": 1, \"assignee\": \"Researcher\", \"description\": \"步骤描述\", \"dependencies\": []}}]"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={{ "type": "json_object" }}
)
data = json.loads(response.choices[0].message.content)
self.tasks = [Task(**t) for t in data["tasks"]]
print("任务规划完成：", [t.description for t in self.tasks])
def execute(self):
while any(t.status == "pending" for t in self.tasks):
for task in self.tasks:
if task.status != "pending":
continue
# 检查依赖项是否都已执行完
if not all(self.tasks[dep - 1].status == "completed" for dep in task.dependencies):
continue
# 搜集依赖任务的上下文数据
context = ""
for dep in task.dependencies:
context += f"\\n【依赖任务 {dep} 的输出结果】:\\n{self.tasks[dep - 1].result}\\n"
# 执行当前任务
print(f"正在指派给 {task.assignee} 执行任务: {task.description}")
task.result = self.call_agent(task.assignee, task.description, context)
task.status = "completed"
print("总目标最终成果：\\n", self.tasks[-1].result)
def call_agent(self, role: str, task_desc: str, context: str) -> str:
system_prompt = AGENT_PROMPTS[role]
user_prompt = f"你当前需要执行的任务是：{task_desc}\\n{context}"
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{{"role": "system", "content": system_prompt}},
{{"role": "user", "content": user_prompt}}
]
)
return response.choices[0].message.content
# 运行 Orchestrator
if __name__ == "__main__":
orchestrator = Orchestrator("研究并撰写一篇关于 Model Context Protocol 在 2026 年最新发展的科技简报")
# 注入模拟的子任务以便演示
orchestrator.tasks = [
Task(id=1, assignee="Researcher", description="搜索并总结关于 MCP 协议在 2026 年的 3 项核心突破技术细节", dependencies=[]),
Task(id=2, assignee="Writer", description="将研究员总结的突破性技术细节整理为一篇 1000 字的干货科技简报", dependencies=[1])
]
orchestrator.execute()四、核心原理与状态管理精要
在上述极简框架中，最核心的理念是显式状态传递（Explicit State Handoff）。每个智能体都在其独立的 API 上下文中运行，相互之间不直接感知对方的存在。Orchestrator 扮演了“共享白板”的角色，它监听整个任务队列的状态，在确认依赖性（Dependencies）满足后，把前一步生成的纯文本结果打包，作为后一步智能体的 User Message 送入。这种确定性的内存流转，避免了代理间因为乱插话、抢发言权而造成的逻辑失控，是工业界构建复杂报表审计智能体（如多层财报审核）时最稳妥的控制架构。
// 典型实现逻辑 / Code outline
// 如需获取该场景下完整可运行的代码库与技术顾问指导，请联系我们
console.log("Loading module: $Python 开发...");
console.log("Configuring agent pipeline: $使用 Python 从零手写一个多代理协作 (Multi-Agent) 框架...");
console.log("Dependencies active. Pipeline initializing...");
// TODO: Custom code hooks for wolaizuo solutions.* 本文为“我来做”动手开发实战教程。如果您不想亲自编写代码，或者需要更深入的企业系统（ERP/CRM）对接与私有化部署，欢迎点击下方按钮预约我们的免费诊断服务。