如果你问我2026年学习AI Agent的最佳方式是什么，我会说：绝对是从零自己动手构建。这不仅对学习至关重要，如果你要打造一个高效、个性化且健壮的生产级AI Agent，从零开始往往是最佳选择。例如，你能找到的所有代码生成Agent（如Claude Code、Codex、Cursor等），都是基于其产品需求定制架构构建的。

当然，这并非否定现有框架的价值。LangChain、LangGraph和LlamaIndex等框架在处理标准任务（例如RAG或自动化工作流）时非常有用。关键在于：在将库应用于真实复杂任务之前，你需要透彻理解其能力与局限。

我至今仍经常使用LangGraph，但更多是在原型设计阶段。它非常适合制作演示，或用于教学Agentic架构的设计模式。

在这篇及后续文章中，我将手把手带你构建一个具备基础能力并包含若干高级特性的AI Agent。同时，我也会展示一些常用且实用的设计模式实现。

你可以在以下Colab笔记本中找到完整代码并自行尝试：https://colab.research.google.com/drive/1a1hAyRo5f-3ct3a2t0m2C-jdaTymhsSY?usp=sharing

AI Agent 究竟是什么？

AI Agent 种类繁多，如今已随处可见。过去的简单聊天机器人（如ChatGPT）现已演变为拥有工具的AI Agent——它们能够进行网页搜索、推理、生成图像等。Agent的复杂度取决于其需要达成的目标。

例如，一个面向网站访客的客服Agent，可能就是一个配备了RAG工具的聊天机器人（用于提供准确、最新的回答），外加一个在找不到可靠答案或需要人工介入时，能自动为客服团队起草邮件的工具。

从本质上讲，一个AI Agent是一个能够执行以下循环的系统：

1. 感知（Perceive） 其环境（理解用户输入）
2. 推理（Reason） 下一步应采取什么行动
3. 行动（Act），通过使用工具或直接回复
4. 学习（Learn） 结果与反馈（第二部分会详细讲解）

今天我们要构建的，是一个使用ReAct（Reasoning + Acting）模式，实现前三项能力的基础Agent。

Basic Agent Architecture

架构总览

在开始编码之前，先理解我们要搭建的整体结构。本次我们将构建三大核心组件：

1. 工具系统（Tool System）：管理所有可用工具的灵活注册表
2. LLM包装器（LLM Wrapper）：与大语言模型交互的抽象层
3. Agent协调器（Agent Orchestrator）：负责整体流程协调的“大脑”

为什么要采用这样的分层架构？

• 工具抽象（Tool Abstraction）：通过工具注册表，我们可以在不改动核心逻辑的情况下，轻松为Agent增加新能力。例如，你需要一个数据库查询函数？只需注册一个新工具即可。这体现了可扩展性原则，也是所有Agent的共性。
• LLM与Agent分离（LLM/Agent Separation）：这对生产系统尤为关键。Agent是协调器，负责管理对话流程、决定何时调用工具、处理整体工作流。LLM只是一个提供推理的组件，它像Agent的“大脑”，但我们需要能够随时“换脑”。

通过这种解耦设计，你可以实现：

• 在不同LLM提供商（如Gemini、OpenAI、Claude）之间切换，无需重写Agent逻辑；
• 实现失败回退（fallback）策略；
• 通过为不同任务分配不同模型来优化成本；
• 更易于测试：可以独立模拟（mock）LLM。

第一步：构建工具系统

让我们从基础设施开始。工具是Agent的“手脚”，使其能够与外部世界交互。

Tool 类

首先，我们需要一种方式来表示单个工具：

“`python
from typing import Dict, List, Callable, Any

class Tool:
def init(
self,
name: str,
description: str,
input_schema: Dict[str, Any],
output_schema: Dict[str, Any],
func: Callable[…, Any],
):
self.name = name
self.description = description
self.input_schema = input_schema
self.output_schema = output_schema
self.func = func

def __call__(self, **kwargs):
    return self.func(**kwargs)

“`

每个工具包含五个关键部分：

• name：唯一标识符
• description：工具的功能描述（这对LLM理解何时使用它至关重要）
• input_schema：定义工具期望的参数
• output_schema：定义工具返回的数据结构
• func：实际执行工作的函数

__call__ 方法使得Tool实例可以像普通函数一样被调用：tool(a=5, b=3)。

Tool Registry

现在我们需要一个中心化的地方来管理所有工具。

