工具 API

工具定义与 StructuredTool

概述

工具是 Agent 可以调用的函数。LangChain 提供了多种方式来定义和使用工具,包括装饰器、类定义和运行时绑定。

graph TD
    A[BaseTool] --> B[StructuredTool]
    A --> C[Tool]
    A --> D[InjectTool]

    E[@tool 装饰器] --> F[自动生成工具]
    F --> B

    B --> G[Agent 可调用]
    C --> G

    style A fill:#e1f5fe
    style B fill:#c8e6c9

@tool 装饰器

tool

将函数转换为 LangChain 工具的装饰器。

from langchain_core.tools import tool

def tool(
    *args: Union[str, Callable],
    name: Optional[str] = None,
    description: Optional[str] = None,
    return_direct: bool = False,
    args_schema: Optional[Type[BaseModel]] = None,
    infer_schema: bool = True,
) -> Union[Tool, Callable[..., Tool]]:
    """
    工具装饰器

    Args:
        *args: 函数或字符串
            - 如果第一个参数是函数,则装饰该函数
            - 如果第一个参数是字符串,则作为工具名称
        name: 工具名称(默认使用函数名)
        description: 工具描述(默认使用函数文档字符串)
        return_direct: 是否直接返回结果
        args_schema: 参数模式(Pydantic 模型)
        infer_schema: 是否自动推断参数模式

    Returns:
        Tool 实例或装饰器

    Example:
        @tool
        def search(query: str) -> str:
            '''搜索网络'''
            return "搜索结果"

        @tool("custom_name", description="自定义描述")
        def my_function(x: int) -> int:
            return x * 2
    """

使用示例

python
from langchain_core.tools import tool

# 基础使用
@tool
def search(query: str) -> str:
    """搜索网络"""
    return f"关于 {query} 的搜索结果"

# 自定义名称和描述
@tool("calculate", description="执行数学计算")
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# 带参数验证
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")
    max_results: int = Field(default=5, description="最大结果数")

@tool("search_advanced", args_schema=SearchInput)
def advanced_search(query: str, max_results: int = 5) -> str:
    """高级搜索"""
    return f"找到 {max_results} 条关于 {query} 的结果"

# 直接返回
@tool(return_direct=True)
def final_answer(answer: str) -> str:
    """返回最终答案"""
    return answer

# 查看工具信息
print(search.name)        # "search"
print(search.description)  # "搜索网络"
print(search.args_schema)  # 参数模式

structured_tool

创建结构化工具的装饰器。

from langchain_core.tools import structured_tool

def structured_tool(
    *args: Union[str, Callable],
    name: Optional[str] = None,
    description: Optional[str] = None,
    return_direct: bool = False,
) -> Union[StructuredTool, Callable[..., StructuredTool]]:
    """
    结构化工具装饰器

    与 tool 的区别在于总是返回 StructuredTool,
    无论函数参数数量如何

    Args:
        *args: 函数或字符串
        name: 工具名称
        description: 工具描述
        return_direct: 是否直接返回

    Returns:
        StructuredTool 实例或装饰器
    """

核心类

BaseTool

所有工具的抽象基类。

from langchain_core.tools import BaseTool

