langchain学习笔记（二）：工具的调用

Tool Calling

定义简单tool

创建工具最简单的方法是使用 @tool 装饰器。

1@tool(description="Returns the current time in yyyy-MM-dd HH:mm:ss format.")
2def get_current_time(*args, **kwargs) -> str:
3    """
4    获取当前系统时间。
5    格式为：yyyy-MM-dd HH:mm:ss
6    """
7    now = datetime.datetime.now()
8    return now.strftime("%Y-%m-%d %H:%M:%S")
9

如果description不写，则默认注释里的就是描述信息，否则则使用description，描述信息需要易于大模型的理解。

1"""
2获取当前系统时间。
3格式为：yyyy-MM-dd HH:mm:ss
4"""
5

llm绑定tool

借助 bind_tools，实现llm与工具的绑定，在底层，这些都会被转换为 OpenAI 工具模式，其格式如下：

1{
2    "name": "...",
3    "description": "...",
4    "parameters": {...}  # JSONSchema
5}
6

代码示例：

1# ========== 方式1：自动工具调用（推荐） ==========
2def demo_simple_tool_calling():
3    """
4    方式1：使用 bind_tools + 手动执行
5    特点：需要手动处理工具调用和结果
6    """
7    print("=" * 50)
8    print("🛠️ 方式1：bind_tools + 手动执行工具")
9    print("=" * 50)
10
11    llm_tools = llm.bind_tools(tools=[get_current_time])
12    messages = [
13        {"role": "user", "content": "请获取当前时间"}
14    ]
15
16    # 第一步：LLM 返回工具调用请求
17    result = llm_tools.invoke(messages)
18    print(f"\n📤 LLM 响应类型: {type(result)}")
19    print(f"📤 LLM 内容: {result.content}")
20    print(f"📤 工具调用请求: {result.tool_calls}")
21
22    # 第二步：手动执行工具，判断大模型的reponse是不是tool_call，如果是tool_call类型，则真正执行工具的调用
23    if result.tool_calls:
24        for tool_call in result.tool_calls:
25            print(f"\n🔧 执行工具: {tool_call['name']}")
26            print(f"🔧 工具参数: {tool_call['args']}")
27
28            if tool_call['name'] == 'get_current_time':
29                # 手动调用工具
30                tool_result = get_current_time.invoke(tool_call['args'])
31                print(f"✅ 工具执行结果: {tool_result}")
32
33                # 第三步：将工具结果返回给 LLM（可选）
34                messages.append(result)  # 添加 AI 的工具调用消息
35                messages.append(ToolMessage(
36                    content=tool_result,
37                    tool_call_id=tool_call['id']
38                ))
39
40                # 让 LLM 根据工具结果生成最终回答
41                final_response = llm_tools.invoke(messages)
42                print(f"\n💬 最终回答: {final_response.content}")
43

模拟工具调用流程，通过ToolMessage

1def demo_simple_tool_calling_2():
2    """
3    方式2：手动构造 AIMessage 和 ToolMessage
4    特点：完全手动控制消息流，适合测试和调试
5    """
6    print("\n" + "=" * 50)
7    print("🛠️ 方式2：手动构造消息流")
8    print("=" * 50)
9
10    # 手动构造 AI 的工具调用消息
11    ai_message = AIMessage(
12        content="",  # 通常为空，因为 AI 选择调用工具而不是直接回答
13        tool_calls=[{
14            "name": "get_current_time",
15            "args": {},
16            "id": "call_123"
17        }]
18    )
19    print(f"\n📤 模拟 AI 工具调用: {ai_message.tool_calls}")
20
21    # 执行工具并创建结果消息
22    tool_result = get_current_time.invoke({})
23    print(f"✅ 工具执行结果: {tool_result}")
24
25    tool_message = ToolMessage(
26        content=tool_result,
27        tool_call_id="call_123"
28    )
29
30    # 构造完整的消息历史
31    messages = [
32        HumanMessage("请获取当前时间"),
33        ai_message,  # AI 的工具调用
34        tool_message,  # 工具执行结果
35    ]
36
37    # LLM 根据工具结果生成最终回答
38    response = llm.invoke(messages)
39    print(f"\n💬 最终回答: {response.content}")
40

