构建你的 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
HelloAgent/
   │
   ├── core/                        # 核心框架层
   │   ├── agent.py                 # Agent基类
   │   ├── llm.py                   # HelloAgentsLLM统一接口
   │   ├── message.py               # 消息系统
   │   ├── config.py                # 配置管理
   │   └── exceptions.py            # 异常体系
   │    
   ├── agent/                       # Agent实现层
   │   ├── simpleAgent.py           # SimpleAgent实现
   │   ├── reActAgent.py            # ReActAgent实现
   │   ├── reflectionAgent.py       # ReflectionAgent实现
   │   ├── planSolveAgent.py        # PlanAndSolveAgent实现
   │   └── prompts                  # 提示词模板
   │     ├── planAndSolvePrompt.py
   │     ├── reActPrompt.py
   │     └── reflectionPrompt.py
   │    
   └── tool/                        # 工具系统层
        ├── tool.py                 # 工具基类
        ├── toolRegistry.py         # 工具注册机制
        ├── toolChain.py            # 工具链管理系统
        ├── toolParameter.py        # 工具参数
        ├── asyncToolExecutor.py    # 异步工具执行器
        └── builtin/                # 内置工具集
            ├── calculator.py       # 计算工具
            ├── search.py           # 搜索工具
            ├── getWeather.py       # 获取天气
            └── getAttraction.py    # 获取景点

在里面有眼熟的，agent/ 目录下面就有第四章中手写过的 ReAct、Plan-and-Resolve 和 Reflection 三种模式的 Agent；我们的 llm 接口则移到 core 了。

另外，第四章的 ReAct 模式中已经有过工具构建的经验，将在 tool/ 目录下面进行定义，包括工具执行、注册发现、工具本体等。

体验现有的 Hello Agent 框架

官方提供了已经实现好的代码，使用 pip 可以进行安装：

1
pip install "hello-agents==0.1.1"

之后我们定义一个 testHelloAgent 来学习一下如何构建这个 Agent，我们来阅读一下。

构建 Agent

1
2
3
4
5
6
7
8
9
10
11
12
# 创建LLM实例 - 框架自动检测provider
llm = HelloAgentsLLM()

# 或手动指定provider（可选）
# llm = HelloAgentsLLM(provider="modelscope")

# 创建SimpleAgent
agent = SimpleAgent(
    name="AI助手",
    llm=llm,
    system_prompt="你是一个有用的AI助手"
)

构建一个简单的 SimpleAgent ，与我们第四章构建 Agent 的方法几乎完全一样：我们将 LLM 调用接口 llm 作为变量传入 Agent。

简单对话

1
2
3
# 基础对话
response = agent.run("你好！请介绍一下自己")
print(response)

我们看到，run 函数同样时作为 Agent 暴露给外界的运行方式。

Function Calling

1
2
3
4
5
6
7
8
9
 添加工具功能（可选）
from hello_agents.tools import CalculatorTool
calculator = CalculatorTool()
# 需要实现7.4.1的MySimpleAgent进行调用，后续章节会支持此类调用方式
# agent.add_tool(calculator)

# 现在可以使用工具了
response = agent.run("请帮我计算 2 + 3 * 4")
print(response)

这一步类似第四章中的构建：将方法作为 callable，注册到 Agent，然后 agent 才能激活工具意图。

测试结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
你好呀！很高兴认识你！

我是 **DeepSeek**，一个由深度求索公司创造的AI助手。我可以帮你解答问题、提供建议、处理文字信息、进行创作，或者陪你聊天解闷。

我的一些特点包括：
- 💬 纯文本交流，但支持读取你上传的图片、PDF、Word、Excel等文件中的文字
- 🌐 支持联网搜索（需要你在Web/App上手动开启）
- 📚 上下文长度1M，可以一次性处理超大文本（比如整本小说）
- 🎤 App端支持语音输入
- 🆓 完全免费

我的知识截止到2025年5月，会用热情、细腻的风格回应你。有什么我可以帮你的吗？尽管说～
根据数学运算的优先级（先乘除，后加减），计算过程如下：

**3 × 4 = 12**  
然后  
**2 + 12 = 14**

所以，`2 + 3 × 4 = 14` ✅

希望这个回答对你有帮助！还有其他问题吗？
历史消息数: 4

core：大模型调用核心

这里是整个客户端的核心，有关大模型的一切，包括大模型配置、大模型调用、上下文管理都在这里实现。

模型最底层接口： llm

我们直接复用第四章中 HelloAgentLLM 的代码，大大方方用。

官方提出的支持多供应商感觉没啥用，用户自己配置一次跑不通就自行查找问题了，让开发者来做自动化适配耗费力气。更何况弄来弄去还是只能支持 openAI 规范，我上个 anthropic 的 key 你不炸了。

标准化消息格式：message

Message 类是为了让上下文的格式进行统一，其继承了 BaseModel 类，这是 python 中专门用于定义数据类的库，只要继承就可以实现方便的数据管理。

另外，Message 依赖 pydantic 库：

1
pip install pydantic

单个 Massage 类对象的信息以 kv 对形式存储，分别为初始化时传入的参数：role: MessageRole 和 content: str。初始化：

1
2
3
4
5
6
7
def __init__(self, content: str, role: MessageRole, **kwargs):
    super().__init__(
        content=content,
        role=role,
        timestamp=kwargs.get('timestamp', datetime.now()),
        metadata=kwargs.get('metadata', {})
    )

role 的取值被固定为四个字符串，这是为了直接对应 OpenAI API 的规范，openAI 规范中，LLM 以下四个字符串来判断该信息的来源和作用:

1
MessageRole = Literal["user", "assistant", "system", "tool"]

其中 assistant 是什么作用？它是模型在响应中写入的一段文字，便于下一次调用时快速了解会话上下文。

to_dict 函数则将 Message 对象转成 openAI 支持的规范字典格式。

1
2
3
4
5
6
def to_dict(self) -> Dict[str, Any]:
    """转换为字典格式（OpenAI API格式）"""
    return {
        "role": self.role,
        "content": self.content
    }

_str_ 函数就是直接将自带格式转成字符串的：

1
2
def __str__(self) -> str:
    return f"[{self.role}] {self.content}"

配置中心：core/config

Config 的作用将各种硬编码参数集中起来，统一配置，有着以下参数：

1
2
3
4
5
6
7
8
9
10
11
12
# LLM配置
default_model: str = "gpt-3.5-turbo"
default_provider: str = "openai"
temperature: float = 0.7
max_tokens: Optional[int] = None

# 系统配置
debug: bool = False
log_level: str = "INFO"

# 其他配置
max_history_length: int = 100

该类的核心是函数 from_env，能够从环境变量中提取参数，以 Config 类对象的形式返回：

1
2
3
4
5
6
7
8
9
@classmethod
def from_env(cls) -> "Config":
    """从环境变量创建配置"""
    return cls(
        debug=os.getenv("DEBUG", "false").lower() == "true",
        log_level=os.getenv("LOG_LEVEL", "INFO"),
        temperature=float(os.getenv("TEMPERATURE", "0.7")),
        max_tokens=int(os.getenv("MAX_TOKENS")) if os.getenv("MAX_TOKENS") else None,
    )

对外暴露接口：core/agent

我们创建一个 Agent 接口，来规定了一个 agent 需要实现哪些方法、应当拥有哪些行为，通过继承 ABC(Abstract Class) 可以实现类的接口化：不能创建实例、方法不能实现。

1
2
class Agent(ABC):
    """Agent基类"""

首先，我们看看其初始化的输入：最重要的是 ``llm: LLM、system_prompt: Optional[str]和config: Optional[Config]` 三个。分别为 openAI 规范的大模型接口，必须传入、字符串形式的系统提示词、自行定义的 Config 类。

另外，如果没有输入 Config 实例，需要创建一个默认配置。

_histor 为 Message 列表，用于存储我们自定义规范化的上下文。

1
2
3
4
5
6
7
8
9
10
11
12
def __init__(
    self,
    name: str,
    llm: LLM,
    system_prompt: Optional[str] = None,
    config: Optional[Config] = None
):
    self.name = name
    self.llm = llm
    self.system_prompt = system_prompt
    self.config = config or Config()
    self._history: list[Message] = []

实现该抽象类的 Agent 类必须包含以下方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@abstractmethod
def run(self, input_text: str, **kwargs) -> str:
    """运行Agent"""
    pass

def add_message(self, message: Message):
    """添加消息到历史记录"""
    self._history.append(message)

def clear_history(self):
    """清空历史记录"""
    self._history.clear()

def get_history(self) -> list[Message]:
    """获取历史记录"""
    return self._history.copy()

def __str__(self) -> str:
    return f"Agent(name={self.name}, provider={self.llm.provider})"

首先通过注解 @abstractmethod 规定实现类必须实现 run 函数，用以调用大模型。

三个管理上下文的函数：add_message clear_history get_history。

tool：工具

Agent 必须能够调用工具，官方的内容编排我实在是无力吐槽，过完 core 直接上手 agent，还一上来就要继承官方给的包，那这跟手写还有什么关系？我直接用你的包不就完了，为啥还要自己实现？

总之，我们先在 tool 包内搞定工具。

tool

我们最先定义一个抽象类 Tool，为工具定义规范：

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
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from .toolParameter import ToolParameter

class Tool(ABC):
    """工具基类"""

    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    @abstractmethod
    def run(self, parameters: Dict[str, Any]) -> str:
        """执行工具"""
        pass

    @abstractmethod
    def get_parameters(self) -> List[ToolParameter]:
        """获取工具参数定义"""
        pass

    def to_openai_schema(self) -> Dict[str, Any]:
        """转换为 OpenAI function calling schema 格式

        用于 FunctionCallAgent，使工具能够被 OpenAI 原生 function calling 使用

        Returns:
            符合 OpenAI function calling 标准的 schema
        """
        parameters = self.get_parameters()

        # 构建 properties
        properties = {}
        required = []

        for param in parameters:
            # 基础属性定义
            prop = {
                "type": param.type,
                "description": param.description
            }

            # 如果有默认值，添加到描述中（OpenAI schema 不支持 default 字段）
            if param.default is not None:
                prop["description"] = f"{param.description} (默认: {param.default})"

            # 如果是数组类型，添加 items 定义
            if param.type == "array":
                prop["items"] = {"type": "string"}  # 默认字符串数组

            properties[param.name] = prop

            # 收集必需参数
            if param.required:
                required.append(param.name)

        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }

初始化中，我们输入了 name 和 description，用来告诉 LLM 这是什么工具、什么时候需要调用这个工具。

在 run 函数中，需要实现工具逻辑，同时将返回值封装为 Dict[str, Any] 的形式。

get_parameters 函数中，我们要向外界暴露出该工具的 name 和 description，便于进行工具的注册发现。里面的 ToolParameter 是一个数据类，用来描述工具更加详细的信息，需要我们自行定义。

最后，我们使用 to_openai_schema 函数，将工具转化为 openAI 格式的 json schema，shcema的格式为：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "type": "function",
    "function": {
        "name": self.name,
        "description": self.description,
        "parameters": {
            "type": "object",
            "properties": properties,
            "required": required
        }
    }
}

ToolParameter

ToolParameter 是一个数据类，需要继承 BaseModel，其内包含的参数有：

1
2
3
4
5
6
"""工具参数定义"""
name: str
type: str
description: str
required: bool = True
default: Any = None

ToolRegistry

有了具体的工具，接下来就要进行注册发现，我们定义一个 ToolRegistry 类作为工具注册表。

类内最重要的成员变量是 _tools 和 _functions，分别用来存储 完整的 Tool 对象 和 函数 + 描述，因为我们支持两种注册方式：Tool & Callable。

1
2
self._tools: dict[str, Tool] = {}
self._functions: dict[str, dict[str, Any]] = {}

工具注册

Tool 的注册很简单，直接传入一个 Tool 就行：

1
2
3
4
5
6
def register_tool(self, tool: Tool):
    """注册Tool对象"""
    if tool.name in self._tools:
        print(f"⚠️ 警告:工具 '{tool.name}' 已存在，将被覆盖。")
    self._tools[tool.name] = tool
    print(f"✅ 工具 '{tool.name}' 已注册。")

register_function 函数为 Callable 的注册，传入参数的 func: Callable[[str], str] 表示的是输入为字符串，返回值也为字符串的函数引用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def register_function(self, name: str, description: str, func: Callable[[str], str]):
    """
    直接注册函数作为工具（简便方式）

    Args:
        name: 工具名称
        description: 工具描述
        func: 工具函数，接受字符串参数，返回字符串结果
    """
    if name in self._functions:
        print(f"⚠️ 警告:工具 '{name}' 已存在，将被覆盖。")

    self._functions[name] = {
        "description": description,
        "func": func
    }
    print(f"✅ 工具 '{name}' 已注册。")

get_tools_description 函数则能够向外界暴露注册表内包含的工具。

自定义 Callable 工具

我们自定义一个数学表达式求值函数工具 tool\builtin\calculator.py。

如果我们想要执行函数工具，或是想要获取这个函数的引用，我们就必须在 ToolRegistry 里追加函数。

函数工具的执行 & 获取

在 ToolRegistry 内追加函数：

1
2
3
4
5
6
7
8
9
10
11
12
def execute_function(self, name: str, *args, **kwargs) -> Any:
    """执行注册的函数工具"""
    if name not in self._functions:
        raise ValueError(f"工具 '{name}' 未注册。")
    func = self._functions[name]["func"]
    return func(*args, **kwargs)

def get_function(self, name: str) -> Callable[[str], str]:
    """获取注册的函数工具"""
    if name not in self._functions:
        raise ValueError(f"工具 '{name}' 未注册。")
    return self._functions[name]["func"]

测试

我们写一个测试函数，使用 ToolRegistry 注册这个函数，并调用函数：

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
from tool.toolRegistry import ToolRegistry
from tool.builtin.calculator import calculator

if __name__ == "__main__":
    """测试自定义计算器工具"""
    print("🧪 测试自定义计算器工具\n")
    # 创建工具注册表
    registry = ToolRegistry()
    # 注册计算器工具
    registry.register_function(
        name="calculator",
        description="简单的数学计算工具，支持基本运算(+,-,*,/)和sqrt函数",
        func=calculator
    )
    # 简单测试用例
    test_cases = [
        "2 + 3",           # 基本加法
        "10 - 4",          # 基本减法
        "5 * 6",           # 基本乘法
        "15 / 3",          # 基本除法
        "sqrt(16)",        # 平方根
    ]
    for i, expression in enumerate(test_cases, 1):
        print(f"测试 {i}: {expression}")
        result = registry.get_function("calculator")(expression)
        # 或者 result = registry.execute_function("calculator", expression)
        print(f"结果: {result}\n")

结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
🧪 测试自定义计算器工具

✅ 工具 'calculator' 已注册。
测试 1: 2 + 3
结果: 5

测试 2: 10 - 4
结果: 6

测试 3: 5 * 6
结果: 30

测试 4: 15 / 3
结果: 5.0

测试 5: sqrt(16)
结果: 4.0

复用 getAttraction 和 getWeather

HelloAgent 在前面 第一章 初识智能体 中给出过两个函数工具，我们可以直接拿来用，通过 ToolRegisty 注册 Callable 就可以用了。

自定义 Tool 工具

我们自定义一个 search.SearchTool 类，继承 Tool 类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class SearchTool(Tool):
    """
    智能混合搜索工具，使用 SerpApi 和 Tavily API 进行网页搜索。
    """

    def __init__(self):
        super().__init__(
            name="search",
            description="一个智能网页搜索引擎。支持混合搜索模式，自动选择最佳搜索源。"
        )
        self.tavily_key = os.getenv("TAVILY_API_KEY")
        self.serpapi_key = os.getenv("SERPAPI_API_KEY")
        self.available_backends = []
        self._setup_backends()
        print(f"SearchTool initialized with backend: {self.available_backends}")

选择搜索引擎

Hello Agent 到现在，已经展示了两种搜索引擎：tavily 和 serpapi，SearchTool 中我们想要让动态选择这两种搜索引擎，所以我们有 _setup_backends 函数：

1
2
3
4
5
6
7
8
9
10
def _setup_backends(self):
    """
    设置可用的搜索后端。
    """
    if self.tavily_key:
        self.available_backends.append("tavily")
    elif self.serpapi_key:
        self.available_backends.append("serpapi")
    if not self.available_backends:
        raise ValueError("No available search backends. Please set TAVILY_API_KEY or SERPAPI_API_KEY.")

根据环境变量中是否有 Key 来选择，优先选择 tavily。

tavily 搜索

在 hello agent 之前的章节中，已经展示过如何使用 tavily api 了，这里定义函数 _search_tavily：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def _search_tavily(self, query: str) -> Dict[str, Any]:
    """使用Tavily搜索"""
    self.tavily_client = TavilyClient(api_key=self.tavily_key)

    response = self.tavily_client.search(query=query, max_results=3)

    if response.get('answer'):
        result = f"💡 AI直接答案:{response['answer']}\n\n"
    else:
        result = ""

    result += "🔗 相关结果:\n"
    for i, item in enumerate(response.get('results', [])[:3], 1):
        result += f"[{i}] {item.get('title', '')}\n"
        result += f"    {item.get('content', '')[:150]}...\n\n"

    return result

先创建客户端，再向 tavily 服务器发送请求，然后处理 response；max_results 表示返回几条结果。

另外，response 是以 JSON 格式响应的：

1
{'query': 'HelloAgent是什么', 'follow_up_questions': None, 'answer': None, 'images': [], 'results': [{'url': 'https://github.com/jjyaoao/helloagents', 'title': 'GitHub - jjyaoao/HelloAgents: A agent framework based on the tutorial hello-agents · GitHub', 'content': "## Navigation Menu. We read every piece of feedback, and take your input very seriously. # Saved searches. ## Use saved searches to filter your results more quickly. # jjyaoao/HelloAgents. ## Folders and files. | Name | | Name | Last commit message | Last commit date |. | Latest commit   History71 Commits   71 Commits | | |. ## Latest commit. ## Repository files navigation. # HelloAgents. License: CC BY-NC-SA 4.0. HelloAgents 是一个基于 OpenAI 原生 API 构建的生产级多智能体框架，集成了工具响应协议（ToolResponse）、上下文工程（HistoryManager/TokenCounter）、会话持久化（SessionStore）、子代理机制（TaskTool）、乐观锁（文件编辑）、熔断器（CircuitBreaker）、Skills 知识外化、TodoWrite 进度管理、DevLog 决策记录、流式输出（SSE）、异步生命周期、可观测性（TraceLogger）、日志系统（四种范式）、LLM/Agent 基类重构等 16 项核心能力，为构建复杂智能体应用提供完整的工程化支持。. **🚀 开发版本（当前分支）**：持续迭代中的最新代码(V1.0.0)，包含新功能和改进，部分实现可能与教程内容存在差异。如需学习教程，请切换到 `learn_version` 分支。. | **云端 API** | OpenAI、DeepSeek、Qwen、Kimi、智谱 GLM | `LLM_BASE_URL=api.deepseek.com` |. | **其他兼容** | 任何 OpenAI 格式接口 | `LLM_BASE_URL=your-endpoint` |. | **Claude** | `base_url` 包含 `anthropic.com` | `LLM_BASE_URL=https://api.anthropic.com` |. `LLM_BASE_URL=https://api.anthropic.com`. | **Google Gemini** | `base_url` 包含 `googleapis.com` 或 `generativelanguage` | `LLM_BASE_URL=https://generativelanguage.googleapis.com` |. `LLM_BASE_URL=https://generativelanguage.googleapis.com`. `git commit -m 'Add some AmazingFeature'`. 本项目采用 CC BY-NC-SA 4.0 许可证 - 查看 LICENSE 文件了解详情。. 详细了解 HelloAgents v1.0.0 的 16 项核心能力：. A agent framework based on the tutorial hello-agents. ### Footer navigation.", 'score': 0.8306082, 'raw_content': None}, {'url': 'https://zhuanlan.zhihu.com/p/1965417697228621653', 'title': '《Hello-Agents》项目正式发布，一起从零学习智能体！ - 知乎专栏', 'content': 'Hello-Agents 是Datawhale社区的系统性智能体学习教程。在如今Agent主要分为两派，一派是Dify，Coze，n8n这类软件工程类Agent，其本质只是将模型作为返回字符串的后端', 'score': 0.82795376, 'raw_content': None}, {'url': 'https://zhuanlan.zhihu.com/p/1965351417196836801', 'title': '《Hello-Agents》项目正式发布，一起从零学习智能体！ - 知乎专栏', 'content': 'Hello-Agents 是Datawhale社区的系统性智能体学习教程。在如今Agent主要分为两派，一派是Dify，Coze，n8n这类软件工程类Agent，其本质只是将模型作为', 'score': 0.82391036, 'raw_content': None}], 'response_time': 0.0, 'request_id': '7e8ed377-8a89-47d9-bc3d-e3ee6a333431'}

