LangChain 系列文章目录

第一章 LangChain 简介和核心包
第二章 LangChain 提示词工程
第三章 LangChain 工作流
第四章 LangChain服务部署与链路监控
第五章 LangChain 消息管理与聊天历史存储
第六章 LangChain多模态输入与自定义输出
第七章 LangChain 工具

---

---

前言

本文主要整理 LangChain 工具相关内容，包括工具的概念、工具对象属性、自定义工具创建方式、内置工具包、第三方工具，以及如何改造第三方工具或将模型能力封装成工具。

---

LangChain工具

一、工具

1.工具是代理、链或聊天模型（LLM）用来与外界交互的接口。

2.在构建代理时，需要提供一个工具对象列表，以便代理可以使用。

3.工具组件（或属性）

• 工具名称-----name
• 工具功能描述-----description
• 工具输入格式描述-----args_schema
• 工具要调用的函数-----function
• 工具的结果是否应直接返回给用户（仅对代理相关）-----return_direct

4.工具的名称、描述和输入格式会作为上下文提供给LLM，可以让LLM明确如何使用工具。

5.给定一组可用工具和提示时，LLM可能请求调用一个或多个工具，并提供适当的参数。

6.工具包：一组在一起使用，以执行某种特定任务的工具对象列表。

7.模型使用工具要点

• 经过微调以便进行工具调用的聊天模型将比未经微调的模型更擅长工具调用。
• 未经微调的模型可能无法使用工具，特别是复杂的工具或需要多次调用工具。
• 如果工具有合适的名称、描述和输入格式，模型能够更好地使用。
• 简单的工具比复杂的工具更容易让模型使用。

二、创建工具

1.LangChain提供四种方式创建工具对象：

• 使用Tool()方法：定义自定义工具。（已废弃使用）。
• 使用@tool装饰器：定义自定义工具的最简单方式。（建议使用）
• 使用StructuredTool.from_function类方法：类似于@tool装饰器，但允许更多配置和同步和异步实现的规范。（建议使用）
• 使用BaseTool继承类：最灵活的方式，提供了最大程度的控制，但需要更多的工作量和代码。

2.工具对象的属性：

• 属性能够获取到工具的信息：

属性	类型	描述
name	str	工具名，提供给LLM或代理的工具集中必须是唯一的。
description	str	工具功能描述。LLM或代理将使用此描述作为上下文。
args_schema	Pydantic BaseModel	输入描述。可选但建议提供，用于为函数的参数提供更多信息或验证预期参数。
return_direct	boolean	是否直接返回。仅对代理相关。当为True时，在调用给定工具后，代理将停止并将结果直接返回给用户。

3.Tool()方法

• 使用Tool(names, func, description)方法，生成单个的工具对象tool。Tool()等同于Tool.from_function()
  • 参数names为自定义的单个工具名称字符串。
  • 参数func为工具执行的方法。
    • 方法需要有一个输入参数input也就是模型生成的数据
  • 参数description为当需要用到工具时，输入到模型的工具描述。
  • 参数args_schema为工具函数的参数描述，值为使用Pydantic定义的格式类
  • 参数return_direct为是否直接返回给用户
• Tool()官方已明确废弃使用
  • func的方法只支持被传入1个参数，否则报错
• 示例

代码

from langchain.agents import Tool
from pydantic import BaseModel, Field
import datetime

# 有单个参数示例
def square(a: float) -> str:
    """实现一个数字的平方计算"""
    result = a**2
    return f"计算结果：{a} 的平方 = {result}"

class NumbersSchema(BaseModel):
    a: float = Field(description="需要平方的数字")

tool = Tool(
    name="平方计算器",
    func=square,
    args_schema=NumbersSchema,
    description="该工具用于执行一个数字的平方运算，当用户需要计算一个数的平方时调用此工具，必须传入一个数字类型的参数a",
    return_direct=False,
)

test_result = tool.invoke({"a": 10.5})
print(test_result)

