从零开始的LLM 5.智能体经典范式构建

发表于 更新于

前言

从零开始学习ai文章系列已完成《动手学深度学习》和《磨菇书》两本书的学习，新开的LLM系列课本来自《Happy-LLM》和《Hello-Agents》，但是内容和排版等个人重新进行整理，因此不会按照原来课本中的章节来写。如果哪里有错误的欢迎指正。或者不清晰的可以直接查看原文部分。

本章内容参考自 第四章 智能体经典范式构建 。原文已经写的很好了，因此这里仅作个人总结笔记记录。

1. 概述

在介绍智能体（Agent）之前，需要先理解两个重要概念：范式（Paradigm） 与 提示词工程（Prompt Engineering）。

1.1 什么是范式

范式（Paradigm）可以理解为解决某类问题的通用模式或设计框架。

在软件开发中，我们熟悉的面向对象、函数式编程都属于编程范式；而在智能体领域，范式则指：

如何组织大模型、工具、记忆以及决策流程，使其能够完成特定任务的一套方法论。

例如：

  • ReAct（Reason + Act）范式
  • Plan-and-Execute（规划-执行）范式
  • Reflection（反思）范式
  • Multi-Agent（多智能体协作）范式

这些范式并不会改变大模型本身，而是在大模型外部构建不同的工作流程，从而获得不同的能力表现。

简单来说：

大模型决定智能体的能力上限，而范式决定智能体如何发挥这些能力。

1.2 什么是提示词工程

提示词工程（Prompt Engineering）是指通过设计输入给大模型的提示词（Prompt），引导模型按照预期方式进行思考和输出。

由于大模型本质上是一个根据上下文预测下一个 Token 的概率模型，因此输入内容会直接影响最终结果。

一个简单的问题：

1
北京到上海有多远？

与：

1
你是一名资深旅游顾问，请从高铁、飞机、自驾三个角度分析北京到上海的距离和出行方案。

模型给出的回答质量通常会存在明显差异。

因此，提示词工程的核心目标是：

  • 提高回答质量
  • 降低模型幻觉
  • 引导模型按照指定格式输出
  • 让模型具备特定角色和行为模式

1.3 范式与提示词工程的关系

范式定义了智能体应该按照什么流程完成任务，而提示词工程则负责将这一流程描述给大模型。

换句话说：

范式负责设计流程，提示词负责让大模型执行流程。

因此，大部分智能体范式都可以理解为：

  1. 先设计一套任务处理流程（范式）；
  2. 再通过提示词将该流程描述给大模型；
  3. 最终让大模型按照该流程进行思考和决策。

从某种程度上来说，许多早期 Agent 系统本质上就是一组精心设计的提示词。随着发展，人们又在此基础上加入了工具调用、记忆管理、状态维护以及工作流控制等机制，最终形成了今天的智能体系统。

1.4 本章内容

理解了范式与提示词工程之间的关系后，接下来将介绍智能体领域最经典的三种范式。

这些范式本质上都是通过不同的流程设计，引导大模型以不同方式完成任务，从而获得推理、规划、反思等能力。

本章将依次介绍：

  • ReAct（Reason + Act）范式：通过“思考 → 行动 → 观察”的循环过程完成任务，是当前最经典、应用最广泛的 Agent 范式之一。
  • Plan-and-Execute（规划-执行）范式：先生成完整任务计划，再按照计划逐步执行，适用于复杂任务拆解与长期任务管理。
  • Reflection（反思）范式：在完成阶段性任务后进行自我审查和总结，根据发现的问题调整后续行动，从而持续提升任务完成质量。

这三种范式构成了现代智能体设计的重要基础思想，许多 Agent 框架和研究工作都可以看作是在这些经典范式之上的扩展与组合。

2. 环境准备

接下来的内容将沿用原文的讲解方式，通过「代码实践 + 原理分析」的形式逐步介绍各种智能体范式。因此，在正式开始之前，需要先完成一些基础环境配置。

2.1 配置 API 密钥

为了提高代码的通用性，我们将模型服务相关配置统一放置在环境变量中，包括：

  • API Key
  • 模型 ID
  • 服务地址（Base URL）

这样在切换不同模型服务商时，只需要修改配置文件，而无需改动业务代码。

首先，在项目根目录下创建一个名为 .env 的文件，并填写对应配置：

1
2
3
4
# .env
LLM_API_KEY="gsk_GF----po"
LLM_MODEL_ID="llama-3.3-70b-versatile"
LLM_BASE_URL="https://api.groq.com/openai/v1"

上述配置仅为示例，你可以根据自己的需求替换为：

  • OpenAI 官方接口
  • 本地部署的模型服务
  • 兼容 OpenAI API 的第三方服务

本文示例中，作者使用的是 Groq 提供的免费模型服务，因此配置内容指向 Groq 的 API 接口。

配置完成后，将 .env 文件放置在运行脚本所在目录（或项目根目录）下，并在程序启动时加载环境变量：

1
2
3
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()

这样后续代码便可以直接从环境变量中读取模型相关配置，而无需在代码中硬编码敏感信息。

2.2 封装基础 LLM 调用函数

为了简化后续 Agent 开发，我们首先对大模型调用过程进行统一封装。

这里与原文保持一致，采用在线 API 的方式调用大模型。这样能够更专注于理解 Agent 范式本身，而无需额外关注模型部署与推理细节。

如果希望使用本地模型，可以参考上一章介绍的模型加载方法，将本地模型封装成与 OpenAI API 类似的调用接口即可。对于后续 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

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量
load_dotenv()