serpapi 搜索

定义函数 _search_serpapi：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def _search_serpapi(self, query: str) -> Dict[str, Any]:
    """
    使用 SerpApi 执行搜索。
    """
    search = GoogleSearch({
        "q": query,
        "api_key": os.getenv("SERPAPI_API_KEY"),
        "num": 3
    })

    results = search.get_dict()

    result = "🔗 Google搜索结果:\n"
    if "organic_results" in results:
        for i, res in enumerate(results["organic_results"][:3], 1):
            result += f"[{i}] {res.get('title', '')}\n"
            result += f"    {res.get('snippet', '')}\n\n"

    return result

我们没有采用之前的 serpapi_client，而是采用了 GoogleSearch，不过也没什么区别，search.get_dict() 的返回值也是 JSON，和 tavily 一样。

实现 Tool 接口

别忘了，SearchTool 是继承了抽象类 Tool 的，必须要实现 run 和 get_parameters 方法，分别实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def run(self, query: str) -> Dict[str, Any]:
    """
    执行搜索操作，根据可用的后端选择最佳搜索源。
    """
    if "tavily" in self.available_backends:
        return self._search_tavily(query)
    elif "serpapi" in self.available_backends:
        return self._search_serpapi(query)
    else:
        raise ValueError("No available search backends to perform the search.")

def get_parameters(self) -> Dict[str, Any]:
    """
    获取工具参数定义。
    """
    return {
        "query": {
            "type": "string",
            "description": "搜索查询字符串",
            "required": True
        }
    }

测试

写一个测试方法，看看能否通过 ToolRegistry，获取到这个 Tool 子类并执行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from tool.builtin.search import SearchTool
from tool.toolRegistry import ToolRegistry

def testSearchTool:
    # 测试 SearchTool 的初始化和搜索功能

    registry = ToolRegistry()
    # 注册 SearchTool
    search_tool = SearchTool()
    registry.register_tool(search_tool)
    query = "antropic风评为什么这么差？"
    print(f"正在搜索: {query}")
    result = registry.execute_tool("search", query=query)
    print(f"搜索结果:\n{result}")

效果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
> python .\__testSearch.py

已选择搜索引擎: ['serpapi']
✅ 工具 'search' 已注册。
正在搜索: antropic风评为什么这么差？
搜索结果:
🔗 Google搜索结果:
[1] Anthropic 真的算是AI 界的良心吗？ : r/ClaudeAI
    我们知道 Altman 减少了安全团队在 openai获得的计算量，而且 gpt4o 还是差劲得要死。他玩弄所有商业伎俩，试图窃取 ...

[2] Critics of Anthropic Have a Point—But So Do Its AI ...
    One of the world's leading AI companies, Anthropic, is calling for an industry-wide mechanism to pause advanced artificial intelligence ...

[3] 如何评价Anthropic 这个据称强到不敢直接发布的大模型 ...
    4 月 8 日消息，Anthropic 昨日（4 月 7 日）发布公告，宣布联合苹果、英伟达、微软、亚马逊云服务（AWS）、谷歌、Linux ...

工具链模式：鸡肋

工具链存在的意义是什么？我们知道，大模型在拥有了工具意图能力的情况下，已经能够自主地按需调用工具了，像工具链这种硬编码的工作流形式，对于大模型而言是否过于鸡肋了？

确实是这样的，工具种类一旦变多，由我们硬编码的特定工作流且不说耗时耗力，其命中率基本上等于0。

所以这节基本上就了解一下就行了，实用价值为0。

异步工具

是为了将多个没有因果关系的工具执行并行化，以节省时间。

创建一个 AsyncToolExecutor 类，核心使用方法：传入一个已经注册好工具的 ToolRegistry 实例创建 AsyncToolExecutor 实例、设定工具任务批次、异步执行。

支持函数类工具

这里要注意的是，官方给的代码，只考虑到了传入工具是 Tool 类实例的情况，于是在 AsyncToolExecutor 的execute_tool_async 函数中只能够执行 Tool 类工具，对于函数类工具就不能支持，然而其官方测试代码却堂而皇之地使用了 calculator 。所以这里我们对函数式工具做出支持。

现在 ToolRegistry 中添加判断工具类型的函数：

1
2
3
4
5
6
7
def is_function(self, name: str) -> bool:
    """判断一个工具名是否为函数类型"""
    return name in self._functions

def is_tool(self, name: str) -> bool:
    """判断一个工具名是否为Tool对象类型"""
    return name in self._tools

随后，修改 AsyncToolExecutor 中的 execute_tool_async 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def execute_tool_async(self, tool_name: str, input_data: str) -> str:
    """异步执行单个工具，能够判断工具是函数还是Tool对象，并调用相应的执行方法"""
    loop = asyncio.get_event_loop()

    if self.registry.is_function(tool_name):
        def _execute():
            return self.registry.execute_function(tool_name, input_data)
    elif self.registry.is_tool(tool_name):
        def _execute():
            return self.registry.execute_tool(tool_name, input_data)
    else:
        raise ValueError(f"工具 '{tool_name}' 未在注册表中找到（既非函数也非Tool对象）")

    result = await loop.run_in_executor(self.executor, _execute)
    return result

完成。

测试

撰写测试函数：

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
from tool.toolRegistry import ToolRegistry
from tool.builtin.calculator import calculator
from tool.builtin.search import SearchTool
from tool.asyncToolExecutor import AsyncToolExecutor

# 使用示例
async def test_parallel_execution():
    """测试并行工具执行"""

    
    registry = ToolRegistry()
    registry.register_function(
        name="calculator",
        description="简单的数学计算工具，支持基本运算(+,-,*,/)和sqrt函数",
        func=calculator
    )
    registry.register_tool(SearchTool())

    executor = AsyncToolExecutor(registry)

    # 定义并行任务
    tasks = [
        {"tool_name": "search", "input_data": "Python编程"},
        {"tool_name": "search", "input_data": "机器学习"},
        {"tool_name": "calculator", "input_data": "2 + 2"},
        {"tool_name": "calculator", "input_data": "sqrt(16)"},
        
    ]

    # 并行执行
    results = await executor.execute_tools_parallel(tasks)

    for i, result in enumerate(results):
        print(f"任务 {i+1} 结果: {result[:100]}...")


if __name__ == "__main__":
    import asyncio
    asyncio.run(test_parallel_execution())

结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
> python .\__testAsyncToolExecutor.py

✅ 工具 'calculator' 已注册。
已选择搜索引擎: ['serpapi']
✅ 工具 'search' 已注册。
🚀 开始并行执行 4 个工具任务
✅ 所有工具任务执行完成
任务 1 结果: 🔗 Google搜索结果:
[1] Python 教程— Python 3.14.6 文档
    Python 是一门易于学习、功能强大的编程语言。它提供了高效的高级数据结构，还能简单有效地面向对象...
任务 2 结果: 🔗 Google搜索结果:
[1] 什么是机器学习？ | Machine Learning
    简单来说，机器学习是指训练一段称为模型的软件，使其能够根据数据做出有用的预测或生成内容（例如文本、图...
任务 3 结果: 4...
任务 4 结果: 4.0..

agent：大模型调用策略

在 core 包内，我们主要实现了大模型的统一配置和规范化的上下文管理，以及给定义的 Agent 规定了一套接口；而 tool 包内，我们已经定义好了工具模板、注册发现，同时还实现了一些具体的工具。

而在 agent 包内，我们终于要开始用 core 包与 tool 包的组件，实现各种策略不同的 Agent。而在此之前，我们还是定义一个标准的 Agent 实现类作为基准，然后再展开各种模式的 Agent。

SimpleAgent

在实现不同策略的 Agent 之前，我们先定义一个标准的 SimpleAgent。

在 core 包内，我们已经定义了一个模板 Agent，我们需要继承它，并实现其中的抽象函数。另外，我们也要在此之上完成一些功能的扩展

初始化

1
2
3
4
5
6
7
8
9
10
11
12
13
def __init__(
    self,
    name: str,
    llm: LLM,
    system_prompt: Optional[str] = None,
    config: Optional[Config] = None,
    tool_registry: Optional['ToolRegistry'] = None,
    enable_tool_calling: bool = True
):
    super().__init__(name, llm, system_prompt, config)
    self.tool_registry = tool_registry
    self.enable_tool_calling = enable_tool_calling and tool_registry is not None
    print(f"✅ {name} 初始化完成，工具调用: {'启用' if self.enable_tool_calling else '禁用'}")

除了接口 Agent 已有的四个输入，还加入了 ToolRegistry 和 enable_tool_calling。

基础问答：通过重写 run 函数

继承 Agent 必须重写 run，这恰好也是我们需要的暴露接口，外界要使用 Agent 就必须调用 run 函数。

run 函数的流程我们已经很熟悉了：加入上下文，然后调用 LLM。只不过其中，我们加入和一个 _get_enhanced_system_prompt 函数，用来强化系统提示词，主要目的就是将 ToolRegistry 中的工具添加到系统提示词，进一步强化工具意图。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def _get_enhanced_system_prompt(self) -> str:
    """构建增强的系统提示词，包含工具信息"""
    base_prompt = self.system_prompt or "你是一个有用的AI助手。"

    if not self.enable_tool_calling or not self.tool_registry:
        return base_prompt

    # 获取工具描述
    tools_description = self.tool_registry.get_tools_description()
    if not tools_description or tools_description == "暂无可用工具":
        return base_prompt

    tools_section = "\n\n## 可用工具\n"
    tools_section += "你可以使用以下工具来帮助回答问题:\n"
    tools_section += tools_description + "\n"

    tools_section += "\n## 工具调用格式\n"
    tools_section += "当需要使用工具时，请使用以下格式:\n"
    tools_section += "`[TOOL_CALL:{tool_name}:{parameters}]`\n"
    tools_section += "例如:`[TOOL_CALL:search:Python编程]` 或 `[TOOL_CALL:memory:recall=用户信息]`\n\n"
    tools_section += "工具调用结果会自动插入到对话中，然后你可以基于结果继续回答。\n"

    return base_prompt + tools_section

此外，在 run 函数尝试调用 LLM 前，如果 Agent 有工具权限，那么就让 _run_with_tools 函数来代理工具执行。

工具意图： _run_with_tools

前面提到，如果 Agent 有使用工具的权限，那么 run 函数并不会直接调用 LLM ，而是将调用 LLM 代理给 _run_with_tools 函数，我们看看这个函数干了什么：

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
def _run_with_tools(self, messages: list, input_text: str, max_tool_iterations: int, **kwargs) -> str:
    """支持工具调用的运行逻辑"""
    current_iteration = 0
    final_response = ""

    while current_iteration < max_tool_iterations:
        # 调用LLM
        response = self.llm.think(messages, **kwargs)
        # 检查是否有工具调用
        tool_calls = self._parse_tool_calls(response)
        if tool_calls:
            print(f"🔧 检测到 {len(tool_calls)} 个工具调用")
            # 执行所有工具调用并收集结果
            tool_results = []
            clean_response = response
            for call in tool_calls:
                result = self._execute_tool_call(call['tool_name'], call['parameters'])
                tool_results.append(result)
                # 从响应中移除工具调用标记
                clean_response = clean_response.replace(call['original'], "")
            # 构建包含工具结果的消息
            messages.append({"role": "assistant", "content": clean_response})
            # 添加工具结果
            tool_results_text = "\n\n".join(tool_results)
            messages.append({"role": "user", "content": f"工具执行结果:\n{tool_results_text}\n\n请基于这些结果给出完整的回答。"})
            current_iteration += 1
            continue
        # 没有工具调用，这是最终回答
        final_response = response
        break
    # 如果超过最大迭代次数，获取最后一次回答
    if current_iteration >= max_tool_iterations and not final_response:
        final_response = self.llm.think(messages, **kwargs)
    # 保存到历史记录
    self.add_message(Message(input_text, "user"))
    self.add_message(Message(final_response, "assistant"))
    print(f"✅ {self.name} 响应完成")
    return final_response

