章节1：Agent核心原理

本章导学

AI Agent（智能体）是当前大模型应用领域最核心的技术方向。本章将从底层原理出发，深入剖析Agent的三大核心能力（规划、记忆、工具使用），讲解主流的推理-行动框架（ReAct）和执行模式（Plan-and-Execute），以及反思机制与评测方法，帮助你建立起对Agent技术体系的系统性认知。

学习目标

理解AI Agent的三大核心能力及其实现原理
掌握ReAct推理-行动框架的工作流程
学会Plan-and-Execute模式的实现方法
理解反思机制与自我纠错的实现逻辑
了解Agent评测方法与能力边界

1.1 Agent三大核心能力

1.1.1 什么是AI Agent

AI Agent（智能体）是一个能自主感知环境、制定计划、调用工具、执行行动并记忆经验的大模型驱动的程序实体。与传统的大模型"单轮问答"不同，Agent具备持续推理和多步行动的能力。

1.1.2 三大核心能力详解

核心能力	能力描述	类比人类
规划（Planning）	将复杂目标分解为子任务，制定执行步骤	先想好怎么做
记忆（Memory）	存储历史信息、上下文、经验教训	记住之前的事
工具（Tools）	调用外部API、数据库、搜索引擎等	使用工具做事

（1）规划（Planning）

规划能力使Agent能够将复杂问题分解为可执行的步骤序列。

关键实现方式：

任务分解（Task Decomposition）：将大目标拆解为子任务
思维链（Chain-of-Thought, CoT）：逐步推理
子目标（Subgoal）设定：为每一步设定中间目标

代码示例：任务规划器

python

from openai import OpenAI
import json

client = OpenAI()

class TaskPlanner:
    """任务规划器：将用户目标分解为可执行的子任务"""
    
    def __init__(self, model="gpt-4"):
        self.model = model
    
    def plan(self, objective: str) -> list[dict]:
        """根据目标制定执行计划"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": """你是一个任务规划专家。
