多工具集成架构设计：Hub-and-Spoke / Pipeline / P2P 三种拓扑全面对比

              ┌─────────┐
              │   Hub   │
              │ (调度器) │
              └────┬────┘
         ┌─────────┼─────────┐
         ▼         ▼         ▼
    ┌────────┐ ┌────────┐ ┌────────┐
    │Spoke A │ │Spoke B │ │Spoke C │
    │ 编码器  │ │ 测试器  │ │ 审查器  │
    └────────┘ └────────┘ └────────┘

class HubScheduler:
    """星型架构的中心调度器"""

    def __init__(self):
        self.agents = {
            "code_writer": Agent("claude-code", "code"),
            "tester": Agent("codex", "test"),
            "reviewer": Agent("opencode", "review"),
            "doc_writer": Agent("llm", "docs"),
        }
        self.task_queue = asyncio.Queue()
        self.results = {}

    async def dispatch(self, task: Task) -> dict:
        """将任务分解并分发到各 Spoke"""
        subtasks = self.decompose(task)
        futures = {}

        for subtask in subtasks:
            agent = self.agents[subtask.target_agent]
            futures[subtask.id] = agent.execute(subtask)

        # 等待所有子任务完成
        results = await asyncio.gather(*futures.values())
        return self.aggregate(dict(zip(futures.keys(), results)))

    async def handle_failure(self, subtask_id: str, error: Exception):
        """Spoke 失败时的补偿策略"""
        # 重试、降级或通知人工介入
        logger.warning(f"Spoke {subtask_id} failed: {error}")
        await self.retry_or_escalate(subtask_id, error)

    def decompose(self, task: Task) -> list[SubTask]:
        """任务分解策略 - 基于规则或 LLM"""
        if task.type == "feature":
            return [
                SubTask(id="code", target_agent="code_writer", input=task.spec),
                SubTask(id="test", target_agent="tester", input=task.spec),
                SubTask(id="review", target_agent="reviewer", depends_on=["code"]),
                SubTask(id="docs", target_agent="doc_writer", depends_on=["code"]),
            ]
        # 其他任务类型...

    def aggregate(self, results: dict) -> dict:
        """聚合结果 - 冲突检测、质量评分"""
        # 检查代码与测试的兼容性
        if results.get("code") and results.get("test"):
            compat = self.check_compatibility(
                results["code"].output,
                results["test"].output
            )
            results["compatibility"] = compat
        return results

┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
│  Stage 1 │───▶│  Stage 2 │───▶│  Stage 3 │───▶│  Stage 4 │
│ 需求解析  │    │ 代码生成  │    │ 测试验证  │    │ 部署执行  │
└─────────┘    └─────────┘    └─────────┘    └─────────┘

class PipelineStage:
    """流水线阶段基类"""

    def __init__(self, name: str, agent: Agent,
                 validator: Callable = None,
                 retry_count: int = 2):
        self.name = name
        self.agent = agent
        self.validator = validator
        self.retry_count = retry_count

    async def execute(self, input_data: dict) -> PipelineResult:
        """执行阶段处理，含验证和重试"""
        last_error = None
        for attempt in range(self.retry_count + 1):
            try:
                result = await self.agent.process(input_data)

                # 质量门控检查
                if self.validator and not self.validator(result):
                    raise QualityGateError(
                        f"Stage {self.name} validation failed"
                    )

                return PipelineResult(
                    stage=self.name,
                    status="success",
                    output=result,
                    metrics=self.collect_metrics(result)
                )
            except Exception as e:
                last_error = e
                logger.warning(
                    f"Stage {self.name} attempt {attempt+1} failed: {e}"
                )

        return PipelineResult(
            stage=self.name,
            status="failed",
            error=last_error
        )


class AgentPipeline:
    """流水线编排器"""

    def __init__(self, stages: list[PipelineStage]):
        self.stages = stages
        self.pipeline_state = PipelineState()

    async def run(self, initial_input: dict) -> PipelineReport:
        """从头到尾执行流水线"""
        current_data = initial_input

        for stage in self.stages:
            # 可选：条件跳过
            if self.should_skip(stage, current_data):
                self.pipeline_state.record_skip(stage.name)
                continue

            result = await stage.execute(current_data)
            self.pipeline_state.record(result)

            if result.status == "failed":
                # 快速失败 或 降级继续
                if self.fail_fast:
                    raise PipelineException(
                        f"Pipeline halted at {stage.name}"
                    )
                self.pipeline_state.record_fallback(stage.name)
                current_data = self.get_fallback_data(stage.name)
            else:
                current_data = result.output

        return self.pipeline_state.generate_report()

    def should_skip(self, stage: PipelineStage, data: dict) -> bool:
        """动态跳过逻辑 - 基于条件判断"""
        if stage.name == "test" and data.get("skip_tests"):
            return True
        if stage.name == "deploy" and not data.get("tests_passed"):
            return True
        return False