多工具调用

1# ========== 额外示例：多个工具 ==========
2@tool
3def calculate(expression: str) -> str:
4    """计算数学表达式。"""
5    try:
6        result = eval(expression)
7        return f"计算结果: {result}"
8    except Exception as e:
9        return f"计算错误: {str(e)}"
10
11
12def demo_multiple_tools():
13    """演示多个工具的使用"""
14    print("\n" + "=" * 50)
15    print("🛠️ 多工具示例")
16    print("=" * 50)
17
18    llm_tools = llm.bind_tools(tools=[get_current_time, calculate])
19    tool_map = {
20        "get_current_time": get_current_time,
21        "calculate": calculate
22    }
23
24    messages = [HumanMessage(content="现在几点？然后帮我计算 15 * 8")]
25
26    max_iterations = 10
27    iteration = 0
28
29    while iteration < max_iterations:
30        iteration += 1
31        response = llm_tools.invoke(messages)
32        messages.append(response)
33
34        if not response.tool_calls:
35            print(f"\n✅ 最终回答: {response.content}")
36            break
37
38        for tool_call in response.tool_calls:
39            tool_name = tool_call['name']
40            print(f"\n🔧 调用工具: {tool_name}")
41
42            tool_result = tool_map[tool_name].invoke(tool_call['args'])
43            print(f"✅ 结果: {tool_result}")
44
45            messages.append(ToolMessage(
46                content=tool_result,
47                tool_call_id=tool_call['id']
48            ))
49

结构化入参

tool 装饰器里，有个 args_schema 参数，可以配置结构化的类

1# ========== 1. 基础结构化入参 ==========
2class SearchInput(BaseModel):
3    """搜索工具的输入参数"""
4    query: str = Field(description="搜索关键词")
5    max_results: int = Field(default=10, description="最大返回结果数", ge=1, le=100)
6    language: str = Field(default="zh", description="语言代码，如 zh, en")
7
8
9@tool(args_schema=SearchInput)
10def search_web(query: str, max_results: int = 10, language: str = "zh") -> str:
11    """在网络上搜索信息。"""
12    return f"搜索 '{query}'，返回 {max_results} 条 {language} 结果"
13
14
15def demo_basic_structured_input():
16    """演示基础结构化入参"""
17    print("=" * 60)
18    print("📝 示例1：基础结构化入参")
19    print("=" * 60)
20
21    llm_tools = llm.bind_tools([search_web])
22    messages = [HumanMessage("搜索 LangChain 相关信息，返回5条结果")]
23
24    response = llm_tools.invoke(messages)
25    print(f"\n🤖 LLM 工具调用:")
26    for tool_call in response.tool_calls:
27        print(f"  工具: {tool_call['name']}")
28        print(f"  参数: {tool_call['args']}")
29
30        # 执行工具
31        result = search_web.invoke(tool_call['args'])
32        print(f"  结果: {result}")
33

嵌套复杂的结构化入参