# 无参数示例
def current_time(_) -> str:
    """Get the current date and time"""
    res = datetime.datetime.now().strftime("%Y年%m月%d日，%H时%M分%S秒")
    return res

tool2 = Tool(
    name="获取当前时间",
    func=current_time,
    description="用于获取当前的日期和时间",
)

test_result = tool2.invoke({"_": None})
print(test_result)

结果

计算结果：10.5 的平方 = 110.25
2026年01月03日，16时25分13秒

4.@tool装饰器

• @tool装饰器是创建自定义工具的最简单方式。
• @tool装饰器参数
  • @tool装饰器默认函数名作为工具名称，也可通过第一参数定义工具名称
    • 第一参数不能直接使用name作参数名，为位置参数，只能是直接值
  • @tool装饰器使用函数的描述注释docstring作为工具描述，必须提供注释。
    • description参数可以代替函数的描述注释
  • @tool装饰器的函数的参数描述，默认会自动生成。
    • 也可以使用args_schema参数定义，值为使用Pydantic定义的格式类
  • @tool装饰器也有return_direct参数。
• @tool装饰器参数用于定义和描述工具，为能够更好的被模型理解。工具应该有有意义的名称、描述和JSON模式，模型理解会更好。
• @tool装饰器的函数可以支持异步

如：async def amultiply(a: int, b: int) -> str:

• @tool装饰器的函数可以有Runnable接口
  • 函数可以直接使用invoke执行
• 示例
  • 乘法示例

代码

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

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

@tool(
    "multiplication-tool",
    description="useful for when you need to multiply two numbers together",
    args_schema=CalculatorInput,
    return_direct=False,          # 得到的结果，要再给到模型处理，所以是False
)
def multiply(a: int, b: int) -> str:
    """Multiply two numbers."""
    result = a * b
    return f"{a} × {b} = {result}"  # 输出要字符串不能数值，因为要再给到模型处理

print(multiply.name)
print(multiply.description)
print(multiply.args)
print(multiply.return_direct)
print(multiply.invoke({"a": 2, "b": 3}))

结果

multiplication-tool

useful for when you need to multiply two numbers together

{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}

False

2 × 3 = 6

• 时间示例（获取当前电脑的时间）

代码

import datetime
from langchain_core.tools import tool

@tool(
    "Current_Time-tool",
    description="Used to retrieve the current date and time",
    return_direct=False,
)
def current_time() -> str:
    """Get the current date and time"""
    res = datetime.datetime.now().strftime('%Y年%m月%d日，%H时%M分%S秒')
    return res

• IP地址示例（获取当前电脑IP地址）

代码

import requests
from langchain_core.tools import tool
import json

@tool(
    "Location-tool",
    description="用于获取当前地理位置信息，返回城市、省份等地理位置数据。",
    return_direct=False,
)
def get_current_location() -> str:
    """获取当前地理位置"""
    try:
        # 使用免费的IP地理位置API
        response = requests.get("http://ip-api.com/json/?lang=zh-CN", timeout=10)
        if response.status_code == 200:
            data = response.json()
            if data["status"] == "success":
                country = data.get("country", "未知")
                region = data.get("regionName", "未知")
                city = data.get("city", "未知")
                location_info = f"{country} {region} {city}"   # 格式化返回结果
                if country == "中国":             # 如果是中国，提供更详细的格式
                    if region and city:
                        if region in city:        # 如果城市名已包含省份信息
                            location_info = city
                        else:
                            location_info = f"{region} {city}"
                    elif city:
                        location_info = city
                    elif region:
                        location_info = region
                return location_info
            else:
                return "无法获取位置信息"
        else:
            return "网络请求失败，无法获取位置信息"

    except requests.exceptions.Timeout:
        return "请求超时，无法获取位置信息"
    except requests.exceptions.RequestException as e:
        return f"网络错误: {str(e)}，无法获取位置信息"
    except json.JSONDecodeError:
        return "数据解析错误，无法获取位置信息"
    except Exception as e:
        return f"获取位置信息时发生错误: {str(e)}，无法获取位置信息"