请将用户的目标分解为3-5个具体的子任务。
每个子任务包含：task_id, description, dependencies（依赖的任务ID列表）。
以JSON数组格式输出。"""},
                {"role": "user", "content": f"目标：{objective}"}
            ],
            response_format={"type": "json_object"}
        )
        plan = json.loads(response.choices[0].message.content)
        return plan.get("tasks", [])

# 使用示例
planner = TaskPlanner()
tasks = planner.plan("调研2025年AI Agent市场趋势并生成分析报告")
for t in tasks:
    print(f"任务{t['task_id']}: {t['description']}")
    if t['dependencies']:
        print(f"  依赖: {t['dependencies']}")

（2）记忆（Memory）

Agent的记忆系统借鉴了人类记忆的分层结构。

三种记忆类型：

记忆类型	存储内容	生命周期	技术实现
短期记忆	当前会话上下文	单轮对话内	上下文窗口（Context Window）
长期记忆	跨会话的知识与经验	持久化	向量数据库（如ChromaDB）
工作记忆	当前正在处理的信息	任务执行中	Token序列 + 注意力机制

代码示例：带记忆的Agent

python

from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# 1. 基于SQLite的持久化记忆
message_history = SQLChatMessageHistory(
    session_id="user_session_001",
    connection_string="sqlite:///agent_memory.db"
)

# 2. 缓冲记忆（保存最近N轮对话）
memory = ConversationBufferMemory(
    chat_memory=message_history,
    return_messages=True,
    k=10  # 保留最近10轮
)

# 3. 创建带记忆的对话链
llm = ChatOpenAI(model="gpt-4")
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 4. 使用
response = conversation.predict(input="我之前说过关于Agent规划能力的内容吗？")
print(response)

（3）工具（Tools）

工具能力是Agent与外部世界交互的接口。通过Function Calling机制，大模型可以动态决定调用哪些工具。

常见工具类型：

🔍 搜索工具（Web Search）
📊 数据查询（SQL/API）
📧 消息发送（Email/Slack）
📁 文件操作（读写/解析）
🧮 计算工具（数学/统计）

代码示例：工具定义与调用

python

from openai import OpenAI
import json

client = OpenAI()

# 1. 定义工具函数
def get_weather(city: str) -> str:
    """模拟查询天气"""
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C",
        "深圳": "阵雨，30°C"
    }
    return weather_data.get(city, f"暂未收录{city}的天气数据")

def calculate(expression: str) -> float:
    """安全的数学计算"""
    import operator
    allowed = {
        '+': operator.add, '-': operator.sub,
        '*': operator.mul, '/': operator.truediv
    }
    parts = expression.split()
    if len(parts) != 3:
        return "仅支持二元运算，如 '10 + 20'"
    a, op, b = parts
    if op not in allowed:
        return f"不支持的运算符: {op}"
    return allowed[op](float(a), float(b))

# 2. 定义工具描述（OpenAI Function Calling格式）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行二元数学运算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '10 + 20'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# 3. Agent执行循环
def agent_loop(user_message: str, max_turns: int = 5):
    """Agent主循环：推理 -> 行动 -> 观察"""
    messages = [{"role": "user", "content": user_message}]
    available_functions = {
        "get_weather": get_weather,
        "calculate": calculate
    }
    
    for turn in range(max_turns):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        
        # 如果没有工具调用，直接返回
        if not message.tool_calls:
            return message.content
        
        messages.append(message)
        
        # 执行工具调用
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            result = available_functions[func_name](**args)
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })
    
    return "已达到最大执行轮次"

# 使用
result = agent_loop("北京今天天气怎么样？再算一下 100 * 3.5 等于多少？")
print(result)

1.2 ReAct推理-行动框架

1.2.1 什么是ReAct

ReAct（Reasoning + Acting）由Shunyu Yao等人于2022年提出，核心思想是推理（Reasoning）与行动（Acting）交替进行，而非先全部想好再执行。

1.2.2 ReAct工作流程

用户输入
   │
   ▼
┌─────────────────┐
│   Thought（思考）  │  ← 推理当前状态和下一步行动
│   → 我需要查询...  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   Action（行动）   │  ← 调用工具或执行操作
│   → 调用搜索工具   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   Observation（观察） │  ← 获取工具执行结果
│   → 搜索结果：...   │
└────────┬────────┘
         │
         ▼  (循环直到得出最终答案)
┌─────────────────┐
│   Final Answer（最终答案）│
└─────────────────┘

ReAct的关键模式是：思考 → 行动 → 观察 → 再思考 → ... → 最终回答

1.2.3 ReAct代码实现

python

from openai import OpenAI
import json

client = OpenAI()

class ReActAgent:
    """ReAct推理-行动Agent"""
    
    SYSTEM_PROMPT = """你是一个具有推理能力的AI助手。请遵循以下格式回答：

思考：分析当前问题，决定下一步行动
行动：调用具体工具（如果不需要工具则输出"无"）
观察：展示工具执行结果
...（可重复思考-行动-观察）
思考：我可以根据以上信息回答用户了
最终答案：给用户的完整回答

