尧图网站建设 
1. 从零到一为什么我们需要封装自己的LangChain工具集如果你已经开始用LangChain构建应用大概率已经体验过它内置的那些“开箱即用”的工具比如GoogleSearchRun或者WikipediaQueryRun。它们很方便点几下鼠标一个能联网搜索的Agent就搭好了。但很快你就会遇到一个现实问题我的业务逻辑、我的内部API、我那个用FastAPI写的奇怪服务怎么让LangChain的Agent去调用这就是我们今天要聊的核心——封装你自己的LangChain工具集。这不仅仅是写个函数那么简单它关乎着你能否将大语言模型LLM的能力无缝、安全、可控地注入到你自己的业务流和数据孤岛中。想象一个场景你有一个电商客服Agent用户问“我上周三买的那个蓝色卫衣发货了吗”。一个理想的Agent应该能1. 理解用户意图查询物流2. 从对话中提取关键信息用户身份、订单时间、商品特征3. 调用你内部的“订单查询API”传入这些参数4. 理解API返回的JSON生成人类可读的回答。这里的第3步就是自定义工具的用武之地。LangChain内置的工具不可能知道你的API长什么样所以你必须自己来“教”它。封装工具集Toolkits则是更进一步的抽象。单个工具就像一把螺丝刀而工具集就是一个完整的“家用维修工具箱”里面包含了拧螺丝、测电、敲钉子等一系列相关的工具。例如一个“数据库操作工具箱”可能包含“执行查询”、“获取表结构”、“写入数据”等多个工具它们共享数据库连接共同完成一个复杂的数据库交互任务。封装工具集意味着你开始以“领域”或“任务”的维度来组织你的工具让Agent的“技能包”更加模块化和专业。所以当你决定封装第一个工具时你实际上是在为你的LLM应用搭建一座通往现实世界数据和服务的桥梁。而当你开始组织工具集时你是在为这座桥梁设计立交桥和收费站让交通数据流和控制流更高效、更有序。接下来我们就从最基础的单个工具封装开始一步步构建起你自己的工具箱。2. 工具Tool的本质一个让LLM学会“动手”的接口在深入代码之前我们必须先理解LangChain中“工具”这个概念的本质。它不是一个魔法黑盒而是一个设计精巧的适配器。简单来说一个Tool就是将一段Python代码一个函数或一个对象的方法“包装”起来并附上一份清晰的“说明书”名称、描述、参数schema使得LLM能够理解这个工具是干什么的、怎么用。为什么需要这份“说明书”因为LLM比如GPT-4本质上是一个文本生成模型它不会直接执行Python代码。它通过“思考”生成文本来决定要调用哪个工具并生成调用这个工具所需的参数。这份“说明书”就是它做决策的依据。一个定义良好的工具其描述应该清晰到让LLM能准确判断在什么场景下使用它。让我们来看一个最基础的例子封装一个查询天气的工具。假设我们有一个简单的函数def get_current_weather(location: str) - str: 根据城市名获取当前天气情况。 # 这里应该是调用真实天气API的代码例如OpenWeatherMap # 为了示例我们返回一个模拟数据 return f{location}的天气是晴朗25摄氏度。如果直接把这个函数丢给AgentAgent是无法使用的。我们需要用LangChain提供的方式来包装它。最直接的方法是使用tool装饰器这是LangChain提供的最便捷的创建工具的方式之一。from langchain.tools import tool tool def get_current_weather(location: str) - str: 根据城市名获取当前天气情况。 return f{location}的天气是晴朗25摄氏度。看代码几乎没变只是加了一个装饰器。但神奇的事情发生了LangChain会自动从这个函数的文档字符串根据城市名获取当前天气情况。和函数签名中提取信息生成一个符合LangChain规范的Tool对象。这个对象包含了名称函数名get_current_weather、描述文档字符串和参数schema一个名为location的字符串参数。注意装饰器方式虽然简便但控制力较弱。例如工具的描述完全依赖于你的文档字符串。如果你的函数逻辑复杂或者你想对输入参数进行更复杂的校验和说明使用StructuredTool类会是更强大的选择。StructuredTool允许你以更声明式、更细致的方式来定义一个工具。你可以明确指定每一个字段from langchain.tools import StructuredTool def get_weather(location: str) - str: return f{location}的天气是晴朗25摄氏度。 weather_tool StructuredTool.from_function( funcget_weather, nameGetCurrentWeather, description查询指定城市的当前天气情况。输入应为城市名称例如‘北京’或‘New York’。, args_schemaNone, # 可以传入一个Pydantic模型来严格定义参数 )这里有几个关键点name工具的名称。这是Agent在“思考”时会看到的标识。起一个清晰、动词开头的名字如GetCurrentWeather、SearchDatabase会帮助LLM更好地理解其用途。description这是重中之重。描述应该清晰、无歧义地说明工具的功能、适用场景以及输入的格式。好的描述是Agent正确使用工具的一半保障。例如“查询天气”就不如“根据完整的城市名称查询当前的温度、湿度和天气状况输入格式如‘北京市’或‘San Francisco, CA’”。func背后实际执行的Python可调用对象。args_schema这是一个高级功能允许你使用Pydantic模型来严格定义参数的名称、类型、描述甚至校验规则。这对于复杂工具至关重要能确保LLM生成的参数符合后端API的期望。实操心得描述Description是你的第一道防线我踩过的第一个坑就是工具描述写得太随意。早期我给一个“用户信息查询”工具的描述是“查询用户数据”。结果Agent在需要找“张三的订单”时也调用了这个工具因为它觉得“订单”也是“用户数据”的一部分。这导致了错误的API调用和混乱的结果。后来我把描述改为“根据用户IDUUID格式查询用户的姓名、注册邮箱和基础档案不包含订单、日志等其他信息。”问题就迎刃而解了。所以花时间打磨工具的描述就像给API写一份优秀的文档能省去后面大量的调试和纠错成本。3. 超越单个函数封装复杂对象与异步工具现实世界的工具 rarely 是一个简单的纯函数。它可能是一个需要初始化的客户端对象一个需要管理状态如数据库连接、会话的类或者是一个异步的协程。LangChain的Tool体系都能很好地支持。场景一封装一个类方法假设我们有一个DatabaseClient类它已经建立了连接我们想暴露它的查询方法作为工具。from langchain.tools import StructuredTool class DatabaseClient: def __init__(self, connection_string): self.conn create_engine(connection_string) def run_query(self, sql: str) - list: 执行SQL查询语句并返回结果列表。 # 执行SQL的逻辑... return results # 初始化客户端 db_client DatabaseClient(your_connection_string) # 将类方法封装为工具 # 注意这里我们需要一个函数来绑定self通常使用lambda或定义一个闭包函数 db_query_tool StructuredTool.from_function( funclambda sql: db_client.run_query(sql), # 绑定实例 nameRunDatabaseQuery, description执行一个只读的SQL SELECT查询语句并返回结果。请确保SQL语法正确。, )这里的关键在于func参数需要的是一个可调用对象。我们通过lambda表达式将实例方法db_client.run_query转换成了一个只需要sql一个参数的函数完美符合Tool的要求。场景二处理异步工具如果你的后端调用是异步的比如使用aiohttp调用异步API那么你的工具函数也应该是一个异步函数。LangChain完全支持异步工具。import aiohttp from langchain.tools import StructuredTool async def async_fetch_data(api_endpoint: str, params: dict) - dict: 异步调用指定的API端点获取数据。 async with aiohttp.ClientSession() as session: async with session.get(api_endpoint, paramsparams) as response: return await response.json() async_api_tool StructuredTool.from_function( funcasync_fetch_data, nameAsyncFetchData, description异步获取数据。参数api_endpoint是URLparams是查询参数字典。, coroutineasync_fetch_data, # 对于异步函数需要显式指定coroutine )当你使用支持异步的Agent如create_react_agent时它可以正确地await这个工具。这能显著提升在I/O密集型场景如同时调用多个外部API下的应用性能。场景三带有复杂参数校验的工具对于输入参数复杂、需要严格校验的工具强烈推荐使用args_schema。这需要借助Pydantic。from pydantic import BaseModel, Field from langchain.tools import StructuredTool class WeatherQueryInput(BaseModel): location: str Field(description城市名称例如‘北京’、‘上海’) unit: str Field(defaultcelsius, description温度单位可选‘celsius’摄氏度或‘fahrenheit’华氏度) def get_weather_advanced(location: str, unit: str celsius) - str: # 实现... pass weather_advanced_tool StructuredTool.from_function( funcget_weather_advanced, nameGetWeatherAdvanced, description查询天气支持选择温度单位。, args_schemaWeatherQueryInput, # 传入Pydantic模型 )这样做的好处是自文档化LLM能清晰地看到每个参数的意义和可选值。输入校验在工具被调用前LangChain会先用Pydantic模型校验参数如果unit的值不是celsius或fahrenheit会提前报错避免无效的API调用。类型安全为整个流程增加了类型提示方便开发和调试。注意使用args_schema时你的工具函数func的参数名必须与Pydantic模型中的字段名完全一致。LangChain会按照模型字段来调用函数。4. 构建工具集Toolkit将散兵游勇组织成特种部队当你拥有了几个相关的工具后比如针对数据库的RunQueryTool、ListTablesTool、GetSchemaTool把它们作为一个整体来管理和使用会更加方便。这就是Toolkit工具集的概念。一个Toolkit本质上就是一个包含多个Tool对象的容器通常还附带一些初始化或共享状态的逻辑。LangChain提供了一些内置的Toolkit如SQLDatabaseToolkit但更多时候我们需要创建自己的。创建自定义Toolkit非常简单通常就是继承BaseToolkit类然后在get_tools方法中返回你的工具列表。为什么需要Toolkit模块化管理将与某个特定领域数据库、文件系统、CRM系统相关的所有工具打包在一起代码结构更清晰。共享状态Toolkit的初始化方法__init__可以用来创建和持有共享资源比如一个数据库连接池。这个连接池可以被它返回的所有工具使用避免了每个工具都独立创建连接的开销和混乱。便于Agent加载很多Agent的初始化方法可以直接接受一个Toolkit对象它会自动提取其中的所有工具比手动传入一个工具列表更简洁。让我们构建一个简单的“文件系统工具箱”from langchain.agents.agent_toolkits.base import BaseToolkit from langchain.tools import Tool import os from typing import List class FileSystemToolkit(BaseToolkit): 一个用于基础文件操作的工具箱。 def __init__(self, base_path: str .): # 可以定义一个基础路径所有文件操作都相对此路径进行增加安全性 self.base_path os.path.abspath(base_path) def get_tools(self) - List[Tool]: # 在这里定义并返回属于这个工具箱的所有工具 return [ self._create_list_files_tool(), self._create_read_file_tool(), self._create_file_stats_tool() ] def _create_list_files_tool(self) - Tool: def list_files(directory: str .) - str: 列出指定目录下的文件和文件夹。 target_path os.path.join(self.base_path, directory) # 安全检查确保目标路径在base_path之下 if not os.path.commonpath([self.base_path, os.path.abspath(target_path)]) self.base_path: return 错误试图访问超出允许范围的路径。 try: items os.listdir(target_path) return \n.join(items) if items else 目录为空。 except FileNotFoundError: return f错误目录 {directory} 不存在。 except PermissionError: return f错误没有权限访问目录 {directory}。 return Tool( nameListDirectoryFiles, funclist_files, description列出指定目录下的内容。输入是目录的相对路径默认为当前目录。 ) def _create_read_file_tool(self) - Tool: # 类似地实现读文件工具这里省略具体函数体 def read_file(filepath: str) - str: 读取文本文件的内容。 full_path os.path.join(self.base_path, filepath) # ... 路径安全检查 ... try: with open(full_path, r, encodingutf-8) as f: return f.read() except Exception as e: return f读取文件失败{str(e)} return Tool( nameReadTextFile, funcread_file, description读取指定文本文件的内容。输入是文件的相对路径。 ) def _create_file_stats_tool(self) - Tool: # 实现获取文件信息的工具 pass # 使用工具箱 fs_toolkit FileSystemToolkit(base_path/safe/workspace) tools fs_toolkit.get_tools() # 获取到三个工具的列表在这个例子中FileSystemToolkit类在初始化时接收一个base_path将其作为文件操作的根目录这是一个重要的安全实践。所有工具方法在执行实际操作前都会检查目标路径是否在base_path之下防止路径穿越攻击。这种共享的安全策略在Toolkit层面统一管理比在每个工具里重复实现要优雅和可靠得多。实操心得在Toolkit初始化中做“脏活累活”把那些繁琐但必要的初始化、配置加载、资源创建如数据库连接、API客户端实例化、认证令牌获取放在Toolkit的__init__方法里。这样当你把Toolkit交给Agent时这些依赖已经准备就绪。例如我的“内部API工具箱”会在__init__里读取配置文件初始化一个带重试和熔断机制的HTTP客户端并获取一个OAuth 2.0令牌。这个客户端实例会被所有工具方法共享既保证了效率复用连接和令牌也统一了错误处理逻辑。5. 高级模式动态工具与元工具在一些更复杂的智能体场景中我们需要的工具可能不是静态的而是根据上下文动态生成或变化的。LangChain的框架也支持这种动态性。动态工具生成假设你有一个工具它的功能取决于运行时的某些状态。例如一个“切换数据源”的工具它执行后会改变另一个“查询数据”工具背后的连接。我们可以通过让工具访问和修改共享状态来实现。from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type class DynamicDataSourceContext(BaseModel): current_source: str default class QueryDataTool(BaseTool): name QueryData description 从当前数据源查询数据。 args_schema: Type[BaseModel] None # 简单工具不需要复杂schema context: DynamicDataSourceContext # 持有共享状态的引用 def _run(self, query: str) - str: # 根据 self.context.current_source 决定查询哪个数据源 if self.context.current_source source_a: return f从源A查询 {query} 的结果。 elif self.context.current_source source_b: return f从源B查询 {query} 的结果。 else: return f从默认源查询 {query} 的结果。 def _arun(self, query: str): raise NotImplementedError(异步不支持) class SwitchDataSourceTool(BaseTool): name SwitchDataSource description 切换当前使用的数据源。 args_schema: Type[BaseModel] None context: DynamicDataSourceContext def _run(self, source_name: str) - str: if source_name in [source_a, source_b, default]: self.context.current_source source_name return f已切换数据源至{source_name} else: return f错误未知数据源 {source_name} def _arun(self, source_name: str): raise NotImplementedError(异步不支持) # 使用 shared_context DynamicDataSourceContext() tool1 QueryDataTool(contextshared_context) tool2 SwitchDataSourceTool(contextshared_context)这里两个工具通过共享同一个context对象实现了状态的联动。当Agent使用SwitchDataSourceTool切换源后后续的QueryDataTool调用就会使用新的数据源。元工具Meta-Tool元工具是指那些不直接完成最终任务而是用于管理、评估或组合其他工具的工具。一个经典的例子是“Human-in-the-loop”工具它暂停Agent的执行将问题抛给人类用户并将用户的输入返回给Agent。from langchain.tools import BaseTool from langchain.callbacks.manager import CallbackManagerForToolRun class HumanInputTool(BaseTool): name AskHuman description 当你需要澄清一个模糊的问题、获取额外信息或确认一个关键操作时使用此工具向人类用户提问。 def _run(self, question: str, run_manager: CallbackManagerForToolRun | None None) - str: # 在实际应用中这里可以通过前端界面、消息队列等方式将问题呈现给用户 # 这里我们模拟一个简单的命令行输入 print(f\n[Agent需要您的输入]: {question}) user_response input(您的回答: ) return user_response将这个工具放入Agent的工具列表后Agent在遇到不确定的情况时就可以主动向人类求助。这极大地增加了复杂、高风险任务中Agent的可靠性和安全性。其他元工具的思路还包括“评估当前计划可行性”、“总结已执行步骤”等它们帮助Agent进行更高层次的规划和反思。6. 实战封装一个企业内部知识库查询工具集现在让我们综合以上所有知识完成一个更贴近真实业务的实战封装一个用于查询企业内部知识库假设是一个Elasticsearch集群的工具集。这个工具集需要包含1. 基础检索工具2. 高级过滤工具3. 获取文档详情的工具。第一步设计共享客户端与基础工具我们首先创建一个KnowledgeBaseToolkit它在初始化时建立Elasticsearch客户端连接。from elasticsearch import Elasticsearch from langchain.agents.agent_toolkits.base import BaseToolkit from langchain.tools import StructuredTool from pydantic import BaseModel, Field from typing import List, Optional import json class KnowledgeBaseToolkit(BaseToolkit): def __init__(self, es_hosts: List[str], index_name: str company-wiki): self.es_client Elasticsearch(es_hosts) self.index_name index_name # 可以在这里进行连接健康检查等初始化操作 def get_tools(self): return [ self._create_basic_search_tool(), self._create_advanced_search_tool(), self._create_get_document_tool() ] def _create_basic_search_tool(self): class BasicSearchInput(BaseModel): query: str Field(description搜索关键词例如‘年假政策’或‘项目报销流程’) size: int Field(default5, description返回结果的最大数量默认5条) def basic_search(query: str, size: int 5) - str: 在知识库中执行基础全文搜索。 body { query: { multi_match: { query: query, fields: [title^2, content] # 标题权重更高 } }, size: size } try: response self.es_client.search(indexself.index_name, bodybody) hits response[hits][hits] if not hits: return 未找到相关文档。 results [] for hit in hits: source hit[_source] results.append(f标题{source.get(title, N/A)}\n摘要{source.get(summary, N/A)[:100]}...\nID{hit[_id]}\n) return \n---\n.join(results) except Exception as e: return f搜索过程中发生错误{str(e)} return StructuredTool.from_function( funcbasic_search, nameBasicKnowledgeSearch, description根据关键词在知识库中进行全文搜索。返回匹配文档的标题、摘要和ID。, args_schemaBasicSearchInput )第二步实现带过滤的高级搜索工具高级搜索工具允许用户通过分类、标签、创建时间等字段进行过滤。def _create_advanced_search_tool(self): class AdvancedSearchInput(BaseModel): query: Optional[str] Field(defaultNone, description可选的关键词搜索) category: Optional[str] Field(defaultNone, description按分类过滤如‘HR’、‘IT’) tags: Optional[List[str]] Field(defaultNone, description按标签过滤如[‘urgent’ ‘policy’]) from_date: Optional[str] Field(defaultNone, description起始日期格式YYYY-MM-DD) def advanced_search(query: Optional[str] None, category: Optional[str] None, tags: Optional[List[str]] None, from_date: Optional[str] None) - str: 执行带过滤条件的高级搜索。 must_conditions [] if query: must_conditions.append({multi_match: {query: query, fields: [title, content]}}) if category: must_conditions.append({term: {category.keyword: category}}) if tags: must_conditions.append({terms: {tags.keyword: tags}}) if from_date: must_conditions.append({range: {created_at: {gte: from_date}}}) body { query: { bool: { must: must_conditions if must_conditions else [{match_all: {}}] } }, size: 10, sort: [{created_at: {order: desc}}] # 按时间倒序 } # ... 执行搜索并格式化结果的代码 ... # 返回格式化的字符串 return formatted_results return StructuredTool.from_function( funcadvanced_search, nameAdvancedKnowledgeSearch, description使用关键词、分类、标签、日期范围等条件组合搜索知识库文档。, args_schemaAdvancedSearchInput )第三步实现获取文档详情的工具当Agent通过搜索工具找到感兴趣的文档ID后可以用这个工具获取全文。def _create_get_document_tool(self): class GetDocInput(BaseModel): doc_id: str Field(description知识库文档的唯一ID) def get_document(doc_id: str) - str: 根据文档ID获取文档的完整内容。 try: response self.es_client.get(indexself.index_name, iddoc_id) source response[_source] # 格式化输出突出重点信息 output f文档ID{doc_id}\n output f标题{source.get(title, N/A)}\n output f分类{source.get(category, N/A)}\n output f标签{, .join(source.get(tags, []))}\n output f创建时间{source.get(created_at, N/A)}\n output --- 内容 ---\n output source.get(content, 内容为空) return output except Exception as e: return f获取文档失败ID ‘{doc_id}’{str(e)} return StructuredTool.from_function( funcget_document, nameGetKnowledgeDocument, description根据文档ID获取知识库中某篇文档的完整详细信息包括标题、分类、标签和全文内容。, args_schemaGetDocInput )整合与使用现在我们可以初始化这个工具箱并将其提供给一个Agent# 初始化工具箱 kb_toolkit KnowledgeBaseToolkit( es_hosts[http://localhost:9200], index_namecompany-wiki-prod ) kb_tools kb_toolkit.get_tools() # 假设我们有一个LLM和Agent执行器 from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4, temperature0) agent create_react_agent(llm, kb_tools, verboseTrue) agent_executor AgentExecutor(agentagent, toolskb_tools, verboseTrue) # 现在Agent就可以使用这些工具了 result agent_executor.invoke({ input: 帮我找一下关于2024年新员工入职培训的IT部门相关文档要最近三个月发布的。 })在这个查询中Agent可能会先使用AdvancedKnowledgeSearch工具设置query为“新员工入职培训”category为“IT”from_date为三个月前的日期。在得到一批结果包含文档ID后如果用户需要查看某篇的具体内容Agent可以再调用GetKnowledgeDocument工具。避坑经验工具描述的协同与防冲突当工具集内工具变多时要特别注意工具描述的区分度。如果BasicKnowledgeSearch和AdvancedKnowledgeSearch的描述过于相似Agent可能会困惑该用哪个。我的经验是在描述中明确各自的使用场景和输入特点。例如基础搜索“快速根据一两个关键词查找相关文档。”高级搜索“当需要结合多个条件如分类、标签、时间进行精确筛选时使用。”此外工具的名称也要有区分度避免使用Search和SearchDocument这样容易混淆的名字。7. 调试、测试与最佳实践封装好工具后在集成到复杂的Agent流程前进行充分的独立测试是至关重要的。这能帮你快速定位是工具逻辑问题还是Agent的理解和调用问题。单元测试你的工具像测试普通Python函数一样测试你的工具。确保在各种边界条件下如空输入、错误参数、网络异常工具都能返回合理的、结构化的错误信息而不是抛出未处理的异常。# 假设测试上面的 basic_search 工具 def test_basic_search(): toolkit KnowledgeBaseToolkit(es_hosts[dummy], index_nametest) tools toolkit.get_tools() search_tool [t for t in tools if t.name BasicKnowledgeSearch][0] # 测试正常查询 result search_tool.run({query: 测试, size: 3}) assert isinstance(result, str) assert 标题 in result or 未找到 in result # 测试空结果 # 可以通过mock Elasticsearch客户端来模拟返回空结果 # ... # 测试异常处理 # 模拟客户端抛出异常检查返回的字符串是否包含友好的错误信息 print(工具基础测试通过。)在LangChain中直接测试工具调用使用Tool.invoke()或直接调用func来验证工具是否能被正确触发并返回预期结果。# 获取工具并直接调用 weather_tool get_weather_advanced_tool # 假设这是之前定义的工具 try: output weather_tool.invoke({location: 北京, unit: celsius}) print(f工具调用成功输出{output}) except Exception as e: print(f工具调用失败{e})集成到Agent前的“沙盒”测试创建一个最简单的Agent只包含你新开发的工具用一些典型的用户问题去测试它。观察Agent的思考链通过设置verboseTrue看它是否正确地选择了你期望的工具并且生成的参数是否符合args_schema的预期。from langchain.agents import initialize_agent, AgentType from langchain_openai import ChatOpenAI llm ChatOpenAI(temperature0) test_agent initialize_agent( tools[weather_tool], # 只放一个工具进行测试 llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, # 使用一个简单的Agent类型 verboseTrue, handle_parsing_errorsTrue # 处理解析错误 ) result test_agent.run(上海今天天气怎么样) print(result)通过verboseTrue的输出你可以清晰地看到Agent的“Thought”思考过程它是否理解了问题并决定调用GetWeatherAdvanced“Action”和“Action Input”它生成的调用动作和参数是否正确例如Action Input应该是{location: 上海, unit: celsius}这样的JSON。“Observation”你的工具返回的结果是什么。“Final Answer”Agent最终给用户的回答。最佳实践清单根据我的经验封装一个健壮、易用的工具集请牢记以下几点描述即契约花80%的精力打磨工具的名称和描述。确保它们清晰、无歧义、能自我解释。好的描述能极大降低Agent的误用率。输入验证与净化永远不要相信LLM生成的输入是完美无缺的。在工具函数内部对输入参数进行类型转换、范围检查、敏感词过滤等。对于文件路径、数据库查询语句等要严防注入攻击。优雅的错误处理工具函数必须捕获所有可能的异常并返回一个对LLM和最终用户都有意义的字符串错误信息。避免抛出未处理的异常导致整个Agent崩溃。返回的信息应能帮助Agent进行后续决策例如“未找到文件请检查路径” vs. “FileNotFoundError”。保持工具功能单一一个工具只做一件事。不要创建一个“万能”工具。单一职责的工具更容易被LLM理解、组合和调试。为工具输出设计格式工具的输出是给LLM看的“观察”。设计清晰、结构化的输出格式如使用\n分隔不同字段使用---分隔不同条目有助于LLM解析并生成更好的最终答案。性能考量如果工具涉及网络请求或复杂计算考虑增加超时、重试和缓存机制。对于频繁调用的工具缓存可以显著提升Agent的响应速度。版本控制与文档像管理API一样管理你的工具集。当工具接口参数、返回值发生变化时要有明确的版本更迭意识并更新对应的描述和文档。封装自己的LangChain工具和工具集是将LLM从“聪明的聊天者”转变为“能干的业务助手”的关键一步。这个过程始于对业务逻辑的深刻理解成于对接口设计的细致打磨。当你看到Agent自如地调用着你亲手封装的服务并组合它们解决复杂问题时那种成就感正是驱动我们不断深入探索AI应用落地的动力。 LangChain自定义工具与工具集封装实战：从基础原理到企业级应用
1. 从零到一为什么我们需要封装自己的LangChain工具集如果你已经开始用LangChain构建应用大概率已经体验过它内置的那些“开箱即用”的工具比如GoogleSearchRun或者WikipediaQueryRun。它们很方便点几下鼠标一个能联网搜索的Agent就搭好了。但很快你就会遇到一个现实问题我的业务逻辑、我的内部API、我那个用FastAPI写的奇怪服务怎么让LangChain的Agent去调用这就是我们今天要聊的核心——封装你自己的LangChain工具集。这不仅仅是写个函数那么简单它关乎着你能否将大语言模型LLM的能力无缝、安全、可控地注入到你自己的业务流和数据孤岛中。想象一个场景你有一个电商客服Agent用户问“我上周三买的那个蓝色卫衣发货了吗”。一个理想的Agent应该能1. 理解用户意图查询物流2. 从对话中提取关键信息用户身份、订单时间、商品特征3. 调用你内部的“订单查询API”传入这些参数4. 理解API返回的JSON生成人类可读的回答。这里的第3步就是自定义工具的用武之地。LangChain内置的工具不可能知道你的API长什么样所以你必须自己来“教”它。封装工具集Toolkits则是更进一步的抽象。单个工具就像一把螺丝刀而工具集就是一个完整的“家用维修工具箱”里面包含了拧螺丝、测电、敲钉子等一系列相关的工具。例如一个“数据库操作工具箱”可能包含“执行查询”、“获取表结构”、“写入数据”等多个工具它们共享数据库连接共同完成一个复杂的数据库交互任务。封装工具集意味着你开始以“领域”或“任务”的维度来组织你的工具让Agent的“技能包”更加模块化和专业。所以当你决定封装第一个工具时你实际上是在为你的LLM应用搭建一座通往现实世界数据和服务的桥梁。而当你开始组织工具集时你是在为这座桥梁设计立交桥和收费站让交通数据流和控制流更高效、更有序。接下来我们就从最基础的单个工具封装开始一步步构建起你自己的工具箱。2. 工具Tool的本质一个让LLM学会“动手”的接口在深入代码之前我们必须先理解LangChain中“工具”这个概念的本质。它不是一个魔法黑盒而是一个设计精巧的适配器。简单来说一个Tool就是将一段Python代码一个函数或一个对象的方法“包装”起来并附上一份清晰的“说明书”名称、描述、参数schema使得LLM能够理解这个工具是干什么的、怎么用。为什么需要这份“说明书”因为LLM比如GPT-4本质上是一个文本生成模型它不会直接执行Python代码。它通过“思考”生成文本来决定要调用哪个工具并生成调用这个工具所需的参数。这份“说明书”就是它做决策的依据。一个定义良好的工具其描述应该清晰到让LLM能准确判断在什么场景下使用它。让我们来看一个最基础的例子封装一个查询天气的工具。假设我们有一个简单的函数def get_current_weather(location: str) - str: 根据城市名获取当前天气情况。 # 这里应该是调用真实天气API的代码例如OpenWeatherMap # 为了示例我们返回一个模拟数据 return f{location}的天气是晴朗25摄氏度。如果直接把这个函数丢给AgentAgent是无法使用的。我们需要用LangChain提供的方式来包装它。最直接的方法是使用tool装饰器这是LangChain提供的最便捷的创建工具的方式之一。from langchain.tools import tool tool def get_current_weather(location: str) - str: 根据城市名获取当前天气情况。 return f{location}的天气是晴朗25摄氏度。看代码几乎没变只是加了一个装饰器。但神奇的事情发生了LangChain会自动从这个函数的文档字符串根据城市名获取当前天气情况。和函数签名中提取信息生成一个符合LangChain规范的Tool对象。这个对象包含了名称函数名get_current_weather、描述文档字符串和参数schema一个名为location的字符串参数。注意装饰器方式虽然简便但控制力较弱。例如工具的描述完全依赖于你的文档字符串。如果你的函数逻辑复杂或者你想对输入参数进行更复杂的校验和说明使用StructuredTool类会是更强大的选择。StructuredTool允许你以更声明式、更细致的方式来定义一个工具。你可以明确指定每一个字段from langchain.tools import StructuredTool def get_weather(location: str) - str: return f{location}的天气是晴朗25摄氏度。 weather_tool StructuredTool.from_function( funcget_weather, nameGetCurrentWeather, description查询指定城市的当前天气情况。输入应为城市名称例如‘北京’或‘New York’。, args_schemaNone, # 可以传入一个Pydantic模型来严格定义参数 )这里有几个关键点name工具的名称。这是Agent在“思考”时会看到的标识。起一个清晰、动词开头的名字如GetCurrentWeather、SearchDatabase会帮助LLM更好地理解其用途。description这是重中之重。描述应该清晰、无歧义地说明工具的功能、适用场景以及输入的格式。好的描述是Agent正确使用工具的一半保障。例如“查询天气”就不如“根据完整的城市名称查询当前的温度、湿度和天气状况输入格式如‘北京市’或‘San Francisco, CA’”。func背后实际执行的Python可调用对象。args_schema这是一个高级功能允许你使用Pydantic模型来严格定义参数的名称、类型、描述甚至校验规则。这对于复杂工具至关重要能确保LLM生成的参数符合后端API的期望。实操心得描述Description是你的第一道防线我踩过的第一个坑就是工具描述写得太随意。早期我给一个“用户信息查询”工具的描述是“查询用户数据”。结果Agent在需要找“张三的订单”时也调用了这个工具因为它觉得“订单”也是“用户数据”的一部分。这导致了错误的API调用和混乱的结果。后来我把描述改为“根据用户IDUUID格式查询用户的姓名、注册邮箱和基础档案不包含订单、日志等其他信息。”问题就迎刃而解了。所以花时间打磨工具的描述就像给API写一份优秀的文档能省去后面大量的调试和纠错成本。3. 超越单个函数封装复杂对象与异步工具现实世界的工具 rarely 是一个简单的纯函数。它可能是一个需要初始化的客户端对象一个需要管理状态如数据库连接、会话的类或者是一个异步的协程。LangChain的Tool体系都能很好地支持。场景一封装一个类方法假设我们有一个DatabaseClient类它已经建立了连接我们想暴露它的查询方法作为工具。from langchain.tools import StructuredTool class DatabaseClient: def __init__(self, connection_string): self.conn create_engine(connection_string) def run_query(self, sql: str) - list: 执行SQL查询语句并返回结果列表。 # 执行SQL的逻辑... return results # 初始化客户端 db_client DatabaseClient(your_connection_string) # 将类方法封装为工具 # 注意这里我们需要一个函数来绑定self通常使用lambda或定义一个闭包函数 db_query_tool StructuredTool.from_function( funclambda sql: db_client.run_query(sql), # 绑定实例 nameRunDatabaseQuery, description执行一个只读的SQL SELECT查询语句并返回结果。请确保SQL语法正确。, )这里的关键在于func参数需要的是一个可调用对象。我们通过lambda表达式将实例方法db_client.run_query转换成了一个只需要sql一个参数的函数完美符合Tool的要求。场景二处理异步工具如果你的后端调用是异步的比如使用aiohttp调用异步API那么你的工具函数也应该是一个异步函数。LangChain完全支持异步工具。import aiohttp from langchain.tools import StructuredTool async def async_fetch_data(api_endpoint: str, params: dict) - dict: 异步调用指定的API端点获取数据。 async with aiohttp.ClientSession() as session: async with session.get(api_endpoint, paramsparams) as response: return await response.json() async_api_tool StructuredTool.from_function( funcasync_fetch_data, nameAsyncFetchData, description异步获取数据。参数api_endpoint是URLparams是查询参数字典。, coroutineasync_fetch_data, # 对于异步函数需要显式指定coroutine )当你使用支持异步的Agent如create_react_agent时它可以正确地await这个工具。这能显著提升在I/O密集型场景如同时调用多个外部API下的应用性能。场景三带有复杂参数校验的工具对于输入参数复杂、需要严格校验的工具强烈推荐使用args_schema。这需要借助Pydantic。from pydantic import BaseModel, Field from langchain.tools import StructuredTool class WeatherQueryInput(BaseModel): location: str Field(description城市名称例如‘北京’、‘上海’) unit: str Field(defaultcelsius, description温度单位可选‘celsius’摄氏度或‘fahrenheit’华氏度) def get_weather_advanced(location: str, unit: str celsius) - str: # 实现... pass weather_advanced_tool StructuredTool.from_function( funcget_weather_advanced, nameGetWeatherAdvanced, description查询天气支持选择温度单位。, args_schemaWeatherQueryInput, # 传入Pydantic模型 )这样做的好处是自文档化LLM能清晰地看到每个参数的意义和可选值。输入校验在工具被调用前LangChain会先用Pydantic模型校验参数如果unit的值不是celsius或fahrenheit会提前报错避免无效的API调用。类型安全为整个流程增加了类型提示方便开发和调试。注意使用args_schema时你的工具函数func的参数名必须与Pydantic模型中的字段名完全一致。LangChain会按照模型字段来调用函数。4. 构建工具集Toolkit将散兵游勇组织成特种部队当你拥有了几个相关的工具后比如针对数据库的RunQueryTool、ListTablesTool、GetSchemaTool把它们作为一个整体来管理和使用会更加方便。这就是Toolkit工具集的概念。一个Toolkit本质上就是一个包含多个Tool对象的容器通常还附带一些初始化或共享状态的逻辑。LangChain提供了一些内置的Toolkit如SQLDatabaseToolkit但更多时候我们需要创建自己的。创建自定义Toolkit非常简单通常就是继承BaseToolkit类然后在get_tools方法中返回你的工具列表。为什么需要Toolkit模块化管理将与某个特定领域数据库、文件系统、CRM系统相关的所有工具打包在一起代码结构更清晰。共享状态Toolkit的初始化方法__init__可以用来创建和持有共享资源比如一个数据库连接池。这个连接池可以被它返回的所有工具使用避免了每个工具都独立创建连接的开销和混乱。便于Agent加载很多Agent的初始化方法可以直接接受一个Toolkit对象它会自动提取其中的所有工具比手动传入一个工具列表更简洁。让我们构建一个简单的“文件系统工具箱”from langchain.agents.agent_toolkits.base import BaseToolkit from langchain.tools import Tool import os from typing import List class FileSystemToolkit(BaseToolkit): 一个用于基础文件操作的工具箱。 def __init__(self, base_path: str .): # 可以定义一个基础路径所有文件操作都相对此路径进行增加安全性 self.base_path os.path.abspath(base_path) def get_tools(self) - List[Tool]: # 在这里定义并返回属于这个工具箱的所有工具 return [ self._create_list_files_tool(), self._create_read_file_tool(), self._create_file_stats_tool() ] def _create_list_files_tool(self) - Tool: def list_files(directory: str .) - str: 列出指定目录下的文件和文件夹。 target_path os.path.join(self.base_path, directory) # 安全检查确保目标路径在base_path之下 if not os.path.commonpath([self.base_path, os.path.abspath(target_path)]) self.base_path: return 错误试图访问超出允许范围的路径。 try: items os.listdir(target_path) return \n.join(items) if items else 目录为空。 except FileNotFoundError: return f错误目录 {directory} 不存在。 except PermissionError: return f错误没有权限访问目录 {directory}。 return Tool( nameListDirectoryFiles, funclist_files, description列出指定目录下的内容。输入是目录的相对路径默认为当前目录。 ) def _create_read_file_tool(self) - Tool: # 类似地实现读文件工具这里省略具体函数体 def read_file(filepath: str) - str: 读取文本文件的内容。 full_path os.path.join(self.base_path, filepath) # ... 路径安全检查 ... try: with open(full_path, r, encodingutf-8) as f: return f.read() except Exception as e: return f读取文件失败{str(e)} return Tool( nameReadTextFile, funcread_file, description读取指定文本文件的内容。输入是文件的相对路径。 ) def _create_file_stats_tool(self) - Tool: # 实现获取文件信息的工具 pass # 使用工具箱 fs_toolkit FileSystemToolkit(base_path/safe/workspace) tools fs_toolkit.get_tools() # 获取到三个工具的列表在这个例子中FileSystemToolkit类在初始化时接收一个base_path将其作为文件操作的根目录这是一个重要的安全实践。所有工具方法在执行实际操作前都会检查目标路径是否在base_path之下防止路径穿越攻击。这种共享的安全策略在Toolkit层面统一管理比在每个工具里重复实现要优雅和可靠得多。实操心得在Toolkit初始化中做“脏活累活”把那些繁琐但必要的初始化、配置加载、资源创建如数据库连接、API客户端实例化、认证令牌获取放在Toolkit的__init__方法里。这样当你把Toolkit交给Agent时这些依赖已经准备就绪。例如我的“内部API工具箱”会在__init__里读取配置文件初始化一个带重试和熔断机制的HTTP客户端并获取一个OAuth 2.0令牌。这个客户端实例会被所有工具方法共享既保证了效率复用连接和令牌也统一了错误处理逻辑。5. 高级模式动态工具与元工具在一些更复杂的智能体场景中我们需要的工具可能不是静态的而是根据上下文动态生成或变化的。LangChain的框架也支持这种动态性。动态工具生成假设你有一个工具它的功能取决于运行时的某些状态。例如一个“切换数据源”的工具它执行后会改变另一个“查询数据”工具背后的连接。我们可以通过让工具访问和修改共享状态来实现。from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type class DynamicDataSourceContext(BaseModel): current_source: str default class QueryDataTool(BaseTool): name QueryData description 从当前数据源查询数据。 args_schema: Type[BaseModel] None # 简单工具不需要复杂schema context: DynamicDataSourceContext # 持有共享状态的引用 def _run(self, query: str) - str: # 根据 self.context.current_source 决定查询哪个数据源 if self.context.current_source source_a: return f从源A查询 {query} 的结果。 elif self.context.current_source source_b: return f从源B查询 {query} 的结果。 else: return f从默认源查询 {query} 的结果。 def _arun(self, query: str): raise NotImplementedError(异步不支持) class SwitchDataSourceTool(BaseTool): name SwitchDataSource description 切换当前使用的数据源。 args_schema: Type[BaseModel] None context: DynamicDataSourceContext def _run(self, source_name: str) - str: if source_name in [source_a, source_b, default]: self.context.current_source source_name return f已切换数据源至{source_name} else: return f错误未知数据源 {source_name} def _arun(self, source_name: str): raise NotImplementedError(异步不支持) # 使用 shared_context DynamicDataSourceContext() tool1 QueryDataTool(contextshared_context) tool2 SwitchDataSourceTool(contextshared_context)这里两个工具通过共享同一个context对象实现了状态的联动。当Agent使用SwitchDataSourceTool切换源后后续的QueryDataTool调用就会使用新的数据源。元工具Meta-Tool元工具是指那些不直接完成最终任务而是用于管理、评估或组合其他工具的工具。一个经典的例子是“Human-in-the-loop”工具它暂停Agent的执行将问题抛给人类用户并将用户的输入返回给Agent。from langchain.tools import BaseTool from langchain.callbacks.manager import CallbackManagerForToolRun class HumanInputTool(BaseTool): name AskHuman description 当你需要澄清一个模糊的问题、获取额外信息或确认一个关键操作时使用此工具向人类用户提问。 def _run(self, question: str, run_manager: CallbackManagerForToolRun | None None) - str: # 在实际应用中这里可以通过前端界面、消息队列等方式将问题呈现给用户 # 这里我们模拟一个简单的命令行输入 print(f\n[Agent需要您的输入]: {question}) user_response input(您的回答: ) return user_response将这个工具放入Agent的工具列表后Agent在遇到不确定的情况时就可以主动向人类求助。这极大地增加了复杂、高风险任务中Agent的可靠性和安全性。其他元工具的思路还包括“评估当前计划可行性”、“总结已执行步骤”等它们帮助Agent进行更高层次的规划和反思。6. 实战封装一个企业内部知识库查询工具集现在让我们综合以上所有知识完成一个更贴近真实业务的实战封装一个用于查询企业内部知识库假设是一个Elasticsearch集群的工具集。这个工具集需要包含1. 基础检索工具2. 高级过滤工具3. 获取文档详情的工具。第一步设计共享客户端与基础工具我们首先创建一个KnowledgeBaseToolkit它在初始化时建立Elasticsearch客户端连接。from elasticsearch import Elasticsearch from langchain.agents.agent_toolkits.base import BaseToolkit from langchain.tools import StructuredTool from pydantic import BaseModel, Field from typing import List, Optional import json class KnowledgeBaseToolkit(BaseToolkit): def __init__(self, es_hosts: List[str], index_name: str company-wiki): self.es_client Elasticsearch(es_hosts) self.index_name index_name # 可以在这里进行连接健康检查等初始化操作 def get_tools(self): return [ self._create_basic_search_tool(), self._create_advanced_search_tool(), self._create_get_document_tool() ] def _create_basic_search_tool(self): class BasicSearchInput(BaseModel): query: str Field(description搜索关键词例如‘年假政策’或‘项目报销流程’) size: int Field(default5, description返回结果的最大数量默认5条) def basic_search(query: str, size: int 5) - str: 在知识库中执行基础全文搜索。 body { query: { multi_match: { query: query, fields: [title^2, content] # 标题权重更高 } }, size: size } try: response self.es_client.search(indexself.index_name, bodybody) hits response[hits][hits] if not hits: return 未找到相关文档。 results [] for hit in hits: source hit[_source] results.append(f标题{source.get(title, N/A)}\n摘要{source.get(summary, N/A)[:100]}...\nID{hit[_id]}\n) return \n---\n.join(results) except Exception as e: return f搜索过程中发生错误{str(e)} return StructuredTool.from_function( funcbasic_search, nameBasicKnowledgeSearch, description根据关键词在知识库中进行全文搜索。返回匹配文档的标题、摘要和ID。, args_schemaBasicSearchInput )第二步实现带过滤的高级搜索工具高级搜索工具允许用户通过分类、标签、创建时间等字段进行过滤。def _create_advanced_search_tool(self): class AdvancedSearchInput(BaseModel): query: Optional[str] Field(defaultNone, description可选的关键词搜索) category: Optional[str] Field(defaultNone, description按分类过滤如‘HR’、‘IT’) tags: Optional[List[str]] Field(defaultNone, description按标签过滤如[‘urgent’ ‘policy’]) from_date: Optional[str] Field(defaultNone, description起始日期格式YYYY-MM-DD) def advanced_search(query: Optional[str] None, category: Optional[str] None, tags: Optional[List[str]] None, from_date: Optional[str] None) - str: 执行带过滤条件的高级搜索。 must_conditions [] if query: must_conditions.append({multi_match: {query: query, fields: [title, content]}}) if category: must_conditions.append({term: {category.keyword: category}}) if tags: must_conditions.append({terms: {tags.keyword: tags}}) if from_date: must_conditions.append({range: {created_at: {gte: from_date}}}) body { query: { bool: { must: must_conditions if must_conditions else [{match_all: {}}] } }, size: 10, sort: [{created_at: {order: desc}}] # 按时间倒序 } # ... 执行搜索并格式化结果的代码 ... # 返回格式化的字符串 return formatted_results return StructuredTool.from_function( funcadvanced_search, nameAdvancedKnowledgeSearch, description使用关键词、分类、标签、日期范围等条件组合搜索知识库文档。, args_schemaAdvancedSearchInput )第三步实现获取文档详情的工具当Agent通过搜索工具找到感兴趣的文档ID后可以用这个工具获取全文。def _create_get_document_tool(self): class GetDocInput(BaseModel): doc_id: str Field(description知识库文档的唯一ID) def get_document(doc_id: str) - str: 根据文档ID获取文档的完整内容。 try: response self.es_client.get(indexself.index_name, iddoc_id) source response[_source] # 格式化输出突出重点信息 output f文档ID{doc_id}\n output f标题{source.get(title, N/A)}\n output f分类{source.get(category, N/A)}\n output f标签{, .join(source.get(tags, []))}\n output f创建时间{source.get(created_at, N/A)}\n output --- 内容 ---\n output source.get(content, 内容为空) return output except Exception as e: return f获取文档失败ID ‘{doc_id}’{str(e)} return StructuredTool.from_function( funcget_document, nameGetKnowledgeDocument, description根据文档ID获取知识库中某篇文档的完整详细信息包括标题、分类、标签和全文内容。, args_schemaGetDocInput )整合与使用现在我们可以初始化这个工具箱并将其提供给一个Agent# 初始化工具箱 kb_toolkit KnowledgeBaseToolkit( es_hosts[http://localhost:9200], index_namecompany-wiki-prod ) kb_tools kb_toolkit.get_tools() # 假设我们有一个LLM和Agent执行器 from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4, temperature0) agent create_react_agent(llm, kb_tools, verboseTrue) agent_executor AgentExecutor(agentagent, toolskb_tools, verboseTrue) # 现在Agent就可以使用这些工具了 result agent_executor.invoke({ input: 帮我找一下关于2024年新员工入职培训的IT部门相关文档要最近三个月发布的。 })在这个查询中Agent可能会先使用AdvancedKnowledgeSearch工具设置query为“新员工入职培训”category为“IT”from_date为三个月前的日期。在得到一批结果包含文档ID后如果用户需要查看某篇的具体内容Agent可以再调用GetKnowledgeDocument工具。避坑经验工具描述的协同与防冲突当工具集内工具变多时要特别注意工具描述的区分度。如果BasicKnowledgeSearch和AdvancedKnowledgeSearch的描述过于相似Agent可能会困惑该用哪个。我的经验是在描述中明确各自的使用场景和输入特点。例如基础搜索“快速根据一两个关键词查找相关文档。”高级搜索“当需要结合多个条件如分类、标签、时间进行精确筛选时使用。”此外工具的名称也要有区分度避免使用Search和SearchDocument这样容易混淆的名字。7. 调试、测试与最佳实践封装好工具后在集成到复杂的Agent流程前进行充分的独立测试是至关重要的。这能帮你快速定位是工具逻辑问题还是Agent的理解和调用问题。单元测试你的工具像测试普通Python函数一样测试你的工具。确保在各种边界条件下如空输入、错误参数、网络异常工具都能返回合理的、结构化的错误信息而不是抛出未处理的异常。# 假设测试上面的 basic_search 工具 def test_basic_search(): toolkit KnowledgeBaseToolkit(es_hosts[dummy], index_nametest) tools toolkit.get_tools() search_tool [t for t in tools if t.name BasicKnowledgeSearch][0] # 测试正常查询 result search_tool.run({query: 测试, size: 3}) assert isinstance(result, str) assert 标题 in result or 未找到 in result # 测试空结果 # 可以通过mock Elasticsearch客户端来模拟返回空结果 # ... # 测试异常处理 # 模拟客户端抛出异常检查返回的字符串是否包含友好的错误信息 print(工具基础测试通过。)在LangChain中直接测试工具调用使用Tool.invoke()或直接调用func来验证工具是否能被正确触发并返回预期结果。# 获取工具并直接调用 weather_tool get_weather_advanced_tool # 假设这是之前定义的工具 try: output weather_tool.invoke({location: 北京, unit: celsius}) print(f工具调用成功输出{output}) except Exception as e: print(f工具调用失败{e})集成到Agent前的“沙盒”测试创建一个最简单的Agent只包含你新开发的工具用一些典型的用户问题去测试它。观察Agent的思考链通过设置verboseTrue看它是否正确地选择了你期望的工具并且生成的参数是否符合args_schema的预期。from langchain.agents import initialize_agent, AgentType from langchain_openai import ChatOpenAI llm ChatOpenAI(temperature0) test_agent initialize_agent( tools[weather_tool], # 只放一个工具进行测试 llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, # 使用一个简单的Agent类型 verboseTrue, handle_parsing_errorsTrue # 处理解析错误 ) result test_agent.run(上海今天天气怎么样) print(result)通过verboseTrue的输出你可以清晰地看到Agent的“Thought”思考过程它是否理解了问题并决定调用GetWeatherAdvanced“Action”和“Action Input”它生成的调用动作和参数是否正确例如Action Input应该是{location: 上海, unit: celsius}这样的JSON。“Observation”你的工具返回的结果是什么。“Final Answer”Agent最终给用户的回答。最佳实践清单根据我的经验封装一个健壮、易用的工具集请牢记以下几点描述即契约花80%的精力打磨工具的名称和描述。确保它们清晰、无歧义、能自我解释。好的描述能极大降低Agent的误用率。输入验证与净化永远不要相信LLM生成的输入是完美无缺的。在工具函数内部对输入参数进行类型转换、范围检查、敏感词过滤等。对于文件路径、数据库查询语句等要严防注入攻击。优雅的错误处理工具函数必须捕获所有可能的异常并返回一个对LLM和最终用户都有意义的字符串错误信息。避免抛出未处理的异常导致整个Agent崩溃。返回的信息应能帮助Agent进行后续决策例如“未找到文件请检查路径” vs. “FileNotFoundError”。保持工具功能单一一个工具只做一件事。不要创建一个“万能”工具。单一职责的工具更容易被LLM理解、组合和调试。为工具输出设计格式工具的输出是给LLM看的“观察”。设计清晰、结构化的输出格式如使用\n分隔不同字段使用---分隔不同条目有助于LLM解析并生成更好的最终答案。性能考量如果工具涉及网络请求或复杂计算考虑增加超时、重试和缓存机制。对于频繁调用的工具缓存可以显著提升Agent的响应速度。版本控制与文档像管理API一样管理你的工具集。当工具接口参数、返回值发生变化时要有明确的版本更迭意识并更新对应的描述和文档。封装自己的LangChain工具和工具集是将LLM从“聪明的聊天者”转变为“能干的业务助手”的关键一步。这个过程始于对业务逻辑的深刻理解成于对接口设计的细致打磨。当你看到Agent自如地调用着你亲手封装的服务并组合它们解决复杂问题时那种成就感正是驱动我们不断深入探索AI应用落地的动力。