多Agent编排实战

章节：专家篇第13章第3节
难度：⭐⭐⭐⭐ 专家级
阅读时间：30-35 分钟
核心收获：掌握多Agent编排的实战技巧和调试方法

16.10 理论要点：编排的核心挑战

16.10.1 三大挑战

16.10.2 编排模式

16.11 实战案例：并行编排

16.11.1 场景：多数据源并行获取

import asyncio

class ParallelOrchestrator:
    def __init__(self):
        self.executors = {
            "wechat": WechatExecutor(),
            "github": GithubExecutor(),
            "search": SearchExecutor(),
            "aihot": AIHOTExecutor()
        }

    async def execute_parallel(self, tasks):
        """并行执行多个任务"""
        coroutines = []
        for task in tasks:
            executor = self.executors[task["source"]]
            coroutines.append(executor.execute(task))

        results = await asyncio.gather(
            *coroutines,
            return_exceptions=True
        )

        return self.handle_results(results)

    def handle_results(self, results):
        """处理结果"""
        successful = []
        failed = []

        for i, result in enumerate(results):
            if isinstance(result, Exception):
                failed.append({"index": i, "error": str(result)})
            else:
                successful.append(result)

        return {
            "successful": successful,
            "failed": failed,
            "success_rate": len(successful) / len(results)
        }

16.11.2 错误处理

class ErrorHandler:
    def __init__(self, max_retries=2):
        self.max_retries = max_retries
        self.circuit_breakers = {}

    async def execute_with_fallback(self, executor, task):
        """执行任务，失败时降级"""
        source = task["source"]

        # 检查熔断器
        if self.is_circuit_open(source):
            return {"source": source, "status": "skipped", "reason": "circuit_open"}

        try:
            result = await executor.execute(task)
            self.record_success(source)
            return result
        except Exception as e:
            retries = self.get_retries(source)
            if retries < self.max_retries:
                self.increment_retries(source)
                return await self.execute_with_fallback(executor, task)
            else:
                self.record_failure(source)
                return {"source": source, "status": "failed", "error": str(e)}

16.12 调试技巧

16.12.1 链路追踪

class Trace:
    def __init__(self, trace_id):
        self.trace_id = trace_id
        self.spans = []

    def start_span(self, name, agent):
        span = {
            "trace_id": self.trace_id,
            "span_id": generate_uuid(),
            "name": name,
            "agent": agent,
            "start_time": time.time()
        }
        self.spans.append(span)
        return span

    def end_span(self, span, result):
        span["end_time"] = time.time()
        span["duration"] = span["end_time"] - span["start_time"]
        span["result"] = result

16.12.2 日志规范

class AgentLogger:
    def log(self, level, agent, message, data=None):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "level": level,
            "agent": agent,
            "message": message,
            "data": data
        }
        # 输出到日志系统
        output_log(log_entry)

16.13 图文说明

📌 待补充：建议绘制以下示意图 1. 并行编排流程图 2. 错误处理流程图 3. 链路追踪图 4. 调试面板原型

16.14 常见错误

📝 版本迭代记录