如果你正在使用 Claude 系列模型进行开发，可能会面临一个现实问题：如何在高性能的 Fable 5 和更具性价比的 Sonnet 5 之间做出选择？直接使用 Fable 5 虽然性能最强，但成本高昂；只用 Sonnet 5 虽然便宜，但在复杂任务上表现有限。Claude 开发团队最近公开的两种多智能体协作模式，正好解决了这个痛点。

根据官方分享，团队内部高频使用 Advisor（顾问）和 Orchestrator（编排者）两种模式，能够在保持高性能的同时显著降低成本。实证数据显示，Advisor 模式可以用 63% 的成本获得 Fable 5 单独使用约 92% 的性能，而 Orchestrator 模式更是能用 46% 的成本获得约 96% 的性能。这种"鱼与熊掌兼得"的方案，对于需要平衡效果与成本的开发团队来说，无疑具有重要参考价值。

本文将深入解析这两种模式的工作原理、适用场景和实现方法，帮助你在实际项目中合理运用多智能体协作策略。无论你是正在构建复杂的 AI 应用，还是希望优化现有智能体系统的成本结构，都能从中获得实用的技术洞察。

1. 多智能体协作的核心价值与问题意识

在单智能体系统中，我们往往面临"性能与成本不可兼得"的困境。高性能模型如 Fable 5 在处理复杂任务时表现出色，但每个 token 的成本较高；而性价比模型如 Sonnet 5 虽然经济实惠，但在需要深度推理或复杂规划的任务上可能力不从心。

多智能体协作的核心思想是"让合适的模型做合适的事"。这类似于一个高效团队的分工协作：经验丰富的专家负责战略规划和关键决策，而执行能力强的成员负责具体实施。这种分工不仅提升了整体效率，还优化了资源分配。

从技术架构角度看，多智能体系统需要解决几个关键问题：

• 智能体间的通信机制如何设计？
• 任务分配和结果聚合的逻辑如何处理？
• 错误处理和回退策略如何实现？
• 成本控制和性能监控如何平衡？

Claude 团队提出的两种模式给出了具体答案。Advisor 模式适用于需要"关键时刻专家指导"的场景，而 Orchestrator 模式更适合"顶层设计+分层执行"的复杂任务。理解这两种模式的差异和适用边界，是构建高效多智能体系统的第一步。

2. Advisor 模式：自下而上的智能求助机制

2.1 核心架构与工作流程

Advisor 模式的基本思想是"执行者主导，专家辅助"。在这种架构中，Sonnet 5 作为主要执行者运行主循环，当遇到需要高级判断或复杂决策的情况时，通过 tool call 机制调用 Fable 5 获取指导建议。

具体的工作流程如下：

1. Sonnet 5 接收任务并开始执行
2. 在执行过程中，当遇到超出自身能力范围的关键决策点时，主动暂停
3. 通过预定义的接口调用 Fable 5，传递当前上下文和待决策问题
4. Fable 5 分析情况后返回具体的指导建议
5. Sonnet 5 根据建议继续执行，直到任务完成或遇到下一个决策点

这种模式的巧妙之处在于，Fable 5 只在自己最擅长的领域（复杂推理、战略规划）被调用，而大量的常规执行工作仍由成本更低的 Sonnet 5 完成。

2.2 关键技术特征与成本优势

Advisor 模式有几个显著的技术特征：

• 调用方向 ：Executor → Advisor（执行者主动求助）
• 调用频率 ：每个任务通常只调用 1-2 次，主要用于关键节点的"纠偏"或"定方向"
• 计费优势 ：绝大多数 token 按 Sonnet 5 的低价计费，只有少量咨询调用按 Fable 5 计价

从官方提供的 SWE-bench Pro 测试数据（482 个题目）可以看出：

• 纯 Sonnet 5：准确率约 75.5%，成本约 0.75 美元
• Sonnet 5 + Fable advisor：准确率约 84%，成本约 1.40 美元
• 纯 Fable 5：准确率约 91.5%，成本约 2.25 美元