class HelloAgentsLLM:
    """
    为本书 "Hello Agents" 定制的LLM客户端。
    它用于调用任何兼容OpenAI接口的服务，并默认使用流式响应。
    """
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化客户端。优先使用传入参数，如果未提供，则从环境变量加载。
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

        if not all([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> None:
        """
        调用大语言模型进行思考，并返回其响应。
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )

            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)
                collected_content.append(content)
            print()  # 在流式输出结束后换行
            return "".join(collected_content)

        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None

# --- 客户端使用示例 ---
if __name__ == '__main__':
    try:
        llmClient = HelloAgentsLLM()

        exampleMessages = [
            {"role": "system", "content": "You are a helpful assistant that writes Python code."},
            {"role": "user", "content": "写一个快速排序算法"}
        ]

        print("--- 调用LLM ---")
        responseText = llmClient.think(exampleMessages)
        if responseText:
            print("\n\n--- 完整模型响应 ---")
            print(responseText)

    except ValueError as e:
        print(e)

2.3 配置 SerpApi API 密钥

在后续实现 Agent 工具调用时，我们会使用 SerpApi 作为搜索引擎接口。因此，需要提前申请并配置对应的 API 密钥。

首先前往 SerpApi 官网 注册账户。SerpApi 提供一定额度的免费调用次数，对于学习和实验场景通常已经足够使用。

注册完成后，在控制台获取你的 API Key，并将其添加到项目根目录下的 .env 文件中：

1
SERPAPI_API_KEY="YOUR_SERPAPI_API_KEY"

3. ReAct（Reason + Act）范式

3.1 ReAct 的工作流程

ReAct（Reason + Act）由 Shunyu Yao 等人在 2022 年提出。其核心思想是模仿人类解决问题的过程，将推理（Reasoning）与行动（Acting）显式结合，使大语言模型能够在思考过程中主动获取外部信息，而不仅仅依赖预训练阶段学习到的知识。

传统的大语言模型通常采用一次性生成（One-shot Generation）的方式：模型读取用户问题后直接给出答案。

而 ReAct 则引入了一个持续迭代的决策过程，使模型能够在推理过程中动态调用工具、获取反馈并调整后续行为。

整个流程可以概括为一个循环：

1
Thought → Action → Observation

其中：

  • Thought（思考）

    智能体的内部推理过程。模型会分析当前任务、制定计划、评估已有信息，并决定下一步需要采取什么行动。

  • Action（行动）

    智能体执行的具体操作，通常表现为调用外部工具。例如：

    1
    Search["华为最新款手机"]

  • Observation（观察）

    工具执行后返回的结果。例如搜索引擎返回的网页摘要、数据库查询结果或 API 响应等。

一个典型的执行过程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
用户问题：
    华为最新发布的手机是哪一款？

循环执行：

    Thought：
        分析当前信息是否足够回答问题

    如果信息不足：
        Action：
            调用搜索工具查询相关信息

        Observation：
            获取搜索结果

        将本次 Action 和 Observation 加入历史记录

    如果信息已经足够：
        输出 Final Answer
        结束循环

随着循环不断进行，新的 Observation 会被追加到上下文中，成为后续推理的依据。模型通过不断地“思考—行动—观察”，逐步缩小问题空间，最终得到答案。

从形式化角度来看，在时间步 t 时，大语言模型策略 \(\pi\) 会根据用户问题 q 以及历史轨迹 \(((a_1,o_1),(a_2,o_2),...,(a_{t-1},o_{t-1}))\) 生成当前的思考 \(th_t\) 和行动 \(a_t\)： \[ (th_t,a_t)=\pi(q,((a_1,o_1),...,(a_{t-1},o_{t-1}))) \] 随后，环境中的工具 T 执行行动 \(a_t\)，并返回新的观察结果： \[ o_t=T(a_t) \] 新的 \((a_t,o_t)\) 将被追加到历史轨迹中，供下一轮推理使用。该过程持续迭代，直到模型认为已经获得足够信息，并输出最终答案。

相比于单纯依赖参数记忆的大语言模型，ReAct 的优势在于能够将模型推理能力与外部工具能力结合起来，从而突破知识时效性和计算能力的限制。

因此，ReAct 特别适用于以下场景：

  • 需要实时信息的任务

    例如查询天气、新闻、股票价格、航班状态等动态变化的信息。

  • 需要外部知识支撑的任务

    例如搜索专业领域资料、阅读文档、检索企业内部知识库等。

  • 需要精确计算的任务

    将数学运算交给计算器或代码执行器完成，避免大语言模型产生计算错误。

  • 需要调用系统能力的任务

    例如访问数据库、调用 API、执行代码、发送邮件或操作文件系统等。

可以发现，ReAct 的本质并不是让模型“知道更多”，而是让模型学会在不知道答案时主动寻找答案。

接下来，我们将构建一个具备外部工具调用能力的 ReAct 智能体，并让其解决一个仅依赖模型参数无法准确回答的问题。例如：

华为最新发布的手机是哪一款？它的主要卖点是什么？

为了回答这个问题，智能体需要首先意识到自身缺乏实时信息，然后主动调用搜索工具获取最新资料，最后基于检索结果完成总结与回答。这正是 ReAct 范式最典型的应用场景。

3.2 工具的定义与实现

如果将大语言模型看作智能体的“大脑”，那么工具（Tools）就是其与外部世界交互的能力。通过工具，智能体不仅能够利用自身的推理能力，还能够获取实时信息、执行计算任务，以及与各种外部系统进行交互。

对于本节的示例任务——回答“华为最新发布的手机是哪一款？它的主要卖点是什么？”，仅依赖模型参数中的知识是不够的。由于手机产品会不断更新，大语言模型掌握的信息很可能已经过时。因此，我们需要为智能体提供一个能够访问互联网的搜索工具。

这里选择使用 SerpApi。它提供了搜索引擎结果的结构化 API 接口，可以直接获取搜索结果，而无需自行实现网页抓取和解析逻辑。

首先安装相关依赖：

1
pip install google-search-results

接下来，我们通过代码实现并管理这个工具。

3.2.1 实现搜索工具的核心逻辑

一个工具通常包含以下三个组成部分：

  1. 名称（Name）

    工具的唯一标识符，用于在 Action 中指定需要调用的工具。

  2. 描述（Description）

    对工具功能的自然语言说明，用于告诉大语言模型该工具能够完成什么任务，以及适用于哪些场景。

    在 ReAct 模式下，模型通常会根据工具描述判断是否需要调用该工具，因此清晰准确的描述十分重要。

  3. 执行逻辑（Execution Logic）

    工具对应的具体实现代码，负责接收输入、执行任务并返回结果。

我们的第一个工具是搜索工具（Search Tool）。它接收一个查询字符串作为输入，通过 SerpApi 调用搜索引擎获取结果，并将搜索内容返回给智能体。

下面实现搜索工具的核心逻辑 search 函数：

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
from serpapi import SerpApiClient
import os

def search(query: str, num_results: int = 7) -> str:
    """
    基于 SerpApi 的网页搜索工具。
    返回结构化搜索结果，包含直接答案和有机结果列表（含 URL）。
    """
    print(f"🔍 正在执行 [SerpApi] 网页搜索: {query}")
    try:
        api_key = os.getenv("SERPAPI_API_KEY")
        if not api_key:
            return "错误: SERPAPI_API_KEY 未配置。"

        params = {
            "engine": "google",
            "q": query,
            "api_key": api_key,
            "gl": "cn",
            "hl": "zh-cn",
            "num": num_results,
        }

        client = SerpApiClient(params)
        results = client.get_dict()

        output_parts = []

        # 1. answer_box：结构不固定，兼容三种形式
        if "answer_box" in results:
            box = results["answer_box"]
            direct = box.get("answer") or box.get("snippet") or box.get("result", "")
            source = box.get("link", "")
            if direct:
                src_str = f"\n来源: {source}" if source else ""
                output_parts.append(f"[直接答案]{src_str}\n{direct}")

        # 2. knowledge_graph：实体类查询的结构化信息
        if "knowledge_graph" in results:
            kg = results["knowledge_graph"]
            desc = kg.get("description", "")
            website = kg.get("website", "")
            attrs = kg.get("attributes", {})
            if desc:
                kg_lines = [f"[知识图谱] {kg.get('title', '')}", desc]
                if website:
                    kg_lines.append(f"官网: {website}")
                if attrs:
                    attr_str = "  ".join(f"{k}: {v}" for k, v in list(attrs.items())[:4])
                    kg_lines.append(attr_str)
                output_parts.append("\n".join(kg_lines))

        # 3. organic_results：保留 link，供 agent 决定是否深入 fetch
        if "organic_results" in results and results["organic_results"]:
            snippets = [
                f"[{i+1}] {res.get('title', '')}\nURL: {res.get('link', '无')}\n{res.get('snippet', '')}"
                for i, res in enumerate(results["organic_results"][:num_results])
            ]
            output_parts.append("\n\n".join(snippets))

        # # 4. related_questions：People Also Ask，含现成 snippet，可作子查询
        # if "related_questions" in results and results["related_questions"]:
        #     rq_lines = ["[相关问题 - 可作为子查询]"]
        #     for rq in results["related_questions"][:3]:
        #         q = rq.get("question", "")
        #         s = rq.get("snippet", "")
        #         l = rq.get("link", "")
        #         if q:
        #             rq_lines.append(f"Q: {q}")
        #             if s:
        #                 rq_lines.append(f"   {s}")
        #             if l:
        #                 rq_lines.append(f"   来源: {l}")
        #     output_parts.append("\n".join(rq_lines))
        #
        # # 5. related_searches：扩词用，agent 可用于判断是否换词重搜
        # if "related_searches" in results and results["related_searches"]:
        #     rs_queries = [r.get("query", "") for r in results["related_searches"][:5] if r.get("query")]
        #     if rs_queries:
        #         output_parts.append("[相关搜索词 - 可用于换词重搜]\n" + "  |  ".join(rs_queries))

        if output_parts:
            return "\n\n---\n\n".join(output_parts)

        return f"没有找到关于 '{query}' 的信息。"

    except Exception as e:
        return f"搜索时发生错误: {e}"

3.2.2 构建通用的工具执行器

目前我们只实现了一个搜索工具，但在实际应用中，智能体往往需要使用多种工具，例如计算器、数据库查询工具、代码执行工具等。

为了统一管理这些工具，我们创建一个 ToolExecutor 类。它负责工具的注册、查询与调用，使智能体无需关心工具的具体实现细节，只需指定工具名称和输入参数即可完成调用。

下面实现通用工具执行器：

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
from typing import Dict, Any

class ToolExecutor:
    """
    一个工具执行器，负责管理和执行工具。
    """
    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}

    def registerTool(self, name: str, description: str, func: callable):
        """
        向工具箱中注册一个新工具。
        """
        if name in self.tools:
            print(f"警告:工具 '{name}' 已存在，将被覆盖。")
        self.tools[name] = {"description": description, "func": func}
        print(f"工具 '{name}' 已注册。")

    def getTool(self, name: str) -> callable:
        """
        根据名称获取一个工具的执行函数。
        """
        return self.tools.get(name, {}).get("func")

    def getAvailableTools(self) -> str:
        """
        获取所有可用工具的格式化描述字符串。
        """
        return "\n".join([
            f"- {name}: {info['description']}" 
            for name, info in self.tools.items()
        ])

3.2.3 测试

现在，我们将搜索工具注册到 ToolExecutor 中，并模拟一次工具调用，以验证整个工具调用流程是否能够正常工作。

下面进行测试：

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
# --- 工具初始化与使用示例 ---
if __name__ == '__main__':
    # 1. 初始化工具执行器
    toolExecutor = ToolExecutor()

    # 2. 注册我们的实战搜索工具
    search_description = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。"
    toolExecutor.registerTool("Search", search_description, search)

    # 3. 打印可用的工具
    print("\n--- 可用的工具 ---")
    print(toolExecutor.getAvailableTools())

    # 4. 智能体的Action调用，这次我们问一个实时性的问题
    print("\n--- 执行 Action: Search['英伟达最新的GPU型号是什么'] ---")
    tool_name = "Search"
    tool_input = "英伟达最新的GPU型号是什么"

    tool_function = toolExecutor.getTool(tool_name)
    if tool_function:
        observation = tool_function(tool_input)
        print("--- 观察 (Observation) ---")
        print(observation)
    else:
        print(f"错误:未找到名为 '{tool_name}' 的工具。")

3.3 ReAct 智能体的编码实现

前面我们已经完成了两个核心组件：

  • LLM 客户端：负责与大语言模型进行交互；
  • ToolExecutor：负责管理和执行工具。

接下来，我们将这些组件组合起来，构建一个完整的 ReAct 智能体。整个智能体的核心职责包括：

  1. 构造提示词并发送给 LLM；
  2. 解析 LLM 返回的 Thought 和 Action；
  3. 调用对应工具执行 Action；
  4. 将执行结果作为 Observation 加入历史记录；
  5. 重复上述过程，直到获得最终答案。

为了便于理解，我们将 ReActAgent 的实现拆分为几个部分进行介绍。

3.3.1 系统提示词设计

提示词是 ReAct 智能体的核心，它决定了模型如何思考以及如何调用工具。

为了让模型按照 ReAct 的方式工作，我们需要通过系统提示词明确告知：

  • 智能体的角色定位；
  • 当前可用的工具列表；
  • 输出格式要求；
  • 用户问题与历史记录。

下面定义 ReAct 使用的提示词模板：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ReAct 提示词模板
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下:
{tools}

请严格按照以下格式进行回应:

Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一:
- `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
- `Finish[最终答案]`:当你认为已经获得最终答案时。
- 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。

现在，请开始解决以下问题:
Question: {question}
History: {history}
"""

该模板主要包含以下几个部分：

  • 角色定义

    指定模型扮演一个能够使用外部工具的智能助手。

  • 工具列表（{tools}）

    动态注入当前可用工具及其描述，帮助模型判断应该调用哪个工具。

  • 输出格式约束

    要求模型按照固定格式输出 Thought 和 Action，便于后续程序解析。

  • 上下文信息（{question}/{history}）

    包含用户问题以及历史交互记录，使模型能够基于已有信息继续推理。

3.3.2 输出解析器的实现

大语言模型返回的内容本质上仍然是一段文本，因此我们需要从文本中提取出 Thought 和 Action。

为此，我们实现几个辅助解析函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def _parse_output(self, text: str):
    """解析LLM的输出，提取Thought和Action。
    """
    # Thought: 匹配到 Action: 或文本末尾
    thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)", text, re.DOTALL)
    # Action: 匹配到文本末尾
    action_match = re.search(r"Action:\s*(.*?)$", text, re.DOTALL)
    thought = thought_match.group(1).strip() if thought_match else None
    action = action_match.group(1).strip() if action_match else None
    return thought, action

def _parse_action(self, action_text: str):
    """解析Action字符串，提取工具名称和输入。
    """
    match = re.match(r"(\w+)\[(.*)\]", action_text, re.DOTALL)
    if match:
        return match.group(1), match.group(2)
    return None, None

其中：

  • **_parse_output** : 从模型输出中提取 Thought 和 Action。
  • **_parse_action** : 解析 Action 内容，获取工具名称以及工具输入参数。

3.3.3 核心循环的实现

ReActAgent 的核心是一个循环。

在每一轮中，智能体都会构造提示词、调用模型、解析结果、执行工具，并将新的 Observation 追加到历史记录中。这个过程会持续进行，直到模型给出最终答案或达到最大迭代次数。

下面实现 ReActAgent 的核心逻辑：

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
def run(self, question: str):
"""
运行ReAct智能体来回答一个问题。
"""
self.history = []  # 每次运行时重置历史记录
current_step = 0

while current_step < self.max_steps:
    current_step += 1
    print(f"--- 第 {current_step} 步 ---")

    # 1. 格式化提示词
    tools_desc = self.tool_executor.getAvailableTools()
    history_str = "\n".join(self.history)
    prompt = REACT_PROMPT_TEMPLATE.format(
        tools=tools_desc,
        question=question,
        history=history_str
    )

    # 2. 调用LLM进行思考
    messages = [{"role": "user", "content": prompt}]
    response_text = self.llm_client.think(messages=messages)

    if not response_text:
        print("错误:LLM未能返回有效响应。")
        break


    # 3. 解析LLM的输出
    thought, action = self._parse_output(response_text)

    if thought:
        print(f"思考: {thought}")

    if not action:
        print("警告:未能解析出有效的Action，流程终止。")
        break

    # 4. 执行Action
    if action.startswith("Finish"):
        # 如果是Finish指令，提取最终答案并结束
        final_answer = re.match(r"Finish\[(.*)\]", action).group(1)
        print(f"🎉 最终答案: {final_answer}")
        return final_answer

    tool_name, tool_input = self._parse_action(action)
    if not tool_name or not tool_input:
        # ... 处理无效Action格式 ...
        continue

    print(f"🎬 行动: {tool_name}[{tool_input}]")

    tool_function = self.tool_executor.getTool(tool_name)
    if not tool_function:
        observation = f"错误:未找到名为 '{tool_name}' 的工具。"
    else:
        observation = tool_function(tool_input)  # 调用真实工具


    print(f"👀 观察: {observation}")

    # 5. 将本轮的Action和Observation添加到历史记录中
    self.history.append(f"Action: {action}")
    self.history.append(f"Observation: {observation}")

    # 循环结束
print("已达到最大步数，流程终止。")
return None

3.3.4 运行实例与分析

至此，我们已经完成了完整的 ReAct 智能体实现。

下面运行一个实际示例，观察智能体如何通过调用搜索工具获取实时信息，并最终生成答案：

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 ReActAgent:
    def __init__(self, llm_client: HelloAgentsLLM, tool_executor: ToolExecutor, max_steps: int = 5):
        self.llm_client = llm_client
        self.tool_executor = tool_executor
        self.max_steps = max_steps
        self.history = []

    def run(self, question: str):
		...

    def _parse_output(self, text: str):
		...

    def _parse_action(self, action_text: str):
		...

if __name__ == '__main__':
    llm = HelloAgentsLLM()
    tool_executor = ToolExecutor()
    search_desc = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。"
    tool_executor.registerTool("Search", search_desc, search)
    agent = ReActAgent(llm_client=llm, tool_executor=tool_executor)
    question = "华为最新的手机是哪一款？它的主要卖点是什么？"
    agent.run(question)

（2026.6.24）

从运行结果可以看到，智能体会先调用搜索工具获取信息，再结合搜索结果完成推理并生成答案。

由于互联网信息会不断变化，你实际运行时得到的结果可能与本书中的示例有所不同。这也体现了 ReAct 范式的优势：能够利用外部工具获取实时信息，而不完全依赖模型参数中的静态知识。

3.4 ReAct 的特点、局限性与调试技巧

通过本章的实现过程，我们已经了解了 ReAct 智能体的工作方式。下面对其特点、局限性以及常见调试方法进行简单总结。

（1）ReAct 的主要特点

  1. 高可解释性

    ReAct 会显式输出 Thought 过程，使我们能够观察智能体的推理步骤，更容易理解和调试其行为。

  2. 动态决策能力

    智能体会根据每一步获得的 Observation 调整后续行动，而不是在任务开始时一次性生成完整计划。

  3. 工具协同能力

    ReAct 将 LLM 的推理能力与外部工具的执行能力结合起来，使模型能够获取实时信息、执行计算以及完成各种外部操作。

（2）ReAct 的局限性

  1. 依赖模型能力

    如果模型的推理能力、指令遵循能力或格式输出能力较弱，整个 ReAct 流程的稳定性也会受到影响。

  2. 执行效率较低

    一个任务通常需要多轮 LLM 调用和工具调用，因此相比直接生成答案会消耗更多时间和成本。

  3. 依赖提示词设计

    提示词的设计会直接影响模型行为，不同模型对提示词的响应也可能存在较大差异。

  4. 缺乏全局规划能力

    ReAct 更倾向于逐步决策，对于需要长期规划的复杂任务并不一定是最优选择。

（3）调试技巧

当 ReAct 智能体运行结果不符合预期时，可以从以下几个方面进行排查：

  • 检查完整 Prompt

    打印最终发送给模型的提示词，确认上下文和历史记录是否正确。

  • 查看模型原始输出

    当解析失败时，优先检查模型返回的原始文本，而不是直接修改解析逻辑。

  • 验证工具调用

    确认工具输入格式正确，并检查工具返回结果是否符合预期。

  • 增加 Few-shot 示例

    在提示词中加入正确示例，通常能够提升模型遵循格式的稳定性。

  • 调整模型与参数

    更换能力更强的模型，或降低 temperature 等参数，有时能够有效减少异常输出。

4. Plan-and-Execute（规划-执行）范式

在理解了 ReAct 这种“边思考边行动”的智能体范式之后，我们接下来介绍一种结构更加清晰的策略：Plan-and-Solve。

顾名思义，该方法将任务显式划分为两个阶段：先规划（Plan），后执行（Solve）。

如果说 ReAct 更像是一名侦探，在信息不断变化的过程中逐步推理并调整方向；那么 Plan-and-Solve 更像一名建筑师，在施工前完成完整设计图（Plan），再按图施工（Solve）。

这一范式的核心思想是：先形成全局规划，再进行局部执行。

4.1 Plan-and-Solve 的工作原理

Plan-and-Solve Prompting 由 Lei Wang 等人在 2023 年提出，其主要动机是缓解思维链在多步骤复杂任务中容易出现的“中途偏移”问题。

与 ReAct 将推理与行动交织进行不同，Plan-and-Solve 将整个过程拆分为两个相对独立的阶段：

（1）规划阶段（Planning Phase）

智能体首先接收完整问题，但不直接求解，而是生成一个分步骤的解决计划。该计划用于指导后续执行过程，本质上是一次对任务结构的显式建模。

形式化表示为： \[ P = \pi_{plan}(q) \] 其中：

  • q：用户输入问题
  • P：生成的执行计划（由多个步骤组成）

（2）执行阶段（Solving Phase）

在获得完整计划后，智能体进入执行阶段。此时模型会按照计划逐步完成每个子任务。

对于第 i 个步骤，其结果 \(s_i\) 的生成依赖于原始问题、完整计划以及之前的执行结果： \[ s_i = \pi_{solve}(q, P, (s_1, ..., s_{i-1})) \] 最终答案由最后一步结果 \(s_n\) 给出。

该流程可以概括为：

  • 先生成整体计划
  • 再按步骤逐步执行
  • 最终汇总得到答案

Plan-and-Solve 特别适用于结构清晰、可分解性强的任务，例如：

  • 多步数学问题

    先列出解题步骤，再逐步计算。

  • 结构化内容生成

    如报告写作，需要先确定整体结构（引言、分析、结论），再逐段生成内容。

  • 代码生成任务

    先设计模块与函数结构，再逐步实现具体功能。

4.2 规划阶段

为了体现 Plan-and-Solve 在结构化推理任务中的优势，本节不依赖工具调用，而是通过提示词设计完成一个纯推理任务。

这类任务的特点是无法通过单步计算直接得到答案，必须先将问题拆解为多个逻辑子步骤，再逐步求解。这正是 Plan-and-Solve 中“先规划，后执行”策略最适合的场景。

我们以如下问题为例：

一个水果店周一卖出了 15 个苹果。周二卖出的数量是周一的两倍。周三比周二少卖 5 个苹果。请问三天总共卖出了多少个苹果？

该问题本身并不复杂，但包含清晰的多步依赖关系，适合作为规划式推理的示例任务。

（1）规划阶段

首先将原始问题拆解为一组有序的子任务，例如：

  • 计算周二的销量
  • 计算周三的销量
  • 计算三天总销量

这些步骤构成后续执行的“计划”。

（2）执行阶段

智能体严格按照规划好的步骤逐步执行计算，每一步都以前一步的结果作为输入，持续推进直到所有步骤完成，最终得到总结果。

下面是规划阶段使用的提示词模板：

1
2
3
4
5
6
7
8
9
10
11
12
PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""

该提示词通过以下方式保证输出稳定性：

  • 角色设定

    将模型定义为“专业的任务规划者”，引导其进行结构化拆解。

  • 任务约束

    明确要求模型只进行步骤分解，而不直接计算最终答案。

  • 格式约束

    强制输出为 Python list 结构，从而便于程序直接解析，无需复杂 NLP 处理。

接下来，我们将该提示词逻辑封装为一个 Planner 类，用于统一生成任务分解计划，这个类也构成了整个规划阶段的核心组件。

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
import ast

# 假定 llm_client.py 中的 HelloAgentsLLM 类已经定义好
# from llm_client import HelloAgentsLLM

class Planner:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def plan(self, question: str) -> list[str]:
        """
        根据用户问题生成一个行动计划。
        """
        prompt = PLANNER_PROMPT_TEMPLATE.format(question=question)

        # 为了生成计划，我们构建一个简单的消息列表
        messages = [{"role": "user", "content": prompt}]

        print("--- 正在生成计划 ---")
        # 使用流式输出来获取完整的计划
        response_text = self.llm_client.think(messages=messages) or ""

        print(f"✅ 计划已生成:\n{response_text}")

        # 解析LLM输出的列表字符串
        try:
            # 找到```python和```之间的内容
            plan_str = response_text.split("```python")[1].split("```")[0].strip()
            # 使用ast.literal_eval来安全地执行字符串，将其转换为Python列表
            plan = ast.literal_eval(plan_str)
            return plan if isinstance(plan, list) else []
        except (ValueError, SyntaxError, IndexError) as e:
            print(f"❌ 解析计划时出错: {e}")
            print(f"原始响应: {response_text}")
            return []
        except Exception as e:
            print(f"❌ 解析计划时发生未知错误: {e}")
            return []

4.3 执行器与状态管理-

在规划器（Planner）生成完整的执行计划后，就需要执行器（Executor）按照计划逐步完成每一个子任务。

与规划器不同，执行器不再负责拆解问题，而是专注于执行当前步骤。同时，它还需要维护整个任务的执行状态，将前面步骤的结果保存下来，并作为后续步骤的输入，从而保证整个推理过程能够连续进行。

因此，执行器的提示词需要包含以下几部分信息：

  • 原始问题：确保模型始终围绕最终目标进行推理。
  • 完整计划：帮助模型了解当前步骤在整个计划中的位置。
  • 历史执行结果：提供已经完成步骤的结果，作为当前步骤的输入。
  • 当前步骤：明确本轮需要完成的具体任务。

下面定义执行器使用的提示词模板：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""

可以看到，相较于 ReAct 的提示词，这里新增了 完整计划（{plan}）。这是规划阶段生成的结果，也是 Plan-and-Solve 与 ReAct 最主要的区别之一：模型在执行之前已经拥有完整的任务规划，而不是边推理边决定下一步行动。

接下来，我们将执行逻辑封装到 Executor 类中。该类负责遍历规划得到的步骤列表，依次调用大语言模型完成每一步任务，并维护整个执行过程中的状态信息。

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

class Executor:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def execute(self, question: str, plan: list[str]) -> str:
        """
        根据计划，逐步执行并解决问题。
        """
        history = ""  # 用于存储历史步骤和结果的字符串

        print("\n--- 正在执行计划 ---")

        for i, step in enumerate(plan):
            print(f"\n-> 正在执行步骤 {i + 1}/{len(plan)}: {step}")

            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question=question,
                plan=plan,
                history=history if history else "无",  # 如果是第一步，则历史为空
                current_step=step
            )

            messages = [{"role": "user", "content": prompt}]

            response_text = self.llm_client.think(messages=messages) or ""

            # 更新历史记录，为下一步做准备
            history += f"步骤 {i + 1}: {step}\n结果: {response_text}\n\n"

            print(f"✅ 步骤 {i + 1} 已完成，结果: {response_text}")

        # 循环结束后，最后一步的响应就是最终答案
        final_answer = response_text
        return final_answer