可用工具：
- search_web(query): 搜索网络信息
- calculate(expression): 数学计算
"""
    
    def __init__(self, model="gpt-4"):
        self.model = model
        self.messages = [{"role": "system", "content": self.SYSTEM_PROMPT}]
    
    def run(self, user_input: str, max_steps: int = 10) -> str:
        """运行ReAct循环"""
        self.messages.append({"role": "user", "content": user_input})
        
        for step in range(max_steps):
            response = client.chat.completions.create(
                model=self.model,
                messages=self.messages
            )
            reply = response.choices[0].message.content
            self.messages.append({"role": "assistant", "content": reply})
            
            # 检查是否输出了最终答案
            if "最终答案" in reply or "Final Answer" in reply:
                # 提取最终答案部分
                answer_start = reply.find("最终答案") if "最终答案" in reply else reply.find("Final Answer")
                return reply[answer_start:]
            
            # 提取行动指令
            if "行动：" in reply or "Action:" in reply:
                action_line = [l for l in reply.split('\n') if "行动：" in l or "Action:" in l]
                if action_line:
                    print(f"  [步骤{step+1}] {action_line[0]}")
        
        return "达到最大执行步数"
    
# 使用示例
agent = ReActAgent()
result = agent.run("2025年最受欢迎的AI编程助手有哪些？请给出排名前3的。")
print(result)

1.2.4 ReAct vs 纯推理 vs 纯行动

方式	特点	适用场景
纯推理（CoT）	只思考不行动	纯知识问答
纯行动（Act-Only）	只行动不思考	简单工具调用
ReAct	思考+行动交替	复杂多步任务

ReAct的优势：通过"思考"帮助模型纠偏，通过"行动"获取外部信息，两者互补大大提升了复杂任务的完成率。

1.3 Plan-and-Execute模式

1.3.1 模式概述

Plan-and-Execute（先规划再执行）是一种将任务拆分为"规划阶段"和"执行阶段"的架构。

Plan（规划）：LLM先生成一个完整的执行计划，包含步骤列表和依赖关系
Execute（执行）：按照计划逐步执行，每步可能调用工具或查询知识

1.3.2 与ReAct的区别

维度	ReAct	Plan-and-Execute
规划时机	边想边做	先想好再做
灵活性	高，可动态调整	中，按计划执行
长任务	可能迷失方向	有全局规划，更稳定
实现复杂度	较低	中等

1.3.3 代码实现

python

from openai import OpenAI
import json
from typing import List, Dict

client = OpenAI()

class PlanAndExecuteAgent:
    """Plan-and-Execute Agent"""
    
    def __init__(self, model="gpt-4"):
        self.model = model
        self.plan = []
        self.results = {}
    
    def create_plan(self, objective: str) -> List[Dict]:
        """阶段1：制定计划"""
        prompt = f"""目标：{objective}
请制定一个详细的执行计划，以JSON数组格式输出。
每个步骤包含：step_id(数字), action(行动描述), tool(所需工具), depends_on(依赖的step_id列表)

可用工具：search_web, analyze_data, write_report, read_file
"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是任务规划专家，输出JSON格式的计划。"},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"}
        )
        plan_data = json.loads(response.choices[0].message.content)
        self.plan = plan_data.get("steps", plan_data.get("plan", []))
        return self.plan
    
    def execute_step(self, step: Dict) -> str:
        """阶段2：执行单个步骤"""
        print(f"执行步骤{step['step_id']}: {step['action']}")
        # 这里模拟工具调用
        if step.get("tool") == "search_web":
            return f"[搜索结果] 关于'{step['action']}'的模拟搜索结果"
        elif step.get("tool") == "analyze_data":
            return f"[分析结果] 数据分析完成"
        return f"[执行完成] {step['action']}"
    
    def execute_plan(self) -> str:
        """按依赖顺序执行计划"""
        executed = set()
        
        while len(executed) < len(self.plan):
            for step in self.plan:
                step_id = step["step_id"]
                if step_id in executed:
                    continue
                # 检查依赖是否都已执行
                deps = step.get("depends_on", [])
                if all(d in executed for d in deps):
                    result = self.execute_step(step)
                    self.results[step_id] = result
                    executed.add(step_id)
        
        return self._summarize()
    
    def _summarize(self) -> str:
        """汇总执行结果"""
        summary = "## 执行报告\n\n"
        for step in self.plan:
            sid = step["step_id"]
            summary += f"**步骤{sid}**: {step['action']}\n"
            summary += f"> 结果: {self.results.get(sid, 'N/A')}\n\n"
        return summary
    
    def run(self, objective: str) -> str:
        """完整流程：先规划再执行"""
        print("【规划阶段】")
        plan = self.create_plan(objective)
        print(f"计划已制定，共{len(plan)}个步骤\n")
        
        print("【执行阶段】")
        result = self.execute_plan()
        return result

