AI Agent 完全指南：从理论到工业实践
Chen Kai BOSS
  2025-08-15 10:45:00 2025-08-15 10:45      18.4k 字  86 分钟  

两年前， ChatGPT 横空出世让我们见识到大语言模型的强大对话能力。但仅仅"对话"还不够——当你问 ChatGPT"帮我订一张明天去上海的机票"时，它只能给你建议，却无法真正完成任务。这就是普通 LLM 的局限：缺乏与外部世界交互的能力。

AI Agent 的出现改变了这一切。 Agent 不仅能理解你的需求，还能自主规划步骤、调用工具、执行操作、从错误中学习，最终完成复杂任务。从 AutoGPT 到 GPT-4 的 Function Calling，从单个智能体到多 Agent 协作系统， Agent 技术正在快速演进，逐步从研究原型走向工业应用。

本文将系统梳理 AI Agent 的核心概念、关键技术、主流框架和实战案例，帮助你构建属于自己的智能代理系统。

Agent 是什么

从"被动回答"到"主动执行"

传统的大语言模型就像一位博学的顾问：你问什么，它答什么。但 Agent 更像一位能干的助手：你提出目标，它会自己拆解任务、查找资料、调用工具、验证结果，直到完成目标。

想象你要组织一场技术分享会。普通 LLM 的交互可能是这样的：

1
2
3
4
5
6
你：帮我策划一场 AI 主题的技术分享会
LLM：可以从以下几个方面准备：
     1. 确定主题和时间
     2. 预订会议室
     3. 邀请讲师
     ...（给出建议后就结束了）

而 Agent 的工作流程则完全不同：

1
2
3
4
5
6
7
8
你：帮我组织一场 AI 主题的技术分享会
Agent：
  [规划] 拆解任务：调研热点话题、预订场地、联系讲师、发送邀请
  [执行] 搜索最近三个月 AI 领域的热门论文...
  [工具] 调用日历 API 查询可用时间...
  [工具] 调用会议室预订系统...
  [反思] 发现讲师时间冲突，重新调整...
  [完成] 已完成预订，邀请邮件已发送

这种"理解目标→自主规划→使用工具→执行反馈"的循环，就是 Agent 的核心特征。

Agent 的正式定义

在学术界， Agent 通常被定义为：

Agent 是一个能够感知环境、自主决策并采取行动以达成目标的计算实体。

这个定义包含三个关键要素：

  1. 感知（ Perception）： Agent 能够接收来自环境的信息，包括用户输入、工具返回结果、外部数据源等
  2. 决策（ Planning & Reasoning）： Agent 基于感知到的信息，结合自身知识进行推理和规划
  3. 行动（ Action）： Agent 执行具体操作，改变环境状态，并观察行动的结果

用一个简单的公式表示：

\[\text{Agent} = \text{LLM} + \text{Planning} + \text{Memory} + \text{Tool Use} \] 其中： - LLM（ Large Language Model）提供核心的理解和生成能力 - Planning 负责任务分解和执行计划制定 - Memory 存储上下文、历史记录和学到的经验 - Tool Use 让 Agent 能调用外部工具扩展能力边界

Agent vs 普通 LLM

让我们用一个对比表格来明确两者的区别：

维度 普通 LLM AI Agent
交互模式 单轮问答或多轮对话 目标驱动的任务执行
能力范围 仅限于生成文本 可调用工具、访问外部资源
规划能力 无，只能根据 prompt 回答 有，能自主拆解复杂任务
记忆机制 仅限于当前对话上下文 长期记忆、经验积累
自主性 被动响应 主动执行、迭代优化
错误处理 无法自我纠正 可以反思并重试

举个具体例子。假设任务是"分析我的 GitHub 仓库的代码质量"：

普通 LLM： 

1
2
3
4
5
用户：分析我的 GitHub 仓库 username/repo 的代码质量
LLM：我无法直接访问你的 GitHub 仓库。可以：
     1. 把代码粘贴给我
     2. 使用代码质量分析工具如 SonarQube
     ...

AI Agent： 

1
2
3
4
5
6
7
8
9
用户：分析我的 GitHub 仓库 username/repo 的代码质量
Agent：
  [工具调用] 使用 GitHub API 克隆仓库...
  [工具调用] 使用 pylint 扫描 Python 文件...
  [分析] 发现 23 个潜在问题：
    - 10 个未使用的导入
    - 5 个命名不规范的变量
    - 8 个过长的函数（>50 行）
  [生成报告] 已生成详细报告并保存到 report.md

这就是 Agent 的强大之处——它不仅理解你的需求，还能真正执行。

Agent 的发展历程

早期探索（ 2020-2022）

AI Agent 的概念并非全新，早在强化学习领域就有 agent 的研究。但真正让 Agent 走向大众视野的是大语言模型的突破。

ReAct（ 2022.10） 是早期的里程碑。研究者发现，让 LLM 交替进行"推理"（ Reasoning）和"行动"（ Acting），能显著提升问题解决能力。 ReAct 的核心 prompt 模板非常简单：

1
2
3
4
5
6
7
8
Question: [用户的问题]
Thought: [我需要做什么]
Action: [调用工具]
Observation: [工具返回结果]
Thought: [分析结果]
Action: [下一步操作]
...
Final Answer: [最终答案]

这个简单的循环奠定了现代 Agent 的基本框架。

爆发期（ 2023）

2023 年是 Agent 技术井喷的一年：

AutoGPT（ 2023.03） 引爆了全民关注。它让 GPT-4 自己给自己分配任务，自己调用工具，自己迭代优化。虽然实用性有限，但证明了 Agent 的可行性。

GPT-4 Function Calling（ 2023.06） 是工程实践的重要突破。 OpenAI 官方支持了结构化的工具调用，让 Agent 开发变得标准化。

LangChain/LangGraph（ 2023） 等框架迅速成熟，提供了开箱即用的 Agent 组件。

Multi-Agent 系统 开始出现：让多个 Agent 协作完成复杂任务，比如 MetaGPT 、 ChatDev 等。

工业化阶段（ 2024-现在）

进入 2024 年， Agent 技术开始从研究走向生产：

  • Claude Computer Use（ 2024.10）： Anthropic 推出的 Computer Use 功能让 Agent 能直接操作电脑界面，标志着通用智能代理的新阶段

  • 企业级 Agent 平台 涌现：各大云厂商推出 Agent 服务，如 AWS Bedrock Agents 、 Azure AI Agent Service

  • 垂直领域 Agent 成熟：代码助手（ GitHub Copilot）、客服机器人、数据分析 Agent 等开始规模化应用

  • 评估体系 完善： AgentBench 、 GAIA 、 AgentBoard 等基准测试建立，为 Agent 能力提供量化标准

现在， Agent 技术已经进入"如何做好"而非"能否实现"的阶段。接下来，让我们深入 Agent 的核心能力。

核心能力一：规划（ Planning）

规划是 Agent 最关键的能力——它决定了 Agent 能否将复杂目标分解为可执行的步骤。

为什么需要规划

人类解决复杂问题时会自然地进行任务分解。比如"做一顿晚餐"这个目标，你会拆分成：

  1. 确定菜单
  2. 列购物清单
  3. 去超市采购
  4. 洗菜切菜
  5. 烹饪
  6. 摆盘上桌

每个子任务还能继续细化。这种层次化的规划能力，正是 Agent 需要具备的。

对于 LLM 来说，直接生成完整的执行步骤往往不可靠——可能遗漏关键步骤，或者顺序混乱。因此， Agent 需要专门的规划机制。

规划的基本方法

单路径规划（ Single-Path）

最简单的规划方式是线性分解：把任务拆成顺序执行的步骤。

Chain of Thought (CoT) 就是这种思路。通过在 prompt 中加入"让我们一步步思考"，引导 LLM 逐步推理：

1
2
3
4
5
6
7
8
prompt = """
任务：计算 (15 + 27) * 3 - 8

让我们一步步思考：
第一步：计算括号内的加法
第二步：乘以 3
第三步：减去 8
"""

对于 Agent 任务，ReAct 是经典的单路径规划框架：

1
2
3
4
5
6
7
8
9
10
class ReActAgent:
    def run(self, task):
        thought = self.think(task)  # 推理
        action = self.plan_action(thought)  # 规划动作
        observation = self.execute(action)  # 执行
        
        if self.is_finished(observation):
            return self.generate_answer(observation)
        else:
            return self.run(self.update_task(observation))  # 递归

单路径规划的优点是简单直接，缺点是遇到错误时难以回溯。

多路径规划（ Multi-Path）

更高级的规划方法会同时探索多个可能的路径。

Tree of Thoughts (ToT) 将推理过程建模为搜索树。每个节点是一个"思考"， Agent 可以： - 生成多个候选思考 - 评估每个思考的质量 - 选择最佳路径继续探索 - 必要时回溯到父节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class ToTAgent:
    def solve(self, problem):
        root = Node(problem)
        frontier = [root]
        
        while frontier:
            node = self.select_best(frontier)
            
            if self.is_solution(node):
                return node.path
            
            # 生成多个子思考
            children = self.generate_thoughts(node, num=5)
            
            # 评估每个子思考
            for child in children:
                child.score = self.evaluate(child)
            
            frontier.extend(children)
            frontier.sort(key=lambda n: n.score, reverse=True)
        
        return None

ToT 特别适合需要试错的任务，比如数学难题、游戏策略等。

层次化任务分解（ HTN）

Hierarchical Task Network (HTN) 是更结构化的规划方法。它将任务分解为多个层次：

1
2
3
4
5
6
7
8
9
10
11
12
13
目标：组织技术大会
├─ 子任务 1：确定主题和议程
│  ├─ 调研行业热点
│  ├─ 设计分会场主题
│  └─ 安排时间表
├─ 子任务 2：联系讲师
│  ├─ 列出候选人名单
│  ├─ 发送邀请邮件
│  └─ 确认参与意愿
└─ 子任务 3：场地和后勤
   ├─ 预订会议中心
   ├─ 准备物料
   └─ 安排餐饮

HTN 规划的代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class HTNPlanner:
    def __init__(self):
        self.task_library = {
            "组织技术大会": ["确定主题", "联系讲师", "安排场地"],
            "确定主题": ["调研热点", "设计议程", "制定时间表"],
            # ... 更多任务分解规则
        }
    
    def decompose(self, task):
        """递归分解任务"""
        if task in self.task_library:
            subtasks = self.task_library[task]
            return [self.decompose(st) for st in subtasks]
        else:
            return task  # 原子任务，不再分解
    
    def plan(self, goal):
        """生成执行计划"""
        task_tree = self.decompose(goal)
        return self.flatten(task_tree)  # 展开为执行序列

HTN 的优势在于可复用的任务模板，缺点是需要预先定义分解规则。

动态规划与重规划

实际执行中，环境可能变化，计划需要调整。自适应规划机制至关重要。

AdaPlanner 是一个典型的自适应规划框架：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class AdaPlanner:
    def execute_with_replanning(self, plan):
        for step in plan:
            result = self.execute_step(step)
            
            if result.success:
                continue  # 继续执行
            else:
                # 执行失败，触发重规划
                failed_context = {
                    "step": step,
                    "error": result.error,
                    "previous_plan": plan
                }
                
                new_plan = self.replan(failed_context)
                return self.execute_with_replanning(new_plan)

重规划的提示词设计：