由于本节示例仅涉及逻辑推理，不需要调用任何外部工具，因此执行器只负责调用 LLM 完成各个步骤的推理。

至此，我们已经分别实现了负责规划的 Planner 和负责执行的 Executor。

最后，我们将二者组合成一个统一的 PlanAndSolveAgent，用于完成整个 Plan-and-Solve 流程。该类负责接收 LLM 客户端，初始化规划器和执行器，并提供统一的 run 方法作为智能体的入口。

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
class PlanAndSolveAgent:
    def __init__(self, llm_client):
        """
        初始化智能体，同时创建规划器和执行器实例。
        """
        self.llm_client = llm_client
        self.planner = Planner(self.llm_client)
        self.executor = Executor(self.llm_client)

    def run(self, question: str):
        """
        运行智能体的完整流程:先规划，后执行。
        """
        print(f"\n--- 开始处理问题 ---\n问题: {question}")

        # 1. 调用规划器生成计划
        plan = self.planner.plan(question)

        # 检查计划是否成功生成
        if not plan:
            print("\n--- 任务终止 --- \n无法生成有效的行动计划。")
            return

        # 2. 调用执行器执行计划
        final_answer = self.executor.execute(question, plan)

        print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")

PlanAndSolveAgent 本身并不负责具体的规划或执行，而是协调 Planner 与 Executor 两个组件完成整个任务流程，使各模块职责更加清晰，也便于后续扩展和维护。

