2.5 工具和智能体

文档摘要

2.5 工具和智能体 — LangChain框架精通核心组件本节导读：深入学习LangChain的工具和智能体系统，掌握工具使用、智能体设计和多智能体协作的实现方法，构建具备自主工作能力的AI系统。学习目标理解智能体和工具的基本概念和架构设计掌握LangChain Agent的使用方法和配置选项熟练使用各种内置工具和自定义工具了解多智能体协作的实现策略实现工具调用和错误处理机制优化智能体的性能和稳定性核心概念工具和智能体是LangChain中最强大的组件之一，它们赋予AI系统使用工具、进行推理和决策的能力，使AI能够完成复杂的任务。智能体的架构原理 !

2.5 工具和智能体 — LangChain框架精通核心组件

本节导读：深入学习LangChain的工具和智能体系统，掌握工具使用、智能体设计和多智能体协作的实现方法，构建具备自主工作能力的AI系统。

学习目标

理解智能体和工具的基本概念和架构设计
掌握LangChain Agent的使用方法和配置选项
熟练使用各种内置工具和自定义工具
了解多智能体协作的实现策略
实现工具调用和错误处理机制
优化智能体的性能和稳定性

核心概念

工具和智能体是LangChain中最强大的组件之一，它们赋予AI系统使用工具、进行推理和决策的能力，使AI能够完成复杂的任务。

智能体的架构原理

![智能体架构示意图：智能体、工具、推理链、决策循环的完整流程]

智能体的核心工作流程包括：

任务理解：理解用户输入的任务需求
工具选择：选择合适的工具来完成任务
工具调用：调用工具并获取结果
决策推理：基于工具结果进行下一步决策
响应生成：生成最终的用户响应

智能体的核心组件

Agent：智能体的核心类，负责决策和工具调用
Tool：工具接口，定义工具的功能和调用方式
LLM：语言模型，提供推理和决策能力
Memory：记忆管理，维护对话历史
Executor：执行器，负责工具调用和结果处理

环境准备 / 前置知识

基础依赖安装


# 工具和智能体基础包
pip install langchain langchain-openai

# 工具相关
pip install langchain-community langchain-core

# 搜索工具
pip install google-search-results serpapi

# 计算工具
pip install numexpr

# 数据处理工具
pip install pandas numpy

# API调用工具
pip install requests httpx

核心导入模块


import os
from typing import List, Dict, Any, Optional
from datetime import datetime

# LangChain智能体相关
from langchain.agents import (
    AgentType,
    Tool,
    AgentExecutor,
    initialize_agent,
    load_tools,
    create_structured_chat_agent,
    create_openai_functions_agent
)
from langchain.tools import tool
from langchain.schema import AgentAction, AgentFinish

# LangChain模型和记忆
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory

# 其他工具
from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.schema import HumanMessage, AIMessage

分步实战

步骤 1：基础智能体配置

首先配置基础的智能体环境：


import os
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.agents import initialize_agent, AgentType
from langchain.tools import tool

# 设置API密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 创建语言模型
llm = ChatOpenAI(
    temperature=0.3,
    model_name="gpt-3.5-turbo"
)

# 创建记忆管理
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建基础计算工具
@tool
def calculate(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2 + 2 * 3"
        
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

# 创建时间工具
@tool
def get_current_time() -> str:
    """获取当前时间
    
    Returns:
        当前时间的字符串表示
    """
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 创建工具列表
tools = [
    Tool(
        name="calculator",
        func=calculate,
        description="用于计算数学表达式，如 '2 + 2 * 3'"
    ),
    Tool(
        name="time",
        func=get_current_time,
        description="用于获取当前时间"
    )
]

# 创建智能体
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    memory=memory,
    verbose=True,
    handle_parsing_errors=True
)

# 使用智能体
try:
    result1 = agent.run("计算 15 + 25 的结果")
    print(f"计算结果: {result1}")
    
    result2 = agent.run("现在几点了？")
    print(f"当前时间: {result2}")
    
    result3 = agent.run("计算 100 * 2 + 50 的结果")
    print(f"计算结果: {result3}")
    
