执行模块

决策模块告诉智能体"应该做什么"，而执行模块则负责"真正去做"。执行模块是智能体将意图转化为现实影响的桥梁——它通过调用API、使用工具、操作环境等方式，将抽象的决策变为具体的行动。没有执行模块，智能体就只是一个"空想家"，永远无法对真实世界产生任何影响。本章将深入探讨执行模块的三大核心能力：API调用、工具使用和环境操作。

API调用：连接外部服务的桥梁

在现代软件生态中，API（应用程序编程接口）是不同系统之间通信的标准方式。智能体通过API调用可以访问海量的外部服务：搜索引擎、数据库、支付系统、邮件服务、社交媒体平台等。API调用能力极大地扩展了智能体的"能力边界"。

智能体的API调用需要解决几个关键问题：

• 接口发现：智能体需要知道有哪些API可用、每个API的功能和参数格式。这通常通过API描述文件（如OpenAPI/Swagger规范）来实现。
• 参数构造：根据决策结果和API规范，自动构造正确的请求参数。这要求智能体具备理解API文档和映射业务逻辑的能力。
• 错误处理：API调用可能失败（网络超时、权限不足、参数错误等）。执行模块需要具备重试、降级和错误报告的能力。
• 结果解析：将API返回的原始响应解析为智能体内部可以使用的结构化数据。

# API调用执行器示例

import json
import time

class APICallExecutor:
    """API调用执行器 - 安全可靠的API调用"""

    def __init__(self):
        self.api_registry = {}     # 已注册的API
        self.call_history = []     # 调用历史
        self.rate_limits = {}      # 速率限制

    def register_api(self, name, endpoint, method, params_schema, auth=None):
        """注册一个可用的API"""
        self.api_registry[name] = {
            "endpoint": endpoint,
            "method": method,
            "params_schema": params_schema,
            "auth": auth,
            "available": True
        }

    def validate_params(self, api_name, params):
        """验证参数是否符合API规范"""
        if api_name not in self.api_registry:
            return False, f"API '{api_name}' 未注册"

        schema = self.api_registry[api_name]["params_schema"]
        for required_param in schema.get("required", []):
            if required_param not in params:
                return False, f"缺少必需参数: {required_param}"

        return True, "参数验证通过"

    def execute(self, api_name, params, max_retries=3):
        """执行API调用（含重试机制）"""
        # 1. 参数验证
        valid, msg = self.validate_params(api_name, params)
        if not valid:
            return {"status": "error", "message": msg}

        api = self.api_registry[api_name]

        # 2. 速率限制检查
        if not self._check_rate_limit(api_name):
            return {"status": "error", "message": "超出速率限制"}

        # 3. 执行调用（含重试）
        for attempt in range(max_retries):
            try:
                result = self._make_call(api, params)
                self._record_call(api_name, params, result, attempt)
                return result
            except Exception as e:
                if attempt == max_retries - 1:
                    self._record_call(api_name, params,
                                     {"status": "failed", "error": str(e)},
                                     attempt)
                    return {"status": "error",
                            "message": f"调用失败(重试{max_retries}次): {str(e)}"}
                time.sleep(2 ** attempt)  # 指数退避

    def _make_call(self, api, params):
        """实际发起HTTP调用（简化示例）"""
        # 实际项目中使用 httpx / requests 等库
        return {
            "status": "success",
            "data": {"mock": True},
            "latency_ms": 150
        }

    def _check_rate_limit(self, api_name):
        """简单的速率限制"""
        current = self.rate_limits.get(api_name, 0)
        if current >= 100:  # 每分钟最多100次
            return False
        self.rate_limits[api_name] = current + 1
        return True

    def _record_call(self, api_name, params, result, attempt):
        """记录调用历史"""
        self.call_history.append({
            "api": api_name,
            "params": params,
            "result": result,
            "attempt": attempt,
            "timestamp": time.time()
        })

# 使用示例
executor = APICallExecutor()
executor.register_api(
    name="send_email",
    endpoint="https://api.mailgun.com/v3/messages",
    method="POST",
    params_schema={"required": ["to", "subject", "body"]}
)

result = executor.execute("send_email", {
    "to": "user@example.com",
    "subject": "周报提醒",
    "body": "请提交本周工作周报。"
})
print(f"执行结果: {result}")

工具使用：智能体的多功能工具箱

工具使用（Tool Use）是LLM驱动的智能体最具标志性的能力之一。通过工具使用，智能体可以突破纯文本生成的限制，执行代码、搜索信息、操作文件、查询数据库等。工具使用使智能体从"只会说"进化为"能说又能做"。

工具使用的核心机制包括：

工具描述：每个工具都有清晰的描述，包括名称、功能说明、参数格式和返回值格式。LLM通过这些描述来理解每个工具的用途和使用方法，从而决定何时调用哪个工具。

