LangChain Agent 从入门到精通:从核心原理到生产级落地

大家好,我是深耕大模型应用开发的技术博主。如果说大模型是智能体的 “大脑”,那Tool(工具)就是智能体的 “双手”—— 它解决了大模型知识滞后、不会精确计算、无法操作外部系统的核心痛点,是 Agent 能力落地的关键基石。

很多同学对 Tool 的理解停留在 “写个函数加个装饰器”,但实际生产中,工具的定义规范、调用机制、错误处理、安全管控才是决定 Agent 稳定性的核心。本文从底层原理、定义方式、调用机制、高级用法、踩坑优化五个维度,系统拆解 LangChain 中的 Tool 技术体系,所有代码适配 LangChain 0.2 + 新版本。

一、Tool 核心概念:大模型的外部能力接口

1.1 什么是 Tool

本质上,Tool 就是大模型可调用的外部能力函数。大模型通过自然语言理解用户需求后,自主判断是否需要调用工具、调用哪个工具、传入什么参数,拿到工具执行结果后再整合生成最终答案。

工具的核心价值:

  • 弥补大模型的固有缺陷:实时信息获取、精确数学计算、结构化数据操作
  • 扩展大模型的能力边界:操作数据库、调用 API、执行代码、控制硬件
  • 实现业务闭环:从 “问答” 升级为 “执行任务”

1.2 LangChain 中 Tool 的核心抽象层级

LangChain 对工具做了多层抽象,从底层到上层依次是:

表格

抽象类定位适用场景
BaseTool所有工具的基类,定义了_run_arun核心方法需要完全自定义逻辑、维护内部状态的复杂工具
StructuredTool支持结构化多参数输入的工具基类多参数、有严格入参校验的业务工具
Tool/@tool装饰器最简工具封装,默认单字符串入参简单功能、快速原型开发

新手最容易踩的坑:默认@tool装饰器只接收一个字符串参数,多参数场景必须用结构化工具定义,否则会出现参数传递错乱。

1.3 工具调用的两种底层模式

LangChain 的工具调用本质分两大流派,直接决定了 Agent 的稳定性:

  1. ReAct 提示词模式:通过提示词引导大模型用固定格式输出工具名和参数,靠正则解析提取。优点是兼容所有大模型,缺点是格式容易出错、稳定性差。
  2. 原生 Function Calling 模式:依赖大模型本身的函数调用能力(如 GPT 系列、文心一言、通义千问等),模型原生输出结构化的工具调用指令。优点是准确率高、格式稳定,是当前生产环境的主流方案。

LangChain 0.1 之后主推的create_tool_calling_agent,就是基于原生函数调用实现的。

二、三种工具定义方式:从简单到复杂

2.1 最简方式:@tool 装饰器(最常用)

这是最便捷的工具定义方式,适合单参数、逻辑简单的工具。只需要给函数加上@tool装饰器,函数名会成为工具名,函数文档字符串会成为工具描述。

from langchain_core.tools import tool # 定义一个简单的计算器工具 @tool def simple_calculate(expression: str) -> str: """ 执行基础算术表达式计算,仅支持加减乘除和括号运算。 输入必须是合法的Python算术表达式字符串,例如 "123 + 456 * 2" """ try: result = eval(expression) return f"计算结果:{result}" except Exception as e: return f"计算失败:{str(e)}" # 查看工具信息 print(f"工具名:{simple_calculate.name}") print(f"工具描述:{simple_calculate.description}") print(f"入参Schema:{simple_calculate.args}")

常用参数说明

  • name:手动指定工具名,默认取函数名
  • description:手动指定工具描述,优先级高于函数文档
  • return_direct:设为 True 时,工具执行结果直接返回给用户,不再经过大模型二次处理,适合结果确定的场景

2.2 结构化输入:Pydantic + StructuredTool

当工具需要多个参数、且有严格的类型校验时,必须用 Pydantic 定义入参模型,配合StructuredTool使用。这是生产环境的标准写法。

from langchain_core.tools import StructuredTool from langchain_core.pydantic_v1 import BaseModel, Field # 1. 定义入参Schema,带字段说明 class WeatherQueryInput(BaseModel): city: str = Field(description="要查询的城市名称,必须是中文城市名,例如:北京、上海") date: str = Field(description="查询日期,格式为YYYY-MM-DD,例如2024-05-20", default="今天") # 2. 实现工具核心逻辑 def get_weather(city: str, date: str) -> str: """查询指定城市指定日期的天气信息""" # 这里模拟调用天气API return f"{date} {city}的天气:晴,温度22~28℃,风力3级,空气质量优" # 3. 创建结构化工具 weather_tool = StructuredTool.from_function( func=get_weather, name="query_weather", description="查询国内城市的天气信息,支持指定日期", args_schema=WeatherQueryInput )