1
2
3
4
5
6
7
8
9
10
replan_prompt = f"""
原计划：{original_plan}
当前步骤：{current_step}
失败原因：{error_message}

请分析失败原因并生成新的计划。注意：
1. 保留已成功的步骤
2. 修正导致失败的策略
3. 考虑替代方案
"""

实战案例：旅行规划 Agent

让我们实现一个完整的旅行规划 Agent：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from langchain.chat_models import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import tool

# 定义工具
@tool
def search_flights(origin: str, destination: str, date: str) -> str:
    """搜索航班信息"""
    # 实际应调用航班 API
    return f"找到 3 个航班： CA123 (¥ 1200), MU456 (¥ 980), CZ789 (¥ 1100)"

@tool
def search_hotels(city: str, checkin: str, checkout: str) -> str:
    """搜索酒店"""
    return f"{city}有 5 家酒店可选，价格区间¥ 300-800/晚"

@tool
def search_attractions(city: str) -> str:
    """搜索景点"""
    return f"{city}热门景点：外滩、豫园、东方明珠、迪士尼"

# 创建规划 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search_flights, search_hotels, search_attractions]

agent_prompt = """你是一个专业的旅行规划助手。

用户目标：{input}

请按以下步骤规划：
1. 理解用户需求（目的地、时间、预算）
2. 搜索交通方式
3. 推荐住宿
4. 安排游玩行程
5. 生成完整计划

可用工具：
- search_flights: 查询航班
- search_hotels: 查询酒店
- search_attractions: 查询景点

开始规划：
{agent_scratchpad}
"""

agent = create_openai_functions_agent(llm, tools, agent_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行
result = executor.invoke({
    "input": "帮我规划 3 天的上海之旅，预算 5000 元，喜欢人文历史"
})

print(result["output"])

运行示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
> Entering AgentExecutor chain...

Thought: 用户要去上海 3 天，预算 5000 元，喜欢人文历史。
        我需要先搜索航班和酒店。

Action: search_flights
Action Input: {"origin": "北京", "destination": "上海", "date": "2024-04-01"}
Observation: 找到 3 个航班： CA123 (¥ 1200), MU456 (¥ 980), CZ789 (¥ 1100)

Thought: MU456 最便宜，选它。现在搜索酒店。

Action: search_hotels
Action Input: {"city": "上海", "checkin": "2024-04-01", "checkout": "2024-04-04"}
Observation: 上海有 5 家酒店可选，价格区间¥ 300-800/晚

Thought: 3 晚住宿预算¥ 1200 左右。查询景点。

Action: search_attractions
Action Input: {"city": "上海"}
Observation: 上海热门景点：外滩、豫园、东方明珠、迪士尼

Thought: 用户喜欢人文历史，迪士尼不太合适。规划完成。

Final Answer:
【上海 3 日游计划】

交通：
- 往返航班 MU456，¥ 980 × 2 = ¥ 1960

住宿：
- 推荐黄浦区酒店（靠近外滩），¥ 400/晚× 3 = ¥ 1200

行程：
第一天：外滩 → 南京路 → 人民广场
第二天：豫园 → 城隍庙 → 田子坊
第三天：上海博物馆 → 新天地 → 返程

总预算：¥ 3160（交通+住宿），剩余¥ 1840 用于餐饮和门票

这个例子展示了完整的规划流程：理解需求→调用工具→组合信息→生成计划。

核心能力二：记忆（ Memory）

如果说规划是 Agent 的"大脑"，那么记忆就是 Agent 的"知识库"。没有记忆， Agent 就像患了失忆症，无法从经验中学习，也无法处理长期任务。

为什么 Agent 需要记忆

LLM 的上下文窗口是有限的。即使是 GPT-4 的 128k tokens，也只能容纳约 10 万字的对话。对于需要持续运行数天甚至数周的 Agent 任务，这远远不够。

更重要的是， Agent 需要记住： - 任务上下文：之前做了什么，还剩什么没做 - 工具使用经验：哪些工具调用成功了，哪些失败了 - 领域知识：积累的专业信息和最佳实践 - 用户偏好：历史交互中暴露的用户习惯

短期记忆 vs 长期记忆

人类的记忆分为短期记忆（工作记忆）和长期记忆。 Agent 也采用类似架构：

短期记忆（ Short-Term Memory）： - 存储当前对话和最近的执行历史 - 直接放在 LLM 的上下文窗口中 - 容量有限，需要不断更新

长期记忆（ Long-Term Memory）： - 存储重要的历史信息和学到的知识 - 通常保存在外部数据库中 - 按需检索加载到短期记忆

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class AgentMemory:
    def __init__(self):
        # 短期记忆：当前会话的消息列表
        self.short_term = []
        
        # 长期记忆：向量数据库
        from langchain.vectorstores import Chroma
        from langchain.embeddings import OpenAIEmbeddings
        
        self.long_term = Chroma(
            embedding_function=OpenAIEmbeddings(),
            persist_directory="./agent_memory"
        )
    
    def add_message(self, role, content):
        """添加到短期记忆"""
        self.short_term.append({"role": role, "content": content})
        
        # 超过限制时，转移到长期记忆
        if len(self.short_term) > 20:
            old_message = self.short_term.pop(0)
            self.long_term.add_texts([old_message["content"]])
    
    def recall(self, query, k=5):
        """从长期记忆中检索相关内容"""
        docs = self.long_term.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]
    
    def get_context(self, current_query):
        """组合短期和长期记忆"""
        # 检索相关的长期记忆
        relevant_history = self.recall(current_query)
        
        # 拼接成完整上下文
        context = relevant_history + self.short_term
        return context

向量数据库：记忆的基础设施

长期记忆通常基于向量数据库实现。基本流程：

  1. 存储：将文本转为向量（ embedding），存入数据库
  2. 检索：将查询转为向量，找到最相似的记忆
  3. 更新：定期清理过时信息，合并重复记忆
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

class VectorMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None
    
    def add(self, text, metadata=None):
        """添加记忆"""
        if self.vectorstore is None:
            self.vectorstore = FAISS.from_texts(
                [text], 
                self.embeddings,
                metadatas=[metadata or {}]
            )
        else:
            self.vectorstore.add_texts([text], [metadata or {}])
    
    def search(self, query, k=5, filter=None):
        """搜索记忆"""
        if self.vectorstore is None:
            return []
        
        docs = self.vectorstore.similarity_search_with_score(
            query, k=k, filter=filter
        )
        
        return [(doc.page_content, score) for doc, score in docs]
    
    def save(self, path):
        """持久化"""
        if self.vectorstore:
            self.vectorstore.save_local(path)
    
    def load(self, path):
        """加载"""
        self.vectorstore = FAISS.load_local(path, self.embeddings)

常用的向量数据库包括： - FAISS： Meta 开源，适合原型开发 - Pinecone：托管服务，易于扩展 - Milvus：开源，适合大规模部署 - Chroma：轻量级，适合中小项目

结构化记忆：超越简单文本

纯文本记忆不够精确。很多 Agent 采用结构化记忆：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class StructuredMemory:
    def __init__(self):
        import sqlite3
        self.conn = sqlite3.connect("agent_memory.db")
        self.cursor = self.conn.cursor()
        
        # 创建表
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS interactions (
                id INTEGER PRIMARY KEY,
                timestamp TEXT,
                task TEXT,
                action TEXT,
                result TEXT,
                success BOOLEAN
            )
        """)
        
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS knowledge (
                id INTEGER PRIMARY KEY,
                entity TEXT,
                attribute TEXT,
                value TEXT,
                source TEXT
            )
        """)
    
    def log_interaction(self, task, action, result, success):
        """记录交互历史"""
        from datetime import datetime
        self.cursor.execute("""
            INSERT INTO interactions 
            (timestamp, task, action, result, success)
            VALUES (?, ?, ?, ?, ?)
        """, (datetime.now().isoformat(), task, action, result, success))
        self.conn.commit()
    
    def store_knowledge(self, entity, attribute, value, source):
        """存储知识三元组"""
        self.cursor.execute("""
            INSERT OR REPLACE INTO knowledge
            (entity, attribute, value, source)
            VALUES (?, ?, ?, ?)
        """, (entity, attribute, value, source))
        self.conn.commit()
    
    def query_knowledge(self, entity=None, attribute=None):
        """查询知识"""
        query = "SELECT * FROM knowledge WHERE 1=1"
        params = []
        
        if entity:
            query += " AND entity = ?"
            params.append(entity)
        if attribute:
            query += " AND attribute = ?"
            params.append(attribute)
        
        self.cursor.execute(query, params)
        return self.cursor.fetchall()

这种结构化记忆适合存储： - 实体关系（如"张三是产品经理"） - 任务状态（如"已完成数据清洗"） - 工具调用记录（如"search_api 在周末超时"）

MARS：多步推理的记忆增强

MARS (Memory-Augmented Reasoning with Search) 是一个专门为 Agent 设计的记忆系统。核心思想：

  1. 在每个推理步骤后，将重要信息存入记忆
  2. 下一步推理前，先检索相关记忆
  3. 将检索结果和当前问题一起送入 LLM
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class MARSAgent:
    def __init__(self, llm, memory):
        self.llm = llm
        self.memory = memory
    
    def solve(self, problem):
        steps = []
        current_state = problem
        
        while not self.is_solved(current_state):
            # 检索相关记忆
            relevant_memories = self.memory.search(
                current_state, k=3
            )
            
            # 组合成 prompt
            prompt = f"""
            问题：{problem}
            当前状态：{current_state}
            
            相关经验：
            {self.format_memories(relevant_memories)}
            
            下一步该做什么？
            """
            
            # LLM 推理
            response = self.llm(prompt)
            action = self.parse_action(response)
            
            # 执行动作
            result = self.execute(action)
            steps.append((action, result))
            
            # 存入记忆
            self.memory.add(
                f"问题：{current_state}\n 动作：{action}\n 结果：{result}"
            )
            
            current_state = self.update_state(current_state, result)
        
        return steps

MARS 特别适合需要多步推理的任务，如数学证明、代码调试等。

Memento：基于事件的记忆管理

Memento 系统将 Agent 的记忆组织为事件流：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional

@dataclass
class MemoryEvent:
    timestamp: datetime
    event_type: str  # "observation", "action", "reflection"
    content: str
    importance: float  # 0-1
    tags: List[str]
    linked_events: List[int]  # 关联事件的 ID

class MementoMemory:
    def __init__(self):
        self.events: List[MemoryEvent] = []
        self.index = {}  # tag -> event_ids
    
    def add_event(self, event: MemoryEvent):
        """添加事件"""
        event_id = len(self.events)
        self.events.append(event)
        
        # 建立索引
        for tag in event.tags:
            if tag not in self.index:
                self.index[tag] = []
            self.index[tag].append(event_id)
    
    def retrieve_by_time(self, start, end):
        """按时间检索"""
        return [e for e in self.events 
                if start <= e.timestamp <= end]
    
    def retrieve_by_tag(self, tag):
        """按标签检索"""
        if tag not in self.index:
            return []
        return [self.events[i] for i in self.index[tag]]
    
    def retrieve_by_importance(self, threshold=0.7):
        """检索重要事件"""
        return [e for e in self.events 
                if e.importance >= threshold]
    
    def get_event_chain(self, event_id):
        """获取事件链"""
        chain = []
        current = self.events[event_id]
        chain.append(current)
        
        for linked_id in current.linked_events:
            chain.extend(self.get_event_chain(linked_id))
        
        return chain

使用示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
memory = MementoMemory()

