AI Agent Harness Engineering 工具生态盘点：从API集成到自定义工具开发的全流程

1. 引入与连接：开启AI Agent工程之旅

1.1 一个引人深思的场景

想象一下，你是一家快速发展的科技公司的CTO。你的团队刚刚开发了一个强大的AI助手，可以帮助客户解决常见问题，但你很快发现，这个助手的能力受到了严重限制——它只能回答它训练过的问题，无法获取实时数据，无法执行特定任务，也无法与你公司现有的系统进行交互。

你的客户开始抱怨：“为什么这个助手不能帮我查看我的订单状态？”“它怎么不知道我们最新的产品价格？”“我想让它帮我生成一份销售报告，但它好像做不到。”

你意识到，你的AI助手需要"装备"更多的"工具"才能真正发挥价值。你需要一种方式，让AI能够安全、高效地使用各种服务、数据库和API，甚至能够根据需要创建新的工具。

恭喜你，你已经进入了AI Agent Harness Engineering的世界——一个正在快速发展的领域，它正在改变我们构建和部署AI系统的方式。

1.2 与你已有知识的连接

如果你曾经开发过应用程序，你可能熟悉API集成的概念——将不同的软件系统连接在一起，使它们能够相互通信。如果你使用过像Zapier或IFTTT这样的自动化工具，你已经体验过将不同服务连接起来创造新功能的力量。

AI Agent Harness Engineering将这些概念提升到了一个新的层次。它不仅仅是连接系统，而是让AI智能体能够自主决定何时、如何以及使用哪些工具来完成任务。这就像给你的AI助手配备了一个万能工具箱，并教会它如何选择和使用正确的工具。

如果你熟悉LangChain、AutoGPT或BabyAGI这些工具，你已经接触过AI Agent的概念。但在这篇文章中，我们将深入探讨更广泛的生态系统，以及如何从简单的API集成过渡到构建复杂的自定义工具。

1.3 为什么这对你很重要

AI Agent Harness Engineering正在迅速成为AI应用开发的核心范式。根据Gartner的预测，到2025年，超过80%的企业AI应用将包含Agent组件，而能够有效管理这些Agent的工具生态系统将成为竞争优势的关键来源。

无论你是一名AI研究员、软件工程师、产品经理还是企业决策者，理解这个生态系统都将帮助你：

• 构建更强大、更灵活的AI应用
• 提高开发效率，减少重复工作
• 确保AI系统的安全性和可控性
• 为未来的AI应用创新打下基础

1.4 我们的学习路径

在这篇文章中，我们将沿着以下路径探索AI Agent Harness Engineering的世界：

1. 概念地图：首先，我们将绘制这个领域的概念地图，帮助你建立整体认知框架。
2. 基础理解：然后，我们将从基础开始，解释核心概念和简单模型。
3. 层层深入：接着，我们将逐步增加复杂度，探讨原理、机制和底层逻辑。
4. 多维透视：我们将从历史、实践、批判和未来等多个角度审视这个领域。
5. 实践转化：然后，我们将进入实践环节，提供具体的操作步骤和案例。
6. 整合提升：最后，我们将总结核心观点，并提供进一步学习的资源。

准备好开始这段旅程了吗？让我们先从构建概念地图开始。

2. 概念地图：建立AI Agent工具生态的整体认知

在深入探讨细节之前，让我们先构建一个整体的概念地图，帮助你理解AI Agent Harness Engineering的各个组成部分以及它们之间的关系。

2.1 核心概念与关键术语

首先，让我们定义一些核心概念和关键术语：

概念	定义	类比
AI Agent	能够感知环境、做出决策并采取行动的自主AI系统	一个能够独立完成任务的智能助手
工具(Tool)	Agent可以使用的功能或服务，用于执行特定任务	工具箱中的锤子、螺丝刀等
工具集(Toolkit)	一组相关工具的集合	一套完整的工具箱
Harness	用于管理和控制Agent使用工具的框架或系统	马具，用于控制和引导马的行动
工具调用(Tool Calling)	Agent使用工具的过程	使用工具箱中的某个工具完成任务
工具开发(Tool Engineering)	创建新工具的过程	设计和制造新的工具
API集成	将外部API连接到Agent的过程	将电器插头插入插座
自定义工具	专门为特定Agent或任务创建的工具	定制的特殊工具
工具注册表	用于存储和管理可用工具的系统	工具目录或索引
权限控制	管理Agent可以使用哪些工具以及如何使用的机制	安全锁和访问权限
工具编排	协调多个工具使用的过程	指挥乐队演奏

