5.4 执行 Execution：从决策到行动
5.4.1 从记忆到行动：执行层的角色
在上一节中，我们深入探讨了 Agent 的记忆系统——从短期的工作记忆到长期的向量记忆库，记忆让 Agent 拥有了"记住过去"的能力。然而，仅有记忆是不够的。一个人如果只会回忆却不会行动，就像一个瘫痪在床的学者，满腹经纶却无法改变任何现实。
执行层（Execution）就是 Agent 的**"手和脚"**。
大脑负责思考和决策，记忆负责存储和检索，而执行层负责将这一切落地为真实的行动。当规划层制定了计划、LLM 选择了工具之后，执行层接过接力棒，真正去调用那个 API、执行那段代码、查询那个数据库，然后把结果带回来交给大脑继续思考。
用一个生活中的类比来理解整个 Agent 架构：
人类类比：
大脑（规划层） → 我想去超市买菜，先列个清单
记忆层 → 我记得冰箱里还有鸡蛋，上次去的超市在东边
执行层（手脚） → 穿鞋出门 → 走到超市 → 挑选蔬菜 → 结账 → 回家如果没有执行层，Agent 就只能"纸上谈兵"——它能告诉你应该做什么，但永远无法真正做点什么。执行层是连接"思考"与"行动"的最后一公里。
5.4.2 执行层架构总览
执行层并非简单的"调用一个函数"，它是一个完整的工程系统。从接收 LLM 的工具调用决策，到最终返回执行结果，中间经历了路由、执行、解析、验证、错误处理等多个环节。
┌─────────────────────────────────────────────────────────────┐
│ 执行层（Execution）架构 │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ LLM 决策 │───▶│ 工具调度器 │───▶│ 工具执行 │ │
│ │ (选择工具+参数)│ │ (Tool Router) │ │ (Tool Exec) │ │
│ └──────────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │
│ ┌───────────────────────────▼────────┐ │
│ │ 结果处理层 │ │
│ │ ┌────────────────────────────────┐ │ │
│ │ │ 1. 解析（Parse） │ │ │
│ │ │ 2. 验证（Validate） │ │ │
│ │ │ 3. 格式化（Format） │ │ │
│ │ │ 4. 错误处理（Error Handle） │ │ │
│ │ │ 5. 重试（Retry） │ │ │
│ │ └───────────────┬────────────────┘ │ │
│ └─────────────────┼──────────────────┘ │
│ ▼ │
│ 反馈给 LLM / 返回用户 │
└─────────────────────────────────────────────────────────────┘让我们逐一拆解这个管道的每个环节。
5.4.3 Function Calling 机制详解
执行层的核心能力来自 Function Calling——让 LLM 不仅能生成文字，还能"调用函数"。这是 2023 年以来 LLM 最重要的一项能力升级，也是 Agent 能从"聊天机器人"进化为"行动智能体"的关键。
Function Calling 的本质是：我们向 LLM 提供一组工具定义（用 JSON Schema 描述），LLM 在对话中根据用户需求自动判断是否需要调用工具、调用哪个工具、传什么参数。
这个过程就像一个人看菜谱做菜：
用户说："北京今天天气怎么样？"
↓
LLM 思考：用户想知道天气 → 我有 get_weather 工具 → 城市参数是"北京"
↓
LLM 返回工具调用：get_weather(city="北京")
↓
执行层：真正调用天气 API → 拿到结果 "晴 25°C"
↓
结果反馈给 LLM → LLM 生成自然语言回复："北京今天晴天，气温 25 摄氏度。"下面是完整的 Function Calling 代码实现：
python
from openai import OpenAI
# 创建 OpenAI 客户端实例
# 实际项目中建议将 API Key 放在环境变量中
client = OpenAI()
# ============================================================
# 第一步：定义工具（JSON Schema 格式）
# 工具定义是 LLM 理解"我能做什么"的唯一途径
# 每个工具需要：名称、描述、参数规范
# ============================================================
tools = [
{
# type 固定为 "function"，表示这是一个函数调用
"type": "function",
"function": {
"name": "get_weather", # 工具名称，LLM 用这个名字来调用
"description": "获取指定城市的实时天气信息", # 描述越清晰，LLM 选择越准确
"parameters": {
"type": "object", # 参数整体是一个 JSON 对象
"properties": {
"city": {
"type": "string", # city 参数是字符串类型
"description": "城市名称，如'北京'、'上海'"
# description 非常关键：它帮助 LLM 理解该填什么值
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"], # 只接受这两个值
"description": "温度单位"
}
},
"required": ["city"] # city 是必填参数，unit 可选
}
}
},
{
"type": "function",
"function": {
"name": "search_database",
"description": "在内部数据库中搜索信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询语句"
},
"limit": {
"type": "integer", # 返回数量是整数
"description": "返回结果数量上限",
"default": 10 # 不传时默认返回 10 条
}
},
"required": ["query"]
}
}
}
]
# ============================================================
# 第二步：调用 LLM，将工具定义一并发送
# ============================================================
response = client.chat.completions.create(
model="gpt-4o",
messages=[
# system 消息设定角色和基本行为
{"role": "system", "content": "你是一个有用的助手，可以调用工具获取信息。"},
# user 消息是用户的实际请求
{"role": "user", "content": "北京今天天气怎么样？"}
],
tools=tools, # 把工具定义传给 LLM
tool_choice="auto" # auto = 让模型自己决定是否调用工具
# 也可以指定 "none"（不调用）或 {"type":"function","function":{"name":"get_weather"}}（强制调某个）
)
# ============================================================
# 第三步：解析 LLM 的响应
# LLM 可能直接回复文字，也可能返回工具调用请求
# ============================================================
message = response.choices[0].message
if message.tool_calls:
# LLM 决定调用工具——遍历所有工具调用
for tool_call in message.tool_calls:
print(f"🛠️ 调用工具: {tool_call.function.name}")
print(f"📥 参数: {tool_call.function.arguments}")
# 注意：此时工具还没有真正执行！
# tool_call.function.arguments 是一个 JSON 字符串，如 '{"city": "北京"}'
# 我们需要自己解析并执行对应的函数
else:
# LLM 直接回复，不需要调用工具
print(f"📝 直接回复: {message.content}")这里有一个关键认知：LLM 本身并不执行任何代码。它只是"说"要调用哪个函数、传什么参数。真正的执行工作由我们的代码完成。LLM 是决策者，执行层是执行者。
5.4.4 工具调用执行流程
让我们用一个更完整的视角来看待工具调用的完整执行流程。从用户输入到最终回复，一个完整的循环包含以下步骤：
用户输入："帮我查一下上海和北京的天气，然后比较哪个更热"
│
▼
┌─ 步骤1：LLM 推理 ──────────────────────────────────────────┐
│ LLM 分析：需要调用两次 get_weather，分别查上海和北京 │
│ 返回两个 tool_call： │
│ ① get_weather(city="上海") │
│ ② get_weather(city="北京") │
└──────────────────────────────────────────────────────────┘
│
▼
┌─ 步骤2：工具路由 ──────────────────────────────────────────┐
│ 执行层收到两个工具调用请求 │
│ 检查 get_weather 是否已注册 → 找到对应的执行函数 │
│ 判断是否可以并行执行 → 两个调用互不依赖，可以并行 │
└──────────────────────────────────────────────────────────┘
│
▼
┌─ 步骤3：工具执行 ──────────────────────────────────────────┐
│ 并行调用天气 API： │
│ ① get_weather("上海") → {"temp": 33, "weather": "晴"} │
│ ② get_weather("北京") → {"temp": 28, "weather": "多云"} │
└──────────────────────────────────────────────────────────┘
│
▼
┌─ 步骤4：结果解析与验证 ────────────────────────────────────┐
│ 检查返回格式是否正确 → 格式符合预期 │
│ 解析 JSON → 提取温度数据 │
│ 统一格式化为 ToolResult 对象 │
└──────────────────────────────────────────────────────────┘
│
▼
┌─ 步骤5：结果反馈给 LLM ───────────────────────────────────┐
│ 将两个工具的执行结果以 tool 消息形式追加到对话中 │
│ LLM 收到结果后进行推理：33 > 28，上海更热 │
│ 生成最终回复："上海 33°C，北京 28°C，上海更热。" │
└──────────────────────────────────────────────────────────┘
│
▼
返回用户这个流程体现了执行层的核心价值：它不仅执行工具，还管理整个执行的生命周期——从路由调度到结果反馈，形成一个闭环。
5.4.5 工具路由与调度
当 Agent 拥有多个工具时，需要一个**工具路由器（Tool Router）**来管理工具的注册、查找和调度。这就像一个公司的前台接待员：接到请求后，要知道该把请求转给哪个部门。
python
from typing import Dict, Any, Callable
from enum import Enum
# 定义工具分类，用于日志记录和策略选择
class ToolCategory(Enum):
SEARCH = "search" # 搜索类工具
CALCULATE = "calculate" # 计算类工具
DATABASE = "database" # 数据库操作类工具
FILE = "file" # 文件操作类工具
class ToolRouter:
"""工具路由器：根据工具调用结果分发到正确的执行器
职责：
1. 维护工具注册表（名称 -> 执行函数）
2. 路由工具调用请求到正确的执行器
3. 支持批量并行执行多个工具调用
"""
def __init__(self):
# 执行函数注册表：工具名 -> 可调用对象
self.executors: Dict[str, Callable] = {}
# 分类映射表：工具名 -> 工具类别
self.category_map: Dict[str, ToolCategory] = {}
def register(self, name: str, executor: Callable, category: ToolCategory):
"""注册一个工具
Args:
name: 工具名称，必须与 LLM 工具定义中的 name 一致
executor: 实际执行的 Python 函数
category: 工具分类，用于路由策略
"""
self.executors[name] = executor
self.category_map[name] = category
def route(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""路由到对应的工具执行器并执行
Args:
tool_name: LLM 返回的工具名称
arguments: LLM 返回的参数字典
Returns:
工具执行的原始结果
Raises:
ValueError: 当工具未注册时
"""
if tool_name not in self.executors:
raise ValueError(f"未知工具: {tool_name}")
executor = self.executors[tool_name]
category = self.category_map[tool_name]
print(f"🔀 路由: {tool_name} -> {category.value}")
# **arguments 将字典解包为关键字参数
# 例如 {"city": "北京"} 会被解包为 executor(city="北京")
return executor(**arguments)
def execute_batch(self, tool_calls: list) -> list:
"""批量执行工具调用（支持并行）
当 LLM 一次返回多个工具调用时，用线程池并行执行，
显著减少总执行时间。例如同时查两个城市天气，串行需 2 秒，
并行只需 1 秒。
Args:
tool_calls: 工具调用列表，每项包含 name 和 arguments
Returns:
结果列表，每项包含工具名、结果/错误、状态
"""
import concurrent.futures
results = []
# 创建线程池，最大 5 个并发
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
futures = {} # future -> tool_name 的映射
# 提交所有工具调用任务
for call in tool_calls:
future = executor.submit(
self.route,
call["name"],
call["arguments"]
)
futures[future] = call["name"]
# 等待所有任务完成，先完成的先处理
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
results.append({
"tool": futures[future],
"result": result,
"status": "success"
})
except Exception as e:
# 某个工具失败不影响其他工具的结果
results.append({
"tool": futures[future],
"error": str(e),
"status": "failed"
})
return results下面是工具路由器的使用示例，展示注册和批量并行调用的效果：
python
# 创建路由器实例
router = ToolRouter()
# 定义两个具体的工具函数
def search_web(query: str, **kwargs):
"""网页搜索工具：根据关键词返回搜索结果"""
return f"搜索结果: {query}"
def calculate(expression: str, **kwargs):
"""计算器工具：执行数学表达式"""
return eval(expression) # 生产环境应使用安全的表达式解析器
# 注册工具到路由器
router.register("search_web", search_web, ToolCategory.SEARCH)
router.register("calculate", calculate, ToolCategory.CALCULATE)
# 模拟 LLM 返回的批量工具调用
# 场景：用户问"AI Agent 有哪些？顺便算一下 2+3×4"
# LLM 可能同时返回搜索和计算两个工具调用
tool_calls = [
{"name": "search_web", "arguments": {"query": "AI Agent"}},
{"name": "calculate", "arguments": {"expression": "2+3*4"}},
]
# 并行执行——两个工具同时运行，总时间 ≈ 较慢的那个
results = router.execute_batch(tool_calls)
print(results)
# 输出示例：
# 🔀 路由: search_web -> search
# 🔀 路由: calculate -> calculate
# [{'tool': 'calculate', 'result': 14, 'status': 'success'},
# {'tool': 'search_web', 'result': '搜索结果: AI Agent', 'status': 'success'}]5.4.6 工具调用编排模式
在实际场景中，Agent 需要执行的工具调用并非总是独立的。根据工具间的依赖关系，可以归纳为三种编排模式：
模式1：顺序执行（Sequential）
┌───┐ ┌───┐ ┌───┐
│ A │───▶│ B │───▶│ C │ B 需要 A 的结果作为输入
└───┘ └───┘ └───┘ 例如：先搜索 → 再翻译 → 再总结
模式2：并行执行（Parallel）
┌───┐
│ A │───┐
└───┘ │
┌───┐ ├──▶ 汇总
│ B │───┤ A、B、C 互不依赖，可同时执行
└───┘ │ 例如：同时查上海、北京、广州的天气
┌───┐ │
│ C │───┘
└───┘
模式3：条件分支（Conditional）
┌───▶ 路径A
┌───┐ │
│判断│──┤ 根据某个结果选择后续路径
└───┘ │ 例如：天气 > 35°C → 提醒防暑
└───▶ 路径B ≤ 35°C → 正常回复这三种模式在实际应用中经常组合使用。例如一个复杂的旅行规划 Agent：先并行查询多个目的地的天气（并行模式），然后根据天气选择最佳目的地（条件分支），最后依次预订机票和酒店（顺序执行）。
5.4.7 结果解析与验证
工具执行完成后，返回的结果格式千差万别——有的是 JSON 字符串，有的是纯文本，有的是字典，有的可能直接就是数字。执行层需要将这些异构结果统一解析为标准格式，才能可靠地反馈给 LLM 或返回给用户。
这就像一个翻译官：不同国家的人说不同语言，翻译官把它们都翻译成统一的中间语言。
python
import json
from typing import Any, Optional
from pydantic import BaseModel, ValidationError, Field
# ============================================================
# 1. 定义统一的结果模型
# 所有工具的返回结果都应被归一化为这个格式
# ============================================================
class ToolResult(BaseModel):
"""统一的工具返回结果格式
无论原始工具返回什么，最终都被包装为 ToolResult
这样下游处理逻辑（LLM 反馈、日志记录等）只需要处理一种格式
"""
success: bool = Field(description="是否执行成功")
data: Optional[Any] = Field(default=None, description="返回数据，类型不限")
error: Optional[str] = Field(default=None, description="失败时的错误信息")
metadata: dict = Field(default_factory=dict, description="元数据（执行时间、重试次数等）")
@classmethod
def ok(cls, data: Any, **metadata):
"""快速构造成功结果"""
return cls(success=True, data=data, metadata=metadata)
@classmethod
def fail(cls, error: str, **metadata):
"""快速构造失败结果"""
return cls(success=False, error=error, metadata=metadata)
# ============================================================
# 2. 结果解析器
# 将各种格式的原始结果统一解析为 ToolResult
# ============================================================
class ResultParser:
"""通用结果解析器：将异构结果归一化"""
@staticmethod
def parse(raw_result: Any) -> ToolResult:
"""将各种格式的原始结果解析为统一格式
处理顺序：
1. 如果已经是 ToolResult，直接返回
2. 如果是字符串，尝试解析为 JSON
3. 如果是字典，提取 success/data/error 字段
4. 其他类型，直接包装为成功结果
"""
# 情况1：已经是标准格式，无需转换
if isinstance(raw_result, ToolResult):
return raw_result
# 情况2：字符串——可能是 JSON 文本，也可能是纯文本
if isinstance(raw_result, str):
try:
parsed = json.loads(raw_result) # 尝试解析为 JSON
if isinstance(parsed, dict):
return ToolResult(
success=parsed.get("success", True),
data=parsed.get("data", parsed),
error=parsed.get("error")
)
except json.JSONDecodeError:
pass # 不是 JSON，当作纯文本处理
return ToolResult.ok(raw_result) # 纯文本直接包装为成功
# 情况3：字典——直接提取字段
if isinstance(raw_result, dict):
return ToolResult(
success=raw_result.get("success", True),
data=raw_result.get("data", raw_result),
error=raw_result.get("error")
)
# 情况4：其他类型（int、float、list 等）
return ToolResult.ok(raw_result)
@staticmethod
def validate(result: ToolResult, expected_schema: dict = None) -> bool:
"""验证结果是否符合预期模式
Args:
result: 已解析的 ToolResult
expected_schema: 期望的数据模式，如 {"type": "number"}
"""
if not result.success:
return False
if expected_schema and result.data:
expected_type = expected_schema.get("type")
# 检查数据类型是否匹配预期
if expected_type == "number" and not isinstance(result.data, (int, float)):
return False
if expected_type == "string" and not isinstance(result.data, str):
return False
return True下面用几个测试用例展示解析器的行为：
python
parser = ResultParser()
# 测试各种输入格式
inputs = [
"简单文本结果", # 纯文本 → 包装为成功
'{"success": true, "data": [1, 2, 3]}', # JSON 字符串 → 解析提取
{"status": "error", "message": "API超时"}, # 字典 → 提取字段
42, # 数字 → 直接包装
]
for inp in inputs:
result = parser.parse(inp)
print(f"输入: {str(inp)[:50]:50s} -> 解析: {result}")5.4.8 错误处理与重试策略
执行层面对的外部世界并不总是可靠的：网络会超时，API 会限流，数据库会断连。一个健壮的执行层必须具备完善的错误处理和重试机制。
错误可以分为三类，处理策略各不相同：
| 错误类型 | 典型场景 | 处理策略 |
|---|---|---|
| 可重试 | 网络超时、API 限流、临时不可用 | 等待后重试（指数退避） |
| 可降级 | 主服务挂了，备用服务可用 | 切换到备用方案 |
| 致命错误 | 权限不足、参数格式错误 | 直接报错，不重试 |
python
import time
import functools
from typing import Type, Tuple
from enum import Enum
class ErrorSeverity(Enum):
"""错误严重级别"""
RETRYABLE = "retryable" # 可重试：网络超时、限流等临时性问题
DEGRADABLE = "degradable" # 可降级：主路径失败，可用备选方案
FATAL = "fatal" # 致命错误：权限不足、参数错误，重试无意义
class RetryStrategy:
"""重试策略配置
核心思想：不是所有错误都值得重试。
只对"临时性"错误重试，且使用指数退避避免雪崩。
"""
def __init__(
self,
max_retries: int = 3, # 最多重试次数
base_delay: float = 1.0, # 首次重试等待时间（秒）
max_delay: float = 30.0, # 最大等待时间上限（秒）
backoff_factor: float = 2.0, # 退避因子：每次等待时间 = base × factor^attempt
retryable_exceptions: Tuple[Type[Exception], ...] = (TimeoutError, ConnectionError)
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.backoff_factor = backoff_factor
self.retryable_exceptions = retryable_exceptions
# 指数退避示意：
# 第1次重试等待 1.0 秒
# 第2次重试等待 2.0 秒
# 第3次重试等待 4.0 秒
# 避免短时间内反复冲击已故障的服务
def with_retry(strategy: RetryStrategy = None):
"""装饰器：为函数添加重试逻辑
用法：
@with_retry()
def call_api(): ...
@with_retry(RetryStrategy(max_retries=5))
def call_api(): ...
"""
if strategy is None:
strategy = RetryStrategy() # 使用默认策略
def decorator(func):
@functools.wraps(func) # 保留原函数的元信息
def wrapper(*args, **kwargs):
last_exception = None # 记录最后一次异常
# 尝试 max_retries + 1 次（首次 + 重试次数）
for attempt in range(strategy.max_retries + 1):
try:
return func(*args, **kwargs) # 尝试执行
except strategy.retryable_exceptions as e:
# 捕获到可重试异常
last_exception = e
if attempt == strategy.max_retries:
break # 已达最大重试次数，退出循环
# 计算指数退避等待时间
delay = min(
strategy.base_delay * (strategy.backoff_factor ** attempt),
strategy.max_delay
)
print(f"⚠️ 第{attempt+1}次重试，等待{delay:.1f}秒... ({e})")
time.sleep(delay) # 等待后重试
except Exception as e:
# 非可重试异常（如参数错误），直接抛出不重试
raise
# 所有重试都失败了，抛出最后的异常
raise last_exception
return wrapper
return decorator下面用重试装饰器保护一个可能失败的外部 API 调用：
python
# 用重试策略装饰一个会随机失败的 API 调用函数
@with_retry(RetryStrategy(max_retries=3, base_delay=0.5))
def call_external_api(endpoint: str, params: dict) -> dict:
"""模拟调用外部API（有 30% 概率失败）"""
import random
if random.random() < 0.3: # 30% 概率模拟网络故障
raise ConnectionError(f"无法连接到 {endpoint}")
return {"status": "ok", "data": f"来自{endpoint}的响应"}
# 测试重试效果
try:
result = call_external_api("https://api.example.com", {"query": "test"})
print(f"✅ 成功: {result}")
except Exception as e:
print(f"❌ 最终失败: {e}")运行时可能的输出：
⚠️ 第1次重试，等待0.5秒... (无法连接到 https://api.example.com)
✅ 成功: {'status': 'ok', 'data': '来自https://api.example.com的响应'}重试策略让原本会直接失败的调用，通过几次重试成功恢复。这在调用第三方 API 时尤其重要——很多服务的 SLA 承诺 99.9% 可用，意味着每 1000 次调用可能有 1 次失败，重试机制能将这个失败率再降低几个数量级。
5.4.9 完整的执行管道
将前面所有组件——工具路由、结果解析、错误处理、重试策略——整合为一个完整的执行管道，这就是生产级 Agent 执行层的核心实现：
python
class ExecutionPipeline:
"""Agent 执行管道：整合路由、解析、重试的完整执行系统
职责：
1. 接收用户任务，驱动 LLM 进行多轮工具调用
2. 管理工具的注册和调度
3. 处理工具执行结果，包括错误恢复
4. 将结果反馈给 LLM，形成闭环
"""
def __init__(self, llm, max_iterations: int = 10):
"""初始化执行管道
Args:
llm: 大语言模型客户端（需支持 tool calling）
max_iterations: 最大迭代轮次，防止无限循环
"""
self.llm = llm
self.router = ToolRouter() # 工具路由器
self.parser = ResultParser() # 结果解析器
self.max_iterations = max_iterations # 安全阀：防止死循环
self.retry_strategy = RetryStrategy() # 默认重试策略
def execute_tool(self, tool_name: str, arguments: dict) -> ToolResult:
"""执行单个工具调用（带重试和错误处理）
这是执行管道的核心方法，每个工具调用都经过：
路由 → 执行 → 重试 → 解析 → 错误处理
"""
try:
# 使用 functools.partial 固定部分参数，构造可重试的调用
executor = functools.partial(self.router.route, tool_name, arguments)
# 套上重试装饰器执行
raw_result = with_retry(self.retry_strategy)(executor)()
# 解析原始结果为统一格式
return self.parser.parse(raw_result)
except Exception as e:
# 任何未恢复的异常都转为失败结果，而非崩溃
return ToolResult.fail(
error=str(e),
tool=tool_name,
arguments=arguments
)
def run(self, task: str) -> str:
"""运行完整的执行管道
这是 Agent 的主循环：
LLM 决策 → 执行工具 → 反馈结果 → LLM 再决策 → ...
直到 LLM 不再需要工具，直接给出最终答案
"""
messages = [{"role": "user", "content": task}] # 初始对话
iterations = 0
while iterations < self.max_iterations:
iterations += 1
# 步骤1：让 LLM 决策（选择工具或直接回复）
response = self.llm.invoke(messages)
# 步骤2：检查是否需要调用工具
if not hasattr(response, 'tool_calls') or not response.tool_calls:
# LLM 不需要工具，直接返回答案
return response.content
# 步骤3：执行 LLM 选择的工具调用
tool_results = []
for tc in response.tool_calls:
result = self.execute_tool(tc.name, tc.arguments)
tool_results.append({
"tool_call_id": tc.id, # 关联工具调用与结果
"name": tc.name,
"result": result
})
# 步骤4：将工具结果反馈给 LLM
messages.append(response) # 将 LLM 的工具调用决策加入对话
for tr in tool_results:
messages.append({
"role": "tool", # tool 角色表示工具返回
"tool_call_id": tr["tool_call_id"], # 关联到对应的调用
"content": json.dumps(tr["result"].model_dump()) # 序列化结果
})
# 回到循环开头，让 LLM 根据工具结果继续决策
# 安全阀触发：超出最大迭代次数
return "达到最大迭代次数，任务未完成"这个执行管道体现了 Agent 的循环驱动本质：不是一次性执行完所有步骤，而是在"思考-行动-观察"的循环中逐步推进。每一轮循环，LLM 都会根据上一轮的工具执行结果来决定下一步行动，直到任务完成或达到安全上限。
5.4.10 超时控制
除了重试，另一个关键的保护机制是超时控制。有些工具不会报错，而是卡住——比如一个慢速 API 永远不返回。如果没有超时控制，Agent 就会无限期等待。
python
import threading
from typing import Any
class TimeoutError(Exception):
"""工具执行超时异常"""
pass
def execute_with_timeout(func, timeout: int, *args, **kwargs) -> Any:
"""在指定时间内执行函数，超时则抛出异常
原理：在子线程中执行目标函数，主线程等待指定时间。
如果超时，主线程不再等待，直接抛出异常。
Args:
func: 要执行的函数
timeout: 超时时间（秒）
*args, **kwargs: 传给 func 的参数
Returns:
func 的返回值
Raises:
TimeoutError: 超时
Exception: func 内部抛出的异常
"""
result = [None] # 用列表存储结果（闭包中可变）
error = [None] # 用列表存储异常
completed = threading.Event() # 线程完成事件
def target():
"""子线程执行的目标函数"""
try:
result[0] = func(*args, **kwargs)
except Exception as e:
error[0] = e
finally:
completed.set() # 标记完成
thread = threading.Thread(target=target)
thread.daemon = True # 设为守护线程：主进程退出时自动结束
thread.start()
if not completed.wait(timeout=timeout):
# 等待 timeout 秒，如果没完成
raise TimeoutError(f"工具执行超时 ({timeout}秒)")
if error[0]:
raise error[0] # 子线程内部发生了异常
return result[0]测试超时场景：
python
# 模拟一个慢速 API 调用
def slow_api_call(delay: int):
"""模拟延迟 delay 秒的 API 调用"""
import time
print(f"⏳ 模拟API调用（延迟{delay}秒）...")
time.sleep(delay)
return f"API结果（延迟{delay}秒）"
# 场景1：超时——函数需要5秒，但只给了2秒
try:
result = execute_with_timeout(slow_api_call, timeout=2, delay=5)
print(f"✅ 结果: {result}")
except TimeoutError as e:
print(f"⏰ 超时: {e}")
# 输出: ⏰ 超时: 工具执行超时 (2秒)
# 场景2：正常——函数需要1秒，给了3秒
try:
result = execute_with_timeout(slow_api_call, timeout=3, delay=1)
print(f"✅ 结果: {result}")
except TimeoutError as e:
print(f"⏰ 超时: {e}")
# 输出: ✅ 结果: API结果（延迟1秒）超时控制和重试策略是执行层的两道安全网：重试处理"失败"，超时处理"卡住"。两者配合使用，才能确保执行层不会因为外部原因而无限阻塞。
5.4.11 常见误区
在实际构建 Agent 执行层的过程中，开发者容易陷入以下误区：
误区一：以为 LLM 会自己执行工具
这是最常见的误解。很多初学者以为给 LLM 定义了工具，它就会自己去调用 API、执行代码。实际上，LLM 只是返回一段结构化文本（工具名 + 参数 JSON），真正的执行必须由你的代码完成。LLM 是"指挥官"，你的执行层才是"士兵"。
python
# ❌ 错误理解：以为 LLM 会自动执行
response = llm.invoke("帮我查北京天气")
# 以为天气数据已经自动获取了——并没有！
# ✅ 正确做法：自己解析并执行
if response.tool_calls:
for tc in response.tool_calls:
result = execute_function(tc.name, tc.arguments) # 你来执行误区二：工具描述写得太模糊
工具的 description 是 LLM 选择工具的唯一依据。如果描述写得含糊，LLM 就会在错误的时候调用错误的工具。
python
# ❌ 不好的描述
"description": "搜索"
# ✅ 好的描述
"description": "在互联网上搜索最新资讯，输入关键词返回相关网页摘要"描述应该回答三个问题：这个工具做什么？什么时候该用它？什么时候不该用它？
误区三：不设最大迭代次数
Agent 的主循环是 while True 式的——LLM 持续决策、执行工具、再决策。如果不设上限，一旦 LLM 陷入"反复调用工具但不收敛"的状态，就会无限循环消耗 token。
python
# ❌ 危险：没有上限
while True:
response = llm.invoke(messages)
# 可能永远不退出...
# ✅ 安全：设置最大迭代次数
max_iterations = 10
while iterations < max_iterations:
# ...误区四：对所有错误都重试
有些错误重试再多次也不会成功——比如 API Key 过期、参数格式不对。对这些"致命错误"重试是浪费时间和资源。
python
# ❌ 对所有异常都重试
except Exception as e:
retry() # 参数错误也重试？毫无意义
# ✅ 区分可重试和不可重试异常
except (TimeoutError, ConnectionError):
retry() # 临时性问题，值得重试
except (ValueError, PermissionError):
raise # 永久性问题，立即失败误区五：工具返回结果不做格式校验
LLM 依赖工具返回的结果继续推理。如果返回了意外格式的数据（比如期望 JSON 却返回了 HTML 错误页面），LLM 可能会"幻觉"出错误结论。
python
# ❌ 不检查结果
result = tool.execute()
return result # 直接返回，不管格式对不对
# ✅ 校验后返回
result = tool.execute()
validated = parser.parse(result)
if not parser.validate(validated, expected_schema={"type": "dict"}):
return ToolResult.fail("返回格式不符合预期")误区六：忽略工具执行的副作用
有些工具具有副作用——发邮件、扣款、删除文件。如果 Agent 误调用了这类工具，后果可能不可逆。对于有副作用的工具，应该加入确认机制或人工审核环节。
5.4.12 本节小结
执行层是 Agent 的"手和脚"——大脑做出了决策，记忆提供了上下文，而执行层负责将这一切转化为真实的行动。让我们回顾本节的核心内容：
1. 执行层的定位
执行层是连接"思考"与"行动"的桥梁。它接收 LLM 的工具调用决策，完成路由调度、工具执行、结果解析、错误恢复，最终将结果反馈给 LLM 形成闭环。
2. Function Calling 机制
通过 JSON Schema 定义工具，LLM 自动选择工具并构造参数。关键认知：LLM 只输出"调用意图"，真正的执行由代码完成。
3. 工具路由与调度
ToolRouter 管理工具注册表，支持单次路由和批量并行执行。三种编排模式：顺序执行、并行执行、条件分支。
4. 结果解析与验证
ToolResult 统一格式 + ResultParser 通用解析器，将异构结果归一化，确保下游处理的一致性。
5. 错误处理与重试
错误三分类（可重试、可降级、致命错误），配合指数退避重试策略和超时控制，构成执行层的安全网。
6. 完整执行管道
将以上组件整合为循环驱动的执行管道：LLM 决策 → 工具执行 → 结果反馈 → 再决策，直到任务完成。
回望整个第 5 章，我们已经走过了 Agent 架构的几个关键组件：规划层是"大脑"，负责制定策略和分解任务；记忆层是"记忆"，负责存储和检索上下文信息；而执行层是"手和脚"，负责将决策落地为行动。三者构成了 Agent 行动的基础闭环。
然而，一个成熟的 Agent 不仅要会行动，还要会"复盘"。执行完毕后，Agent 是否达成了目标？哪些做得好，哪些可以改进？下次遇到类似问题，能否做得更快更好？这些问题，就需要**反思（Reflection）**机制来回答。
在下一节 5.5 中，我们将探讨 Agent 的反思能力——如何让 Agent 在行动之后回看自己的执行过程，自我评估、发现不足、持续改进。反思让 Agent 从"会做事"进化为"越做越好"，这是通向真正智能的关键一步。