# 添加观察事件
memory.add_event(MemoryEvent(
    timestamp=datetime.now(),
    event_type="observation",
    content="用户询问如何优化数据库查询",
    importance=0.8,
    tags=["数据库", "查询优化", "用户问题"],
    linked_events=[]
))

# 添加行动事件
memory.add_event(MemoryEvent(
    timestamp=datetime.now(),
    event_type="action",
    content="查询数据库 schema",
    importance=0.6,
    tags=["数据库", "工具调用"],
    linked_events=[0]  # 关联到第一个事件
))

# 检索
recent_db_events = memory.retrieve_by_tag("数据库")

记忆压缩与遗忘

记忆不能无限增长。 Agent 需要"遗忘"机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class AdaptiveMemory:
    def __init__(self, max_size=1000):
        self.memories = []
        self.max_size = max_size
    
    def add(self, content, importance=0.5):
        self.memories.append({
            "content": content,
            "importance": importance,
            "access_count": 0,
            "last_access": datetime.now()
        })
        
        # 超过容量时清理
        if len(self.memories) > self.max_size:
            self.cleanup()
    
    def cleanup(self):
        """清理低价值记忆"""
        # 计算每条记忆的保留分数
        for mem in self.memories:
            recency = (datetime.now() - mem["last_access"]).days
            mem["score"] = (
                mem["importance"] * 0.5 +
                mem["access_count"] * 0.3 -
                recency * 0.2
            )
        
        # 排序并保留前 N 条
        self.memories.sort(key=lambda m: m["score"], reverse=True)
        self.memories = self.memories[:self.max_size]
    
    def retrieve(self, query, k=5):
        # 使用向量搜索（省略细节）
        results = self.vector_search(query, k)
        
        # 更新访问统计
        for mem in results:
            mem["access_count"] += 1
            mem["last_access"] = datetime.now()
        
        return results

这种基于重要性、访问频率和时间衰减的遗忘机制，让 Agent 能够聚焦于最相关的记忆。

核心能力三：工具使用（ Tool Use）

工具使用是 Agent 突破 LLM 固有限制的关键。通过工具， Agent 能够： - 访问实时信息（搜索引擎、数据库） - 执行精确计算（计算器、代码解释器） - 操作外部系统（ API 调用、文件操作） - 感知多模态信息（图像识别、语音转文字）

Function Calling：标准化的工具接口

Function Calling 是现代 Agent 工具使用的基础。以 OpenAI 的实现为例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import openai
import json

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# Agent 调用
def run_agent(user_message):
    messages = [{"role": "user", "content": user_message}]
    
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # 让模型决定是否调用工具
    )
    
    # 检查是否需要调用工具
    message = response.choices[0].message
    
    if message.tool_calls:
        # 执行工具调用
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            # 调用实际函数
            if function_name == "get_weather":
                result = get_weather(**arguments)
            
            # 将结果返回给模型
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
        
        # 获取最终回复
        final_response = openai.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        return final_response.choices[0].message.content
    else:
        return message.content

# 实际工具实现
def get_weather(city, unit="celsius"):
    # 这里应调用真实天气 API
    return json.dumps({
        "city": city,
        "temperature": 22,
        "condition": "晴",
        "unit": unit
    })

# 测试
result = run_agent("北京今天天气怎么样？")
print(result)  # 输出：北京今天晴，温度 22 摄氏度

Function Calling 的优势： - 结构化输出：模型生成 JSON 格式的参数，易于解析 - 类型安全：参数有明确的类型定义 - 自动选择：模型决定何时调用哪个工具 - 链式调用：可以连续调用多个工具

Claude 的工具使用

Anthropic 的 Claude 也支持工具使用，语法略有不同：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "search_database",
        "description": "在数据库中搜索信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                }
            },
            "required": ["query"]
        }
    }
]

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "查询用户 ID 12345 的订单记录"}
    ]
)

# 处理工具调用
if message.stop_reason == "tool_use":
    tool_use = next(
        block for block in message.content 
        if block.type == "tool_use"
    )
    
    # 执行工具
    result = search_database(tool_use.input["query"])
    
    # 继续对话
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "查询用户 ID 12345 的订单记录"},
            {"role": "assistant", "content": message.content},
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": result
                    }
                ]
            }
        ]
    )

Computer Use：突破界面限制

2024 年 10 月， Anthropic 推出的 Computer Use 功能是工具使用的重大突破。 Agent 不再局限于预定义的 API，而是可以直接操作电脑：

  • 移动鼠标和点击
  • 键盘输入
  • 截图并理解界面
  • 跨应用操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from anthropic import Anthropic

client = Anthropic()

def computer_use_agent(task):
    messages = [
        {"role": "user", "content": task}
    ]
    
    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            tools=[
                {
                    "type": "computer_20241022",
                    "name": "computer",
                    "display_width_px": 1920,
                    "display_height_px": 1080
                }
            ],
            messages=messages
        )
        
        # 检查是否有操作
        if response.stop_reason == "tool_use":
            for block in response.content:
                if block.type == "tool_use":
                    action = block.input
                    
                    # 执行操作（鼠标、键盘、截图）
                    result = execute_computer_action(action)
                    
                    # 更新对话
                    messages.append({"role": "assistant", "content": response.content})
                    messages.append({
                        "role": "user",
                        "content": [{
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        }]
                    })
        else:
            # 任务完成
            return response.content

# 示例任务
computer_use_agent("打开浏览器，搜索'AI Agent 教程'，并总结前三个结果")

Computer Use 开启了通用自动化的大门，但也带来安全性挑战——需要严格的权限控制和沙箱隔离。

工具生态：从基础到高级

基础工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 搜索工具
from langchain.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()
result = search.run("AI Agent 最新进展")

# 计算器
from langchain.tools import PythonREPLTool

calculator = PythonREPLTool()
result = calculator.run("import math; math.sqrt(12345)")

# 文件操作
from langchain.tools import FileManagementToolkit
from langchain.tools.file_management import (
    ReadFileTool,
    WriteFileTool,
    ListDirectoryTool
)

file_tools = FileManagementToolkit(
    root_dir="./workspace"
).get_tools()

数据库工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from langchain_community.utilities import SQLDatabase
from langchain.tools import QuerySQLDataBaseTool

# 连接数据库
db = SQLDatabase.from_uri("sqlite:///users.db")

# 创建查询工具
db_tool = QuerySQLDataBaseTool(db=db)

# 自然语言转 SQL
from langchain.chains import create_sql_query_chain
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI()
chain = create_sql_query_chain(llm, db)

query = chain.invoke({"question": "有多少活跃用户？"})
# 输出： SELECT COUNT(*) FROM users WHERE status = 'active'

result = db_tool.run(query)

API 集成工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langchain.tools import APIOperation
from langchain.utilities import RequestsWrapper

# 定义 API 工具
api_spec = """
openapi: 3.0.0
info:
  title: GitHub API
  version: 1.0.0
paths:
  /repos/{owner}/{repo}/issues:
    get:
      summary: List repository issues
      parameters:
        - name: owner
          in: path
          required: true
        - name: repo
          in: path
          required: true
"""

requests_wrapper = RequestsWrapper(headers={"Authorization": "token YOUR_TOKEN"})
github_tool = APIOperation.from_openapi_spec(api_spec, requests_wrapper)

# 使用
issues = github_tool.run({
    "owner": "langchain-ai",
    "repo": "langchain"
})

工具编排：组合多个工具

复杂任务需要多个工具协作。工具编排是关键能力：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from langchain.agents import AgentType, initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool

# 定义多个工具
def analyze_data(data_path):
    """分析数据文件"""
    import pandas as pd
    df = pd.read_csv(data_path)
    return df.describe().to_string()

def visualize_data(data_path, chart_type):
    """生成数据可视化"""
    import pandas as pd
    import matplotlib.pyplot as plt
    
    df = pd.read_csv(data_path)
    if chart_type == "bar":
        df.plot.bar()
    elif chart_type == "line":
        df.plot.line()
    
    plt.savefig("chart.png")
    return "图表已保存到 chart.png"

def send_report(recipient, content):
    """发送报告邮件"""
    # 实际应调用邮件 API
    return f"报告已发送给 {recipient}"

# 创建工具列表
tools = [
    Tool(name="AnalyzeData", func=analyze_data, 
         description="分析 CSV 文件，返回统计信息"),
    Tool(name="VisualizeData", func=visualize_data,
         description="生成数据图表，需要指定图表类型"),
    Tool(name="SendReport", func=send_report,
         description="发送报告邮件")
]

# 创建 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

# 执行复杂任务
agent.run("""
分析 sales_data.csv 文件，生成柱状图，
然后把分析结果和图表发送给 manager@company.com
""")

执行过程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
> Entering AgentChain...

Thought: 需要先分析数据，然后生成图表，最后发送报告

Action: AnalyzeData
Action Input: sales_data.csv
Observation: 
       销售额      数量
count   12.0    12.0
mean   8500.0   85.0
std    1200.0   15.0
...

Thought: 数据已分析，现在生成柱状图

Action: VisualizeData
Action Input: {"data_path": "sales_data.csv", "chart_type": "bar"}
Observation: 图表已保存到 chart.png

Thought: 准备发送报告

Action: SendReport
Action Input: {
  "recipient": "manager@company.com",
  "content": "销售数据分析完成，平均销售额 8500，详见附件图表"
}
Observation: 报告已发送给 manager@company.com

Final Answer: 任务已完成。已分析数据，生成图表，并发送报告。

动态工具发现

更高级的 Agent 能够自动发现和学习新工具：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, tool_spec):
        """注册新工具"""
        self.tools[tool_spec["name"]] = {
            "function": tool_spec["function"],
            "description": tool_spec["description"],
            "parameters": tool_spec["parameters"],
            "examples": tool_spec.get("examples", [])
        }
    
    def discover(self, task_description):
        """根据任务描述推荐工具"""
        from sklearn.metrics.pairwise import cosine_similarity
        from sentence_transformers import SentenceTransformer
        
        model = SentenceTransformer('all-MiniLM-L6-v2')
        task_embedding = model.encode([task_description])
        
        tool_descriptions = [
            self.tools[name]["description"] 
            for name in self.tools
        ]
        tool_embeddings = model.encode(tool_descriptions)
        
        similarities = cosine_similarity(task_embedding, tool_embeddings)[0]
        
        # 返回最相关的工具
        top_indices = similarities.argsort()[-3:][::-1]
        return [list(self.tools.keys())[i] for i in top_indices]

# 使用
registry = ToolRegistry()

registry.register({
    "name": "weather_api",
    "function": get_weather,
    "description": "获取城市天气信息",
    "parameters": {"city": "string"},
    "examples": ["北京天气如何", "上海明天会下雨吗"]
})

# 动态发现
recommended = registry.discover("我想知道深圳的气温")
# 返回：["weather_api", ...]

核心能力四：反思（ Reflection）

反思是 Agent 从错误中学习、持续改进的能力。没有反思， Agent 会重复同样的错误；有了反思， Agent 能够自我纠正、优化策略。

为什么需要反思

人类解决问题时会不断反思： - "这个方法不 work，换个思路试试" - "上次失败是因为参数设置不对" - "这个策略在类似场景下效果更好"

Agent 同样需要这种元认知能力。

Self-Refine：迭代优化

