从零开始理解 Agent(五):从临时工到正式团队——多智能体协作与编排

举报
AGENT魔方 发表于 2026/04/10 16:24:36 2026/04/10
【摘要】 本文是「从零开始理解 Agent」系列第五篇,围绕多智能体协作与编排展开,以极简开源项目 nanoAgent 为基础,逐层解析 OpenClaw、Claude Code 等 AI Agent 的核心概念。

专栏1.png

欢迎阅读从零开始理解 Agent」系列 —— 我们将从从一个极简开源项目 nanoAgent[1] 出发,逐层拆解 OpenClaw / Claude Code 等 AI Agent 背后的全部核心概念。

★ 项目地址:https://github.com/sanbuphy/nanoAgent

作者:十一

上一篇我们实现了 SubAgent——主 Agent 可以临时派出一个"专家"来干活。但我们也明确定义了 SubAgent 的本质:一次性临时工,生成 → 干活 → 返回摘要 → 消亡,没有身份,没有记忆。

这在很多场景下够用了。但想想现实中的软件开发团队:后端工程师写完 API 后,前端工程师需要知道接口长什么样;测试工程师发现 bug 后,需要告诉开发去修;开发修完后,测试还得再验一遍——同一个人,被多次找到,而且他还记得上次做了什么。

SubAgent 做不到这些。每次调用都是一个全新的、失忆的临时工。

那怎么办?答案是:从临时工升级为正式团队。

★ 关于本篇代码的说明:

和第四篇一样,本篇的 agent-teams.py 是我们新开发的文件GitHub 源码[2]),不在 nanoAgent 原始仓库中。它在 agent-subagent.py 的基础上,用两个类(Agent + Team)实现了多智能体团队协作。

一、临时工 vs 正式员工:差什么?

1.png

从临时工升级为正式团队,需要补齐三样东西:

1. 能跨多轮对话存活的持久智能体 —— Agent 有记忆,被多次 chat() 调用时记得之前做过什么,不会像 SubAgent 那样每次都失忆

2. 身份与生命周期管理 —— Agent 有名字、有角色,被创建(入职)、持续存活(干活)、最终解散(离职),而不是用完即弃

3. 智能体之间的通信通道 —— Agent 之间可以互相发消息(点对点或广播),而不是彼此隔离、互相看不到

接下来看代码怎么实现。


二、核心实现:两个类搞定一切

整个 agent-teams.py 只有 270 行,核心新增是两个类:Agent 和 Team。工具层(read/write/edit/bash)和 Agent 循环完全复用前几篇的代码。

2.1 Agent 类:有状态的持久智能体

先回忆 SubAgent 的实现——一个函数:

# SubAgent(第四篇)—— 一个函数,用完就没
def subagent(role, task):
    sub_messages = [...]  # 局部变量,函数返回即消亡
    for _ in range(10):
        ...
    return result  # 返回后 sub_messages 被垃圾回收,一切归零

现在看 Teams 中的 Agent——一个类:

class Agent:
    def __init__(self, name, role):
        self.name = name                # 身份:有名字
        self.role = role                # 身份:有角色
        self.inbox = []                 # 通信:收件箱
        self.messages = [               # 记忆:持久保持
            {"role": "system", "content": f"You are {name}, a {role}. Be concise and focused."}
        ]

区别只有一个,但意义巨大:messages 从函数的局部变量变成了对象的实例属性。

局部变量在函数返回后就被垃圾回收。实例属性只要对象还活着,就一直在。这意味着你可以对同一个 Agent 多次调用 chat(),每次的对话历史都会累积在 self.messages 中——Agent 记得之前做过什么

2.2 chat( ) 方法:带收件箱的 Agent 循环

def chat(self, task):
    # 第 1 步:如果 inbox 有新消息,先读取并消化
    if self.inbox:
        mail = "\n".join(f"[来自 {m['from']}]: {m['content']}"for m in self.inbox)
        self.messages.append({"role": "user", "content": f"你收到了团队成员的消息:\n{mail}"})
        resp = client.chat.completions.create(model=MODEL, messages=self.messages)
        self.messages.append(resp.choices[0].message)
        self.inbox.clear()

    # 第 2 步:执行本次任务(和之前的 Agent 循环一样)
    self.messages.append({"role": "user", "content": task})
    for _ in range(10):
        response = client.chat.completions.create(model=MODEL, messages=self.messages, tools=tools)
        message = response.choices[0].message
        self.messages.append(message)
        ifnot message.tool_calls:
            return message.content
        for tc in message.tool_calls:
            # ... 执行工具,追加结果(和第一篇完全一样)