这个函数的核心是一个关于 max_tool_iterations = 3 的循环，每个单次循环内，都直接调用了 LLM，查看 LLM 是否有工具意图（从 response 中查找），如果有工具意图，那就调用自身的 _execute_tool_call 函数调用工具，并得到调用结果，拼接到一个 Message 实例中。如此一来，核心循环完成时，Message 内已经有多次工具调用的结果，此时再调用 LLM，LLM 有足够的信息判断工具是否调用成功，随后再完成响应。

流式响应：stream_run

流式响应，就是让 LLM 响应的增量片段实时地传输给用户，无需等待增量片段响应完毕再统一响应。

不过当前有一个问题是，我们底层的大模型调用接口 think 采用的是同步响应，也即：

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
def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
    """
    调用大语言模型进行思考，并返回其响应。
    """
    print(f"🧠 正在调用 {self.model} 模型...")
    try:
        # 通过LLM接口调用大模型，response为流式响应(stream=True)
        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

如果我们想要在 Agent 中实现流式响应，光靠 think 函数是不可能的。所以我们不得不对 LLM 类进行改动，添加一个 think_stream 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def think_stream(self, messages: List[Dict[str, str]], temperature: float = 0):
    """
    流式调用大语言模型进行思考，逐步返回其响应。
    """
    print(f"🧠 正在流式调用 {self.model} 模型...")
    try:
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            stream=True,
        )
        for chunk in response:
            if not chunk.choices:
                continue
            content = chunk.choices[0].delta.content or ""
            yield content

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

通过 yield 直接将增量片段返回。随后，我们在 SimpleAgent 中定义一个函数接受被 yield 的增量片段：

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
def stream_run(self, input_text: str, **kwargs) -> Iterator[str]:
    """
    自定义的流式运行方法
    """
    print(f"🌊 {self.name} 开始流式处理: {input_text}")

    messages = []

    if self.system_prompt:
        messages.append({"role": "system", "content": self.system_prompt})

    for msg in self._history:
        messages.append({"role": msg.role, "content": msg.content})

    messages.append({"role": "user", "content": input_text})

    # 流式调用LLM
    full_response = ""
    print("📝 实时响应: ", end="")
    for chunk in self.llm.think_stream(messages, **kwargs):
        full_response += chunk
        print(chunk, end="", flush=True)
        yield chunk

    print()  # 换行

    # 保存完整对话到历史记录
    self.add_message(Message(input_text, "user"))
    self.add_message(Message(full_response, "assistant"))
    print(f"✅ {self.name} 流式响应完成")

这样一来，就可以在用户端流式响应了。

工具管理

对于工具的管理，SimpleAgent 内部确实存在一个变量 ToolRegistry，但是外部没有权力直接使用 ToolRegistry 的方法，因此如果我们想要为 SimpleAgent 管理工具，应该对外暴露代理函数：

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
def add_tool(self, tool) -> None:
    """添加工具到Agent（便利方法）"""
    if not self.tool_registry:
        self.tool_registry = ToolRegistry()
        self.enable_tool_calling = True

    self.tool_registry.register_tool(tool)
    print(f"🔧 工具 '{tool.name}' 已添加")

def add_tool(self, func: Callable[[str], str]) -> None:
    """添加函数作为工具到Agent（便利方法）"""
    if not self.tool_registry:
        self.tool_registry = ToolRegistry()
        self.enable_tool_calling = True

    # 使用函数名称作为工具名称
    tool_name = func.__name__
    description = func.__doc__ or "无描述"

    self.tool_registry.register_function(
        name=tool_name,
        description=description,
        func=func
    )
    print(f"🔧 函数工具 '{tool_name}' 已添加")

def has_tools(self) -> bool:
    """检查是否有可用工具"""
    return self.enable_tool_calling and self.tool_registry is not None

def remove_tool(self, tool_name: str) -> bool:
    """移除工具（便利方法）"""
    if self.tool_registry:
        self.tool_registry.unregister(tool_name)
        return True
    return False

def list_tools(self) -> list:
    """列出所有可用工具"""
    if self.tool_registry:
        return self.tool_registry.get_tools_description()
    return []

对于添加函数 add_tool ，我们进行了重载，同时支持 Tool 类工具和函数式工具。

除此之外，工具的增删改查全部都是在 ToolRegistry 的基础上的代理。

测试

编写函数测试 SimpleAgent 的主要功能：

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
# test_simple_agent.py
from dotenv import load_dotenv
from core.llm import LLM
from agent.simpleAgent import SimpleAgent
from tool.toolRegistry import ToolRegistry
from tool.builtin.calculator import calculator
from tool.builtin.search import SearchTool


# 加载环境变量
load_dotenv()

# 创建LLM实例
llm = LLM()

# 测试1:基础对话Agent（无工具）
print("=== 测试1:基础对话 ===")
basic_agent = SimpleAgent(
    name="基础助手",
    llm=llm,
    system_prompt="你是一个友好的AI助手，请用简洁明了的方式回答问题。"
)

response1 = basic_agent.run("你好，请介绍一下自己")
print(f"基础对话响应: {response1}\n")

# 测试2:带工具的Agent
print("=== 测试2:工具增强对话 ===")
tool_registry = ToolRegistry()

tool_registry.register_function(
    name="calculator",
    description="简单的数学计算工具，支持基本运算(+,-,*,/)和sqrt函数",
    func=calculator
)

enhanced_agent = SimpleAgent(
    name="增强助手",
    llm=llm,
    system_prompt="你是一个智能助手，可以使用工具来帮助用户。",
    tool_registry=tool_registry,
    enable_tool_calling=True
)

response2 = enhanced_agent.run("请帮我计算 15 * 8 + 32")
print(f"工具增强响应: {response2}\n")

# 测试3:流式响应
print("=== 测试3:流式响应 ===")
print("流式响应: ", end="")
for chunk in basic_agent.stream_run("请解释什么是人工智能"):
    pass  # 内容已在stream_run中实时打印

# 测试4:动态添加工具，使用重载的 add_tool 方法
print("\n=== 测试4:动态工具管理 ===")
print(f"添加工具前: {basic_agent.has_tools()}")
basic_agent.add_tool(calculator)
print(f"添加工具后: {basic_agent.has_tools()}")
basic_agent.add_tool(SearchTool)
print(f"可用工具: {basic_agent.list_tools()}")

# 查看对话历史
print(f"\n对话历史: {len(basic_agent.get_history())} 条消息")

结果：

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
> python .\__testSimpleAgent.py

=== 测试1:基础对话 ===
✅ 基础助手 初始化完成，工具调用: 禁用
🤖 基础助手 正在处理: 你好，请介绍一下自己
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

✅ 基础助手 响应完成
基础对话响应: 你好！我是一个基于人工智能的助手，由计算机生成，用于回答问题、提供信息、帮助解决问题。我能处理文字，支持多种语言，但没有意识和情感，会尽力提供准确、有用的回答。有什么我可以帮忙的吗？

=== 测试2:工具增强对话 ===
✅ 工具 'calculator' 已注册。
✅ 增强助手 初始化完成，工具调用: 启用
🤖 增强助手 正在处理: 请帮我计算 15 * 8 + 32
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🔧 检测到 1 个工具调用
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

✅ 增强助手 响应完成
工具增强响应: 计算 15 × 8 + 32 的结果：

15 × 8 = 120
120 + 32 = 152

所以最终答案是 **152**。

=== 测试3:流式响应 ===
流式响应: 🌊 基础助手 开始流式处理: 请解释什么是人工智能
📝 实时响应: 🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

人工智能（AI）是指让计算机或机器模拟人类智能行为的技术。它通过分析数据、识别模式和学习经验，来完成通常需要人类智慧的任务，比如理解语言、辨识图像、做决策和解决问题。简单来说，就是让机器具备“学习”和“思考”的能力，像你的语音助手、推荐算法或自动驾驶汽车，都是人工智能的应用。
✅ 基础助手 流式响应完成

=== 测试4:动态工具管理 ===
添加工具前: False
✅ 工具 'calculator' 已注册。
🔧 函数工具 'calculator' 已添加
添加工具后: True
可用工具: - calculator: 简单的数学计算函数

对话历史: 4 条消息

其中，测试3是流式响应，在对话框内流式输出。

ReActAgent

ReAct 模式中，大语言模型需要经历 思考-行动-观察 的流程，而第四章中我们实现的 ReActAgent，每一个循环分别三次调用 LLM，我们需要让 ReActAgent 继承抽象类 Agent。

grill-with-docs 的建议

使用 /grill-with-docs 设计时，它提出了以下问题，我把我认为比较重要的问题贴上来：

ReflectionAgent 需要集成工具（Tool）吗？

看 reflectionPrompt.py 里的三个模板：

  1. initial — 根据 {task} 生成答案
  2. reflect — 根据 {task} + {content} 做自我批评
  3. refine — 根据 {task} + {last_attempt} + {feedback} 改进答案

这三个阶段都没有工具调用的语义。这是一个纯”生成 → 自省 → 改进”的链路，和 ReActAgent 的”思考 → 行动 → 观察”完全不同。

我的建议：不需要集成 ToolRegistry。ReflectionAgent 的核心价值是自我反思与迭代优化，工具调用是 ReActAgent
的职责。加工具反而让两个 Agent 的边界模糊。

反思循环次数——执行一轮”生成→反思→改进”就结束，还是支持多轮迭代？

ReActAgent 有 max_iterations（默认 5）来控制 Thought-Action 循环。ReflectionAgent 类似地可以：

  • 方案A：固定一轮 —— generate → reflect → refine，结束。简单直接。
  • 方案B：可配置多轮 —— 把上一轮的 refine 结果再送入 reflect → refine，循环直到 max_iterations 或者 feedback
    中模型自己判定”已经够好了”。

我的建议：方案B，可配置多轮，默认 2-3 轮。反思的本质就是反复打磨，只做一轮有时候改进不够明显。保留一个 max_reflections
参数（默认 2），让调用方可以调整。

构造函数参数设计

ReActAgent 的构造函数签名为：

def init(self, name, llm, system_prompt=None, config=None,
tool_registry=None, max_iterations=5, is_stream=False, verbose=True)

ReflectionAgent 目前是个空壳。按我们的决定，它需要这些参数：

┌──────────────────────────────────┬────────────────────────────────────────────────────┐
│ 参数 │ 建议 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ name, llm, system_prompt, config │ 继承基类，保留 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ tool_registry │ 不要 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ max_reflections │ 加，默认 2 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ is_stream │ 不要，你已明确不需要流式 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ verbose │ 保留，打印每轮的反思/优化日志 │
├──────────────────────────────────┼────────────────────────────────────────────────────┤
│ prompts │ 加，允许外部覆盖提示词模板，默认用 DEFAULT_PROMPTS │
└──────────────────────────────────┴────────────────────────────────────────────────────┘

我的建议是：

def init(self, name, llm, system_prompt=None, config=None,
max_reflections=2, prompts=None, verbose=True)

  • prompts 默认 None，运行时取 DEFAULT_PROMPTS，允许用户自定义三个阶段的提示词
  • max_reflections 控制反思-优化循环的最大次数
  • verbose=True 打印每轮的反思结果和优化结果

循环终止条件

多轮反思循环需要一个终止条件，否则会跑到 max_reflections 上限。有两种思路：

  • 方案A（纯计数）：跑满 max_reflections 轮，不提前终止。简单稳定，每轮都在前一轮基础上打磨。
  • 方案B（智能判定）：每轮 reflect 之后检查 feedback 内容，如果模型在 feedback 中明确表示”答案已经足够好了 /
    没有更多改进空间”，就提前终止，避免浪费 token。

我的建议：方案A，纯计数。让模型在 prompt 里自己判断”还需要不需要改进”不可靠，容易过早终止或者反复纠结。而且
max_reflections 默认只有 2，每轮 refine 都在进步，不差那一轮。简单可预测。

(这里由我出面采用可选择智能判断+纯计数：我们加上一个输入参数auto_finish:bool，如果true就支持智能终止，如果false就纯计数终止)

好，加 auto_finish: bool，默认 False（纯计数），设为 True 时智能判定提前终止。

run() 的返回值设计

ReActAgent 的 run(input_text) -> str 只返回最终答案。但对于 ReflectionAgent，中间产物（初始答案、每轮的
feedback、每轮的优化结果）也有价值。

  • 方案A（纯字符串）：run(input_text) -> str，只返回最终 refined 答案，和 ReActAgent 保持一致。
  • 方案B（结构化返回）：run(input_text) -> dict，返回 {“final_answer”: …, “initial_answer”: …, “reflections”:
    [{“feedback”: …, “refined”: …}, …]}，让调用方能拿到完整的反思链路。

我的建议：方案A。和基类 Agent 的签名 -> str 保持一致，而且 SimpleAgent、ReActAgent 都返回 str。中间产物可以通过
verbose=True 打印出来，调用方如果真想拿到中间结果，可以看 history。

auto_finish=True 时的智能终止机制

如果开启智能终止，需要一个明确的信号来判定”不用再改了”。有两种做法：

  • 方案A（prompt 内嵌）：在 reflect 提示词中要求模型，如果认为无需改进就输出特定标记（比如
    [NO_IMPROVEMENT_NEEDED]），Agent 解析到就终止。
  • 方案B（对比判定）：把 refined 答案和上一轮答案做相似度对比（比如用 LLM 或简单字符串相似度），如果变化很小就终止。

我的建议：方案A。方案B 要么引入额外的 LLM 调用（浪费），要么用字符串相似度不可靠。方案A 只需在 reflect prompt
里加一句”如果你认为答案已经完美无需修改，请回复 [NO_IMPROVEMENT_NEEDED]”，然后正则匹配即可，零额外成本。

历史记录管理

ReActAgent 会在 _history 中保存用户输入和最终回复。ReflectionAgent 有多轮中间产物（initial answer、每轮的
feedback、每轮的 refined answer），这些是否都要存入 self._history？

  • 方案A（只存最终）：和 ReActAgent 一样，只存 user input + final answer，保持 history 干净。
  • 方案B（全存）：把每轮的反思和优化都作为 assistant message 存入 history，完整记录思考链路。

我的建议：方案B。反思过程本身就是 Agent 的”思考”，存入 history 让后续对话有完整的上下文。而且 ReflectionAgent
没有工具调用，不会让 history 膨胀过快。

auto_finish 的判定逻辑细节

方案A 说了在 reflect prompt 中要求模型输出 [NO_IMPROVEMENT_NEEDED] 标记。具体怎么实现？

  • 做法①：在 reflect 的 prompt 模板里追加一句”如果你认为无需改进，输出 [NO_IMPROVEMENT_NEEDED]”，然后正则匹配。这会改动
    DEFAULT_PROMPTS。
  • 做法②：不改 DEFAULT_PROMPTS，在 Agent 内部拿到 reflect 结果后，额外调一次 LLM
    小请求判定”这反馈是否意味着无需改进”。多一次 LLM 调用。
  • 做法③：不改 prompt 模板，直接用正则检测 reflect 结果中是否包含类似”完美/无需修改/已经很好”等关键词。不可靠。

我的建议：做法①。改动最小，零额外成本，判定最可靠。DEFAULT_PROMPTS[“reflect”]
末尾加一句类似”如果答案已经足够好、无需改进，请只回复 [NO_IMPROVEMENT_NEEDED]”。

