Agent 核心组件详解

┌─────────────────────────────────────────────────────────────────┐
│                    Agent 四大核心组件架构                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│                      ┌─────────────┐                            │
│                      │   用户目标   │                            │
│                      └──────┬──────┘                            │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    1. 规划器 (Planning)                    │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  职责: 理解目标 → 分解任务 → 制定计划 → 选择策略    │  │  │
│  │  │  技术: CoT思维链 | ToT思维树 | ReAct推理行动       │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    2. 记忆库 (Memory)                      │  │
│  │  ┌────────────┐  ┌────────────┐  ┌────────────┐          │  │
│  │  │  短期记忆   │  │  长期记忆   │  │  工作记忆   │          │  │
│  │  │ (会话上下文) │  │ (向量数据库) │  │ (任务状态)  │          │  │
│  │  └────────────┘  └────────────┘  └────────────┘          │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    3. 工具集 (Tools)                       │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐  │  │
│  │  │搜索工具│ │代码执行│ │数据库  │ │文件操作│ │API调用 │  │  │
│  │  └────────┘ └────────┘ └────────┘ └────────┘ └────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    4. 执行器 (Action)                      │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  职责: 选择工具 → 构造参数 → 安全执行 → 获取反馈    │  │  │
│  │  │  技术: 沙箱隔离 | 权限控制 | 错误处理 | 结果验证    │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│                      ┌─────────────┐                            │
│                      │   执行结果   │                            │
│                      └──────┬──────┘                            │
│                             ↓                                   │
│                      ┌─────────────┐                            │
│                      │  反馈给规划器 │ ←── 迭代优化循环          │
│                      └─────────────┘                            │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                      规划器内部架构                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   用户目标: "分析销售数据并生成报告"                             │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    意图理解层                             │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  输入: "分析销售数据并生成报告"                      │  │  │
│  │  │  输出: 意图 = {分析数据, 生成报告, 可视化}           │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    任务分解层                             │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Task 1: 连接数据库获取销售数据                      │  │  │
│  │  │  Task 2: 数据清洗和预处理                            │  │  │
│  │  │  Task 3: 统计分析与可视化                            │  │  │
│  │  │  Task 4: 生成报告文档                                │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    依赖分析层                             │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Task1 → Task2 → Task3 → Task4 (线性依赖)           │  │  │
│  │  │  或: Task1 → [Task2, Task3] → Task4 (并行可能)      │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    执行计划生成                           │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Plan = [                                           │  │  │
│  │  │    {task: Task1, tool: database, priority: 1},      │  │  │
│  │  │    {task: Task2, tool: python, priority: 2},        │  │  │
│  │  │    {task: Task3, tool: python, priority: 3},        │  │  │
│  │  │    {task: Task4, tool: document, priority: 4}       │  │  │
│  │  │  ]                                                  │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    CoT 思维链示例                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  问题: 一个商店有 23 个苹果，卖出了 17 个，又进货了 8 个，       │
│        现在有多少个苹果？                                        │
│                                                                 │
│  CoT 推理过程:                                                  │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │ Step 1: 初始有 23 个苹果                                  │  │
│  │ Step 2: 卖出 17 个，剩余 23 - 17 = 6 个                   │  │
│  │ Step 3: 进货 8 个，现有 6 + 8 = 14 个                     │  │
│  │ Answer: 现在有 14 个苹果                                  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

# CoT 规划器实现
class CoTPlanner:
    def __init__(self, llm):
        self.llm = llm
    
    def plan(self, goal: str) -> list[str]:
        """
        使用思维链进行任务规划
        """
        prompt = f"""
        目标: {goal}
        
        请一步步思考如何完成这个目标：
        1. 首先分析需要做什么
        2. 然后分解为具体步骤
        3. 最后确定每步需要的工具
        
        思考过程：
        """
        
        # LLM 生成推理过程
        reasoning = self.llm.generate(prompt)
        
        # 从推理中提取步骤
        steps = self.extract_steps(reasoning)
        
        return steps
    
    def extract_steps(self, reasoning: str) -> list[str]:
        """从推理文本中提取步骤"""
        import re
        pattern = r"Step \d+:(.+)"
        matches = re.findall(pattern, reasoning)
        return [m.strip() for m in matches]

┌─────────────────────────────────────────────────────────────────┐
│                    ToT 思维树示例                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  目标: "帮我写一篇技术博客"                                      │
│                                                                 │
│                        ┌─────────────┐                          │
│                        │   根节点     │                          │
│                        │ 确定主题方向 │                          │
│                        └──────┬──────┘                          │
│               ┌───────────────┼───────────────┐                 │
│               ↓               ↓               ↓                 │
│        ┌──────────┐    ┌──────────┐    ┌──────────┐            │
│        │ AI技术   │    │ 架构设计 │    │ 性能优化 │            │
│        └────┬─────┘    └────┬─────┘    └────┬─────┘            │
│             │               │               │                   │
│     ┌───────┴───────┐       │       ┌───────┴───────┐          │
│     ↓               ↓       │       ↓               ↓          │
│ ┌────────┐    ┌────────┐    │   ┌────────┐    ┌────────┐      │
│ │LLM原理 │    │Agent实战│    │   │缓存策略│    │JVM调优 │      │
│ └────────┘    └────────┘    │   └────────┘    └────────┘      │
│                             │                                 │
│                     评估选择最优路径 → 执行                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