Self-Refine 是最简单的反思模式：生成→评估→改进。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class SelfRefineAgent:
    def __init__(self, llm):
        self.llm = llm
    
    def generate(self, task):
        """生成初始输出"""
        prompt = f"任务：{task}\n 请给出解决方案："
        return self.llm(prompt)
    
    def evaluate(self, task, output):
        """评估输出质量"""
        prompt = f"""
        任务：{task}
        当前方案：{output}
        
        请评估这个方案，指出存在的问题和改进方向。
        按以下格式回答：
        评分（ 1-10）：
        问题：
        改进建议：
        """
        return self.llm(prompt)
    
    def refine(self, task, output, feedback):
        """根据反馈改进"""
        prompt = f"""
        任务：{task}
        当前方案：{output}
        反馈意见：{feedback}
        
        请根据反馈改进方案：
        """
        return self.llm(prompt)
    
    def run(self, task, max_iterations=3):
        """迭代优化"""
        output = self.generate(task)
        
        for i in range(max_iterations):
            feedback = self.evaluate(task, output)
            
            # 解析评分
            score = self.parse_score(feedback)
            print(f"迭代 {i+1}，评分：{score}")
            
            if score >= 8:
                break  # 质量足够好
            
            output = self.refine(task, output, feedback)
        
        return output

应用示例：

1
2
3
4
5
6
7
8
agent = SelfRefineAgent(ChatOpenAI(model="gpt-4"))

result = agent.run("写一个 Python 函数，计算斐波那契数列")

# 输出过程：
# 迭代 1，评分： 5（初始代码效率低）
# 迭代 2，评分： 7（改用动态规划，但缺少注释）
# 迭代 3，评分： 9（添加详细注释和类型提示）

Reflexion：从失败中学习

Reflexion 框架更进一步，让 Agent 从失败案例中提取教训。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class ReflexionAgent:
    def __init__(self, llm):
        self.llm = llm
        self.memory = []  # 存储失败案例和反思
    
    def act(self, task):
        """执行任务"""
        # 检索相关的历史反思
        relevant_reflections = self.retrieve_reflections(task)
        
        prompt = f"""
        任务：{task}
        
        历史经验：
        {self.format_reflections(relevant_reflections)}
        
        请制定执行计划：
        """
        
        plan = self.llm(prompt)
        result = self.execute(plan)
        
        return result
    
    def reflect(self, task, trajectory, result):
        """反思执行过程"""
        prompt = f"""
        任务：{task}
        执行轨迹：{trajectory}
        最终结果：{result}
        
        请分析：
        1. 哪些步骤是正确的？
        2. 哪些步骤出了问题？
        3. 为什么会失败？
        4. 下次应该怎么做？
        
        请给出具体的反思和改进建议：
        """
        
        reflection = self.llm(prompt)
        
        # 存入记忆
        self.memory.append({
            "task": task,
            "reflection": reflection,
            "timestamp": datetime.now()
        })
        
        return reflection
    
    def run_with_retry(self, task, max_attempts=3):
        """带重试的执行"""
        for attempt in range(max_attempts):
            trajectory = []
            
            try:
                result = self.act(task)
                
                if self.is_successful(result):
                    return result
                else:
                    # 失败，进行反思
                    reflection = self.reflect(task, trajectory, result)
                    print(f"尝试 {attempt+1} 失败，反思：{reflection}")
            
            except Exception as e:
                reflection = self.reflect(task, trajectory, str(e))
                print(f"尝试 {attempt+1} 出错，反思：{reflection}")
        
        return None

RISE：结构化的错误分析

RISE (Reflective Iterative Self-Evaluation) 提供更结构化的反思框架：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class RISEAgent:
    def __init__(self, llm):
        self.llm = llm
    
    def execute_with_rise(self, task):
        # 阶段 1：初始执行
        output = self.initial_attempt(task)
        
        # 阶段 2：识别错误（ Recognize）
        errors = self.recognize_errors(task, output)
        
        if not errors:
            return output
        
        # 阶段 3：调查原因（ Investigate）
        root_causes = self.investigate_causes(errors)
        
        # 阶段 4：策略制定（ Strategize）
        strategy = self.formulate_strategy(root_causes)
        
        # 阶段 5：执行改进（ Execute）
        improved_output = self.execute_strategy(task, strategy)
        
        return improved_output
    
    def recognize_errors(self, task, output):
        """识别错误"""
        prompt = f"""
        任务：{task}
        输出：{output}
        
        请列出所有存在的错误或不足：
        1.
        2.
        3.
        ...
        """
        return self.llm(prompt)
    
    def investigate_causes(self, errors):
        """调查根本原因"""
        prompt = f"""
        发现的错误：
        {errors}
        
        请使用"5 个为什么"分析法，找出根本原因：
        
        错误 1：
        - 为什么？
        - 为什么？
        - ...
        - 根本原因：
        
        错误 2：
        ...
        """
        return self.llm(prompt)
    
    def formulate_strategy(self, root_causes):
        """制定改进策略"""
        prompt = f"""
        根本原因：
        {root_causes}
        
        请为每个根本原因制定具体的改进策略：
        
        原因 1 → 策略 1：
        原因 2 → 策略 2：
        ...
        """
        return self.llm(prompt)

基于奖励模型的反思

更高级的反思可以使用奖励模型来量化输出质量：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from transformers import AutoModelForSequenceClassification, AutoTokenizer

class RewardBasedReflection:
    def __init__(self, llm, reward_model_name):
        self.llm = llm
        self.reward_model = AutoModelForSequenceClassification.from_pretrained(
            reward_model_name
        )
        self.tokenizer = AutoTokenizer.from_pretrained(reward_model_name)
    
    def compute_reward(self, task, output):
        """计算输出的奖励分数"""
        input_text = f"任务：{task}\n 输出：{output}"
        inputs = self.tokenizer(input_text, return_tensors="pt")
        
        with torch.no_grad():
            reward = self.reward_model(**inputs).logits[0][0].item()
        
        return reward
    
    def optimize_with_reward(self, task, num_samples=5):
        """生成多个候选，选择奖励最高的"""
        candidates = []
        
        for _ in range(num_samples):
            output = self.llm(task)
            reward = self.compute_reward(task, output)
            candidates.append((output, reward))
        
        # 选择最佳输出
        best_output, best_reward = max(candidates, key=lambda x: x[1])
        
        # 反思：分析为什么这个输出更好
        reflection = self.analyze_best_output(task, candidates, best_output)
        
        return best_output, reflection
    
    def analyze_best_output(self, task, all_candidates, best):
        """分析最佳输出的特点"""
        others = [c[0] for c in all_candidates if c[0] != best]
        
        prompt = f"""
        任务：{task}
        
        最佳输出：
        {best}
        
        其他候选：
        {chr(10).join(others)}
        
        请分析最佳输出为什么更好，总结其特点：
        """
        
        return self.llm(prompt)

协作式反思

在多 Agent 系统中， Agent 之间可以互相反思：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class CollaborativeReflection:
    def __init__(self, agents):
        self.agents = agents
    
    def peer_review(self, task, outputs):
        """同行评审"""
        reviews = {}
        
        for agent_id, output in outputs.items():
            reviews[agent_id] = []
            
            # 其他 agent 评审这个输出
            for reviewer_id, reviewer in self.agents.items():
                if reviewer_id != agent_id:
                    review = reviewer.review(task, output)
                    reviews[agent_id].append({
                        "reviewer": reviewer_id,
                        "feedback": review
                    })
        
        return reviews
    
    def consensus_reflection(self, task, outputs, reviews):
        """基于评审达成共识"""
        # 汇总所有反馈
        all_feedback = []
        for agent_reviews in reviews.values():
            all_feedback.extend([r["feedback"] for r in agent_reviews])
        
        # 生成共识性的反思
        prompt = f"""
        任务：{task}
        
        各 Agent 的输出：
        {self.format_outputs(outputs)}
        
        收到的反馈：
        {chr(10).join(all_feedback)}
        
        请综合所有反馈，总结：
        1. 达成共识的优秀做法
        2. 存在争议的问题
        3. 改进建议
        """
        
        consensus = self.agents[0].llm(prompt)
        return consensus

反思能力让 Agent 不再是"一次性"的执行者，而是能够持续学习和改进的智能系统。

主流框架实现

理解了核心能力后，下面介绍如何用现有框架快速构建 Agent 。

LangChain：最流行的 Agent 框架

LangChain 是目前最广泛使用的 Agent 框架，提供了丰富的组件和预置模板。

基础 Agent 构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 定义工具
@tool
def search(query: str) -> str:
    """搜索互联网信息"""
    # 实际应调用搜索 API
    return f"搜索结果：{query} 的相关信息..."

@tool
def calculator(expression: str) -> str:
    """执行数学计算"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except:
        return "计算错误"

# 创建 prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来回答问题。"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# 初始化 LLM 和工具
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search, calculator]

# 创建 Agent
agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

# 执行
result = executor.invoke({
    "input": "搜索 2024 年诺贝尔物理学奖获得者，并计算他们的平均年龄"
})

带记忆的 Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor

# 创建记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建带记忆的 prompt
prompt_with_memory = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# 创建 Agent
agent = create_openai_functions_agent(llm, tools, prompt_with_memory)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 多轮对话
executor.invoke({"input": "我喜欢 Python 编程"})
executor.invoke({"input": "推荐一些学习资源"})  # Agent 会记住你喜欢 Python

LangGraph：流程编排的新范式

LangGraph 是 LangChain 团队推出的新框架，用图结构来定义 Agent 的执行流程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_action: str

# 定义节点
def call_model(state):
    """调用 LLM"""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

def call_tool(state):
    """执行工具"""
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]
    
    # 执行工具
    tool_output = tools[tool_call["name"]].invoke(tool_call["args"])
    
    return {"messages": [ToolMessage(
        content=tool_output,
        tool_call_id=tool_call["id"]
    )]}

def should_continue(state):
    """决定是否继续"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "continue"
    return "end"

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# 设置入口
workflow.set_entry_point("agent")

# 添加边
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# 编译
app = workflow.compile()

# 执行
result = app.invoke({
    "messages": [HumanMessage(content="北京今天天气如何？")]
})

LangGraph 的优势： - 可视化：流程图一目了然 - 灵活：支持复杂的条件分支和循环 - 调试友好：每个节点的状态都可追踪

AutoGPT：自主任务规划

AutoGPT 是最早引起轰动的自主 Agent 项目。其核心思想：让 GPT-4 自己给自己分配任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
class AutoGPTAgent:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.task_list = []
    
    def run(self, goal):
        """运行 AutoGPT 主循环"""
        self.task_list = [{"id": 1, "task": goal}]
        completed_tasks = []
        
        while self.task_list:
            # 获取当前任务
            current_task = self.task_list.pop(0)
            print(f"\n 执行任务：{current_task['task']}")
            
            # 生成新任务
            new_tasks = self.generate_new_tasks(
                current_task,
                completed_tasks,
                goal
            )
            
            # 执行当前任务
            result = self.execute_task(current_task['task'])
            completed_tasks.append({
                "task": current_task['task'],
                "result": result
            })
            
            # 优先级排序
            self.task_list = self.prioritize_tasks(
                self.task_list + new_tasks,
                goal
            )
            
            # 检查是否完成目标
            if self.check_goal_completion(goal, completed_tasks):
                break
        
        return completed_tasks
    
    def generate_new_tasks(self, current_task, completed_tasks, goal):
        """生成新任务"""
        prompt = f"""
        总目标：{goal}
        
        当前任务：{current_task['task']}
        
        已完成任务：
        {self.format_tasks(completed_tasks)}
        
        基于当前任务的结果，生成接下来需要完成的新任务列表。
        每个任务应该是独立可执行的。
        
        以 JSON 格式返回：
        [
            {{ "id": 1, "task": "任务描述" }},
            {{ "id": 2, "task": "任务描述" }}
        ]
        """
        
        response = self.llm(prompt)
        return json.loads(response)
    
    def execute_task(self, task):
        """执行单个任务"""
        # 检索相关记忆
        context = self.memory.search(task)
        
        prompt = f"""
        任务：{task}
        
        相关上下文：
        {context}
        
        可用工具：
        {self.format_tools()}
        
        请完成这个任务，返回执行结果。
        """
        
        result = self.llm_with_tools(prompt)
        
        # 存入记忆
        self.memory.add(f"任务：{task}\n 结果：{result}")
        
        return result
    
    def prioritize_tasks(self, tasks, goal):
        """任务优先级排序"""
        prompt = f"""
        总目标：{goal}
        
        待完成任务：
        {self.format_tasks(tasks)}
        
        请对这些任务按优先级排序，考虑：
        1. 哪些任务对实现目标最关键
        2. 哪些任务有依赖关系
        3. 哪些任务可以并行
        
        返回排序后的任务列表（ JSON 格式）：
        """
        
        response = self.llm(prompt)
        return json.loads(response)