# 使用示例
agent = PlanAndExecuteAgent()
result = agent.run("调研AI Agent市场，撰写一份分析报告")
print(result)

1.4 反思机制与自我纠错

1.4.1 为什么需要反思

大模型在推理过程中可能出现以下问题：

幻觉：生成不准确的事实
遗漏：漏掉关键信息或步骤
逻辑错误：推理链条断裂

反思机制（Reflection）让Agent能够审视自己的输出，发现并纠正错误。

1.4.2 反思的工作流程

Agent生成回答
    │
    ▼
思考者（Self-Critic）：检查回答的正确性
    │
    ├── 发现问题 ──→ 修正回答 ──→ 再次检查
    │
    └── 通过检查 ──→ 输出最终结果

1.4.3 代码实现

python

from openai import OpenAI
import json

client = OpenAI()

class ReflectiveAgent:
    """带反思机制的Agent"""
    
    def __init__(self, model="gpt-4"):
        self.model = model
        self.history = []
    
    def generate(self, prompt: str) -> str:
        """生成初步回答"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    
    def reflect(self, question: str, answer: str) -> dict:
        """反思评估回答质量"""
        reflection_prompt = f"""问题：{question}
回答：{answer}

请评估这个回答，检查：
1. 是否存在事实错误？
2. 是否遗漏了重要信息？
3. 逻辑是否严谨？
4. 回答是否完整？