5.StructuredTool

• StructuredTool.from_function()方法提供了比@tool装饰器更多的可配置性，如同时支持同步/异步函数。
• 使用StructuredTool.from_function()方法，直接指定函数作为工具。参数
  • func：(必需)指定工具的同步函数。当在同步上下文中调用工具时，会自动使用同步函数执行。
  • coroutine (可选)指定工具的异步版本（async函数）。当在异步上下文中调用工具时，会自动使用异步函数执行。
    • 要求：与func参数一致（参数名和类型相同）。
  • name：(可选)：自定义工具名称。默认值为函数名。
  • description：(可选)工具的描述文本，帮助 LLM 理解其用途。默认值为函数的注释。
  • args_schema：(可选)用Pydantic模型定义输入参数的验证规则。
  • return_direct：(可选)若为True，工具结果直接返回用户，不被LLM处理。默认值：False。
  • handle_tool_error：(可选)自定义错误处理函数
• StructuredTool.from_function()方法返回的对象可以有Runnable接口
  • 可以直接使用invoke执行
• 乘法示例

代码

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

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

def multiply(a: int, b: int) -> str:
    """Multiply two numbers."""
    result = a * b
    return f"{a} × {b} = {result}"

async def amultiply(a: int, b: int) -> str:
    """Multiply two numbers."""
    result = a * b
    return f"{a} × {b} = {result}"

async def main():
    calculator = StructuredTool.from_function(
        func=multiply, coroutine=amultiply, 
        name="calculator",    
        description="useful for when you need to multiply two numbers together",
        args_schema=CalculatorInput,
        return_direct=True,
    )
    print(calculator.invoke({"a": 2, "b": 3}))
    print(await calculator.ainvoke({"a": 2, "b": 5}))

# 运行异步主函数
asyncio.run(main())

结果

2 × 3 = 6
2 × 5 = 10

6.处理工具错误

• 错误处理策略，以便代理可以从错误中恢复并继续执行，而不是报错退出。
• 使用StructuredTool.from_function()方法的handle_tool_error参数指定一个错误处理方式。
• handle_tool_error可设置为False/True、字符串值或函数。
  • handle_tool_error默认设置为False，直接会报错（ToolException）。
  • 将handle_tool_error设置为True，直接返回错误信息字符串，而不报错。

代码

from langchain_core.tools import StructuredTool
from langchain_core.tools import ToolException

def get_weather(city: str) -> int:
    """获取给定城市的天气。"""
    raise ToolException(f"错误：没有名为{city}的城市。")    # 故意触发错误

get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error=True,
)
response = get_weather_tool.invoke({"city": "foobar"})
print(response)

结果

错误：没有名为foobar的城市。
# 如果是ToolException()，会输出：Tool execution error

• 将handle_tool_error设置为字符串，直接返回指定字符串，而不报错。

代码

from langchain_core.tools import StructuredTool
from langchain_core.tools import ToolException

def get_weather(city: str) -> int:
    """获取给定城市的天气。"""
    raise ToolException(f"错误：没有名为{city}的城市。")

get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error="没找到这个城市",
)

response = get_weather_tool.invoke({"city": "foobar"})
print(response)

结果

没找到这个城市
# ToolException()内的内容不起作用

• 将handle_tool_error设置为一个函数，直接返回函数结果，而不报错。

代码

from langchain_core.tools import StructuredTool
from langchain_core.tools import ToolException

def get_weather(city: str) -> int:
    """获取给定城市的天气。"""
    raise ToolException(f"错误：没有名为{city}的城市。")

def _handle_error(error: ToolException) -> str:
    return f"工具执行期间发生以下错误：`{error.args[0]}`"

get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error=_handle_error,
)

response = get_weather_tool.invoke({"city": "foobar"})
print(response)

结果

工具执行期间发生以下错误：`错误：没有名为foobar的城市。`
# ToolException()内的内容不起作用

三、调用内置工具包

1.使用内置工具包，创建代理执行器