“`python
from typing import Union, Literal
from pydantic import BaseModel

class ToolRegistry:
def init(self):
self.tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
self.tools[tool.name] = tool
def get(self, name: str) -> Tool:
if name not in self.tools.keys():
raise ValueError(f”Tool ‘{name}’ not found”)
return self.tools[name]
def list_tools(self) -> List[Dict[str, Any]]:
return [
{
“name”: tool.name,
“description”: tool.description,
“input_schema”: tool.input_schema.model_json_schema(),
}
for tool in self.tools.values()
]
def get_tool_call_args_type(self) -> Union[BaseModel]:
input_args_models = [tool.input_schema for tool in self.tools.values()]
tool_call_args = Union[tuple(input_args_models)]
return tool_call_args
def get_tool_names(self) -> Literal[None]:
return Literal[*self.tools.keys()]
“`

Registry 是所有能力的中央目录，主要用于注册和获取工具。

• list_tools()：告知 LLM 可用能力
  该方法生成所有可用工具的机器可读描述。将其放入系统提示词（system prompt）中，LLM 便能知晓自己可以访问哪些能力。返回格式如下：

json
[
{
"name": "add",
"description": "Add two numbers",
"input_schema": {
"type": "object",
"properties": {
"a": {"type": "integer"},
"b": {"type": "integer"}
},
"required": ["a", "b"]
}
},
{
"name": "multiply",
"description": "Multiply two numbers",
"input_schema": {...}
}
]

这个 JSON Schema 明确告知 LLM 如何调用每个工具。没有它，LLM 可能会“幻觉”出不存在的工具名，或以错误的格式传递参数。

• get_tool_call_args_type()：运行时参数校验
  该方法创建所有可能工具参数 Schema 的 Union 类型。在 Python 类型系统中，Union 表示“其中之一”。例如，如果你有两个工具，就会得到 Union[ToolAddArgs, ToolMultiplyArgs]。

这很重要：当 LLM 返回一次工具调用时，Pydantic 会校验参数是否匹配这些 Schema 中的某一个。如果 LLM 尝试传递 {"a": "five", "b": 3}（将整数写成了字符串），Pydantic 会在工具执行前捕获错误。这能防止运行时异常并提供清晰的反馈。

注：我们使用 Pydantic，因为它正成为 LLM API 进行工具调用与结构化输出的事实标准。当然，你也可以在 Pydantic 模型与简易 JSON 之间轻松转换。

• get_tool_names()：防止工具名幻觉
  该方法生成一个仅包含有效工具名的 Literal 类型，例如 Literal["add", "multiply"]。这是一个强约束：LLM 只能返回 Registry 中真实存在的工具名。

没有它，LLM 可能会自信地调用一个你从未创建过的“divide”工具。通过结构化输出与 Literal 约束，LLM 被迫从允许的集合中选择；如果试图使用无效名称，API 会拒绝响应并让模型重试。

这三个方法共同构建了一个稳健的类型安全体系，弥合了 LLM 的概率世界与 Python 代码的确定世界之间的鸿沟，将模糊的请求转化为可验证、可执行的函数调用。

下面看看如何使用工具抽象类与 Registry 来创建并注册新工具。

注册我们的第一个工具

首先创建两个简单的工具函数来演示系统：

“`python
def add(a: int, b: int) -> int:
return a + b

def multiply(a: int, b: int) -> int:
return a * b
“`

以上是执行时会被调用的工具函数。

接下来，使用 Pydantic 为每个工具定义输入参数 Schema：

“`python
class ToolAddArgs(BaseModel):
a: int
b: int

class ToolMultiplyArgs(BaseModel):
a: int
b: int
“`

“`python
registry = ToolRegistry()

registry.register(
Tool(
name=”add”,
description=”Add two numbers”,
input_schema=ToolAddArgs,
output_schema={“result”: “int”},
func=add,
)
)
registry.register(
Tool(
name=”multiply”,
description=”Multiply two numbers”,
input_schema=ToolMultiplyArgs,
output_schema={“result”: “int”},
func=multiply,
)
)
“`

第二步：使用 Pydantic 实现类型安全

你可能会问：“为什么不直接使用字典或 JSON？” 这涉及到结构化输出与类型安全的核心问题。

在与大语言模型（LLM）协作时，一个主要挑战是确保其返回的数据格式能够被你的代码可靠地解析和处理。即使到了2026年，我们已经拥有了在工具调用上训练有素的可靠 LLM，但“幻觉”问题依然存在，因此类型与结构校验至关重要。