这种方式的优势:

  • 大模型能清晰识别每个参数的含义和格式要求,调用准确率大幅提升
  • 自带参数类型校验,非法参数会直接拦截,避免工具执行报错
  • 支持默认值、枚举值、参数约束,适配复杂业务场景

2.3 完全自定义:继承 BaseTool

当工具需要维护内部状态、初始化配置、实现异步逻辑时,直接继承BaseTool是最灵活的方式。

from langchain_core.tools import BaseTool from typing import Optional, Type from langchain_core.pydantic_v1 import BaseModel class DatabaseQueryTool(BaseTool): # 工具基础信息 name = "database_query" description = "执行业务数据库的只读SQL查询,仅支持SELECT语句,返回查询结果" args_schema: Type[BaseModel] = WeatherQueryInput # 自定义入参Schema # 自定义属性:数据库连接 db_config: dict = {} def __init__(self, db_config: dict): super().__init__() self.db_config = db_config # 初始化数据库连接... def _run(self, sql: str) -> str: """同步执行逻辑,必须实现""" # 执行SQL查询,返回结果 return f"SQL执行结果:xxx" async def _arun(self, sql: str) -> str: """异步执行逻辑,可选实现""" # 异步执行SQL return f"异步执行结果:xxx"

三、工具调用核心机制:让大模型精准调用

3.1 工具描述的黄金法则

工具调用准确率 90% 取决于描述写得好不好。写工具描述请遵循三个原则:

  1. 明确适用场景:说清楚 “什么时候该用这个工具”,比如 “当用户问天气相关问题时使用”
  2. 明确输入要求:说明参数格式、约束条件,比如 “城市名必须是中文,SQL 必须是 SELECT 语句”
  3. 明确输出内容:说明工具返回什么结果,避免大模型对返回值产生误解

反面教材:"计算工具"—— 大模型完全不知道什么时候用、怎么传参 正面教材:"执行精确的数学算术计算,当用户需要计算数值、公式运算时使用。输入为合法的算术表达式字符串,不支持文字描述的计算需求"

3.2 bind_tools:给大模型绑定工具

LangChain 提供了bind_tools方法,一键把工具列表转换成大模型支持的函数调用格式,自动处理不同厂商的格式差异。

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 给大模型绑定工具 llm_with_tools = llm.bind_tools([simple_calculate, weather_tool]) # 调用大模型,它会自主判断是否调用工具 response = llm_with_tools.invoke("北京明天天气怎么样?") print(response.tool_calls) # 查看大模型生成的工具调用指令

输出的tool_calls包含工具名、参数、调用 ID,是结构化的标准格式,不会出现格式错乱。

3.3 工具调用的完整执行流程

一次完整的 Agent 工具调用分为 5 步:

  1. 推理:大模型接收用户问题,结合工具列表,判断是否需要调用工具
  2. 解析:提取工具名称、调用参数,生成工具调用指令
  3. 执行:调用对应工具函数,传入参数,获取执行结果
  4. 回填:把工具执行结果回填到对话上下文中
  5. 生成:大模型结合工具结果,生成最终回答

这个流程在AgentExecutor中被自动循环执行,直到大模型认为任务完成。

四、常用内置与第三方工具盘点

LangChain 生态内置了大量现成工具,不用重复造轮子,重点介绍几个高频使用的:

4.1 代码执行工具

from langchain_community.tools.python.tool import PythonREPLTool # Python代码执行工具,可以执行任意Python代码 python_tool = PythonREPLTool() # 注意:生产环境慎用,存在代码注入风险,必须加沙箱和权限控制

4.2 搜索引擎工具

最常用的实时信息获取工具,推荐 Tavily(专为大模型优化的搜索 API):

from langchain_community.tools.tavily_search import TavilySearchResults # 需先安装:pip install tavily-python,配置TAVILY_API_KEY search_tool = TavilySearchResults(max_results=3)

4.3 文件系统工具

from langchain_community.tools.file_management import ReadFileTool, WriteFileTool # 文件读写工具,可指定工作目录 read_tool = ReadFileTool() write_tool = WriteFileTool()