1# ========== 2. 复杂结构化入参（嵌套对象） ==========
2class Address(BaseModel):
3    """地址信息"""
4    street: str = Field(description="街道地址")
5    city: str = Field(description="城市")
6    country: str = Field(default="中国", description="国家")
7    postal_code: Optional[str] = Field(None, description="邮政编码")
8
9
10class UserInfo(BaseModel):
11    """用户信息"""
12    name: str = Field(description="用户姓名")
13    age: int = Field(description="年龄", ge=0, le=150)
14    email: str = Field(description="邮箱地址")
15    address: Address = Field(description="地址信息")
16    tags: List[str] = Field(default=[], description="用户标签")
17
18    @field_validator('email')
19    def validate_email(cls, v):
20        if '@' not in v:
21            raise ValueError('邮箱格式不正确')
22        return v
23
24
25@tool(args_schema=UserInfo)
26def create_user(name: str, age: int, email: str, address: Address, tags: List[str] = None) -> str:
27    """创建新用户。"""
28    tags = tags or []
29    return f"创建用户: {name}, {age}岁, {email}, 地址: {address.city}, 标签: {tags}"
30
31
32def demo_nested_structured_input():
33    """演示嵌套结构化入参"""
34    print("\n" + "=" * 60)
35    print("📝 示例2：嵌套结构化入参")
36    print("=" * 60)
37
38    # 手动测试
39    user_data = {
40        "name": "张三",
41        "age": 25,
42        "email": "zhangsan@example.com",
43        "address": {
44            "street": "中关村大街1号",
45            "city": "北京",
46            "country": "中国",
47            "postal_code": "100000"
48        },
49        "tags": ["VIP", "技术"]
50    }
51
52    tool_call = {
53        "name": "create_user",
54        "args": user_data,
55        "id": "call_123",  # 必须提供 tool_call_id
56        "type": "tool_call"
57    }
58
59    result = create_user.invoke(tool_call)
60    print(f"\n✅ 执行结果: {result}")
61

结构化输出

response_format 是 @tool 装饰器的一个重要参数，用于控制工具返回值的格式和处理方式。

代码示例

1# ========== 4. 结构化出参 (content_and_artifact) ==========
2class ProductInfo(BaseModel):
3    """产品信息"""
4    id: int
5    name: str
6    price: float
7    stock: int
8    category: str
9
10
11class SearchResult(BaseModel):
12    """搜索结果"""
13    total: int
14    products: List[ProductInfo]
15    page: int
16    page_size: int
17
18
19class ProductSearchInput(BaseModel):
20    """产品搜索输入"""
21    keyword: str = Field(description="搜索关键词")
22    category: Optional[str] = Field(None, description="产品分类")
23    min_price: Optional[float] = Field(None, description="最低价格")
24    max_price: Optional[float] = Field(None, description="最高价格")
25    page: int = Field(default=1, description="页码", ge=1)
26    page_size: int = Field(default=10, description="每页数量", ge=1, le=100)
27
28
29@tool(
30    args_schema=ProductSearchInput,
31    response_format="content_and_artifact"
32)
33def search_products(
34        keyword: str,
35        category: Optional[str] = None,
36        min_price: Optional[float] = None,
37        max_price: Optional[float] = None,
38        page: int = 1,
39        page_size: int = 10
40) -> Tuple[str, SearchResult]:
41    """搜索产品。"""
42    # 模拟数据库查询
43    mock_products = [
44        ProductInfo(id=1, name="iPhone 15", price=6999, stock=50, category="手机"),
45        ProductInfo(id=2, name="MacBook Pro", price=12999, stock=30, category="电脑"),
46        ProductInfo(id=3, name="AirPods Pro", price=1999, stock=100, category="耳机"),
47    ]
48
49    # 过滤产品
50    filtered = mock_products
51    if category:
52        filtered = [p for p in filtered if p.category == category]
53    if min_price:
54        filtered = [p for p in filtered if p.price >= min_price]
55    if max_price:
56        filtered = [p for p in filtered if p.price <= max_price]
57
58    # 构造结果
59    result = SearchResult(
60        total=len(filtered),
61        products=filtered,
62        page=page,
63        page_size=page_size
64    )
65
66    # content: 给 LLM 的简洁描述
67    content = f"找到 {result.total} 个产品"
68    if category:
69        content += f"，分类: {category}"
70
71    # artifact: 完整的结构化数据
72    return content, result
73
74
75def demo_structured_output():
76    """演示结构化出参"""
77    print("\n" + "=" * 60)
78    print("📝 示例4：结构化出参 (content_and_artifact)")
79    print("=" * 60)
80
81    llm_with_tools = llm.bind_tools([search_products])
82
83    messages = [HumanMessage(content="搜索价格在5000以上的手机")]
84
85    # LLM 决策
86    response = llm_with_tools.invoke(messages)
87
88    if response.tool_calls:
89        tool_call = response.tool_calls[0]
90        print(f"\n🤖 LLM 决定调用工具:")
91        print(f"   工具: {tool_call['name']}")
92        print(f"   参数: {tool_call['args']}")
93
94        # 执行工具
95        ## 这里返回的是ToolMessage
96        result = search_products.invoke(tool_call)
97
98        print(f"\n✅ 工具执行结果:")
99        print(f"   Content: {result.content}")
100        print(f"   Artifact 类型: {result.artifact}")
101        print(f"   产品数量: {result.artifact.total}")
102