2.2 概念层次与关系

AI Agent Harness Engineering的概念可以分为以下几个层次：

1. 基础层：核心概念（Agent、工具、调用）
2. 连接层：API集成、工具注册表、权限控制
3. 构建层：自定义工具开发、工具集创建
4. 编排层：多工具协调、工作流管理
5. 优化层：性能优化、安全性增强、用户体验提升

这些层次之间存在着密切的关系。例如，没有基础层的概念，就无法理解连接层的API集成；没有连接层的基础设施，就无法进行构建层的自定义工具开发。

2.3 学科定位与边界

AI Agent Harness Engineering是一个跨学科领域，它结合了以下学科的知识：

• 人工智能：特别是强化学习、规划和决策理论
• 软件工程：API设计、系统架构、测试和部署
• 安全工程：权限管理、沙箱技术、风险评估
• 人机交互：用户体验设计、自然语言接口
• DevOps：持续集成、部署监控、资源管理

同时，我们也需要明确它的边界：

• 它不完全等同于AI开发，而是专注于AI与工具的交互
• 它不完全等同于API开发，而是关注API如何被Agent使用
• 它不完全等同于自动化，而是强调Agent的自主性和决策能力

2.4 AI Agent工具生态系统概览

让我们用一个简单的图示来展示AI Agent工具生态系统的主要组件和它们之间的关系：

这个图示展示了AI Agent工具生态系统的五个主要层次：

1. 用户层：与Agent交互的用户或应用程序
2. Agent层：核心AI系统，包括规划、执行和记忆组件
3. Harness层：管理和控制Agent与工具交互的中间层
4. 工具层：Agent可以使用的各种工具
5. 资源层：工具访问的底层资源

在接下来的章节中，我们将深入探讨每个层次的细节，以及它们如何协同工作。

3. 基础理解：AI Agent工具生态的直观认识

现在我们已经建立了整体概念框架，让我们深入到基础层，建立对AI Agent工具生态的直观认识。

3.1 核心概念的生活化解释

让我们从最基本的概念开始，用生活化的比喻来解释AI Agent和工具的关系。

3.1.1 AI Agent是什么？

你可以把AI Agent想象成一个智能助手，比如一个私人助理。这个助理有能力理解你的指令，思考如何完成任务，然后采取行动。

但是，就像一个没有任何工具的助理一样，一个没有工具的AI Agent能力非常有限。它可以回答问题，进行对话，但无法完成需要与外界交互的任务。

3.1.2 工具是什么？

工具就是Agent可以使用的功能或服务，就像助理可以使用的各种设备和服务。例如：

• 一个"计算器"工具，就像助理手里的计算器
• 一个"天气查询"工具，就像助理可以查看天气应用
• 一个"邮件发送"工具，就像助理可以帮你发送邮件
• 一个"数据库查询"工具，就像助理可以查阅档案柜

没有这些工具，Agent就像一个被蒙住眼睛、捆住双手的助理，空有智慧却无法施展。

3.1.3 Harness是什么？

Harness是用来管理和控制Agent使用工具的框架或系统，就像管理助理工作的一套规则和流程。它确保：

• Agent只能使用它被允许使用的工具（权限控制）
• Agent使用工具的方式是安全的（安全控制）
• Agent使用工具的过程被记录和监控（审计追踪）
• 多个工具可以协调工作（流程编排）

想象一下，如果你的助理可以不受限制地使用你的银行账户、发送你的私人邮件或访问你的机密文件，那会有多危险？Harness就是确保Agent安全、可控地使用工具的关键。

3.2 简化模型与类比

让我们用一个更具体的类比来理解整个系统：餐厅厨房。

3.2.1 餐厅厨房类比

• AI Agent = 主厨
• 工具 = 厨房设备（刀具、炉灶、搅拌机等）和食材
• Harness = 厨房管理系统（食谱、安全规定、设备使用指南）
• 工具调用 = 主厨使用某个设备或食材
• API集成 = 从外部供应商获取食材
• 自定义工具 = 专门为这家餐厅设计的特殊设备
• 用户 = 点餐的顾客

