万字长文实战教程：用Python从零构建一个具备工具调用能力的Agent

引言

在人工智能技术飞速发展的今天，“Agent”（智能体）这个概念正在从学术殿堂走向实际应用。你可能已经听说过GPT-4、Claude等大型语言模型，它们展现了惊人的语言理解和生成能力。但是，这些模型也有明显的局限性：它们无法直接访问实时信息，无法执行特定的计算任务，也无法与外部系统交互。

这就是为什么我们需要为AI Agent赋予"工具调用能力"。想象一下，如果AI能够像人类一样使用计算器、搜索网络、查询数据库，甚至控制智能家居设备，那将会带来多么强大的应用场景！

在这篇万字长文中，我将带领你从零开始，用Python构建一个具备工具调用能力的智能Agent。我们不仅会探讨理论概念，还会编写完整的可运行代码，让你真正掌握这项前沿技术。

无论你是AI初学者还是有经验的开发者，这篇教程都将为你提供宝贵的实践经验。让我们开始这段激动人心的旅程吧！

1. 核心概念解析

1.1 什么是Agent？

在人工智能领域，Agent（智能体）是指能够感知环境、做出决策并采取行动的实体。这个概念最早可以追溯到20世纪50年代的AI研究，但近年来随着大语言模型的兴起，Agent的概念又获得了新的内涵。

现代AI Agent通常具备以下特征：

1. 感知能力：能够接收和理解外部信息
2. 推理能力：能够基于信息进行思考和决策
3. 行动能力：能够执行特定的任务或操作
4. 学习能力：能够从经验中改进自己的行为

1.2 工具调用的重要性

为什么Agent需要工具调用能力？让我们通过一个简单的例子来理解：

假设用户问：“2023年中国的GDP增长率是多少？”

一个没有工具调用能力的AI模型可能会：

• 基于训练数据给出一个估计值（但数据可能过时）
• 坦诚自己不知道最新数据

而一个具备工具调用能力的Agent则会：

1. 识别到需要获取最新经济数据
2. 选择合适的工具（如网络搜索）
3. 执行搜索获取最新信息
4. 基于搜索结果给出准确回答

这就是工具调用的价值所在：它让AI Agent能够突破自身知识的局限，与真实世界进行交互。

1.3 Agent的核心架构

一个典型的具备工具调用能力的Agent通常包含以下核心组件：

1. 感知模块：负责接收和理解用户输入
2. 推理引擎：负责分析问题、制定计划
3. 工具注册表：管理可用工具的信息
4. 工具执行器：负责调用和执行具体工具
5. 记忆模块：存储对话历史和重要信息
6. 响应生成器：基于工具返回结果生成最终回答

让我们用一个简单的示意图来表示这些组件的关系：

2. 问题背景与挑战

2.1 传统AI的局限性

在深入探讨如何构建工具调用Agent之前，让我们先了解一下传统AI系统面临的挑战：

1. 知识截止日期：大型语言模型的训练数据有固定的截止日期，无法获取截止日期之后的信息。
2. 计算能力限制：虽然AI模型可以进行简单计算，但在复杂数学运算、精确统计分析方面仍有不足。
3. 环境交互缺失：传统AI无法直接与外部环境交互，如控制设备、修改文件等。
4. 专业领域知识：对于高度专业化的领域，通用AI模型可能缺乏必要的专业知识。

2.2 工具调用Agent的发展历程

工具调用Agent的发展并不是一蹴而就的，它经历了几个重要阶段：

时间阶段	主要特点	代表性工作
早期阶段 (2010年前)	基于规则的系统，工具调用能力有限	Siri (早期版本), 专家系统
中期阶段 (2010-2020)	结合机器学习，有限的自主工具选择	Google Assistant, Alexa
近期阶段 (2020至今)	大语言模型驱动，高度自主的工具调用	GPT-4 with Tools, Claude 2.1

2.3 技术挑战

构建一个有效的工具调用Agent面临多个技术挑战：

1. 工具选择：如何根据用户需求选择最合适的工具？
2. 参数提取：如何从用户输入中提取工具所需的参数？
3. 错误处理：当工具调用失败时，如何优雅地处理并尝试替代方案？
4. 多步推理：有些任务需要多个工具按顺序调用，如何规划这一流程？
5. 结果整合：如何将多个工具的返回结果整合成一个连贯的回答？

3. 系统设计与架构

3.1 整体架构设计

