初面网初面网

多代理系统

多代理系统（Multi-Agent System）是指由多个独立的 Agent 组成的系统，它们各司其职、相互协作，共同完成复杂任务。

为什么需要多代理

单一 Agent 的局限：

  • 能力有限，什么都做但什么都不精
  • 上下文有限，复杂任务容易丢失重点
  • 难以处理需要专业知识的任务

多代理的优势：

  • 专业化：每个代理专注特定领域
  • 可扩展：可以灵活添加新代理
  • 容错性：一个代理失败不影响整体
  • 更自然：模拟真实团队的协作方式

多代理架构模式

1. 顺序执行模式

任务按顺序传递给各个代理：

User → Agent A → Agent B → Agent C → Output

适用场景：流水线式任务，每一步依赖上一步结果

class SequentialAgents:
    def __init__(self, agents):
        self.agents = agents  # [agent1, agent2, agent3]

    def run(self, task):
        result = task
        for agent in self.agents:
            result = agent.run(result)
        return result

# 示例：写作流程
writing_pipeline = SequentialAgents([
    planner_agent,      # 规划文章结构
    writer_agent,       # 撰写内容
    editor_agent,       # 编辑校对
])

article = writing_pipeline.run("写一篇关于 AI 的科普文章")

2. 并行执行模式

多个代理同时处理任务的不同部分：

        ┌─→ Agent A ──┐
User → ─┤─→ Agent B ──┼─→ 汇总 → Output
        └─→ Agent C ──┘

适用场景：任务可分解，结果需要汇总

import concurrent.futures

class ParallelAgents:
    def __init__(self, agents):
        self.agents = agents

    def run(self, task):
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = [executor.submit(agent.run, task) for agent in self.agents]
            results = [f.result() for f in concurrent.futures.as_completed(futures)]
        return self.aggregate(results)

    def aggregate(self, results):
        # 汇总结果
        pass

# 示例：多角度分析
analysis_team = ParallelAgents([
    technical_agent,    # 技术角度
    business_agent,     # 商业角度
    market_agent,       # 市场角度
])

report = analysis_team.run("分析 AI 对软件行业的影响")

3. 层级模式

上下级代理，层层汇报：

        ┌─→ 子代理 1
主代理 ─┤─→ 子代理 2
        └─→ 子代理 3

适用场景：复杂任务的分解与协调

class HierarchicalAgents:
    def __init__(self, manager, workers):
        self.manager = manager  # 主代理
        self.workers = workers  # 子代理列表

    def run(self, task):
        # 1. 主代理规划任务
        subtasks = self.manager.plan(task)

        # 2. 子代理并行执行
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = [executor.submit(worker.run, subtask)
                      for worker, subtask in zip(self.workers, subtasks)]
            results = [f.result() for f in concurrent.futures.as_completed(futures)]

        # 3. 主代理汇总结果
        return self.manager.summarize(results)

# 示例：项目管理
project_team = HierarchicalAgents(
    manager=project_manager_agent,
    workers=[
        frontend_agent,
        backend_agent,
        tester_agent,
    ]
)

project_team.run("开发一个电商网站")

4. 讨论模式

代理之间相互讨论，达成共识：

       ┌──────┐
   ┌───┤Agent A├──┐
   │   └──────┘  │
User┤        ┌───▼──┐    ┌──────┐
   └──┬──────┤Agent B├──┤      │
      │      └───────┘    │ 投票 │
      └───────────────────┤      │
                          └──────┘

适用场景：需要多方观点综合的决策

class DebateAgents:
    def __init__(self, agents, max_rounds=3):
        self.agents = agents
        self.max_rounds = max_rounds

    def run(self, task):
        # 初始化：每个代理提出观点
        messages = [{"role": "user", "content": task}]

        for round in range(self.max_rounds):
            # 每个代理轮流发言
            for agent in self.agents:
                response = agent.run(messages)
                messages.append({"role": "assistant", "content": response})

        # 最终投票决定
        return self.vote(messages)

    def vote(self, messages):
        # 基于所有讨论做出最终决定
        pass

代理间通信

1. 共享消息队列

from queue import Queue