4.4 运行实例与分析

下面给出 main 函数部分，通过创建 PlanAndSolveAgent 实例并调用 run() 方法即可完成整个推理流程。

1
2
3
4
5
6
7
8
9
10
# --- 5. 主函数入口 ---
if __name__ == '__main__':
    try:
        llm_client = HelloAgentsLLM()
        agent = PlanAndSolveAgent(llm_client)
        question = "一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果？"
        agent.run(question)
    except ValueError as e:
        print(e)

image-20260626011214809

从运行结果可以看到，Plan-and-Solve 将整个任务划分为规划和执行两个独立阶段。

首先，Planner 根据原始问题生成了一份结构化的执行计划，将复杂问题拆解为多个有序的子任务。随后，Executor 严格按照计划依次执行每个步骤，并将前一步的执行结果作为后续步骤的输入，使整个推理过程能够连续进行。

最终，智能体顺利完成所有步骤，并得到正确答案 70。

相比于直接一步生成答案，Plan-and-Solve 将复杂推理拆分为多个简单步骤，使整个求解过程更加清晰，也降低了长链路推理过程中出现偏差的可能性。

5. Reflection（反思）范式

在前面介绍的 ReAct 和 Plan-and-Solve 范式中，智能体完成任务后便结束整个执行流程。然而，模型生成的结果并不总是正确，可能存在事实错误、逻辑漏洞或表达不够完善等问题。