• 工具包是一组在一起使用以执行某种特定任务的工具对象列表。
  • 要获取可用的现成工具包完整列表，访问

https://www.aidoczh.com/langchain/v0.2/docs/integrations/toolkits/

2.示例：数据库工具包和代理

• 使用SQLDatabase的SQLDatabaseToolkit()，操作langchain.db数据库
  • db参数为数据库对象
  • llm参数为模型对象
  • 返回一个SQL工具套件生成器对象Toolkit。它不是工具对象，也不是工具对象列表，是一个工具生成器。
• 使用create_sql_agent()方法，用于将SQLDatabaseToolkit工具生成器创建成AgentExecutor代理执行对象，内部已经封装了操作数据库的工作流。
  • llm：用于驱动agent语言模型。如果agent_type是"tool-calling"，则该llm必须支持工具调用功能。
  • toolkit：Toolkit实例，为agent提供操作数据库所需的工具生成器。
  • agent_type：agent的类型，可选值：“tool-calling”，“openai-tools”，“openai-functions”，“zero-shot-react-description”（默认，万能兼容所有模型）
  • verbose：是否显示详细的执行日志，默认False。

代码

from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="Pro/deepseek-ai/DeepSeek-V3",
    openai_api_key="************",
    openai_api_base="https://api.siliconflow.cn/v1",
    temperature=0,         # 精准任务必须设为0，杜绝发散思考
    max_tokens=8000,
)

db = SQLDatabase.from_uri(
    database_uri=r"sqlite:///D:\python\envs\langchain2\db\database.db"
)
toolkit = SQLDatabaseToolkit(db=db, llm=model)

agent_executor = create_sql_agent(
    llm=model,
    toolkit=toolkit,
    verbose=False,                                # 如果需要查看过程可以设为True
    agent_type="zero-shot-react-description",     # 此类型兼容所有模型
)

# 需要精准的提问指令，直接指定SQL行为，否则可能失败
input = "请执行SQL语句统计sqlite数据库中student表的总行数，给出具体的数字答案"
result = agent_executor.invoke(input)
print(result)

结果

{'input': '请执行SQL语句统计sqlite数据库中students表的总行数，给出具体的数字答案', 'output': 'The "studnet" table contains a total of 10 rows.'}

• 返回SQL工具套件生成器对象Toolkit后，使用get_tools()方法，返回一个 List[BaseTool]标准工具对象列表，Agent能直接调用的最终格式。工具对象列表包含4个标准SQL工具：
  • sql_db_list_tables
    • 作用：查询并返回「当前数据库里所有的表名」；
    • Agent调用时机：Agent不知道数据库里有什么表时，第一步一定会先调用这个工具，获取表名清单。
  • sql_db_schema
    • 作用：传入指定表名，返回该表的「完整建表语句 + 字段名 + 字段类型 + 主键 + 约束」；
    • Agent调用时机：拿到表名后，必须调用这个工具才能知道表的结构，这是生成正确SQL语句的前提，是最核心的SQL工具。
  • sql_db_check_query
    • 作用：传入一段SQL语句，校验语法是否合法、是否能在当前数据库执行（只校验不执行）；
    • Agent调用时机：Agent用大模型生成SQL语句后，会先调用这个工具做语法预检，避免执行错误SQL导致程序崩溃，是容错核心。
  • sql_db_query
    • 作用：传入合法的SQL查询语句，执行该SQL并返回查询结果（支持SELECT/COUNT/SUM等所有查询类SQL）；
    • Agent调用时机：SQL语法校验通过后，最终调用这个工具获取数据库的真实数据，是最终执行环节。

代码

from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor

model = ChatOpenAI(
    model="Pro/deepseek-ai/DeepSeek-V3",
    openai_api_key="************",
    openai_api_base="https://api.siliconflow.cn/v1",
    temperature=0,
    max_tokens=8000,
)

db = SQLDatabase.from_uri(
    database_uri=r"sqlite:///D:\python\envs\langchain2\db\database.db"
)
toolkit = SQLDatabaseToolkit(db=db, llm=model)
sql_tools = toolkit.get_tools()
print(sql_tools)

