第14课：Agent编排

🎼 Agent编排：协调多个Agent步骤

复杂任务往往需要多个步骤、多个Agent协作完成。Agent编排（Orchestration）是设计和管理这些步骤之间关系的技术，确保任务高效、可靠地执行。

📖 编排模式

Agent编排模式
├── 顺序编排 (Sequential)
│   └── A → B → C → D
├── 并行编排 (Parallel)
│   └── A ┬→ B
│         └→ C → 合并
├── 条件编排 (Conditional)
│   └── A → [条件?] → B / C
├── 循环编排 (Loop)
│   └── A → B → [满足条件?] → C / 回到B
└── DAG编排 (有向无环图)
    └── 复杂依赖关系

📊 编排框架对比

💻 代码实现：Agent编排引擎

# Agent编排引擎 - 支持多种编排模式
import json, time
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, field
from enum import Enum

class StepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class Step:
    # 编排步骤
    name: str
    agent: Callable
    depends_on: List[str] = field(default_factory=list)
    condition: Optional[Callable] = None
    retry: int = 0
    status: StepStatus = StepStatus.PENDING
    result: Any = None
    error: Optional[str] = None

class OrchestrationEngine:
    # Agent编排引擎
    def __init__(self):
        self.steps: Dict[str, Step] = {}
        self.context: Dict[str, Any] = {}
        self.execution_log: List[Dict] = []
    
    def add_step(self, name, agent, depends_on=None, condition=None, retry=0):
        self.steps[name] = Step(
            name=name, agent=agent,
            depends_on=depends_on or [],
            condition=condition, retry=retry
        )
    
    def run(self, initial_input=None) -> Dict:
        # 执行编排
        self.context["input"] = initial_input
        completed = set()
        max_iter = len(self.steps) * 3
        
        for _ in range(max_iter):
            # 找到可执行的步骤
            ready = self._get_ready_steps(completed)
            if not ready:
                if len(completed) == len(self.steps):
                    break  # 全部完成
                # 检查是否死锁
                pending = {n for n, s in self.steps.items() if s.status == StepStatus.PENDING}
                if not pending:
                    break
                print(f"⚠️ 可能存在循环依赖，待执行: {pending}")
                break
            
            for step in ready:
                # 检查条件
                if step.condition and not step.condition(self.context):
                    step.status = StepStatus.SKIPPED
                    completed.add(step.name)
                    continue
                
                # 执行步骤
                step.status = StepStatus.RUNNING
                for attempt in range(step.retry + 1):
                    try:
                        result = step.agent(self.context)
                        step.result = result
                        step.status = StepStatus.SUCCESS
                        self.context[step.name] = result
                        print(f"  ✅ {step.name}: {str(result)[:60]}")
                        break
                    except Exception as e:
                        step.error = str(e)
                        if attempt < step.retry:
                            print(f"  🔄 {step.name} 重试({attempt+1})")
                        else:
                            step.status = StepStatus.FAILED
                            print(f"  ❌ {step.name}: {e}")
                
                completed.add(step.name)
                self.execution_log.append({
                    "step": step.name, "status": step.status.value,
                    "result": str(step.result)[:100] if step.result else None
                })
        
        return self._build_result()
    
    def _get_ready_steps(self, completed):
        ready = []
        for name, step in self.steps.items():
            if step.status != StepStatus.PENDING:
                continue
            if all(dep in completed for dep in step.depends_on):
                ready.append(step)
        return ready
    
    def _build_result(self):
        success = sum(1 for s in self.steps.values() if s.status == StepStatus.SUCCESS)
        total = len(self.steps)
        return {
            "success": success == total,
            "completed_steps": success,
            "total_steps": total,
            "context": {k: v for k, v in self.context.items() if not k.startswith("_")},
            "log": self.execution_log,
        }

# ===== 定义Agent步骤 =====
def research_agent(context):
    # 研究Agent
    topic = context.get("input", "未知主题")
    return {"research": f"关于'{topic}'的研究报告：发现3个关键要点", "facts": ["要点1", "要点2", "要点3"]}

def outline_agent(context):
    # 大纲Agent
    research = context.get("research_agent", {}).get("research", "")
    return {"outline": f"基于研究的文章大纲：1.引言 2.分析 3.结论"}

def draft_agent(context):
    # 草稿Agent
    outline = context.get("outline_agent", {}).get("outline", "")
    return {"draft": f"根据大纲撰写的文章草稿（约500字）"}

def review_agent(context):
    # 审核Agent
    draft = context.get("draft_agent", {}).get("draft", "")
    return {"review": "审核意见：结构清晰，建议增加实例", "approved": True}

def publish_agent(context):
    # 发布Agent
    approved = context.get("review_agent", {}).get("approved", False)
    if approved:
        return {"published": True, "url": "https://example.com/article/1"}
    return {"published": False, "reason": "未通过审核"}

# ===== 测试 =====
engine = OrchestrationEngine()
engine.add_step("research_agent", research_agent)
engine.add_step("outline_agent", outline_agent, depends_on=["research_agent"])
engine.add_step("draft_agent", draft_agent, depends_on=["outline_agent"])
engine.add_step("review_agent", review_agent, depends_on=["draft_agent"])
engine.add_step("publish_agent", publish_agent, depends_on=["review_agent"])

print("🎯 执行编排：写一篇文章")
result = engine.run(initial_input="AI Agent的未来发展")
print(f"\n🏁 结果: {result['completed_steps']}/{result['total_steps']}步完成")
print(f"📊 成功: {result['success']}")

🏋️ 实战练习

深入理解：Agent编排核心原理

Agent编排的五种模式：顺序(A-B-C简单可控)、并行(A|B|C高效快速)、路由(A-B或C按需分派)、层级(管理者-工作者分工明确)、协作(A-B-C灵活创新)。LangGraph状态图编排是最新的生产级方案，支持循环、条件分支、状态持久化。关键设计原则：先Chain后Agent、能用固定流程不要引入动态决策。

进阶实现：Agent编排引擎

以下是针对Agent编排主题的进阶实现，包含顺序/并行/条件路由三种编排模式等核心功能。代码经过实机运行验证。

# AgentOrchestrator - Agent编排进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class AgentOrchestrator:
    # Agent编排进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = AgentOrchestrator({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

Agent编排最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：Agent编排的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。