structure_tool

StructuredTool 是 LangChain 里的一个类，专门用于处理多参数输入的工具。 与基础的 Tool 类（通常只接受一个字符串输入）不同，StructuredTool 利用 Pydantic 来定义和校验复杂的输入结构（Schema）。

相比 @tool 装饰器提供了更多的灵活性和控制能力。

1# ========== 5. 使用 StructuredTool 创建工具 ==========
2class CalculatorInput(BaseModel):
3    """计算器输入"""
4    expression: str = Field(description="数学表达式，如 '2 + 3 * 4'")
5    precision: int = Field(default=2, description="小数精度", ge=0, le=10)
6
7
8class CalculatorOutput(BaseModel):
9    """计算器输出"""
10    expression: str
11    result: float
12    formatted_result: str
13
14
15def calculate_func(expression: str, precision: int = 2) -> Dict[str, Any]:
16    """计算数学表达式"""
17    try:
18        result = eval(expression)
19        formatted = f"{result:.{precision}f}"
20        return {
21            "expression": expression,
22            "result": result,
23            "formatted_result": formatted
24        }
25    except Exception as e:
26        return {
27            "expression": expression,
28            "result": 0,
29            "formatted_result": f"错误: {str(e)}"
30        }
31
32
33calculator_tool = StructuredTool.from_function(
34    func=calculate_func,
35    name="calculator",
36    description="计算数学表达式",
37    args_schema=CalculatorInput,
38    return_direct=False
39)
40
41
42def demo_structured_tool():
43    """演示 StructuredTool"""
44    print("\n" + "=" * 60)
45    print("📝 示例5：StructuredTool 创建工具")
46    print("=" * 60)
47
48    test_cases = [
49        {"expression": "15 + 25 * 3", "precision": 2},
50        {"expression": "100 / 3", "precision": 4},
51        {"expression": "2 ** 10", "precision": 0}
52    ]
53
54    for params in test_cases:
55        result = calculator_tool.invoke(params)
56        print(f"\n表达式: {params['expression']}")
57        print(f"结果: {result}")
58

base_tool

BaseTool 是 LangChain 中所有工具的基类。通过继承它可以创建完全自定义的工具，拥有最大的灵活性和控制能力。

核心参数与方法：

