代理(Agents)基础概念
代理(Agents)是LangChain中一个强大的概念，它允许语言模型决定使用哪些工具、何时使用这些工具，以及如何使用这些工具来完成任务。代理本质上是将LLM作为推理引擎，使其能够：
- 确定需要采取的行动执行这些行动观察结果重复上述过程直到任务完成
代理的工作流程
代理的典型工作流程如下：
- 接收用户输入决定下一步行动执行行动（通常是调用工具）根据工具的输出结果进行下一步决策如此循环直到任务完成
代理的组成部分
一个完整的代理系统通常包含以下组件：
- LLM/聊天模型：作为代理的大脑，负责决策工具(Tools)：代理可以使用的功能集合代理类型：决定代理如何使用LLM来决定行动记忆(Memory)：存储对话历史和状态代理执行器(AgentExecutor)：协调以上组件的运行时
工具(Tools)介绍
工具是代理用来与世界交互的接口。在LangChain中，一个工具由以下组件组成：
- 名称(name)：在提供给LLM的工具集中必须是唯一的描述(description)：描述工具的功能，LLM将使用此描述作为上下文参数模式(args_schema)：定义工具接受的参数执行函数：实际执行工具功能的代码直接返回标志(return_direct)：仅对代理相关，决定是否直接返回结果给用户
工具的重要性
设计良好的工具对代理的性能至关重要：
- 经过微调以进行工具调用的模型将比未经微调的模型更擅长使用工具工具的名称、描述和参数模式应精心选择，以提高模型的使用效果简单的工具通常比复杂的工具更容易被模型使用
自定义工具创建
LangChain提供了三种创建自定义工具的方式：
- 使用
@tool装饰器（最简单）使用StructuredTool.from_function类方法（更多配置选项）通过子类化BaseTool（最灵活但需要更多代码）下面我们将详细介绍这三种方法。
方法一：使用@tool装饰器
这是定义自定义工具的最简单方式。该装饰器默认使用函数名称作为工具名称，并使用函数的文档字符串作为工具的描述。
from langchain_core.tools import toolfrom langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_core.prompts import ChatPromptTemplate# 定义一个简单的计算器工具@tooldef calculator(operation: str) -> float: """执行基本数学运算。 参数: operation: 要执行的数学表达式，例如 '2 + 2' 或 '(3 * 4) / 2' 返回: 计算结果 """ try: return eval(operation) except Exception as e: return f"计算错误: {str(e)}"# 定义一个异步工具@toolasync def async_calculator(operation: str) -> float: """异步执行基本数学运算。 参数: operation: 要执行的数学表达式，例如 '2 + 2' 或 '(3 * 4) / 2' 返回: 计算结果 """ try: return eval(operation) except Exception as e: return f"计算错误: {str(e)}"# 自定义工具名称和描述@tool(name="天气查询", description="查询指定城市的当前天气情况")def get_weather(city: str) -> str: """这是一个模拟的天气查询工具，实际应用中应该调用真实的天气API。 参数: city: 要查询天气的城市名称 返回: 天气信息 """ # 在实际应用中，这里应该调用天气API weather_data = { "北京": "晴朗，26°C", "上海": "多云，24°C", "广州": "小雨，28°C", "深圳": "阴天，27°C" } return weather_data.get(city, f"没有找到{city}的天气信息")# 创建代理llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个助手，可以使用提供的工具来帮助用户。"), ("human", "{input}")])tools = [calculator, get_weather]agent = create_openai_functions_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 使用代理result = agent_executor.invoke({"input": "北京今天天气怎么样？然后计算23.5乘以4"})print(result["output"])方法二：使用StructuredTool.from_function
StructuredTool.from_function类方法提供了比@tool装饰器更多的配置选项，同时不需要太多额外代码。
from langchain_core.tools import StructuredToolfrom typing import Dict, List, Optionalfrom pydantic import BaseModel, Field# 定义参数模式class TranslationInput(BaseModel): text: str = Field(..., description="要翻译的文本") source_language: str = Field(..., description="源语言，例如'中文'、'英语'") target_language: str = Field(..., description="目标语言，例如'中文'、'英语'")# 定义翻译函数def translate_text(text: str, source_language: str, target_language: str) -> str: """将文本从源语言翻译为目标语言。 这是一个模拟的翻译工具，实际应用中应该调用真实的翻译API。 """ # 在实际应用中，这里应该调用翻译API if source_language == "中文" and target_language == "英语": translations = { "你好": "Hello", "谢谢": "Thank you", "再见": "Goodbye" } return translations.get(text, f"无法翻译: {text}") elif source_language == "英语" and target_language == "中文": translations = { "Hello": "你好", "Thank you": "谢谢", "Goodbye": "再见" } return translations.get(text, f"无法翻译: {text}") else: return f"不支持从{source_language}翻译到{target_language}"# 创建结构化工具translation_tool = StructuredTool.from_function( func=translate_text, name="文本翻译", description="将文本从一种语言翻译为另一种语言", args_schema=TranslationInput, return_direct=False)# 使用工具from langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_core.prompts import ChatPromptTemplatellm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个翻译助手，可以使用提供的工具来帮助用户翻译文本。"), ("human", "{input}")])tools = [translation_tool]agent = create_openai_functions_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 使用代理result = agent_executor.invoke({"input": "请将'你好'翻译成英语"})print(result["output"])方法三：通过子类化BaseTool
通过子类化BaseTool可以获得最大的灵活性，但需要编写更多代码。
from langchain_core.tools import BaseToolfrom typing import Optional, Typefrom pydantic import BaseModel, Field# 定义参数模式class SearchInput(BaseModel): query: str = Field(..., description="搜索查询") max_results: int = Field(5, description="返回的最大结果数量")# 创建自定义工具类class SearchTool(BaseTool): name = "网络搜索" description = "搜索互联网以查找有关特定查询的信息" args_schema: Type[BaseModel] = SearchInput return_direct: bool = False def _run(self, query: str, max_results: int = 5) -> str: """运行搜索工具""" # 在实际应用中，这里应该调用搜索API search_results = { "人工智能": [ "人工智能(AI)是计算机科学的一个分支，致力于创建能够模拟人类智能的系统。", "机器学习是人工智能的一个子领域，专注于让系统从数据中学习。", "深度学习是机器学习的一种方法，使用神经网络进行学习。", "自然语言处理(NLP)是AI的一个分支，专注于让计算机理解和生成人类语言。", "计算机视觉是AI的另一个分支，专注于让计算机理解和解释视觉信息。" ], "Python编程": [ "Python是一种高级、解释型、通用编程语言。", "Python强调代码可读性，使用缩进而不是花括号来分隔代码块。", "Python支持多种编程范式，包括面向对象、命令式和函数式编程。", "Python有大量的库和框架，如NumPy、Pandas、Django和Flask。", "Python是数据科学和机器学习中最流行的编程语言之一。" ] } results = search_results.get(query, [f"没有找到关于'{query}'的搜索结果"]) return "\n".join(results[:max_results]) async def _arun(self, query: str, max_results: int = 5) -> str: """异步运行搜索工具""" # 简单地调用同步方法，在实际应用中应该使用异步API return self._run(query, max_results)# 使用自定义工具search_tool = SearchTool()# 创建代理from langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_core.prompts import ChatPromptTemplatellm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个搜索助手，可以使用提供的工具来帮助用户查找信息。"), ("human", "{input}")])tools = [search_tool]agent = create_openai_functions_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 使用代理result = agent_executor.invoke({"input": "告诉我关于人工智能的信息"})print(result["output"])工具错误处理
在使用工具时，错误处理是一个重要的考虑因素，特别是当代理需要从错误中恢复并继续执行时。LangChain提供了多种错误处理策略。
使用ToolException
可以通过抛出ToolException并设置handle_tool_error来处理工具错误：
from langchain_core.tools import ToolException, tool# 使用默认的handle_tool_error=True@tool(handle_tool_error=True)def divide(a: int, b: int) -> float: """将两个数相除。 参数: a: 被除数 b: 除数 返回: 除法结果 """ if b == 0: raise ToolException("除数不能为零") return a / b# 使用固定字符串作为错误处理@tool(handle_tool_error="除法计算出错，请检查输入")def divide_with_message(a: int, b: int) -> float: """将两个数相除。 参数: a: 被除数 b: 除数 返回: 除法结果 """ if b == 0: raise ToolException("除数不能为零") return a / b# 使用函数处理错误def custom_error_handler(e: ToolException) -> str: return f"计算错误: {str(e)}。请提供有效的输入。"@tool(handle_tool_error=custom_error_handler)def divide_with_custom_handler(a: int, b: int) -> float: """将两个数相除。 参数: a: 被除数 b: 除数 返回: 除法结果 """ if b == 0: raise ToolException("除数不能为零") return a / b内置工具和工具包
LangChain提供了大量内置工具和工具包，可以直接使用或自定义。
使用内置工具
from langchain_community.tools.wikipedia.tool import WikipediaQueryRunfrom langchain_community.utilities.wikipedia import WikipediaAPIWrapper# 创建维基百科工具wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())# 查看工具信息print(f"工具名称: {wikipedia.name}")print(f"工具描述: {wikipedia.description}")print(f"工具参数: {wikipedia.args}")# 使用工具result = wikipedia.invoke("人工智能")print(result)自定义内置工具
可以修改内置工具的名称、描述和参数模式：
from langchain_community.tools.wikipedia.tool import WikipediaQueryRunfrom langchain_community.utilities.wikipedia import WikipediaAPIWrapperfrom langchain_core.pydantic_v1 import BaseModel, Field# 定义自定义参数模式class CustomWikipediaInput(BaseModel): query: str = Field(..., description="要在维基百科上搜索的查询") lang: str = Field("zh", description="维基百科的语言版本，默认为中文(zh)")# 创建并自定义维基百科工具wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())wikipedia.name = "维基百科搜索"wikipedia.description = "在维基百科上搜索信息，适用于查找事实性知识"wikipedia.args_schema = CustomWikipediaInput使用工具包(Toolkits)
工具包是一组旨在一起使用以执行特定任务的工具集合。
from langchain_community.agent_toolkits import SQLDatabaseToolkitfrom langchain_community.utilities.sql_database import SQLDatabasefrom langchain_openai import ChatOpenAI# 连接到SQLite数据库db = SQLDatabase.from_uri("sqlite:///langchain.db")# 创建SQL数据库工具包toolkit = SQLDatabaseToolkit(db=db, llm=ChatOpenAI(temperature=0))# 获取工具包中的所有工具tools = toolkit.get_tools()print(f"工具包中的工具数量: {len(tools)}")for i, tool in enumerate(tools): print(f"工具 {i+1}: {tool.name} - {tool.description}")# 创建代理from langchain.agents import create_sql_agentfrom langchain_openai import ChatOpenAIllm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)agent_executor = create_sql_agent( llm=llm, toolkit=toolkit, verbose=True)# 使用代理result = agent_executor.invoke({"input": "描述full_llm_cache表的结构"})print(result["output"])实际应用案例：构建多功能助手
下面是一个结合多种工具的实际应用案例，构建一个多功能助手：
import osfrom langchain_core.tools import toolfrom langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_community.tools.wikipedia.tool import WikipediaQueryRunfrom langchain_community.utilities.wikipedia import WikipediaAPIWrapperfrom langchain_community.tools.python.tool import PythonREPLToolfrom datetime import datetime# 设置API密钥os.environ["OPENAI_API_KEY"] = "你的OpenAI API密钥"# 定义计算器工具@tooldef calculator(operation: str) -> float: """执行基本数学运算。 参数: operation: 要执行的数学表达式，例如 '2 + 2' 或 '(3 * 4) / 2' 返回: 计算结果 """ try: return eval(operation) except Exception as e: return f"计算错误: {str(e)}"# 定义日期时间工具@tooldef get_current_datetime() -> str: """获取当前的日期和时间。 返回: 当前的日期和时间，格式为'YYYY-MM-DD HH:MM:SS' """ return datetime.now().strftime("%Y-%m-%d %H:%M:%S")# 创建维基百科工具wikipedia_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())wikipedia_tool.name = "维基百科搜索"wikipedia_tool.description = "在维基百科上搜索信息，适用于查找事实性知识"# 创建Python REPL工具python_repl = PythonREPLTool()python_repl.name = "Python执行器"python_repl.description = "执行Python代码并返回结果，适用于复杂计算和数据处理"# 定义翻译工具@tooldef translate(text: str, target_language: str) -> str: """将文本翻译为目标语言。 参数: text: 要翻译的文本 target_language: 目标语言，例如'中文'、'英语'、'日语'等 返回: 翻译后的文本 """ # 在实际应用中，这里应该调用翻译API translations = { ("Hello", "中文"): "你好", ("Thank you", "中文"): "谢谢", ("Goodbye", "中文"): "再见", ("你好", "英语"): "Hello", ("谢谢", "英语"): "Thank you", ("再见", "英语"): "Goodbye" } key = (text, target_language) return translations.get(key, f"无法将'{text}'翻译为{target_language}")# 创建代理llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个多功能助手，可以使用各种工具来帮助用户。 你可以进行计算、查询信息、翻译文本、获取当前时间，甚至执行Python代码。 请根据用户的需求选择合适的工具，并提供有用的回答。"""), ("human", "{input}")])tools = [calculator, get_current_datetime, wikipedia_tool, python_repl, translate]agent = create_openai_functions_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 使用多功能助手queries = [ "计算25乘以16再减去30", "现在是什么时间？", "告诉我关于人工智能的信息", "使用Python计算斐波那契数列的前10个数", "将'你好'翻译成英语"]for query in queries: print(f"\n问题: {query}") result = agent_executor.invoke({"input": query}) print(f"回答: {result['output']}")高级工具使用技巧
1. 组合多个工具
有时候，我们需要将多个工具组合起来解决复杂问题。可以通过创建一个新工具，该工具内部调用其他工具：
@tooldef comprehensive_search(query: str) -> str: """执行综合搜索，包括维基百科和其他来源。 参数: query: 搜索查询 返回: 综合搜索结果 """ # 调用维基百科工具 wiki_result = wikipedia_tool.invoke(query) # 可以在这里添加其他搜索源 # other_result = other_search_tool.invoke(query) return f"维基百科结果:\n{wiki_result}\n\n" # 可以添加其他搜索结果2. 工具的参数验证
使用Pydantic模型可以为工具参数添加验证：
from pydantic import BaseModel, Field, validatorclass CalculatorInput(BaseModel): operation: str = Field(..., description="要执行的数学表达式") @validator("operation") def validate_operation(cls, v): # 安全检查，防止执行危险代码 if any(keyword in v for keyword in ["import", "exec", "eval", "os.", "sys."]): raise ValueError("不允许执行系统或导入相关操作") return v@tool(args_schema=CalculatorInput)def safe_calculator(operation: str) -> float: """安全地执行基本数学运算。 参数: operation: 要执行的数学表达式，例如 '2 + 2' 或 '(3 * 4) / 2' 返回: 计算结果 """ try: return eval(operation) except Exception as e: return f"计算错误: {str(e)}"3. 工具的文档和示例
为工具提供详细的文档和示例可以帮助LLM更好地理解如何使用工具：
@tooldef format_date(date_str: str, input_format: str, output_format: str) -> str: """将日期从一种格式转换为另一种格式。 参数: date_str: 要格式化的日期字符串，例如 '2023-01-15' input_format: 输入日期的格式，例如 '%Y-%m-%d' output_format: 输出日期的格式，例如 '%d/%m/%Y' 示例: - 输入: date_str='2023-01-15', input_format='%Y-%m-%d', output_format='%d/%m/%Y' 输出: '15/01/2023' - 输入: date_str='01/15/2023', input_format='%m/%d/%Y', output_format='%Y年%m月%d日' 输出: '2023年01月15日' 格式代码: %Y - 四位数年份 (例如 2023) %m - 两位数月份 (01-12) %d - 两位数日期 (01-31) %H - 两位数小时 (00-23) %M - 两位数分钟 (00-59) %S - 两位数秒钟 (00-59) 返回: 格式化后的日期字符串 """ from datetime import datetime try: date_obj = datetime.strptime(date_str, input_format) return date_obj.strftime(output_format) except Exception as e: return f"日期格式化错误: {str(e)}"结论
LangChain的代理和工具系统提供了强大的功能，使我们能够构建智能应用，让语言模型与外部世界交互。通过自定义工具，我们可以扩展模型的能力，使其能够执行各种任务，从简单的计算到复杂的数据处理。
关键要点：
- 工具是代理与世界交互的接口，由名称、描述、参数模式和执行函数组成LangChain提供了三种创建工具的方式：
@tool装饰器、StructuredTool.from_function和子类化BaseTool错误处理是工具设计的重要部分，可以使用ToolException和handle_tool_error来处理错误LangChain提供了大量内置工具和工具包，可以直接使用或自定义工具的名称、描述和参数模式对模型的使用效果有重要影响通过合理设计和组合工具，我们可以构建功能强大的代理系统，解决各种复杂问题。
下一章：08_LangGraph基础 - 我们将探索LangGraph，这是LangChain的一个扩展，用于构建更复杂的工作流程和状态机。