在这个类比中，主厨（Agent）需要根据顾客的订单（用户指令），使用厨房中的设备和食材（工具），按照食谱和安全规定（Harness）来制作美食。

如果没有合适的工具，即使是最好的主厨也无法做出美食；如果没有适当的管理，厨房可能会变得混乱，甚至发生危险。

3.2.2 工具箱类比

另一个有用的类比是工具箱：

• AI Agent = 工人
• 工具集 = 工具箱
• 单个工具 = 工具箱中的锤子、螺丝刀、扳手等
• Harness = 工具箱的组织系统和安全指南
• 工具选择 = 工人根据任务选择合适的工具
• 工具使用 = 工人正确使用工具完成任务

在这个类比中，工人（Agent）需要根据任务（用户需求）从工具箱（工具集）中选择合适的工具，并按照安全指南（Harness）正确使用它们。

3.3 直观示例与案例

让我们通过一个具体的例子来看看AI Agent如何使用工具：

3.3.1 旅行规划助手

假设你有一个旅行规划AI助手，你让它帮你规划一次从北京到上海的周末旅行。

没有工具的话，助手只能告诉你：“我很乐意帮你规划旅行，但我无法查看实时航班信息或酒店空房情况。”

有了工具的话，对话可能是这样的：

你：帮我规划一次从北京到上海的周末旅行，预算5000元以内。

助手：好的，我来帮你规划。首先，让我查看一下周末的航班信息。
【助手使用"航班查询"工具】

助手：我找到了几个合适的航班选项。现在让我查看一下上海的酒店空房情况。
【助手使用"酒店查询"工具】

助手：根据你的预算，我找到了几个不错的酒店。让我再查一下上海周末的天气情况，以便给你推荐一些活动。
【助手使用"天气查询"工具】

助手：好的，根据我的查询结果，我为你规划了以下行程：

1. 周六上午乘坐CA1234航班从北京到上海
2. 入住上海外滩附近的XYZ酒店
3. 周六下午参观豫园（天气预报显示天气晴朗）
4. 周六晚上在南京路步行街用餐
5. 周日上午参观上海博物馆
6. 周日下午乘坐CA5678航班返回北京

总预算约为4500元，符合你的要求。你觉得这个规划如何？

在这个例子中，助手使用了三个工具：航班查询、酒店查询和天气查询，来完成一个它原本无法完成的任务。

3.3.2 数据分析师助手

让我们再看一个更专业的例子——数据分析师助手：

你：分析一下我们公司第三季度的销售数据，重点关注华东地区的表现，并生成一份可视化报告。

助手：好的，我来帮你分析。首先，让我连接到公司的销售数据库。
【助手使用"数据库连接"工具】

助手：现在让我查询第三季度的销售数据，特别是华东地区的数据。
【助手使用"SQL查询"工具】

助手：数据已获取，让我进行一些基本的统计分析。
【助手使用"数据分析"工具】

助手：分析完成，现在让我创建一些可视化图表。
【助手使用"数据可视化"工具】

助手：最后，让我将所有内容整合成一份报告。
【助手使用"报告生成"工具】

助手：你的报告已准备好！以下是主要发现：

1. 华东地区第三季度销售额同比增长23%，超过公司平均水平
2. 产品A在华东地区的销量最高，占总销售额的45%
3. 9月份是销售最高峰，主要归功于中秋促销活动
4. 上海和杭州是华东地区销售最好的两个城市

[包含图表的完整报告]

你需要我对任何特定方面进行更深入的分析吗？

这个例子展示了AI Agent如何使用一系列专业工具来完成复杂的数据分析任务。

3.4 常见误解澄清

在我们继续深入之前，让我们澄清一些关于AI Agent和工具的常见误解：

3.4.1 误解1：“Agent越强大，需要的工具越少”

事实：正好相反。Agent越强大，它能够有效使用的工具就越多，也就需要更多的工具来发挥其全部潜力。就像一个技能高超的工匠需要更多、更专业的工具一样。

3.4.2 误解2：“给Agent越多工具越好”

事实：工具的质量和相关性比数量更重要。给Agent太多不相关的工具会导致"选择悖论"——Agent可能会困惑于选择哪个工具，或者选择错误的工具。

3.4.3 误解3：“Agent会自动知道如何使用任何工具”

事实：Agent需要明确的指导才能使用工具。这包括工具的描述、使用方法、输入输出格式等。没有这些信息，即使是最强大的Agent也无法正确使用工具。