# ToT 规划器实现
class ToTPlanner:
    def __init__(self, llm, max_depth=3, beam_width=3):
        self.llm = llm
        self.max_depth = max_depth
        self.beam_width = beam_width  # 保留的最优路径数
    
    def plan(self, goal: str) -> list[str]:
        """
        使用思维树探索最优规划
        """
        # 初始化根节点
        root = ThoughtNode(goal=goal, depth=0)
        
        # 广度优先探索
        current_nodes = [root]
        
        for depth in range(self.max_depth):
            next_nodes = []
            
            for node in current_nodes:
                # 生成子节点（多种可能路径）
                children = self.generate_thoughts(node)
                next_nodes.extend(children)
            
            # 评估并选择最优路径
            next_nodes = self.evaluate_and_select(next_nodes)
            current_nodes = next_nodes
            
            # 检查是否已找到解决方案
            if any(node.is_solution for node in current_nodes):
                break
        
        # 返回最优路径
        best_path = self.get_best_path(current_nodes)
        return best_path
    
    def generate_thoughts(self, node: ThoughtNode) -> list[ThoughtNode]:
        """生成多个可能的下一步思考"""
        prompt = f"""
        当前状态: {node.state}
        目标: {node.goal}
        
        请生成 3 个可能的下一步思考方向：
        """
        
        thoughts = self.llm.generate(prompt)
        return [ThoughtNode(state=t, parent=node) for t in thoughts]
    
    def evaluate_and_select(self, nodes: list[ThoughtNode]) -> list[ThoughtNode]:
        """评估节点价值，选择最优的 beam_width 个"""
        scored_nodes = []
        
        for node in nodes:
            score = self.evaluate_node(node)
            scored_nodes.append((score, node))
        
        # 按分数排序，选择前 beam_width 个
        scored_nodes.sort(key=lambda x: x[0], reverse=True)
        return [node for _, node in scored_nodes[:self.beam_width]]

┌─────────────────────────────────────────────────────────────────┐
│                    ReAct 循环流程                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                     ReAct 循环                            │  │
│  │                                                          │  │
│  │    ┌─────────┐                                           │  │
│  │    │ Thought │ ←─────────────────────────────┐           │  │
│  │    │ (思考)   │                              │           │  │
│  │    └────┬────┘                              │           │  │
│  │         ↓                                   │           │  │
│  │    ┌─────────┐                              │           │  │
│  │    │ Action  │  执行工具调用                 │           │  │
│  │    │ (行动)   │                              │           │  │
│  │    └────┬────┘                              │           │  │
│  │         ↓                                   │           │  │
│  │    ┌───────────┐                            │           │  │
│  │    │Observation│  观察执行结果               │           │  │
│  │    │ (观察)    │                            │           │  │
│  │    └────┬──────┘                            │           │  │
│  │         │                                   │           │  │
│  │         └───────────────────────────────────┘           │  │
│  │                   (循环直到完成)                         │  │
│  │                                                          │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
│  示例执行过程:                                                  │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │ Thought: 用户想知道北京天气，我需要查询天气 API           │  │
│  │ Action: weather_api(city="北京")                         │  │
│  │ Observation: {"temp": 25, "weather": "晴"}               │  │
│  │ Thought: 已获取天气信息，可以回答用户了                   │  │
│  │ Answer: 北京今天天气晴朗，气温 25°C                       │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

# ReAct 规划器实现
class ReActPlanner:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
    
    def run(self, goal: str, max_iterations: int = 10):
        """
        ReAct 循环执行
        """
        context = f"目标: {goal}\n"
        
        for i in range(max_iterations):
            # Thought: 思考下一步
            thought = self.think(context)
            context += f"Thought: {thought}\n"
            
            # 检查是否可以给出最终答案
            if "Answer:" in thought or self.is_complete(thought):
                return self.extract_answer(thought)
            
            # Action: 决定执行什么动作
            action, action_input = self.decide_action(thought)
            context += f"Action: {action}({action_input})\n"
            
            # 执行动作
            observation = self.execute_action(action, action_input)
            context += f"Observation: {observation}\n"
        
        return "超过最大迭代次数，任务未完成"
    
    def think(self, context: str) -> str:
        """思考下一步"""
        prompt = f"""
        {context}
        
        请思考下一步应该做什么。如果已经有足够信息，给出最终答案。
        
        格式：
        - 如果需要执行操作: Thought: [你的思考]
        - 如果可以回答: Answer: [最终答案]
        """
        return self.llm.generate(prompt)
    
    def decide_action(self, thought: str) -> tuple[str, dict]:
        """决定执行什么动作"""
        prompt = f"""
        基于思考: {thought}
        
        可用工具: {list(self.tools.keys())}
        
        请选择要执行的工具和参数，格式: tool_name|{{"param": "value"}}
        """
        response = self.llm.generate(prompt)
        action, params = self.parse_action(response)
        return action, params
    
    def execute_action(self, action: str, params: dict) -> str:
        """执行动作并返回观察结果"""
        if action in self.tools:
            tool = self.tools[action]
            return tool.execute(**params)
        return f"错误: 未知工具 {action}"