工具选择：面对多个可用工具，智能体需要根据当前任务选择最合适的工具。这要求智能体理解工具的适用场景和局限性。

工具编排：复杂任务往往需要多个工具的组合使用。智能体需要规划工具调用的顺序，处理工具之间的数据传递和依赖关系。

结果验证：工具执行完成后，智能体需要验证结果是否符合预期。如果结果不理想，可能需要调整参数重新调用或选择替代工具。

环境操作：直接改变世界

除了通过API和工具间接影响环境，智能体有时需要直接操作环境。这包括文件系统操作（读写文件、创建目录）、数据库操作（增删改查记录）、系统配置（修改环境变量、启动服务）等。

环境操作与API调用的关键区别在于：环境操作通常是"有副作用的"——它会永久性地改变环境状态。因此，环境操作需要更严格的安全控制：

• 权限控制：明确智能体可以操作哪些资源、执行哪些操作。采用最小权限原则，只授予完成任务所必需的权限。
• 操作审计：记录每一次环境操作的详细信息，包括操作者、操作内容、操作时间和操作结果。
• 回滚机制：提供操作撤销能力。当操作产生非预期后果时，可以回滚到操作前的状态。
• 沙箱隔离：在隔离的环境中执行操作，防止错误操作影响生产系统。

# 环境操作执行器示例：安全的文件系统操作

class EnvironmentOperator:
    """环境操作执行器 - 安全可控的环境操作"""

    def __init__(self, allowed_paths, sandbox=True):
        self.allowed_paths = allowed_paths  # 允许操作的路径
        self.sandbox = sandbox              # 是否启用沙箱模式
        self.operation_log = []             # 操作日志
        self.undo_stack = []                # 回滚栈

    def write_file(self, path, content):
        """安全写入文件"""
        # 1. 权限检查
        if not self._check_permission(path):
            return {"status": "denied", "message": f"无权限操作: {path}"}

        # 2. 备份原文件（用于回滚）
        backup = self._backup_file(path)

        # 3. 执行写入
        try:
            if not self.sandbox:
                with open(path, "w", encoding="utf-8") as f:
                    f.write(content)

            # 4. 记录操作
            self.undo_stack.append({
                "action": "write_file",
                "path": path,
                "backup": backup
            })

            self.operation_log.append({
                "action": "write_file",
                "path": path,
                "size": len(content),
                "timestamp": time.time()
            })

            return {"status": "success", "path": path}

        except Exception as e:
            # 写入失败时恢复备份
            if backup:
                self._restore_backup(path, backup)
            return {"status": "error", "message": str(e)}

    def undo(self):
        """撤销最近一次操作"""
        if not self.undo_stack:
            return {"status": "empty", "message": "无可撤销操作"}

        last_op = self.undo_stack.pop()
        if last_op["action"] == "write_file":
            self._restore_backup(last_op["path"], last_op["backup"])
            return {"status": "undone", "action": last_op["action"]}

    def _check_permission(self, path):
        """检查路径权限"""
        import os
        abs_path = os.path.abspath(path)
        return any(abs_path.startswith(p) for p in self.allowed_paths)

    def _backup_file(self, path):
        """备份文件"""
        import os
        if os.path.exists(path):
            with open(path, "r", encoding="utf-8") as f:
                return f.read()
        return None

    def _restore_backup(self, path, backup):
        """恢复备份"""
        if backup is not None:
            with open(path, "w", encoding="utf-8") as f:
                f.write(backup)

# 使用示例
operator = EnvironmentOperator(
    allowed_paths=["/tmp/agent_workspace"],
    sandbox=True
)
result = operator.write_file(
    "/tmp/agent_workspace/report.txt",
    "这是智能体生成的报告内容。"
)
print(f"操作结果: {result}")

执行模块的可靠性设计

执行模块是智能体与真实世界交互的最后一环，其可靠性至关重要。一个设计良好的执行模块应当具备以下特性：

幂等性：相同的操作执行多次应当产生相同的结果。这对于网络调用尤为重要——当超时发生时，重试不应导致重复操作。

超时控制：每个操作都应当设置合理的超时时间，避免智能体因等待某个操作而无限阻塞。

优雅降级：当某个执行能力不可用时（如外部API宕机），智能体应当能够降级到备选方案或通知用户，而不是直接崩溃。

执行确认：对于重要操作，执行模块应当提供确认机制——在执行前请求确认，执行后报告结果。

本章小结

执行模块是智能体将决策转化为现实影响的关键组件。它通过API调用连接外部服务，通过工具使用扩展能力边界，通过环境操作直接改变世界。可靠的执行模块需要具备参数验证、错误处理、重试机制、权限控制、操作审计和回滚能力等工程特性。安全性是执行模块设计中的首要考量——必须设定不可逾越的安全红线。下一章我们将探讨智能体的学习能力，看看它如何从执行结果中持续改进。