手写一个Mini MetaGPT:从零搭建多Agent协作框架

📅 2026/7/10 2:04:28 👁️ 阅读次数 📝 编程学习
手写一个Mini MetaGPT:从零搭建多Agent协作框架

上周和一个做AI架构的朋友聊天,他说了一句话让我想了很久:

“现在大家都在用LangChain、MetaGPT这些框架,但多数人只是调API,真遇到业务场景要定制,反而不会了。”

我问他为什么,他说:“因为不知道框架背后是怎么work的。”

所以我决定干一件事——手写一个Mini版本的MetaGPT,把多Agent协作的核心逻辑讲清楚。

一、MetaGPT到底干了什么?

先看一张图,理清MetaGPT的核心架构:

MetaGPT的核心理念就一句话:把软件工程的SOP(标准作业程序)注入到多Agent协作中

传统做法是单个Agent从头写到尾,但MetaGPT的思路是——既然一个软件公司有产品经理、架构师、工程师、测试,为什么不让Agent也各司其职?

每个Agent只干自己最擅长的那件事,然后通过结构化输出串联起来。

二、核心架构拆解

MetaGPT的设计分两层:

2.1 基础组件层

  • 环境(Environment):Agent协作的工作空间和消息总线
  • 记忆(Memory):存储和检索历史消息
  • 角色(Role):封装专业技能和工作流程
  • 行动(Action):执行模块化的子任务
  • 工具(Tool):提供外部能力(搜索、计算等)

2.2 协作层

基础组件之上,MetaGPT提供了两种协作机制:

  1. 知识共享(Knowledge Sharing):Agent之间通过环境日志共享信息
  2. 封装工作流(Encapsulated Workflow):利用SOP将复杂任务拆解为子任务

关键设计思路:每个Agent的输出是标准化的结构化数据,下一个Agent可以直接解析使用。

三、手写Mini MetaGPT

下面我来写一个简化版的多Agent协作框架。完整代码不到200行,但能跑通"需求→代码"的完整流程。

3.1 基础定义

from pydantic import BaseModel, Fieldfrom typing importList, Optional, Dict, Anyfrom openai import OpenAIimport json# ============ 基础数据结构 ============classMessage(BaseModel): """Agent之间的消息""" sender: str receiver: str content: str msg_type: str = "text"# text | structured metadata: Dict[str, Any] = {}classRoleContext(BaseModel): """角色上下文""" name: str profile: str goal: str constraints: str = "" memory: List[Message] = []classAction(BaseModel): """原子行动""" name: str description: str output_schema: Dict[str, Any] = {}classAgent: """基础Agent类""" def__init__(self, context: RoleContext, llm_client: OpenAI, model: str = "gpt-4"): self.context = context self.llm = llm_client self.model = model defthink(self, task: str, context: str = "") -> str: """思考并决定下一步""" prompt = f"""你是一个{self.context.profile}。目标:{self.context.goal}约束:{self.context.constraints}当前任务:{task}上下文:{context}请给出你的专业输出。""" resp = self.llm.chat.completions.create( model=self.model, messages=[{"role": "system", "content": prompt}], temperature=0.7 ) return resp.choices[0].message.content defact(self, thought: str) -> Message: """执行行动并返回消息""" return Message( sender=self.context.name, receiver="environment", content=thought, msg_type="structured" )

实际生产中可以进一步扩展Action的标准化输出模式、重试机制等,但核心逻辑就是:每个Agent用角色提示词驱动 + 标准化输入输出

3.2 环境与消息总线

class Environment: """协作环境——消息总线""" def__init__(self): self.messages: List[Message] = [] self.agents: Dict[str, Agent] = {} defregister(self, agent: Agent): self.agents[agent.context.name] = agent defpublish(self, msg: Message): self.messages.append(msg) print(f"[{msg.sender} → {msg.receiver}]: {msg.content[:80]}...") defget_history(self, for_agent: str = None) -> str: """获取Agent的历史上下文""" relevant = [m for m inself.messages ifnot for_agent or m.receiver in [for_agent, "environment"]] return"\n".join([f"{m.sender}: {m.content}"for m in relevant[-10:]])