┌─────────────────────────────────────────────────────────────────┐
│                    Agent 三层记忆架构                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                 第一层：短期记忆                           │  │
│  │                  Short-Term Memory                        │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  特点: 会话级别、有限容量、快速访问                  │  │  │
│  │  │  存储: 当前会话的对话历史                            │  │  │
│  │  │  技术: 滑动窗口、摘要压缩                            │  │  │
│  │  │  生命周期: 会话结束即清除                            │  │  │
│  │  │                                                     │  │  │
│  │  │  示例:                                              │  │  │
│  │  │  [User] 帮我查北京天气                              │  │  │
│  │  │  [Agent] 北京今天晴，25°C                           │  │  │
│  │  │  [User] 明天呢？  ←── 需要记住上下文中的"北京"      │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                 第二层：长期记忆                           │  │
│  │                  Long-Term Memory                         │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  特点: 持久化、无限容量、语义检索                    │  │  │
│  │  │  存储: 重要对话、学习知识、用户偏好                  │  │  │
│  │  │  技术: 向量数据库(FAISS/Milvus/Pinecone)            │  │  │
│  │  │  生命周期: 永久保存                                  │  │  │
│  │  │                                                     │  │  │
│  │  │  存储流程:                                          │  │  │
│  │  │  文本 → Embedding模型 → 向量 → 向量数据库            │  │  │
│  │  │                                                     │  │  │
│  │  │  检索流程:                                          │  │  │
│  │  │  查询 → Embedding → 相似度搜索 → 返回相关记忆       │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                 第三层：工作记忆                           │  │
│  │                  Working Memory                           │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  特点: 任务级别、临时状态、执行跟踪                  │  │  │
│  │  │  存储: 当前任务状态、中间结果、执行进度              │  │  │
│  │  │  技术: 状态机、任务队列                              │  │  │
│  │  │  生命周期: 任务完成即清除                            │  │  │
│  │  │                                                     │  │  │
│  │  │  示例: 多步骤任务执行状态                            │  │  │
│  │  │  Task: "分析销售数据并生成报告"                      │  │  │
│  │  │  ├── Step 1: 读取数据 ✓ (result: DataFrame)         │  │  │
│  │  │  ├── Step 2: 数据清洗 ⏳ (in_progress)               │  │  │
│  │  │  ├── Step 3: 生成报告 (pending)                      │  │  │
│  │  │  └── Current: 正在执行 Step 2                        │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

from typing import Optional
from dataclasses import dataclass, field
from datetime import datetime
import json
 
# 向量数据库接口（示例使用 FAISS）
class VectorDatabase:
    def __init__(self, embedding_model: str = "text-embedding-3-small"):
        self.embedding_model = embedding_model
        self.index = None  # FAISS 索引
        self.metadata = []  # 存储元数据
    
    def add(self, text: str, metadata: dict = None):
        """添加记忆到向量数据库"""
        embedding = self.get_embedding(text)
        # 向量化并存储
        self.index.add(embedding)
        self.metadata.append({"text": text, **metadata})
    
    def search(self, query: str, top_k: int = 5) -> list[dict]:
        """语义搜索相关记忆"""
        query_embedding = self.get_embedding(query)
        distances, indices = self.index.search(query_embedding, top_k)
        return [self.metadata[i] for i in indices[0]]
    
    def get_embedding(self, text: str) -> list[float]:
        """获取文本向量（调用 Embedding API）"""
        # 实际实现调用 OpenAI 或其他 Embedding 服务
        pass
 
 
# 短期记忆实现
class ShortTermMemory:
    def __init__(self, max_tokens: int = 4000):
        self.messages = []  # 对话历史
        self.max_tokens = max_tokens
        self.current_tokens = 0
    
    def add(self, role: str, content: str):
        """添加消息到短期记忆"""
        message = {"role": role, "content": content, "timestamp": datetime.now()}
        self.messages.append(message)
        self.current_tokens += self.count_tokens(content)
        
        # 超过限制时压缩或丢弃旧消息
        while self.current_tokens > self.max_tokens:
            removed = self.messages.pop(0)
            self.current_tokens -= self.count_tokens(removed["content"])
    
    def get_context(self) -> list[dict]:
        """获取当前上下文"""
        return self.messages
    
    def count_tokens(self, text: str) -> int:
        """估算 token 数量"""
        return len(text) // 4  # 简化估算
 
 
# 长期记忆实现
class LongTermMemory:
    def __init__(self, db_path: str = "./memory_db"):
        self.vector_db = VectorDatabase()
        self.db_path = db_path
    
    def store(self, text: str, importance: str = "normal", tags: list = None):
        """存储到长期记忆"""
        metadata = {
            "timestamp": datetime.now().isoformat(),
            "importance": importance,
            "tags": tags or []
        }
        self.vector_db.add(text, metadata)
    
    def recall(self, query: str, top_k: int = 5) -> list[dict]:
        """从长期记忆中检索"""
        return self.vector_db.search(query, top_k)
    
    def store_conversation(self, user_input: str, agent_response: str):
        """存储重要对话"""
        # 判断对话重要性
        importance = self.evaluate_importance(user_input, agent_response)
        
        if importance in ["high", "critical"]:
            combined = f"User: {user_input}\nAgent: {agent_response}"
            self.store(combined, importance=importance)
 
 
# 工作记忆实现
class WorkingMemory:
    def __init__(self):
        self.current_task = None
        self.step_results = {}  # 步骤执行结果
        self.task_status = "idle"  # idle, running, completed, failed
    
    def start_task(self, task_id: str, plan: list):
        """开始新任务"""
        self.current_task = {
            "id": task_id,
            "plan": plan,
            "current_step": 0,
            "total_steps": len(plan)
        }
        self.task_status = "running"
        self.step_results = {}
    
    def update_step(self, step_index: int, result: str, status: str):
        """更新步骤状态"""
        self.step_results[step_index] = {
            "result": result,
            "status": status,
            "timestamp": datetime.now()
        }
        
        if status == "completed":
            self.current_task["current_step"] = step_index + 1
        elif status == "failed":
            self.task_status = "failed"
        
        # 检查任务是否完成
        if self.current_task["current_step"] >= self.current_task["total_steps"]:
            self.task_status = "completed"
    
    def get_current_state(self) -> dict:
        """获取当前任务状态"""
        return {
            "task": self.current_task,
            "status": self.task_status,
            "completed_steps": len([s for s in self.step_results.values() if s["status"] == "completed"]),
            "results": self.step_results
        }
 
 