class MessageQueue:
    def __init__(self):
        self.queue = Queue()

    def send(self, from_agent, to_agent, message):
        self.queue.put({
            "from": from_agent,
            "to": to_agent,
            "content": message
        })

    def receive(self, agent_name):
        messages = []
        while not self.queue.empty():
            msg = self.queue.get()
            if msg["to"] == agent_name:
                messages.append(msg["content"])
        return messages

2. 共享内存

class SharedMemory:
    def __init__(self):
        self.data = {}

    def write(self, key, value):
        self.data[key] = value

    def read(self, key):
        return self.data.get(key)

    def read_all(self):
        return self.data

3. 状态同步

class AgentState:
    def __init__(self):
        self.states = {}

    def update(self, agent_name, state):
        self.states[agent_name] = state

    def get_all_states(self):
        return self.states

    def get_state(self, agent_name):
        return self.states.get(agent_name)

实战：CrewAI

CrewAI 是一个流行的多代理框架：

from crewai import Agent, Task, Crew

# 定义代理
researcher = Agent(
    role="研究员",
    goal="收集 AI 领域的最新信息",
    backstory="你是一个专业的技术研究员",
    verbose=True
)

writer = Agent(
    role="作家",
    goal="将研究内容写成通俗易懂的科普文章",
    backstory="你是一个科普作家，擅长用生动的语言解释技术",
    verbose=True
)

# 定义任务
research_task = Task(
    description="调研 AI Agent 的最新发展",
    agent=researcher,
    expected_output="一份详细的调研报告"
)

write_task = Task(
    description="将调研报告改写成科普文章",
    agent=writer,
    expected_output="一篇 2000 字的科普文章"
)

# 创建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process="sequential"  # sequential 或 hierarchical
)

result = crew.kickoff()
print(result)

实战：AutoGen

微软的 AutoGen 框架：

from autogen import ConversableAgent, GroupChat, GroupChatManager

# 创建代理
assistant = ConversableAgent(
    name="assistant",
    system是一个助手，帮助用户完成任务。_message="你",
    llm_config={"model": "gpt-4"}
)

critic = ConversableAgent(
    name="critic",
    system_message="你是一个评论家，负责审查和提出改进建议。",
    llm_config={"model": "gpt-4"}
)

# 创建群聊
group_chat = GroupChat(
    agents=[assistant, critic],
    messages=[],
    max_round=5
)

# 创建管理器
manager = GroupChatManager(groupchat=group_chat)

# 启动对话
assistant.initiate_chat(
    manager,
    message="写一首关于春天的诗，然后让评论家点评"
)

多代理最佳实践

1. 角色定义清晰

# ❌ 模糊
agent = Agent(role="助手", goal="帮助用户")

# ✅ 清晰
agent = Agent(
    role="技术顾问",
    goal="为用户提供专业的技术建议",
    backstory="你有 10 年软件开发经验，熟悉 AI、云计算等领域"
)

2. 控制代理数量

  • 2-4 个代理通常足够
  • 太多代理会增加通信复杂度
  • 考虑用分层结构而非扁平结构

3. 设计清晰的流程

任务输入 → 明确分工 → 定义接口 → 处理冲突 → 结果汇总

4. 错误处理

class RobustMultiAgent:
    def __init__(self, agents):
        self.agents = agents

    def run(self, task):
        results = {}
        for name, agent in self.agents.items():
            try:
                results[name] = agent.run(task)
            except Exception as e:
                results[name] = f"Error: {str(e)}"

        # 如果所有代理都失败
        if all("Error" in r for r in results.values()):
            return "所有代理都执行失败"

        return self.select_best(results)

适用场景

场景推荐模式示例
内容创作顺序调研 → 写作 → 审核
市场分析并行多角度分析 → 汇总
项目管理层级经理 → 开发/测试
决策制定讨论多方辩论 → 投票

总结

多代理系统的核心思想是"分而治之"：

  • 专业化：每个代理专注特定任务
  • 协作化：代理之间有效沟通
  • 可扩展：灵活添加新代理

选择合适的架构模式，让多个 Agent 协同工作，可以处理远超单个 Agent 能力的复杂任务。

更新于 2026/3/3

上一页
工具调用实战