except Exception as e:
    print(f"智能体运行错误: {str(e)}")

步骤 2：使用内置工具

LangChain提供了许多内置工具，可以直接使用：


from langchain.agents import load_tools
from langchain.agents import AgentType

# 加载内置工具
built_in_tools = load_tools([
    'google-serper',  # Google搜索
    'llm-math',      # 数学计算
    'wikipedia',     # Wikipedia查询
    'python-repl'    # Python代码执行
], llm=llm)

# 创建智能体
built_in_agent = initialize_agent(
    tools=built_in_tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True
)

# 使用内置工具
try:
    result1 = built_in_agent.run("搜索最新的AI技术发展")
    print(f"搜索结果: {result1}")
    
    result2 = built_in_agent.run("计算 123 + 456 * 2 的结果")
    print(f"计算结果: {result2}")
    
    result3 = built_in_agent.run("搜索关于机器学习的基本概念")
    print(f"Wikipedia结果: {result3}")
    
except Exception as e:
    print(f"内置工具错误: {str(e)}")

步骤 3：创建自定义工具

创建更复杂的自定义工具：


from langchain.tools import tool
from langchain.schema import Document
import json
import requests

# 创建文件操作工具
@tool
def read_file(filename: str) -> str:
    """读取文件内容
    
    Args:
        filename: 文件路径
        
    Returns:
        文件内容
    """
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        return f"文件 {filename} 不存在"
    except Exception as e:
        return f"读取文件错误: {str(e)}"

@tool
def write_file(filename: str, content: str) -> str:
    """写入文件内容
    
    Args:
        filename: 文件路径
        content: 要写入的内容
        
    Returns:
        操作结果
    """
    try:
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"文件 {filename} 写入成功"
    except Exception as e:
        return f"写入文件错误: {str(e)}"

# 创建API调用工具
@tool
def fetch_weather(city: str) -> str:
    """获取天气信息
    
    Args:
        city: 城市名称
        
    Returns:
        天气信息
    """
    try:
        # 模拟的天气数据
        weather_data = {
            "北京": "晴天，气温25°C",
            "上海": "多云，气温28°C",
            "广州": "雨天，气温30°C",
            "深圳": "晴天，气温29°C"
        }
        
        if city in weather_data:
            return f"{city}的天气：{weather_data[city]}"
        else:
            return f"暂不支持{city}的天气查询"
    except Exception as e:
        return f"获取天气错误: {str(e)}"

# 创建工具列表
custom_tools = [
    Tool(
        name="read_file",
        func=read_file,
        description="读取文件内容，支持文本文件"
    ),
    Tool(
        name="write_file",
        func=write_file,
        description="写入文件内容，支持文本文件"
    ),
    Tool(
        name="weather",
        func=fetch_weather,
        description="获取指定城市的天气信息"
    )
]

# 创建智能体
custom_agent = initialize_agent(
    tools=custom_tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True
)

# 使用自定义工具
try:
    result1 = custom_agent.run("创建一个包含姓名和年龄的JSON数组")
    print(f"创建结果: {result1}")
    
    result2 = custom_agent.run("查询北京的天气情况")
    print(f"天气结果: {result2}")
    
    result3 = custom_agent.run("查询上海的天气情况")
    print(f"天气结果: {result3}")
    
except Exception as e:
    print(f"自定义工具错误: {str(e)}")

步骤 4：使用结构化智能体

使用结构化的智能体，提供更精确的控制：


from langchain.agents import create_structured_chat_agent
from langchain.prompts import ChatPromptTemplate

# 创建结构化提示模板
structured_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手，能够使用各种工具来帮助用户完成任务。
    
可用的工具：
- calculator: 计算数学表达式
- time: 获取当前时间
- weather: 查询天气信息

使用格式：
1. 分析用户需求
2. 选择合适的工具
3. 调用工具并获取结果
4. 生成最终回答