使用示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
agent = AutoGPTAgent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=[search, calculator, file_ops],
    memory=VectorMemory()
)

result = agent.run("研究并总结 2024 年 AI 领域的重大突破")

# AutoGPT 会自主分解任务：
# 任务 1：搜索 2024 年 AI 领域的新闻
# 任务 2：筛选重大突破
# 任务 3：深入研究每个突破
# 任务 4：撰写总结报告
# 任务 5：保存到文件

AutoGPT 的问题： - 成本高：每个决策都要调用 LLM - 不稳定：容易陷入死循环或偏离目标 - 难控制：自主性太强，不易干预

BabyAGI：任务管理的极简实现

BabyAGI 是 AutoGPT 的简化版，核心只有 140 行代码，但包含了关键思想。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from collections import deque

class BabyAGI:
    def __init__(self, llm, vectorstore):
        self.llm = llm
        self.vectorstore = vectorstore
        self.task_queue = deque([])
        self.task_id_counter = 1
    
    def add_task(self, task_description):
        """添加任务到队列"""
        task = {
            "id": self.task_id_counter,
            "description": task_description
        }
        self.task_queue.append(task)
        self.task_id_counter += 1
    
    def execute_task(self, task):
        """执行任务"""
        # 检索相关上下文
        context = self.vectorstore.similarity_search(
            task["description"], k=5
        )
        
        prompt = f"""
        任务：{task['description']}
        
        上下文：
        {self.format_context(context)}
        
        请完成这个任务：
        """
        
        result = self.llm(prompt)
        
        # 存储结果
        self.vectorstore.add_texts([
            f"任务：{task['description']}\n 结果：{result}"
        ])
        
        return result
    
    def create_new_tasks(self, result, task_description, objective):
        """基于结果创建新任务"""
        prompt = f"""
        你是一个任务创建 AI 。
        
        目标：{objective}
        上一个任务：{task_description}
        上一个任务的结果：{result}
        
        当前任务列表：
        {self.format_task_queue()}
        
        基于结果，创建新的任务来帮助实现目标。
        不要创建重复的任务。
        
        以 JSON 数组格式返回新任务：
        ["任务 1", "任务 2", ...]
        """
        
        response = self.llm(prompt)
        new_tasks = json.loads(response)
        
        for task in new_tasks:
            self.add_task(task)
    
    def prioritize_tasks(self, objective):
        """重新排列任务优先级"""
        prompt = f"""
        目标：{objective}
        
        当前任务列表：
        {self.format_task_queue()}
        
        请对任务重新排序，把最重要的任务放在前面。
        返回排序后的任务 ID 列表（ JSON 数组）：
        """
        
        response = self.llm(prompt)
        task_ids = json.loads(response)
        
        # 重新排列队列
        new_queue = deque([])
        for tid in task_ids:
            for task in self.task_queue:
                if task["id"] == tid:
                    new_queue.append(task)
                    break
        
        self.task_queue = new_queue
    
    def run(self, objective, max_iterations=10):
        """主循环"""
        # 添加初始任务
        self.add_task(objective)
        
        iteration = 0
        while self.task_queue and iteration < max_iterations:
            iteration += 1
            
            # 执行队首任务
            task = self.task_queue.popleft()
            print(f"\n[{iteration}] 执行：{task['description']}")
            
            result = self.execute_task(task)
            print(f"结果：{result}")
            
            # 创建新任务
            self.create_new_tasks(result, task["description"], objective)
            
            # 重新排序
            if self.task_queue:
                self.prioritize_tasks(objective)
        
        print("\n 任务完成！")

AGENTS：模块化的框架设计

AGENTS 框架提供了更模块化的设计：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
from abc import ABC, abstractmethod

# 抽象基类
class Agent(ABC):
    @abstractmethod
    def perceive(self, observation):
        """感知环境"""
        pass
    
    @abstractmethod
    def plan(self, goal):
        """制定计划"""
        pass
    
    @abstractmethod
    def act(self, action):
        """执行动作"""
        pass
    
    @abstractmethod
    def reflect(self, feedback):
        """反思改进"""
        pass

# 具体实现
class ResearchAgent(Agent):
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.current_plan = None
    
    def perceive(self, observation):
        """处理观察到的信息"""
        # 提取关键信息
        summary = self.llm(f"总结以下信息：\n{observation}")
        
        # 更新记忆
        self.memory.add(summary)
        
        return summary
    
    def plan(self, goal):
        """制定研究计划"""
        context = self.memory.recall(goal)
        
        prompt = f"""
        研究目标：{goal}
        
        已知信息：
        {context}
        
        请制定详细的研究计划，包括：
        1. 需要搜索的关键词
        2. 需要阅读的文献
        3. 需要验证的假设
        4. 预期的输出
        """
        
        self.current_plan = self.llm(prompt)
        return self.current_plan
    
    def act(self, action):
        """执行研究动作"""
        if action["type"] == "search":
            return self.tools["search"](action["query"])
        elif action["type"] == "read":
            return self.tools["read"](action["url"])
        elif action["type"] == "analyze":
            return self.llm(f"分析：{action['content']}")
    
    def reflect(self, feedback):
        """反思研究过程"""
        prompt = f"""
        原计划：{self.current_plan}
        反馈：{feedback}
        
        请反思：
        1. 计划执行得如何？
        2. 遇到了什么问题？
        3. 应该如何调整？
        """
        
        reflection = self.llm(prompt)
        self.memory.add(f"反思：{reflection}")
        
        return reflection
    
    def run(self, goal):
        """完整的研究流程"""
        # 制定计划
        plan = self.plan(goal)
        
        # 执行计划
        results = []
        for step in self.parse_plan(plan):
            observation = self.act(step)
            summary = self.perceive(observation)
            results.append(summary)
        
        # 反思改进
        feedback = self.evaluate_results(results)
        self.reflect(feedback)
        
        # 生成最终报告
        report = self.generate_report(goal, results)
        return report

各框架对比：

框架 优势 劣势 适用场景
LangChain 生态丰富、易上手 抽象层次复杂 通用 Agent 快速开发
LangGraph 流程可视化、灵活 学习曲线陡峭 复杂流程编排
AutoGPT 高度自主 成本高、不稳定 探索性任务
BabyAGI 简单清晰 功能有限 学习和原型
AGENTS 模块化、可扩展 需要自己实现很多 定制化需求

多 Agent 协作

单个 Agent 能力有限，多个 Agent 协作能解决更复杂的问题。就像一个团队，不同成员各司其职，协同完成任务。

为什么需要多 Agent

考虑"写一本技术书"这个任务： - 研究员 Agent：收集资料、调研技术 - 作者 Agent：撰写章节内容 - 编辑 Agent：审阅和修改 - 设计师 Agent：制作配图和排版

每个 Agent 专注于自己擅长的领域，协作效率远超单个全能 Agent 。

Multi-Agent Debate：辩论式协作

让多个 Agent 对同一问题展开辩论，通过思想碰撞得到更好的答案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class DebateAgent:
    def __init__(self, name, llm, stance):
        self.name = name
        self.llm = llm
        self.stance = stance
        self.arguments = []
    
    def make_argument(self, topic, opponent_arguments):
        """提出论点"""
        prompt = f"""
        辩题：{topic}
        你的立场：{self.stance}
        
        对方的论点：
        {chr(10).join(opponent_arguments)}
        
        请提出你的论点，要：
        1. 支持自己的立场
        2. 反驳对方的观点
        3. 提供证据和逻辑
        """
        
        argument = self.llm(prompt)
        self.arguments.append(argument)
        return argument

class DebateSystem:
    def __init__(self, agents, judge_llm):
        self.agents = agents
        self.judge_llm = judge_llm
    
    def debate(self, topic, rounds=3):
        """进行辩论"""
        for round_num in range(rounds):
            print(f"\n--- 第 {round_num + 1} 轮 ---")
            
            for agent in self.agents:
                # 获取其他 Agent 的论点
                opponent_arguments = [
                    arg for other in self.agents 
                    if other != agent
                    for arg in other.arguments
                ]
                
                # 发表论点
                argument = agent.make_argument(topic, opponent_arguments)
                print(f"\n{agent.name}：{argument}")
        
        # 评委总结
        return self.judge_debate(topic)
    
    def judge_debate(self, topic):
        """评判辩论结果"""
        all_arguments = {
            agent.name: agent.arguments 
            for agent in self.agents
        }
        
        prompt = f"""
        辩题：{topic}
        
        各方论点：
        {self.format_arguments(all_arguments)}
        
        请作为公正的评委：
        1. 总结各方的核心观点
        2. 分析论点的优缺点
        3. 给出综合结论
        """
        
        judgment = self.judge_llm(prompt)
        return judgment

# 使用示例
topic = "AI 会取代程序员吗？"

agent_pro = DebateAgent("支持方", llm, "AI 会取代程序员")
agent_con = DebateAgent("反对方", llm, "AI 不会取代程序员")

debate = DebateSystem([agent_pro, agent_con], judge_llm)
result = debate.debate(topic, rounds=3)

print(f"\n 评委总结：\n{result}")

Consensus：共识机制

ReConcile 等共识框架让多个 Agent 通过投票和协商达成一致。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class ConsensusAgent:
    def __init__(self, name, llm):
        self.name = name
        self.llm = llm
    
    def propose_solution(self, problem):
        """提出解决方案"""
        prompt = f"问题：{problem}\n 请提出你的解决方案："
        return self.llm(prompt)
    
    def vote(self, problem, proposals):
        """对方案投票"""
        prompt = f"""
        问题：{problem}
        
        候选方案：
        {self.format_proposals(proposals)}
        
        请投票选择最佳方案，并说明理由。
        返回格式：
        {{ 
            "choice": 方案编号,
            "reason": "理由"
         }}
        """
        
        response = self.llm(prompt)
        return json.loads(response)
    
    def revise_solution(self, original, feedback):
        """根据反馈修改方案"""
        prompt = f"""
        原方案：{original}
        反馈意见：{feedback}
        
        请改进方案：
        """
        return self.llm(prompt)

