LangChain 中文文档

主题

构建自定义 SQL 代理

在本教程中，我们将使用 LangGraph 构建一个能够回答关于 SQL 数据库问题的自定义智能体（agent）。

LangChain 提供了内置的智能体实现，这些实现使用了 LangGraph 的原语。如果需要更深度的定制，可以直接在 LangGraph 中实现智能体。本指南演示了一个 SQL 智能体的示例实现。你也可以查看这里的教程，它使用更高层级的 LangChain 抽象来构建 SQL 智能体。

构建基于 SQL 数据库的问答系统需要执行模型生成的 SQL 查询。这样做存在固有的风险。请确保你的数据库连接权限始终根据智能体的需求被限制在尽可能小的范围内。这将减轻（尽管不能完全消除）构建模型驱动系统的风险。

预构建的智能体让我们能够快速开始，但我们依赖系统提示（system prompt）来约束其行为——例如，我们指示智能体总是从“列出表”工具开始，并且在执行查询之前总是运行查询检查器工具。

在 LangGraph 中，我们可以通过自定义智能体来实施更高程度的控制。在这里，我们实现一个简单的 ReAct 智能体设置，为特定的工具调用设置专用节点。我们将使用与预构建智能体相同的状态（state）。

概念 ​

我们将涵盖以下概念：

设置 ​

安装 ​

bash
pip install langchain  langgraph  langchain-community

LangSmith ​

设置 LangSmith 以检查你的链（chain）或智能体内部发生的情况。然后设置以下环境变量：

shell
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

1. 选择 LLM ​

选择一个支持工具调用（tool-calling）的模型：

OpenAI
Anthropic
Azure
Google Gemini
AWS Bedrock
HuggingFace

👉 阅读 OpenAI 聊天模型集成文档

shell
pip install -U "langchain[openai]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-4.1")
python
import os
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-..."

model = ChatOpenAI(model="gpt-4.1")

下面示例中显示的输出使用了 OpenAI。

2. 配置数据库 ​

你将为本教程创建一个 SQLite 数据库。SQLite 是一个轻量级数据库，易于设置和使用。我们将加载 chinook 数据库，这是一个代表数字媒体商店的示例数据库。

为了方便起见，我们已将数据库 (Chinook.db) 托管在一个公共的 GCS 存储桶上。

python
import requests, pathlib

url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
local_path = pathlib.Path("Chinook.db")

if local_path.exists():
    print(f"{local_path} already exists, skipping download.")
else:
    response = requests.get(url)
    if response.status_code == 200:
        local_path.write_bytes(response.content)
        print(f"File downloaded and saved as {local_path}")
    else:
        print(f"Failed to download the file. Status code: {response.status_code}")

我们将使用 langchain_community 包中一个方便的 SQL 数据库包装器来与数据库交互。该包装器提供了一个简单的接口来执行 SQL 查询和获取结果：

python
from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///Chinook.db")