prompt = hub.pull("hwchase17/react")
text_agent = create_react_agent(
    llm=model,
    tools=sql_tools,
    prompt=prompt,
)
agent_executor = AgentExecutor(agent=text_agent, tools=sql_tools)

input = "请执行SQL语句统计sqlite数据库中student表的总行数，给出具体的数字答案"
result = agent_executor.invoke({"input": input})
print(result)

结果

[QuerySQLDatabaseTool(description="Input ..., llm_kwargs={}))]

{'input': '请执行SQL语句统计sqlite数据库中student表的总行数，给出具体的数字答案', 'output': 'The student table contains a total of 10 rows.'}

四、调用第三方工具

1.第三方工具

• LangChain拥有大量第三方工具。
  • 可用工具列表：https://python.langchain.com/docs/integrations/tools/
  • 部分工具可以参考《LangChain》章节中的代理
• LangChain的第三方工具。使用的方式与自定义工具一样。
• 以维基百科工具为例
  • 使用维基百科工具需要下载维基百科包：pip install -qU wikipedia
  • 使用WikipediaAPIWrapper()方法，API包装器：用于执行维基百科搜索和获取页面摘要。返回包装器实例。参数：
    • top_k_results：返回前k个搜索结果（默认3个）
    • doc_content_chars_max：限制文档内容的最大字符数（默认4000）
    • lang：设置语言（默认’en’）
  • 使用WikipediaQueryRun()方法，生成一个Tool类：
    • 继承BaseTool基类。封装了WikipediaAPIWrapper的功能，使其可以作为Agent的工具使用。
    • api_wrapper参数用于指定包装器实例
    • 提供了标准的Tool接口，包括name、description等属性
    • 内置了description说明内容。

代码

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

api_wrapper = WikipediaAPIWrapper(
    top_k_results=1, 
    doc_content_chars_max=100, 
    lang="zh"
)
tool = WikipediaQueryRun(api_wrapper=api_wrapper)
print(tool.invoke({"query": "langchain"}))
print(tool.name)
print(tool.description)
print(tool.args)
print(tool.return_direct)

结果

Page: LangChain
Summary: LangChain 是一个应用框架，旨在简化使用大型语言模型的应用程序。作为一个语言模型集成框架，LangChain 的用例与一般语言模型的用例有很大

wikipedia

A wrapper around Wikipedia. Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects. Input should be a search query.

{'query': {'description': 'query to look up on wikipedia', 'title': 'Query', 'type': 'string'}}

False

• 可以定义第三方工具的名称、描述和输入格式描述。

代码

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from pydantic import BaseModel, Field

api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)

class WikiInputs(BaseModel):
    """维基百科工具的输入。"""
    query: str = Field(
        description="query to look up in Wikipedia, should be 3 or less words"
    )

tool = WikipediaQueryRun(
    name="wiki-tool",
    description="look up things in wikipedia",
    args_schema=WikiInputs,
    api_wrapper=api_wrapper,
    return_direct=True,
)

print(tool.run("langchain"))
print(f"Name: {tool.name}")
print(f"Description: {tool.description}")
print(f"args schema: {tool.args}")
print(f"returns directly?: {tool.return_direct}")

2.修改第三方工具–带实时性的Tavily搜索工具

• 第三方工具能够实现某个功能，但在对话环境下可能存在局限性，比如第三方搜索工具，直接搜索会有时效性的局限（比如搜索昨天新闻，搜索工具直接搜索，是不知道昨天是什么日期的）。修改第三方工具，弥补局限性。
• 使用BaseTool基类被继承，定义新工具类。在新工具类中修改第三方工具。
• 新工具类需要定义的属性和方法：
  • name工具名称字符串，description工具描述字符串，args_schema参数限制pydantic类，return_direct是否直接返回给用户，需要修改的第三方工具。
  • 在工具类的__init__()方法中增加修改的第三方工具类
  • 在_run()方法中实现修改逻辑
• 示例
  • 修改第三方工具TavilySearch，让其每次搜索都会带上当前时间。
  • time_search_tool.py代码可以被引用使用

time_search_tool.py代码

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import BaseTool
from typing import Type, Any
from pydantic import BaseModel, Field
import datetime

# 创建带时间前缀的搜索工具
class TimeAwareSearchInput(BaseModel):
    query: str = Field(description="搜索查询内容")

class TimeAwareSearchTool(BaseTool):
    name: str = "time_aware_search"               # 添加类型注解
    description: str = (
        "智能搜索工具，会自动在搜索查询前添加当前时间信息，特别适用于需要最新信息的查询"                                           # 添加类型注解
    )
    args_schema: Type[BaseModel] = TimeAwareSearchInput
    return_direct: bool = False
    search_tool: Any = Field(default=None)       # 正确声明 search_tool 属性

    def __init__(self):
        super().__init__()
        self.search_tool = TavilySearchResults(
            max_results=2, 
            tavily_api_key="************"
        )

    def _run(self, query: str) -> str:
        # 获取当前时间
        current_time = datetime.datetime.now()
        time_str = current_time.strftime("%Y年%m月%d日 %H时%M分")
        enhanced_query = f"当前时间是{time_str}，{query}"  # 在查询前添加时间信息
        # 执行搜索
        try:
            results = self.search_tool.invoke({"query": enhanced_query})
            return str(results)
        except Exception as e:
            return f"搜索失败: {str(e)}"

    async def _arun(self, query: str) -> str:
        return self._run(query)

if __name__ == "__main__":
    search_tool = TimeAwareSearchTool()  # 创建时间感知的搜索工具实例

3.修改第三方工具–将视觉模型能力创建成工具

• 将一个有特殊能力的模型包装成一个工具，让agent在运作时，决定是否使用该工具来完成特定的任务
• 豆包的视频分析模型。
  • 模型可以分析图片中的试卷内容。识别几何图形，给出解题方案。
• 示例
  • 代理执行器有记忆功能，如果当前输入没有图片，会查找之前的图片继续分析

代码

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import Optional
from langchain_openai import ChatOpenAI
from volcenginesdkarkruntime import Ark
import base64
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from time_search_tool import TimeAwareSearchTool

current_image_path = None           # 最近使用图片的路径
API_KEY = "************"
MODEL_ID = "doubao-1-5-thinking-vision-pro-250428"
client = Ark(api_key=API_KEY)        # 初始化火山引擎视觉模型客户端

# 将指定路径图片转为base64格式化的数据URL
def encode_image(image_path):
    with open(image_path, "rb") as image_file:
        base64_data = base64.b64encode(image_file.read()).decode("utf-8")
    image_format = image_path.lower().split(".")[-1]
    if image_format == "jpg":
        image_format = "jpeg"
    return f"data:image/{image_format};base64,{base64_data}"

def call_vision_api(query: str, image_path: Optional[str] = None) -> str:
    """
    调用火山引擎视觉理解API，优先使用传入的图片路径；若未传入则从记忆中读取。
    参数:
        image_path: 图片的本地路径（可选，首次提问需传入，后续可省略）
        query: 关于图片的具体问题
    """
    try:
        global current_image_path     # 确定当前使用的图片路径

        if image_path:           # 如果有传入图片路径，更新全局变量
            current_image_path = image_path
        # 如果没有传入图片路径，使用全局变量中的路径
        if not image_path and current_image_path:
            image_path = current_image_path
        if not image_path:
            return "错误：请先提供图片路径"

        image_url = encode_image(image_path) # 转换图片为base64格式化的数据URL

        # 豆包视频模型的请求内容
        messages = [{
            "role": "user",
            "content": [
                {"type": "image_url", "image_url": {"url": image_url}},
                {"type": "text", "text": query},
            ],
        }]

        # 调用API
        response = client.chat.completions.create(
            model=MODEL_ID,
            messages=messages,
            response_format={"type": "text", "style": "concise"},
        )
        result = response.choices[0].message.content

        return result
    except Exception as e:
        return f"API调用失败: {str(e)}"

# 定义输入模式（图片路径设为可选，首次传入后后续可省略）
class VisionInput(BaseModel):
    image_path: Optional[str] = Field(
        default=None,
        description="图片的本地路径（如：D:\\img\\test.jpg），首次提问需提供，后续提问可省略",
    )
    query: str = Field(description="关于图片的具体问题")

# 封装工具
vision_tool = StructuredTool(
    name="VisionUnderstanding",
    func=call_vision_api,
    description="用于处理与图片相关的问题（如识别内容、解答题目等）。首次调用需传入image_path，后续可仅传query",
    args_schema=VisionInput,
)

# 封装了TavilySearchResults工具，增加了当前时间信息
search_tool = TimeAwareSearchTool()
tools = [vision_tool, search_tool]            # 工具列表

# 总的LLM模型
model = ChatOpenAI(
    model="doubao-seed-1-6-flash-250615",
    openai_api_key=************
    openai_api_base="https://ark.cn-beijing.volces.com/api/v3",
    temperature=0.2,
)

# 创建Agent提示词（加入对话历史，让模型参考上下文）
agent_prompt = ChatPromptTemplate.from_messages( [
    (
        "system",
        "你是一个智能助手。在处理用户问题时，请严格遵循以下规则："
        "1. 当用户提供图片路径和问题时，使用视觉理解工具(VisionUnderstanding)分析图片内容。"
        "2. 对于数学题目的解答，要详细说明解题思路和步骤。"
        "3. 当用户询问前面题目的相关内容时，优先查看对话历史记录进行回答。"
        "4. 如果需要补充知识点或解题技巧，使用搜索工具获取准确信息。"
        "5. 在回答问题时，要基于工具返回的实际结果，不要随意推测或编造。"
        "6. 如果用户没有提供图片路径，需要提醒用户提供。"
        "7. 当用户要求总结考点时，要结合之前的对话历史，全面分析题目特点。"
        "请确保严格执行以上规则，给出清晰、准确的答案。",
    ),
    ("placeholder", "{chat_history}"),  # 引入对话历史记忆
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),  # 存储Agent思考过程
])