在开始编码之前，让我们先设计系统的整体架构。我们将采用模块化设计，确保每个组件都有明确的职责，同时又能高效协作。

3.2 核心模块详解

3.2.1 记忆模块

记忆模块是Agent的"大脑"，负责存储和检索重要信息。我们将实现三种类型的记忆：

1. 短期记忆：存储当前对话的上下文信息
2. 长期记忆：存储用户的偏好、历史交互等长期信息
3. 工作记忆：存储任务执行过程中的中间结果

class MemoryModule:
    def __init__(self):
        self.short_term_memory = []
        self.long_term_memory = {}
        self.working_memory = {}
    
    def add_to_short_term(self, message):
        """添加信息到短期记忆"""
        self.short_term_memory.append(message)
        # 限制短期记忆的大小，避免无限增长
        if len(self.short_term_memory) > 10:
            self.short_term_memory.pop(0)
    
    def add_to_long_term(self, key, value):
        """添加信息到长期记忆"""
        self.long_term_memory[key] = value
    
    def get_from_long_term(self, key):
        """从长期记忆中获取信息"""
        return self.long_term_memory.get(key)
    
    def set_working_memory(self, key, value):
        """设置工作记忆"""
        self.working_memory[key] = value
    
    def get_working_memory(self, key):
        """获取工作记忆"""
        return self.working_memory.get(key)
    
    def clear_working_memory(self):
        """清空工作记忆"""
        self.working_memory = {}

3.2.2 工具注册表

工具注册表负责管理所有可用的工具，包括工具的描述、参数要求等元数据。

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, name, func, description, parameters):
        """注册一个新工具"""
        self.tools[name] = {
            'function': func,
            'description': description,
            'parameters': parameters
        }
    
    def get_tool(self, name):
        """获取工具信息"""
        return self.tools.get(name)
    
    def list_tools(self):
        """列出所有可用工具"""
        return {name: {
            'description': tool['description'],
            'parameters': tool['parameters']
        } for name, tool in self.tools.items()}

3.2.3 推理引擎

推理引擎是Agent的"决策中心"，负责分析用户需求、选择工具并规划执行流程。

class ReasoningEngine:
    def __init__(self, memory, tool_registry):
        self.memory = memory
        self.tool_registry = tool_registry
    
    def analyze_query(self, query):
        """分析用户查询"""
        # 这里我们会使用自然语言处理技术来分析查询
        # 为了简化演示，我们使用一个基于规则的方法
        analysis = {
            'needs_tool': False,
            'suggested_tools': [],
            'parameters': {}
        }
        
        # 检查是否需要数学计算
        if any(keyword in query.lower() for keyword in ['计算', '算一下', 'math', 'calculate']):
            analysis['needs_tool'] = True
            analysis['suggested_tools'].append('calculator')
        
        # 检查是否需要网络搜索
        if any(keyword in query.lower() for keyword in ['搜索', '查找', '最新', 'search', 'find', 'latest']):
            analysis['needs_tool'] = True
            analysis['suggested_tools'].append('web_search')
        
        return analysis
    
    def select_tool(self, analysis):
        """选择合适的工具"""
        if not analysis['needs_tool']:
            return None
        
        # 简单选择第一个建议的工具
        # 在实际应用中，这里可以有更复杂的选择逻辑
        if analysis['suggested_tools']:
            return self.tool_registry.get_tool(analysis['suggested_tools'][0])
        
        return None
    
    def extract_parameters(self, query, tool):
        """从查询中提取工具所需参数"""
        # 这里同样简化处理
        # 在实际应用中，可以使用命名实体识别(NER)等技术
        parameters = {}
        return parameters

3.3 数学模型：工具选择决策

在工具选择过程中，我们可以使用一个简单的决策模型来评估不同工具的适用性。假设我们有一组工具 T={t1,t2,...,tn}T = \{t_1, t_2, ..., t_n\}T={t1​,t2​,...,tn​}，每个工具都有一组特征 Ft={ft1,ft2,...,ftm}F_t = \{f_{t1}, f_{t2}, ..., f_{tm}\}Ft​={ft1​,ft2​,...,ftm​}，我们可以使用以下公式来计算工具 $t$ 对查询 $q$ 的适用性分数：

$$S(t,q)=\sum _{i=1}^m{w}_i\cdot f_{ti}(q)$$

其中 $w_i$ 是特征 $i$ 的权重，$f_{ti}(q)$ 是工具 $t$ 的特征 $i$ 对查询 $q$ 的匹配程度。