3.4.4 误解4：“工具调用只需要考虑功能，不需要考虑安全”

事实：安全是工具调用最重要的考虑因素之一。Agent使用工具可能会访问敏感数据、修改系统状态或产生实际影响。没有适当的安全控制，可能会导致严重的后果。

3.4.5 误解5：“AI Agent工具生态就是LangChain”

事实：LangChain是AI Agent工具生态中的一个重要组件，但不是全部。这个生态系统包括许多其他框架、工具和服务，我们将在后面的章节中详细介绍。

4. 层层深入：AI Agent工具生态的技术深度解析

现在我们已经建立了基础理解，让我们逐步深入，探索AI Agent工具生态的技术细节。我们将从基本原理开始，然后逐步深入到细节、底层逻辑和高级应用。

4.1 第一层：基本原理与运作机制

4.1.1 Agent如何决定使用哪个工具？

Agent决定使用哪个工具的过程通常包括以下几个步骤：

1. 理解任务：Agent首先需要理解用户的请求或任务目标。
2. 评估能力：Agent评估自己是否能够直接完成任务，还是需要使用工具。
3. 工具选择：如果需要工具，Agent从可用工具中选择最合适的一个或多个。
4. 工具使用：Agent生成工具调用请求，执行工具调用。
5. 结果处理：Agent处理工具返回的结果，决定下一步操作。

这个过程可以是单轮的（一次工具调用完成任务），也可以是多轮的（多次工具调用，可能使用不同工具）。

从技术角度来看，Agent通常使用以下几种方法来决定使用哪个工具：

1. 提示工程(Prompt Engineering)：在提示词中明确描述可用工具，让LLM自己决定使用哪个工具。
2. 函数调用(Function Calling)：使用LLM的内置函数调用功能，如OpenAI的Function Calling或Anthropic的Tool Use。
3. 规划器(Planner)：使用专门的规划组件来生成工具使用计划。
4. 强化学习(Reinforcement Learning)：通过训练让Agent学习如何选择和使用工具。

让我们更详细地了解一下这些方法：

4.1.1.1 提示工程方法

提示工程是最简单的方法，它通过精心设计的提示词，让LLM了解可用工具并决定如何使用它们。

一个典型的工具使用提示词可能如下：

你是一个有用的助手，你可以使用以下工具来帮助用户完成任务：

1. 天气查询工具：输入城市名称，返回该城市的当前天气。使用格式：[weather:城市名]
2. 计算器工具：输入数学表达式，返回计算结果。使用格式：[calculator:表达式]
3. 搜索工具：输入查询内容，返回搜索结果。使用格式：[search:查询]

当你需要使用工具时，请只输出工具调用格式，不要包含其他内容。当你收到工具返回的结果后，再继续处理。

如果不需要使用工具，或者已经收集到足够的信息来回答用户的问题，请直接回答用户。

然后，当用户提问时，交互可能如下：

用户：北京今天的天气怎么样？如果下雨的话，我需要带伞。

助手：[weather:北京]

系统：北京今天天气：多云，气温22-28°C，无降水。

助手：北京今天多云，气温22-28°C，不会下雨，所以你不需要带伞。

这种方法的优点是简单，不需要特殊的API或框架支持。缺点是LLM可能不会始终遵循指定的格式，特别是在处理复杂任务时。

4.1.1.2 函数调用方法

函数调用是现代LLM提供的一种内置功能，它允许LLM以结构化的方式请求调用外部函数。

以OpenAI的Function Calling为例，你可以这样定义工具：

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称",
                    },
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function", 
        "function": {
            "name": "calculator",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如'2 + 2 * 3'",
                    },
                },
                "required": ["expression"],
            },
        },
    }
]

然后，当LLM决定使用工具时，它会返回一个结构化的响应，告诉你要调用哪个函数以及传递什么参数：

