1.5 实战：构建第一个 LangChain Tool

引言：从理论到实践

小白理解 - 什么是 LangChain Tool？
想象 AI Agent 是一个超级助手，但它只会"说话"（生成文本）。
如果你想让它：
查天气？❌ 它不知道
搜索网页？❌ 它做不到
计算复杂数学？❌ 它会算错
Tool 就是给 AI 助手的"技能包"！
AI Agent（只会说话）
    │
    ├── 天气 Tool → 现在能查天气了！
    ├── 搜索 Tool → 现在能搜索了！
    └── 计算 Tool → 现在算数准了！
1
2
3
4
5
你在这节课学的，就是如何给 AI 助手制作技能包！

到目前为止，你已经学习了函数、装饰器、模块化——现在是时候将这些知识综合运用，构建一个真实可用的 LangChain Tool。

在本节中，我们将：

理解 LangChain Tool 的结构
使用 @tool 装饰器创建自定义工具
实现错误处理和类型验证
将 Tool 集成到 Agent 中
完整示例：天气查询 Tool

学习目标

✅ 理解 LangChain Tool 的基本结构
✅ 掌握 @tool 装饰器的使用
✅ 实现健壮的错误处理
✅ 集成 Tool 到 ReAct Agent
✅ 测试和调试 Tool

第一部分：LangChain Tool 的解剖

Tool 的核心要素

一个 LangChain Tool 需要：

名称（name）：Tool 的唯一标识符
描述（description）：告诉 LLM 这个 Tool 做什么
参数（args_schema）：Tool 接受的参数
执行函数（func）：Tool 的实际逻辑

小白理解 - 为什么 Tool 需要这些？
想象你是 AI，收到用户请求"北京明天天气怎么样"：
你有一堆 Tool：搜索、计算、天气、翻译...
你怎么选？ 看每个 Tool 的描述（description）
天气 Tool 描述："获取指定城市的天气信息" ← 选这个！
怎么使用？ 看参数说明（args_schema）
需要参数：city（城市名）
于是调用：weather_tool(city="北京")
描述写得好，AI 才知道什么时候该用这个 Tool！

最简单的 Tool

python

from langchain.tools import tool

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# Tool 的属性
print(get_current_time.name)         # get_current_time
print(get_current_time.description)  # 获取当前时间

小白理解 - @tool 做了什么？
还记得装饰器那节课吗？@tool 就是 LangChain 提供的装饰器：
python
@tool
def get_current_time() -> str:
    """获取当前时间"""  ← 这行会变成 Tool 的描述！
    ...
1
2
3
4
@tool 装饰器会：
把函数名字 → tool.name
把文档字符串 → tool.description
把参数 → tool.args_schema
把函数包装成 AI 可以调用的格式
一行 @tool，普通函数秒变 AI 技能！

第二部分：创建带参数的 Tool

基础示例

python

from langchain.tools import tool

@tool
def search_wikipedia(query: str) -> str:
    """
    在维基百科上搜索信息
    
    Args:
        query: 搜索关键词
    
    Returns:
        搜索结果摘要
    """
    # 实际实现会调用 Wikipedia API
    return f"维基百科搜索结果：{query}..."

# 调用
result = search_wikipedia.invoke({"query": "Python programming"})
print(result)

使用 Pydantic 定义参数结构

python

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

class CalculatorInput(BaseModel):
    """计算器输入参数"""
    expression: str = Field(description="要计算的数学表达式")