关键在第 1 步:每次 chat() 开始前,Agent 会先检查收件箱。如果有其他 Agent 发来的消息,就先读取、消化(让 LLM 处理一下),然后清空收件箱。这样 Agent 在执行任务时,已经知道了队友们的最新进展。

2.3 receive() 方法:通信通道

def receive(self, sender, message):
    self.inbox.append({"from": sender, "content": message})

就这一行。往收件箱里追加一条消息。简单到不需要解释。


三、Team 类:生命周期管理与通信编排

class Team:
    def __init__(self):
        self.agents = {}  # name → Agent

    def hire(self, name, role):
        """招募:创建一个持久 Agent"""
        agent = Agent(name, role)
        self.agents[name] = agent
        return agent

    def send(self, from_name, to_name, message):
        """点对点通信"""
        self.agents[to_name].receive(from_name, message)

    def broadcast(self, from_name, message):
        """广播:给团队所有其他人发消息"""
        for name, agent in self.agents.items():
            if name != from_name:
                agent.receive(from_name, message)

    def disband(self):
        """解散:所有 Agent 生命周期结束"""
        self.agents.clear()

四个方法,对应团队协作的四个动作:

2.png


四、完整协作流程

def run_team(task):
    team = Team()

    # 第 1 阶段:组建团队
    members = plan_team(task)  # LLM 自动拆分角色
    for m in members:
        team.hire(m["name"], m["role"])

    # 第 2 阶段:逐个执行,每人干完广播成果
    for m in members:
        agent = team.agents[m["name"]]
        result = agent.chat(m["task"])
        team.broadcast(m["name"], f"我完成了任务。摘要: {result[:200]}")

    # 第 3 阶段:最后一个成员做二次审查
    reviewer = team.agents[members[-1]["name"]]
    review = reviewer.chat("请根据团队成果做最终审查")

    # 第 4 阶段:解散
    team.disband()

用一个具体例子来说明。假设输入 "创建一个 TODO 应用,包含 Python 后端和 HTML 前端":

[PM] 分析任务,组建团队...
[团队] 3 人:
  1. alice — backend developer → 用 FastAPI 创建 TODO 后端 API
  2. bob — frontend developer → 创建 HTML 前端页面
  3. carol — test engineer → 验证前后端能正常工作

============================================================
  第 1 阶段: 招募团队
============================================================
  [创建] alice (backend developer)
  [创建] bob (frontend developer)
  [创建] carol (test engineer)

============================================================
  第 2 阶段: 协作开发
============================================================

── [1/3] alice 开始工作 ──
  [alice] write({"path": "app.py", ...})
  [alice] → 已创建 app.py,包含 GET/POST/DELETE 三个接口...
  [广播] alice → 全体: 我完成了任务。摘要: 已创建 app.py...

── [2/3] bob 开始工作 ──
  (bob 的 inbox 里有 alice 的广播,他知道后端接口长什么样)
  [bob] write({"path": "index.html", ...})
  [bob] → 已创建 index.html,调用了 alice 定义的 API 接口...
  [广播] bob → 全体: 我完成了任务。摘要: 已创建 index.html...

── [3/3] carol 开始工作 ──
  (carol 的 inbox 里有 alice 和 bob 的广播)
  [carol] read({"path": "app.py"})
  [carol] read({"path": "index.html"})
  [carol] bash({"command": "python -c 'import app; print(\"OK\")'"})
  [carol] → 后端代码语法正确,前端页面已创建,接口调用地址匹配...
  [广播] carol → 全体: 我完成了任务。摘要: 验证通过...

============================================================
  第 3 阶段: carol 做最终审查