# 创建Agent执行器
agent = create_tool_calling_agent(llm=model, tools=tools, prompt=agent_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 创建带记忆的Agent执行器
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]
agent_with_chat_history = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

def main():
    # 第1次提问：需传入图片路径
    image_path = r"D:\python\envs\langchain2\img\ff.jpg"
    query1 = "图片中的第十六题怎么做？"
    res1 = agent_with_chat_history.invoke(
        input={"input": f"图片路径：{image_path}，问题：{query1}"},
        config={"configurable": {"session_id": "123"}},
    )
    print(res1["output"])
    print("\n" + "-" * 50 + "\n")

    # 第2次提问：无需传入图片路径，自动复用之前的图片
    query2 = "第17题怎么做？"
    res2 = agent_with_chat_history.invoke(
        {"input": query2},
        config={"configurable": {"session_id": "123"}},
    )
    print(res2["output"])
    print("\n" + "-" * 50 + "\n")

    # 第3次提问：继续复用图片
    query3 = "请总结一下前面这两道题的考点是什么？"
    res3 = agent_with_chat_history.invoke(
        {"input": query3},
        config={"configurable": {"session_id": "123"}},
    )
    print(res3["output"])

if __name__ == "__main__":
    main()

---

总结

本文整理了 LangChain 工具体系的核心用法，包括工具对象的名称、描述、输入格式、执行函数和返回方式；也介绍了 Tool、@tool、StructuredTool、BaseTool 等创建方式，以及 SQLDatabaseToolkit、Wikipedia、Tavily 等内置或第三方工具的调用与改造方法。实际项目中，建议优先使用 @tool 或 StructuredTool.from_function 创建工具，并为工具提供清晰的名称、描述和参数结构，这样更有利于大模型正确理解和调用工具。