class ReConcileSystem:
    def __init__(self, agents):
        self.agents = agents
    
    def reach_consensus(self, problem, max_rounds=5):
        """达成共识"""
        proposals = {}
        
        # 第一轮：每个 Agent 提出方案
        for agent in self.agents:
            proposal = agent.propose_solution(problem)
            proposals[agent.name] = proposal
        
        for round_num in range(max_rounds):
            print(f"\n--- 第 {round_num + 1} 轮投票 ---")
            
            # 投票
            votes = {}
            for agent in self.agents:
                vote = agent.vote(problem, proposals)
                votes[agent.name] = vote
            
            # 统计结果
            vote_counts = self.count_votes(votes)
            winner = max(vote_counts, key=vote_counts.get)
            
            # 检查是否达成共识（超过 2/3 支持）
            if vote_counts[winner] >= len(self.agents) * 2 / 3:
                print(f"\n 达成共识！选择方案：{winner}")
                return proposals[winner]
            
            # 未达成共识，收集反馈并修改方案
            feedback = self.collect_feedback(votes)
            
            for agent_name, proposal in proposals.items():
                agent = next(a for a in self.agents if a.name == agent_name)
                revised = agent.revise_solution(
                    proposal,
                    feedback.get(agent_name, "")
                )
                proposals[agent_name] = revised
        
        # 未能达成共识，返回得票最多的方案
        final_winner = max(vote_counts, key=vote_counts.get)
        return proposals[final_winner]

DOWN：分层协作框架

DOWN (Divide, Operate, Write, Navigate) 框架将任务分层：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class DOWNFramework:
    def __init__(self):
        self.coordinator = CoordinatorAgent()  # 协调者
        self.specialists = []  # 专家 Agent 列表
        self.executor = ExecutorAgent()  # 执行者
    
    def add_specialist(self, specialist):
        self.specialists.append(specialist)
    
    def solve(self, problem):
        # D: Divide - 分解任务
        subtasks = self.coordinator.divide_task(problem)
        
        # O: Operate - 分配给专家
        results = {}
        for subtask in subtasks:
            specialist = self.select_specialist(subtask)
            result = specialist.operate(subtask)
            results[subtask["id"]] = result
        
        # W: Write - 整合结果
        integrated = self.coordinator.write_summary(results)
        
        # N: Navigate - 导航下一步
        next_action = self.coordinator.navigate(integrated, problem)
        
        if next_action == "complete":
            return integrated
        else:
            # 递归处理
            return self.solve(next_action)
    
    def select_specialist(self, subtask):
        """选择最适合的专家"""
        scores = []
        for specialist in self.specialists:
            score = specialist.evaluate_capability(subtask)
            scores.append((specialist, score))
        
        best_specialist, _ = max(scores, key=lambda x: x[1])
        return best_specialist

# 定义专家 Agent
class CodeSpecialist(Agent):
    def evaluate_capability(self, subtask):
        if "代码" in subtask["description"] or "编程" in subtask["description"]:
            return 0.9
        return 0.1
    
    def operate(self, subtask):
        # 编写代码
        return self.generate_code(subtask["description"])

class ResearchSpecialist(Agent):
    def evaluate_capability(self, subtask):
        if "调研" in subtask["description"] or "搜索" in subtask["description"]:
            return 0.9
        return 0.1
    
    def operate(self, subtask):
        # 执行调研
        return self.research(subtask["description"])

# 使用
framework = DOWNFramework()
framework.add_specialist(CodeSpecialist())
framework.add_specialist(ResearchSpecialist())

result = framework.solve("开发一个天气查询网站")

MetaGPT：软件公司模拟

MetaGPT 模拟了完整的软件团队协作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class SoftwareCompany:
    def __init__(self):
        self.ceo = CEOAgent()
        self.product_manager = ProductManagerAgent()
        self.architect = ArchitectAgent()
        self.engineer = EngineerAgent()
        self.qa = QAAgent()
    
    def develop_product(self, requirement):
        """完整的产品开发流程"""
        
        # 1. CEO 制定目标
        goal = self.ceo.set_goal(requirement)
        
        # 2. PM 撰写 PRD
        prd = self.product_manager.write_prd(goal)
        
        # 3. 架构师设计系统
        design = self.architect.design_system(prd)
        
        # 4. 工程师实现代码
        code = self.engineer.write_code(design)
        
        # 5. QA 测试
        test_results = self.qa.test(code, prd)
        
        # 6. 迭代改进
        if test_results["pass"]:
            return code
        else:
            # 反馈给工程师修复
            fixed_code = self.engineer.fix_bugs(
                code,
                test_results["issues"]
            )
            return self.qa.test(fixed_code, prd)

class ProductManagerAgent(Agent):
    def write_prd(self, goal):
        """撰写产品需求文档"""
        prompt = f"""
        产品目标：{goal}
        
        请撰写详细的 PRD，包括：
        1. 用户故事
        2. 功能列表
        3. 非功能需求
        4. 验收标准
        """
        return self.llm(prompt)

class ArchitectAgent(Agent):
    def design_system(self, prd):
        """设计系统架构"""
        prompt = f"""
        PRD：{prd}
        
        请设计系统架构，包括：
        1. 技术栈选择
        2. 模块划分
        3. 数据库设计
        4. API 设计
        5. 部署方案
        """
        return self.llm(prompt)

class EngineerAgent(Agent):
    def write_code(self, design):
        """编写代码"""
        # 根据设计逐个模块实现
        modules = self.parse_modules(design)
        code = {}
        
        for module in modules:
            code[module["name"]] = self.generate_module_code(module)
        
        return code

Agent 通信协议

多 Agent 系统需要标准化的通信协议：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from dataclasses import dataclass
from enum import Enum

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    QUERY = "query"

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    message_type: MessageType
    content: dict
    timestamp: datetime

class MessageBus:
    def __init__(self):
        self.agents = {}
        self.message_queue = []
    
    def register(self, agent):
        """注册 Agent"""
        self.agents[agent.name] = agent
    
    def send(self, message: AgentMessage):
        """发送消息"""
        if message.message_type == MessageType.BROADCAST:
            # 广播给所有 Agent
            for agent in self.agents.values():
                if agent.name != message.sender:
                    agent.receive(message)
        else:
            # 点对点发送
            receiver = self.agents.get(message.receiver)
            if receiver:
                receiver.receive(message)
    
    def subscribe(self, agent_name, topic):
        """订阅主题"""
        # 实现发布-订阅模式
        pass

# Agent 基类
class CommunicatingAgent:
    def __init__(self, name, message_bus):
        self.name = name
        self.message_bus = message_bus
        self.inbox = []
        
        message_bus.register(self)
    
    def send_message(self, receiver, message_type, content):
        """发送消息"""
        message = AgentMessage(
            sender=self.name,
            receiver=receiver,
            message_type=message_type,
            content=content,
            timestamp=datetime.now()
        )
        self.message_bus.send(message)
    
    def receive(self, message):
        """接收消息"""
        self.inbox.append(message)
        self.process_message(message)
    
    def process_message(self, message):
        """处理消息"""
        if message.message_type == MessageType.REQUEST:
            response = self.handle_request(message.content)
            self.send_message(
                message.sender,
                MessageType.RESPONSE,
                response
            )

多 Agent 协作是 AI Agent 技术的前沿，能够解决单个 Agent 难以应对的复杂问题。

评估与基准测试

如何评估一个 Agent 的能力？这需要标准化的基准测试。

AgentBench：综合能力评估

AgentBench 是首个大规模的 Agent 评估基准，包含 8 个不同环境：

  1. 操作系统任务（ OS）：文件操作、程序安装
  2. 数据库操作（ DB）： SQL 查询、数据分析
  3. 知识图谱（ KG）：实体关系推理
  4. 数字卡牌游戏（ DCG）：策略规划
  5. 横向思维谜题（ LTP）：创造性思维
  6. 家庭环境导航（ ALFWorld）：多步交互
  7. 网页导航（ WebShop）：电商任务
  8. 网页问答（ Mind2Web）：复杂交互

评估示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 操作系统任务评估
task = {
    "description": "在/home/user 目录下创建一个名为 project 的文件夹，并在其中创建 README.md",
    "environment": "ubuntu_container",
    "success_criteria": [
        "目录/home/user/project 存在",
        "文件/home/user/project/README.md 存在"
    ]
}

def evaluate_os_task(agent, task):
    # 初始化环境
    env = DockerEnvironment("ubuntu:latest")
    
    # Agent 执行
    agent.run(task["description"])
    
    # 检查成功标准
    success = all([
        env.check_condition(criterion)
        for criterion in task["success_criteria"]
    ])
    
    return {
        "success": success,
        "steps": agent.get_trajectory(),
        "cost": agent.get_token_usage()
    }

AgentBench 的评估结果显示： - GPT-4 在大部分任务上表现最好（总分 35.1%） - 开源模型如 LLaMA 明显落后（< 10%） - 所有模型在复杂交互任务上都有很大提升空间

GAIA：真实世界问答

GAIA (General AI Assistants) 基准专注于真实世界的问答任务，特点：

  • 需要多步推理：不是一次搜索就能解决
  • 需要工具使用：计算器、搜索、代码执行
  • 答案明确：便于自动评估

示例任务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
gaia_task = {
    "question": "2023 年诺贝尔物理学奖获得者中，哪位的 h-index 最高？",
    "level": "medium",
    "expected_steps": [
        "搜索 2023 年诺贝尔物理学奖获得者",
        "查询每位获得者的 Google Scholar 页面",
        "提取 h-index 数值",
        "比较并返回最高者"
    ],
    "ground_truth": "Anne L'Huillier (h-index: 82)"
}

def evaluate_gaia(agent, task):
    answer = agent.run(task["question"])
    
    # 标准化答案比较
    correct = compare_answers(answer, task["ground_truth"])
    
    return {
        "correct": correct,
        "agent_answer": answer,
        "expected_answer": task["ground_truth"]
    }

GAIA 2 扩展版加入了更多挑战： - 多模态输入（图片、音频） - 需要隐私保护的个人助手任务 - 长期记忆依赖的场景

AgentBoard：全方位评估

AgentBoard 提供更细粒度的能力评估：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class AgentBoard:
    def evaluate(self, agent):
        scores = {}
        
        # 1. 规划能力
        scores["planning"] = self.test_planning(agent)
        
        # 2. 工具使用
        scores["tool_use"] = self.test_tool_use(agent)
        
        # 3. 错误恢复
        scores["error_recovery"] = self.test_error_recovery(agent)
        
        # 4. 长期任务
        scores["long_horizon"] = self.test_long_horizon(agent)
        
        # 5. 多模态理解
        scores["multimodal"] = self.test_multimodal(agent)
        
        return scores
    
    def test_planning(self, agent):
        """测试规划能力"""
        tasks = [
            "组织一次 10 人的户外徒步活动",
            "规划一周的健康饮食计划",
            "准备一场技术分享会"
        ]
        
        scores = []
        for task in tasks:
            plan = agent.plan(task)
            score = self.evaluate_plan_quality(plan)
            scores.append(score)
        
        return sum(scores) / len(scores)
    
    def evaluate_plan_quality(self, plan):
        """评估计划质量"""
        # 检查完整性、合理性、可执行性
        criteria = {
            "completeness": 0.3,  # 步骤是否完整
            "feasibility": 0.3,   # 是否可行
            "efficiency": 0.2,    # 是否高效
            "detail": 0.2         # 细节是否充分
        }
        
        # 使用评估模型打分
        scores = self.evaluator_llm.score(plan, criteria)
        return sum(scores.values())

自动化评估流程

