Langchain 自定义插件

文字转SQL

1. 安装必要的库

首先，确保你已经安装了所需的库：

pip install langchain sqlalchemy openai

2. 创建 SQL 查询工具

我们将创建一个自定义的 SQL 查询工具，该工具可以执行 SQL 查询并返回结果。这里我们使用 SQLite 数据库作为示例。

2.1 定义工具类

from langchain.tools import BaseTool
from sqlalchemy import create_engine, text

class SQLQueryTool(BaseTool):
    name = "sql_query_tool"
    description = "A tool that executes SQL queries and returns the results."

    def __init__(self, db_url: str):
        self.engine = create_engine(db_url)

    def run(self, query: str) -> str:
        with self.engine.connect() as connection:
            result = connection.execute(text(query))
            rows = result.fetchall()
            columns = result.keys()
            return [dict(zip(columns, row)) for row in rows]

2.2 初始化工具

db_url = "sqlite:///example.db"  # 替换为你的数据库 URL
sql_query_tool = SQLQueryTool(db_url)

3. 创建自然语言到 SQL 的转换工具

我们将使用一个简单的自然语言处理库（如 sqlglot）来将自然语言文本转换为 SQL 查询。

3.1 安装 sqlglot

pip install sqlglot

3.2 定义转换工具类

import sqlglot
from langchain.tools import BaseTool

class NLtoSQLTool(BaseTool):
    name = "nl_to_sql_tool"
    description = "A tool that converts natural language to SQL queries."

    def run(self, nl_query: str) -> str:
        try:
            sql_query = sqlglot.transpile(nl_query, read="english", write="sql")[0]
            return sql_query
        except Exception as e:
            return f"Error converting natural language to SQL: {e}"

3.3 初始化转换工具

nl_to_sql_tool = NLtoSQLTool()

4. 创建 LangChain 环境

我们将创建一个 LangChain 环境，并将两个工具添加到其中。

4.1 创建 Agent

我们将使用 OpenAI 的 LLM 和 ReAct 策略来创建一个 Agent。

from langchain.llms import OpenAI
from langchain.agents import initialize_agent, Tool

# 初始化 OpenAI LLM
llm = OpenAI(api_key="your_openai_api_key")  # 替换为你的 OpenAI API 密钥

# 创建工具列表
tools = [
    Tool(
        name=sql_query_tool.name,
        func=sql_query_tool.run,
        description=sql_query_tool.description
    ),
    Tool(
        name=nl_to_sql_tool.name,
        func=nl_to_sql_tool.run,
        description=nl_to_sql_tool.description
    )
]

# 初始化 Agent
agent = initialize_agent(tools, llm, agent="react-docstore", verbose=True)

5. 使用 Agent 执行自然语言查询

现在我们可以使用这个 Agent 来执行自然语言查询。

nl_query = "查询年龄大于25岁的用户"
response = agent.run(nl_query)
print(response)

6. 完整代码示例

以下是完整的代码示例，包含了上述所有步骤。

from langchain.tools import BaseTool
from sqlalchemy import create_engine, text
import sqlglot
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, Tool

# 定义 SQL 查询工具类
class SQLQueryTool(BaseTool):
    name = "sql_query_tool"
    description = "A tool that executes SQL queries and returns the results."

    def __init__(self, db_url: str):
        self.engine = create_engine(db_url)

    def run(self, query: str) -> str:
        with self.engine.connect() as connection:
            result = connection.execute(text(query))
            rows = result.fetchall()
            columns = result.keys()
            return [dict(zip(columns, row)) for row in rows]

# 初始化 SQL 查询工具
db_url = "sqlite:///example.db"  # 替换为你的数据库 URL
sql_query_tool = SQLQueryTool(db_url)

# 定义自然语言到 SQL 的转换工具类
class NLtoSQLTool(BaseTool):
    name = "nl_to_sql_tool"
    description = "A tool that converts natural language to SQL queries."

    def run(self, nl_query: str) -> str:
        try:
            sql_query = sqlglot.transpile(nl_query, read="english", write="sql")[0]
            return sql_query
        except Exception as e:
            return f"Error converting natural language to SQL: {e}"

# 初始化自然语言到 SQL 的转换工具
nl_to_sql_tool = NLtoSQLTool()

# 初始化 OpenAI LLM
llm = OpenAI(api_key="your_openai_api_key")  # 替换为你的 OpenAI API 密钥