# 统一记忆管理器
class AgentMemory:
    def __init__(self):
        self.short_term = ShortTermMemory()
        self.long_term = LongTermMemory()
        self.working = WorkingMemory()
    
    def remember(self, content: str, memory_type: str = "short", **kwargs):
        """
        存储记忆
        
        Args:
            content: 记忆内容
            memory_type: short/long/working
        """
        if memory_type == "short":
            self.short_term.add("system", content)
        elif memory_type == "long":
            self.long_term.store(content, **kwargs)
        elif memory_type == "working":
            # 工作记忆通过任务管理
            pass
    
    def recall(self, query: str = None, memory_type: str = "all") -> str:
        """
        检索记忆
        
        Args:
            query: 查询内容
            memory_type: short/long/working/all
        """
        results = []
        
        if memory_type in ["short", "all"]:
            context = self.short_term.get_context()
            results.append(f"[短期记忆] {context}")
        
        if memory_type in ["long", "all"] and query:
            memories = self.long_term.recall(query)
            results.append(f"[长期记忆] {memories}")
        
        if memory_type in ["working", "all"]:
            state = self.working.get_current_state()
            results.append(f"[工作记忆] {state}")
        
        return "\n".join(results)

┌─────────────────────────────────────────────────────────────────┐
│                    RAG 检索增强生成流程                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────┐                                               │
│  │   用户查询   │                                               │
│  │"如何优化SQL" │                                               │
│  └──────┬──────┘                                               │
│         ↓                                                       │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   1. 查询向量化                           │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Query → Embedding Model → Vector [0.1, 0.3, ...]  │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   2. 向量相似度搜索                       │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  向量数据库中搜索相似向量                           │  │  │
│  │  │  cosine_similarity(query_vec, stored_vec)          │  │  │
│  │  │                                                     │  │  │
│  │  │  Top-K 结果:                                        │  │  │
│  │  │  1. "SQL索引优化技巧" (similarity: 0.92)           │  │  │
│  │  │  2. "数据库查询性能调优" (similarity: 0.88)        │  │  │
│  │  │  3. "慢查询分析与优化" (similarity: 0.85)          │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   3. 构建增强提示                         │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Context: [检索到的相关记忆]                        │  │  │
│  │  │                                                     │  │  │
│  │  │  相关知识:                                         │  │  │
│  │  │  - SQL索引优化技巧...                              │  │  │
│  │  │  - 数据库查询性能调优...                           │  │  │
│  │  │                                                     │  │  │
│  │  │  用户问题: 如何优化SQL？                            │  │  │
│  │  │  请基于以上知识回答用户问题。                       │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   4. LLM 生成回答                         │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  基于检索到的上下文生成准确回答                     │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    Agent 常用工具分类                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   信息获取类工具                          │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐            │  │
│  │  │搜索引擎│ │天气API │ │股票API │ │新闻API │            │  │
│  │  └────────┘ └────────┘ └────────┘ └────────┘            │  │
│  │  特点: 获取实时、动态的外部信息                           │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   代码执行类工具                          │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐            │  │
│  │  │Python  │ │Bash    │ │SQL     │ │Docker  │            │  │
│  │  │执行器  │ │终端    │ │查询    │ │容器    │            │  │
│  │  └────────┘ └────────┘ └────────┘ └────────┘            │  │
│  │  特点: 在沙箱环境安全执行代码                             │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   数据操作类工具                          │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐            │  │
│  │  │文件读写│ │数据库  │ │Excel   │ │CSV     │            │  │
│  │  │        │ │操作    │ │处理    │ │解析    │            │  │
│  │  └────────┘ └────────┘ └────────┘ └────────┘            │  │
│  │  特点: 操作本地或远程数据                                 │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                   通信交互类工具                          │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐            │  │
│  │  │邮件    │ │HTTP    │ │消息    │ │Webhook │            │  │
│  │  │发送    │ │请求    │ │推送    │ │触发    │            │  │
│  │  └────────┘ └────────┘ └────────┘ └────────┘            │  │
│  │  特点: 与外部系统通信交互                                 │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    Function Calling 流程                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  Step 1: 工具定义                                              │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  tools = [                                                │  │
│  │    {                                                      │  │
│  │      "type": "function",                                  │  │
│  │      "function": {                                        │  │
│  │        "name": "get_weather",                             │  │
│  │        "description": "获取指定城市的天气信息",            │  │
│  │        "parameters": {                                    │  │
│  │          "type": "object",                                │  │
│  │          "properties": {                                  │  │
│  │            "city": {                                      │  │
│  │              "type": "string",                            │  │
│  │              "description": "城市名称，如: 北京、上海"     │  │
│  │            }                                              │  │
│  │          },                                               │  │
│  │          "required": ["city"]                             │  │
│  │        }                                                  │  │
│  │      }                                                    │  │
│  │    }                                                      │  │
│  │  ]                                                        │  │
│  └──────────────────────────────────────────────────────────┘  │
│                             ↓                                   │
│  Step 2: LLM 决策调用                                         │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  用户输入: "北京今天天气怎么样？"                          │  │
│  │                                                           │  │
│  │  LLM 分析:                                                │  │
│  │  - 用户想查询天气                                         │  │
│  │  - 城市是"北京"                                           │  │
│  │  - 匹配到工具 get_weather                                 │  │
│  │                                                           │  │
│  │  LLM 返回:                                                │  │
│  │  {                                                        │  │
│  │    "function_call": {                                     │  │
│  │      "name": "get_weather",                               │  │
│  │      "arguments": "{\"city\": \"北京\"}"                  │  │
│  │    }                                                      │  │
│  │  }                                                        │  │
│  └──────────────────────────────────────────────────────────┘  │
│                             ↓                                   │
│  Step 3: 执行工具并返回结果                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  # 执行工具                                               │  │
│  │  result = get_weather(city="北京")                        │  │
│  │  # result = {"temp": 25, "weather": "晴", "humidity": 60}│  │
│  │                                                           │  │
│  │  # 将结果返回给 LLM                                       │  │
│  │  messages.append({                                        │  │
│  │    "role": "tool",                                        │  │
│  │    "content": json.dumps(result)                          │  │
│  │  })                                                       │  │
│  └──────────────────────────────────────────────────────────┘  │
│                             ↓                                   │
│  Step 4: LLM 生成最终回复                                     │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  LLM 基于工具返回结果生成自然语言回复:                     │  │
│  │  "北京今天天气晴朗，气温 25°C，湿度 60%，适合外出活动。"   │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