{
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"city\":\"北京\",\"unit\":\"celsius\"}"
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

然后，你负责执行这个函数调用，并将结果返回给LLM：

messages = [
    {"role": "user", "content": "北京今天的天气怎么样？"},
    {"role": "assistant", "tool_calls": [tool_call_object]},
    {"role": "tool", "tool_call_id": "call_abc123", "content": "北京今天天气：多云，气温22-28°C"}
]

最后，LLM使用这些信息生成最终回答。

函数调用方法的优点是结构化、可靠，LLM经过专门训练来生成正确的函数调用格式。缺点是需要LLM支持这种功能，并且需要编写更多的代码来处理函数调用的执行。

4.1.1.3 规划器方法

规划器方法使用专门的组件来生成详细的工具使用计划，而不是让LLM在每一步临时决定。

一个典型的规划器可能会：

1. 分析任务目标
2. 生成完成任务的步骤序列
3. 为每个步骤指定需要使用的工具
4. 执行计划并根据需要调整

这种方法对于复杂任务特别有用，因为它可以提前规划整个流程，而不是依赖LLM的临时决策。

我们将在后面的章节中更详细地讨论规划器。

4.1.1.4 强化学习方法

强化学习方法通过训练Agent来学习如何选择和使用工具。在这种方法中：

1. Agent在环境中尝试使用不同的工具
2. 根据任务完成情况获得奖励或惩罚
3. 通过强化学习算法更新策略
4. 随着时间的推移，Agent学会了如何有效地选择和使用工具

这种方法的优点是Agent可以学习到非常复杂的工具使用策略，甚至可能发现人类没有想到的工具使用方式。缺点是需要大量的训练数据和计算资源，并且训练过程可能不稳定。

4.1.2 工具调用的基本流程

无论使用哪种方法，工具调用的基本流程通常如下：

1. 用户输入：用户提出问题或请求。
2. 任务分析：Agent分析用户请求，确定任务目标。
3. 工具选择：Agent选择合适的工具（如果需要）。
4. 参数生成：Agent生成工具调用所需的参数。
5. 权限检查：系统检查Agent是否有权限使用该工具。
6. 工具执行：系统执行工具调用，获取结果。
7. 结果处理：Agent处理工具返回的结果。
8. 迭代决策：Agent决定是否需要继续使用工具，或者是否可以生成最终回答。
9. 最终回答：Agent生成最终回答，返回给用户。

让我们用一个流程图来可视化这个过程：

这个流程图展示了工具调用的基本流程，包括决策点和可能的迭代过程。

4.1.3 工具的基本结构

一个典型的工具通常包括以下几个部分：

1. 元数据：工具的名称、描述、版本等基本信息。
2. 参数定义：工具接受的输入参数，包括类型、格式、验证规则等。
3. 执行逻辑：工具的核心功能实现。
4. 返回值定义：工具返回的结果格式。
5. 错误处理：处理可能出现的错误情况。
6. 安全控制：访问权限、数据脱敏等安全相关的逻辑。

让我们用一个简单的Python例子来展示工具的基本结构：

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field

# 参数定义
class WeatherParams(BaseModel):
    city: str = Field(..., description="城市名称")
    unit: Optional[str] = Field(default="celsius", description="温度单位，可选值：celsius、fahrenheit")

# 返回值定义
class WeatherResult(BaseModel):
    city: str
    temperature: float
    unit: str
    description: str
    humidity: float
    wind_speed: float

# 工具实现
class WeatherTool:
    def __init__(self, api_key: str):
        self.name = "weather_query"
        self.description = "获取指定城市的当前天气信息"
        self.version = "1.0.0"
        self.api_key = api_key
        # 权限设置
        self.required_permissions = ["weather:read"]
    
    def validate_params(self, params: Dict[str, Any]) -> WeatherParams:
        """验证输入参数"""
        return WeatherParams(**params)
    
    def check_permissions(self, user_permissions: list) -> bool:
        """检查用户权限"""
        return any(perm in self.required_permissions for perm in user_permissions)
    
    def execute(self, params: Dict[str, Any], user_permissions: list) -> Dict[str, Any]:
        """执行工具"""
        # 检查权限
        if not self.check_permissions(user_permissions):
            return {
                "success": False,
                "error": "权限不足，无法使用天气查询工具"
            }
        
        try:
            # 验证参数
            validated_params = self.validate_params(params)
            
            # 调用外部API（这里用模拟数据代替）
            result = self._call_weather_api(validated_params)
            
            return {
                "success": True,
                "data": result.dict()
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def _call_weather_api(self, params: WeatherParams) -> WeatherResult:
        """调用天气API（模拟实现）"""
        # 实际应用中，这里会调用真实的天气API
        # 这里用模拟数据代替
        return WeatherResult(
            city=params.city,
            temperature=25.5,
            unit=params.unit,
            description="多云",
            humidity=65.0,
            wind_speed=12.3
        )

这个例子展示了一个相对完整的工具实现，包括参数验证、权限检查、执行逻辑和错误处理。

4.2 第二层：细节、例外与特殊情况

在了解了基本原理之后，让我们深入探讨一些更细节的内容，包括例外情况和特殊场景。

4.2.1 工具调用的错误处理

工具调用过程中可能会出现各种错误，我们需要有良好的错误处理机制：

4.2.1.1 常见错误类型

1. 参数错误：Agent提供的参数不符合要求（类型错误、格式错误、缺少必要参数等）。
2. 权限错误：Agent没有权限使用该工具或访问特定数据。
3. 执行错误：工具执行过程中出现错误（外部API不可用、网络错误、逻辑错误等）。
4. 超时错误：工具执行时间过长，超过了允许的时间限制。
5. 资源限制错误：工具使用的资源（如API调用次数、计算资源等）超出了限制。

4.2.1.2 错误处理策略

1. 优雅降级：当主要工具不可用时，使用备用工具或方法。
2. 重试机制：对于临时性错误（如网络波动），进行有限次数的重试。
3. 清晰反馈：向Agent提供清晰、有用的错误信息，帮助它理解问题并可能纠正。
4. 日志记录：记录所有错误，以便后续分析和改进。
5. 人工介入：对于严重错误或无法自动处理的情况，触发人工介入流程。

让我们用一个例子来展示如何实现错误处理：

import time
from functools import wraps
from typing import Callable, Any, Dict

def handle_errors(max_retries: int = 3, retry_delay: float = 1.0):
    """错误处理装饰器"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Dict[str, Any]:
            retries = 0
            last_error = None
            
            while retries <= max_retries:
                try:
                    return func(*args, **kwargs)
                except ParameterError as e:
                    # 参数错误，不重试，直接返回明确的错误信息
                    return {
                        "success": False,
                        "error_type": "parameter_error",
                        "error": str(e),
                        "suggestion": e.suggestion if hasattr(e, 'suggestion') else None
                    }
                except PermissionError as e:
                    # 权限错误，不重试
                    return {
                        "success": False,
                        "error_type": "permission_error",
                        "error": str(e)
                    }
                except TimeoutError as e:
                    # 超时错误，可以考虑重试
                    last_error = e
                    retries += 1
                    if retries <= max_retries:
                        time.sleep(retry_delay * retries)  # 指数退避
                except TemporaryError as e:
                    # 临时性错误，重试
                    last_error = e
                    retries += 1
                    if retries <= max_retries:
                        time.sleep(retry_delay * retries)
                except Exception as e:
                    # 其他错误，记录但不重试
                    return {
                        "success": False,
                        "error_type": "unknown_error",
                        "error": str(e)
                    }
            
            # 所有重试都失败了
            return {
                "success": False,
                "error_type": "retry_exhausted",
                "error": f"操作失败，已重试{max_retries}次。最后错误: {str(last_error)}"
            }
        
        return wrapper
    return decorator

# 自定义异常类
class ParameterError(ValueError):
    def __init__(self, message: str, suggestion: str = None):
        super().__init__(message)
        self.suggestion = suggestion

class TemporaryError(RuntimeError):
    """临时性错误，可以重试"""
    pass

这个例子展示了一个错误处理装饰器，它可以处理不同类型的错误，并采取相应的策略。

4.2.2 多工具协调

在许多情况下，完成一个任务需要使用多个工具，这就涉及到多工具协调的问题。

4.2.2.1 多工具协调的模式

1. 顺序执行：一个接一个地使用工具，每个工具的输出可能是下一个工具的输入。
2. 并行执行：同时使用多个工具，然后合并结果。
3. 条件执行：根据前面工具的结果，决定使用哪个后续工具。
4. 循环执行：重复使用一个或多个工具，直到满足某个条件。

让我们用一个例子来展示多工具协调：

from typing import List, Dict, Any, Callable

class ToolOrchestrator:
    """工具编排器"""
    
    def __init__(self, tools: Dict[str, Any]):
        self.tools = tools
    
    def execute_sequential(self, tool_calls: List[Dict[str, Any]]) -> Dict[str, Any]:
        """顺序执行多个工具调用"""
        results = {}
        context = {}
        
        for i, tool_call in enumerate(tool_calls):
            tool_name = tool_call["name"]
            params = tool_call["params"]
            
            # 如果参数中引用了之前的结果，进行替换
            params = self._resolve_references(params, context)
            
            # 执行工具调用
            result = self.tools[tool_name].execute(params)
            
            # 存储结果
            results[f"step_{i}"] = result
            context[tool_call.get("output_var", f"step_{i}")] = result
            
            # 如果失败，决定是否继续
            if not result.get("success", False) and not tool_call.get("continue_on_failure", False):
                return {
                    "success": False,
                    "error": f"步骤 {i+1} 失败: {result.get('error', '未知错误')}",
                    "partial_results": results
                }
        
        return {
            "success": True,
            "results": results,
            "final_context": context
        }
    
    def execute_parallel(self, tool_calls: List[Dict[str, Any]]) -> Dict[str, Any]:
        """并行执行多个工具调用"""
        import concurrent.futures
        
        def execute_tool(tool_call: Dict[str, Any]) -> Dict[str, Any]:
            tool_name = tool_call["name"]
            params = tool_call["params"]
            return {
                "name": tool_name,
                "result": self.tools[tool_name].execute(params)
            }
        
        results = {}
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future_to_tool = {
                executor.submit(execute_tool, tool_call): tool_call
                for tool_call in tool_calls
            }
            
            for future in concurrent.futures.as_completed(future_to_tool):
                tool_call = future_to_tool[future]
                try:
                    result = future.result()
                    results[result["name"]] = result["result"]
                except Exception as e:
                    results[tool_call["name"]] = {
                        "success": False,
                        "error": str(e)
                    }
        
        # 检查是否有失败
        all_success = all(result.get("success", False) for result in results.values())
        
        return {
            "success": all_success,
            "results": results
        }
    
    def _resolve_references(self, params: Any, context: Dict[str, Any]) -> Any:
        """解析参数中的引用"""
        if isinstance(params, str):
            # 简单的引用解析，例如 ${step_0.data}
            import re
            pattern = r'\$\{([^}]+)\}'
            
            def replace_match(match):
                ref = match.group(1)
                try:
                    # 尝试从上下文中获取值
                    parts = ref.split('.')
                    value = context
                    for part in parts:
                        value = value[part]
                    return str(value)
                except (KeyError, TypeError):
                    return match.group(0)
            
            return re.sub(pattern, replace_match, params)
        elif isinstance(params, dict):
            return {k: self._resolve_references(v, context) for k, v in params.items()}
        elif isinstance(params, list):
            return [self._resolve_references(item, context) for item in params]
        else:
            return params

这个例子展示了一个简单的工具编排器，它可以顺序或并行执行多个工具调用，并处理工具之间的依赖关系。

4.2.3 上下文管理

当Agent使用多个工具时，它需要管理上下文信息，包括：

1. 对话历史：之前的用户输入和助手回复。
2. 工具调用历史：之前使用的工具和返回的结果。
3. 中间结果：在多步骤过程中生成的中间数据。
4. 任务状态：当前任务的进展情况。

有效的上下文管理对于Agent能够连贯地完成复杂任务至关重要。

让我们看一个上下文管理器的例子：

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Message:
    role: str  # "user", "assistant", "system", "tool"
    content: Optional[str] = None
    tool_calls: Optional[List[Dict[str, Any]]] = None
    tool_call_id: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class ToolCallRecord:
    tool_name: str
    params: Dict[str, Any]
    result: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)

class ContextManager:
    """上下文管理器"""
    
    def __init__(self, max_messages: int = 50, max_tool_records: int = 100):
        self.messages: List[Message] = []
        self.tool_records: List[ToolCallRecord] = []
        self.task_state: Dict[str, Any] = {}
        self.max_messages = max_messages
        self.max_tool_records = max_tool_records
    
    def add_message(self, message: Message) -> None:
        """添加消息"""
        self.messages.append(message)
        # 如果超过限制，删除最早的消息
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]
    
    def add_tool_record(self, record: ToolCallRecord) -> None:
        """添加工具调用记录"""
        self.tool_records.append(record)
        # 如果超过限制，删除最早的记录
        if len(self.tool_records) > self.max_tool_records:
            self.tool_records = self.tool_records[-self.max_tool_records:]
    
    def update_task_state(self, key: str, value: Any) -> None:
        """更新任务状态"""
        self.task_state[key] = value
    
    def get_task_state(self, key: str, default: Any = None) -> Any:
        """获取任务状态"""
        return self.task_state.get(key, default)
    
    def get_recent_messages(self, n: int = 10) -> List[Message]:
        """获取最近的n条消息"""
        return self.messages[-n:]
    
    def get_tool_history(self, tool_name: Optional[str] = None) -> List[ToolCallRecord]:
        """获取工具调用历史，可按工具名称过滤"""
        if tool_name:
            return [record for record in self.tool_records if record.tool_name == tool_name]
        return self.tool_records
    
    def to_dict(self) -> Dict[str, Any]:
        """将上下文转换为字典"""
        return {
            "messages": [
                {
                    "role": msg.role,
                    "content": msg.content,
                    "tool_calls": msg.tool_calls,
                    "tool_call_id": msg.tool_call_id,
                    "timestamp": msg.timestamp.isoformat()
                }
                for msg in self.messages
            ],
            "tool_records": [
                {
                    "tool_name": record.tool_name,
                    "params": record.params,
                    "result": record.result,
                    "timestamp": record.timestamp.isoformat()
                }
                for record in self.tool_records
            ],
            "task_state": self.task_state
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "ContextManager":
        """从字典创建上下文管理器"""
        manager = cls()
        
        # 恢复消息
        for msg_data in data.get("messages", []):
            msg = Message(
                role=msg_data["role"],
                content=msg_data.get("content"),
                tool_calls=msg_data.get("tool_calls"),
                tool_call_id=msg_data.get("tool_call_id"),
                timestamp=datetime.fromisoformat(msg_data["timestamp"])
            )
            manager.messages.append(msg)
        
        # 恢复工具记录
        for record_data in data.get("tool_records", []):
            record = ToolCallRecord(
                tool_name=record_data["tool_name"],
                params=record_data["params"],
                result=record_data["result"],
                timestamp=datetime.fromisoformat(record_data["timestamp"])
            )
            manager.tool_records.append(record)
        
        # 恢复任务状态
        manager.task_state = data.get("task_state", {})
        
        return manager
    
    def save_to_file(self, filepath: str) -> None:
        """保存上下文到文件"""
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)
    
    @classmethod
    def load_from_file(cls, filepath: str) -> "ContextManager":
        """从文件加载上下文"""
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        return cls.from_dict(data)

这个例子展示了一个相对完整的上下文管理器，它可以管理对话历史、工具调用记录和任务状态。

4.2.4 工具结果的处理与总结

工具返回的原始结果可能包含大量信息，Agent需要能够有效地处理和总结这些结果，以便：

1. 提取关键信息
2. 过滤无关内容
3. 将结果转换为自然语言
4. 决定下一步行动

让我们看一个工具结果处理器的例子：

from typing import Dict, Any, List, Optional
from dataclasses import dataclass

@dataclass
class ProcessedResult:
    """处理后的结果"""
    success: bool
    summary: str  # 自然语言摘要
    key_points: List[str]  # 关键点
    raw_data: Optional[Dict[str, Any]] = None  # 原始数据（可选）
    error: Optional[str] = None  # 错误信息（如果有）
    suggestions: Optional[List[str]] = None  # 后续行动建议

class ResultProcessor:
    """工具结果处理器"""
    
    def __init__(self, llm_client: Any):
        self.llm_client = llm_client
    
    def process(self, tool_name: str, raw_result: Dict[str, Any], context: Optional[Dict[str, Any]] = None) -> ProcessedResult:
        """处理工具结果"""
        if not raw_result.get("success", False):
            # 处理错误情况
            return ProcessedResult(
                success=False,
                summary=f"执行{tool_name}时出错: {raw_result.get('error', '未知错误')}",
                key_points=[],
                error=raw_result.get("error"),
                suggestions=self._generate_error_suggestions(tool_name, raw_result)
            )
        
        # 成功情况，使用LLM处理结果
        data = raw_result.get("data", {})
        summary, key_points = self._summarize_with_llm(tool_name, data, context)
        suggestions = self._generate_suggestions(tool_name, data, context)
        
        return ProcessedResult(
            success=True,
            summary=summary,
            key_points=key_points,
            raw_data=data,
            suggestions=suggestions
        )
    
    def _summarize_with_llm(self, tool_name: str, data