# 创建工具列表
tools = [
    Tool(
        name=sql_query_tool.name,
        func=sql_query_tool.run,
        description=sql_query_tool.description
    ),
    Tool(
        name=nl_to_sql_tool.name,
        func=nl_to_sql_tool.run,
        description=nl_to_sql_tool.description
    )
]

# 初始化 Agent
agent = initialize_agent(tools, llm, agent="react-docstore", verbose=True)

# 使用 Agent 执行自然语言查询
nl_query = "查询年龄大于25岁的用户"
response = agent.run(nl_query)
print(response)

调用API

1. 自定义插件的基本概念

LangChain 的插件机制允许你通过 Python 类来封装特定的功能。一个插件通常包含以下几个部分：

• 初始化 (__init__)：用于设置插件所需的配置或依赖。
• 核心功能方法：实现插件的核心逻辑。
• 与 LangChain 的集成：通过 LangChain 提供的接口（如 Tool 或 Runnable）将插件注册到系统中。

2. 创建一个调用外部 API 的插件

假设我们有一个场景：需要调用一个天气 API 来获取某个城市的实时天气信息。我们将创建一个自定义插件来实现这一功能。

步骤 1: 定义插件类

首先，我们需要定义一个 Python 类来封装调用天气 API 的逻辑。

import requests

class WeatherAPIPlugin:
    def __init__(self, api_key):
        """
        初始化插件，设置 API 密钥。
        :param api_key: 天气 API 的密钥
        """
        self.api_key = api_key
        self.base_url = "https://api.weatherapi.com/v1/current.json"

    def get_weather(self, city):
        """
        获取指定城市的实时天气信息。
        :param city: 城市名称
        :return: 天气信息的字符串描述
        """
        params = {
            "key": self.api_key,
            "q": city
        }
        try:
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()  # 检查请求是否成功
            data = response.json()
            weather_info = (
                f"当前 {city} 的天气情况:\n"
                f"温度: {data['current']['temp_c']}°C\n"
                f"湿度: {data['current']['humidity']}%\n"
                f"天气状况: {data['current']['condition']['text']}"
            )
            return weather_info
        except requests.exceptions.RequestException as e:
            return f"无法获取天气信息: {str(e)}"

步骤 2: 将插件封装为 LangChain 的 Tool

LangChain 提供了 Tool 类，用于将自定义功能集成到链式逻辑中。我们需要将 WeatherAPIPlugin 的功能封装为一个 Tool。

from langchain.tools import Tool

# 假设你的 API 密钥如下
API_KEY = "your_weather_api_key_here"

# 实例化插件
weather_plugin = WeatherAPIPlugin(api_key=API_KEY)

# 将插件封装为 LangChain 的 Tool
weather_tool = Tool(
    name="Weather API Tool",
    func=weather_plugin.get_weather,
    description="获取指定城市的实时天气信息。输入应为城市名称，例如 '北京'。"
)

步骤 3: 使用插件

现在我们可以将这个插件集成到 LangChain 的工作流中。以下是一个简单的示例，展示如何调用插件并获取天气信息。

from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI

# 初始化 LLM（这里使用 OpenAI 的 GPT 模型）
llm = OpenAI(temperature=0, openai_api_key="your_openai_api_key_here")