3.3 角色定义——虚拟软件团队

# 定义各角色的专业提示词PROFILES = { "product_manager": RoleContext( name="产品经理", profile="资深产品经理,擅长将用户需求转化为PRD", goal="将用户需求转化为详细的产品需求文档", constraints="输出必须包含:功能列表、优先级、验收标准" ), "architect": RoleContext( name="架构师", profile="高级软件架构师,精通系统设计", goal="根据PRD输出技术方案和系统设计", constraints="输出必须包含:技术栈选择、模块划分、接口设计" ), "engineer": RoleContext( name="工程师", profile="全栈工程师,擅长代码实现", goal="根据技术方案编写完整的实现代码", constraints="代码必须可运行,包含模块化和错误处理" ), "qa": RoleContext( name="测试工程师", profile="高级QA工程师,擅长Code Review", goal="审查代码质量,发现潜在问题", constraints="必须检查:语法错误、逻辑漏洞、安全隐患" )}

3.4 工作流引擎——SOP编排

class SOPWorkflow: """标准作业流程编排""" def__init__(self, env: Environment): self.env = env defexecute(self, user_request: str) -> str: # 第一步:产品经理写PRD pm = self.env.agents["产品经理"] pm_thought = pm.think( f"用户需求:{user_request}\n请输出完整的PRD文档" ) self.env.publish(pm.act(pm_thought)) # 第二步:架构师设计技术方案 arch = self.env.agents["架构师"] context = self.env.get_history("架构师") arch_thought = arch.think( "根据PRD输出技术方案", context ) self.env.publish(arch.act(arch_thought)) # 第三步:工程师编写代码 eng = self.env.agents["工程师"] context = self.env.get_history("工程师") code = eng.think( "根据技术方案编写代码,注意代码质量和错误处理", context ) self.env.publish(eng.act(code)) return code

完整的SOP还可以包含QA审查环节和循环修正机制,核心思路是一样的。

3.5 运行入口

def main(): # 初始化LLM客户端(以DeepSeek为例) client = OpenAI( api_key="sk-your-api-key", base_url="https://api.deepseek.com" ) # 创建环境,注册Agent env = Environment() for role in PROFILES.values(): agent = Agent(role, client, model="deepseek-chat") env.register(agent) # 创建SOP工作流,执行 workflow = SOPWorkflow(env) result = workflow.execute("写一个命令行贪吃蛇游戏") print(result)if __name__ == "__main__": main()

代码已通过Python语法校验 ✅。使用时替换为真实的API Key即可运行。

四、MetaGPT的优劣势分析

维度优势劣势
任务分解SOP标准化,输出可控角色固定,难动态扩展
协作效率多Agent并行,各司其职偶尔引用不存在的资源文件
代码质量有QA环节兜底易调用未定义的类或变量
扩展性模块化设计,可替换组件固定流水线,难以插入UI设计等角色

核心洞察:MetaGPT适合需求明确的工程化任务,不适合需要创意探索的场景。

五、为什么理解这些对你重要?

我见过太多团队引入AI框架的方式是这样的:

  1. 装个LangChain/LangGraph
  2. 调几个Agent示例
  3. 发现跟业务对不上
  4. 放弃

问题的根源不是框架不好,而是你不知道框架背后是怎么决策的

当你理解了:

  • Agent的本质是Role + LLM + Memory + Tool
  • 协作的本质是结构化输出 + 消息总线
  • SOP的本质是任务序列 + 标准化接口

你就能在任何一个框架之上快速定制出适合自己业务的方案。

框架是用来"用"的,不是用来"学"的。

学AI大模型的正确顺序,千万不要搞错了

🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!

有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!

就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!

我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费