Pydantic 模型充当了数据契约的角色，它们能够：

1. 自动校验输入数据的有效性。
2. 在数据无效时提供清晰、可读的错误信息。
3. 为集成开发环境（IDE）提供更好的自动补全支持，提升开发体验。
4. 生成现代 LLM 可直接使用的 JSON Schema，用于驱动结构化输出。

接下来，我们定义 Agent 可能执行的动作类型：

“`python

获取类型安全的工具名和参数类型

ToolNameLiteral = registry.get_tool_names()
ToolArgsUnion = registry.get_tool_call_args_type()

class ToolCall(BaseModel):
action: Literal[“tool”]
thought: str
tool_name: ToolNameLiteral
args: ToolArgsUnion

class FinalAnswer(BaseModel):
action: Literal[“final”]
answer: str

LLMResponse = Union[ToolCall, FinalAnswer]
“`

这个数据结构强制实现了 ReAct（推理-行动）模式。LLM 必须：

• 选择一个动作类型（"tool" 或 "final"）。
• 如果调用工具：提供思考过程（thought）、工具名称（tool_name）以及符合该工具 Schema 的有效参数（args）。
• 如果给出最终答案：直接提供答案文本（answer）。

其中，ToolNameLiteral 确保了 LLM 只能调用已注册的真实工具；ToolArgsUnion 则确保了传入的参数必须匹配对应工具所期望的 Schema。

第三步：LLM 包装器

现在，我们将集成 Google 的 Gemini API。本教程选择 Gemini 是因为它提供了免费的 API 配额。当然，你也可以使用其他服务提供商，只需根据其 API 文档调整此类的实现即可。

“`python
import json
from google import genai
from google.genai import types

class GeminiLLM:
def init(self, client, tool_registry, model=”gemini-2.5-flash”):
self.client = client
self.model = model
self.tool_registry = tool_registry
self.system_instruction = self._create_system_instruction()
“`

系统提示词

“`python
def _create_system_instruction(self) -> str:
tools_description = json.dumps(
self.tool_registry.list_tools(),
indent=2
)

system_prompt = """

You are a conversational AI agent that can interact with external tools.
CRITICAL RULES (MUST FOLLOW):
– You are NOT allowed to perform operations internally that could be performed by an available tool.
– If a tool exists that can perform any part of the task, you MUST use that tool.
– You MUST NOT skip tools, even for simple or obvious steps.
– You MUST NOT combine multiple operations into a single step unless a tool explicitly supports it.
– You may ONLY produce a final answer when no available tool can further advance the task.
TOOL USAGE RULES:
– Each tool call must perform exactly ONE meaningful operation.
– If the task requires multiple operations, you MUST call tools sequentially.
– If multiple tools could apply, choose the most specific one.
RESPONSE FORMAT (STRICT):
– You MUST respond ONLY in valid JSON.
– Never include explanations outside JSON.
– You must choose exactly one action per response.
Tool call format:
{
“action”: “tool”,
“thought”: “…”,
“tool_name”: “…”,
“inputs”: { … }
}
Final answer format:
{
“action”: “final”,
“answer”: “…”
}””” + “nnAvailable tools with description:n” + tools_description
return system_prompt
“`

为什么规则要如此严格？大型语言模型（LLM）通常被训练为“尽可能提供帮助”，这可能导致它们倾向于在内部进行计算或推理。然而，为了确保Agent具备可观测性与可靠性，我们强制其在每一步都使用工具。这种做法带来了以下优势：

• 记录与调试：能够记录并调试Agent的每一步操作。
• 可替换性：在不修改Agent核心逻辑的前提下，可以替换工具的具体实现。
• 独立测试：工具可以独立于Agent进行测试。
• 审计轨迹：保留清晰、完整的操作审计轨迹。

这正是ReAct（推理-行动）模式的核心：显式的推理（“thought”）与显式的动作（“tool_name” + “inputs”）相结合。

为Gemini格式化对话历史

不同的LLM服务提供商通常采用不同的消息格式。以下代码将通用的历史记录格式转换为Gemini API所需的格式：

python
def _format_gemini_chat_history(self, history: list[dict]) -> list:
formatted_history = []
for message in history:
if message["role"] == "user":
formatted_history.append(types.Content(
role="user",
parts=[
types.Part.from_text(text=message["content"])
]
)
)
if message["role"] == "assistant":
formatted_history.append(types.Content(
role="model",
parts=[
types.Part.from_text(text=message["content"])
]
)
)
if message["role"] == "tool":
formatted_history.append(types.Content(
role="tool",
parts=[
types.Part.from_function_response(
name=message["tool_name"],
response={'result': message["tool_response"]},
)
]
)
)
return formatted_history