完整的评估流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class AgentEvaluator:
    def __init__(self, benchmarks):
        self.benchmarks = benchmarks
        self.results = []
    
    def run_evaluation(self, agent, num_tasks=100):
        """运行评估"""
        for i in range(num_tasks):
            task = self.sample_task()
            
            try:
                # 执行任务
                start_time = time.time()
                result = agent.run(task["input"])
                elapsed = time.time() - start_time
                
                # 评估结果
                score = self.score_result(result, task["expected"])
                
                self.results.append({
                    "task_id": task["id"],
                    "success": score >= 0.8,
                    "score": score,
                    "time": elapsed,
                    "tokens": agent.get_token_usage()
                })
            
            except Exception as e:
                self.results.append({
                    "task_id": task["id"],
                    "success": False,
                    "error": str(e)
                })
        
        return self.compute_metrics()
    
    def compute_metrics(self):
        """计算指标"""
        total = len(self.results)
        success_count = sum(1 for r in self.results if r.get("success"))
        
        avg_score = np.mean([r["score"] for r in self.results if "score" in r])
        avg_time = np.mean([r["time"] for r in self.results if "time" in r])
        total_tokens = sum(r["tokens"] for r in self.results if "tokens" in r)
        
        return {
            "success_rate": success_count / total,
            "average_score": avg_score,
            "average_time": avg_time,
            "total_cost": total_tokens * 0.00001,  # 假设每 1k tokens = $0.01
            "reliability": self.compute_reliability()
        }
    
    def compute_reliability(self):
        """计算可靠性（相同任务多次执行的一致性）"""
        # 对部分任务重复执行 3 次
        consistency_scores = []
        
        for task_type in self.task_types:
            results = [r for r in self.results if r["type"] == task_type]
            if len(results) >= 3:
                scores = [r["score"] for r in results[:3]]
                variance = np.var(scores)
                consistency_scores.append(1 - variance)
        
        return np.mean(consistency_scores)

人类评估

某些任务需要人类评估：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def human_evaluation(agent_outputs, tasks):
    """人类评估界面"""
    scores = []
    
    for output, task in zip(agent_outputs, tasks):
        print(f"\n 任务：{task['description']}")
        print(f"Agent 输出：{output}")
        
        # 多维度评分
        relevance = int(input("相关性(1-5)："))
        accuracy = int(input("准确性(1-5)："))
        completeness = int(input("完整性(1-5)："))
        usefulness = int(input("实用性(1-5)："))
        
        scores.append({
            "relevance": relevance,
            "accuracy": accuracy,
            "completeness": completeness,
            "usefulness": usefulness,
            "overall": (relevance + accuracy + completeness + usefulness) / 4
        })
    
    return scores

评估是持续改进 Agent 的基础。通过系统化的测试，我们能发现 Agent 的弱点，有针对性地优化。

实战：构建完整的 Agent 项目

理论和框架都了解了，动手构建一个完整的 Agent 项目：智能数据分析助手。

项目需求

目标：构建一个能够自动分析数据、生成报告的 Agent 。

功能： 1. 接收 CSV/Excel 文件 2. 自动进行数据清洗 3. 探索性数据分析（ EDA） 4. 生成可视化图表 5. 撰写分析报告 6. 回答用户的数据相关问题

系统架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 系统架构
"""
┌─────────────────────────────────────────┐
│           User Interface                │
│    (接收文件、展示结果、对话)            │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│        Main Agent                       │
│    (协调、规划、决策)                    │
└──┬───────┬────────┬────────┬───────────┘
   │       │        │        │
   ▼       ▼        ▼        ▼
┌─────┐ ┌──────┐ ┌──────┐ ┌──────────┐
│ Data │ │ EDA  │ │ Viz  │ │  Report  │
│ Clean │ │ Agent │ │ Agent │ │  Agent   │
└─────┘ └──────┘ └──────┘ └──────────┘
   │       │        │        │
   └───────┴────────┴────────┴───────────┐
                                          │
                ┌─────────────────────────▼──┐
                │     Shared Memory          │
                │  (数据、中间结果、报告)     │
                └────────────────────────────┘
"""

核心实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from langchain.chat_models import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.memory import ConversationBufferMemory

# ===== 工具定义 =====

@tool
def load_data(file_path: str) -> str:
    """加载数据文件"""
    try:
        if file_path.endswith('.csv'):
            df = pd.read_csv(file_path)
        elif file_path.endswith('.xlsx'):
            df = pd.read_excel(file_path)
        else:
            return "不支持的文件格式"
        
        # 保存到共享内存
        DataStore.save("raw_data", df)
        
        return f"成功加载数据：{df.shape[0]}行 × {df.shape[1]}列\n 列名：{list(df.columns)}"
    except Exception as e:
        return f"加载失败：{str(e)}"

@tool
def clean_data() -> str:
    """清洗数据"""
    df = DataStore.load("raw_data")
    
    # 删除重复行
    df = df.drop_duplicates()
    
    # 处理缺失值
    numeric_cols = df.select_dtypes(include=['number']).columns
    df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].mean())
    
    categorical_cols = df.select_dtypes(include=['object']).columns
    df[categorical_cols] = df[categorical_cols].fillna(df[categorical_cols].mode().iloc[0])
    
    # 保存清洗后的数据
    DataStore.save("clean_data", df)
    
    return f"数据清洗完成：删除了{len(DataStore.load('raw_data')) - len(df)}行重复数据"

@tool
def describe_data() -> str:
    """获取数据统计描述"""
    df = DataStore.load("clean_data")
    return df.describe().to_string()

@tool
def analyze_correlation() -> str:
    """分析变量相关性"""
    df = DataStore.load("clean_data")
    numeric_df = df.select_dtypes(include=['number'])
    
    corr = numeric_df.corr()
    
    # 找出强相关的变量对
    strong_corr = []
    for i in range(len(corr.columns)):
        for j in range(i+1, len(corr.columns)):
            if abs(corr.iloc[i, j]) > 0.7:
                strong_corr.append(
                    f"{corr.columns[i]} 与 {corr.columns[j]}: {corr.iloc[i, j]:.2f}"
                )
    
    DataStore.save("correlation", corr)
    
    return "相关性分析完成。\n 强相关关系：\n" + "\n".join(strong_corr)

@tool
def create_visualization(chart_type: str, x_column: str, y_column: str = None) -> str:
    """创建可视化图表
    
    Args:
        chart_type: 图表类型 (histogram, scatter, bar, box)
        x_column: X 轴列名
        y_column: Y 轴列名（可选）
    """
    df = DataStore.load("clean_data")
    
    plt.figure(figsize=(10, 6))
    
    if chart_type == "histogram":
        plt.hist(df[x_column], bins=30)
        plt.xlabel(x_column)
        plt.ylabel("频数")
    
    elif chart_type == "scatter" and y_column:
        plt.scatter(df[x_column], df[y_column])
        plt.xlabel(x_column)
        plt.ylabel(y_column)
    
    elif chart_type == "bar":
        df[x_column].value_counts().plot(kind='bar')
        plt.xlabel(x_column)
        plt.ylabel("计数")
    
    elif chart_type == "box":
        df.boxplot(column=x_column)
    
    plt.title(f"{chart_type.title()} - {x_column}")
    
    # 保存图表
    filename = f"chart_{chart_type}_{x_column}.png"
    plt.savefig(filename)
    plt.close()
    
    DataStore.add_chart(filename)
    
    return f"图表已生成：{filename}"

@tool
def generate_report() -> str:
    """生成分析报告"""
    df = DataStore.load("clean_data")
    charts = DataStore.get_charts()
    
    report = f"""
# 数据分析报告

## 数据概览
- 总行数：{len(df)}
- 总列数：{len(df.columns)}
- 列名：{', '.join(df.columns)}

## 数据统计
{df.describe().to_string()}

## 可视化图表
"""
    
    for chart in charts:
        report += f"\n![{chart}]({chart})\n"
    
    # 保存报告
    DataStore.save("report", report)
    with open("analysis_report.md", "w") as f:
        f.write(report)
    
    return "报告已生成： analysis_report.md"

# ===== 数据存储 =====

class DataStore:
    """共享数据存储"""
    _data = {}
    _charts = []
    
    @classmethod
    def save(cls, key, value):
        cls._data[key] = value
    
    @classmethod
    def load(cls, key):
        return cls._data.get(key)
    
    @classmethod
    def add_chart(cls, filename):
        cls._charts.append(filename)
    
    @classmethod
    def get_charts(cls):
        return cls._charts

# ===== 主 Agent =====

class DataAnalysisAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        
        self.tools = [
            load_data,
            clean_data,
            describe_data,
            analyze_correlation,
            create_visualization,
            generate_report
        ]
        
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        
        self.agent = self._create_agent()
    
    def _create_agent(self):
        """创建 Agent"""
        from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的数据分析师助手。

你的工作流程：
1. 加载数据文件
2. 清洗数据（处理缺失值、重复值）
3. 进行探索性数据分析
4. 创建合适的可视化图表
5. 生成分析报告

