云信AI部署

LangChain自定义工具开发：Python在API集成中的应用

云信安装大师
90
AI 质量分
2 5 月, 2025
3 分钟阅读
0 阅读

LangChain自定义工具开发：Python在API集成中的应用

引言

在当今AI应用开发中，LangChain已成为连接大语言模型(LLM)与外部系统的强大框架。本文将带你学习如何开发自定义工具(Tool)来实现API集成，让LLM能够访问外部数据和服务。通过Python实现这一功能，你可以为你的AI应用添加无限可能。

准备工作

在开始之前，请确保：

  1. 已安装Python 3.8+环境
  2. 已安装LangChain核心库
  3. 了解基本的API调用概念
代码片段 
pip install langchain openai requests

基础概念：什么是LangChain工具？

工具(Tool)是LangChain中让LLM与外界交互的接口。每个工具本质上是一个Python函数，LLM可以根据需要调用这些函数来获取信息或执行操作。

第一步：创建基础API工具

让我们从一个简单的天气API查询工具开始：

代码片段 
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import requests

class WeatherCheckInput(BaseModel):
    """输入参数模型：城市名称"""
    city: str = Field(..., description="需要查询天气的城市名称")

class WeatherTool(BaseTool):
    name = "weather_checker"
    description = "根据城市名称查询当前天气情况"
    args_schema: Type[BaseModel] = WeatherCheckInput

    def _run(self, city: str):
        """实际执行API调用的方法"""
        try:
            # 这里使用模拟的天气API，实际使用时替换为真实API
            response = requests.get(f"https://api.example.com/weather?city={city}")
            data = response.json()

            if response.status_code == 200:
                return f"{city}当前天气: {data['condition']}, 温度: {data['temp']}°C"
            else:
                return f"无法获取{city}的天气信息: {data.get('message', '未知错误')}"

        except Exception as e:
            return f"查询天气时出错: {str(e)}"

    async def _arun(self, city: str):
        """异步版本（可选）"""
        raise NotImplementedError("此工具不支持异步调用")

代码解析：

  1. BaseModel定义了工具的输入参数结构
  2. BaseTool是所有工具的基类
  3. name和description非常重要 – LLM通过这些信息决定何时使用该工具
  4. _run方法是核心逻辑所在

第二步：集成到LangChain代理中

创建好工具后，我们需要将其添加到代理中：

代码片段 
from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI

# 初始化LLM (这里使用ChatGPT)
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")

# 创建工具实例
weather_tool = WeatherTool()

# 初始化代理
agent = initialize_agent(
    tools=[weather_tool],
    llm=llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试运行
response = agent.run("上海现在的天气怎么样？")
print(response)

第三步：处理复杂API响应

当API返回复杂数据结构时，我们需要适当处理：

代码片段 
class StockInfoInput(BaseModel):
    symbol: str = Field(..., description="股票代码，如AAPL代表苹果公司")

class StockTool(BaseTool):
    name = "stock_checker"
    description = "查询股票实时信息"
    args_schema: Type[BaseModel] = StockInfoInput

    def _run(self, symbol: str):
        try:
            # 模拟股票API调用
            response = requests.get(f"https://api.example.com/stocks/{symbol}")
            data = response.json()

            if response.status_code == 200:
                # 格式化复杂响应为易读字符串
                info = [
                    f"公司: {data['companyName']}",
                    f"当前价格: ${data['latestPrice']}",
                    f"今日变动: {data['changePercent']*100:.2f}%",
                    f"市值: ${data['marketCap']/1e9:.2f}B",
                    f"52周最高/最低: ${data['week52High']}/{data['week52Low']}"
                ]
                return "\n".join(info)
            else:
                return f"无法获取{symbol}的股票信息"

        except Exception as e:
            return f"查询股票时出错: {str(e)}"

API集成的实践经验与注意事项

  1. 错误处理：

    • API可能不可用或返回错误格式
    • 始终检查HTTP状态码和响应内容

  2. 速率限制：

    • API通常有调用频率限制
    • 考虑添加缓存机制减少调用次数

  3. 认证安全：

    • API密钥等敏感信息不要硬编码在代码中
    • 使用环境变量存储凭据：
      代码片段 
      import os
api_key = os.getenv("WEATHER_API_KEY")<br>

  4. 响应格式化：

    • LLM处理结构化文本比JSON更好
    • 将复杂数据转换为自然语言描述

  5. 描述优化：

    代码片段 
    description = """
查询指定城市的实时天气情况。
输入应为城市名称字符串。
输出包含天气状况和温度信息。
"""

高级技巧：动态参数工具

有些API需要动态参数，我们可以这样实现：

代码片段 
class CustomAPITool(BaseTool):
    name = "custom_api"

    def __init__(self, api_config):
        super().__init__()
        self.description = api_config["description"]
        self.api_endpoint = api_config["endpoint"]

    def _run(self, **kwargs):
        params = {k:v for k,v in kwargs.items() if v is not None}
        response = requests.get(self.api_endpoint, params=params)
        return self._format_response(response.json())

def create_dynamic_tools(api_configs):
    """根据配置动态创建多个工具"""
    return [CustomAPITool(config) for config in api_configs]

# 使用示例：
apis = [
    {
        "description": "航班查询",
        "endpoint": "https://api.example.com/flights"
    },
    {
        "description": "酒店搜索",
        "endpoint": "https://api.example.com/hotels"
    }
]

dynamic_tools = create_dynamic_tools(apis)

LangChain工具的替代方案：Function Calling

如果你使用的是支持Function Calling的模型(如GPT-4)，可以直接利用该特性：

代码片段 
from langchain.chat_models import ChatOpenAI

def get_current_weather(location, unit="celsius"):
    """获取指定位置的当前天气"""
    # ...实现同上...

llm_with_tools = ChatOpenAI(model="gpt-4").bind(
    functions=[{
        "name": "get_current_weather",
        "description": "获取指定位置的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "城市和地区"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }]
)

总结与最佳实践

通过本文我们学习了：

  1. LangChain自定义工具的完整开发流程 ✅
  2. API集成中的关键注意事项 🚨
  3. Python实现的各种技巧和模式 💡

最佳实践建议：

  • 保持工具单一职责：每个工具只做一件事并做好它
  • 编写清晰的描述：帮助LLM准确判断何时使用该工具
  • 限制权限范围：只授予必要的API访问权限
  • 监控API使用：记录所有外部调用以便调试

完整的示例代码可以在GitHub仓库中找到：[示例仓库链接]

现在你已经掌握了LangChain自定义工具的开发和API集成方法，试着为你自己的项目创建一些有用的工具吧！

原创 高质量