什么是 AI Agent

┌─────────────────────────────────────────────────────────┐
│                      AI Agent 架构                       │
├─────────────────────────────────────────────────────────┤
│                                                         │
│    用户输入 ──→ ┌─────────────────────────┐             │
│                 │     LLM（大脑/决策层）    │             │
│                 │  ┌─────┬─────┬─────┬─────┐│             │
│                 │  │规划器│记忆库│工具集│执行器││             │
│                 │  └─────┴─────┴─────┴─────┘│             │
│                 └───────────┬─────────────┘             │
│                             ↓                           │
│                      执行行动/返回结果                    │
│                             ↓                           │
│                       环境反馈                           │
└─────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────┐
│                    传统软件 vs AI Agent                     │
├──────────────────────┬─────────────────────────────────────┤
│       传统软件        │            AI Agent                 │
├──────────────────────┼─────────────────────────────────────┤
│  预定义流程（If-Else）│  自主决策（LLM 推理）               │
│  功能导向            │  目标导向                           │
│  被动响应            │  主动执行                           │
│  固定规则            │  适应环境变化                       │
│  需要详细指令        │  理解意图自动执行                   │
│  无法处理未知场景    │  可处理未见过的情况                 │
└──────────────────────┴─────────────────────────────────────┘

                    ┌─────────────┐
                    │   用户目标   │
                    └──────┬──────┘
                           ↓
    ┌──────────────────────────────────────────────┐
    │                  规划器（Planning）            │
    │  ┌────────────────────────────────────────┐  │
    │  │ 1. 理解目标                              │  │
    │  │ 2. 分解任务（Task Decomposition）        │  │
    │  │ 3. 生成执行计划                          │  │
    │  └────────────────────────────────────────┘  │
    └──────────────────────┬───────────────────────┘
                           ↓
    ┌──────────────────────────────────────────────┐
    │                  记忆库（Memory）             │
    │  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
    │  │ 短期记忆  │  │ 长期记忆  │  │ 工作记忆  │   │
    │  │(上下文)   │  │(向量DB)   │  │(任务状态) │   │
    │  └──────────┘  └──────────┘  └──────────┘   │
    └──────────────────────┬───────────────────────┘
                           ↓
    ┌──────────────────────────────────────────────┐
    │                  工具集（Tools）              │
    │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐│
    │  │搜索工具│ │代码执行│ │数据库  │ │文件操作││
    │  └────────┘ └────────┘ └────────┘ └────────┘│
    └──────────────────────┬───────────────────────┘
                           ↓
    ┌──────────────────────────────────────────────┐
    │                  执行器（Action）             │
    │  ┌────────────────────────────────────────┐  │
    │  │ 1. 调用工具执行任务                      │  │
    │  │ 2. 获取执行结果                          │  │
    │  │ 3. 反馈给规划器进行迭代                  │  │
    │  └────────────────────────────────────────┘  │
    └──────────────────────┬───────────────────────┘
                           ↓
                    ┌─────────────┐
                    │   最终结果   │
                    └─────────────┘

# Agent 规划流程示例
class AgentPlanner:
    def plan(self, user_goal: str) -> list[Task]:
        """
        规划流程：
        1. 理解用户意图
        2. 分解为子任务
        3. 确定任务依赖关系
        4. 生成执行计划
        """
        # Step 1: 理解意图
        intent = self.llm.understand(user_goal)
        
        # Step 2: 分解任务
        subtasks = self.llm.decompose(intent)
        
        # Step 3: 规划顺序（考虑依赖）
        execution_plan = self.create_plan(subtasks)
        
        return execution_plan
 
# 示例：用户目标 "帮我分析上季度销售数据并生成报告"
# 分解结果：
# Task 1: 连接数据库，获取销售数据
# Task 2: 数据清洗和预处理
# Task 3: 统计分析和可视化
# Task 4: 生成报告文档

# 三层记忆架构
class AgentMemory:
    def __init__(self):
        self.short_term = ShortTermMemory()   # 当前会话上下文
        self.long_term = LongTermMemory()     # 向量数据库持久化
        self.working = WorkingMemory()        # 任务执行状态
    
    def remember(self, info: str, memory_type: str):
        """存储记忆"""
        if memory_type == "short":
            self.short_term.add(info)
        elif memory_type == "long":
            self.long_term.store(info)
        else:
            self.working.update(info)
    
    def recall(self, query: str) -> str:
        """检索记忆"""
        # 优先从工作记忆获取
        if self.working.has(query):
            return self.working.get(query)
        
        # 向量检索长期记忆
        return self.long_term.search(query)

# 工具定义示例
from typing import Literal
from pydantic import BaseModel
 
class SearchTool(BaseModel):
    """搜索工具"""
    name: str = "web_search"
    description: str = "搜索互联网获取实时信息"
    
    def execute(self, query: str) -> str:
        # 调用搜索 API
        return search_engine.search(query)
 
class CodeExecutor(BaseModel):
    """代码执行工具"""
    name: str = "code_execute"
    description: str = "执行 Python 代码"
    
    def execute(self, code: str) -> str:
        # 在沙箱环境中执行
        return sandbox.run(code)
 
# 工具注册
tools = [SearchTool(), CodeExecutor(), DatabaseTool(), ...]