print(f"Dialect: {db.dialect}")
print(f"Available tables: {db.get_usable_table_names()}")
print(f'Sample output: {db.run("SELECT * FROM Artist LIMIT 5;")}')
Dialect: sqlite
Available tables: ['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
Sample output: [(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), (5, 'Alice In Chains')]

3. 添加用于数据库交互的工具 ​

使用 langchain_community 包中可用的 SQLDatabase 包装器来与数据库交互。该包装器提供了一个简单的接口来执行 SQL 查询和获取结果：

python
from langchain_community.agent_toolkits import SQLDatabaseToolkit

toolkit = SQLDatabaseToolkit(db=db, llm=model)

tools = toolkit.get_tools()

for tool in tools:
    print(f"{tool.name}: {tool.description}\n")
sql_db_query: 此工具的输入是一个详细且正确的 SQL 查询，输出是数据库的结果。如果查询不正确，将返回错误信息。如果返回错误，请重写查询、检查查询，然后重试。如果遇到“字段列表中的未知列 'xxxx'”问题，请使用 sql_db_schema 查询正确的表字段。

sql_db_schema: 此工具的输入是一个逗号分隔的表名列表，输出是这些表的模式和示例行。务必先调用 sql_db_list_tables 来确认这些表确实存在！示例输入：table1, table2, table3

sql_db_list_tables: 输入是一个空字符串，输出是数据库中表的逗号分隔列表。

sql_db_query_checker: 在执行查询之前，使用此工具再次检查你的查询是否正确。在使用 sql_db_query 执行查询之前，务必先使用此工具！

4. 定义应用步骤 ​

我们为以下步骤构建专用节点：

  • 列出数据库表
  • 调用“获取模式”工具
  • 生成查询
  • 检查查询

将这些步骤放在专用节点中，让我们能够（1）在需要时强制进行工具调用，以及（2）自定义与每个步骤关联的提示（prompt）。

python
from typing import Literal

from langchain.messages import AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

get_schema_tool = next(tool for tool in tools if tool.name == "sql_db_schema")
get_schema_node = ToolNode([get_schema_tool], name="get_schema")

run_query_tool = next(tool for tool in tools if tool.name == "sql_db_query")
run_query_node = ToolNode([run_query_tool], name="run_query")

# 示例：创建一个预定的工具调用
def list_tables(state: MessagesState):
    tool_call = {
        "name": "sql_db_list_tables",
        "args": {},
        "id": "abc123",
        "type": "tool_call",
    }
    tool_call_message = AIMessage(content="", tool_calls=[tool_call])

    list_tables_tool = next(tool for tool in tools if tool.name == "sql_db_list_tables")
    tool_message = list_tables_tool.invoke(tool_call)
    response = AIMessage(f"Available tables: {tool_message.content}")

    return {"messages": [tool_call_message, tool_message, response]}

# 示例：强制模型创建一个工具调用
def call_get_schema(state: MessagesState):
    # 注意，LangChain 强制要求所有模型都接受 `tool_choice="any"`
    # 以及 `tool_choice=<工具名称字符串>`。
    llm_with_tools = model.bind_tools([get_schema_tool], tool_choice="any")
    response = llm_with_tools.invoke(state["messages"])

    return {"messages": [response]}

generate_query_system_prompt = """
You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct {dialect} query to run,
then look at the results of the query and return the answer. Unless the user
specifies a specific number of examples they wish to obtain, always limit your
query to at most {top_k} results.

You can order the results by a relevant column to return the most interesting
examples in the database. Never query for all the columns from a specific table,
only ask for the relevant columns given the question.

DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.
""".format(
    dialect=db.dialect,
    top_k=5,
)

def generate_query(state: MessagesState):
    system_message = {
        "role": "system",
        "content": generate_query_system_prompt,
    }
    # 我们在这里不强制进行工具调用，以允许模型在获得解决方案时自然响应。
    llm_with_tools = model.bind_tools([run_query_tool])
    response = llm_with_tools.invoke([system_message] + state["messages"])

    return {"messages": [response]}

check_query_system_prompt = """
You are a SQL expert with a strong attention to detail.
Double check the {dialect} query for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins

If there are any of the above mistakes, rewrite the query. If there are no mistakes,
just reproduce the original query.

You will call the appropriate tool to execute the query after running this check.
""".format(dialect=db.dialect)

def check_query(state: MessagesState):
    system_message = {
        "role": "system",
        "content": check_query_system_prompt,
    }

    # 生成一个用于检查的人工用户消息
    tool_call = state["messages"][-1].tool_calls[0]
    user_message = {"role": "user", "content": tool_call["args"]["query"]}
    llm_with_tools = model.bind_tools([run_query_tool], tool_choice="any")
    response = llm_with_tools.invoke([system_message, user_message])
    response.id = state["messages"][-1].id

    return {"messages": [response]}

:::js

typescript

// 为模式和查询执行创建工具节点
const getSchemaNode = new ToolNode([getSchemaTool]);
const runQueryNode = new ToolNode([queryTool]);

// 示例：创建一个预定的工具调用
async function listTables(state: typeof MessagesAnnotation.State) {
  const toolCall = {
name: "sql_db_list_tables",
args: {},
id: "abc123",
type: "tool_call" as const,
  };
  const toolCallMessage = new AIMessage({
content: "",
tool_calls: [toolCall],
  });

  const toolMessage = await listTablesTool.invoke({});
  const response = new AIMessage(`Available tables: ${toolMessage}`);

  return { messages: [toolCallMessage, new ToolMessage({ content: toolMessage, tool_call_id: "abc123" }), response] };
}

// 示例：强制模型创建一个工具调用
async function callGetSchema(state: typeof MessagesAnnotation.State) {
  const llmWithTools = model.bindTools([getSchemaTool], {
tool_choice: "any",
  });
  const response = await llmWithTools.invoke(state.messages);

  return { messages: [response] };
}

const topK = 5;

const generateQuerySystemPrompt = `
You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct ${dialect}
query to run, then look at the results of the query and return the answer. Unless
the user specifies a specific number of examples they wish to obtain, always limit
your query to at most ${topK} results.

You can order the results by a relevant column to return the most interesting
examples in the database. Never query for all the columns from a specific table,
only ask for the relevant columns given the question.

DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.
`;

async function generateQuery(state: typeof MessagesAnnotation.State) {
  const systemMessage = new SystemMessage(generateQuerySystemPrompt);
  // 我们在这里不强制进行工具调用，以允许模型在获得解决方案时自然响应。
  const llmWithTools = model.bindTools([queryTool]);
  const response = await llmWithTools.invoke([

LangChain 中文文档

Copyright © 王码码