然后我们选择适用性分数最高的工具：

$$t_{best}=\arg \underset{t\in T}{\max }S(t,q)$$

这个模型可以根据实际需求进行扩展和优化。

4. 基础工具实现

4.1 计算器工具

让我们从一个简单但实用的计算器工具开始：

import math

def calculator(expression):
    """
    计算器工具：执行数学计算
    
    参数:
        expression (str): 数学表达式字符串
    
    返回:
        float: 计算结果
    """
    try:
        # 安全计算表达式
        # 注意：在实际应用中，应该使用更安全的表达式求值方法
        allowed_names = {
            'abs': abs,
            'round': round,
            'min': min,
            'max': max,
            'sum': sum,
            'pow': pow,
            'sqrt': math.sqrt,
            'sin': math.sin,
            'cos': math.cos,
            'tan': math.tan,
            'log': math.log,
            'pi': math.pi,
            'e': math.e
        }
        
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return {
            'success': True,
            'result': result,
            'expression': expression
        }
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
            'expression': expression
        }

4.2 网络搜索工具

接下来，让我们实现一个网络搜索工具。为了演示方便，我们将使用一个模拟的搜索功能：

import requests
from bs4 import BeautifulSoup
import random

def web_search(query, num_results=5):
    """
    网络搜索工具：模拟网络搜索功能
    
    参数:
        query (str): 搜索查询
        num_results (int): 返回结果数量
    
    返回:
        dict: 搜索结果
    """
    # 在实际应用中，你可以使用Google Search API, Bing Search API等
    # 这里我们使用一个模拟的搜索结果
    
    # 模拟搜索结果数据库
    mock_results = {
        "人工智能": [
            {"title": "人工智能的未来发展", "snippet": "人工智能是当前科技领域最热门的话题之一..."},
            {"title": "机器学习入门指南", "snippet": "机器学习是人工智能的一个重要分支..."},
            {"title": "深度学习在图像识别中的应用", "snippet": "深度学习技术在图像识别领域取得了重大突破..."}
        ],
        "Python编程": [
            {"title": "Python入门教程", "snippet": "Python是一种简单易学的编程语言..."},
            {"title": "Python数据分析实战", "snippet": "Python在数据分析领域有广泛的应用..."},
            {"title": "Django Web开发指南", "snippet": "Django是Python最流行的Web框架之一..."}
        ],
        "机器学习": [
            {"title": "监督学习算法详解", "snippet": "监督学习是机器学习的主要类型之一..."},
            {"title": "无监督学习的实际应用", "snippet": "无监督学习在聚类和降维等任务中有重要应用..."}
        ]
    }
    
    # 查找匹配的结果
    results = []
    for key, value in mock_results.items():
        if key in query:
            results.extend(value)
    
    # 如果没有找到匹配结果，返回通用结果
    if not results:
        results = [
            {"title": f"关于'{query}'的搜索结果", "snippet": f"这是关于'{query}'的一些基本信息..."},
            {"title": f"{query}的相关话题", "snippet": f"与'{query}'相关的话题还有很多..."}
        ]
    
    # 随机打乱并限制结果数量
    random.shuffle(results)
    results = results[:num_results]
    
    return {
        'success': True,
        'query': query,
        'results': results,
        'total_results': len(results)
    }

4.3 天气查询工具

让我们再添加一个天气查询工具：

import datetime
import random