Reflection 的核心思想，就是让智能体在完成任务后对自己的输出进行检查与修正，形成一个"执行—反思—优化"的闭环，不断提升最终结果的质量。

5.1 Reflection 机制的核心思想

Reflection 的思想来源于人类解决问题的过程：完成一份初稿后会进行校对，解出一道数学题后会进行验算，根据发现的问题不断修改，直到得到更满意的结果。

这一思想在多个研究中得到了体现，例如 Shinn 等人在 2023 年提出的 Reflexion 框架。其整体流程可以概括为三个阶段：

（1）执行

智能体首先完成一次任务，生成初始结果。这里的执行过程可以采用前面介绍的 ReAct、Plan-and-Solve，或其他 Agent 范式。

（2）反思

随后，智能体开始审查自己的输出，对执行结果进行分析，并生成改进建议。

反思通常会重点关注以下几个方面：

  • 是否存在事实错误；
  • 推理过程是否合理；
  • 是否遗漏了关键条件；
  • 是否存在更优的解决方案。

最终，反思阶段会输出一份反馈（Feedback），指出当前结果存在的问题以及改进方向。

其形式化表示为： \[ F_i=\pi_{reflect}(Task,O_i) \] 其中：

  • Task：原始任务；
  • \(O_i\)：第 \(i\) 次迭代产生的结果；
  • \(F_i\)：针对当前结果生成的反馈。