这意味着组合方案用 63% 的成本获得了 Fable 5 单独使用约 92% 的性能，性价比提升非常明显。

2.3 适用场景与实现要点

Advisor 模式特别适合以下场景：

• 代码开发任务 ：Sonnet 5 负责具体编码，在架构设计或复杂算法实现时求助 Fable 5
• 文档撰写 ：常规内容由 Sonnet 5 完成，在需要深度分析或战略规划部分调用 Fable 5
• 数据分析 ：基础处理用 Sonnet 5，复杂统计推断或趋势分析求助 Fable 5

实现时的关键要点：

• 需要明确定义"求助触发条件"，避免过度调用或调用不足
• 设计清晰的信息传递格式，确保上下文传递的完整性
• 建立错误处理机制，当 Fable 5 不可用时要有回退方案

3. Orchestrator 模式：自上而下的任务编排体系

3.1 架构设计与执行逻辑

Orchestrator 模式采用"专家规划，团队执行"的思路。在这种模式下，Fable 5 作为编排者负责整体任务规划和分解，然后将子任务分配给多个 Sonnet 5 worker 并行执行。

具体执行流程：

1. Fable 5 接收原始任务，进行任务分析和规划
2. 将复杂任务分解为多个可并行执行的子任务
3. 为每个子任务创建 Sonnet 5 worker 实例
4. 各 worker 独立执行分配的子任务
5. Fable 5 收集各 worker 的执行结果并进行整合
6. 输出最终的任务结果

这种模式的核心优势在于充分发挥了 Fable 5 的规划能力和 Sonnet 5 的并行执行效率，特别适合处理可分解的复杂任务。

3.2 性能表现与成本分析

根据 BrowseComp 完整集的测试数据：

• 全 Sonnet 5：准确率 77.8%，成本 16.01 美元
• Fable 5 lead + Sonnet 5 workers：准确率 86.8%，成本 18.53 美元
• 全 Fable 5：准确率 90.8%，成本 40.56 美元

Orchestrator 模式用 46% 的成本获得了 Fable 5 单独使用约 96% 的性能。相比纯 Sonnet 5 方案，准确率提升约 9 个百分点，成本仅增加约 2.5 美元，边际效益十分显著。

3.3 典型应用场景与实施考虑

Orchestrator 模式在以下场景中表现优异：

• 研究类任务 ：需要大量资料搜集和综合分析的项目
• 批量处理 ：对大量相似项目进行并行处理和分析
• 复杂问题求解 ：需要多角度分析并合成最终结论的任务

实施时需要重点考虑：

• 任务分解的粒度要合理，避免过度细分导致通信开销过大
• 需要设计有效的结果聚合算法，确保最终输出的质量
• 要考虑 worker 间的依赖关系和执行顺序问题
• 需要建立监控机制，及时发现和处理失败的子任务

4. 两种模式的深度对比与选择指南

4.1 控制流与职责划分对比

对比维度	Advisor 模式	Orchestrator 模式
控制流方向	自下而上（执行者主动求助）	自上而下（编排者主动委派）
Fable 5 角色	被动顾问，提供关键时刻指导	主动规划者，负责任务分解和分配
Sonnet 5 角色	主要执行者，承担大部分工作	工作节点，执行具体子任务
通信模式	间歇性咨询，调用频率低	集中规划+结果收集，调用有阶段性

4.2 成本结构与适用场景对比

Advisor 模式更适合 ：

• 单任务流中需要关键节点指导的场景
• 任务整体不难，但包含少数复杂决策点
• 成本敏感且对性能要求不是极致的项目
• 开发者的主要目标是"用得起的好性能"

Orchestrator 模式更适合 ：

• 可并行处理的批量任务或研究项目
• 需要顶层设计和系统规划的场景
• 对最终质量要求较高且任务可分解的项目
• 追求"接近顶级性能的最佳性价比"

4.3 选择决策树