请严格按照这个流程进行。"""),
    ("user", "{input}"),
    ("assistant", "{agent_scratchpad}"),
])

# 创建结构化智能体
structured_agent = create_structured_chat_agent(
    llm=llm,
    tools=tools,
    prompt=structured_prompt,
    verbose=True
)

# 使用结构化智能体
try:
    result1 = structured_agent.invoke({
        "input": "计算 2的10次方是多少？"
    })
    print(f"计算结果: {result1}")
    
    result2 = structured_agent.invoke({
        "input": "现在是什么时间？"
    })
    print(f"时间结果: {result2}")
    
    result3 = structured_agent.invoke({
        "input": "查询上海的天气情况"
    })
    print(f"天气结果: {result3}")
    
except Exception as e:
    print(f"结构化智能体错误: {str(e)}")

步骤 5：实现智能体错误处理和重试机制


import time
from langchain.agents import AgentExecutor
from langchain.agents.format_scratchpad import format_to_openai_function_messages
from langchain.output_parsers import OpenAIFunctionsAgentOutputParser

class SmartAgentWithRetry:
    def __init__(self, tools, llm, max_retries=3):
        self.tools = tools
        self.llm = llm
        self.max_retries = max_retries
        self.agent = self._create_agent()
        
    def _create_agent(self):
        """创建智能体"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个智能助手，能够使用各种工具来帮助用户完成任务。
            
可用的工具：
- calculator: 计算数学表达式
- time: 获取当前时间
- weather: 查询天气信息

使用格式：
1. 分析用户需求
2. 选择合适的工具
3. 调用工具并获取结果
4. 生成最终回答