class AgentExecutor:
    def execute(self, task: Task, tools: list) -> Result:
        """
        执行流程：
        1. 选择合适的工具
        2. 构造调用参数
        3. 执行并获取结果
        4. 错误处理和重试
        """
        # Step 1: 选择工具
        tool = self.select_tool(task, tools)
        
        # Step 2: 构造参数
        params = self.llm.generate_params(task, tool)
        
        # Step 3: 执行
        try:
            result = tool.execute(**params)
        except Exception as e:
            # Step 4: 错误处理
            result = self.handle_error(e, task)
        
        return result

┌─────────────────────────────────────────────────────────────┐
│                    Agent 工作流程（ReAct 模式）               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐  │
│  │ 用户输入 │───→│ 理解目标 │───→│ 规划任务 │───→│ 执行行动 │  │
│  └─────────┘    └─────────┘    └─────────┘    └────┬────┘  │
│                                                     │       │
│                    ┌────────────────────────────────┘       │
│                    ↓                                        │
│  ┌──────────────────────────────────────────────────────┐  │
│  │                     观察与反思                        │  │
│  │  ┌────────────┐  ┌────────────┐  ┌────────────┐      │  │
│  │  │ 执行成功？  │  │ 需要调整？  │  │ 任务完成？  │      │  │
│  │  └─────┬──────┘  └─────┬──────┘  └─────┬──────┘      │  │
│  │        │               │               │              │  │
│  │        ↓               ↓               ↓              │  │
│  │  ┌──────────┐   ┌──────────┐   ┌──────────┐          │  │
│  │  │ 继续执行  │   │ 调整策略  │   │ 返回结果  │          │  │
│  │  └──────────┘   └──────────┘   └──────────┘          │  │
│  └──────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# ReAct 循环示例
def react_loop(agent, user_goal):
    """
    ReAct 循环：
    Thought → Action → Observation → Thought → ...
    """
    context = user_goal
    
    while not agent.is_finished():
        # Thought: 思考下一步
        thought = agent.think(context)
        print(f"Thought: {thought}")
        
        # Action: 决定执行什么动作
        action, action_input = agent.decide_action(thought)
        print(f"Action: {action}({action_input})")
        
        # Observation: 执行并观察结果
        observation = agent.execute(action, action_input)
        print(f"Observation: {observation}")
        
        # 更新上下文
        context += f"\nThought: {thought}\nAction: {action}\nObservation: {observation}"
    
    return agent.get_final_answer()

用户: 帮我查一下北京明天的天气，并给出穿衣建议

Thought: 我需要先搜索北京明天的天气信息
Action: web_search("北京明天天气")
Observation: 北京明天晴，气温 15-25°C，风力 3 级

Thought: 已获取天气信息，现在需要给出穿衣建议
Action: generate_advice("晴天，15-25°C，3级风")
Observation: 建议穿着轻薄外套，早晚温差较大，注意保暖

Thought: 我已经完成了用户的请求
Final Answer: 北京明天天气晴朗，气温 15-25°C，风力 3 级。
              建议穿着轻薄外套，早晚温差较大，注意保暖。

┌─────────────────────────────────────────────────────────────┐
│                    Agent 应用领域分类                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │   代码助手    │  │   研究助手    │  │   智能客服    │      │
│  │  (Devin等)   │  │ (Perplexity) │  │  (智能对话)   │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
│                                                             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │  数据分析     │  │  自动化运维   │  │  个人助理     │      │
│  │  (报表生成)   │  │ (监控告警)   │  │ (日程管理)   │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│                    Agent 发展演进                             │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│  AutoGPT          BabyAGI          MetaGPT         LangChain │
│  (自主执行)       (任务驱动)       (多Agent协作)    (框架化)   │
│                                                              │
│  ┌────────┐      ┌────────┐       ┌────────┐      ┌────────┐│
│  │ 单Agent │ ──→ │ 任务队列 │ ──→ │ 多Agent │ ──→ │ 工业级  ││
│  │ 自主执行 │     │ 循环执行 │      │ 角色协作 │     │ 框架生态 ││
│  └────────┘      └────────┘       └────────┘      └────────┘│
│                                                              │
│  2022.10          2023.04           2023.06         2023.10  │
│                                                              │
└──────────────────────────────────────────────────────────────┘

# 记忆存储流程
def store_memory(text: str):
    # 1. 文本向量化
    embedding = embedding_model.encode(text)
    
    # 2. 存入向量数据库
    vector_db.insert(
        id=generate_id(),
        vector=embedding,
        metadata={"text": text, "timestamp": now()}
    )
 
# 记忆检索流程
def recall_memory(query: str, top_k=5):
    # 1. 查询向量化
    query_embedding = embedding_model.encode(query)
    
    # 2. 向量相似度搜索
    results = vector_db.search(query_embedding, top_k=top_k)
    
    return results

# 1. 工具定义
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    }
]
 
# 2. LLM 决定调用哪个工具
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "北京天气怎么样"}],
    tools=tools
)
 
# 3. LLM 返回工具调用请求
# response.tool_calls[0].function.name = "get_weather"
# response.tool_calls[0].function.arguments = {"city": "北京"}
 
# 4. 执行工具并返回结果
result = get_weather("北京")

def handle_failure(agent, task, error):
    # 1. 分析失败原因
    reflection = agent.reflect(f"任务执行失败: {error}")
    
    # 2. 生成替代方案
    alternative = agent.plan_alternative(reflection)
    
    # 3. 重试或调整目标
    if alternative.feasible:
        return agent.execute(alternative)
    else:
        return "无法完成任务，原因：..."