第14章：高级Agent模式实践¶

深入探索Plan-and-Execute、多Agent协作和自主研究Agent

📚 学习目标¶

本Notebook将带你：

• ✅ 理解Plan-and-Execute模式
• ✅ 实现多Agent协作系统
• ✅ 构建自主研究Agent
• ✅ 掌握Agent评估和优化

预计时间¶

• Plan-and-Execute实现：30分钟
• 多Agent协作：40分钟
• 自主研究Agent：30分钟
• 评估优化：20分钟

---

In [ ]:

# 导入必要的库
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import json

# 设置（使用模拟的LLM客户端）
class MockLLMClient:
    """模拟LLM客户端"""
    async def generate(self, prompt: str, **kwargs) -> str:
        if "plan" in prompt.lower():
            return json.dumps({
                "steps": [
                    "Step 1: Search for information",
                    "Step 2: Analyze results",
                    "Step 3: Generate final answer"
                ]
            })
        return "Mock response"

llm_client = MockLLMClient()
print("Mock LLM client initialized")

# 导入必要的库 import asyncio from typing import List, Dict, Any, Optional from dataclasses import dataclass import json # 设置（使用模拟的LLM客户端） class MockLLMClient: """模拟LLM客户端""" async def generate(self, prompt: str, **kwargs) -> str: if "plan" in prompt.lower(): return json.dumps({ "steps": [ "Step 1: Search for information", "Step 2: Analyze results", "Step 3: Generate final answer" ] }) return "Mock response" llm_client = MockLLMClient() print("Mock LLM client initialized")

1. Plan-and-Execute Agent¶

1.1 核心概念¶

Plan-and-Execute = 规划阶段 + 执行阶段

In [ ]:

@dataclass
class Plan:
    steps: List[str]
    reasoning: str

class PlannerAgent:
    """规划Agent"""
    async def plan(self, task: str) -> Plan:
        return Plan(
            steps=["Gather info", "Process", "Answer"],
            reasoning="Default plan"
        )

class ExecutorAgent:
    """执行Agent"""
    async def execute(self, plan: Plan) -> Dict:
        results = []
        for step in plan.steps:
            results.append(f"Executed: {step}")
        return {"final_answer": "All steps completed"}

# 测试
planner = PlannerAgent()
plan = await planner.plan("Test task")
print(f"Plan: {plan.steps}")

executor = ExecutorAgent()
result = await executor.execute(plan)
print(f"Result: {result}")

@dataclass class Plan: steps: List[str] reasoning: str class PlannerAgent: """规划Agent""" async def plan(self, task: str) -> Plan: return Plan( steps=["Gather info", "Process", "Answer"], reasoning="Default plan" ) class ExecutorAgent: """执行Agent""" async def execute(self, plan: Plan) -> Dict: results = [] for step in plan.steps: results.append(f"Executed: {step}") return {"final_answer": "All steps completed"} # 测试 planner = PlannerAgent() plan = await planner.plan("Test task") print(f"Plan: {plan.steps}") executor = ExecutorAgent() result = await executor.execute(plan) print(f"Result: {result}")

---

2. 多Agent协作¶

2.1 Manager-Worker模式¶

In [ ]:

class Agent:
    """基础Agent类"""
    def __init__(self, name: str, role: str):
        self.name = name
        self.role = role
    
    async def process(self, task: str) -> Dict:
        await asyncio.sleep(0.1)
        return {"from": self.name, "result": f"Processed: {task}"}

class MultiAgentSystem:
    """多Agent系统"""
    def __init__(self):
        self.agents = []
    
    def add_agent(self, agent: Agent):
        self.agents.append(agent)
    
    async def collaborate(self, task: str) -> Dict:
        print(f"Task: {task}")
        print(f"Team size: {len(self.agents)}")
        
        # 并行执行
        results = await asyncio.gather(*[
            agent.process(task) for agent in self.agents
        ])
        
        return {"results": results}

# 测试
system = MultiAgentSystem()
system.add_agent(Agent("Researcher", "research"))
system.add_agent(Agent("Analyst", "analyze"))

result = await system.collaborate("Research AI trends")
print(f"Completed with {len(result['results'])} agents")

class Agent: """基础Agent类""" def __init__(self, name: str, role: str): self.name = name self.role = role async def process(self, task: str) -> Dict: await asyncio.sleep(0.1) return {"from": self.name, "result": f"Processed: {task}"} class MultiAgentSystem: """多Agent系统""" def __init__(self): self.agents = [] def add_agent(self, agent: Agent): self.agents.append(agent) async def collaborate(self, task: str) -> Dict: print(f"Task: {task}") print(f"Team size: {len(self.agents)}") # 并行执行 results = await asyncio.gather(*[ agent.process(task) for agent in self.agents ]) return {"results": results} # 测试 system = MultiAgentSystem() system.add_agent(Agent("Researcher", "research")) system.add_agent(Agent("Analyst", "analyze")) result = await system.collaborate("Research AI trends") print(f"Completed with {len(result['results'])} agents")

---

3. 练习¶

练习1：改进规划器¶

让PlannerAgent根据任务复杂度生成不同数量的步骤。

In [ ]:

# TODO: 实现自适应规划
class AdaptivePlanner(PlannerAgent):
    async def plan(self, task: str) -> Plan:
        # 分析任务复杂度
        complexity = len(task.split())  # 简化版
        
        if complexity < 5:
            steps = ["Quick research", "Answer"]
        else:
            steps = ["Deep research", "Analyze", "Synthesize", "Answer"]
        
        return Plan(steps=steps, reasoning=f"Adaptive plan (complexity: {complexity})")

# 测试
adaptive_planner = AdaptivePlanner()
plan1 = await adaptive_planner.plan("Simple task")
plan2 = await adaptive_planner.plan("This is a much more complex task that needs detailed analysis")

print(f"Simple task plan: {plan1.steps}")
print(f"Complex task plan: {plan2.steps}")

# TODO: 实现自适应规划 class AdaptivePlanner(PlannerAgent): async def plan(self, task: str) -> Plan: # 分析任务复杂度 complexity = len(task.split()) # 简化版 if complexity < 5: steps = ["Quick research", "Answer"] else: steps = ["Deep research", "Analyze", "Synthesize", "Answer"] return Plan(steps=steps, reasoning=f"Adaptive plan (complexity: {complexity})") # 测试 adaptive_planner = AdaptivePlanner() plan1 = await adaptive_planner.plan("Simple task") plan2 = await adaptive_planner.plan("This is a much more complex task that needs detailed analysis") print(f"Simple task plan: {plan1.steps}") print(f"Complex task plan: {plan2.steps}")

---

恭喜完成第14章的学习！ 🎉