五、高级进阶:生产级工具的必备能力

5.1 工具错误处理与重试

工具执行失败是常态,原生 Agent 遇到错误直接中断,生产环境必须做异常兜底。

from langchain_core.tools import tool import functools import time def tool_retry(max_retries=3, delay=1): """工具重试装饰器""" def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if i == max_retries - 1: return f"工具执行失败,错误信息:{str(e)}" time.sleep(delay) return wrapper return decorator @tool @tool_retry(max_retries=3) def stable_api_call(param: str) -> str: """带重试机制的外部API调用工具""" # 调用外部API return "调用成功"

5.2 return_direct 直接返回结果

对于结果确定、不需要大模型二次加工的工具,开启return_direct可以减少一次大模型调用,降低延迟和成本。

@tool(return_direct=True) def get_system_status(query: str) -> str: """查询系统运行状态,结果直接返回给用户""" return "系统当前运行正常,CPU使用率30%,内存使用率45%"

5.3 带状态的工具

工具可以维护内部状态,比如会话级别的用户信息、连接池等,通过自定义类工具实现。 典型场景:数据库连接池、用户身份校验、会话级缓存。

5.4 工具权限管控

生产环境必须对工具做权限分级,不是所有用户都能调用所有工具。可以在工具执行前加入权限校验逻辑,根据用户角色判断是否允许调用。

六、完整实战:搭建多工具智能 Agent

下面给出一个可直接运行的完整示例,整合自定义工具 + Agent 执行器:

from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor # 1. 定义工具 @tool def calculate(expression: str) -> str: """执行精确的数学计算,输入为合法的算术表达式字符串""" try: return str(eval(expression)) except: return "计算表达式错误" @tool def get_current_time(format: str) -> str: """ 获取当前系统时间 参数format: 时间格式,可选值为 '12小时制' 或 '24小时制' """ from datetime import datetime if format == "12小时制": return datetime.now().strftime("%Y-%m-%d %I:%M:%S %p") return datetime.now().strftime("%Y-%m-%d %H:%M:%S") tools = [calculate, get_current_time] # 2. 初始化大模型 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 3. 定义Agent提示词 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个可靠的智能助手,遇到计算和时间相关问题请调用工具,不要自己编造结果。"), ("human", "{input}"), ("agent_scratchpad", "{agent_scratchpad}") ]) # 4. 创建Agent并执行 agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 打印执行过程,调试用 max_iterations=5, # 最大迭代次数,防止死循环 handle_parsing_errors=True # 自动处理解析错误 ) # 5. 运行测试 result = agent_executor.invoke({"input": "现在的时间用24小时制是多少?再帮我算一下1234 * 5678等于多少"}) print("最终答案:", result["output"])

七、高频踩坑与优化指南

7.1 工具调用准确率低?多半是描述写废了

  • 工具名要见名知意,不要用缩写和模糊命名
  • 描述里必须写清楚 “适用场景 + 输入要求 + 返回内容” 三要素
  • 相似功能的工具不要定义太多,会让大模型混淆,优先合并成一个多参数工具

7.2 参数传递错误?结构化输入是解药

  • 超过 1 个参数的工具,必须用 Pydantic 定义入参 Schema
  • 每个字段都要加Field描述,说明参数含义和格式
  • 复杂参数(如日期、JSON)要明确指定格式示例

7.3 工具执行超时?加超时控制

  • 外部 API、代码执行类工具必须加超时限制,避免阻塞整个 Agent
  • 可以用timeout-decorator库给工具函数加超时装饰器

7.4 安全红线:避免任意代码执行

  • PythonREPLTool这类代码执行工具,生产环境绝对不能直接暴露给用户
  • 所有用户输入的参数都要做校验和过滤,防止注入攻击
  • 敏感操作类工具必须加二次确认机制

7.5 性能优化:减少无效工具调用

  • 给 Agent 增加任务分类前置判断,简单问题直接回答,不进入工具调用流程
  • 结果确定的工具开启return_direct,节省 Token 和延迟
  • 批量任务优先合并成一次工具调用,减少大模型交互次数

写在最后

Tool 是大模型从 “对话” 走向 “执行” 的核心载体,看似简单的函数封装,背后涉及语义理解、参数解析、错误处理、安全管控等一系列工程问题。原型开发可以靠@tool快速起步,但生产落地一定要做好结构化定义、异常兜底和权限管控。