from typing import Callable, Any
from pydantic import BaseModel, Field
from dataclasses import dataclass
import json
 
# 工具基类
@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    execute: Callable
    
    def to_openai_format(self) -> dict:
        """转换为 OpenAI Function Calling 格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
            }
        }
 
 
# 具体工具实现示例
class WeatherTool(Tool):
    def __init__(self):
        super().__init__(
            name="get_weather",
            description="获取指定城市的实时天气信息",
            parameters={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如: 北京、上海、广州"
                    }
                },
                "required": ["city"]
            },
            execute=self._execute
        )
    
    def _execute(self, city: str) -> dict:
        """实际执行天气查询"""
        # 调用天气 API
        import requests
        response = requests.get(f"https://api.weather.com/{city}")
        return response.json()
 
 
class CodeExecutorTool(Tool):
    def __init__(self, sandbox):
        super().__init__(
            name="execute_python",
            description="在安全沙箱中执行 Python 代码，返回执行结果",
            parameters={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "要执行的 Python 代码"
                    },
                    "timeout": {
                        "type": "integer",
                        "description": "执行超时时间（秒）",
                        "default": 30
                    }
                },
                "required": ["code"]
            },
            execute=self._execute
        )
        self.sandbox = sandbox
    
    def _execute(self, code: str, timeout: int = 30) -> str:
        """在沙箱中安全执行代码"""
        return self.sandbox.run(code, timeout=timeout)
 
 
class DatabaseQueryTool(Tool):
    def __init__(self, db_connection):
        super().__init__(
            name="query_database",
            description="执行 SQL 查询并返回结果",
            parameters={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL 查询语句"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回结果数量限制",
                        "default": 100
                    }
                },
                "required": ["query"]
            },
            execute=self._execute
        )
        self.db = db_connection
    
    def _execute(self, query: str, limit: int = 100) -> list:
        """执行数据库查询"""
        # 安全检查：防止危险 SQL
        if self.is_dangerous_query(query):
            raise ValueError("危险 SQL 操作被拒绝")
        
        cursor = self.db.cursor()
        cursor.execute(query)
        return cursor.fetchmany(limit)
    
    def is_dangerous_query(self, query: str) -> bool:
        """检查是否为危险查询"""
        dangerous_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
        return any(kw in query.upper() for kw in dangerous_keywords)
 
 
# 工具注册管理器
class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, tool: Tool):
        """注册工具"""
        self.tools[tool.name] = tool
    
    def get(self, name: str) -> Tool:
        """获取工具"""
        return self.tools.get(name)
    
    def list_tools(self) -> list[dict]:
        """列出所有工具（OpenAI 格式）"""
        return [tool.to_openai_format() for tool in self.tools.values()]
    
    def execute(self, name: str, **kwargs) -> Any:
        """执行工具"""
        tool = self.get(name)
        if not tool:
            raise ValueError(f"未知工具: {name}")
        return tool.execute(**kwargs)
 
 
# 使用示例
def main():
    # 创建工具注册器
    registry = ToolRegistry()
    
    # 注册工具
    registry.register(WeatherTool())
    registry.register(CodeExecutorTool(sandbox=CodeSandbox()))
    registry.register(DatabaseQueryTool(db=get_db_connection()))
    
    # 获取 OpenAI 格式的工具定义
    tools = registry.list_tools()
    
    # 调用 LLM（伪代码）
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "北京天气怎么样？"}],
        tools=tools
    )
    
    # 处理工具调用
    if response.choices[0].message.tool_calls:
        for tool_call in response.choices[0].message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            result = registry.execute(function_name, **arguments)
            print(f"工具 {function_name} 返回: {result}")

class ToolSelector:
    """智能工具选择器"""
    
    def __init__(self, llm, tools: list[Tool]):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
    
    def select_tool(self, task: str, context: dict = None) -> tuple[Tool, dict]:
        """
        根据任务选择最合适的工具
        
        Returns:
            (选中的工具, 参数字典)
        """
        # 构建工具描述
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])
        
        prompt = f"""
        任务: {task}
        上下文: {context or '无'}
        
        可用工具:
        {tool_descriptions}
        
        请选择最合适的工具并提供参数，格式：
        工具名称|{{参数}}
        """
        
        response = self.llm.generate(prompt)
        tool_name, params = self.parse_response(response)
        
        return self.tools[tool_name], params
    
    def parse_response(self, response: str) -> tuple[str, dict]:
        """解析 LLM 返回的工具选择"""
        parts = response.strip().split("|")
        tool_name = parts[0].strip()
        params = json.loads(parts[1]) if len(parts) > 1 else {}
        return tool_name, params

┌─────────────────────────────────────────────────────────────────┐
│                    执行器内部架构                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   来自规划器的任务                                              │
│         ↓                                                       │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    1. 任务解析层                          │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  - 解析任务类型                                     │  │  │
│  │  │  - 确定需要的工具                                   │  │  │
│  │  │  - 提取关键参数                                     │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    2. 安全检查层                          │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  - 权限验证：用户是否有权限执行此操作               │  │  │
│  │  │  - 参数校验：参数是否合法、是否包含危险内容         │  │  │
│  │  │  - 资源限制：检查资源使用是否超限                   │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    3. 沙箱执行层                          │  │
│  │  ┌────────────────────────────────────────────────────┐  │
│  │  │  ┌─────────────────────────────────────────────┐   │  │  │
│  │  │  │              隔离沙箱环境                    │   │  │  │
│  │  │  │  ┌───────────┐  ┌───────────┐              │   │  │  │
│  │  │  │  │ 文件隔离   │  │ 网络隔离  │              │   │  │  │
│  │  │  │  └───────────┘  └───────────┘              │   │  │  │
│  │  │  │  ┌───────────┐  ┌───────────┐              │   │  │  │
│  │  │  │  │ 进程隔离   │  │ 资源限制  │              │   │  │  │
│  │  │  │  └───────────┘  └───────────┘              │   │  │  │
│  │  │  │                                             │   │  │  │
│  │  │  │         执行工具调用                        │   │  │  │
│  │  │  └─────────────────────────────────────────────┘   │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    4. 结果处理层                          │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  - 执行结果格式化                                   │  │  │
│  │  │  - 错误处理和重试                                   │  │  │
│  │  │  - 日志记录                                         │  │  │
│  │  │  - 反馈给规划器                                     │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                             ↓                                   │
│                      返回执行结果                               │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    安全沙箱机制                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │                    沙箱隔离维度                            │  │
│  ├──────────────────────────────────────────────────────────┤  │
│  │                                                          │  │
│  │  ┌─────────────────────────────────────────────────────┐│  │
│  │  │  1. 文件系统隔离                                     ││  │
│  │  │  ┌─────────────────────────────────────────────┐    ││  │
│  │  │  │  - 限制可访问目录（如 /tmp/sandbox/）         │    ││  │
│  │  │  │  - 禁止访问敏感文件（/etc/passwd 等）         │    ││  │
│  │  │  │  - 文件操作审计                              │    ││  │
│  │  │  └─────────────────────────────────────────────┘    ││  │
│  │  └─────────────────────────────────────────────────────┘│  │
│  │                                                          │  │
│  │  ┌─────────────────────────────────────────────────────┐│  │
│  │  │  2. 网络隔离                                         ││  │
│  │  │  ┌─────────────────────────────────────────────┐    ││  │
│  │  │  │  - 白名单域名访问                            │    ││  │
│  │  │  │  - 禁止内网访问                              │    ││  │
│  │  │  │  - 网络请求监控                              │    ││  │
│  │  │  └─────────────────────────────────────────────┘    ││  │
│  │  └─────────────────────────────────────────────────────┘│  │
│  │                                                          │  │
│  │  ┌─────────────────────────────────────────────────────┐│  │
│  │  │  3. 进程隔离                                         ││  │
│  │  │  ┌─────────────────────────────────────────────┐    ││  │
│  │  │  │  - Docker 容器隔离                           │    ││  │
│  │  │  │  - 或子进程隔离                              │    ││  │
│  │  │  │  - 禁止 fork 炸弹等                          │    ││  │
│  │  │  └─────────────────────────────────────────────┘    ││  │
│  │  └─────────────────────────────────────────────────────┘│  │
│  │                                                          │  │
│  │  ┌─────────────────────────────────────────────────────┐│  │
│  │  │  4. 资源限制                                         ││  │
│  │  │  ┌─────────────────────────────────────────────┐    ││  │
│  │  │  │  - CPU 时间限制（如 30 秒）                   │    ││  │
│  │  │  │  - 内存限制（如 512MB）                      │    ││  │
│  │  │  │  - 磁盘空间限制                              │    ││  │
│  │  │  │  - 网络带宽限制                              │    ││  │
│  │  │  └─────────────────────────────────────────────┘    ││  │
│  │  └─────────────────────────────────────────────────────┘│  │
│  │                                                          │  │
│  └──────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

import subprocess
import docker
import tempfile
import os
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum
 
class ExecutionStatus(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"
    PERMISSION_DENIED = "permission_denied"
 
 
@dataclass
class ExecutionResult:
    status: ExecutionStatus
    output: str
    error: Optional[str] = None
    exit_code: int = 0
    execution_time: float = 0.0
 
 
# 沙箱环境基类
class Sandbox:
    """安全沙箱基类"""
    
    def run(self, code: str, timeout: int = 30) -> ExecutionResult:
        raise NotImplementedError
 
 
# Docker 沙箱实现
class DockerSandbox(Sandbox):
    """Docker 容器沙箱 - 最安全的隔离方式"""
    
    def __init__(self, image: str = "python:3.10-slim"):
        self.client = docker.from_env()
        self.image = image
    
    def run(self, code: str, timeout: int = 30) -> ExecutionResult:
        """在 Docker 容器中执行代码"""
        import time
        start_time = time.time()
        
        try:
            # 创建临时文件
            with tempfile.NamedTemporaryFile(
                mode='w', suffix='.py', delete=False
            ) as f:
                f.write(code)
                temp_file = f.name
            
            # 运行容器
            container = self.client.containers.run(
                self.image,
                command=f"python /code/script.py",
                volumes={temp_file: {'bind': '/code/script.py', 'mode': 'ro'}},
                mem_limit="512m",
                cpu_period=100000,
                cpu_quota=50000,  # 50% CPU
                network_disabled=True,  # 禁用网络
                timeout=timeout,
                remove=True
            )
            
            execution_time = time.time() - start_time
            
            return ExecutionResult(
                status=ExecutionStatus.SUCCESS,
                output=container.decode('utf-8'),
                execution_time=execution_time
            )
            
        except subprocess.TimeoutExpired:
            return ExecutionResult(
                status=ExecutionStatus.TIMEOUT,
                output="",
                error=f"执行超时（{timeout}秒）"
            )
        except Exception as e:
            return ExecutionResult(
                status=ExecutionStatus.FAILED,
                output="",
                error=str(e)
            )
        finally:
            # 清理临时文件
            if os.path.exists(temp_file):
                os.unlink(temp_file)
 
 
# 进程沙箱实现（较轻量）
class ProcessSandbox(Sandbox):
    """子进程沙箱 - 轻量级隔离"""
    
    def __init__(self, allowed_dirs: list[str] = None):
        self.allowed_dirs = allowed_dirs or [tempfile.gettempdir()]
    
    def run(self, code: str, timeout: int = 30) -> ExecutionResult:
        """在受限子进程中执行代码"""
        import time
        start_time = time.time()
        
        # 安全检查
        if self.contains_dangerous_code(code):
            return ExecutionResult(
                status=ExecutionStatus.PERMISSION_DENIED,
                output="",
                error="代码包含危险操作"
            )
        
        try:
            # 使用 subprocess 执行
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=self.allowed_dirs[0],  # 限制工作目录
                env=self.get_safe_env()  # 安全的环境变量
            )
            
            execution_time = time.time() - start_time
            
            if result.returncode == 0:
                return ExecutionResult(
                    status=ExecutionStatus.SUCCESS,
                    output=result.stdout,
                    exit_code=result.returncode,
                    execution_time=execution_time
                )
            else:
                return ExecutionResult(
                    status=ExecutionStatus.FAILED,
                    output=result.stdout,
                    error=result.stderr,
                    exit_code=result.returncode,
                    execution_time=execution_time
                )
                
        except subprocess.TimeoutExpired:
            return ExecutionResult(
                status=ExecutionStatus.TIMEOUT,
                output="",
                error=f"执行超时（{timeout}秒）"
            )
    
    def contains_dangerous_code(self, code: str) -> bool:
        """检查危险代码"""
        dangerous_patterns = [
            "import os", "import subprocess", "import sys",
            "__import__", "eval(", "exec(",
            "open(", "file(", "input(",
            "rm -rf", "shutdown", "reboot"
        ]
        return any(pattern in code for pattern in dangerous_patterns)
    
    def get_safe_env(self) -> dict:
        """获取安全的环境变量"""
        return {
            "PATH": "/usr/bin:/bin",
            "PYTHONPATH": "",
            "HOME": tempfile.gettempdir()
        }
 
 
# 执行器主类
class AgentExecutor:
    """Agent 执行器"""
    
    def __init__(
        self,
        tools: dict,
        sandbox: Sandbox = None,
        max_retries: int = 3,
        default_timeout: int = 30
    ):
        self.tools = tools
        self.sandbox = sandbox or ProcessSandbox()
        self.max_retries = max_retries
        self.default_timeout = default_timeout
    
    def execute(self, task: dict) -> ExecutionResult:
        """
        执行任务
        
        Args:
            task: {
                "tool": "工具名称",
                "action": "具体操作",
                "params": {"参数": "值"},
                "timeout": 30
            }
        """
        tool_name = task.get("tool")
        params = task.get("params", {})
        timeout = task.get("timeout", self.default_timeout)
        
        # 1. 检查工具是否存在
        if tool_name not in self.tools:
            return ExecutionResult(
                status=ExecutionStatus.FAILED,
                output="",
                error=f"未知工具: {tool_name}"
            )
        
        tool = self.tools[tool_name]
        
        # 2. 权限检查
        if not self.check_permission(tool, params):
            return ExecutionResult(
                status=ExecutionStatus.PERMISSION_DENIED,
                output="",
                error="权限不足"
            )
        
        # 3. 参数验证
        validation_result = self.validate_params(tool, params)
        if not validation_result.valid:
            return ExecutionResult(
                status=ExecutionStatus.FAILED,
                output="",
                error=f"参数验证失败: {validation_result.message}"
            )
        
        # 4. 执行（带重试）
        for attempt in range(self.max_retries):
            try:
                result = tool.execute(**params)
                return ExecutionResult(
                    status=ExecutionStatus.SUCCESS,
                    output=str(result)
                )
            except Exception as e:
                if attempt == self.max_retries - 1:
                    return ExecutionResult(
                        status=ExecutionStatus.FAILED,
                        output="",
                        error=f"执行失败（重试 {self.max_retries} 次）: {str(e)}"
                    )
                # 等待后重试
                time.sleep(2 ** attempt)
    
    def check_permission(self, tool, params: dict) -> bool:
        """检查执行权限"""
        # 实现权限检查逻辑
        return True
    
    def validate_params(self, tool, params: dict):
        """验证参数"""
        from dataclasses import dataclass
        
        @dataclass
        class ValidationResult:
            valid: bool
            message: str = ""
        
        # 使用工具定义的参数 schema 验证
        schema = tool.parameters
        
        # 检查必需参数
        required = schema.get("required", [])
        for param in required:
            if param not in params:
                return ValidationResult(
                    valid=False,
                    message=f"缺少必需参数: {param}"
                )
        
        return ValidationResult(valid=True)
 
 
# 使用示例
def main():
    # 创建沙箱
    sandbox = DockerSandbox(image="python:3.10-slim")
    
    # 创建执行器
    executor = AgentExecutor(
        tools={
            "execute_python": CodeExecutorTool(sandbox),
            "get_weather": WeatherTool()
        },
        sandbox=sandbox,
        max_retries=3
    )
    
    # 执行任务
    result = executor.execute({
        "tool": "execute_python",
        "params": {"code": "print('Hello, World!')"},
        "timeout": 10
    })
    
    print(f"状态: {result.status}")
    print(f"输出: {result.output}")

class ErrorHandler:
    """执行错误处理器"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def handle_error(
        self,
        error: Exception,
        task: dict,
        execution_history: list
    ) -> dict:
        """
        分析错误并生成恢复策略
        
        Returns:
            {
                "strategy": "retry" | "alternative" | "abort",
                "alternative_action": {...},  # 替代方案
                "reason": "错误原因分析"
            }
        """
        prompt = f"""
        任务执行失败，请分析原因并给出解决方案：
        
        任务: {task}
        错误: {str(error)}
        历史执行: {execution_history}
        
        请选择处理策略：
        1. retry - 重新尝试相同操作
        2. alternative - 使用替代方案
        3. abort - 放弃任务
        
        返回格式：
        策略|原因|替代方案（如果有）
        """
        
        response = self.llm.generate(prompt)
        return self.parse_response(response)
    
    def parse_response(self, response: str) -> dict:
        """解析 LLM 返回的错误处理策略"""
        parts = response.strip().split("|")
        return {
            "strategy": parts[0].strip(),
            "reason": parts[1].strip() if len(parts) > 1 else "",
            "alternative_action": parts[2].strip() if len(parts) > 2 else None
        }