def weather_forecast(location, days=1):
    """
    天气预报工具：查询指定位置的天气
    
    参数:
        location (str): 位置名称
        days (int): 查询天数
    
    返回:
        dict: 天气预报结果
    """
    # 天气状况选项
    weather_conditions = ['晴朗', '多云', '阴天', '小雨', '中雨', '大雨', '雷阵雨', '小雪', '中雪']
    
    # 生成天气预报
    forecast = []
    base_temp = random.randint(10, 30)  # 基础温度
    
    for i in range(days):
        date = (datetime.datetime.now() + datetime.timedelta(days=i)).strftime("%Y-%m-%d")
        condition = random.choice(weather_conditions)
        temp_high = base_temp + random.randint(0, 10)
        temp_low = base_temp - random.randint(0, 10)
        humidity = random.randint(30, 90)
        wind_speed = random.randint(0, 30)
        
        forecast.append({
            'date': date,
            'condition': condition,
            'temperature_high': temp_high,
            'temperature_low': temp_low,
            'humidity': humidity,
            'wind_speed': wind_speed
        })
    
    return {
        'success': True,
        'location': location,
        'forecast': forecast,
        'query_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }

5. Agent核心实现

5.1 工具执行器

工具执行器负责实际调用工具并处理返回结果：

class ToolExecutor:
    def __init__(self, tool_registry):
        self.tool_registry = tool_registry
    
    def execute_tool(self, tool_name, parameters):
        """执行指定工具"""
        tool = self.tool_registry.get_tool(tool_name)
        
        if not tool:
            return {
                'success': False,
                'error': f'工具 "{tool_name}" 未找到'
            }
        
        try:
            # 执行工具函数
            result = tool['function'](**parameters)
            return result
        except Exception as e:
            return {
                'success': False,
                'error': f'执行工具时出错: {str(e)}'
            }

5.2 响应生成器

响应生成器负责将工具执行结果转化为自然语言回答：

class ResponseGenerator:
    def __init__(self, memory):
        self.memory = memory
    
    def generate_response(self, query, tool_result=None):
        """生成自然语言响应"""
        if not tool_result:
            return f"我理解你的问题是关于：'{query}'。不过目前我无法直接回答这个问题，让我尝试使用其他方法帮你解决。"
        
        if not tool_result.get('success', False):
            return f"抱歉，在处理你的问题时遇到了错误：{tool_result.get('error', '未知错误')}"
        
        # 根据不同的工具结果生成不同的响应
        if 'calculator' in str(tool_result):
            return self._generate_calculator_response(tool_result)
        elif 'web_search' in str(tool_result):
            return self._generate_search_response(tool_result)
        elif 'weather_forecast' in str(tool_result):
            return self._generate_weather_response(tool_result)
        else:
            return f"已成功处理你的请求，结果如下：{tool_result.get('result', '无详细结果')}"
    
    def _generate_calculator_response(self, result):
        """生成计算器响应"""
        expression = result.get('expression', '')
        calc_result = result.get('result', '')
        return f"计算结果：{expression} = {calc_result}"
    
    def _generate_search_response(self, result):
        """生成搜索响应"""
        query = result.get('query', '')
        search_results = result.get('results', [])
        
        response = f"关于'{query}'的搜索结果：\n\n"
        
        for i, item in enumerate(search_results, 1):
            response += f"{i}. {item['title']}\n"
            response += f"   {item['snippet']}\n\n"
        
        return response
    
    def _generate_weather_response(self, result):
        """生成天气响应"""
        location = result.get('location', '')
        forecast = result.get('forecast', [])
        query_time = result.get('query_time', '')
        
        response = f"{location}的天气预报（查询时间：{query_time}）：\n\n"
        
        for item in forecast:
            response += f"{item['date']}：{item['condition']}\n"
            response += f"   温度：{item['temperature_low']}°C ~ {item['temperature_high']}°C\n"
            response += f"   湿度：{item['humidity']}%\n"
            response += f"   风速：{item['wind_speed']} km/h\n\n"
        
        return response

5.3 主Agent类

现在，让我们将所有组件整合到一个主Agent类中：

class ToolCallingAgent:
    def __init__(self):
        # 初始化组件
        self.memory = MemoryModule()
        self.tool_registry = ToolRegistry()
        self.tool_executor = ToolExecutor(self.tool_registry)
        self.reasoning_engine = ReasoningEngine(self.memory, self.tool_registry)
        self.response_generator = ResponseGenerator(self.memory)
        
        # 注册默认工具
        self._register_default_tools()
    
    def _register_default_tools(self):
        """注册默认工具"""
        self.tool_registry.register_tool(
            name='calculator',
            func=calculator,
            description='执行数学计算，支持基本运算和常见数学函数',
            parameters=[
                {
                    'name': 'expression',
                    'type': 'string',
                    'description': '数学表达式字符串，如"2+2"或"sin(pi/2)"',
                    'required': True
                }
            ]
        )
        
        self.tool_registry.register_tool(
            name='web_search',
            func=web_search,
            description='搜索网络信息',
            parameters=[
                {
                    'name': 'query',
                    'type': 'string',
                    'description': '搜索查询字符串',
                    'required': True
                },
                {
                    'name': 'num_results',
                    'type': 'integer',
                    'description': '返回结果数量，默认为5',
                    'required': False,
                    'default': 5
                }
            ]
        )
        
        self.tool_registry.register_tool(
            name='weather_forecast',
            func=weather_forecast,
            description='查询指定位置的天气预报',
            parameters=[
                {
                    'name': 'location',
                    'type': 'string',
                    'description': '位置名称，如"北京"、"上海"',
                    'required': True
                },
                {
                    'name': 'days',
                    'type': 'integer',
                    'description': '查询天数，默认为1',
                    'required': False,
                    'default': 1
                }
            ]
        )
    
    def process_query(self, query):
        """处理用户查询"""
        # 将用户查询添加到记忆
        self.memory.add_to_short_term({'role': 'user', 'content': query})
        
        # 分析查询
        analysis = self.reasoning_engine.analyze_query(query)
        
        # 选择工具
        tool = self.reasoning_engine.select_tool(analysis)
        
        if tool:
            # 提取参数（简化实现）
            parameters = self._extract_parameters(query, tool)
            
            # 执行工具
            tool_name = [name for name, t in self.tool_registry.tools.items() if t == tool][0]
            tool_result = self.tool_executor.execute_tool(tool_name, parameters)
            
            # 生成响应
            response = self.response_generator.generate_response(query, tool_result)
        else:
            # 没有工具可用，生成默认响应
            response = self.response_generator.generate_response(query)
        
        # 将响应添加到记忆
        self.memory.add_to_short_term({'role': 'assistant', 'content': response})
        
        return response
    
    def _extract_parameters(self, query, tool):
        """从查询中提取参数（简化实现）"""
        parameters = {}
        
        # 这里是一个简化的参数提取逻辑
        # 在实际应用中，可以使用更复杂的NLP技术
        tool_name = [name for name, t in self.tool_registry.tools.items() if t == tool][0]
        
        if tool_name == 'calculator':
            # 提取数学表达式
            # 这是一个简化实现，实际应用中需要更复杂的逻辑
            parameters['expression'] = query
            
        elif tool_name == 'web_search':
            # 提取搜索查询
            # 移除可能的搜索关键词
            search_keywords = ['搜索', '查找', 'search', 'find', '请搜索', '帮我搜索']
            clean_query = query
            for keyword in search_keywords:
                clean_query = clean_query.replace(keyword, '')
            parameters['query'] = clean_query.strip()
            
        elif tool_name == 'weather_forecast':
            # 提取位置
            # 这是一个简化实现
            locations = ['北京', '上海', '广州', '深圳', '杭州', '南京', '成都', '武汉', '西安']
            found_location = None
            for location in locations:
                if location in query:
                    found_location = location
                    break
            
            if found_location:
                parameters['location'] = found_location
            else:
                parameters['location'] = '北京'  # 默认位置
        
        return parameters

6. 实战应用

6.1 基本使用示例

让我们通过一些示例来看看我们的Agent如何工作：

# 创建Agent实例
agent = ToolCallingAgent()

# 示例1: 数学计算
print("示例1: 数学计算")
query1 = "计算 25 * (4 + 2) / 3"
response1 = agent.process_query(query1)
print(f"用户: {query1}")
print(f"Agent: {response1}\n")

# 示例2: 网络搜索
print("示例2: 网络搜索")
query2 = "搜索人工智能"
response2 = agent.process_query(query2)
print(f"用户: {query2}")
print(f"Agent: {response2}\n")

# 示例3: 天气预报
print("示例3: 天气预报")
query3 = "北京的天气怎么样"
response3 = agent.process_query(query3)
print(f"用户: {query3}")
print(f"Agent: {response3}\n")

6.2 进阶应用：多步骤任务处理

现在，让我们扩展我们的Agent，使其能够处理需要多个工具调用的复杂任务：

class AdvancedToolCallingAgent(ToolCallingAgent):
    def __init__(self):
        super().__init__()
        self.max_iterations = 5  # 最大迭代次数，防止无限循环
    
    def process_complex_query(self, query):
        """处理复杂查询，可能需要多个工具调用"""
        # 将用户查询添加到记忆
        self.memory.add_to_short_term({'role': 'user', 'content': query})
        
        # 初始化工作记忆
        self.memory.set_working_memory('original_query', query)
        self.memory.set_working_memory('collected_information', [])
        
        # 多步处理循环
        iterations = 0
        final_response = ""
        
        while iterations < self.max_iterations:
            iterations += 1
            
            # 分析当前状态
            current_query = self._formulate_current_query(iterations)
            analysis = self.reasoning_engine.analyze_query(current_query)
            
            # 决定是否需要继续调用工具
            if not analysis['needs_tool'] or self._task_is_complete():
                final_response = self._synthesize_final_response()
                break
            
            # 选择和执行工具
            tool = self.reasoning_engine.select_tool(analysis)
            if tool:
                tool_name = [name for name, t in self.tool_registry.tools.items() if t == tool][0]
                parameters = self._extract_parameters(current_query, tool)
                tool_result = self.tool_executor.execute_tool(tool_name, parameters)
                
                # 保存工具结果到工作记忆
                collected_info = self.memory.get_working_memory('collected_information')
                collected_info.append({
                    'tool': tool_name,
                    'query': current_query,
                    'result': tool_result
                })
                self.memory.set_working_memory('collected_information', collected_info)
        
        # 如果达到最大迭代次数，生成一个总结性响应
        if iterations >= self.max_iterations and not final_response:
            final_response = self._synthesize_partial_response()
        
        # 将响应添加到记忆
        self.memory.add_to_short_term({'role': 'assistant', 'content': final_response})
        
        return final_response
    
    def _formulate_current_query(self, iteration):
        """根据当前迭代次数和已有信息制定当前查询"""
        original_query = self.memory.get_working_memory('original_query')
        collected_info = self.memory.get_working_memory('collected_information')
        
        # 简化实现：第一轮使用原始查询，后续轮次基于已有信息
        if iteration == 1:
            return original_query
        else:
            # 这里可以实现更复杂的查询生成逻辑
            # 例如，基于已收集的信息确定下一步需要什么
            return f"基于已收集的信息，继续处理原始查询: {original_query}"
    
    def _task_is_complete(self):
        """判断任务是否完成"""
        # 简化实现：在实际应用中，这里可以有更复杂的判断逻辑
        collected_info = self.memory.get_working_memory('collected_information')
        return len(collected_info) >= 2  # 假设收集2个工具结果就足够了
    
    def _synthesize_final_response(self):
        """合成最终响应"""
        original_query = self.memory.get_working_memory('original_query')
        collected_info = self.memory.get_working_memory('collected_information')
        
        response = f"针对你的问题：'{original_query}'，我已收集到以下信息：\n\n"
        
        for i, info in enumerate(collected_info, 1):
            response += f"信息源 {i} ({info['tool']})：\n"
            if info['result'].get('success', False):
                # 根据工具类型格式化结果
                if info['tool'] == 'calculator':
                    response += f"  计算结果: {info['result'].get('expression', '')} = {info['result'].get('result', '')}\n"
                elif info['tool'] == 'web_search':
                    response += f"  搜索了: {info['result'].get('query', '')}\n"
                    results = info['result'].get('results', [])
                    if results:
                        response += f"  找到 {len(results)} 个相关结果\n"
                elif info['tool'] == 'weather_forecast':
                    response += f"  查询了: {info['result'].get('location', '')} 的天气\n"
            else:
                response += f"  获取信息时出错: {info['result'].get('error', '未知错误')}\n"
            response += "\n"
        
        response += "基于以上信息，我已经完成了对你问题的分析。"
        return response
    
    def _synthesize_partial_response(self):
        """合成部分响应（当达到最大迭代次数时）"""
        original_query = self.memory.get_working_memory('original_query')
        collected_info = self.memory.get_working_memory('collected_information')
        
        response = f"针对你的问题：'{original_query}'，我已经尝试收集信息，但还需要更多步骤才能完成分析。\n\n"
        response += f"目前已收集到 {len(collected_info)} 条信息：\n"
        
        for i, info in enumerate(collected_info, 1):
            response += f"{i}. 使用了 {info['tool']} 工具\n"
        
        return response

6.3 实际应用场景：旅行助手

让我们构建一个专门的旅行助手Agent，它可以帮助用户规划旅行：

class TravelAssistantAgent(AdvancedToolCallingAgent):
    def __init__(self):
        super().__init__()
        # 注册旅行相关的额外工具
        self._register_travel_tools()
    
    def _register_travel_tools(self):
        """注册旅行相关工具"""
        # 这里我们可以添加航班查询、酒店预订等工具
        # 为了演示，我们添加一个简单的旅行推荐工具
        
        def travel_recommendation(destination, interest=None):
            """
            旅行推荐工具：提供目的地的旅行建议
            
            参数:
                destination (str): 目的地名称
                interest (str): 兴趣点，如"历史"、"自然"、"美食"等
            
            返回:
                dict: 旅行推荐结果
            """
            recommendations = {
                '北京': {
                    'default': ['故宫博物院', '天安门广场', '长城'],
                    '历史': ['故宫博物院', '天坛', '颐和园'],
                    '自然': ['香山公园', '奥林匹克森林公园', '玉渊潭公园'],
                    '美食': ['王府井小吃街', '簋街', '护国寺小吃']
                },
                '上海': {
                    'default': ['外滩', '东方明珠', '豫园'],
                    '历史': ['豫园', '上海博物馆', '中共一大会址'],
                    '自然': ['外滩', '世纪公园', '上海植物园'],
                    '美食': ['南京东路美食街', '豫园小吃', '田子坊美食']
                },
                '杭州': {
                    'default': ['西湖', '灵隐寺', '雷峰塔'],
                    '历史': ['灵隐寺', '岳王庙', '宋城'],
                    '自然': ['西湖', '千岛湖', '西溪湿地'],
                    '美食': ['河坊街', '知味观', '楼外楼']
                }
            }
            
            dest_recommendations = recommendations.get(destination, {'default': [f'{destination}的主要景点']})
            selected_recommendations = dest_recommendations.get(interest, dest_recommendations['default'])
            
            return {
                'success': True,
                'destination': destination,
                'interest': interest,
                'recommendations': selected_recommendations
            }
        
        self.tool_registry.register_tool(
            name='travel_recommendation',
            func=travel_recommendation,
            description='提供目的地的旅行建议和景点推荐',
            parameters=[
                {
                    'name': 'destination',
                    'type': 'string',
                    'description': '目的地名称，如"北京"、"上海"',
                    'required': True
                },
                {
                    'name': 'interest',
                    'type': 'string',
                    'description': '兴趣点，如"历史"、"自然"、"美食"等',
                    'required': False
                }
            ]
        )
    
    def plan_trip(self, destination, duration=3, interests=None):
        """
        规划旅行行程
        
        参数:
            destination (str): 目的地
            duration (int): 旅行天数
            interests (list): 兴趣列表
            
        返回:
            str: 旅行计划
        """
        # 构建查询
        query = f"帮我规划一个{duration}天的{destination}旅行"
        if interests:
            query += f"，我对{', '.join(interests)}特别感兴趣"
        
        # 使用高级处理功能
        return self.process_complex_query(query)

7. 性能优化与扩展

7.1 优化推理引擎

现在的推理引擎比较简单，让我们优化一下，使其更智能地选择工具：

class OptimizedReasoningEngine(ReasoningEngine):
    def __init__(self, memory, tool_registry):
        super().__init__(memory, tool_registry)
        self.tool_selection_history = {}  # 记录工具选择历史
    
    def analyze_query(self, query):
        """更智能地分析用户查询"""
        analysis = super().analyze_query(query)
        
        # 扩展分析，考虑对话历史
        short_term_memory = self.memory.short_term_memory
        
        # 检查是否有历史对话可以提供上下文
        if len(short_term_memory) > 1:
            # 这里可以实现更复杂的上下文分析
            pass
        
        return analysis
    
    def select_tool(self, analysis, context=None):
        """
        基于分析结果和上下文选择工具
        
        参数:
            analysis: 查询分析结果
            context: 额外的上下文信息
        
        返回:
            选择的工具
        """
        if not analysis['needs_tool']:
            return None
        
        tools = self.tool_registry.list_tools()
        tool_scores = {}
        
        # 为每个工具评分
        for tool_name, tool_info in tools.items():
            score = 0
            
            # 基础分数：检查工具是否在建议列表中
            if tool_name in analysis['suggested_tools']:
                score += 10
            
            # 描述匹配度
            description = tool_info['description']
            for keyword in analysis.get('keywords', []):
                if keyword in description:
                    score += 5
            
            # 历史选择记录（如果有的话）
            if tool_name in self.tool_selection_history:
                score += min(self.tool_selection_history[tool_name], 5)  # 最多加5分
            
            tool_scores[tool_name] = score
        
        # 选择分数最高的工具
        if tool_scores:
            best_tool_name = max(tool_scores.items(), key=lambda x: x[1])[0]
            
            # 更新选择历史
            if best_tool_name in self.tool_selection_history:
                self.tool_selection_history[best_tool_name] += 1
            else:
                self.tool_selection_history[best_tool_name] = 1
            
            return self.tool_registry.get_tool(best_tool_name)
        
        return None

7.2 添加异步支持

为了提高性能，特别是在处理多个工具调用时，让我们添加异步支持：

import asyncio
import concurrent.futures

class AsyncToolExecutor(ToolExecutor):
    def __init__(self, tool_registry):
        super().__init__(tool_registry)
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
    
    async def execute_tool_async(self, tool_name, parameters):
        """异步执行工具"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.executor, 
            self.execute_tool, 
            tool_name, 
            parameters
        )
    
    async def execute_multiple_tools(self, tool_calls):
        """
        并行执行多个工具调用
        
        参数:
            tool_calls: 工具调用列表，格式为[(tool_name, parameters), ...]
        
        返回:
            工具执行结果列表
        """
        tasks = []
        for tool_name, parameters in tool_calls:
            task = self.execute_tool_async(tool_name, parameters)
            tasks.append(task)
        
        return await asyncio.gather(*tasks)

7.3 错误恢复与重试机制

为了使我们的Agent更加健壮，让我们添加错误恢复和重试机制：

class RobustToolExecutor(AsyncToolExecutor):
    def __init__(self, tool_registry, max_retries=3, retry_delay=1):
        super().__init__(tool_registry)
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    
    async def execute_tool_with_retry(self, tool_name, parameters):
        """带重试机制的工具执行"""
        retries = 0
        last_error = None
        
        while retries < self.max_retries:
            try:
                result = await self.execute_tool_async(tool_name, parameters)
                
                if result.get('success', False):
                    return result
                else:
                    # 检查错误类型，决定是否重试
                    error_msg = result.get('error', '')
                    if self._is_retriable_error(error_msg):
                        last_error = error_msg
                        retries += 1
                        await asyncio.sleep(self.retry_delay * retries)  # 指数退避
                    else:
                        return result  # 不可重试的错误，直接返回
            except Exception as e:
                last_error = str(e)
                retries += 1
                await asyncio.sleep(self.retry_delay * retries)
        
        # 达到最大重试次数，返回失败结果
        return {
            'success': False,
            'error': f'工具执行失败，已重试{self.max_retries}次。最后错误: {last_error}',
            'retries_exhausted': True
        }
    
    def _is_retriable_error(self, error_msg):
        """判断错误是否可重试"""
        retriable_keywords = ['timeout', 'connection', 'temporarily', 'busy', 'rate limit']
        return any(keyword in error_msg.lower() for keyword in retriable_keywords)

8. 实际项目案例：智能客服系统

8.1 项目概述

让我们将我们的Agent技术应用到一个实际项目中：智能客服系统。这个系统可以：

1. 回答常见问题
2. 查询订单状态
3. 提供产品信息
4. 转接人工客服

8.2 系统设计

8.3 核心实现

class CustomerServiceAgent(ToolCallingAgent):
    def __init__(self):
        super().__init__()
        self._register_customer_service_tools()
        self.context = {}  # 存储对话上下文
    
    def _register_customer_service_tools(self):
        """注册客服相关工具"""
        # 订单查询工具
        def order_query(order_id):
            """
            订单查询工具
            
            参数:
                order_id (str): 订单号
            
            返回:
                dict: 订单信息
            """
            # 模拟订单数据库
            mock_orders = {
                'ORD123456': {
                    'order_id': 'ORD123456',
                    'product': '无线耳机',
                    'quantity': 1,
                    'price': 299,
                    'status': '已发货',
                    'tracking_number': 'SF1234567890',
                    'estimated_delivery': '2023-12-20'
                },
                'ORD789012': {
                    'order_id': 'ORD789012',
                    'product': '智能手表',
                    'quantity': 1,
                    'price': 1299,
                    'status': '待发货',
                    'estimated_delivery': '2023-12-25'
                }
            }
            
            order = mock_orders.get(order_id)
            if order:
                return {
                    'success': True,
                    'order': order
                }
            else:
                return {
                    'success': False,
                    'error': f'未找到订单号 {order_id}'
                }
        
        self.tool_registry.register_tool(
            name='order_query',
            func=order_query,
            description='查询订单信息',
            parameters=[
                {
                    'name': 'order_id',
                    'type': 'string',
                    'description': '订单号',
                    'required': True
                }
            ]
        )
        
        # 常见问题查询工具
        def faq_query(category=None):
            """
            常见问题查询工具
            
            参数:
                category (str): 问题分类，可选
            
            返回:
                dict: 常见问题和答案
            """
            faq_database