============================================================
  (carol 被第二次调用 chat(),她还记得第一次测试的结果)
  [carol] → 最终审查:后端 app.py 包含 3 个接口(GET/POST/DELETE),
            前端 index.html 已正确引用后端地址,代码验证通过,可以交付。

注意 carol 被调用了两次 chat() :第一次做测试,第二次做审查。第二次时她还记得第一次做了什么——这就是"持久记忆"的价值。SubAgent 做不到这一点,因为每次调用都是一个全新的、失忆的函数。


五、三大核心能力的代码对照

回到开头提出的三个要求,逐一对照:

能力 1:能跨多轮对话存活的持久智能体

# SubAgent:局部变量,函数返回即消亡
def subagent(role, task):
    sub_messages = [...]  # 🔴 生命周期 = 这个函数调用
    ...
    return result         # sub_messages 被回收

# Teams Agent:实例属性,对象存活就一直在
class Agent:
    def __init__(self, ...):
        self.messages = [...]  # 🟢 生命周期 = Agent 对象的生命周期

    def chat(self, task):
        self.messages.append(...)  # 每次调用都往同一个列表里追加
        ...
        # 第 1 次 chat():messages = [system, user1, assistant1]
        # 第 2 次 chat():messages = [system, user1, assistant1, user2, assistant2]
        # Agent 在第 2 次时能看到第 1 次的全部历史

能力 2:身份与生命周期管理

team = Team()

# 入职:Agent 被创建,开始存活
alice = team.hire("alice", "backend developer")
bob   = team.hire("bob",   "frontend developer")

# 存活期间:可以多次交互
alice.chat("创建后端 API")
alice.chat("添加认证中间件")   # alice 记得第一次创建的 API

# 解散:所有 Agent 生命周期结束
team.disband()                 # alice、bob 都消亡了

能力 3:智能体之间的通信通道

# 点对点:alice 告诉 bob 接口格式
team.send("alice", "bob", "API 接口: GET /todos, POST /todos")

# 广播:alice 告诉所有人
team.broadcast("alice", "后端已完成,接口文档见 API.md")

# bob 下次 chat() 时,会先读 inbox 中的消息
bob.chat("创建前端页面")  # bob 已经知道了 API 接口格式

▍六、SubAgent vs Teams:什么时候用哪个?

1.png

句话总结:任务简单、互不相关用 SubAgent;需要协作、需要记忆用 Teams


七、系列总结

五篇文章,从一个 115 行的极简 Agent 出发,逐层叠加能力

1.png

从第四篇的 subagent() 函数到第五篇的 Agent 类,变化只有一个:messages 从局部变量变成了实例属性。但这一个变化,让 Agent 从"用完即弃的临时工"进化为了"有记忆、有身份、能协作的团队成员"。

这就是软件工程中最朴素的道理:数据放在哪里,决定了它的生命周期;生命周期决定了能力边界。

但能力越强,副作用也越大——Agent 干的活越多、协作越复杂,messages 就越长。长到撑爆 LLM 的 context window 怎么办?在 第六篇:上下文压缩 中,我们用一个 30 行的函数来解决这个"自我窒息"问题。

本文基于 agent-teams.pyGitHub 源码[2]分析。完整系列:第一篇[3] → 第二篇[4] → 第三篇[5] → 第四篇[6] → 第五篇(本文) → 第六篇


相关链接

[1] nanoAgent: https://github.com/sanbuphy/nanoAgent

[2] GitHub 源码: https://github.com/GitHubxsy/nanoAgent/blob/main/agent-teams.py

[3] 第一篇: https://bbs.huaweicloud.com/blogs/6870960c06054b64b5d7cf3a3759a793

[4] 第二篇: https://bbs.huaweicloud.com/blogs/1f1b0f3bcba8413282c64209ac223a83

[5] 第三篇: https://bbs.huaweicloud.com/blogs/06d0d3ac8fba488fa344eb35b07b3ff9

[6] 第四篇: https://bbs.huaweicloud.com/blogs/1d212d5ebb7347758977d65f1f85666e


容器.jpg

关注 AGENT 魔方公众号,回复 Agent

免费领取「从零开始理解 Agent」全套资料包

加速入门和掌握 Agent:

资料二维码.jpg


【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。