这种抽象至关重要。我们的Agent核心逻辑使用简单、与提供商无关的通用消息格式，而每个LLM封装器（wrapper）则负责处理与特定提供商API的格式转换细节。

启用结构化输出的响应生成

python
def generate(self, history: list[dict]) -> str:
gemini_history_format = self._format_gemini_chat_history(history)
response = self.client.models.generate_content(
model=self.model,
contents=gemini_history_format,
config=types.GenerateContentConfig(
temperature=0,
response_mime_type="application/json",
response_schema=LLMResponse,
system_instruction=self.system_instruction,
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True)
),
)
return response.text

关键参数解析：
* temperature=0：设置为0以获得确定性、可复现的模型输出。
* response_mime_type="application/json"：强制模型以JSON格式进行响应。
* response_schema=LLMResponse：使用预定义的Pydantic模型对输出的JSON进行结构化和验证。
* automatic_function_calling=...disable=True：禁用模型的自动函数调用功能，将工具调用的控制权完全交由Agent逻辑处理。

第四步：构建Agent协调器

现在，我们将所有组件整合到一个核心的Agent类中，由其负责管理整个对话循环的执行流程。

python
class Agent:
def __init__(self, llm, tool_registry, max_steps=5):
self.llm = llm
self.tool_registry = tool_registry
self.history = []
self.max_steps = max_steps

max_steps参数用于限制Agent的最大思考步数，这是一个重要的安全机制，可以防止因逻辑错误或意外输入导致Agent陷入无限循环。

实现ReAct循环

Agent的核心执行逻辑遵循经典的“思考-行动-观察”（ReAct）循环模式，其run方法实现如下：

“`python
def run(self, user_input: str):
self.history.append({“role”: “user”, “content”: user_input})

for step in range(self.max_steps):
    # 1. 思考：基于完整对话历史，获取LLM的决策
    llm_output = self.llm.generate(self.history)
    action = json.loads(llm_output)

    # 2. 行动与观察：根据决策类型执行相应操作
    if action["action"] == "tool":
        # 记录思考过程
        self.history.append(
            {"role": "assistant", "content": llm_output}
        )
        # 执行工具调用
        tool = self.tool_registry.get(action["tool_name"])
        result = tool(**action["args"])
        # 记录工具执行结果（观察）
        self.history.append(
            {"role": "tool", "tool_name": tool.name, "tool_response": result}
        )
        continue

    if action["action"] == "final":
        self.history.append(
            {"role": "assistant", "content": llm_output}
        )
        return action["answer"]

raise RuntimeError("Agent did not terminate within max_steps")

“`

循环步骤解析：
1. 初始化：将用户输入添加到对话历史中，为LLM提供完整上下文。
2. 决策生成：LLM基于当前历史，生成一个结构化的决策（调用工具或给出最终答案）。
3. 工具调用路径：
* 将LLM的“思考”决策记录到历史。
* 从工具注册表中获取指定工具并执行。
* 将工具返回的结果作为“观察”记录到历史，循环进入下一步。
4. 最终答案路径：将最终答案记录到历史，并返回答案，结束循环。
5. 安全终止：如果循环达到max_steps仍未得到最终答案，则抛出异常，防止无限执行。

第五步：整合与运行

初始化并创建简单的对话界面

以下代码展示了如何初始化客户端、创建LLM与Agent，并启动一个简单的命令行对话循环。

“`python
from google import genai

初始化客户端（需要提供你的API密钥）

client = genai.Client(api_key=GEMINI_API_KEY)

创建LLM和Agent实例

llm = GeminiLLM(client, registry)
agent = Agent(llm, registry)

def chat_with_agent(agent: Agent):
print(“Welcome! Type ‘exit’ to quit.n”)
while True:
user_input = input(“You: “)
if user_input.lower() in [“exit”, “quit”, “q”]:
print(“Goodbye!”)
break
try:
response = agent.run(user_input)
print(f”Agent: {response}”)
except RuntimeError as e:
print(f”Agent error: {e}”)
except Exception as e:
print(f”Unexpected error: {e}”)

启动对话

chat_with_agent(agent)
“`

示例：观察Agent的执行流程

向Agent提问：“What is 5 plus 3, then multiply the result by 2?”