如果工具调用失败，请重试最多3次。如果仍然失败，请告诉用户无法完成该任务。"""),
            ("user", "{input}"),
            ("assistant", "{agent_scratchpad}"),
        ])
        
        agent = create_structured_chat_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt,
            verbose=True
        )
        
        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            max_iterations=10,
            verbose=True,
            handle_parsing_errors=True,
            return_only_outputs=True
        )
    
    def run_with_retry(self, input_text: str) -> str:
        """带重试机制的智能体运行"""
        for attempt in range(self.max_retries):
            try:
                result = self.agent.invoke({"input": input_text})
                return result.get("output", "没有可用的输出")
            except Exception as e:
                if attempt == self.max_retries - 1:
                    return f"经过{self.max_retries}次重试后仍然失败：{str(e)}"
                print(f"第{attempt + 1}次尝试失败，正在重试...")
                time.sleep(1)
        
        return "智能体运行失败"
    
    def run_with_timeout(self, input_text: str, timeout: int = 30) -> str:
        """带超时机制的智能体运行"""
        import signal
        
        def timeout_handler(signum, frame):
            raise TimeoutError("智能体运行超时")
        
        # 设置超时处理
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout)
        
        try:
            result = self.run_with_retry(input_text)
            return result
        except TimeoutError as e:
            return str(e)
        finally:
            signal.alarm(0)  # 取消超时设置

# 创建带重试机制的智能体
smart_agent = SmartAgentWithRetry(tools, llm)

# 使用带重试机制的智能体
test_cases = [
    "计算 2的10次方是多少？",
    "查询北京的天气情况",
    "现在是什么时间？",
    "计算 100 / 0 的结果",  # 这个会触发错误
    "查询火星的天气情况"    # 这个可能不支持
]

for test_case in test_cases:
    print(f"\n测试用例: {test_case}")
    try:
        result = smart_agent.run_with_timeout(test_case, timeout=10)
        print(f"结果: {result}")
    except Exception as e:
        print(f"错误: {str(e)}")

完整示例

完整的多工具智能体系统

下面是一个完整的多工具智能体系统实现：


import os
import json
import time
from datetime import datetime
from typing import List, Dict, Any, Optional
from pathlib import Path

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.agents import (
    initialize_agent,
    AgentType,
    Tool,
    create_structured_chat_agent
)
from langchain.prompts import ChatPromptTemplate
from langchain.tools import tool

class MultiToolAgentSystem:
    """多工具智能体系统"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.llm = self._create_llm()
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        self.tools = self._create_tools()
        self.agent = self._create_agent()
        
    def _create_llm(self):
        """创建语言模型"""
        return ChatOpenAI(
            temperature=0.3,
            model_name="gpt-3.5-turbo",
            api_key=self.api_key
        )
    
    def _create_tools(self):
        """创建工具集合"""
        @tool
        def calculate(expression: str) -> str:
            """计算数学表达式"""
            try:
                result = eval(expression)
                return f"计算结果: {result}"
            except:
                return f"计算错误: 无法计算表达式 '{expression}'"
        
        @tool
        def create_note(title: str, content: str) -> str:
            """创建笔记"""
            try:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filename = f"notes/{timestamp}_{title}.md"
                os.makedirs("notes", exist_ok=True)
                
                with open(filename, 'w', encoding='utf-8') as f:
                    f.write(f"# {title}\n\n创建时间: {datetime.now()}\n\n{content}")
                
                return f"笔记创建成功: {filename}"
            except Exception as e:
                return f"笔记创建失败: {str(e)}"
        
        @tool
        def list_notes() -> str:
            """列出所有笔记"""
            try:
                notes_dir = Path("notes")
                if not notes_dir.exists():
                    return "没有找到任何笔记"
                
                notes = []
                for file_path in notes_dir.glob("*.md"):
                    notes.append(f"- {file_path.name}")
                
                return f"找到 {len(notes)} 个笔记:\n" + "\n".join(notes)
            except Exception as e:
                return f"读取笔记列表失败: {str(e)}"
        
        @tool
        def search_web(query: str) -> str:
            """搜索网络信息"""
            try:
                # 模拟网络搜索
                search_results = {
                    "人工智能": "人工智能是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。",
                    "机器学习": "机器学习是人工智能的一个子集，专注于开发能够从数据中学习和改进的算法。",
                    "深度学习": "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的工作方式。",
                    "自然语言处理": "自然语言处理是人工智能的一个分支，专注于使计算机能够理解、解释和生成人类语言。"
                }
                
                if query in search_results:
                    return f"{query}：{search_results[query]}"
                else:
                    return f"未找到关于 '{query}' 的信息"
            except Exception as e:
                return f"搜索失败: {str(e)}"
        
        return [
            Tool(
                name="calculator",
                func=calculate,
                description="计算数学表达式，如 '2 + 2 * 3'"
            ),
            Tool(
                name="create_note",
                func=create_note,
                description="创建笔记，参数: title(标题), content(内容)"
            ),
            Tool(
                name="list_notes",
                func=list_notes,
                description="列出所有已创建的笔记"
            ),
            Tool(
                name="search_web",
                func=search_web,
                description="搜索网络信息，参数: query(搜索关键词)"
            )
        ]
    
    def _create_agent(self):
        """创建智能体"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个多功能的智能助手，能够使用以下工具帮助用户：

可用工具：
- calculator: 计算数学表达式
- create_note: 创建笔记
- list_notes: 列出笔记
- search_web: 搜索网络信息

使用规则：
1. 仔细分析用户需求
2. 选择最合适的工具
3. 调用工具并获取结果
4. 基于结果生成友好、有帮助的回答
5. 如果工具不可用或调用失败，请告知用户

请提供准确、有用、友好的回答。"""),
            ("user", "{input}"),
            ("assistant", "{agent_scratchpad}"),
        ])
        
        agent = create_structured_chat_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt,
            verbose=True
        )
        
        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            max_iterations=15,
            verbose=True,
            handle_parsing_errors=True,
            return_only_outputs=True
        )
    
    def run(self, input_text: str) -> str:
        """运行智能体"""
        try:
            result = self.agent.invoke({"input": input_text})
            return result.get("output", "没有可用的输出")
        except Exception as e:
            return f"智能体运行错误: {str(e)}"
    
    def get_memory_summary(self) -> str:
        """获取记忆摘要"""
        try:
            memory_data = self.memory.load_memory_variables({})
            if "chat_history" in memory_data:
                return f"记忆摘要：{len(memory_data['chat_history'])} 条对话记录"
            else:
                return "记忆摘要：无对话记录"
        except Exception as e:
            return f"获取记忆摘要失败: {str(e)}"
    
    def clear_memory(self):
        """清除记忆"""
        try:
            self.memory.clear()
            return "记忆已清除"
        except Exception as e:
            return f"清除记忆失败: {str(e)}"