# 定义一个完整的代码修复流水线
fix_pipeline = AgentPipeline(stages=[
    PipelineStage(
        name="analyze",
        agent=CodexAgent(),
        validator=lambda r: r.confidence > 0.7,
        retry_count=1,
    ),
    PipelineStage(
        name="generate_fix",
        agent=ClaudeCodeAgent(),
        validator=lambda r: r.patch.is_valid(),
        retry_count=2,
    ),
    PipelineStage(
        name="test_fix",
        agent=TestRunnerAgent(),
        validator=lambda r: r.pass_rate > 0.95,
        retry_count=3,
    ),
    PipelineStage(
        name="commit",
        agent=GitAgent(),
        validator=lambda r: r.push_successful,
        retry_count=1,
    ),
])

# 执行
report = await fix_pipeline.run({
    "bug_report": "Memory leak in connection pool",
    "repo_path": "/opt/data/my-project",
    "skip_tests": False,
})

    ┌──────┐         ┌──────┐
    │Agent A├────────▶│Agent B│
    └───┬──┘         └───┬──┘
        │                │
        ▼                ▼
    ┌──────┐         ┌──────┐
    │Agent D├◀────────│Agent C│
    └──────┘         └──────┘
    (网状互连，自由协商)

class P2PAgent:
    """对等网络中的 Agent 节点"""

    def __init__(self, name: str, capabilities: list[str]):
        self.name = name
        self.capabilities = set(capabilities)
        self.peers: dict[str, "P2PAgent"] = {}
        self.message_bus = MessageBus()
        self.state = AgentState.IDLE

    def register_peer(self, peer: "P2PAgent"):
        """注册对等节点"""
        self.peers[peer.name] = peer
        peer.peers[self.name] = self  # 双向注册

    async def handle_task_proposal(self, proposal: TaskProposal):
        """收到任务提案后的响应策略"""
        # 评估自身能力匹配度
        match_score = self.evaluate_match(proposal)

        if match_score > self.ACCEPT_THRESHOLD:
            # 接受任务
            self.state = AgentState.WORKING
            await self.message_bus.broadcast(
                TaskAcceptance(
                    proposer=proposal.sender,
                    accepter=self.name,
                    task=proposal.task,
                )
            )
            result = await self.execute(proposal.task)
            await self.message_bus.broadcast(TaskResult(
                agent=self.name, result=result
            ))
        else:
            # 拒绝或转发给更合适的节点
            best_peer = self.find_best_peer(proposal)
            if best_peer:
                await self.message_bus.send(
                    best_peer.name,
                    TaskForward(original=proposal, suggested_by=self.name)
                )
            else:
                await self.message_bus.broadcast(
                    TaskRejection(task=proposal.task, reason="no_match")
                )

    def find_best_peer(self, proposal: TaskProposal) -> str:
        """寻找最合适的对等节点"""
        candidates = []
        for peer_name, peer in self.peers.items():
            if peer.state != AgentState.WORKING:
                score = peer.evaluate_match(proposal)
                candidates.append((peer_name, score))

        if candidates:
            return max(candidates, key=lambda x: x[1])[0]
        return None


class P2PNetwork:
    """P2P 网络协调层"""

    def __init__(self):
        self.agents: dict[str, P2PAgent] = {}
        self.message_bus = MessageBus()

    def add_agent(self, agent: P2PAgent):
        """添加 Agent 并自动建立对等连接"""
        self.agents[agent.name] = agent
        # 与所有已有 Agent 建立连接
        for existing in self.agents.values():
            if existing.name != agent.name:
                agent.register_peer(existing)

    async def submit_task(self, task: Task) -> TaskReport:
        """提交任务到网络，由网络自主协商处理"""
        proposal = TaskProposal(
            sender="orchestrator",
            task=task,
            deadline=task.deadline,
            priority=task.priority,
        )

        # 广播任务提案
        await self.message_bus.broadcast(proposal)

        # 等待网络自发形成协作
        return await self.collect_results(task)

# 场景：5 个审查 Agent 对同一段代码进行多维度审查
network = P2PNetwork()

# 添加各具专长的 Agent
agents = [
    P2PAgent("security", ["security_audit", "vuln_scan"]),
    P2PAgent("performance", ["perf_analysis", "bottleneck_detect"]),
    P2PAgent("style", ["lint_check", "format_check"]),
    P2PAgent("logic", ["logic_review", "edge_case_detect"]),
    P2PAgent("docs", ["doc_check", "api_compatibility"]),
]

for a in agents:
    network.add_agent(a)

# 提交审查任务
report = await network.submit_task(Task(
    type="code_review",
    input={
        "code_path": "src/auth/module.py",
        "diff": "...",
        "context": "OAuth2 认证模块重构",
    },
))
# 网络自动协商：security 处理安全、performance 处理性能...
# 最终聚合为一份综合审查报告

需要多工具集成？
│
├─ 任务能否清晰分解为独立子任务？
│  ├─ 是 → 子任务间有依赖吗？
│  │       ├─ 有 → Pipeline（有固定顺序的流水线）
│  │       └─ 无 → Hub-and-Spoke（并行分发）
│  │
│  └─ 否 → 任务边界是否模糊/需要动态协商？
│           ├─ 是 → P2P（自主协作网络）
│           └─ 否 → Hub-and-Spoke（集中管控）
│
└─ 容错性是否是第一优先级？
    └─ 是 → P2P（无单点故障）