步骤 1： LLM接收问题并生成第一个工具调用。
json
{
"action": "tool",
"thought": "I need to first add 5 and 3",
"tool_name": "add",
"args": {"a": 5, "b": 3}
}

步骤 2： Agent执行 add(5, 3)，得到结果 8。

步骤 3： LLM根据上一步的结果生成下一个工具调用。
json
{
"action": "tool",
"thought": "Now I need to multiply 8 by 2",
"tool_name": "multiply",
"args": {"a": 8, "b": 2}
}

步骤 4： Agent执行 multiply(8, 2)，得到结果 16。

步骤 5： LLM生成最终答案。
json
{
"action": "final",
"answer": "The result is 16"
}

整个过程清晰展示了Agent如何将复杂任务分解为离散步骤、按需调用工具，并最终给出答案。这种透明性使得AI Agent更易于调试，也更具可信度。

架构设计的重要性

你可能会觉得，对于一个简单的计算任务，这套架构似乎过于复杂。然而，这套基础架构的强大之处恰恰体现在以下几个方面：

1. 可扩展性（Extensibility）

例如，要添加一个天气查询功能，只需定义工具函数并注册即可，无需修改Agent的核心逻辑。

“`python
class WeatherArgs(BaseModel):
city: str

def get_weather(city: str) -> str:
# 此处实现API调用
return f”Weather in {city}: Sunny, 72°F”

registry.register(Tool(
name=”get_weather”,
description=”Get current weather for a city”,
input_schema=WeatherArgs,
output_schema={“weather”: “str”},
func=get_weather
))
“`

LLM会通过系统提示词自动“学习”并使用这个新工具。

2. 提供商灵活性（Provider Flexibility）

如果需要从Gemini切换到OpenAI，只需实现一个遵循相同接口的OpenAILLM类，然后替换即可，其他部分保持不变。

“`python
class OpenAILLM:
def init(self, client, tool_registry, model=”gpt-4″):
# 结构类似，但调用不同的API
pass
def generate(self, history: list[dict]) -> str:
# OpenAI特定的实现
pass

替换LLM

llm = OpenAILLM(openai_client, registry)
agent = Agent(llm, registry) # 其他所有代码无需改动
“`

3. 可测试性（Testability）

每个组件都可以独立测试：
* 对单个工具函数进行单元测试。
* 通过模拟（Mock）LLM来测试Agent的逻辑。
* 进行端到端测试以验证整个流程。

4. 可观测性（Observability）

每一步交互都被记录在历史中，这带来了强大的调试和分析能力：
* 记录所有工具调用，便于问题排查。
* 分析工具的使用频率，优化系统设计。
* 识别Agent的薄弱环节或常见错误。
* 基于此构建监控和分析仪表盘。

当前成果总结

至此，我们已经构建了一个具备以下特性的基础AI Agent：
* 模块化的工具系统：易于添加和管理功能。
* 类型安全的结构化输出：确保工具调用的可靠性。
* 与提供商无关的LLM集成：便于切换底层模型。
* ReAct推理模式：实现了思考与行动的循环。
* 清晰的关注点分离：使系统更易于理解和维护。

然而，这只是一个起点。当前的Agent仍然是无状态的（对话间没有记忆），缺乏人工干预机制，并且可观测性能力有限。

下一步：迈向生产级Agent

在后续部分，我们将为这个Agent增添生产环境所需的关键能力：

1. 长期记忆（Long-term Memory）：利用向量数据库存储和检索过往对话，使Agent能够从历史交互中学习。
2. 人在回路（Human-in-the-Loop, HITL）：在关键操作（如高风险工具调用）前暂停，等待人工审核批准。
3. 高级可观测性（Advanced Observability）：集成结构化日志、分布式链路追踪和性能监控。
4. 错误恢复（Error Recovery）：优雅地处理工具调用失败，并实现智能重试逻辑。

这些特性对于构建可靠、可控、可投入实际应用的AI系统至关重要。

动手实践

最好的学习方式是亲手构建。你可以基于本文提供的代码框架进行扩展和实验。

• 增加你自己的工具（API、数据库查询、文件操作）
• 试验不同的 system prompt
• 尝试不同的 LLM 提供商
• 故意“整坏”它看看会发生什么！

完整的项目代码可在以下 Colab 笔记本中获取并运行：https://colab.research.google.com/drive/1a1hAyRo5f-3ct3a2t0m2C-jdaTymhsSY?usp=sharing

关注“鲸栖”小程序，掌握最新AI资讯