（3）优化

获得反馈后，智能体会结合原始任务、当前结果以及反馈信息，对答案进行重新生成，得到新的输出。

形式化表示为： \[ O_{i+1}=\pi_{refine}(Task,O_i,F_i) \] 新的结果随后又可以进入下一轮反思，从而形成持续迭代的优化过程，直到达到预设的迭代次数，或模型认为结果已经足够完善。

相比于前面介绍的两种范式，Reflection 最大的特点在于为智能体增加了一条内部反馈回路，使其能够主动发现并修正自身的问题，而不是一次生成答案后立即结束任务。

对于复杂推理、代码生成、文档写作等任务，这种多轮自我修正机制通常能够获得比单次生成更高质量的结果。

5.2 案例设定与记忆模块设计

为了更直观地展示 Reflection 范式在实际任务中的工作流程，本节将结合一个代码生成案例，引入一个简单的短期记忆（Memory）模块。

Reflection 的核心思想并不是一次性生成最终答案，而是在完成一次初始执行后，不断经历反思（Reflection）→优化（Optimization）的迭代过程，使模型能够根据已有结果持续发现问题、提出改进，并逐步提升最终答案的质量。

如果每一轮都将全部历史上下文直接输入给负责反思的模型，不仅会引入大量冗余信息，还会增加 Token 消耗，并影响模型关注真正重要的信息。因此，我们需要一个记忆模块，对历史信息进行统一管理，并按需组织为后续推理所需的上下文。

本节将通过一个简单的代码生成任务来演示这一过程。

目标任务：编写一个 Python 函数，找出 1 到 n 之间所有的素数（Prime Numbers）。