每轮 refine 的消息构建方式

ReflectionAgent 的三个阶段是三次独立的 LLM 调用。以 max_reflections=2 为例：

第1轮: initial(生成) → reflect(反思) → refine(优化)
第2轮: reflect(再反思) → refine(再优化) → 返回

每轮 reflect 的输入是 {task} + {content}（上一步的答案），refine 的输入是 {task} + {last_attempt} + {feedback}。

那第二轮 reflect 的 {content} 应该传什么？

  • 方案A（只传最新）：传上一轮的 refined 答案，不带上之前的 feedback。每轮独立。
  • 方案B（累积上下文）：把之前的 initial + feedback₁ + refined₁ 拼成一个上下文，让模型看到完整演进过程。

我的建议：方案A。refine 已经把 feedback 吸收了，输出的 refined 答案就是”综合了反馈的结果”，下一轮 reflect
只需要审视这个最新版。累积上下文会让消息越来越长，且模型可能被旧反馈干扰。

verbose=True 时的日志输出格式

每轮反思循环中会输出哪些信息？参考 ReActAgent 的 verbose 会打印 Thought、Action、Observation。ReflectionAgent
的日志同样需要清晰展示反思链路。

我建议每轮输出类似：

[ReflectionAgent] 第 1/2 轮反思
📝 初始答案: <initial answer 摘要>
🔍 反思反馈: <feedback 摘要>
✨ 优化结果: <refined answer 摘要>

system_prompt 参数的处理

基类 Agent.init 接受 system_prompt 参数，但 ReActAgent 实际上忽略它，始终用自己的 REACT_PROMPT。ReflectionAgent
的三个阶段各有自己的提示词（来自 prompts 参数的三个 key），跟 system_prompt 没有直接关系。

我的建议：忽略 system_prompt，和 ReActAgent 保持一致。prompts 参数完全替代 system_prompt 的作用。

prompts 参数的合并策略

用户传入自定义 prompts 时，是完全替换 DEFAULT_PROMPTS，还是合并（只覆盖传入的 key，未传入的保持默认）？

比如用户只传 {“reflect”: “自定义反思提示词”}，那 initial 和 refine 是否保留默认？

我的建议：合并策略。更灵活，用户只需覆盖他想定制的部分。

初始化

除了 Agent 基类中包含的输入参数，我们还需要给定：

  • max_steps：最大迭代次数，因为 ReAct 需要多次观察才能得出最 LLM 认为最完备的响应；
  • custom_prompt：让用户能够自定义自己的 ReAct 提示词模板；
  • verbose：是否将 思考-行动-观察 的全过程打印出来；
  • is_stream：是否流式输出。
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
def __init__(
    self,
    name: str,
    llm: LLM,
    system_prompt: Optional[str] = None,
    config: Optional[Config] = None,
    tool_registry: Optional[ToolRegistry] = None,
    max_iterations: int = 5,
    is_stream: bool = False,
    verbose: bool = True,
):
    super().__init__(name, llm, system_prompt, config)
    self.tool_registry = tool_registry
    self.max_iterations = max_iterations
    self.is_stream = is_stream
    self.verbose = verbose
    self._execution_trace: List[str] = []   # 本轮执行轨迹

    # 统计工具数量
    tool_count = 0
    if tool_registry:
        desc = tool_registry.get_tools_description()
        if desc and desc != "暂无可用工具":
            tool_count = len([l for l in desc.split("\n") if l.strip().startswith("-")])

    print(
        f"✅ {name} (ReAct) 初始化完成 | "
        f"流式: {'开' if is_stream else '关'} | "
        f"最大迭代: {max_iterations} | "
        f"工具: {tool_count}个"
    )

ReActPrompt

在之前的 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
REACT_PROMPT = """你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题，然后调用合适的工具来获取信息，最终给出准确的答案。

## 可用工具
{tools}

## 工作流程
请严格按照以下格式进行回应，每次只能执行一个步骤:

Thought: 分析当前问题，思考需要什么信息或采取什么行动。
Action: 选择一个行动，格式必须是以下之一:
- `{{tool_name}}[{{tool_input}}]` - 调用指定工具
- `Finish[最终答案]` - 当你有足够信息给出最终答案时

## 重要提醒
1. 每次回应必须包含Thought和Action两部分
2. 工具调用的格式必须严格遵循:工具名[参数]
3. 只有当你确信有足够信息回答问题时，才使用Finish
4. 如果工具返回的信息不够，继续使用其他工具或相同工具的不同参数

## 当前任务
**Question:** {question}

## 执行历史
{history}

现在开始你的推理和行动:
"""

同步输出 _run_batch

由于我们要支持流式输出和同步输出，所以重写的 run 函数只用来当个代理就行，检查输入的 is_stream，判断走哪个代理函数。如果是同步输出，就走 _run_batch，否则走 _run_stream，这里给出前者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def _run_batch(self, input_text: str, **kwargs) -> str:
    """批量模式：每轮用 llm.think() 获取完整响应。"""
    messages = self._build_initial_messages(input_text)

    for i in range(self.max_iterations):
        if self.verbose:
            print(f"\n--- ReAct 第 {i + 1}/{self.max_iterations} 轮 ---")

        response = self.llm.think(messages, **kwargs)
        if response is None:
            return "❌ LLM 调用失败，请检查 API 配置。"

        result = self._process_response(response, messages)
        if result is not None:
            self._save_to_history(input_text, result)
            return result

    return self._force_final_answer(messages, input_text, **kwargs)

可以看到，大致流程就是：1.拼接提示词；2.调用 LLM；3.解析 LLM 的响应；4.后处理&拼接最终结果。

先后调用了多个函数，接下来细说。

_build_initial_messages

我们知道，openAI 的提示词标准，将提示词分为 user、 assistant、 system、 tool 四种 role，该函数主要关注的是如何构建符合 openAI 规范的提示词。

首先将 ReActPrompt 和 工具列表作为系统提示词，role 为 system，值得关注的是 ToolRegistry 的 get_tools_description 返回的就是符合 openAI 规范的 kv 对，其 role 则是 tool。将 tool 包含在 system 内，更容易激发 LLM 对工具意图。

随后将历史消息加入，Message 类的函数 to_dict 同样返回带有 role 的提示词。

最后，加入用户消息，至此，提示词完成，它能够完全符合 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
def _build_initial_messages(self, input_text: str) -> List[Dict[str, str]]:
    """构建初始消息列表：system（含 ReAct 格式说明 + 工具列表）
    + 历史消息 + 当前用户消息。"""
    messages: List[Dict[str, str]] = []

    # 1. 系统消息
    tools_desc = (
        self.tool_registry.get_tools_description()
        if self.tool_registry
        else "暂无可用工具"
    )
    system_content = REACT_PROMPT.format(
        tools=tools_desc,
        question=input_text,
        history="（尚未执行任何操作）",
    )
    messages.append({"role": "system", "content": system_content})

    # 2. 历史消息（来自父类 self._history）
    for msg in self._history:
        messages.append(msg.to_dict())

    # 3. 当前用户消息
    messages.append({"role": "user", "content": input_text})

    return messages

_process_response

回想第四章中的做法，让 LLM 进行 think 之后，我们需要从冗长的结构化 JSON 输出中提取出我们事先让 LLM 输出的关键词，也即 Thought、Action 和 Finish。

不过，这个函数也是嵌套的：先是调用 _parse_react_output 函数，提取三个关键词的文本，然后在本函数体内进行校验，比如是否存在 Action（工具意图）、是否存在 Finish（LLM 认为答案完备，可以结束会话了）。

如果存在工具意图，那么就要调用 _execute_action 函数，从而调用工具，随后将结果写入 Obversation。

_force_final_answer

注意到这个函数，在 _run_batch 和 _run_stream 中的最后均调用了，其作用就是为了给响应兜底。

流式输出 _run_stream

和 SimpleAgent 的做法一致，使用 llm.think_stream 函数，使用 yield 实现增量片段即时返回。

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
def _run_stream(self, input_text: str, **kwargs) -> str:
    """流式模式：每轮用 llm.think_stream() 实时输出 chunk。

    等本轮全部收完后再解析（不做增量解析），保证解析正确性。
    """
    messages = self._build_initial_messages(input_text)

    for i in range(self.max_iterations):
        if self.verbose:
            print(f"\n--- ReAct 第 {i + 1}/{self.max_iterations} 轮 ---")

        full_response = ""
        for chunk in self.llm.think_stream(messages, **kwargs):
            if chunk:
                print(chunk, end="", flush=True)
                full_response += chunk
        print()  # 本轮流式结束，换行

        if not full_response:
            return "❌ LLM 流式调用失败，请检查 API 配置。"

        result = self._process_response(full_response, messages)
        if result is not None:
            self._save_to_history(input_text, result)
            return result

    return self._force_final_answer(messages, input_text, **kwargs)

测试

书写测试函数，查看 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
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
# __testReActAgent.py
# ReActAgent 功能测试 —— 覆盖批量/流式模式、函数工具/Tool对象、执行轨迹、历史管理
from dotenv import load_dotenv
from core.llm import LLM
from agent.reActAgent import ReActAgent
from tool.toolRegistry import ToolRegistry
from tool.tool import Tool
from tool.builtin.calculator import calculator

load_dotenv()

llm = LLM()

# ================================================================
# 测试1: 批量模式 + 函数工具（calculator 通过 register_function 注册）
# ================================================================
print("=" * 60)
print("测试1: 批量模式 + 函数工具")
print("=" * 60)

registry1 = ToolRegistry()
registry1.register_function(
    name="calculator",
    description="数学计算工具，支持加减乘除和 sqrt 函数",
    func=calculator,
)

agent1 = ReActAgent(
    name="ReAct数学助手",
    llm=llm,
    system_prompt="你是一个擅长数学的AI助手。",
    tool_registry=registry1,
    max_iterations=5,
    is_stream=False,
    verbose=True,
)

response1 = agent1.run("请帮我计算 15 * 8 + 32 的结果")
print(f"\n📌 最终响应: {response1}\n")

# ================================================================
# 测试2: 批量模式 + Tool 对象（自定义 EchoTool 验证 is_tool 分支）
# ================================================================
print("=" * 60)
print("测试2: 批量模式 + Tool 对象")
print("=" * 60)


class EchoTool(Tool):
    """回显工具：把输入原样返回，用于测试 Tool 对象执行路径"""

    def __init__(self):
        super().__init__(
            name="echo",
            description="回显工具，将输入内容原样返回。可用于确认信息。",
        )

    def run(self, parameters: dict) -> str:
        return f"[回显] {parameters.get('input', '')}"

    def get_parameters(self):
        return []


registry2 = ToolRegistry()
registry2.register_tool(EchoTool())

agent2 = ReActAgent(
    name="ReAct回显助手",
    llm=llm,
    system_prompt="你是一个测试助手，需要回显确认用户输入。",
    tool_registry=registry2,
    max_iterations=3,
    is_stream=False,
    verbose=True,
)

response2 = agent2.run("请用 echo 工具帮我确认: Hello ReAct!")
print(f"\n📌 最终响应: {response2}\n")

# ================================================================
# 测试3: 流式模式（is_stream=True，边输出边收集）
# ================================================================
print("=" * 60)
print("测试3: 流式模式")
print("=" * 60)

registry3 = ToolRegistry()
registry3.register_function(
    name="calculator",
    description="数学计算工具",
    func=calculator,
)

agent3 = ReActAgent(
    name="ReAct流式助手",
    llm=llm,
    system_prompt="你是一个擅长数学的AI助手。",
    tool_registry=registry3,
    max_iterations=5,
    is_stream=True,   # 开启流式
    verbose=True,
)

response3 = agent3.run("帮我算一下 100 / 4 + 7")
print(f"\n📌 最终响应: {response3}\n")

# ================================================================
# 测试4: 执行轨迹（get_execution_trace）
# ================================================================
print("=" * 60)
print("测试4: 执行轨迹")
print("=" * 60)

# 复用 agent1（已有历史），再跑一个问题
trace_agent = ReActAgent(
    name="ReAct轨迹助手",
    llm=llm,
    tool_registry=registry1,  # 用 calculator 的那个 registry
    max_iterations=5,
    is_stream=False,
    verbose=False,
)

trace_agent.run("计算 (3 + 5) * 2")
trace = trace_agent.get_execution_trace()
print(f"执行轨迹共 {len(trace)} 步:")
for i, step in enumerate(trace, 1):
    print(f"  --- 步骤 {i} ---")
    print(f"  {step[:200]}{'...' if len(step) > 200 else ''}")

# ================================================================
# 测试5: 对话历史管理（get_history / clear_history）
# ================================================================
print("\n" + "=" * 60)
print("测试5: 对话历史管理")
print("=" * 60)

hist_agent = ReActAgent(
    name="ReAct历史助手",
    llm=llm,
    tool_registry=registry1,
    max_iterations=5,
    is_stream=False,
    verbose=False,
)

# 先跑一轮
hist_agent.run("1 + 1 等于几？")
print(f"第一轮后历史条数: {len(hist_agent.get_history())}")

# 再跑一轮
hist_agent.run("2 + 2 等于几？")
print(f"第二轮后历史条数: {len(hist_agent.get_history())}")

# 清空
hist_agent.clear_history()
print(f"清空后历史条数: {len(hist_agent.get_history())}")

# 再跑一轮确认清空后正常工作
hist_agent.run("3 + 3 等于几？")
print(f"清空后重新对话，历史条数: {len(hist_agent.get_history())}")

print("\n" + "=" * 60)
print("✅ 所有测试完成")
print("=" * 60)

结果：

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
> python __testReActAgent.py

============================================================
测试1: 批量模式 + 函数工具
============================================================
✅ 工具 'calculator' 已注册。
✅ ReAct数学助手 (ReAct) 初始化完成 | 流式: 关 | 最大迭代: 5 | 工具: 1个

--- ReAct 第 1/5 轮 ---
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 Thought: 我需要计算 15 * 8 + 32 的结果，我可以使用计算器工具直接计算整个表达式。
🔧 Action: calculator[15 * 8 + 32]
👁️  Observation: 152

--- ReAct 第 2/5 轮 ---
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 Thought: 计算器返回了结果 152，现在我可以给出最终答案。
🎯 最终答案: 152

📌 最终响应: 152

============================================================
测试2: 批量模式 + Tool 对象
============================================================
✅ 工具 'echo' 已注册。
✅ ReAct回显助手 (ReAct) 初始化完成 | 流式: 关 | 最大迭代: 3 | 工具: 1个

--- ReAct 第 1/3 轮 ---
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 Thought: 用户想要确认 "Hello ReAct!"，我可以使用 echo 工具来回显这个消息，然后返回结果。
🔧 Action: echo[Hello ReAct!]
👁️  Observation: [回显] Hello ReAct!

--- ReAct 第 2/3 轮 ---
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 Thought: echo工具已成功回显了"Hello ReAct!"，任务完成。
🎯 最终答案: 已确认: Hello ReAct!

📌 最终响应: 已确认: Hello ReAct!

============================================================
测试3: 流式模式
============================================================
✅ 工具 'calculator' 已注册。
✅ ReAct流式助手 (ReAct) 初始化完成 | 流式: 开 | 最大迭代: 5 | 工具: 1个

--- ReAct 第 1/5 轮 ---
🧠 正在流式调用 deepseek-v4-pro 模型...
Thought: 用户要求计算 100 / 4 + 7。这是一个简单的算术表达式，需要先除后加。我将使用计算器工具来得出准确结果。
Action: calculator[100/4+7]
🧠 Thought: 用户要求计算 100 / 4 + 7。这是一个简单的算术表达式，需要先除后加。我将使用计算器工具来得出准确结果。
🔧 Action: calculator[100/4+7]
👁️  Observation: 32.0

--- ReAct 第 2/5 轮 ---
🧠 正在流式调用 deepseek-v4-pro 模型...
Thought: 计算器返回了结果 32.0，表达式 100/4+7 等于 32。可以给出最终答案了。
Action: Finish[32]
🧠 Thought: 计算器返回了结果 32.0，表达式 100/4+7 等于 32。可以给出最终答案了。
🎯 最终答案: 32