在分析时，注意：
- 先理解数据的结构和含义
- 识别数据中的模式和异常
- 选择恰当的图表类型
- 提供清晰的解释和洞察
"""),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])
        
        agent = create_openai_functions_agent(self.llm, self.tools, prompt)
        
        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True,
            max_iterations=10
        )
    
    def analyze(self, file_path):
        """自动分析数据文件"""
        prompt = f"""
        请分析数据文件：{file_path}
        
        执行完整的分析流程：
        1. 加载数据
        2. 清洗数据
        3. 统计描述
        4. 相关性分析
        5. 创建 3-5 个有意义的可视化图表
        6. 生成完整报告
        """
        
        return self.agent.invoke({"input": prompt})
    
    def chat(self, question):
        """对话式查询"""
        return self.agent.invoke({"input": question})

# ===== 使用示例 =====

if __name__ == "__main__":
    # 创建 Agent
    agent = DataAnalysisAgent()
    
    # 自动分析
    print("=== 自动分析模式 ===")
    result = agent.analyze("sales_data.csv")
    print("\n 分析完成！")
    print(result["output"])
    
    # 对话模式
    print("\n=== 对话模式 ===")
    
    while True:
        question = input("\n 你的问题（输入'quit'退出）：")
        if question.lower() == 'quit':
            break
        
        response = agent.chat(question)
        print(f"\nAgent: {response['output']}")

运行效果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
=== 自动分析模式 ===

> Entering AgentExecutor chain...

Thought: 我需要先加载数据文件
Action: load_data
Action Input: {"file_path": "sales_data.csv"}
Observation: 成功加载数据： 1000 行 × 5 列
列名：['date', 'product', 'quantity', 'price', 'revenue']

Thought: 数据已加载，现在清洗数据
Action: clean_data
Observation: 数据清洗完成：删除了 15 行重复数据

Thought: 获取统计描述
Action: describe_data
Observation: 
       quantity      price      revenue
count   985.000   985.000     985.000
mean     45.230    99.850    4513.205
std      12.450    25.320    1123.450
...

Thought: 分析相关性
Action: analyze_correlation
Observation: 相关性分析完成。
强相关关系：
quantity 与 revenue: 0.89
price 与 revenue: 0.76

Thought: 创建可视化图表
Action: create_visualization
Action Input: {"chart_type": "histogram", "x_column": "revenue"}
Observation: 图表已生成： chart_histogram_revenue.png

Action: create_visualization
Action Input: {"chart_type": "scatter", "x_column": "quantity", "y_column": "revenue"}
Observation: 图表已生成： chart_scatter_quantity_revenue.png

Action: create_visualization
Action Input: {"chart_type": "bar", "x_column": "product"}
Observation: 图表已生成： chart_bar_product.png

Thought: 生成最终报告
Action: generate_report
Observation: 报告已生成： analysis_report.md

Final Answer: 数据分析完成！

主要发现：
1. 数据集包含 985 条有效销售记录
2. 销售数量和收入有很强的正相关关系（ 0.89）
3. 价格对收入也有较强影响（ 0.76）
4. 已生成 3 个可视化图表
5. 完整报告已保存到 analysis_report.md

=== 对话模式 ===

你的问题：哪个产品销量最高？

Agent: 让我查询一下...

Action: load_data
...（ Agent 会调用相应工具回答）

Agent: 根据数据分析，产品 A 的销量最高，共售出 12,450 件。

扩展功能

这个基础版本可以继续扩展：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# 1. 添加更多工具
@tool
def detect_outliers(column: str) -> str:
    """检测异常值"""
    df = DataStore.load("clean_data")
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    outliers = df[(df[column] < Q1 - 1.5*IQR) | (df[column] > Q3 + 1.5*IQR)]
    return f"发现{len(outliers)}个异常值"

@tool
def predict_trend(column: str, periods: int = 30) -> str:
    """预测趋势"""
    from statsmodels.tsa.arima.model import ARIMA
    df = DataStore.load("clean_data")
    model = ARIMA(df[column], order=(1,1,1))
    fitted = model.fit()
    forecast = fitted.forecast(steps=periods)
    return f"未来{periods}期预测：{forecast.tolist()}"

# 2. 添加多 Agent 协作
class SpecialistAgents:
    """专家 Agent 团队"""
    
    def __init__(self):
        self.cleaning_agent = DataCleaningAgent()
        self.viz_agent = VisualizationAgent()
        self.report_agent = ReportAgent()
    
    def collaborate(self, task):
        # 清洗 Agent 处理数据质量
        cleaned = self.cleaning_agent.process()
        
        # 可视化 Agent 创建图表
        charts = self.viz_agent.create_charts(cleaned)
        
        # 报告 Agent 整合结果
        report = self.report_agent.generate(cleaned, charts)
        
        return report

# 3. 添加反思机制
class ReflectiveAnalysisAgent(DataAnalysisAgent):
    def analyze_with_reflection(self, file_path):
        # 第一次分析
        result = self.analyze(file_path)
        
        # 自我评估
        evaluation = self.evaluate_analysis(result)
        
        # 如果质量不够，改进
        if evaluation["score"] < 0.8:
            improvements = self.suggest_improvements(evaluation)
            result = self.refine_analysis(improvements)
        
        return result

这个完整的项目展示了如何将 Agent 的核心能力整合到实际应用中。

最佳实践与注意事项

构建生产级 Agent 系统，需要注意以下要点：

提示词工程

好的提示词是 Agent 成功的关键：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# ❌ 差的提示词
prompt = "分析这个数据"

# ✅ 好的提示词
prompt = """
你是一个专业的数据分析师。

任务：分析提供的销售数据

步骤：
1. 首先检查数据质量（缺失值、异常值）
2. 计算关键指标（总销售额、平均客单价、增长率）
3. 识别趋势和模式
4. 提供可行的业务建议

输出格式：
- 数据质量报告
- 关键指标表格
- 趋势分析（附图表）
- 业务建议（ 3-5 条）

注意：
- 使用专业术语
- 提供具体数字
- 解释分析逻辑
"""

错误处理

健壮的错误处理机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class RobustAgent:
    def execute_with_retry(self, action, max_retries=3):
        """带重试的执行"""
        for attempt in range(max_retries):
            try:
                result = self.execute(action)
                return result
            
            except RateLimitError:
                wait_time = 2 ** attempt  # 指数退避
                time.sleep(wait_time)
            
            except ToolExecutionError as e:
                # 尝试修复工具调用
                fixed_action = self.fix_tool_call(action, e)
                result = self.execute(fixed_action)
                return result
            
            except Exception as e:
                if attempt == max_retries - 1:
                    # 最后一次尝试仍失败，降级处理
                    return self.fallback_strategy(action, e)
                else:
                    logging.error(f"尝试 {attempt + 1} 失败：{e}")

成本控制

LLM 调用成本可能很高：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class CostAwareAgent:
    def __init__(self, budget_limit=10.0):
        self.budget_limit = budget_limit  # 美元
        self.total_cost = 0.0
    
    def call_llm(self, prompt, model="gpt-4"):
        # 估算成本
        estimated_tokens = len(prompt.split()) * 1.3
        estimated_cost = estimated_tokens / 1000 * 0.03  # GPT-4 价格
        
        if self.total_cost + estimated_cost > self.budget_limit:
            # 超出预算，降级到便宜模型
            model = "gpt-3.5-turbo"
            estimated_cost = estimated_tokens / 1000 * 0.001
        
        response = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # 记录实际成本
        actual_tokens = response.usage.total_tokens
        actual_cost = actual_tokens / 1000 * (0.03 if model == "gpt-4" else 0.001)
        self.total_cost += actual_cost
        
        return response.choices[0].message.content

安全性

Agent 能执行代码和操作系统命令，必须严格控制权限：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class SecureAgent:
    def __init__(self):
        # 白名单：允许的操作
        self.allowed_operations = [
            "read_file",
            "write_file",
            "search_web"
        ]
        
        # 黑名单：禁止的命令
        self.forbidden_patterns = [
            r"rm\s+-rf",
            r"del\s+/f",
            r"DROP\s+TABLE",
            r"eval\(",
        ]
    
    def validate_action(self, action):
        """验证动作是否安全"""
        if action["type"] not in self.allowed_operations:
            raise SecurityError(f"不允许的操作：{action['type']}")
        
        # 检查命令内容
        for pattern in self.forbidden_patterns:
            if re.search(pattern, str(action), re.IGNORECASE):
                raise SecurityError(f"检测到危险命令：{pattern}")
        
        return True
    
    def execute_in_sandbox(self, code):
        """在沙箱中执行代码"""
        import docker
        
        client = docker.from_env()
        container = client.containers.run(
            "python:3.9-slim",
            command=f"python -c '{code}'",
            remove=True,
            network_disabled=True,  # 禁用网络
            mem_limit="256m",  # 限制内存
            cpu_quota=50000,  # 限制 CPU
            timeout=30  # 超时限制
        )
        
        return container.decode()

可观测性

生产环境需要完善的监控：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import logging
from datetime import datetime

class ObservableAgent:
    def __init__(self):
        self.logger = logging.getLogger("agent")
        self.metrics = {
            "total_tasks": 0,
            "successful_tasks": 0,
            "failed_tasks": 0,
            "total_cost": 0.0,
            "average_latency": 0.0
        }
    
    def execute_with_logging(self, task):
        """带日志的执行"""
        task_id = self.generate_task_id()
        start_time = datetime.now()
        
        self.logger.info(f"[{task_id}] 开始执行任务：{task}")
        
        try:
            result = self.execute(task)
            
            elapsed = (datetime.now() - start_time).total_seconds()
            
            self.logger.info(f"[{task_id}] 任务成功，耗时{elapsed}秒")
            
            self.metrics["successful_tasks"] += 1
            self.update_metrics(elapsed)
            
            return result
        
        except Exception as e:
            self.logger.error(f"[{task_id}] 任务失败：{str(e)}")
            self.metrics["failed_tasks"] += 1
            raise
        
        finally:
            self.metrics["total_tasks"] += 1
    
    def export_metrics(self):
        """导出指标供监控系统使用"""
        return {
            **self.metrics,
            "success_rate": self.metrics["successful_tasks"] / self.metrics["total_tasks"],
            "timestamp": datetime.now().isoformat()
        }

未来展望

AI Agent 技术仍在快速演进，未来几个重要方向：

多模态 Agent

未来的 Agent 将不限于文本，能够处理图像、音频、视频：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class MultimodalAgent:
    def process(self, inputs):
        """处理多模态输入"""
        if inputs["type"] == "image":
            # 图像理解
            description = self.vision_model.describe(inputs["data"])
            
        elif inputs["type"] == "audio":
            # 语音识别
            text = self.speech_to_text(inputs["data"])
            
        elif inputs["type"] == "video":
            # 视频分析
            frames = self.extract_keyframes(inputs["data"])
            analysis = self.analyze_frames(frames)
        
        # 综合处理
        return self.llm.process(description, text, analysis)

具身智能

Agent 将从纯软件走向物理世界，控制机器人：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class EmbodiedAgent:
    def __init__(self, robot):
        self.robot = robot
        self.perception = VisionSystem()
        self.planning = MotionPlanner()
    
    def execute_task(self, instruction):
        """执行物理世界任务"""
        # 感知环境
        scene = self.perception.understand_scene()
        
        # 规划动作序列
        plan = self.planning.plan(instruction, scene)
        
        # 执行动作
        for action in plan:
            self.robot.execute(action)
            feedback = self.robot.get_feedback()
            
            # 动态调整
            if feedback["success"]:
                continue
            else:
                plan = self.replan(action, feedback)

终身学习

Agent 将能够从每次交互中学习，持续改进：

1
2
3
4
5
6
7
8
9
10
11
12
class LifelongLearningAgent:
    def learn_from_experience(self, trajectory, feedback):
        """从经验中学习"""
        # 提取教训
        lesson = self.extract_lesson(trajectory, feedback)
        
        # 更新知识库
        self.knowledge_base.add(lesson)
        
        # 微调模型（可选）
        if len(self.experiences) >= 100:
            self.fine_tune_model(self.experiences)

通用人工智能（ AGI）之路

Agent 是通向 AGI 的重要一步。未来的 Agent 将具备：

  • 跨领域迁移：在一个领域学到的能力可以迁移到其他领域
  • 元学习：学会如何学习，快速适应新任务
  • 创造性：不仅执行任务，还能提出新想法
  • 价值对齐：理解人类价值观，做出符合伦理的决策

总结

AI Agent 代表了人工智能从"理解"到"行动"的跨越。本文系统介绍了：

  1. Agent 基础：定义、核心组件、与普通 LLM 的区别
  2. 核心能力：规划、记忆、工具使用、反思四大支柱
  3. 主流框架： LangChain 、 LangGraph 、 AutoGPT 、 BabyAGI 等
  4. 多 Agent 协作：辩论、共识、分层协作等模式
  5. 评估方法： AgentBench 、 GAIA 、 AgentBoard 等基准
  6. 实战项目：完整的数据分析 Agent 实现
  7. 最佳实践：提示词工程、错误处理、安全性、可观测性

Agent 技术正在从研究走向生产，从玩具走向工具。无论你是研究者、工程师还是产品经理，现在都是深入了解和应用 Agent 技术的最佳时机。

希望这篇完全指南能帮助你构建属于自己的智能代理系统。 Agent 的时代，才刚刚开始。

  • 本文标题：AI Agent 完全指南：从理论到工业实践
  • 本文作者：Chen Kai
  • 创建时间：2025-08-15 10:45:00
  • 本文链接：https://www.chenk.top/AI-Agent%E5%AE%8C%E5%85%A8%E6%8C%87%E5%8D%97%EF%BC%9A%E4%BB%8E%E7%90%86%E8%AE%BA%E5%88%B0%E5%B7%A5%E4%B8%9A%E5%AE%9E%E8%B7%B5/
  • 版权声明：本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！
 评论