┌─────────────────────────────────────────────────────────────────┐
│                    四大组件协作流程                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户目标: "帮我分析 sales.csv 文件并生成销售报告"               │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Step 1: 规划器 (Planning)                                │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Thought: 需要读取 CSV、分析数据、生成报告           │  │  │
│  │  │  Plan:                                               │  │  │
│  │  │  1. 读取 sales.csv                                   │  │  │
│  │  │  2. 数据统计分析                                      │  │  │
│  │  │  3. 生成可视化图表                                    │  │  │
│  │  │  4. 输出报告文档                                      │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Step 2: 记忆库 (Memory)                                  │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  - 工作记忆: 初始化任务状态                          │  │  │
│  │  │  - 短期记忆: 记录当前目标                            │  │  │
│  │  │  - 长期记忆: 检索类似任务经验                        │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Step 3: 工具集 (Tools) - 工具选择                        │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  可用工具:                                          │  │  │
│  │  │  - file_reader: 读取文件                            │  │  │
│  │  │  - python_executor: 执行 Python 代码                │  │  │
│  │  │  - chart_generator: 生成图表                        │  │  │
│  │  │  - document_writer: 生成文档                        │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Step 4: 执行器 (Action) - 执行任务 1                     │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  Action: file_reader(file="sales.csv")              │  │  │
│  │  │  Result: DataFrame with 1000 rows                  │  │  │
│  │  │  Memory Update: 存储数据到工作记忆                   │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Step 5: 循环执行 → 反思 → 调整                          │  │
│  │  ┌────────────────────────────────────────────────────┐  │  │
│  │  │  规划器: 分析执行结果，决定下一步                    │  │  │
│  │  │  执行器: 执行任务 2, 3, 4                           │  │  │
│  │  │  记忆库: 更新任务进度和中间结果                      │  │  │
│  │  └────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────┬───────────────────────────────┘  │
│                             ↓                                   │
│                      最终结果: sales_report.pdf                 │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

