SQL 代理

原文：SQL Agent

一句话

使用 LangGraph 构建自定义 SQL 代理，能够回答关于 SQL 数据库的问题。

什么时候翻这页

当你需要构建一个能够与 SQL 数据库交互的自定义代理，并且需要比预构建代理更精细的控制时。

核心概念

Tools (工具)：用于从 SQL 数据库读取的工具
LangGraph Graph API：包括 state (状态)、nodes (节点)、edges (边) 和 conditional edges (条件边)
Human-in-the-loop (人机交互)：在执行 SQL 查询前进行人工审查的流程

怎么做

1. 安装依赖

pip install langchain langgraph

2. 配置 LangSmith

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

3. 选择 LLM

选择支持工具调用的模型，如 OpenAI、Anthropic、Azure、Google Gemini、AWS Bedrock、HuggingFace 或 OpenRouter。

4. 配置数据库

使用 SQLite 数据库作为示例：

import sqlite3
con = sqlite3.connect("Chinook.db")

5. 添加数据库交互工具

使用 @tool 装饰器创建数据库工具：

@tool
def sql_db_list_tables() -> str:
    """列出数据库中的表"""
    # 实现细节

@tool
def sql_db_schema(table_names: str) -> str:
    """获取表的架构和示例行"""
    # 实现细节

@tool
def sql_db_query(query: str) -> str:
    """执行 SQL 查询"""
    # 实现细节

6. 定义应用步骤

创建专用节点处理不同步骤：

列出数据库表
调用"获取架构"工具
生成查询
检查查询

7. 实现代理

使用 StateGraph 构建工作流：

from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

def should_continue(state: MessagesState) -> Literal[END, "check_query"]:
    # 实现条件逻辑

builder = StateGraph(MessagesState)
builder.add_node(list_tables)
builder.add_node(call_get_schema)
builder.add_node(get_schema_node, "get_schema")
builder.add_node(generate_query)
builder.add_node(check_query)
builder.add_node(run_query_node, "run_query")

# 添加边和条件边
# ...

agent = builder.compile()

8. 实现人机循环审查

使用 LangGraph 的中断功能在执行 SQL 查询前进行人工审查：

from langgraph.types import interrupt
from langgraph.checkpoint.memory import InMemorySaver

@tool
def run_query_tool_with_interrupt(config: RunnableConfig, **tool_input):
    request = {
        "action": run_query_tool.name,
        "args": tool_input,
        "description": "Please review the tool call",
    }
    response = interrupt([request])
    # 处理不同类型的响应
    # ...

命令 / API 速查

工具创建

@tool
def tool_name(params) -> return_type:
    """工具描述"""
    # 实现

StateGraph 创建

from langgraph.graph import StateGraph, END, START

builder = StateGraph(MessagesState)
builder.add_node(node_function, node_name)
builder.add_edge(source_node, target_node)
builder.add_conditional_edges(
    source_node,
    condition_function,
    {
        "option1": "target_node1",
        "option2": "target_node2",
    }
)
graph = builder.compile(checkpointer=InMemorySaver())

中断功能

from langgraph.types import interrupt, Command

# 创建中断请求
request = {
    "action": tool_name,
    "args": tool_args,
    "description": "请审查此工具调用"
}
response = interrupt([request])

# 恢复执行
Command(resume={"type": "accept"})  # 或 "edit" 或 "response"

与 Hello-Agents / Claude Code 的联系

与 Hello-Agents 中 LangGraph 章节相比，本教程展示了如何构建更复杂的 SQL 代理工作流，包括自定义节点、条件边和人机交互功能。Hello-Agents 提供了 LangGraph 的基础概念，而本教程展示了如何将这些概念应用于实际的数据库交互场景。

初学者易错点

数据库权限：确保数据库连接权限尽可能最小化，仅授予代理所需的权限
SQL 注入风险：不要信任模型生成的 SQL，始终进行验证
工具调用顺序：确保按照正确的顺序调用工具（先列出表，再获取架构，然后生成查询）
中断处理：在使用人机交互功能时，正确处理不同类型的响应（接受、编辑、响应）

语义检索