📌 最终响应: 32

============================================================
测试4: 执行轨迹
============================================================
✅ ReAct轨迹助手 (ReAct) 初始化完成 | 流式: 关 | 最大迭代: 5 | 工具: 1个
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

执行轨迹共 1 步:
  --- 步骤 1 ---
  Thought: 我需要计算 (3+5)*2 的结果。计算器工具支持加减乘除，应该可以直接计算整个表达式。
Action: calculator[(3+5)*2]
Observation: 16

============================================================
测试5: 对话历史管理
============================================================
✅ ReAct历史助手 (ReAct) 初始化完成 | 流式: 关 | 最大迭代: 5 | 工具: 1个
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

第一轮后历史条数: 2
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

第二轮后历史条数: 4
清空后历史条数: 0
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

清空后重新对话，历史条数: 2

============================================================
✅ 所有测试完成
============================================================

ReflectionAgent

第四章中提到，Reflection 的模式是 执行 => 反思 => 优化，我们同样继承抽象类 Agent 进行实现。

不过，我们在之前实现的过程中，没有集成工具，所以这个 Agent 上我们也不集成工具，而是专注于为 LLM 赋予拆解任务的能力。

另外，也不再支持流式输出。

reflectionPrompt

官方给出了标准的提示词，我们之间拿来用。不过，我们在里面改一点：如果 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
DEFAULT_PROMPTS = {
    "initial": """
请根据以下要求完成任务:

任务: {task}

请提供一个完整、准确的回答。
""",
    "reflect": """
请仔细审查以下回答，并找出可能的问题或改进空间:

# 原始任务:
{task}

# 当前回答:
{content}

请分析这个回答的质量，指出不足之处，并提出具体的改进建议。
如果回答已经完美、无需任何改进，请只回复 [NO_IMPROVEMENT_NEEDED]。
""",
    "refine": """
请根据反馈意见改进你的回答:
 
# 原始任务:
{task}

# 上一轮回答:
{last_attempt}

# 反馈意见:
{feedback}

请提供一个改进后的回答。
"""
}

继承 Agent

创建 ReflectionAgent，其初始化：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def __init__(
    self,
    name: str,
    llm: LLM,
    system_prompt: Optional[str] = None,
    config: Optional[Config] = None,
    max_reflections: int = 2,
    auto_finish: bool = False,
    prompts: Optional[Dict[str, str]] = None,
    verbose: bool = True,
):
    super().__init__(name, llm, system_prompt, config)
    self.max_reflections = max_reflections
    self.auto_finish = auto_finish
    self.verbose = verbose

    # 合并提示词：用户传入的覆盖默认值
    self.prompts = {**DEFAULT_PROMPTS, **(prompts or {})}

    print(
        f"✅ {name} (Reflection) 初始化完成 | "
        f"最大反思轮数: {max_reflections} | "
        f"智能终止: {'开' if auto_finish else '关'}"
    )

相比起 Agent 已有的多个参数，我们还添加了以下参数：

  • max_reflection：执行 => 反思 => 优化 的次数；

  • auto_finish：是否智能终止，即是否由 LLM 决定是否接受循环；

  • verbose：是否打印中间结果，与 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
def _generate_initial(self, task: str, **kwargs) -> Optional[str]:
    """第一阶段：生成初始答案。"""
    messages = [
        {"role": "system", "content": self.prompts["initial"].format(task=task)},
        {"role": "user", "content": task},
    ]
    return self.llm.think(messages, **kwargs)

def _reflect(self, task: str, content: str, **kwargs) -> Optional[str]:
    """第二阶段：反思当前答案，给出改进反馈。"""
    prompt = self.prompts["reflect"].format(task=task, content=content)
    messages = [
        {"role": "system", "content": prompt},
        {"role": "user", "content": "请审查以上回答并给出反馈意见。"},
    ]
    return self.llm.think(messages, **kwargs)

def _refine(self, task: str, last_attempt: str, feedback: str, **kwargs) -> Optional[str]:
    """第三阶段：根据反馈优化答案。"""
    messages = [
        {"role": "system", "content": self.prompts["refine"].format(
            task=task, last_attempt=last_attempt, feedback=feedback
        )},
        {"role": "user", "content": "请根据反馈意见提供改进后的回答。"},
    ]
    return self.llm.think(messages, **kwargs)

这三个函数相当干净：几乎只是代理了一下 llm.think 函数，将 LLM 的输出直接返回。在实现上也相当类似，唯一的区别是输入的提示词，以告诉 LLM 这个三大阶段的哪个阶段。

run 函数

还是重新 run 函数，但是相比起 ReActAgent 的实现来说就简单多了，我们已经定义好了三大阶段的独立函数，在 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
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
def run(self, input_text: str, **kwargs) -> str:
    """运行 Reflection Agent。

    流程: 初始生成 → 循环(反思 → 优化) → 返回最终答案
    """
    # 阶段 1: 初始生成
    initial_answer = self._generate_initial(input_text, **kwargs)
    if initial_answer is None:
        return "❌ LLM 调用失败，请检查 API 配置。"

    if self.verbose:
        print(f"\n📝 初始答案: {initial_answer[:120]}{'...' if len(initial_answer) > 120 else ''}")

    # 保存用户输入 + 初始答案到 history
    self._save_to_history(input_text, initial_answer)

    current_answer = initial_answer

    # 阶段 2: 反思-优化循环
    for i in range(self.max_reflections):
        if self.verbose:
            print(f"\n--- Reflection 第 {i + 1}/{self.max_reflections} 轮 ---")

        # 反思
        feedback = self._reflect(input_text, current_answer, **kwargs)
        if feedback is None:
            if self.verbose:
                print("⚠️ 反思阶段 LLM 调用失败，返回当前答案")
            return current_answer

        if self.verbose:
            print(f"🔍 反思反馈: {feedback[:120]}{'...' if len(feedback) > 120 else ''}")

        # 智能终止检测
        if self._check_no_improvement(feedback):
            if self.auto_finish:
                if self.verbose:
                    print("✅ 模型判定无需改进，提前终止")
                self.add_message(Message(feedback, "assistant"))
                return current_answer
            # auto_finish=False: 忽略标记，继续执行 refine

        # 优化
        refined_answer = self._refine(input_text, current_answer, feedback, **kwargs)
        if refined_answer is None:
            if self.verbose:
                print("⚠️ 优化阶段 LLM 调用失败，返回当前答案")
            return current_answer

        if self.verbose:
            print(f"✨ 优化结果: {refined_answer[:120]}{'...' if len(refined_answer) > 120 else ''}")

        # 保存本轮反思产物到 history
        self.add_message(Message(feedback, "assistant"))
        self.add_message(Message(refined_answer, "assistant"))

        current_answer = refined_answer

    return current_answer

测试

写一个函数进行测试：

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
# __testReflectionAgent.py
# ReflectionAgent 功能测试 —— 覆盖反思循环、智能终止、历史管理、自定义提示词
from dotenv import load_dotenv
from core.llm import LLM
from agent.reflectionAgent import ReflectionAgent

load_dotenv()

llm = LLM()

# ================================================================
# 测试1: 基础反思模式（auto_finish=False，走满 max_reflections 轮）
# ================================================================
print("=" * 60)
print("测试1: 基础反思模式（max_reflections=2, 纯计数）")
print("=" * 60)

agent1 = ReflectionAgent(
    name="Reflection基础助手",
    llm=llm,
    max_reflections=2,
    auto_finish=False,
    verbose=True,
)

response1 = agent1.run("请解释什么是递归，并给出一个简单的Python示例")
print(f"\n📌 最终响应: {response1}\n")

# ================================================================
# 测试2: 智能终止模式（auto_finish=True）
# ================================================================
print("=" * 60)
print("测试2: 智能终止模式（auto_finish=True）")
print("=" * 60)

agent2 = ReflectionAgent(
    name="Reflection智能终止助手",
    llm=llm,
    max_reflections=3,
    auto_finish=True,
    verbose=True,
)

response2 = agent2.run("1 + 1 等于几？")
print(f"\n📌 最终响应: {response2}\n")

# ================================================================
# 测试3: 单轮反思（max_reflections=1）
# ================================================================
print("=" * 60)
print("测试3: 单轮反思（max_reflections=1）")
print("=" * 60)

agent3 = ReflectionAgent(
    name="Reflection单轮助手",
    llm=llm,
    max_reflections=1,
    auto_finish=False,
    verbose=True,
)

response3 = agent3.run("用一句话概括机器学习的定义")
print(f"\n📌 最终响应: {response3}\n")

# ================================================================
# 测试4: 对话历史管理（get_history / clear_history）
# ================================================================
print("=" * 60)
print("测试4: 对话历史管理")
print("=" * 60)

hist_agent = ReflectionAgent(
    name="Reflection历史助手",
    llm=llm,
    max_reflections=1,
    auto_finish=False,
    verbose=False,
)

# 第一轮
hist_agent.run("Python 是什么？")
# 历史应包含: 1条user + 1条assistant(初始) + 1条assistant(反馈) + 1条assistant(优化)
print(f"第一轮后历史条数: {len(hist_agent.get_history())}")
for i, msg in enumerate(hist_agent.get_history()):
    print(f"  [{i}] {msg.role}: {msg.content[:80]}{'...' if len(msg.content) > 80 else ''}")

# 第二轮
hist_agent.run("Java 是什么？")
print(f"第二轮后历史条数: {len(hist_agent.get_history())}")

# 清空
hist_agent.clear_history()
print(f"清空后历史条数: {len(hist_agent.get_history())}")

# 清空后正常工作
hist_agent.run("C++ 是什么？")
print(f"清空后重新对话，历史条数: {len(hist_agent.get_history())}")

print()

# ================================================================
# 测试5: 自定义提示词（合并策略）
# ================================================================
print("=" * 60)
print("测试5: 自定义提示词")
print("=" * 60)

custom_prompts = {
    "initial": "请用专业的语气完成任务: {task}",
}

agent5 = ReflectionAgent(
    name="Reflection自定义助手",
    llm=llm,
    max_reflections=1,
    auto_finish=False,
    prompts=custom_prompts,
    verbose=False,
)

# 验证合并: initial 被覆盖，reflect 和 refine 保持默认
assert agent5.prompts["initial"] == custom_prompts["initial"], "initial 应被覆盖"
assert "[NO_IMPROVEMENT_NEEDED]" in agent5.prompts["reflect"], "reflect 应保留默认"
print("✅ 提示词合并验证通过")
print(f"  initial: {'已覆盖' if agent5.prompts['initial'] == custom_prompts['initial'] else '未覆盖'}")
print(f"  reflect: {'保持默认' if '[NO_IMPROVEMENT_NEEDED]' in agent5.prompts['reflect'] else '异常'}")
print(f"  refine:  {'保持默认' if agent5.prompts['refine'] != '' else '异常'}")
print()

response5 = agent5.run("请用一句话总结 Docker 的作用")
print(f"\n📌 最终响应: {response5}\n")

# ================================================================
print("=" * 60)
print("✅ 所有测试完成")
print("=" * 60)

PlanAndSolveAgent

Plan-and-Solve 的模式是将问题拆分成若干子问题，再依次解决子问题。

提示词

