摘要: 本篇作为实战系列的开篇，将带你从零搭建智能体开发的底层基础设施（LLM Client 封装），并深入解析 ReAct (Reasoning and Acting) 范式的核心理论与数学形式化表达。我们将亲手实现第一个外部工具“Google Search”及其调度器 ToolExecutor，为构建真正的自主智能体打下坚实的地基。
关键词: Agent Architecture, ReAct, LLM Wrapper, Tool Use, SerpApi
本文是基于Datawhale的hello-agent开源项目做的一些笔记，内容仅供参考，原PDF以及代码可以去github仓库获取https://datawhalechina.github.io/hello-agents

4. 智能体经典范式构建 (Introduction)

在上一章中，我们深入探讨了作为现代智能体“大脑”的大语言模型（LLM），了解了其 Transformer 架构及能力边界。现在，是时候将理论转化为实践，构建真正的智能体了。

一个现代智能体，其核心在于将 LLM 的推理能力与外部世界联通。它能理解意图、拆解任务，并通过调用工具（代码解释器、搜索引擎、API）来达成目标。然而，智能体也面临着幻觉、推理循环、错误使用工具等挑战。

为了组织智能体的“思考”与“行动”，业界涌现了三种经典范式，本章将逐一实现：

• ReAct (Reasoning and Acting)：边想边做，动态调整。
• Plan-and-Solve：三思而后行，先计划后执行。
• Reflection：自我反思，通过自我批判优化结果。

💡 深度解析 (Deep Dive)：
你可能会问：“为什么已有 LangChain、LlamaIndex 还要重复造轮子？”

1. 去黑盒化：高度抽象的框架隐藏了底层机制，亲手实现有助于理解设计原理。
2. 工程祛魅：处理格式解析、重试机制、死循环保护等“脏活”，是培养系统设计能力的必经之路。
3. 定制能力：掌握原理后，当标准组件无法满足需求时，你才有能力进行深度定制。

---

4.1 环境准备与基础工具定义

在构建具体范式前，我们需要搭建开发环境并定义基础组件，以避免重复劳动 。

4.1.1 安装依赖库

实战主要使用 Python (建议 3.10+)。我们需要安装 openai 用于交互，以及 python-dotenv 管理密钥 [cite: 18, 22]。

pip install openai python-dotenv

4.1.2 配置 API 密钥

为了代码通用性，我们将模型配置统一在环境变量中 。

1. 在项目根目录创建 .env 文件。
2. 添加以下内容（可指向 OpenAI 或兼容接口的本地/第三方服务）：

# .env file
LLM_API_KEY="YOUR-API-KEY"
LLM_MODEL_ID="gpt-3.5-turbo" # 或其他模型
LLM_BASE_URL="[https://api.openai.com/v1](https://api.openai.com/v1)" # 或你的代理地址/本地地址

4.1.3 封装基础 LLM 调用函数

为了让主逻辑更清晰，我们定义一个专属的 LLM 客户端类 HelloAgentsLLM，封装交互细节 。

🛠️ 实战映射 (Implementation)：
在工程实践中，直接在业务逻辑里写 client.chat.completions.create 是反模式。通常我们会封装一个 ModelClient 或 LLMService，负责处理 Retry (重试)、Fallback (降级)、Logging (日志) 以及 Latency Monitoring (延时监控)。

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

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

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

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

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

    def think(self, messages: List[Dict[str, str]], temperature: float = 0.7) -> str:
        """
        调用大语言模型进行思考，并返回其响应
        """
        print(f"正在调用 {self.model} 模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True, # 强制开启流式输出
            )
            
            # 处理流式响应
            print("大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True) # 实时打印到控制台
                collected_content.append(content)
            
            print() # 在流式输出结束后换行
            return "".join(collected_content)

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

# --- 简单测试 ---
if __name__ == '__main__':
    try:
        llm_client = HelloAgentsLLM()
        example_messages = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "写一个 Python 的快速排序算法"}
        ]
        print("--- 调用 LLM ---")
        response_text = llm_client.think(example_messages)
        
        if response_text:
            print("\n\n--- 完整模型响应 ---")
            print(response_text)
            
    except ValueError as e:
        print(e)

---

4.2 ReAct: 协同思考与行动

在准备好 LLM 客户端后，我们将构建第一个也是最经典的范式：ReAct (Reason + Act)。它由 Shunyu Yao 于 2022 年提出，核心思想是模仿人类解决问题的方式：形成“思考-行动-观察”的循环 。

4.2.1 ReAct 的工作流程

在 ReAct 之前，主流方法分为两类 ：

1. 纯思考 (Pure Reasoning)：如 CoT (Chain-of-Thought)，擅长逻辑推理，但无法交互，易产生幻觉。
2. 纯行动 (Pure Acting)：模型直接输出动作，缺乏规划和纠错。

ReAct 通过 Prompt Engineering 引导模型遵循固定轨迹 ：