# 初始化代理（Agent），并将插件添加到工具列表中
agent = initialize_agent(
    tools=[weather_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 调用插件
response = agent.run("请告诉我北京的天气情况。")
print(response)

3. 示例运行结果

假设天气 API 返回的数据如下：

{
  "current": {
    "temp_c": 15,
    "humidity": 60,
    "condition": {
      "text": "晴朗"
    }
  }
}

运行上述代码后，输出可能如下：

当前 北京 的天气情况:
温度: 15°C
湿度: 60%
天气状况: 晴朗

智能家居插件

1. 背景知识：LangChain 的 Agent 和 Tool

LangChain 提供了一个灵活的架构，允许我们定义 Tool（工具）和 Agent（代理）。Tool 是封装特定功能的模块，而 Agent 则负责根据用户输入选择合适的 Tool 并执行任务。

对于家庭硬件的控制，我们可以将硬件接口封装为一个 Tool，然后通过 LangChain 的 Agent 来调用它。

2. 家庭硬件的典型场景

假设你有一个家庭自动化系统，其中包含以下硬件：

• 灯光控制器：可以打开或关闭灯光。
• 温度传感器：可以读取当前室内温度。
• 窗帘控制器：可以打开或关闭窗帘。

我们将这些硬件的功能封装为 LangChain 的 Tool，并通过自然语言交互来控制它们。

3. 创建自定义插件

3.1 硬件接口模拟

为了演示方便，我们假设有一个 Python API 可以与家庭硬件通信。以下是一个简单的模拟接口：

class HomeHardwareAPI:
    def __init__(self):
        self.light_status = "off"
        self.temperature = 22.5  # 模拟当前温度
        self.curtain_status = "closed"

    def toggle_light(self, action):
        if action.lower() == "on":
            self.light_status = "on"
            return "Light turned on."
        elif action.lower() == "off":
            self.light_status = "off"
            return "Light turned off."
        else:
            return "Invalid action."

    def get_temperature(self):
        return f"Current temperature is {self.temperature}°C."

    def control_curtains(self, action):
        if action.lower() == "open":
            self.curtain_status = "open"
            return "Curtains opened."
        elif action.lower() == "close":
            self.curtain_status = "closed"
            return "Curtains closed."
        else:
            return "Invalid action."

3.2 封装为 LangChain 工具

接下来，我们将上述硬件接口封装为 LangChain 的 Tool：

from langchain.agents import Tool
from langchain.chains.conversation.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent

# 假设我们已经有一个硬件 API 实例
home_hardware_api = HomeHardwareAPI()

# 定义工具函数
def toggle_light(action):
    return home_hardware_api.toggle_light(action)

def get_temperature():
    return home_hardware_api.get_temperature()

def control_curtains(action):
    return home_hardware_api.control_curtains(action)

# 将工具封装为 LangChain 的 Tool
tools = [
    Tool(
        name="Toggle Light",
        func=lambda action: toggle_light(action),
        description="Use this tool to turn the light on or off. Input should be 'on' or 'off'."
    ),
    Tool(
        name="Get Temperature",
        func=get_temperature,
        description="Use this tool to get the current room temperature."
    ),
    Tool(
        name="Control Curtains",
        func=lambda action: control_curtains(action),
        description="Use this tool to open or close the curtains. Input should be 'open' or 'close'."
    )
]

4. 初始化 Agent

LangChain 的 Agent 负责解析用户的自然语言输入并调用相应的 Tool。以下是初始化 Agent 的代码：

# 初始化对话记忆
memory = ConversationBufferMemory(memory_key="chat_history")

# 使用 OpenAI 的 GPT-3.5-turbo 模型
llm = ChatOpenAI(temperature=0)

# 初始化 Agent
agent_chain = initialize_agent(
    tools, llm, agent="chat-conversational-react-description", verbose=True, memory=memory
)

5. 实际场景调用

5.1 用户输入示例

假设用户通过自然语言与系统交互，以下是一些可能的输入及其响应：

# 示例 1: 打开灯光
response = agent_chain.run("Please turn the light on.")
print(response)  # 输出: Light turned on.

# 示例 2: 获取当前温度
response = agent_chain.run("What is the current temperature?")
print(response)  # 输出: Current temperature is 22.5°C.

# 示例 3: 打开窗帘
response = agent_chain.run("Can you open the curtains?")
print(response)  # 输出: Curtains opened.

# 示例 4: 关闭灯光
response = agent_chain.run("Turn the light off.")
print(response)  # 输出: Light turned off.

6. 运行结果分析

• 当用户输入“Please turn the light on.”时，Agent 会识别出这是对灯光的操作请求，并调用 toggle_light 工具。
• 当用户输入“What is the current temperature?”时，Agent 会调用 get_temperature 工具返回当前温度。
• 类似地，其他命令也会被正确解析并执行。

7. 扩展功能

7.1 添加更多硬件支持

如果需要支持更多的硬件（例如空调、音响等），只需扩展 HomeHardwareAPI 类并添加相应的 Tool。

7.2 集成真实硬件

在实际应用中，你可以通过 MQTT、HTTP API 或其他协议与真实硬件通信。例如：

• 使用 MQTT 订阅/发布消息来控制灯光和窗帘。
• 通过 HTTP 请求获取传感器数据。

7.3 自然语言优化

为了让 Agent 更好地理解复杂的用户输入，可以训练一个定制化的语言模型，或者使用更高级的提示工程技巧。

8. 总结

通过 LangChain 的 Tool 和 Agent，我们可以轻松实现家庭硬件的智能化控制。本文展示了如何将硬件接口封装为 LangChain 的工具，并通过自然语言交互调用这些工具。希望这个示例能帮助你更好地理解和应用 LangChain！