# 使用示例
if __name__ == "__main__":
    # 初始化系统
    api_key = "your-openai-api-key"
    agent_system = MultiToolAgentSystem(api_key)
    
    # 测试用例
    test_cases = [
        "计算 25 * 4 + 10 的结果",
        "创建一个关于人工智能的笔记，内容简要介绍AI的发展历史",
        "列出所有笔记",
        "搜索机器学习的基本概念",
        "计算 2的8次方"
    ]
    
    # 运行测试
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n=== 测试用例 {i}: {test_case} ===")
        try:
            result = agent_system.run(test_case)
            print(f"结果: {result}")
            time.sleep(1)  # 避免API调用过于频繁
        except Exception as e:
            print(f"错误: {str(e)}")
    
    # 显示记忆摘要
    print(f"\n=== 记忆摘要 ===")
    print(agent_system.get_memory_summary())

常见问题 FAQ

Q1：如何选择合适的Agent类型？

A：根据任务复杂度选择：简单任务用ZERO_SHOT_REACT_DESCRIPTION，复杂任务用STRUCTURED_CHAT；根据推理需求选择：需要多步推理用REACT_CHAT，需要函数调用用OPENAI_FUNCTIONS；根据稳定性需求选择：需要稳定输出用STRUCTURED_CHAT，需要灵活性用ZERO_SHOT。

Q2：如何优化Agent的性能？

A：优化工具调用频率、使用缓存机制、减少不必要的工具调用、优化提示词设计、使用批量处理、合理设置最大迭代次数、监控token使用量、选择合适的模型温度参数。

Q3：如何处理Agent的失败情况？

A：实现重试机制、设置超时控制、提供详细的错误信息、实现降级处理、使用备选方案、记录错误日志、定期更新模型和工具、优化错误提示。

Q4：如何管理多个Agent的协作？

A：设计明确的任务分配机制、实现任务队列和优先级管理、建立通信协议、处理依赖关系、实现进度监控、设置冲突解决机制、实现容错和回滚机制。

Q5：如何确保Agent的安全性？

A：实现输入验证、设置内容过滤、限制工具权限、实现访问控制、监控异常行为、定期安全审计、使用安全的API调用、实现敏感信息保护。

最佳实践与避坑

最佳实践

合理设计工具：工具接口清晰、功能单一、错误处理完善
优化提示词：明确指令、提供上下文、设置约束条件
实现监控：监控执行过程、记录关键指标、分析性能瓶颈
渐进式开发：从简单功能开始，逐步增加复杂度
充分测试：单元测试、集成测试、性能测试、安全测试

常见陷阱

过度复杂化：工具和Agent设计过于复杂，难以维护
忽视错误处理：缺乏完善的错误处理机制
性能问题：频繁的API调用导致性能下降
资源浪费：不必要的工具调用和模型推理
安全问题：没有充分的安全考虑和防护措施

本节小结

本节详细介绍了LangChain中工具和智能体的核心概念和实现方法，包括：

智能体原理：理解Agent的决策流程和工具调用机制
工具使用：内置工具和自定义工具的使用方法
Agent配置：不同类型Agent的特点和适用场景
错误处理：完善的错误处理和重试机制
完整实践：从基础工具到协作系统的完整实现

掌握工具和智能体是构建高级AI应用的关键能力。在下一章中，我们将深入学习链式编程，进一步提升LangChain的应用深度。

延伸阅读

官方文档：LangChain Agent模块文档
相关章节：本教程 3.3 节链式调用
推荐资源：《智能体系统设计与实现》
开源项目：LangChain Agent示例库
技术博客：多智能体协作的最佳实践

关键词：工具,智能体,Agent,LangChain,协作系统,自定义工具,错误处理
难度：高级
预计阅读：40分钟