class BaseTool(Serializable, Runnable[ToolInput, ToolOutput]):
    """工具基类"""

    name: str = ""
    """工具名称(在 Agent 调用时使用)"""

    description: str = ""
    """工具描述(传递给 LLM)"""

    args_schema: Type[BaseModel] = BaseModel
    """参数验证模式"""

    return_direct: bool = False
    """是否直接返回结果而不继续推理"""

    # 核心方法
    @abstractmethod
    def _run(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """
        同步运行工具

        Args:
            *args: 位置参数
            **kwargs: 关键字参数

        Returns:
            工具执行结果
        """

    async def _arun(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """
        异步运行工具(可选实现)

        默认调用 _run
        """
        return await asyncio.to_thread(self._run, *args, **kwargs)

    # Runnable 接口实现
    def invoke(
        self,
        input: Union[str, Dict[str, Any]],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Any:
        """
        调用工具(Runnable 接口)

        Args:
            input: 工具输入(字符串或字典)
            config: 运行配置
            **kwargs: 额外参数

        Returns:
            工具结果
        """

    def run(
        self,
        tool_input: Union[str, Dict[str, Any]],
        verbose: Optional[bool] = None,
        start_color: Optional[str] = "green",
        color: Optional[str] = "green",
        **kwargs: Any,
    ) -> Any:
        """
        运行工具(兼容方法)

        Args:
            tool_input: 工具输入
            verbose: 是否打印详细日志
            start_color: 日志颜色
            color: 输出颜色
            **kwargs: 额外参数

        Returns:
            工具结果
        """

创建自定义工具

python
from langchain_core.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")

class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "自定义搜索工具"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        """同步执行"""
        return f"搜索结果: {query}"

    async def _arun(self, query: str) -> str:
        """异步执行"""
        return f"异步搜索结果: {query}"

# 使用
tool = CustomSearchTool()
result = tool.invoke({"query": "Python"})

StructuredTool

支持结构化输入的工具。

from langchain_core.tools import StructuredTool

class StructuredTool(BaseTool):
    """结构化工具"""

    func: Union[
        Callable[..., Any],
        Callable[..., Awaitable[Any]]
    ]
    """底层函数"""

    coroutine: Optional[Callable[..., Awaitable[Any]]] = None
    """异步函数(如果与 func 不同)"""

    # 构造方法
    def __init__(
        self,
        name: str,
        description: str,
        func: Callable[..., Any],
        args_schema: Type[BaseModel],
        return_direct: bool = False,
        coroutine: Optional[Callable[..., Awaitable[Any]]] = None,
        **kwargs: Any,
    ):
        """
        初始化结构化工具

        Args:
            name: 工具名称
            description: 工具描述
            func: 同步函数
            args_schema: 参数模式
            return_direct: 是否直接返回
            coroutine: 异步函数(可选)
            **kwargs: 额外参数
        """

创建方式

python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式")

def calculate(expression: str) -> str:
    """计算表达式"""
    try:
        return str(eval(expression))
    except:
        return "错误"

# 方式1: 直接创建
calculator = StructuredTool.from_function(
    func=calculate,
    name="calculator",
    description="执行数学计算",
    args_schema=CalculatorInput
)

# 方式2: 使用 from_function 类方法
calculator = StructuredTool.from_function(
    calculate,
    name="calculator",
    description="执行数学计算",
    args_schema=CalculatorInput
)

# 方式3: 使用装饰器
from langchain_core.tools import tool

@tool
def calculator_tool(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

Tool

简单工具,接受单个字符串输入。

from langchain_core.tools import Tool

class Tool(BaseTool):
    """简单工具(单字符串输入)"""

    func: Callable[[str], Any]
    """工具函数"""

    # 构造方法
    def __init__(
        self,
        name: str,
        func: Callable[[str], Any],
        description: str,
        return_direct: bool = False,
        **kwargs: Any,
    ):
        """
        初始化简单工具

        Args:
            name: 工具名称
            func: 工具函数(接受单个字符串)
            description: 工具描述
            return_direct: 是否直接返回
            **kwargs: 额外参数
        """

创建方式

python
from langchain_core.tools import Tool

# 方式1: 直接实例化
search_tool = Tool(
    name="search",
    func=lambda query: f"搜索结果: {query}",
    description="搜索网络"
)

# 方式2: 使用 from_function
def search_func(query: str) -> str:
    return f"搜索结果: {query}"

search_tool = Tool.from_function(
    func=search_func,
    name="search",
    description="搜索网络"
)

# 方式3: 使用装饰器
@tool
def search(query: str) -> str:
    """搜索网络"""
    return f"搜索结果: {query}"

InjectTool

注入工具,直接返回预设值而不执行。

from langchain_core.tools import InjectTool

class InjectTool(BaseTool):
    """注入工具"""

    value: Any
    """要返回的预设值"""

    def _run(self, *args: Any, **kwargs: Any) -> Any:
        """直接返回预设值"""
        return self.value

ToolInvocation

工具调用记录。

from langchain_core.tools import ToolInvocation

class ToolInvocation(TypedDict):
    """工具调用"""

    name: str
    """工具名称"""

    id: Optional[str]
    """调用 ID"""

    args: Dict[str, Any]
    """工具参数"""

tool 装饰器函数

便捷的函数式工具创建。

from langchain_core.tools import tool

# 无参数
@tool
def simple_tool(input: str) -> str:
    """简单工具"""
    return input.upper()

# 带名称
@tool("my_tool")
def named_tool(input: str) -> str:
    """命名工具"""
    return input.upper()

# 带描述
@tool(description="这是一个自定义工具")
def described_tool(input: str) -> str:
    """带描述的工具"""
    return input.upper()

# 完整参数
@tool(
    name="complete_tool",
    description="完整的工具定义",
    return_direct=True
)
def complete(input: str) -> str:
    """完整工具"""
    return input.upper()

# 使用参数模式
from pydantic import BaseModel

class InputSchema(BaseModel):
    text: str
    times: int = 1

@tool(args_schema=InputSchema)
def complex_tool(text: str, times: int = 1) -> str:
    """复杂工具"""
    return text * times

使用示例

python
from langchain_core.tools import tool, Tool, StructuredTool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# ========== 示例1: 使用 @tool 装饰器 ==========

@tool
def search(query: str) -> str:
    """搜索网络获取信息"""
    return f"关于 {query} 的搜索结果"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# ========== 示例2: 自定义工具类 ==========

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位")

class WeatherTool(BaseTool):
    name = "get_weather"
    description = "获取指定城市的天气"
    args_schema = WeatherInput

    def _run(self, city: str, unit: str = "celsius") -> str:
        return f"{city} 今天是晴天,温度 25°{unit}"

    async def _arun(self, city: str, unit: str = "celsius") -> str:
        # 模拟 API 调用
        return f"{city}(异步查询)今天晴天,25°{unit}"

weather = WeatherTool()

# ========== 示例3: StructuredTool ==========

from pydantic import BaseModel

class EmailInput(BaseModel):
    to: str = Field(description="收件人邮箱")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")

def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    # 实际实现会调用邮件 API
    return f"邮件已发送给 {to}"

email_tool = StructuredTool.from_function(
    func=send_email,
    name="send_email",
    description="发送邮件",
    args_schema=EmailInput
)

# ========== 示例4: Tool(简单输入) ==========

simple_search = Tool(
    name="simple_search",
    func=lambda q: f"搜索 {q}",
    description="简单搜索工具"
)

# ========== 示例5: 与 Agent 集成 ==========

tools = [search, calculator, weather, email_tool]

llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

result = agent_executor.invoke({
    "input": "北京天气怎么样?顺便计算 25 * 4"
})

# ========== 示例6: 异步工具 ==========

import asyncio

@tool
async def async_search(query: str) -> str:
    """异步搜索"""
    await asyncio.sleep(0.1)
    return f"异步搜索结果: {query}"

# ========== 示例7: 直接调用工具 ==========

# 方式1: invoke
result = search.invoke("Python")

# 方式2: 直接调用
result = search.func("Python")

# 方式3: run
result = search.run("Python")

内置工具

模块 工具 描述 langchain_community.tools Shell 执行 shell 命令 langchain_community.tools PythonREPL 执行 Python 代码 langchain_community.tools RequestsGetTool HTTP GET 请求 langchain_community.tools RequestsPostTool HTTP POST 请求 langchain_community.tools SerpAPIWrapper Google 搜索 langchain_community.tools WikipediaQueryRun Wikipedia 查询 langchain_community.tools Shell 执行命令

相关 API