1from langchain_core.tools import BaseTool
2from pydantic import BaseModel, Field
3from typing import Type, Any, Optional
4
5class MyToolInput(BaseModel):
6    """工具输入参数 Schema"""
7    param1: str = Field(description="参数1描述")
8    param2: int = Field(default=10, description="参数2描述")
9
10class MyTool(BaseTool):
11    """自定义工具"""
12    
13    # ========== 必须定义的属性 ==========
14    name: str = "my_tool"                    # 工具名称
15    description: str = "这是一个自定义工具"    # 工具描述
16    args_schema: Type[BaseModel] = MyToolInput  # 输入参数 Schema
17    
18    # ========== 可选属性 ==========
19    return_direct: bool = False              # 是否直接返回结果
20    response_format: str = "content"         # 响应格式
21    
22    # ========== 自定义属性 ==========
23    custom_config: str = "default"           # 可以添加任意自定义属性
24    
25    # ========== 必须实现的方法 ==========
26    def _run(self, param1: str, param2: int = 10) -> str:
27        """同步执行方法（必须实现）"""
28        return f"执行结果: {param1}, {param2}"
29    
30    # ========== 可选实现的方法 ==========
31    async def _arun(self, param1: str, param2: int = 10) -> str:
32        """异步执行方法（可选）"""
33        return f"异步执行结果: {param1}, {param2}"
34

代码示例

1# ========== 6. 继承 BaseTool 实现完全自定义 ==========
2class WeatherInput(BaseModel):
3    """天气查询输入"""
4    city: str = Field(description="城市名称")
5    date: Optional[str] = Field(None, description="日期 (YYYY-MM-DD)，默认今天")
6    unit: str = Field(default="celsius", description="温度单位: celsius 或 fahrenheit")
7
8
9class WeatherOutput(BaseModel):
10    """天气查询输出"""
11    city: str
12    date: str
13    temperature: float
14    condition: str
15    humidity: int
16    wind_speed: float
17
18
19class WeatherTool(BaseTool):
20    name: str = "get_weather"
21    description: str = "获取指定城市的天气信息"
22    args_schema: type[BaseModel] = WeatherInput
23    response_format: str = "content_and_artifact"
24
25    # 自定义属性
26    api_key: str = ""
27
28    def _run(
29            self,
30            city: str,
31            date: Optional[str] = None,
32            unit: str = "celsius"
33    ) -> Tuple[str, WeatherOutput]:
34        """同步执行"""
35        # 模拟 API 调用
36        date = date or datetime.datetime.now().strftime("%Y-%m-%d")
37
38        weather_data = {
39            "北京": {"temp": 25, "condition": "晴天", "humidity": 45, "wind": 3.5},
40            "上海": {"temp": 28, "condition": "多云", "humidity": 60, "wind": 4.2},
41            "广州": {"temp": 32, "condition": "小雨", "humidity": 75, "wind": 2.8},
42        }
43
44        data = weather_data.get(city, {"temp": 20, "condition": "未知", "humidity": 50, "wind": 3.0})
45        temp = data["temp"]
46
47        if unit == "fahrenheit":
48            temp = temp * 9 / 5 + 32
49
50        output = WeatherOutput(
51            city=city,
52            date=date,
53            temperature=temp,
54            condition=data["condition"],
55            humidity=data["humidity"],
56            wind_speed=data["wind"]
57        )
58
59        content = f"{city} {date}: {output.condition}, {temp}°{'F' if unit == 'fahrenheit' else 'C'}"
60
61        return content, output
62
63    async def _arun(self, *args, **kwargs):
64        """异步执行"""
65        return self._run(*args, **kwargs)
66
67
68def demo_base_tool():
69    """演示继承 BaseTool"""
70    print("\n" + "=" * 60)
71    print("📝 示例6：继承 BaseTool 实现自定义工具")
72    print("=" * 60)
73
74    weather_tool = WeatherTool(api_key="your-api-key")
75
76    test_cases = [
77        {"city": "北京"},
78        {"city": "上海", "unit": "fahrenheit"},
79        {"city": "广州", "date": "2024-01-15"}
80    ]
81
82    for params in test_cases:
83        tool_call = {
84            "name": weather_tool.name,
85            "args": params,
86            "id": "call_123" + params["city"],
87            "type": "tool_call"
88        }
89        result = weather_tool.invoke(tool_call)
90        print(f"\n查询: {params}")
91        print(f"Content: {result.content}")
92        print(f"Artifact: {result.artifact.model_dump()}")
93