输出JSON格式：
{{
    "has_issues": true/false,
    "issues": ["问题1", "问题2"],
    "suggestions": ["改进建议1", "改进建议2"],
    "score": 0-100
}}
"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": reflection_prompt}],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def improve(self, question: str, original: str, feedback: dict) -> str:
        """根据反馈改进回答"""
        improvement_prompt = f"""原始问题：{question}
原始回答：{original}
发现的问题：{feedback.get('issues', [])}
改进建议：{feedback.get('suggestions', [])}

请根据上述反馈，提供一个经过改进的完整回答。
"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": improvement_prompt}]
        )
        return response.choices[0].message.content
    
    def run(self, question: str, max_reflections: int = 3) -> str:
        """带反思循环的完整执行"""
        answer = self.generate(question)
        self.history.append(("initial", answer))
        
        for i in range(max_reflections):
            print(f"\n--- 反思轮次 {i+1} ---")
            feedback = self.reflect(question, answer)
            
            print(f"评分: {feedback.get('score')}")
            print(f"问题: {feedback.get('issues', [])}")
            
            if not feedback.get("has_issues", True):
                print("✅ 回答通过反思检查！")
                return answer
            
            answer = self.improve(question, answer, feedback)
            self.history.append((f"refine_{i+1}", answer))
        
        print("⚠️ 已达到最大反思轮次")
        return answer

# 使用示例
agent = ReflectiveAgent()
result = agent.run(
    "请解释大模型中的注意力机制（Attention），"
    "包括它的基本原理和在Transformer中的应用。"
)
print(f"\n=== 最终回答 ===\n{result}")

1.4.4 反思的高级模式

模式	描述	适用场景
单轮反思	生成→检查→修正一次	简单纠错
多轮反思	反复检查直到满意	高质量要求
外部评估	用另一个LLM或规则评估	客观性要求高
结构化反思	按检查清单逐项评估	领域专业任务

1.5 Agent评测与能力边界

1.5.1 评测维度

评测维度	说明	评测指标
任务完成率	Agent是否成功完成任务	成功率（Success Rate）
执行效率	完成任务的速度和资源消耗	步数、Token消耗、耗时
鲁棒性	面对异常输入的处理能力	错误恢复率
工具使用	是否准确调用工具和解析结果	工具调用准确率
记忆准确度	是否准确回忆历史信息	信息召回率
安全性	是否产生有害输出或被注入	安全率

1.5.2 评估工具与基准

基准/框架	说明	官网
GAIA	通用AI助手评测基准	https://huggingface.co/gaia-benchmark
AgentBench	多维度Agent能力评测	https://github.com/THUDM/AgentBench
SWE-bench	软件工程Agent评测	https://www.swebench.com/
ToolBench	工具使用能力评测	https://github.com/OpenBMB/ToolBench

1.5.3 评测代码示例

python

import time
from typing import Callable, Dict, List

class AgentEvaluator:
    """Agent评测器"""
    
    def __init__(self):
        self.results = []
    
    def evaluate(
        self,
        agent_func: Callable,
        test_cases: List[Dict],
        timeout: int = 60
    ) -> Dict:
        """运行评测"""
        summary = {
            "total": len(test_cases),
            "success": 0,
            "failed": 0,
            "timeout": 0,
            "total_tokens": 0,
            "total_time": 0
        }
        
        for case in test_cases:
            start_time = time.time()
            try:
                result = agent_func(case["input"])
                elapsed = time.time() - start_time
                
                is_success = self._check_result(result, case["expected"])
                if is_success:
                    summary["success"] += 1
                else:
                    summary["failed"] += 1
                    
                self.results.append({
                    "case": case["name"],
                    "success": is_success,
                    "time": elapsed
                })
                
            except Exception as e:
                summary["failed"] += 1
                self.results.append({
                    "case": case["name"],
                    "success": False,
                    "error": str(e)
                })
        
        summary["success_rate"] = summary["success"] / summary["total"] * 100
        return summary
    
    def _check_result(self, result, expected) -> bool:
        """检查结果是否符合预期（简化版）"""
        if isinstance(expected, str):
            return expected.lower() in str(result).lower()
        elif callable(expected):
            return expected(result)
        return result == expected
    
    def report(self) -> str:
        """生成评测报告"""
        success_count = sum(1 for r in self.results if r["success"])
        total = len(self.results)
        avg_time = sum(r.get("time", 0) for r in self.results) / total if total else 0
        
        report = f"""
## Agent评测报告

### 总览
- 用例总数: {total}
- 通过: {success_count}
- 失败: {total - success_count}
- 成功率: {success_count/total*100:.1f}%
- 平均耗时: {avg_time:.2f}s

