第 22 期：多 Agent 协作编排

多 Agent 协作模式

graph TB
    User[用户请求] --> Orchestrator[编排 Agent]
    Orchestrator --> |研究任务| Researcher[研究 Agent]
    Orchestrator --> |写作任务| Writer[写作 Agent]
    Orchestrator --> |审核任务| Reviewer[审核 Agent]
    
    Researcher --> |搜索结果| Orchestrator
    Writer --> |初稿| Reviewer
    Reviewer --> |修改意见| Writer
    Writer --> |终稿| Orchestrator
    Orchestrator --> User

方式一：Sub-Workflow 嵌套

在主 Workflow 中调用子 Workflow，每个子 Workflow 扮演一个 Agent 角色。

graph TB
    subgraph "主 Workflow"
        Start[输入: 撰写报告需求] --> SubWF1
        SubWF1[Sub-WF: 数据收集 Agent] --> SubWF2
        SubWF2[Sub-WF: 分析 Agent] --> SubWF3
        SubWF3[Sub-WF: 写作 Agent] --> End[输出: 完整报告]
    end

子 Workflow：数据收集 Agent

# sub-workflow-data-collector.yaml
inputs:
  - topic: string
nodes:
  - type: tool
    name: google_search
    params:
      query: "{{topic}} 最新数据 统计"
  - type: tool
    name: web_scraper
    params:
      url: "{{google_search.first_result_url}}"
  - type: llm
    prompt: "提取以下网页中的关键数据点: {{web_scraper.content}}"
outputs:
  - key_data: "{{llm.text}}"

方式二：Agent 链式调用 (通过 API)

import requests

class AgentOrchestrator:
    """多 Agent 协作编排器"""
    
    def __init__(self):
        self.agents = {
            "researcher": "app-researcher-token",
            "writer": "app-writer-token",
            "reviewer": "app-reviewer-token",
        }
    
    def call_agent(self, agent_name: str, query: str) -> str:
        token = self.agents[agent_name]
        response = requests.post(
            "http://localhost/api/completion-messages",
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json"
            },
            json={
                "inputs": {},
                "query": query,
                "response_mode": "blocking",
                "user": "orchestrator"
            }
        )
        return response.json()["answer"]
    
    def run_pipeline(self, task: str) -> dict:
        # 1. 研究 Agent 收集信息
        research = self.call_agent("researcher",
            f"请对以下主题进行深度调研: {task}")
        
        # 2. 写作 Agent 撰写初稿
        draft = self.call_agent("writer",
            f"基于以下调研资料撰写报告:\n{research}")
        
        # 3. 审核 Agent 审稿
        review = self.call_agent("reviewer",
            f"请审核以下报告并给出修改意见:\n{draft}")
        
        # 4. 写作 Agent 修改终稿
        final = self.call_agent("writer",
            f"根据以下审核意见修改报告:\n审核意见:{review}\n原文:{draft}")
        
        return {
            "research": research,
            "draft": draft,
            "review": review,
            "final_report": final
        }

# 使用
orchestrator = AgentOrchestrator()
result = orchestrator.run_pipeline("2026年 AI Agent 市场趋势分析")
print(result["final_report"])

协作模式对比