一聚教程网:一个值得你收藏的教程网站

最新下载

热门教程

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

时间:2026-05-22 08:45:01 编辑:袖梨 来源:一聚教程网

在人工智能领域,多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