@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """
    计算数学表达式
    
    Args:
        expression: 数学表达式（如 "2 + 3 * 4"）
    
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

第三部分：实现真实的 Tool

完整示例：天气查询 Tool

python

"""
天气查询 Tool
演示如何创建一个真实可用的 LangChain Tool
"""

from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Optional
import json


class WeatherInput(BaseModel):
    """天气查询输入参数"""
    location: str = Field(description="城市名称，如 '北京' 或 'Beijing'")
    unit: str = Field(
        default="celsius",
        description="温度单位：'celsius' 或 'fahrenheit'"
    )


@tool(args_schema=WeatherInput)
def get_weather(location: str, unit: str = "celsius") -> str:
    """
    获取指定城市的天气信息
    
    这个工具可以查询任何城市的实时天气，包括：
    - 当前温度
    - 天气状况（晴、雨、雪等）
    - 湿度
    - 风速
    
    Args:
        location: 城市名称
        unit: 温度单位
    
    Returns:
        JSON 格式的天气信息
    """
    # 参数验证
    if not location:
        return json.dumps({"error": "城市名称不能为空"})
    
    if unit not in ["celsius", "fahrenheit"]:
        return json.dumps({"error": "单位必须是 celsius 或 fahrenheit"})
    
    # 模拟 API 调用
    # 实际应用中会调用 OpenWeatherMap 等 API
    weather_data = {
        "location": location,
        "temperature": 25 if unit == "celsius" else 77,
        "unit": "°C" if unit == "celsius" else "°F",
        "condition": "晴天",
        "humidity": "60%",
        "wind_speed": "15 km/h",
        "forecast": "未来三天天气晴朗"
    }
    
    return json.dumps(weather_data, ensure_ascii=False, indent=2)


# 测试 Tool
if __name__ == "__main__":
    result = get_weather.invoke({
        "location": "北京",
        "unit": "celsius"
    })
    print(result)

带错误处理和重试的 Tool

python

from langchain.tools import tool
from pydantic import BaseModel, Field
import time
from typing import Optional


class WebSearchInput(BaseModel):
    """网络搜索输入参数"""
    query: str = Field(description="搜索查询")
    max_results: int = Field(default=5, description="最大结果数")


@tool(args_schema=WebSearchInput)
def search_web(query: str, max_results: int = 5) -> str:
    """
    搜索网络并返回结果
    
    Args:
        query: 搜索查询
        max_results: 返回的最大结果数
    
    Returns:
        搜索结果列表（JSON 格式）
    """
    # 参数验证
    if not query or len(query.strip()) == 0:
        return json.dumps({"error": "搜索查询不能为空"})
    
    if max_results < 1 or max_results > 10:
        return json.dumps({"error": "max_results 必须在 1-10 之间"})
    
    # 重试逻辑
    max_retries = 3
    for attempt in range(max_retries):
        try:
            # 模拟 API 调用
            results = [
                {
                    "title": f"搜索结果 {i+1}: {query}",
                    "url": f"https://example.com/result{i+1}",
                    "snippet": f"关于 {query} 的相关信息..."
                }
                for i in range(max_results)
            ]
            
            return json.dumps({
                "query": query,
                "results": results,
                "total": len(results)
            }, ensure_ascii=False, indent=2)
            
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(1)
                continue
            return json.dumps({
                "error": f"搜索失败: {str(e)}"
            })

第四部分：集成到 Agent

创建 ReAct Agent

python

"""
完整的 Agent + Tools 示例
"""

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field
import json


# Tool 1: 天气查询
class WeatherInput(BaseModel):
    location: str = Field(description="城市名称")


@tool(args_schema=WeatherInput)
def get_weather(location: str) -> str:
    """获取城市天气"""
    return json.dumps({
        "location": location,
        "temperature": "25°C",
        "condition": "晴天"
    }, ensure_ascii=False)


# Tool 2: 计算器
class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式")


@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"结果: {result}"
    except Exception as e:
        return f"错误: {str(e)}"


# Tool 3: 网络搜索
class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")


@tool(args_schema=SearchInput)
def search_web(query: str) -> str:
    """搜索网络"""
    return json.dumps({
        "query": query,
        "results": [
            {"title": "结果1", "url": "http://example.com"}
        ]
    }, ensure_ascii=False)


def create_my_agent():
    """创建配置好的 Agent"""
    # 初始化 LLM
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        temperature=0
    )
    
    # 工具列表
    tools = [get_weather, calculator, search_web]
    
    # 创建 Agent
    agent = create_react_agent(
        model=llm,
        tools=tools
    )
    
    return agent


def main():
    """演示 Agent 使用"""
    agent = create_my_agent()
    
    # 测试查询
    queries = [
        "北京的天气怎么样？",
        "计算 123 * 456",
        "搜索 Python 教程"
    ]
    
    for query in queries:
        print(f"\n用户: {query}")
        response = agent.invoke({"messages": [("user", query)]})
        print(f"Agent: {response['messages'][-1].content}")


if __name__ == "__main__":
    # 注意：需要设置 OPENAI_API_KEY 环境变量
    # import os
    # os.environ["OPENAI_API_KEY"] = "your-api-key"
    
    # 演示单个 Tool
    result = get_weather.invoke({"location": "上海"})
    print(result)

第五部分：Tool 开发最佳实践

1. 详细的描述（Description）

python

@tool
def search_database(query: str) -> str:
    """
    在数据库中搜索信息
    
    这个工具可以搜索公司内部数据库，包括：
    - 产品信息
    - 客户记录  
    - 订单历史
    
    使用场景：
    - 当用户询问产品详情时
    - 当用户查询订单状态时
    - 当需要查找客户信息时
    
    注意事项：
    - 查询必须具体明确
    - 支持模糊搜索
    - 最多返回 20 条结果
    
    Args:
        query: 搜索查询字符串
        
    Returns:
        JSON 格式的搜索结果
    """
    pass

💡 为什么描述重要？
LLM 通过描述来决定何时使用这个 Tool。描述越详细，Tool 被正确使用的概率越高。

2. 参数验证

python

from pydantic import BaseModel, Field, validator

class EmailInput(BaseModel):
    """邮件发送参数"""
    to: str = Field(description="收件人邮箱")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")
    
    @validator("to")
    def validate_email(cls, v):
        """验证邮箱格式"""
        if "@" not in v:
            raise ValueError("无效的邮箱地址")
        return v
    
    @validator("subject")
    def validate_subject(cls, v):
        """验证主题不为空"""
        if not v.strip():
            raise ValueError("邮件主题不能为空")
        return v

3. 错误处理

python

@tool
def risky_operation(param: str) -> str:
    """可能失败的操作"""
    try:
        # 实际操作
        result = perform_operation(param)
        return json.dumps({"success": True, "data": result})
    
    except ValueError as e:
        return json.dumps({"success": False, "error": f"参数错误: {e}"})
    
    except ConnectionError as e:
        return json.dumps({"success": False, "error": f"网络错误: {e}"})
    
    except Exception as e:
        return json.dumps({
            "success": False,
            "error": f"未知错误: {e}",
            "type": type(e).__name__
        })

4. 结构化输出

python

@tool
def get_user_info(user_id: str) -> str:
    """获取用户信息"""
    # ✅ 推荐：返回 JSON
    return json.dumps({
        "user_id": user_id,
        "name": "张三",
        "email": "zhang@example.com",
        "status": "active"
    }, ensure_ascii=False)
    
    # ❌ 不推荐：返回纯文本
    # return "用户ID: 123, 姓名: 张三, 邮箱: zhang@example.com"

第六部分：测试和调试

单元测试

python

import pytest
from your_tools import get_weather, calculator

def test_get_weather():
    """测试天气查询"""
    result = get_weather.invoke({"location": "北京"})
    assert "北京" in result
    assert "temperature" in result

def test_calculator_valid():
    """测试计算器：有效输入"""
    result = calculator.invoke({"expression": "2 + 3"})
    assert "5" in result

def test_calculator_invalid():
    """测试计算器：无效输入"""
    result = calculator.invoke({"expression": "invalid"})
    assert "错误" in result

调试技巧

python

@tool
def debug_tool(param: str) -> str:
    """带调试的 Tool"""
    import logging
    logging.basicConfig(level=logging.DEBUG)
    
    logger = logging.getLogger(__name__)
    logger.debug(f"Tool 被调用，参数: {param}")
    
    try:
        result = process(param)
        logger.debug(f"处理成功，结果: {result}")
        return result
    except Exception as e:
        logger.error(f"处理失败: {e}", exc_info=True)
        raise

完整项目示例

python

"""
完整的 Tool 库示例
展示如何组织多个 Tools
"""

from langchain.tools import tool
from pydantic import BaseModel, Field
import json
from typing import Optional


# ============= 工具定义 =============

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


@tool(args_schema=WeatherInput)
def get_weather(location: str, unit: str = "celsius") -> str:
    """获取天气信息"""
    return json.dumps({
        "location": location,
        "temperature": 25,
        "unit": unit,
        "condition": "晴天"
    }, ensure_ascii=False)


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


@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"结果: {result}"
    except Exception as e:
        return f"错误: {str(e)}"


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


@tool(args_schema=SearchInput)
def search_web(query: str, max_results: int = 5) -> str:
    """搜索网络"""
    results = [
        {"title": f"结果 {i+1}", "snippet": f"关于 {query}..."}
        for i in range(max_results)
    ]
    return json.dumps(results, ensure_ascii=False)


# ============= 工具注册表 =============

ALL_TOOLS = [get_weather, calculator, search_web]


def get_tools_by_category(category: str) -> list:
    """按类别获取工具"""
    categories = {
        "weather": [get_weather],
        "math": [calculator],
        "search": [search_web],
        "all": ALL_TOOLS
    }
    return categories.get(category, [])


# ============= 使用示例 =============

if __name__ == "__main__":
    # 测试所有工具
    print("=== 测试天气工具 ===")
    print(get_weather.invoke({"location": "北京"}))
    
    print("\n=== 测试计算器 ===")
    print(calculator.invoke({"expression": "123 + 456"}))
    
    print("\n=== 测试搜索 ===")
    print(search_web.invoke({"query": "Python", "max_results": 3}))

本节总结

你学到了什么

✅ LangChain Tool 的基本结构
✅ 使用 @tool 装饰器创建 Tool
✅ 使用 Pydantic 定义参数结构
✅ 实现错误处理和验证
✅ 集成 Tool 到 Agent
✅ Tool 开发最佳实践

关键要点

详细的描述：让 LLM 知道何时使用
参数验证：Pydantic schema
错误处理：返回结构化错误信息
JSON 输出：便于解析
测试：确保 Tool 正确工作

下一步

现在你已经掌握了创建 LangChain Tool 的技能！在下一节，我们将回顾本章的所有内容，并完成高难度的编码挑战。

下一节：1.6 小结和复习

1.5 实战：构建第一个 LangChain Tool ​

引言：从理论到实践 ​

学习目标 ​

第一部分：LangChain Tool 的解剖 ​

Tool 的核心要素 ​

最简单的 Tool ​

第二部分：创建带参数的 Tool ​

基础示例 ​

使用 Pydantic 定义参数结构 ​

第三部分：实现真实的 Tool ​

完整示例：天气查询 Tool ​

带错误处理和重试的 Tool ​

第四部分：集成到 Agent ​

创建 ReAct Agent ​

第五部分：Tool 开发最佳实践 ​

1. 详细的描述（Description） ​

2. 参数验证 ​

3. 错误处理 ​

4. 结构化输出 ​

第六部分：测试和调试 ​

单元测试 ​

调试技巧 ​

完整项目示例 ​

本节总结 ​

你学到了什么 ​

关键要点 ​

下一步 ​

1.5 实战：构建第一个 LangChain Tool

引言：从理论到实践

学习目标

第一部分：LangChain Tool 的解剖

Tool 的核心要素

最简单的 Tool

第二部分：创建带参数的 Tool

基础示例

使用 Pydantic 定义参数结构

第三部分：实现真实的 Tool

完整示例：天气查询 Tool

带错误处理和重试的 Tool

第四部分：集成到 Agent

创建 ReAct Agent

第五部分：Tool 开发最佳实践

1. 详细的描述（Description）

2. 参数验证

3. 错误处理

4. 结构化输出

第六部分：测试和调试

单元测试

调试技巧

完整项目示例

本节总结

你学到了什么

关键要点

下一步