在实际项目中选择模式时，可以遵循以下决策流程：

1. 分析任务特性

   • 任务是否可分解为独立子任务？是 → 考虑 Orchestrator
   • 任务是否线性执行但有关键决策点？是 → 考虑 Advisor

2. 评估成本约束

   • 预算紧张但接受适度性能损失？ → 优先 Advisor
   • 追求最佳性价比且任务可并行？ → 优先 Orchestrator

3. 考虑技术复杂度

   • 团队经验有限，希望简单实现？ → 优先 Advisor
   • 有分布式系统经验，能处理复杂协调？ → 可考虑 Orchestrator

5. 实战示例：基于 Claude API 的多智能体实现

5.1 环境准备与基础配置

在开始编码前，需要确保环境准备就绪：

# 安装必要的 Python 包
pip install anthropic python-dotenv

创建配置文件 .env ：

ANTHROPIC_API_KEY=your_api_key_here

基础工具类设置：

# utils.py
import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

class ClaudeClient:
    def __init__(self):
        self.client = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
    
    def call_model(self, model: str, prompt: str, max_tokens: int = 1000):
        response = self.client.messages.create(
            model=model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

5.2 Advisor 模式完整实现示例

下面是一个完整的 Advisor 模式实现，模拟代码审查场景：

# advisor_pattern.py
import re
from utils import ClaudeClient

class CodeAdvisorSystem:
    def __init__(self):
        self.client = ClaudeClient()
        self.complexity_threshold = 0.7  # 复杂度阈值，超过则求助专家
        
    def assess_complexity(self, code_snippet: str) -> float:
        """评估代码复杂度，决定是否需要专家指导"""
        complexity_indicators = [
            len(re.findall(r'def ', code_snippet)),  # 函数数量
            len(re.findall(r'class ', code_snippet)),  # 类数量
            len(re.findall(r'if |for |while ', code_snippet)),  # 控制结构
        ]
        return sum(complexity_indicators) / 10  # 简化复杂度计算
    
    def sonnet_review(self, code: str) -> str:
        """Sonnet 5 执行基础代码审查"""
        prompt = f"""
请对以下代码进行基础审查，重点关注语法错误、基础逻辑问题：
{code}

请提供简洁的审查意见：
"""
        return self.client.call_model("claude-3-5-sonnet-20241022", prompt)
    
    def fable_advisor(self, code: str, context: str) -> str:
        """Fable 5 提供高级架构建议"""
        prompt = f"""
作为高级架构师，请对以下代码进行深度审查：
代码：{code}
审查上下文：{context}

请从架构设计、性能优化、可维护性角度提供专业建议：
"""
        return self.client.call_model("claude-3-5-fable-20241022", prompt)
    
    def review_code(self, code: str, context: str = "") -> dict:
        """完整的代码审查流程"""
        complexity = self.assess_complexity(code)
        
        # Sonnet 5 执行基础审查
        base_review = self.sonnet_review(code)
        
        result = {
            'base_review': base_review,
            'expert_advice': None,
            'complexity_score': complexity,
            'advisor_called': False
        }
        
        # 复杂度超过阈值时求助 Fable 5
        if complexity > self.complexity_threshold:
            expert_advice = self.fable_advisor(code, context)
            result['expert_advice'] = expert_advice
            result['advisor_called'] = True
            
        return result

# 使用示例
if __name__ == "__main__":
    system = CodeAdvisorSystem()
    
    sample_code = """
def process_data(data):
    results = []
    for item in data:
        if item['type'] == 'A':
            # 复杂处理逻辑
            transformed = transform_complex_a(item)
            results.append(transformed)
        elif item['type'] == 'B':
            # 另一个复杂分支
            transformed = transform_complex_b(item)
            results.append(transformed)
    return results
"""
    
    review_result = system.review_code(sample_code, "数据处理函数审查")
    print(f"复杂度评分: {review_result['complexity_score']}")
    print(f"是否求助专家: {review_result['advisor_called']}")
    print(f"基础审查: {review_result['base_review']}")
    if review_result['expert_advice']:
        print(f"专家建议: {review_result['expert_advice']}")

5.3 Orchestrator 模式完整实现示例

以下是一个研究任务编排系统的实现：

# orchestrator_pattern.py
import asyncio
from typing import List, Dict
from utils import ClaudeClient

class ResearchOrchestrator:
    def __init__(self):
        self.client = ClaudeClient()
        
    async def fable_planning(self, research_topic: str) -> List[str]:
        """Fable 5 进行任务规划和分解"""
        prompt = f"""
请对以下研究主题进行任务分解，生成3-5个可并行研究的子任务：
研究主题：{research_topic}

请返回具体的子任务列表，每个任务应该独立可执行：
"""
        planning_result = self.client.call_model(
            "claude-3-5-fable-20241022", prompt
        )
        
        # 解析规划结果，提取子任务
        subtasks = [
            line.strip().replace('- ', '').replace('* ', '')
            for line in planning_result.split('\n')
            if line.strip() and line.strip()[0] in ['-', '*', '1', '2', '3', '4', '5']
        ]
        return subtasks[:4]  # 限制最多4个子任务
    
    async def sonnet_worker(self, subtask: str, worker_id: int) -> Dict:
        """Sonnet 5 工作节点执行子任务"""
        prompt = f"""
请深入研究以下子任务，提供详细的分析报告：
子任务：{subtask}

请从多个角度进行分析，提供有深度的见解：
"""
        try:
            result = self.client.call_model(
                "claude-3-5-sonnet-20241022", prompt
            )
            return {
                'worker_id': worker_id,
                'subtask': subtask,
                'result': result,
                'status': 'success'
            }
        except Exception as e:
            return {
                'worker_id': worker_id,
                'subtask': subtask,
                'result': str(e),
                'status': 'failed'
            }
    
    async def execute_research(self, research_topic: str) -> Dict:
        """执行完整的研究任务"""
        print("🎯 Fable 5 开始任务规划...")
        subtasks = await self.fable_planning(research_topic)
        print(f"📋 规划完成，生成 {len(subtasks)} 个子任务")
        
        print("🚀 启动 Sonnet 5 worker 并行执行...")
        tasks = [
            self.sonnet_worker(subtask, i) 
            for i, subtask in enumerate(subtasks)
        ]
        
        worker_results = await asyncio.gather(*tasks)
        
        print("📊 Fable 5 开始结果整合...")
        final_report = self.integrate_results(research_topic, worker_results)
        
        return {
            'research_topic': research_topic,
            'subtasks': subtasks,
            'worker_results': worker_results,
            'final_report': final_report
        }
    
    def integrate_results(self, topic: str, results: List[Dict]) -> str:
        """整合各worker的研究结果"""
        successful_results = [
            r for r in results if r['status'] == 'success'
        ]
        
        integration_prompt = f"""
请基于以下各子任务的研究结果，整合成关于'{topic}'的完整研究报告：

子任务研究成果：
{chr(10).join([f"子任务 {r['worker_id']+1}: {r['subtask']}{chr(10)}结果: {r['result']}" for r in successful_results])}

请生成结构完整、逻辑清晰的研究报告：
"""
        
        return self.client.call_model(
            "claude-3-5-fable-20241022", integration_prompt
        )

# 使用示例
async def main():
    orchestrator = ResearchOrchestrator()
    
    research_result = await orchestrator.execute_research(
        "多智能体系统在软件开发中的应用前景"
    )
    
    print(f"研究主题: {research_result['research_topic']}")
    print(f"生成子任务数量: {len(research_result['subtasks'])}")
    print(f"最终报告长度: {len(research_result['final_report'])} 字符")
    print("\n最终报告摘要:")
    print(research_result['final_report'][:500] + "...")

if __name__ == "__main__":
    asyncio.run(main())

6. 性能优化与成本控制策略

6.1 Token 使用优化技巧

在多智能体系统中，token 使用效率直接影响成本。以下是一些实用优化策略：

# token_optimizer.py
class TokenOptimizer:
    def __init__(self):
        self.compression_threshold = 500  # 超过500token考虑压缩
    
    def compress_context(self, context: str, target_ratio: float = 0.6) -> str:
        """压缩上下文信息，保留关键内容"""
        if len(context.split()) < self.compression_threshold:
            return context
            
        compression_prompt = f"""
请将以下文本压缩到原长度的{target_ratio*100}%，保留所有关键信息：
{context}

压缩后的文本：
"""
        # 这里可以使用更便宜的模型进行压缩
        compressed = self.client.call_model(
            "claude-3-haiku-20240307", compression_prompt
        )
        return compressed
    
    def optimize_prompt(self, original_prompt: str, model_type: str) -> str:
        """根据模型类型优化提示词"""
        if model_type == "sonnet":
            # Sonnet 5 需要更详细的指令
            return f"""
请逐步思考并详细回答以下问题：
{original_prompt}

请确保回答完整且逻辑清晰：
"""
        else:  # Fable 5
            # Fable 5 可以处理更简洁的指令
            return f"""
请提供专业、深入的解答：
{original_prompt}
"""

6.2 智能调用频率控制

避免过度调用高价模型是关键的成本控制手段：

# call_controller.py
import time
from collections import deque

class CallController:
    def __init__(self, max_calls_per_minute: int = 10):
        self.call_history = deque()
        self.max_calls = max_calls_per_minute
    
    def should_call_fable(self, complexity: float, recent_calls: int) -> bool:
        """基于复杂度和调用频率决定是否调用 Fable 5"""
        if complexity < 0.3:
            return False  # 简单任务不需要专家
        elif complexity > 0.8:
            return True   # 复杂任务必须求助
        
        # 中等复杂度任务基于频率控制
        current_time = time.time()
        # 清理1分钟前的记录
        while (self.call_history and 
               current_time - self.call_history[0] > 60):
            self.call_history.popleft()
        
        if len(self.call_history) < self.max_calls:
            return True
        else:
            return False  # 频率超限，本次不调用
    
    def record_call(self):
        """记录 Fable 5 调用时间"""
        self.call_history.append(time.time())

7. 常见问题与故障排查

7.1 模式选择与配置问题

问题现象	可能原因	解决方案
成本超出预期	Advisor 模式调用频率过高	调整复杂度阈值，减少不必要的专家调用
性能提升不明显	任务不适合多智能体协作	重新评估任务特性，选择单模型方案
响应时间过长	Orchestrator 模式任务分解过细	调整任务粒度，减少协调开销
结果质量不稳定	子任务间缺乏一致性约束	加强结果整合逻辑，添加一致性检查

7.2 API 使用与错误处理

# error_handler.py
class ErrorHandler:
    @staticmethod
    def handle_api_error(error, model_type: str, fallback_model: str = None):
        """处理 API 调用错误"""
        error_messages = {
            "rate_limit": "API调用频率超限，建议添加限流控制",
            "authentication": "API密钥错误，检查密钥配置",
            "overload": "服务暂时过载，建议重试机制",
            "content_filter": "内容触发安全过滤，调整提示词"
        }
        
        error_type = str(error).lower()
        for key, message in error_messages.items():
            if key in error_type:
                print(f"⚠️ {message}")
                
                if fallback_model and key in ["overload", "rate_limit"]:
                    print(f"🔄 切换到备用模型: {fallback_model}")
                    return fallback_model
                
        # 默认错误处理
        print(f"❌ 未知错误: {error}")
        return None
    
    @staticmethod
    def validate_response(response: str, min_length: int = 10) -> bool:
        """验证响应质量"""
        if not response or len(response.strip()) < min_length:
            return False
        
        # 检查是否包含常见错误模式
        error_patterns = [
            "I cannot", "I'm sorry", "无法", "错误"
        ]
        return not any(pattern in response for pattern in error_patterns)

7.3 性能监控与调试

建立有效的监控体系可以帮助及时发现和解决问题：

# performance_monitor.py
import time
import json
from datetime import datetime

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {
            'total_calls': 0,
            'fable_calls': 0,
            'sonnet_calls': 0,
            'total_tokens': 0,
            'total_cost': 0.0
        }
        self.call_logs = []
    
    def log_call(self, model: str, tokens_used: int, cost: float):
        """记录每次调用详情"""
        timestamp = datetime.now().isoformat()
        
        self.metrics['total_calls'] += 1
        self.metrics['total_tokens'] += tokens_used
        self.metrics['total_cost'] += cost
        
        if 'fable' in model.lower():
            self.metrics['fable_calls'] += 1
        else:
            self.metrics['sonnet_calls'] += 1
        
        self.call_logs.append({
            'timestamp': timestamp,
            'model': model,
            'tokens': tokens_used,
            'cost': cost
        })
    
    def get_cost_breakdown(self) -> dict:
        """获取成本分析报告"""
        total = self.metrics['total_cost']
        fable_ratio = (self.metrics['fable_calls'] / 
                      max(1, self.metrics['total_calls']))
        
        return {
            'total_cost': total,
            'fable_contribution': total * fable_ratio,
            'sonnet_contribution': total * (1 - fable_ratio),
            'cost_per_call': total / max(1, self.metrics['total_calls']),
            'tokens_per_dollar': self.metrics['total_tokens'] / max(0.01, total)
        }
    
    def generate_report(self) -> str:
        """生成性能报告"""
        breakdown = self.get_cost_breakdown()
        
        report = f"""
多智能体系统性能报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

📊 调用统计:
- 总调用次数: {self.metrics['total_calls']}
- Fable 5 调用: {self.metrics['fable_calls']}
- Sonnet 5 调用: {self.metrics['sonnet_calls']}

💰 成本分析:
- 总成本: ${breakdown['total_cost']:.4f}
- Fable 5 占比: ${breakdown['fable_contribution']:.4f}
- Sonnet 5 占比: ${breakdown['sonnet_contribution']:.4f}
- 每次调用平均成本: ${breakdown['cost_per_call']:.4f}
- 每美元获得token数: {breakdown['tokens_per_dollar']:.0f}

💡 优化建议:
"""
        if breakdown['fable_contribution'] > breakdown['sonnet_contribution']:
            report += "- 考虑降低 Fable 5 调用频率或优化调用时机"
        else:
            report += "- 成本结构合理，可继续保持当前策略"
            
        return report

8. 最佳实践与工程化建议

8.1 架构设计原则

在实际项目中应用多智能体模式时，建议遵循以下原则：

明确职责边界

• 每个智能体应该有清晰的职责范围
• 避免功能重叠导致的冗余调用
• 建立标准的通信协议和数据格式

渐进式复杂度

• 从简单的 Advisor 模式开始，逐步过渡到复杂的 Orchestrator 模式
• 每个阶段都要有明确的验收标准和回退方案
• 建立 A/B 测试机制，验证模式切换的效果

容错与降级

• 设计完善的错误处理机制
• 确保单个智能体故障不影响整体系统
• 建立降级方案，当高价模型不可用时能自动切换

8.2 配置管理与环境隔离

# config_manager.py
import yaml
from typing import Dict, Any

class ConfigManager:
    def __init__(self, config_path: str = "config.yaml"):
        self.config_path = config_path
        self.config = self.load_config()
    
    def load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        except FileNotFoundError:
            return self.get_default_config()
    
    def get_default_config(self) -> Dict[str, Any]:
        """获取默认配置"""
        return {
            'models': {
                'advisor': 'claude-3-5-fable-20241022',
                'executor': 'claude-3-5-sonnet-20241022',
                'fallback': 'claude-3-haiku-20240307'
            },
            'thresholds': {
                'complexity_threshold': 0.7,
                'cost_alert_threshold': 10.0,
                'retry_max_attempts': 3
            },
            'optimization': {
                'context_compression': True,
                'prompt_optimization': True,
                'caching_enabled': False
            }
        }
    
    def get_model_config(self, pattern_type: str) -> Dict[str, Any]:
        """获取特定模式的模型配置"""
        pattern_configs = {
            'advisor': {
                'primary': self.config['models']['executor'],
                'advisor': self.config['models']['advisor'],
                'threshold': self.config['thresholds']['complexity_threshold']
            },
            'orchestrator': {
                'planner': self.config['models']['advisor'],
                'workers': self.config['models']['executor'],
                'fallback': self.config['models']['fallback']
            }
        }
        return pattern_configs.get(pattern_type, {})

8.3 安全与合规考虑

在多智能体系统中，需要特别注意以下安全事项：

数据隐私保护

• 敏感信息在智能体间传递时需要加密
• 建立数据脱敏机制，避免隐私泄露
• 定期清理临时数据和日志文件

使用合规性

• 遵守各模型供应商的使用条款
• 建立内容审核机制，避免生成不当内容
• 监控使用模式，确保符合公平使用原则

访问控制

• 严格管理 API 密钥和访问权限
• 建立使用配额和限流机制
• 记录完整的操作日志用于审计

9. 实际项目集成案例

9.1 代码审查系统集成

以下展示如何将 Advisor 模式集成到现有的代码审查流程中：

# code_review_integration.py
class CodeReviewPipeline:
    def __init__(self, advisor_system):
        self.advisor_system = advisor_system
        self.quality_gate = 0.8  # 质量门槛
    
    def process_pull_request(self, pr_data: dict) -> dict:
        """处理 Pull Request 审查"""
        results = []
        
        for file_change in pr_data['changes']:
            review_result = self.advisor_system.review_code(
                file_change['content'],
                f"PR#{pr_data['id']} - {file_change['filename']}"
            )
            
            results.append({
                'filename': file_change['filename'],
                'review_result': review_result,
                'passed': self.evaluate_review(review_result)
            })
        
        return {
            'pr_id': pr_data['id'],
            'overall_pass': all(r['passed'] for r in results),
            'file_reviews': results
        }
    
    def evaluate_review(self, review_result: dict) -> bool:
        """评估审查结果是否通过质量门槛"""
        if not review_result['advisor_called']:
            # 简单文件，基础审查通过即可
            return len(review_result['base_review']) > 50  # 有实质内容
        else:
            # 复杂文件，需要专家认可
            expert_advice = review_result['expert_advice'] or ""
            return "建议通过" in expert_advice or "架构合理" in expert_advice

9.2 研究分析平台集成

对于需要大量资料分析的研究平台，Orchestrator 模式可以大幅提升效率：

# research_platform.py
class ResearchPlatform:
    def __init__(self, orchestrator):
        self.orchestrator = orchestrator
        self.cache = {}  # 简单的结果缓存
    
    async def batch_research(self, topics: List[str]) -> Dict[str, Any]:
        """批量研究多个主题"""
        results = {}
        
        for topic in topics:
            if topic in self.cache:
                results[topic] = self.cache[topic]
                continue
                
            print(f"🔍 开始研究: {topic}")
            research_result = await self.orchestrator.execute_research(topic)
            results[topic] = research_result
            self.cache[topic] = research_result
            
            # 简单的限流控制
            await asyncio.sleep(1)
        
        return {
            'total_topics': len(topics),
            'cached_results': len([t for t in topics if t in self.cache]),
            'new_research': len(topics) - len([t for t in topics if t in self.cache]),
            'results': results
        }

通过本文的详细解析和实战示例，你应该对 Claude 多智能体协作的两种核心模式有了深入理解。在实际项目中，关键是找到适合自己业务场景的模式组合，并建立完善的监控优化机制。这种智能体协作思路不仅适用于 Claude 系列模型，其设计理念也可以迁移到其他多模型协作场景中。