class Agent:
    """完整 Agent 实现：四大组件协作"""
    
    def __init__(self, llm, tools: list, sandbox=None):
        # 初始化四大组件
        self.planner = ReActPlanner(llm, tools)
        self.memory = AgentMemory()
        self.tools = ToolRegistry()
        self.executor = AgentExecutor(
            tools=self.tools.tools,
            sandbox=sandbox
        )
        
        # 注册工具
        for tool in tools:
            self.tools.register(tool)
    
    def run(self, goal: str, max_iterations: int = 10) -> str:
        """
        执行用户目标
        
        Args:
            goal: 用户目标
            max_iterations: 最大迭代次数
        """
        # 初始化工作记忆
        self.memory.working.start_task(
            task_id=self.generate_task_id(),
            plan=[]
        )
        
        # 记录目标到短期记忆
        self.memory.short_term.add("user", goal)
        
        context = f"目标: {goal}\n"
        
        for iteration in range(max_iterations):
            # 1. 规划：思考下一步
            thought = self.planner.think(context)
            context += f"Thought: {thought}\n"
            
            # 2. 检查是否完成
            if self.is_task_complete(thought):
                answer = self.extract_answer(thought)
                # 存储到长期记忆
                self.memory.long_term.store_conversation(goal, answer)
                return answer
            
            # 3. 决策：选择工具和参数
            action, params = self.planner.decide_action(thought)
            context += f"Action: {action}({params})\n"
            
            # 4. 执行：在沙箱中安全执行
            execution_result = self.executor.execute({
                "tool": action,
                "params": params
            })
            
            # 5. 观察：记录执行结果
            observation = self.format_observation(execution_result)
            context += f"Observation: {observation}\n"
            
            # 6. 更新记忆
            self.memory.working.update_step(
                step_index=iteration,
                result=observation,
                status="completed" if execution_result.status == ExecutionStatus.SUCCESS else "failed"
            )
        
        return "超过最大迭代次数，任务未完成"
    
    def is_task_complete(self, thought: str) -> bool:
        """检查任务是否完成"""
        return "Answer:" in thought or "Final:" in thought
    
    def extract_answer(self, thought: str) -> str:
        """提取最终答案"""
        if "Answer:" in thought:
            return thought.split("Answer:")[1].strip()
        return thought
    
    def format_observation(self, result: ExecutionResult) -> str:
        """格式化观察结果"""
        if result.status == ExecutionStatus.SUCCESS:
            return result.output[:500]  # 限制长度
        else:
            return f"Error: {result.error}"
    
    def generate_task_id(self) -> str:
        """生成任务 ID"""
        import uuid
        return str(uuid.uuid4())[:8]

# 简化的 Function Calling 流程
response = llm.chat(
    messages=[{"role": "user", "content": "北京天气"}],
    tools=[{"name": "get_weather", "parameters": {...}}]
)
 
if response.tool_calls:
    result = execute_tool(response.tool_calls[0])
    final_answer = llm.chat(
        messages=[..., {"role": "tool", "content": result}]
    )

Planning → Memory → Tools → Action → 反馈 → Planning → ...

# 存储流程
text → Embedding Model → Vector [0.1, 0.3, ...] → Vector DB
 
# 检索流程
query → Embedding Model → Query Vector
      → Vector DB.similarity_search(query_vector, top_k=5)
      → 返回语义相关的记忆

用户目标 → 规划器(思考) → 记忆库(检索) → 工具集(选择) → 执行器(执行) → 反馈 → 循环