官方依然分别给出了 plan 和 solve 的提示词，我们直接拿来用：

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
# 默认规划器提示词模板
DEFAULT_PLANNER_PROMPT = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划:
```python
["步骤1", "步骤2", "步骤3", ...]
``
"""

#  默认执行器提示词模板
DEFAULT_EXECUTOR_PROMPT = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决"当前步骤"，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

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

# 当前步骤:
{current_step}

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

grill-with-docs 的建议

使用 /grill-with-docs 设计时，它提出了以下问题，我把我认为比较重要的问题贴上来：

Plan 阶段的输出解析

DEFAULT_PLANNER_PROMPT 要求 LLM 输出：

[“步骤1”, “步骤2”, “步骤3”, …]

怎么把这个字符串解析成真正的 Python list？

  • 方案A（ast.literal_eval）：安全解析，只接受字面量。但如果 LLM 输出格式稍有偏差（多了注释、少了引号）就直接失败。
  • 方案B（正则提取）：用 re.findall 匹配 “…” 字符串，容错性高。
  • 方案C（先 ast 后正则兜底）：先尝试 ast.literal_eval，失败则用正则提取引号内字符串。

我的建议：方案C。ast.literal_eval 正常情况最可靠，正则做兜底避免 LLM 格式不稳定导致整个 Agent 崩溃。

所有子任务执行完后，如何产出最终答案？

执行循环按计划逐步产出每个步骤的结果。执行完最后一步后：

  • 方案A（最后一步即最终答案）：执行完计划中所有步骤后，最后一步的结果就是最终答案。综合能力交给规划器——如果用户问的问题
    需要综合，规划器自然会把”总结”作为最后一步。
  • 方案B（最终汇总调用）：所有步骤执行完后，额外调一次 LLM，把所有步骤及其结果汇总成一个完整答案。

我的建议：方案A。规划器的职责就是把问题拆成合理的步骤序列，最后一步天然可以承担”给出最终答案”的角色。方案B 每次多花一次
LLM 调用，且与规划器的职责重叠。

步骤间的历史累积方式

Executor prompt 有一个 {history} 占位符。随着步骤推进，{history} 需要逐步填充。有两种做法：

  • 方案A（扁平累积）：{history} 是一段文本，格式为：
    步骤1: [步骤1描述]
    结果1: [步骤1的结果]

步骤2: [步骤2描述]
结果2: [步骤2的结果]

  • 方案B（只传上一步结果）：{history} 只包含上一步的结果，不附带更早的历史。

我的建议：方案A。后续步骤可能需要前面所有步骤的上下文（比如步骤3需要步骤1和步骤2的结果）。方案B
信息丢失，可能让后续步骤做出错误判断。

run() 的返回值设计

和 ReflectionAgent 一样的问题：

  • 方案A：run(input_text) -> str，只返回最终答案（最后一步的结果），和基类一致。
  • 方案B：run(input_text) -> dict，返回 {“final_answer”: …, “plan”: […], “step_results”: […]}
    让调用方能拿到完整执行链路。

我的建议：方案A。和 ReflectionAgent、ReActAgent 保持一致，只返回 str。中间产物（计划列表、每步结果）通过 verbose=True
打印，也可以通过 get_history() 追溯。

Plan 解析失败处理

选择：直接报错返回，不做兜底。

ast.literal_eval 和正则都失败，说明 LLM
输出的格式严重偏离要求。不降级为”整个问题当单步执行”，那样会隐藏问题根源。让调用方看到明确的错误信息。

self._history 管理

选择：保存（用户输入 + 计划 + 每步结果），全部标为 assistant。

[0] user: “用户原始问题”
[1] assistant: “📋 计划: [‘步骤1’, ‘步骤2’, ‘步骤3’]”
[2] assistant: “步骤1: … 结果: …”
[3] assistant: “步骤2: … 结果: …”
[4] assistant: “步骤3: … 结果: …”

和 ReflectionAgent 的全量存储策略一致，提供完整追溯链。

system_prompt 参数

选择：忽略。和 ReflectionAgent 一致，planner/executor 提示词完全替代 system_prompt 的作用。

verbose 输出格式

选择：中文日志，沿用 ReflectionAgent 风格：

📋 计划 (3 步):
1. 搜索相关数据
2. 计算统计指标
3. 给出最终结论

— 执行步骤 1/3 —
✅ 步骤 1 完成: <摘要>

— 执行步骤 2/3 —
✅ 步骤 2 完成: <摘要>

— 执行步骤 3/3 —
✅ 步骤 3 完成: <摘要>
🏁 最终答案: <最后一步结果>

最终完整设计汇总

PlanAndSolveAgent
├── 构造函数: (name, llm, system_prompt=None, config=None, verbose=True)
├── run(input_text) -> str:
│ ├── 阶段1 (_plan): LLM 生成步骤列表
│ │ ├── 解析: ast.literal_eval, 失败 → regex 兜底, 都失败 → 报错
│ │ └── 空列表 → 报错
│ ├── 阶段2 (_solve): 顺序执行每个步骤
│ │ ├── 每次调用 executor prompt，传入 {question}, {plan}, {history}, {current_step}
│ │ ├── history 扁平累积: “步骤N: … \n结果N: …”
│ │ └── LLM 返回 None → 中止，返回已累积结果
│ ├── 返回: 最后一步的结果 (str)
│ └── history: 全量保存 user输入 + 计划 + 每步结果

PlanAndSolveAgent 类的设计

第四章中，我们曾经构建过简易的 Plan-and-solve 模式的 Agent，其主要运行逻辑是：1.（Plan）首次调用 LLM，将初始任务拆分成若干个子任务；2.（Solve）对于若干个子任务，均调用 LLM 以解决，解决完所有子任务后，也就是完成了初始任务。

之后的就自己去看吧，与第四章并无本质区别。

总结

本章内容与第四章相似，我们仍然以从零到有构建四种 Agent 为终极目标，但比起第四章的即兴式编写代码，我们的开发路径遵照 底层 => 模板 => 成品 ，将上下文管理、大模型参数配置、Agent 的构建等进行了标准化，以这些开发成果为基础构建了一套最基本的 Agent 框架。

在 core 包内，我们从底层上对 Agent 进行了支持：标准的上下文存储对象 Message 将上下文的格式由简单的字符串固定为了严格的对象；我们构建了 LLM 参数统一配置中心 Config，避免了 LLM 参数的分散配置；随后我们使用 openAI 包，消防第四章的 HelloAgentLLm 类，构建了一个大模型调用接口 LLM；最后，我们构建了一个 Agent 抽象类，为构建一个 Agent 提供了规范。

在 tool 包内，我们主要对两种工具进行了支持：对象类和函数类。对于对象类，我们也构建了一个抽象类 Tool 作为工具的基类。而对于工具管理上，我们构建了一个 ToolRegistry，对工具进行统一的注册发现、管理和调用。随后，我们基于 Tool 接口和 Callable 函数引用构建了几个工具，并进行了测试。

在 agent 包内，我们基于 Agent 接口，对第四章中构建过的四种 Agent 进行了重构，使每一种 Agent 有着统一的构建方式和使用方法。

不过，当前的这个框架仅仅完成了最简单的功能，还有相当多的拓展空间，比如记忆、上下文工程、Agent 协议等，这些将在后续的章节中进行。

本章代码

core

llm.py

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
import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict
# 加载 .env 文件中的环境变量
load_dotenv()

class LLM:
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        # 1.初始化客户端。优先使用传入参数，如果未提供，则从环境变量加载。
        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文件中定义。")

        # 2.构建一个 OpenAI 规范的 LLM 调用接口
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用大语言模型进行思考，并返回其响应。
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            # 通过LLM接口调用大模型，response为流式响应(stream=True)
            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

        def think_stream(self, messages: List[Dict[str, str]], temperature: float = 0):
        """
        流式调用大语言模型进行思考，逐步返回其响应。
        """
        print(f"🧠 正在流式调用 {self.model} 模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                yield content

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

message.py

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
"""消息系统"""
from typing import Optional, Dict, Any, Literal
from datetime import datetime
from pydantic import BaseModel

# 定义消息角色的类型，限制其取值
MessageRole = Literal["user", "assistant", "system", "tool"]

class Message(BaseModel):
    """消息类"""
    
    content: str
    role: MessageRole
    timestamp: datetime = None
    metadata: Optional[Dict[str, Any]] = None
    
    def __init__(self, content: str, role: MessageRole, **kwargs):
        super().__init__(
            content=content,
            role=role,
            timestamp=kwargs.get('timestamp', datetime.now()),
            metadata=kwargs.get('metadata', {})
        )
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式（OpenAI API格式）"""
        return {
            "role": self.role,
            "content": self.content
        }
    
    def __str__(self) -> str:
        return f"[{self.role}] {self.content}"

config.py

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
"""配置管理"""
import os
from typing import Optional, Dict, Any
from pydantic import BaseModel

class Config(BaseModel):
    """HelloAgents配置类"""
    
    # LLM配置
    default_model: str = "gpt-3.5-turbo"
    default_provider: str = "openai"
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    
    # 系统配置
    debug: bool = False
    log_level: str = "INFO"
    
    # 其他配置
    max_history_length: int = 100
    
    @classmethod
    def from_env(cls) -> "Config":
        """从环境变量创建配置"""
        return cls(
            debug=os.getenv("DEBUG", "false").lower() == "true",
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            temperature=float(os.getenv("TEMPERATURE", "0.7")),
            max_tokens=int(os.getenv("MAX_TOKENS")) if os.getenv("MAX_TOKENS") else None,
        )
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return self.dict()

agent.py

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
"""Agent基类"""
from abc import ABC, abstractmethod
from typing import Optional, Any
from .message import Message
from .llm import LLM
from .config import Config

class Agent(ABC):
    """Agent基类"""
    
    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None
    ):
        self.name = name
        self.llm = llm
        self.system_prompt = system_prompt
        self.config = config or Config()
        self._history: list[Message] = []
    
    @abstractmethod
    def run(self, input_text: str, **kwargs) -> str:
        """运行Agent"""
        pass
    
    def add_message(self, message: Message):
        """添加消息到历史记录"""
        self._history.append(message)
    
    def clear_history(self):
        """清空历史记录"""
        self._history.clear()
    
    def get_history(self) -> list[Message]:
        """获取历史记录"""
        return self._history.copy()
    
    def __str__(self) -> str:
        return f"Agent(name={self.name}, provider={self.llm.provider})"

tool

tool.py

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
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from .toolParameter import ToolParameter

class Tool(ABC):
    """工具基类"""

    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    @abstractmethod
    def run(self, parameters: Dict[str, Any]) -> str:
        """执行工具"""
        pass

    @abstractmethod
    def get_parameters(self) -> List[ToolParameter]:
        """获取工具参数定义"""
        pass

    def to_openai_schema(self) -> Dict[str, Any]:
        """转换为 OpenAI function calling schema 格式

        用于 FunctionCallAgent，使工具能够被 OpenAI 原生 function calling 使用

        Returns:
            符合 OpenAI function calling 标准的 schema
        """
        parameters = self.get_parameters()

        # 构建 properties
        properties = {}
        required = []

        for param in parameters:
            # 基础属性定义
            prop = {
                "type": param.type,
                "description": param.description
            }

            # 如果有默认值，添加到描述中（OpenAI schema 不支持 default 字段）
            if param.default is not None:
                prop["description"] = f"{param.description} (默认: {param.default})"

            # 如果是数组类型，添加 items 定义
            if param.type == "array":
                prop["items"] = {"type": "string"}  # 默认字符串数组

            properties[param.name] = prop

            # 收集必需参数
            if param.required:
                required.append(param.name)

        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }

toolParameter.py

1
2
3
4
5
6
7
8
9
10
from pydantic import BaseModel
from typing import Any

class ToolParameter(BaseModel):
    """工具参数定义"""
    name: str
    type: str
    description: str
    required: bool = True
    default: Any = None

toolRegistry.py

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
from typing import Callable, Any, Dict
from .tool import Tool

class ToolRegistry:
    """HelloAgents工具注册表"""

    def __init__(self):
        self._tools: dict[str, Tool] = {}
        self._functions: dict[str, dict[str, Any]] = {}

    def register_tool(self, tool: Tool):
        """注册Tool对象"""
        if tool.name in self._tools:
            print(f"⚠️ 警告:工具 '{tool.name}' 已存在，将被覆盖。")
        self._tools[tool.name] = tool
        print(f"✅ 工具 '{tool.name}' 已注册。")
        
    def register_function(self, name: str, description: str, func: Callable[[str], str]):
        """
        直接注册函数作为工具（简便方式）

        Args:
            name: 工具名称
            description: 工具描述
            func: 工具函数，接受字符串参数，返回字符串结果
        """
        if name in self._functions:
            print(f"⚠️ 警告:工具 '{name}' 已存在，将被覆盖。")

        self._functions[name] = {
            "description": description,
            "func": func
        }
        print(f"✅ 工具 '{name}' 已注册。")

    def execute_function(self, name: str, *args, **kwargs) -> Any:
        """执行注册的函数工具"""
        if name not in self._functions:
            raise ValueError(f"工具 '{name}' 未注册。")
        func = self._functions[name]["func"]
        return func(*args, **kwargs)
    
    def get_function(self, name: str) -> Callable[[str], str]:
        """获取注册的函数工具"""
        if name not in self._functions:
            raise ValueError(f"工具 '{name}' 未注册。")
        return self._functions[name]["func"]
    
    def execute_tool(self, name: str, *args, **kwargs) -> Any:
        """执行注册的Tool对象"""
        if name not in self._tools:
            raise ValueError(f"工具 '{name}' 未注册。")
        tool = self._tools[name]
        return tool.run(*args, **kwargs)
    
    def get_tool(self, name: str) -> Tool:
        """获取注册的Tool对象"""
        if name not in self._tools:
            raise ValueError(f"工具 '{name}' 未注册。")
        return self._tools[name]

    def is_function(self, name: str) -> bool:
        """判断一个工具名是否为函数类型"""
        return name in self._functions

    def is_tool(self, name: str) -> bool:
        """判断一个工具名是否为Tool对象类型"""
        return name in self._tools

    def get_tools_description(self) -> str:
        """获取所有可用工具的格式化描述字符串"""
        descriptions = []

        # Tool对象描述
        for tool in self._tools.values():
            descriptions.append(f"- {tool.name}: {tool.description}")

        # 函数工具描述
        for name, info in self._functions.items():
            descriptions.append(f"- {name}: {info['description']}")

        return "\n".join(descriptions) if descriptions else "暂无可用工具"

builtin/calculator.py

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
# my_calculator_tool.py
import ast
import operator
import math
def calculator(expression: str) -> str:
    """简单的数学计算函数"""
    if not expression.strip():
        return "计算表达式不能为空"

    # 支持的基本运算
    operators = {
        ast.Add: operator.add,      # +
        ast.Sub: operator.sub,      # -
        ast.Mult: operator.mul,     # *
        ast.Div: operator.truediv,  # /
    }

    # 支持的基本函数
    functions = {
        'sqrt': math.sqrt,
        'pi': math.pi,
    }

    try:
        node = ast.parse(expression, mode='eval')
        result = _eval_node(node.body, operators, functions)
        return str(result)
    except:
        return "计算失败，请检查表达式格式"

def _eval_node(node, operators, functions):
    """简化的表达式求值"""
    if isinstance(node, ast.Constant):
        return node.value
    elif isinstance(node, ast.BinOp):
        left = _eval_node(node.left, operators, functions)
        right = _eval_node(node.right, operators, functions)
        op = operators.get(type(node.op))
        return op(left, right)
    elif isinstance(node, ast.Call):
        func_name = node.func.id
        if func_name in functions:
            args = [_eval_node(arg, operators, functions) for arg in node.args]
            return functions[func_name](*args)
    elif isinstance(node, ast.Name):
        if node.id in functions:
            return functions[node.id]

builtin/search.py

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
from typing import Any, Callable, Dict, Optional
from serpapi import SerpApiClient, GoogleSearch
from tavily import TavilyClient
import os
from ..tool import Tool
from dotenv import load_dotenv

load_dotenv()  # Load environment variables from .env file


class SearchTool(Tool):
    """
    智能混合搜索工具，使用 SerpApi 和 Tavily API 进行网页搜索。
    """

    def __init__(self):
        super().__init__(
            name="search",
            description="一个智能网页搜索引擎。支持混合搜索模式，自动选择最佳搜索源。"
        )
        self.tavily_key = os.getenv("TAVILY_API_KEY")
        self.serpapi_key = os.getenv("SERPAPI_API_KEY")
        self.available_backends = []
        self._setup_backends()
        print(f"已选择搜索引擎: {self.available_backends}")


    def _setup_backends(self):
        """
        设置可用的搜索后端。
        """
        if self.tavily_key:
            self.available_backends.append("tavily")
        elif self.serpapi_key:
            self.available_backends.append("serpapi")
        if not self.available_backends:
            raise ValueError("No available search backends. Please set TAVILY_API_KEY or SERPAPI_API_KEY.")
        
    def run(self, query: str) -> Dict[str, Any]:
        """
        执行搜索操作，根据可用的后端选择最佳搜索源。
        """
        if "tavily" in self.available_backends:
            return self._search_tavily(query)
        elif "serpapi" in self.available_backends:
            return self._search_serpapi(query)
        else:
            raise ValueError("No available search backends to perform the search.")
        
    def _search_tavily(self, query: str) -> Dict[str, Any]:
        """使用Tavily搜索"""
        self.tavily_client = TavilyClient(api_key=self.tavily_key)

        response = self.tavily_client.search(query=query, max_results=3)

        if response.get('answer'):
            result = f"💡 AI直接答案:{response['answer']}\n\n"
        else:
            result = ""

        result += "🔗 相关结果:\n"
        for i, item in enumerate(response.get('results', [])[:3], 1):
            result += f"[{i}] {item.get('title', '')}\n"
            result += f"    {item.get('content', '')[:150]}...\n\n"

        return result

    def _search_serpapi(self, query: str) -> Dict[str, Any]:
        """
        使用 SerpApi 执行搜索。
        """
        search = GoogleSearch({
            "q": query,
            "api_key": os.getenv("SERPAPI_API_KEY"),
            "num": 3
        })

        results = search.get_dict()

        result = "🔗 Google搜索结果:\n"
        if "organic_results" in results:
            for i, res in enumerate(results["organic_results"][:3], 1):
                result += f"[{i}] {res.get('title', '')}\n"
                result += f"    {res.get('snippet', '')}\n\n"

        return result
    

    def get_parameters(self) -> Dict[str, Any]:
        """
        获取工具参数定义。
        """
        return {
            "query": {
                "type": "string",
                "description": "搜索查询字符串",
                "required": True
            }
        }

builtin/getArrraction.py

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
# 这个工具查询当地地区的景点
import os
from tavily import TavilyClient

def get_attraction(city: str, weather: str) -> str:
    """
    根据城市和天气，使用Tavily Search API搜索并返回优化后的景点推荐。
    """
    # 1. 从环境变量中读取API密钥
    api_key = os.environ.get('TAVILY_API_KEY')
    if not api_key:
        return "错误:未配置TAVILY_API_KEY环境变量。"

    # 2. 初始化Tavily客户端
    tavily = TavilyClient(api_key=api_key)
    
    # 3. 构造一个精确的查询
    query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"
    
    try:
        # 4. 调用API，include_answer=True会返回一个综合性的回答
        response = tavily.search(query=query, search_depth="basic", include_answer=True)
        
        # 5. Tavily返回的结果已经非常干净，可以直接使用
        # response['answer'] 是一个基于所有搜索结果的总结性回答
        if response.get("answer"):
            return response["answer"]
        
        # 如果没有综合性回答，则格式化原始结果
        formatted_results = []
        for result in response.get("results", []):
            formatted_results.append(f"- {result['title']}: {result['content']}")
        
        if not formatted_results:
             return "抱歉，没有找到相关的旅游景点推荐。"

        return "根据搜索，为您找到以下信息:\n" + "\n".join(formatted_results)

    except Exception as e:
        return f"错误:执行Tavily搜索时出现问题 - {e}"

builtin/getWeather.py

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
# 这里定义 get_weather(city: str) 工具，查询天气
# 这里有公共的 http api: wttr.in（ http://wttr.in/重庆）
import requests

def get_weather(city: str) -> str:

    # format参数代表让服务器响应的格式，j1就是JSON
    url = f"https://wttr.in/{city}?format=j1"

    try:
        # 使用 requests 包发起网络请求
        response = requests.get(url)
        # 检查是否成功，即状态码为200，若非200，则异常退出
        response.raise_for_status()
        # 解析返回 json 数据
        data = response.json()

        # 提取当前天气状况
        current_condition = data['current_condition'][0]
        weather_desc = current_condition['weatherDesc'][0]['value']
        temp_c = current_condition['temp_C']
        
        # 格式化成自然语言返回
        return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
    
    except requests.exceptions.RequestException as e:
        # 处理网络错误
        return f"错误:查询天气时遇到网络问题 - {e}"
    except (KeyError, IndexError) as e:
        # 处理数据解析错误
        return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"

toolChain.py

1

asyncToolExecutor.py

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
import asyncio
import concurrent.futures
from typing import Dict, Any, List, Callable
from tool.toolRegistry import ToolRegistry

class AsyncToolExecutor:
    """异步工具执行器"""

    def __init__(self, registry: ToolRegistry, max_workers: int = 4):
        self.registry = registry
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)

    async def execute_tool_async(self, tool_name: str, input_data: str) -> str:
        """异步执行单个工具，能够判断工具是函数还是Tool对象，并调用相应的执行方法"""
        loop = asyncio.get_event_loop()

        if self.registry.is_function(tool_name):
            def _execute():
                return self.registry.execute_function(tool_name, input_data)
        elif self.registry.is_tool(tool_name):
            def _execute():
                return self.registry.execute_tool(tool_name, input_data)
        else:
            raise ValueError(f"工具 '{tool_name}' 未在注册表中找到（既非函数也非Tool对象）")

        result = await loop.run_in_executor(self.executor, _execute)
        return result


    async def execute_tools_parallel(self, tasks: List[Dict[str, str]]) -> List[str]:
        """并行执行多个工具"""
        print(f"🚀 开始并行执行 {len(tasks)} 个工具任务")

        # 创建异步任务
        async_tasks = []
        for task in tasks:
            tool_name = task["tool_name"]
            input_data = task["input_data"]
            async_task = self.execute_tool_async(tool_name, input_data)
            async_tasks.append(async_task)

        # 等待所有任务完成
        results = await asyncio.gather(*async_tasks)

        print(f"✅ 所有工具任务执行完成")
        return results

    def __del__(self):
        """清理资源"""
        if hasattr(self, 'executor'):
            self.executor.shutdown(wait=True)

agent

simpleAgent.py

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
270
271
from core.llm import LLM
from core.agent import Agent
from core.config import Config
from core.message import Message
from tool.toolRegistry import ToolRegistry
from typing import Optional, Iterator
from typing import Callable
import re


class SimpleAgent(Agent):
    """一个简单的Agent实现，使用LLM进行对话。"""

    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None,
        tool_registry: Optional['ToolRegistry'] = None,
        enable_tool_calling: bool = True
    ):
        super().__init__(name, llm, system_prompt, config)
        self.tool_registry = tool_registry
        self.enable_tool_calling = enable_tool_calling and tool_registry is not None
        print(f"✅ {name} 初始化完成，工具调用: {'启用' if self.enable_tool_calling else '禁用'}")

    def run(self, input_text: str, max_tool_iterations: int = 3, **kwargs) -> str:
        """
        重写的运行方法 - 实现简单对话逻辑，支持可选工具调用
        """
        print(f"🤖 {self.name} 正在处理: {input_text}")

        # 构建消息列表
        messages = []

        # 添加系统消息（可能包含工具信息）
        enhanced_system_prompt = self._get_enhanced_system_prompt()
        messages.append({"role": "system", "content": enhanced_system_prompt})

        # 添加历史消息
        for msg in self._history:
            messages.append({"role": msg.role, "content": msg.content})

        # 添加当前用户消息
        messages.append({"role": "user", "content": input_text})

        # 如果没有启用工具调用，使用简单对话逻辑
        if not self.enable_tool_calling:
            response = self.llm.think(messages, **kwargs)
            self.add_message(Message(input_text, "user"))
            self.add_message(Message(response, "assistant"))
            print(f"✅ {self.name} 响应完成")
            return response

        # 支持多轮工具调用的逻辑
        return self._run_with_tools(messages, input_text, max_tool_iterations, **kwargs)

    def _get_enhanced_system_prompt(self) -> str:
        """构建增强的系统提示词，包含工具信息"""
        base_prompt = self.system_prompt or "你是一个有用的AI助手。"

        if not self.enable_tool_calling or not self.tool_registry:
            return base_prompt

        # 获取工具描述
        tools_description = self.tool_registry.get_tools_description()
        if not tools_description or tools_description == "暂无可用工具":
            return base_prompt

        tools_section = "\n\n## 可用工具\n"
        tools_section += "你可以使用以下工具来帮助回答问题:\n"
        tools_section += tools_description + "\n"

        tools_section += "\n## 工具调用格式\n"
        tools_section += "当需要使用工具时，请使用以下格式:\n"
        tools_section += "`[TOOL_CALL:{tool_name}:{parameters}]`\n"
        tools_section += "例如:`[TOOL_CALL:search:Python编程]` 或 `[TOOL_CALL:memory:recall=用户信息]`\n\n"
        tools_section += "工具调用结果会自动插入到对话中，然后你可以基于结果继续回答。\n"

        return base_prompt + tools_section

    def _run_with_tools(self, messages: list, input_text: str, max_tool_iterations: int, **kwargs) -> str:
        """支持工具调用的运行逻辑"""
        current_iteration = 0
        final_response = ""

        while current_iteration < max_tool_iterations:
            # 调用LLM
            response = self.llm.think(messages, **kwargs)

            # 检查是否有工具调用
            tool_calls = self._parse_tool_calls(response)

            if tool_calls:
                print(f"🔧 检测到 {len(tool_calls)} 个工具调用")
                # 执行所有工具调用并收集结果
                tool_results = []
                clean_response = response

                for call in tool_calls:
                    result = self._execute_tool_call(call['tool_name'], call['parameters'])
                    tool_results.append(result)
                    # 从响应中移除工具调用标记
                    clean_response = clean_response.replace(call['original'], "")

                # 构建包含工具结果的消息
                messages.append({"role": "assistant", "content": clean_response})

                # 添加工具结果
                tool_results_text = "\n\n".join(tool_results)
                messages.append({"role": "user", "content": f"工具执行结果:\n{tool_results_text}\n\n请基于这些结果给出完整的回答。"})

                current_iteration += 1
                continue

            # 没有工具调用，这是最终回答
            final_response = response
            break

        # 如果超过最大迭代次数，获取最后一次回答
        if current_iteration >= max_tool_iterations and not final_response:
            final_response = self.llm.think(messages, **kwargs)

        # 保存到历史记录
        self.add_message(Message(input_text, "user"))
        self.add_message(Message(final_response, "assistant"))
        print(f"✅ {self.name} 响应完成")

        return final_response

    def _parse_tool_calls(self, text: str) -> list:
        """解析文本中的工具调用"""
        pattern = r'\[TOOL_CALL:([^:]+):([^\]]+)\]'
        matches = re.findall(pattern, text)

        tool_calls = []
        for tool_name, parameters in matches:
            tool_calls.append({
                'tool_name': tool_name.strip(),
                'parameters': parameters.strip(),
                'original': f'[TOOL_CALL:{tool_name}:{parameters}]'
            })

        return tool_calls

    def _execute_tool_call(self, tool_name: str, parameters: str) -> str:
        """执行工具调用"""
        if not self.tool_registry:
            return f"❌ 错误:未配置工具注册表"

        try:
            # 智能参数解析
            if tool_name == 'calculator':
                # 计算器工具直接传入表达式
                result = self.tool_registry.execute_tool(tool_name, parameters)
            else:
                # 其他工具使用智能参数解析
                param_dict = self._parse_tool_parameters(tool_name, parameters)
                tool = self.tool_registry.get_tool(tool_name)
                if not tool:
                    return f"❌ 错误:未找到工具 '{tool_name}'"
                result = tool.run(param_dict)

            return f"🔧 工具 {tool_name} 执行结果:\n{result}"

        except Exception as e:
            return f"❌ 工具调用失败:{str(e)}"

    def _parse_tool_parameters(self, tool_name: str, parameters: str) -> dict:
        """智能解析工具参数"""
        param_dict = {}

        if '=' in parameters:
            # 格式: key=value 或 action=search,query=Python
            if ',' in parameters:
                # 多个参数:action=search,query=Python,limit=3
                pairs = parameters.split(',')
                for pair in pairs:
                    if '=' in pair:
                        key, value = pair.split('=', 1)
                        param_dict[key.strip()] = value.strip()
            else:
                # 单个参数:key=value
                key, value = parameters.split('=', 1)
                param_dict[key.strip()] = value.strip()
        else:
            # 直接传入参数，根据工具类型智能推断
            if tool_name == 'search':
                param_dict = {'query': parameters}
            elif tool_name == 'memory':
                param_dict = {'action': 'search', 'query': parameters}
            else:
                param_dict = {'input': parameters}

        return param_dict


    def stream_run(self, input_text: str, **kwargs) -> Iterator[str]:
        """
        自定义的流式运行方法
        """
        print(f"🌊 {self.name} 开始流式处理: {input_text}")

        messages = []

        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})

        for msg in self._history:
            messages.append({"role": msg.role, "content": msg.content})

        messages.append({"role": "user", "content": input_text})

        # 流式调用LLM
        full_response = ""
        print("📝 实时响应: ", end="")
        for chunk in self.llm.think_stream(messages, **kwargs):
            full_response += chunk
            print(chunk, end="", flush=True)
            yield chunk

        print()  # 换行

        # 保存完整对话到历史记录
        self.add_message(Message(input_text, "user"))
        self.add_message(Message(full_response, "assistant"))
        print(f"✅ {self.name} 流式响应完成")

    def add_tool(self, tool) -> None:
        """添加工具到Agent（便利方法）"""
        if not self.tool_registry:
            self.tool_registry = ToolRegistry()
            self.enable_tool_calling = True

        self.tool_registry.register_tool(tool)
        print(f"🔧 工具 '{tool.name}' 已添加")

    def add_tool(self, func: Callable[[str], str]) -> None:
        """添加函数作为工具到Agent（便利方法）"""
        if not self.tool_registry:
            self.tool_registry = ToolRegistry()
            self.enable_tool_calling = True

        # 使用函数名称作为工具名称
        tool_name = func.__name__
        description = func.__doc__ or "无描述"

        self.tool_registry.register_function(
            name=tool_name,
            description=description,
            func=func
        )
        print(f"🔧 函数工具 '{tool_name}' 已添加")

    def has_tools(self) -> bool:
        """检查是否有可用工具"""
        return self.enable_tool_calling and self.tool_registry is not None
    
    def remove_tool(self, tool_name: str) -> bool:
        """移除工具（便利方法）"""
        if self.tool_registry:
            self.tool_registry.unregister(tool_name)
            return True
        return False
    
    def list_tools(self) -> list:
        """列出所有可用工具"""
        if self.tool_registry:
            return self.tool_registry.get_tools_description()
        return []

reActAgent.py

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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
"""
ReAct (Reasoning + Acting) Agent 实现
基于 Thought → Action → Observation 循环的推理-行动模式

继承自 Agent 抽象类（非 SimpleAgent），独立实现完整的 ReAct 循环。
"""
import re
from typing import Dict, Any, Optional, List
from core.agent import Agent
from core.llm import LLM
from core.config import Config
from core.message import Message
from tool.toolRegistry import ToolRegistry
from agent.prompts.reActPrompt import REACT_PROMPT


class ReActAgent(Agent):
    """ReAct 模式 Agent。

    通过 Thought → Action → Observation 循环进行推理和工具调用。
    run() 始终返回 str；is_stream=True 时内部走流式路径，
    边输出 chunk 边收集，对外签名不变。
    """

    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None,
        tool_registry: Optional[ToolRegistry] = None,
        max_iterations: int = 5,
        is_stream: bool = False,
        verbose: bool = True,
    ):
        super().__init__(name, llm, system_prompt, config)
        self.tool_registry = tool_registry
        self.max_iterations = max_iterations
        self.is_stream = is_stream
        self.verbose = verbose
        self._execution_trace: List[str] = []   # 本轮执行轨迹

        # 统计工具数量
        tool_count = 0
        if tool_registry:
            desc = tool_registry.get_tools_description()
            if desc and desc != "暂无可用工具":
                tool_count = len([l for l in desc.split("\n") if l.strip().startswith("-")])

        print(
            f"✅ {name} (ReAct) 初始化完成 | "
            f"流式: {'开' if is_stream else '关'} | "
            f"最大迭代: {max_iterations} | "
            f"工具: {tool_count}个"
        )

    # ================================================================
    # 公共入口
    # ================================================================

    def run(self, input_text: str, **kwargs) -> str:
        """运行 ReAct Agent。

        is_stream=False → 批量模式，完整接收每轮响应后解析
        is_stream=True  → 流式模式，实时打印 chunk，行为对调用方透明
        """
        self._execution_trace = []

        if self.is_stream:
            return self._run_stream(input_text, **kwargs)
        else:
            return self._run_batch(input_text, **kwargs)

    # ================================================================
    # 批量模式
    # ================================================================

    def _run_batch(self, input_text: str, **kwargs) -> str:
        """批量模式：每轮用 llm.think() 获取完整响应。"""
        messages = self._build_initial_messages(input_text)

        for i in range(self.max_iterations):
            if self.verbose:
                print(f"\n--- ReAct 第 {i + 1}/{self.max_iterations} 轮 ---")

            response = self.llm.think(messages, **kwargs)
            if response is None:
                return "❌ LLM 调用失败，请检查 API 配置。"

            result = self._process_response(response, messages)
            if result is not None:
                self._save_to_history(input_text, result)
                return result

        return self._force_final_answer(messages, input_text, **kwargs)

    # ================================================================
    # 流式模式
    # ================================================================

    def _run_stream(self, input_text: str, **kwargs) -> str:
        """流式模式：每轮用 llm.think_stream() 实时输出 chunk。

        等本轮全部收完后再解析（不做增量解析），保证解析正确性。
        """
        messages = self._build_initial_messages(input_text)

        for i in range(self.max_iterations):
            if self.verbose:
                print(f"\n--- ReAct 第 {i + 1}/{self.max_iterations} 轮 ---")

            full_response = ""
            for chunk in self.llm.think_stream(messages, **kwargs):
                if chunk:
                    print(chunk, end="", flush=True)
                    full_response += chunk
            print()  # 本轮流式结束，换行

            if not full_response:
                return "❌ LLM 流式调用失败，请检查 API 配置。"

            result = self._process_response(full_response, messages)
            if result is not None:
                self._save_to_history(input_text, result)
                return result

        return self._force_final_answer(messages, input_text, **kwargs)

    # ================================================================
    # 消息构建
    # ================================================================

    def _build_initial_messages(self, input_text: str) -> List[Dict[str, str]]:
        """构建初始消息列表：system（含 ReAct 格式说明 + 工具列表）
        + 历史消息 + 当前用户消息。"""
        messages: List[Dict[str, str]] = []

        # 1. 系统消息
        tools_desc = (
            self.tool_registry.get_tools_description()
            if self.tool_registry
            else "暂无可用工具"
        )
        system_content = REACT_PROMPT.format(
            tools=tools_desc,
            question=input_text,
            history="（尚未执行任何操作）",
        )
        messages.append({"role": "system", "content": system_content})

        # 2. 历史消息（来自父类 self._history）
        for msg in self._history:
            messages.append(msg.to_dict())

        # 3. 当前用户消息
        messages.append({"role": "user", "content": input_text})

        return messages

    # ================================================================
    # 响应处理（批量 / 流式共用）
    # ================================================================

    def _process_response(
        self, response: str, messages: List[Dict]
    ) -> Optional[str]:
        """处理 LLM 的一轮完整响应。

        Returns:
            None  → 已执行工具并将 Observation 追加到 messages，继续循环
            str   → 这是最终答案，调用方应立刻返回
        """
        parsed = self._parse_react_output(response)

        # 打印 Thought
        if self.verbose and parsed["thought"]:
            t = parsed["thought"]
            print(f"🧠 Thought: {t[:120]}{'...' if len(t) > 120 else ''}")

        # —— 情况 1：检测到 Finish ——
        if parsed["finish"]:
            if self.verbose:
                f = parsed["finish"]
                print(f"🎯 最终答案: {f[:100]}{'...' if len(f) > 100 else ''}")
            return parsed["finish"]

        # —— 情况 2：检测到 Action ——
        if parsed["action"]:
            observation = self._execute_action(
                parsed["action"], parsed["action_input"] or ""
            )

            if self.verbose:
                print(f"🔧 Action: {parsed['action']}[{parsed['action_input']}]")
                o = observation
                print(f"👁️  Observation: {o[:150]}{'...' if len(o) > 150 else ''}")

            # 记录本轮执行轨迹
            trace = (
                f"Thought: {parsed['thought']}\n"
                f"Action: {parsed['action']}[{parsed['action_input']}]\n"
                f"Observation: {observation}"
            )
            self._execution_trace.append(trace)

            # 将本轮输出追加到对话上下文
            messages.append({"role": "assistant", "content": response})
            messages.append({"role": "user", "content": f"Observation: {observation}"})

            return None

        # —— 情况 3：格式异常，提示 LLM 重新输出 ——
        if self.verbose:
            print("⚠️  响应格式异常（无 Action / Finish），提示 LLM 重试")
        messages.append({"role": "assistant", "content": response})
        messages.append({
            "role": "user",
            "content": (
                "请严格遵循 ReAct 格式。每次回复必须包含 Thought 和以下两者之一：\n"
                "  - Action: 工具名[参数]  （调用工具）\n"
                "  - Finish[你的最终答案]   （结束任务）"
            ),
        })
        return None

    # ================================================================
    # 解析器
    # ================================================================

    def _parse_react_output(self, text: str) -> Dict[str, Optional[str]]:
        """解析 ReAct 格式的 LLM 输出。

        支持的格式:
          Thought: 推理内容...
          Action: tool_name[input]
          Finish[最终答案]

        Returns:
          {"thought": ..., "action": ..., "action_input": ..., "finish": ...}
          finish 与 action 互斥；未匹配到的字段为 None。
        """
        result: Dict[str, Optional[str]] = {
            "thought": None,
            "action": None,
            "action_input": None,
            "finish": None,
        }

        # Thought —— 从 "Thought:" 到下一个 "Action" / "Finish" 或文本末尾
        m = re.search(
            r"Thought:\s*(.+?)(?=\n\s*(?:Action|Finish)|\Z)",
            text, re.DOTALL | re.IGNORECASE,
        )
        if m:
            result["thought"] = m.group(1).strip()

        # Finish —— 优先检测，一旦命中就不再解析 Action
        m = re.search(r"Finish\s*\[(.+)\]", text, re.DOTALL | re.IGNORECASE)
        if m:
            result["finish"] = m.group(1).strip()
            return result

        # Action
        m = re.search(r"Action:\s*(\w+)\s*\[([^\]]+)\]", text, re.IGNORECASE)
        if m:
            result["action"] = m.group(1).strip()
            result["action_input"] = m.group(2).strip()

        return result

    # ================================================================
    # 工具执行
    # ================================================================

    def _execute_action(self, tool_name: str, tool_input: str) -> str:
        """执行工具，自动判断工具类型（函数 / Tool 对象）。"""
        if not self.tool_registry:
            return "❌ 错误: 未配置工具注册表"

        try:
            if self.tool_registry.is_function(tool_name):
                result = self.tool_registry.execute_function(tool_name, tool_input)
            elif self.tool_registry.is_tool(tool_name):
                result = self.tool_registry.execute_tool(
                    tool_name, {"input": tool_input}
                )
            else:
                return f"❌ 错误: 工具 '{tool_name}' 未在注册表中找到"

            return str(result) if result is not None else "（工具执行完成，无返回内容）"

        except Exception as e:
            return f"❌ 工具调用失败: {str(e)}"

    # ================================================================
    # 兜底 & 辅助
    # ================================================================

    def _force_final_answer(
        self, messages: List[Dict], input_text: str, **kwargs
    ) -> str:
        """达到最大迭代次数后，强制 LLM 输出最终答案。"""
        if self.verbose:
            print(f"⚠️  达到最大迭代次数 ({self.max_iterations})，强制要求最终答案")

        messages.append({
            "role": "user",
            "content": (
                "你已达到最大执行轮数。"
                "请立即用 Finish[你的答案] 格式给出最终回答，不要再调用工具。"
            ),
        })

        if self.is_stream:
            full = ""
            for chunk in self.llm.think_stream(messages, **kwargs):
                if chunk:
                    print(chunk, end="", flush=True)
                    full += chunk
            print()
        else:
            full = self.llm.think(messages, **kwargs)

        if full:
            parsed = self._parse_react_output(full)
            if parsed["finish"]:
                self._save_to_history(input_text, parsed["finish"])
                return parsed["finish"]

        # 最后兜底
        result = full or "❌ 无法获取有效回答"
        self._save_to_history(input_text, result)
        return result

    def _save_to_history(self, input_text: str, response: str):
        """保存对话到 Agent 历史。"""
        self.add_message(Message(input_text, "user"))
        self.add_message(Message(response, "assistant"))

    # ================================================================
    # 便利方法
    # ================================================================

    def get_execution_trace(self) -> List[str]:
        """返回本轮完整的执行轨迹（每轮一条 Thought/Action/Observation）。"""
        return self._execution_trace.copy()

reflectionAgent.py

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
"""
Reflection Agent 实现
基于 生成 → 反思 → 优化 循环的自我改进模式

继承自 Agent 抽象类，无工具调用，通过内部反思循环优化答案质量。
"""
import re
from typing import Dict, Optional, List
from core.agent import Agent
from core.llm import LLM
from core.config import Config
from core.message import Message
from agent.prompts.reflectionPrompt import DEFAULT_PROMPTS


class ReflectionAgent(Agent):
    """Reflection 模式 Agent。

    通过 生成 → 反思 → 优化 循环进行自我改进。
    run() 始终返回 str；不支持流式输出。
    """

    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None,
        max_reflections: int = 2,
        auto_finish: bool = False,
        prompts: Optional[Dict[str, str]] = None,
        verbose: bool = True,
    ):
        super().__init__(name, llm, system_prompt, config)
        self.max_reflections = max_reflections
        self.auto_finish = auto_finish
        self.verbose = verbose

        # 合并提示词：用户传入的覆盖默认值
        self.prompts = {**DEFAULT_PROMPTS, **(prompts or {})}

        print(
            f"✅ {name} (Reflection) 初始化完成 | "
            f"最大反思轮数: {max_reflections} | "
            f"智能终止: {'开' if auto_finish else '关'}"
        )

    # ================================================================
    # 公共入口
    # ================================================================

    def run(self, input_text: str, **kwargs) -> str:
        """运行 Reflection Agent。

        流程: 初始生成 → 循环(反思 → 优化) → 返回最终答案
        """
        # 阶段 1: 初始生成
        initial_answer = self._generate_initial(input_text, **kwargs)
        if initial_answer is None:
            return "❌ LLM 调用失败，请检查 API 配置。"

        if self.verbose:
            print(f"\n📝 初始答案: {initial_answer[:120]}{'...' if len(initial_answer) > 120 else ''}")

        # 保存用户输入 + 初始答案到 history
        self._save_to_history(input_text, initial_answer)

        current_answer = initial_answer

        # 阶段 2: 反思-优化循环
        for i in range(self.max_reflections):
            if self.verbose:
                print(f"\n--- Reflection 第 {i + 1}/{self.max_reflections} 轮 ---")

            # 反思
            feedback = self._reflect(input_text, current_answer, **kwargs)
            if feedback is None:
                if self.verbose:
                    print("⚠️ 反思阶段 LLM 调用失败，返回当前答案")
                return current_answer

            if self.verbose:
                print(f"🔍 反思反馈: {feedback[:120]}{'...' if len(feedback) > 120 else ''}")

            # 智能终止检测
            if self._check_no_improvement(feedback):
                if self.auto_finish:
                    if self.verbose:
                        print("✅ 模型判定无需改进，提前终止")
                    self.add_message(Message(feedback, "assistant"))
                    return current_answer
                # auto_finish=False: 忽略标记，继续执行 refine

            # 优化
            refined_answer = self._refine(input_text, current_answer, feedback, **kwargs)
            if refined_answer is None:
                if self.verbose:
                    print("⚠️ 优化阶段 LLM 调用失败，返回当前答案")
                return current_answer

            if self.verbose:
                print(f"✨ 优化结果: {refined_answer[:120]}{'...' if len(refined_answer) > 120 else ''}")

            # 保存本轮反思产物到 history
            self.add_message(Message(feedback, "assistant"))
            self.add_message(Message(refined_answer, "assistant"))

            current_answer = refined_answer

        return current_answer

    # ================================================================
    # 三个阶段
    # ================================================================

    def _generate_initial(self, task: str, **kwargs) -> Optional[str]:
        """第一阶段：生成初始答案。"""
        messages = [
            {"role": "system", "content": self.prompts["initial"].format(task=task)},
            {"role": "user", "content": task},
        ]
        return self.llm.think(messages, **kwargs)

    def _reflect(self, task: str, content: str, **kwargs) -> Optional[str]:
        """第二阶段：反思当前答案，给出改进反馈。"""
        prompt = self.prompts["reflect"].format(task=task, content=content)
        messages = [
            {"role": "system", "content": prompt},
            {"role": "user", "content": "请审查以上回答并给出反馈意见。"},
        ]
        return self.llm.think(messages, **kwargs)

    def _refine(self, task: str, last_attempt: str, feedback: str, **kwargs) -> Optional[str]:
        """第三阶段：根据反馈优化答案。"""
        messages = [
            {"role": "system", "content": self.prompts["refine"].format(
                task=task, last_attempt=last_attempt, feedback=feedback
            )},
            {"role": "user", "content": "请根据反馈意见提供改进后的回答。"},
        ]
        return self.llm.think(messages, **kwargs)

    # ================================================================
    # 辅助方法
    # ================================================================

    def _check_no_improvement(self, feedback: str) -> bool:
        """检测反馈中是否包含无需改进的标记。"""
        return bool(re.search(r'\[NO_IMPROVEMENT_NEEDED\]', feedback))

    def _save_to_history(self, input_text: str, response: str):
        """保存用户输入和回答到 Agent 历史。"""
        self.add_message(Message(input_text, "user"))
        self.add_message(Message(response, "assistant"))

planAndSolveAgent.py

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
"""
Plan-and-Solve Agent 实现
基于 拆分子任务 → 顺序执行子任务 的两阶段模式

继承自 Agent 抽象类，无工具调用，规划器分解问题，执行器逐步求解。
"""
import ast
import re
from typing import Optional, List, Dict
from core.agent import Agent
from core.llm import LLM
from core.config import Config
from core.message import Message
from agent.prompts.planAndSolvePrompt import DEFAULT_PLANNER_PROMPT, DEFAULT_EXECUTOR_PROMPT


class PlanAndSolveAgent(Agent):
    """Plan-and-Solve 模式 Agent。

    阶段 1（Plan）：将复杂问题分解为有序的子任务列表。
    阶段 2（Solve）：按顺序执行每个子任务，累积历史上下文。
    run() 始终返回 str；不支持流式输出。
    """

    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None,
        verbose: bool = True,
    ):
        super().__init__(name, llm, system_prompt, config)
        self.verbose = verbose
        self._step_results: List[Dict[str, str]] = []

        print(f"✅ {name} (PlanAndSolve) 初始化完成")

    # ================================================================
    # 公共入口
    # ================================================================

    def run(self, input_text: str, **kwargs) -> str:
        """运行 Plan-and-Solve Agent。

        流程: 规划（分解子任务） → 执行（逐步求解） → 返回最终答案
        """
        self._step_results = []

        # 阶段 1: 规划
        plan = self._plan(input_text, **kwargs)
        if plan is None:
            return "❌ 规划阶段 LLM 调用失败，请检查 API 配置。"
        if not plan:
            return "❌ 规划阶段出错：未能生成有效的子任务计划。"

        if self.verbose:
            print(f"\n📋 计划 ({len(plan)} 步):")
            for j, step in enumerate(plan, 1):
                print(f"  {j}. {step}")

        # 保存用户输入和计划到 history
        self._save_to_history(input_text, self._format_plan(plan))

        # 阶段 2: 逐步执行
        history_text = ""

        for i, step in enumerate(plan):
            if self.verbose:
                print(f"\n--- 执行步骤 {i + 1}/{len(plan)} ---")

            result = self._execute_step(input_text, plan, history_text, step, **kwargs)
            if result is None:
                if self.verbose:
                    print(f"⚠️ 步骤 {i + 1} LLM 调用失败，返回已累积结果")
                break

            self._step_results.append({"step": step, "result": result})

            if self.verbose:
                print(f"✅ 步骤 {i + 1} 完成: {result[:120]}{'...' if len(result) > 120 else ''}")

            # 更新历史
            history_text += f"步骤 {i + 1}: {step}\n结果 {i + 1}: {result}\n\n"

            # 保存每步结果到 history
            self.add_message(Message(
                f"步骤 {i + 1}: {step}\n结果: {result}", "assistant"
            ))

        # 返回最后一步的结果
        if self._step_results:
            final = self._step_results[-1]["result"]
            if self.verbose:
                print(f"\n🏁 最终答案: {final[:120]}{'...' if len(final) > 120 else ''}")
            return final

        return "❌ 执行阶段未能产出任何结果。"

    # ================================================================
    # 阶段 1: 规划
    # ================================================================

    def _plan(self, question: str, **kwargs) -> Optional[List[str]]:
        """调用规划器 LLM，生成子任务列表。"""
        messages = [
            {"role": "system", "content": DEFAULT_PLANNER_PROMPT.format(question=question)},
            {"role": "user", "content": question},
        ]
        response = self.llm.think(messages, **kwargs)
        if response is None:
            return None
        return self._parse_plan(response)

    def _parse_plan(self, text: str) -> List[str]:
        """解析 LLM 输出的计划。

        优先使用 ast.literal_eval，失败后用正则提取引号内字符串兜底。
        """
        # 尝试定位 ```python ... ``` 代码块中的内容
        code_block = text
        m = re.search(r"```(?:python)?\s*(.+?)```", text, re.DOTALL)
        if m:
            code_block = m.group(1).strip()

        # 1) ast.literal_eval
        try:
            result = ast.literal_eval(code_block)
            if isinstance(result, list) and all(isinstance(s, str) for s in result):
                return result
        except (ValueError, SyntaxError):
            pass

        # 2) 正则兜底：匹配所有被双引号或单引号包裹的字符串
        steps = re.findall(r"""["']([^"']+)["']""", code_block)
        if steps:
            return steps

        return []

    # ================================================================
    # 阶段 2: 执行
    # ================================================================

    def _execute_step(
        self,
        question: str,
        plan: List[str],
        history: str,
        current_step: str,
        **kwargs,
    ) -> Optional[str]:
        """执行单个子任务，返回该步骤的结果。"""
        plan_text = "\n".join(f"{i + 1}. {s}" for i, s in enumerate(plan))
        history_text = history if history else "（尚无已完成的步骤）"

        messages = [
            {
                "role": "system",
                "content": DEFAULT_EXECUTOR_PROMPT.format(
                    question=question,
                    plan=plan_text,
                    history=history_text,
                    current_step=current_step,
                ),
            },
            {"role": "user", "content": f"请执行当前步骤: {current_step}"},
        ]
        return self.llm.think(messages, **kwargs)

    # ================================================================
    # 辅助方法
    # ================================================================

    def _format_plan(self, plan: List[str]) -> str:
        """格式化计划为可读字符串，用于存储到 history。"""
        lines = [f"{i + 1}. {s}" for i, s in enumerate(plan)]
        return "📋 执行计划:\n" + "\n".join(lines)

    def _save_to_history(self, input_text: str, response: str):
        """保存用户输入和回答到 Agent 历史。"""
        self.add_message(Message(input_text, "user"))
        self.add_message(Message(response, "assistant"))

    # ================================================================
    # 便利方法
    # ================================================================

    def get_step_results(self) -> List[Dict[str, str]]:
        """返回本轮每个步骤的名称和结果。"""
        return self._step_results.copy()