第4周 Day 3：多 Agent 协作-如何实现 Agent 团队高效协同作业

在人工智能领域，多Agent协作系统正成为处理复杂任务的高效解决方案。本文将深入探讨如何通过分工协作实现1+1>2的效果。

第4周 Day 3：多 Agent 协作——让 Agent 们"组队干活"

经过短暂休整，我们继续AI学习之旅。前文讨论的都是独立工作的Agent，但现实任务往往需要团队配合。

日常工作中的团队协作随处可见：

1. 内容创作：资料收集→文案撰写→排版校对
2. 项目开发：方案设计→程序编写→测试部署

专业分工能显著提升效率，这正是多Agent系统的核心价值。

---

一、什么是多 Agent 协作？

通过多个Agent协同工作，构建专业化团队处理复杂任务。

	单 Agent	多 Agent 团队
能力	全而不精	术业有专攻
复杂度	简单任务	复杂流程
可维护性	牵一发而动全身	模块化设计
扩展性	功能臃肿	灵活扩展
容错	单点故障	故障隔离

---

二、流水线协作

任务接力模式：前序Agent输出作为后续Agent输入。

研究员(查资料) → 写手(写稿) → 编辑(审校)

适合步骤明确、存在依赖关系的任务场景。

真实运行效果

核心代码

class Agent:
    def __init__(self, name, role, system_prompt):
        self.system_prompt = system_prompt  # 角色定义    def process(self, task):
        return call_llm(self.system_prompt, task)class Researcher(Agent):
    def __init__(self):
        super().__init__(
            name="研究员",
            system_prompt="你是资料研究员，搜集信息并整理成要点列表"
        )class PipelineTeam:
    def collaborate(self, task):
        current_input = task
        for agent in self.members:
            output = agent.process(current_input)
            current_input = output
        return current_input

---

三、代码详解（逐部分讲解）

3.1 API 配置

API_URL = "https://coding.dashscope.aliyuncs.com/v1/chat/completions"
API_HEADERS = {
    "Content-Type": "application/json",
    "Authorization": "Bearer sk-sp-xxxxx"
}
API_MODEL = "glm-5"

参数	作用	类比
API_URL	服务地址	电话号码
API_HEADERS	身份凭证	身份证明
API_MODEL	模型选择	通话对象

3.2 call_llm 函数

def call_llm(system_prompt, user_prompt):
    payload = {
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    }
    response = requests.post(API_URL, headers=API_HEADERS, json=payload)
    return response.json()["choices"][0]["message"]["content"]

功能说明：实现与LLM的交互通信。

参数	作用	示例
system_prompt	角色定义	"资料研究员"
user_prompt	具体指令	"查询仓鼠习性"

典型交互示例：

1. System："你现在是医生"
2. User："头痛怎么缓解？"

3.3 Agent 基类

class Agent:
    def __init__(self, name, role, system_prompt):
        self.name = name
        self.system_prompt = system_prompt
        self.work_log = []    def process(self, task):
        result = call_llm(self.system_prompt, task