• Thought (思考)：内心独白。分析情况，分解任务。
• Action (行动)：具体的动作，如调用工具 Search['query']。
• Observation (观察)：工具执行后的返回结果。

数学形式化表达

在每个时间步 ，智能体策略 根据初始问题 和之前的轨迹生成当前的思考 和行动 ：

随后，环境中的工具 执行行动并返回观察 ：

这个循环持续进行，直到模型认为找到了最终答案。

流程可视化

---

4.2.2 工具的定义与实现

如果说 LLM 是大脑，工具 (Tools) 就是“手和脚” 。我们将实现一个具备联网搜索能力的 ReAct 智能体。

1. 安装与配置搜索工具

我们选用 SerpApi 来获取结构化的 Google 搜索结果 。

pip install google-search-results

你需要注册 SerpApi 获取密钥，并加入 .env 文件 ：

SERPAPI_API_KEY="YOUR_SERPAPI_API_KEY"

2. 实现搜索工具核心逻辑

一个良好定义的工具包含：Name (标识符), Description (用途描述，供 LLM 判断), Execution Logic (函数) 。

以下是封装好的 search 工具，它会智能解析 SerpApi 的返回结果，优先提取 Knowledge Graph 或 Answer Box ：

from serpapi import GoogleSearch
import os

def search(query: str) -> str:
    """
    一个基于 SerpApi 的实战网页搜索引擎工具。
    它会智能地解析搜索结果，优先返回直接答案或知识图谱信息。
    """
    print(f"正在执行 [SerpApi] 网页搜索: {query}")
    try:
        api_key = os.getenv("SERPAPI_API_KEY")
        if not api_key:
            return "错误: SERPAPI_API_KEY 未在 .env 文件中配置"

        params = {
            "engine": "google",
            "q": query,
            "api_key": api_key,
            "gl": "cn",    # 国家代码
            "hl": "zh-cn", # 语言代码
        }
        
        search = GoogleSearch(params)
        results = search.get_dict()
        
        # --- 智能解析逻辑 ---
        # 1. 优先寻找 Answer Box (如直接的定义、计算结果)
        if "answer_box" in results:
             box = results["answer_box"]
             if "answer" in box:
                 return box["answer"]
             if "snippet" in box:
                 return box["snippet"]
             if "result" in box: # 针对计算器等
                 return box["result"]

        # 2. 其次寻找 Knowledge Graph (知识图谱)
        if "knowledge_graph" in results:
             kg = results["knowledge_graph"]
             title = kg.get("title", "")
             desc = kg.get("description", "")
             return f"{title}: {desc}"

        # 3. 如果没有直接答案，返回前三个自然搜索结果的摘要
        if "organic_results" in results:
            snippets = []
            for i, res in enumerate(results["organic_results"][:3]):
                title = res.get("title", "")
                snippet = res.get("snippet", "")
                snippets.append(f"[{i+1}] {title}\n{snippet}")
            return "\n\n".join(snippets)

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

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

(注：原文代码使用了 SerpApiClient 和 client.get_dict()，这里根据常见 google-search-results 库用法微调为 GoogleSearch 类，逻辑保持一致)

💡 深度解析 (Deep Dive)：
原文中的“智能解析”逻辑非常关键。直接把巨大的 JSON 丢给 LLM 会消耗大量 Token 且引入噪声。Preprocessing (预处理) 工具的返回值，提取最关键字段（Summary, Snippet），是提升 Agent 性能的常用工程技巧。

3. 构建通用的工具执行器 (ToolExecutor)

我们需要一个管理器来注册和调度这些工具 。

from typing import Dict, Any, Callable

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

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

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

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

4. 测试工具链

让我们将 Search 工具注册进去，模拟一次调用 。

if __name__ == '__main__':
    # 1. 初始化执行器
    tool_executor = ToolExecutor()
    
    # 2. 注册搜索工具
    # 注意：Description 必须写得足够清晰，引导 LLM 何时使用它
    search_desc = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。"
    tool_executor.register_tool("Search", search_desc, search)
    
    # 3. 打印可用工具
    print("\n--- 可用的工具 ---")
    print(tool_executor.get_available_tools())
    
    # 4. 模拟 Agent 决定调用的动作
    print("\n--- 执行 Action: Search['英伟达最新的 GPU 型号是什么'] ---")
    tool_name = "Search"
    tool_input = "英伟达最新的 GPU 型号是什么"
    
    tool_function = tool_executor.get_tool(tool_name)
    if tool_function:
        observation = tool_function(tool_input)
        print("\n--- 观察 (Observation) ---")
        print(observation)
    else:
        print(f"错误: 未找到名为 '{tool_name}' 的工具。")

运行结果示例:
[1] GeForce RTX 50 系列显卡...
[2] 比较 GeForce 系列最新一代显卡...

至此，我们已经为智能体配备了连接真实互联网的 Search 工具，并有了 HelloAgentsLLM 作为大脑接口。