之所以选择该任务，是因为它非常适合作为 Reflection 的演示案例，主要体现在以下几个方面：

  1. 存在清晰的优化空间。 大语言模型首次生成的代码通常能够正确完成任务，但往往采用较为朴素的实现方式，例如逐个判断每个数字是否为素数，整体效率较低。
  2. 反思点明确。 模型可以较容易发现当前实现存在的问题，例如时间复杂度较高、存在重复计算、缺少边界条件处理等。
  3. 优化方向清晰。 根据反思结果，可以进一步将算法优化为仅判断至平方根、跳过偶数、使用埃拉托斯特尼筛法（Sieve of Eratosthenes）等更高效的实现，从而形成一个完整的"生成—反思—优化"迭代过程。

由于 Reflection 需要不断参考历史执行结果，因此我们首先实现一个用于保存历史轨迹的短期记忆模块，其整体结构如下：

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 typing import List, Dict, Any, Optional

class Memory:
    """
    一个简单的短期记忆模块，用于存储智能体的行动与反思轨迹。
    """

    def __init__(self):
        """
        初始化一个空列表来存储所有记录。
        """
        self.records: List[Dict[str, Any]] = []

    def add_record(self, record_type: str, content: str):
        """
        向记忆中添加一条新记录。
        参数:
        - record_type (str): 记录的类型 ('execution' 或 'reflection')。
        - content (str): 记录的具体内容 (例如，生成的代码或反思的反馈)。
        """
        record = {"type": record_type, "content": content}
        self.records.append(record)
        print(f"📝 记忆已更新，新增一条 '{record_type}' 记录。")

    def get_trajectory(self) -> str:
        """
        将所有记忆记录格式化为一个连贯的字符串文本，用于构建提示词。
        """
        trajectory_parts = []
        for record in self.records:
            if record['type'] == 'execution':
                trajectory_parts.append(f"--- 上一轮尝试 (代码) ---\n{record['content']}")
            elif record['type'] == 'reflection':
                trajectory_parts.append(f"--- 评审员反馈 ---\n{record['content']}")

        return "\n\n".join(trajectory_parts)

    def get_last_execution(self) -> Optional[str]:
        """
        获取最近一次的执行结果 (例如，最新生成的代码)。
        如果不存在，则返回 None。
        """
        for record in reversed(self.records):
            if record['type'] == 'execution':
                return record['content']
        return None

这个 Memory 类的设计十分简单，主要承担以下几个职责：

  • records：使用列表按时间顺序保存每一轮迭代的信息，包括模型生成的结果、执行结果以及对应的反思内容。
  • add_record()：向记忆中新增一条记录，完成一次迭代轨迹的保存。
  • get_trajectory()：将历史记录序列化为文本，供后续提示词直接引用，使模型能够快速了解之前的尝试过程，而无需重新组织上下文。
  • get_last_execution()：获取最近一次执行产生的结果，作为下一轮 Reflection 的输入，便于模型针对最新方案进行分析与优化。

5.3 Reflection 智能体的编码实现

完成 Memory 模块后，我们便可以开始构建 ReflectionAgent 的核心逻辑。

与 ReAct 等范式不同，Reflection 更强调对已有结果进行持续改进。因此，整个智能体的工作流程可以概括为：

首先完成一次初始执行，随后围绕当前结果不断进行反思（Reflection）→优化（Optimization）迭代，直到达到预期效果。

在这一过程中，Memory 模块负责记录每一次生成结果和反思内容，为后续迭代提供历史上下文。

为了实现这一流程，我们需要分别设计不同阶段所使用的提示词，使模型在不同阶段承担不同的职责。

（1）提示词设计

Reflection 智能体并不会始终使用同一套提示词，而是根据当前所处阶段动态切换 Prompt，使模型分别完成代码生成、结果评审以及方案优化等不同任务。

① 初始执行提示词（Execution Prompt）

该提示词用于智能体的第一次生成，仅负责根据用户需求完成任务，不考虑历史反馈或优化建议，其目标是生成一个能够解决问题的初始方案。

1
2
3
4
5
6
7
8
INITIAL_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。请根据以下要求，编写一个Python函数。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。

要求: {task}

请直接输出代码，不要包含任何额外的解释。
"""

② 反思提示词（Reflection Prompt）

这是 Reflection 机制中最重要的一部分。该提示词引导模型扮演代码评审员（Reviewer）的角色，对上一轮生成的代码进行分析，指出存在的问题，并给出具体、可执行的改进建议，而不是直接修改代码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
REFLECT_PROMPT_TEMPLATE = """
你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。
你的任务是审查以下Python代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。

# 原始任务:
{task}

# 待审查的代码:
```python
{code}
```

请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。
如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。
如果代码在算法层面已经达到最优，才能回答“无需改进”。

请直接输出你的反馈，不要包含任何额外的解释。
"""

③ 优化提示词（Refinement Prompt）

在获得反思结果后，该提示词负责引导模型根据反馈重新生成代码。它会结合上一轮代码及对应的评审意见，对已有实现进行修正和优化，从而得到质量更高的新版本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
REFINE_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。

# 原始任务:
{task}

# 你上一轮尝试的代码:
{last_code_attempt}
评审员的反馈：
{feedback}

请根据评审员的反馈，生成一个优化后的新版本代码。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
请直接输出优化后的代码，不要包含任何额外的解释。
"""

可以发现，Reflection 将代码生成与代码评审拆分为两个独立阶段，使模型能够分别专注于"解决问题"和"发现问题"。这种角色分离能够有效减少模型一次性完成所有工作的负担，从而提高迭代优化的效果。

（2）智能体封装与实现

完成提示词设计后，我们便可以将 Memory 模块与三类 Prompt 进行整合，实现完整的 ReflectionAgent。

整个 Agent 的执行流程如下：

  1. 使用 Execution Prompt 生成初始代码；将生成结果保存到 Memory；
  2. 使用 Reflection Prompt 对最新代码进行分析，并生成评审意见；将反思内容保存到 Memory；
  3. 使用 Refinement Prompt 结合代码与反馈重新生成优化后的代码；
  4. 重复执行第 2～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
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