### 详细结果
"""
        for r in self.results:
            status = "✅" if r["success"] else "❌"
            report += f"- {status} {r['case']}"
            if "time" in r:
                report += f" ({r['time']:.2f}s)"
            if "error" in r:
                report += f" - 错误: {r['error']}"
            report += "\n"
        
        return report

# 使用示例
evaluator = AgentEvaluator()

test_cases = [
    {"name": "天气查询", "input": "北京天气", "expected": "25"},
    {"name": "数学计算", "input": "计算 15 + 27", "expected": "42"},
    {"name": "知识问答", "input": "AI Agent的中文全称", "expected": "智能体"}
]

# 模拟一个Agent函数
def my_agent(input_text: str) -> str:
    # 实际评测时替换为真实Agent
    responses = {"北京天气": "北京今天晴天，25°C", "计算 15 + 27": "42"}
    return responses.get(input_text, "未知输入")

summary = evaluator.evaluate(my_agent, test_cases)
print(evaluator.report())

1.5.4 Agent的能力边界

Agent虽然强大，但存在固有的能力边界：

边界	说明	缓解策略
上下文窗口限制	无法处理超长历史	滑动窗口、摘要压缩
幻觉	生成不准确的信息	添加验证步骤、RAG
工具安全性	可能执行危险操作	权限控制、人工审批
成本	多次LLM调用增加成本	Token优化、缓存
鲁棒性	对输入变化敏感	Prompt工程设计
可解释性	决策过程不够透明	详细日志、思维链展示

小结与练习

知识小结

Agent三大核心能力：规划（任务分解）、记忆（短期/长期/工作记忆）、工具（Function Calling）
ReAct框架：思考→行动→观察循环，推理与行动交替进行
Plan-and-Execute模式：先制定完整计划再逐步执行，适合长任务
反思机制：通过自我评估和修正提升回答质量
Agent评测：从任务完成率、效率、鲁棒性等多维度评估

练习

练习1：实现一个带记忆的Agent

实现一个Agent，它能记住用户在前几轮对话中提到的信息
（如姓名、偏好），并在后续对话中正确回忆和使用这些信息。

练习2：实现ReAct Agent

实现一个ReAct Agent，能够执行以下操作：
1. 搜索网络信息（模拟）
2. 执行数学计算
3. 基于搜索结果和计算结果回答用户问题

练习3：反思机制实战

为你的Agent添加反思机制：
1. 生成回答后让另一个LLM实例检查错误
2. 如果发现问题，自动修正
3. 最多反思3轮，输出最终结果

练习4：Agent评测

设计5个测试用例用于评测你的Agent，然后：
1. 使用文本评测器运行评测
2. 分析失败原因
3. 根据评测结果优化Agent

下一章预告：章节2-多智能体系统 - 从单个Agent到多Agent协作的进阶之路

章节1：Agent核心原理 ​

本章导学 ​

学习目标 ​

1.1 Agent三大核心能力 ​

1.1.1 什么是AI Agent ​

1.1.2 三大核心能力详解 ​

（1）规划（Planning） ​

（2）记忆（Memory） ​

（3）工具（Tools） ​

1.2 ReAct推理-行动框架 ​

1.2.1 什么是ReAct ​

1.2.2 ReAct工作流程 ​

1.2.3 ReAct代码实现 ​

1.2.4 ReAct vs 纯推理 vs 纯行动 ​

1.3 Plan-and-Execute模式 ​

1.3.1 模式概述 ​

1.3.2 与ReAct的区别 ​

1.3.3 代码实现 ​

1.4 反思机制与自我纠错 ​

1.4.1 为什么需要反思 ​

1.4.2 反思的工作流程 ​

1.4.3 代码实现 ​

1.4.4 反思的高级模式 ​

1.5 Agent评测与能力边界 ​

1.5.1 评测维度 ​

1.5.2 评估工具与基准 ​

1.5.3 评测代码示例 ​

1.5.4 Agent的能力边界 ​

小结与练习 ​

知识小结 ​

练习 ​

练习1：实现一个带记忆的Agent ​

练习2：实现ReAct Agent ​

练习3：反思机制实战 ​

练习4：Agent评测 ​

章节1：Agent核心原理

本章导学

学习目标

1.1 Agent三大核心能力

1.1.1 什么是AI Agent

1.1.2 三大核心能力详解

（1）规划（Planning）

（2）记忆（Memory）

（3）工具（Tools）

1.2 ReAct推理-行动框架

1.2.1 什么是ReAct

1.2.2 ReAct工作流程

1.2.3 ReAct代码实现

1.2.4 ReAct vs 纯推理 vs 纯行动

1.3 Plan-and-Execute模式

1.3.1 模式概述

1.3.2 与ReAct的区别

1.3.3 代码实现

1.4 反思机制与自我纠错

1.4.1 为什么需要反思

1.4.2 反思的工作流程

1.4.3 代码实现

1.4.4 反思的高级模式

1.5 Agent评测与能力边界

1.5.1 评测维度

1.5.2 评估工具与基准

1.5.3 评测代码示例

1.5.4 Agent的能力边界

小结与练习

知识小结

练习

练习1：实现一个带记忆的Agent

练习2：实现ReAct Agent

练习3：反思机制实战

练习4：Agent评测