class ReflectionAgent:
    def __init__(self, llm_client, max_iterations=3):
        self.llm_client = llm_client
        self.memory = Memory()
        self.max_iterations = max_iterations

    def run(self, task: str):
        print(f"\n--- 开始处理任务 ---\n任务: {task}")

        # --- 1. 初始执行 ---
        print("\n--- 正在进行初始尝试 ---")
        initial_prompt = INITIAL_PROMPT_TEMPLATE.format(task=task)
        initial_code = self._get_llm_response(initial_prompt)
        self.memory.add_record("execution", initial_code)

        # --- 2. 迭代循环:反思与优化 ---
        for i in range(self.max_iterations):
            print(f"\n--- 第 {i + 1}/{self.max_iterations} 轮迭代 ---")

            # a. 反思
            print("\n-> 正在进行反思...")
            last_code = self.memory.get_last_execution()
            reflect_prompt = REFLECT_PROMPT_TEMPLATE.format(task=task, code=last_code)
            feedback = self._get_llm_response(reflect_prompt)
            self.memory.add_record("reflection", feedback)

            # b. 检查是否需要停止
            if "无需改进" in feedback:
                print("\n✅ 反思认为代码已无需改进，任务完成。")
                break

            # c. 优化
            print("\n-> 正在进行优化...")
            refine_prompt = REFINE_PROMPT_TEMPLATE.format(
                task=task,
                last_code_attempt=last_code,
                feedback=feedback
            )
            refined_code = self._get_llm_response(refine_prompt)
            self.memory.add_record("execution", refined_code)

        final_code = self.memory.get_last_execution()
        print(f"\n--- 任务完成 ---\n最终生成的代码:\n```python\n{final_code}\n```")
        return final_code

    def _get_llm_response(self, prompt: str) -> str:
        """一个辅助方法，用于调用LLM并获取完整的流式响应。"""
        messages = [{"role": "user", "content": prompt}]
        response_text = self.llm_client.think(messages=messages) or ""
        return response_text

需要注意的是，当前实现中的 ReflectionAgent 并未显式利用完整的历史轨迹进行推理，而是采用了一种更为轻量的“单步状态驱动”策略。

在每一轮迭代中，反思（Reflection）与优化（Optimization）阶段均仅依赖于最近一次的代码输出，即通过 get_last_execution() 获取当前版本，而不会显式引入此前所有迭代的完整历史记录。因此，该实现本质上是一种基于“最新状态”的逐步改进过程，而非严格意义上基于全局轨迹的反思机制。

这种设计的优势在于实现简单、上下文长度固定、Token 开销较低，适合快速迭代与教学演示场景；但其局限性在于模型无法回顾早期决策与修改路径，可能导致重复修正相同问题，或在复杂任务中出现信息遗忘与优化震荡。

因此，该版本更准确地可以描述为一种轻量级迭代优化型 Reflection Agent，而非完全依赖历史轨迹的增强型 Reflection 系统。

5.4 运行实例与分析

为了验证 Reflection Agent 的实际效果，我们在本节给出一个完整的运行示例。读者可以直接复现实验结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
if __name__ == '__main__':
    # 1. 初始化LLM客户端 (请确保你的 .env 和 llm_client.py 文件配置正确)
    try:
        llm_client = HelloAgentsLLM()
    except Exception as e:
        print(f"初始化LLM客户端时出错: {e}")
        exit()

    # 2. 初始化 Reflection 智能体，设置最多迭代2轮
    agent = ReflectionAgent(llm_client, max_iterations=2)

    # 3. 定义任务并运行智能体
    task = "编写一个Python函数，找出1到n之间所有的素数 (prime numbers)。"
    agent.run(task)

这个运行实例展示了 Reflection 机制是如何驱动智能体进行深度优化的:

  1. 有效“批判”是优化的起点 在第一轮反思中，由于引入了“极其严格”且“以算法效率为核心”的提示，智能体没有停留在“功能正确”的初级解，而是识别出原始实现中 \(O(n \cdot \sqrt{n})\) 的时间复杂度瓶颈，并进一步提出算法级改进方向——埃拉托斯特尼筛法。
  2. 迭代式性能提升 在接收到明确反馈后，智能体进入优化阶段，并成功将算法重构为更高效的筛法实现，使时间复杂度降至 \(O(n \log n)\)，完成了第一次具有实际意义的性能跃迁。
  3. 收敛判断与终止机制 在第二轮反思中，智能体对已有方案进行了更高层次的评估：既肯定了当前算法的高效性，也进一步讨论了分段筛法等更高级优化方向。但最终其判断为“在一般场景下已无进一步优化必要”，从而触发终止条件，使整个优化流程自然收敛。

这个案例充分证明，一个设计良好的 Reflection 机制，其价值不仅在于修复错误，更在于驱动解决方案在质量和效率上实现阶梯式的提升，这使其成为构建复杂、高质量智能体的关键技术之一。

5.5 Reflection 机制的成本收益分析

尽管 Reflection 机制在提升任务解决质量方面表现突出，但这种能力并非“免费午餐”。在实际应用中，需要在性能收益与系统成本之间进行权衡。

（1）主要成本

  1. 模型调用开销增加 这是最直接的成本来源。每增加一轮 Reflection 迭代，通常需要至少两次大模型调用（反思阶段 + 优化阶段）。随着迭代轮数上升，API 调用费用与算力消耗会线性甚至倍增增长。
  2. 任务延迟显著增加 Reflection 本质上是一个严格串行的优化流程，每一轮优化都必须等待上一轮反思完成后才能继续推进。因此整体执行链路被显著拉长，在多轮迭代情况下尤为明显，不适用于低延迟或实时性要求较高的任务。
  3. 提示工程复杂度上升 如前文案例所示，Reflection 的效果高度依赖阶段化提示词设计（执行 / 反思 / 优化）。如何为不同阶段构造稳定、可控且具有引导性的 prompt，本身就需要较高的工程调试成本与经验积累。

（2）核心收益

  1. 解决方案质量的阶梯式提升 Reflection 的核心价值在于“迭代优化能力”。它能够将初始的“可用解”逐步提升为“高质量解”，实现从功能正确到性能最优、从逻辑可用到结构严谨的跨越。这种质量跃迁在复杂任务中尤为关键。
  2. 鲁棒性与可靠性增强 通过“生成—审查—修正”的闭环机制，智能体能够主动识别初始解中的逻辑缺陷、边界问题或潜在错误，从而显著提升最终输出的稳定性与可靠性。

（3）小结与适用场景

总体而言，Reflection 机制是一种典型的“以计算与时间换取质量”的策略。其适用场景通常具备以下特征：

  • 对输出质量、准确性与可靠性要求较高
  • 对任务完成时间不敏感

例如：

  • 关键业务代码生成与技术方案设计
  • 科学研究中的复杂推导与分析任务
  • 高价值决策支持与深度规划系统

相对而言，在以下场景中其性价比可能不足：

  • 强实时性任务（如在线交互、低延迟问答）
  • 容忍一定误差的轻量级生成任务

在这些情况下，采用更轻量的 ReAct 或 Plan-and-Solve 等范式，往往能够在成本与效果之间取得更优平衡。