第五章：实战项目 — Text2SQL

发表于 2026-05-08 分类于 python ， rd Changyan：本文字数： 2.7k 阅读时长 ≈ 10 分钟

本节课通过Text2SQL实战项目，用三大框架分别实现”自然语言查数据库”，在真实场景中感受框架差异与选型要点。

5.1 项目概述

目标：用户输入自然语言问题（如”哪个音乐类型的歌曲最长？”），Agent 自动查询数据库并返回结果。

使用数据库：Chinook SQLite（示例音乐商店数据库，11 张表）

技术栈：

Python 3.11+
SQLite
LangChain / LangGraph / DeepAgents

数据库 Schema 概览：

Album ← Artist
  ↓
Track ← Genre, MediaType
  ↓
InvoiceLine → Invoice → Customer
  ↓
PlaylistTrack → Playlist
Employee → Customer

安装依赖：

1	pip install langchain langgraph deepagents langchain-openai langchain-community

下载数据库：

import requests, pathlib

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

if not local_path.exists():
    response = requests.get(url)
    if response.status_code == 200:
        local_path.write_bytes(response.content)

5.2 LangChain 实现

"""
Text2SQL — LangChain 实现
特点：简单直接，但缺乏流程控制和状态管理
"""
import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# 1. 连接数据库
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(f"可用表: {db.get_usable_table_names()}")

# 2. 初始化模型和工具包
llm = ChatOpenAI(model="gpt-4o", temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()

# 3. 定义提示词
system_prompt = """你是一个 SQL 数据库查询助手。
给定用户的自然语言问题，请：
1. 先查看数据库中有哪些表
2. 查看相关表的结构
3. 编写正确的 SQL 查询
4. 执行查询并返回结果

注意：
- 只使用 SELECT 语句，不要使用 INSERT/UPDATE/DELETE
- 查询结果限制最多 10 条
- 如果查询出错，请修正后重试
"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# 4. 创建 Agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True,
)

# 5. 运行
question = "哪个音乐类型的歌曲平均时长最长？"
result = agent_executor.invoke({"input": question})
print(f"\n问题: {question}")
print(f"回答: {result['output']}")

LangChain 实现的问题：

问题	原因
无法强制先查看表结构	完全依赖 LLM 自主决定调用哪些工具
无法自动检查 SQL 正确性	没有”检查”步骤，直接执行
无人工审批	AgentExecutor 不支持 interrupt
错误恢复有限	只能靠 `handle_parsing_errors`
无状态持久化	对话结束后状态丢失

5.3 LangGraph 实现

"""
Text2SQL — LangGraph 实现
特点：精确控制每个步骤，支持查询检查和人工审批
"""
import os
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, SystemMessage
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver

# 1. 连接数据库
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()

# 获取具体工具引用
list_tables_tool = next(t for t in tools if t.name == "sql_db_list_tables")
get_schema_tool = next(t for t in tools if t.name == "sql_db_schema")
query_tool = next(t for t in tools if t.name == "sql_db_query")
query_checker_tool = next(t for t in tools if t.name == "sql_db_query_checker")

# 2. 定义节点 — 每个步骤精确控制

def list_tables(state: MessagesState):
    """第一步：强制列出所有表（不由 LLM 决定）"""
    tool_call = {
        "name": "sql_db_list_tables",
        "args": {},
        "id": "list_tables_1",
        "type": "tool_call",
    }
    tool_call_msg = AIMessage(content="", tool_calls=[tool_call])
    tool_msg = list_tables_tool.invoke(tool_call)
    response = AIMessage(content=f"可用表: {tool_msg.content}")
    return {"messages": [tool_call_msg, tool_msg, response]}

def call_get_schema(state: MessagesState):
    """第二步：让 LLM 决定查看哪些表的结构"""
    llm_with_schema = llm.bind_tools([get_schema_tool], tool_choice="any")
    response = llm_with_schema.invoke(state["messages"])
    return {"messages": [response]}

get_schema_node = ToolNode([get_schema_tool], name="get_schema")

def generate_query(state: MessagesState):
    """第三步：生成 SQL 查询"""
    system_prompt = f"""你是一个 SQL 专家。根据用户的自然语言问题，
    编写正确的 {db.dialect} SQL 查询。

    规则：
    - 只使用 SELECT 语句
    - 限制最多返回 10 条结果
    - 只查询相关列，不要 SELECT *
    - 不要使用 DML 语句（INSERT/UPDATE/DELETE）
    """
    system_msg = SystemMessage(content=system_prompt)
    llm_with_query = llm.bind_tools([query_tool])
    response = llm_with_query.invoke([system_msg] + state["messages"])
    return {"messages": [response]}

def check_query(state: MessagesState):
    """第四步：检查 SQL 查询是否正确"""
    check_prompt = f"""你是 SQL 专家。检查以下 {db.dialect} 查询是否有常见错误：
    - NOT IN 与 NULL 值的问题
    - UNION 与 UNION ALL 的误用
    - 数据类型不匹配
    - 函数参数数量错误

    如果有错误，修正查询。如果没有错误，保持原样。"""

    last_msg = state["messages"][-1]
    tool_call = last_msg.tool_calls[0]
    user_msg = {"role": "user", "content": tool_call["args"]["query"]}

    llm_with_checker = llm.bind_tools([query_tool], tool_choice="any")
    response = llm_with_checker.invoke([
        SystemMessage(content=check_prompt),
        user_msg,
    ])
    response.id = last_msg.id
    return {"messages": [response]}

run_query_node = ToolNode([query_tool], name="run_query")

# 3. 定义条件边
def should_continue(state: MessagesState) -> Literal["check_query", "__end__"]:
    """判断 LLM 是否想调用工具"""
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "check_query"
    return END

# 4. 构建图
builder = StateGraph(MessagesState)

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

builder.add_edge(START, "list_tables")
builder.add_edge("list_tables", "call_get_schema")
builder.add_edge("call_get_schema", "get_schema")
builder.add_edge("get_schema", "generate_query")
builder.add_conditional_edges("generate_query", should_continue)
builder.add_edge("check_query", "run_query")
builder.add_edge("run_query", "generate_query")

agent = builder.compile(checkpointer=InMemorySaver())

# 5. 运行
question = "哪个音乐类型的歌曲平均时长最长？"

for step in agent.stream(
    {"messages": [{"role": "user", "content": question}]},
    config={"configurable": {"thread_id": "1"}},
    stream_mode="values",
):
    last_msg = step["messages"][-1]
    last_msg.pretty_print()

执行流程图：

flowchart TD
START([START]) --> LT["list_tables<br/>强制列出所有表"]
LT --> CGS["call_get_schema<br/>LLM 决定查看哪些表"]
CGS --> GS["get_schema<br/>获取表结构"]
GS --> GQ["generate_query<br/>生成 SQL"]
GQ --> SC{有工具调用?}
SC -->|否| END([END])
SC -->|是| CQ["check_query<br/>检查 SQL 正确性"]
CQ --> RQ["run_query<br/>执行 SQL"]
RQ --> GQ

style START fill:#e8f5e9
style END fill:#ffebee
style LT fill:#e3f2fd
style CGS fill:#e3f2fd
style GS fill:#fff9c4
style GQ fill:#e3f2fd
style CQ fill:#f3e5f5
style RQ fill:#fff9c4

5.4 DeepAgents 实现

"""
Text2SQL — DeepAgents 实现
特点：极简代码，内置规划、文件系统、上下文管理
"""
import os
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

# 1. 定义数据库工具

def list_tables() -> str:
    """列出数据库中所有可用的表名。
    当你需要了解数据库结构时使用此工具。"""
    from langchain_community.utilities import SQLDatabase
    db = SQLDatabase.from_uri("sqlite:///Chinook.db")
    tables = db.get_usable_table_names()
    return f"可用表: {', '.join(tables)}"

def get_table_schema(table_names: str) -> str:
    """获取指定表的结构和样例数据。
    参数 table_names: 逗号分隔的表名，如 'Album,Artist'
    在编写 SQL 之前，务必先调用此工具了解表结构。"""
    from langchain_community.utilities import SQLDatabase
    db = SQLDatabase.from_uri("sqlite:///Chinook.db")
    return db.get_table_info_no_throw(table_names.split(","))

def run_sql_query(query: str) -> str:
    """执行 SQL 查询并返回结果。
    参数 query: 要执行的 SQL SELECT 语句
    注意：只执行 SELECT 查询，不执行修改操作。
    如果查询出错，会返回错误信息。"""
    from langchain_community.utilities import SQLDatabase
    db = SQLDatabase.from_uri("sqlite:///Chinook.db")
    try:
        result = db.run_no_throw(query)
        return result if result else "查询返回空结果"
    except Exception as e:
        return f"查询错误: {str(e)}"

def check_sql_query(query: str) -> str:
    """检查 SQL 查询是否有常见错误。
    参数 query: 要检查的 SQL 语句
    在执行查询前，使用此工具检查 SQL 是否正确。"""
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    check_prompt = f"""检查以下 SQL 查询是否有常见错误：
- NOT IN 与 NULL 值问题
- UNION vs UNION ALL 误用
- 数据类型不匹配
- 函数参数错误
- JOIN 条件不正确

SQL: {query}

如果有错误，返回修正后的 SQL。如果没有错误，返回 'OK'。"""
    response = llm.invoke(check_prompt)
    return response.content

# 2. 创建 Deep Agent
agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[list_tables, get_table_schema, run_sql_query, check_sql_query],
    system_prompt="""你是一个专业的 SQL 数据库查询助手。

## 工作流程
1. 用户提出问题后，先用 list_tables 了解数据库有哪些表
2. 用 get_table_schema 查看相关表的结构
3. 编写 SQL 查询，用 check_sql_query 检查正确性
4. 用 run_sql_query 执行查询
5. 用自然语言解释查询结果

## 注意事项
- 只使用 SELECT 语句，不要修改数据
- 查询结果限制 10 条
- 如果查询出错，分析错误并修正后重试
- 使用虚拟文件系统保存中间结果（如复杂查询的拆解步骤）
""",
    checkpointer=InMemorySaver(),
)

# 3. 运行
question = "哪个音乐类型的歌曲平均时长最长？"

result = agent.invoke(
    {"messages": [{"role": "user", "content": question}]},
    config={"configurable": {"thread_id": "text2sql-1"}},
)
print(result["messages"][-1].content)

# 4. 测试更复杂的问题
complex_questions = [
    "列出每个国家的客户数量和总消费额，按消费额降序排列",
    "找到购买了最多种类音乐的客户，列出其购买的所有音乐类型",
]

for q in complex_questions:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": q}]},
        config={"configurable": {"thread_id": "text2sql-complex"}},
    )
    print(f"\nQ: {q}")
    print(f"A: {result['messages'][-1].content}")

DeepAgents 自动做的事情（你不需要写代码）：

flowchart TD
A[用户提问] --> B[write_todos 自动规划]
B --> C[待办列表]
C --> D1["✅ 1. 列出数据库表"]
C --> D2["✅ 2. 查看 Genre 和 Track 表结构"]
C --> D3["✅ 3. 编写 SQL 查询"]
C --> D4["✅ 4. 检查 SQL 正确性"]
C --> D5["✅ 5. 执行查询"]
C --> D6["✅ 6. 解释结果"]

D1 --> E[上下文管理 自动]
D2 --> E
D3 --> E
D4 --> E
D5 --> E
D6 --> F[智能决策 自动]

E --> E1[大的工具结果自动卸载]
E --> E2[对话过长时自动摘要]

F --> F1[自主决定调用哪些工具]
F --> F2[出错时自动修正重试]
F --> F3[虚拟文件系统保存中间结果]

style B fill:#e3f2fd
style E fill:#fff9c4
style F fill:#e8f5e9

5.5 三框架实现对比

维度	LangChain	LangGraph	DeepAgents
代码行数	~50 行	~120 行	~60 行
工具定义	使用 SQLDatabaseToolkit	使用 SQLDatabaseToolkit	自定义工具函数
流程控制	无（LLM 完全自主）	精确（每个节点定义）	半自主（提示词引导）
SQL 检查	无	有（check_query 节点）	有（check_sql_query 工具）
人工审批	不支持	支持（interrupt）	支持（interrupt_on）
错误处理	有限（handle_parsing_errors）	节点级别	工具级别 + 自动重试
状态持久化	无	有（Checkpointer）	有（继承自 LangGraph）
上下文管理	无	无	有（自动压缩 + 卸载）
规划能力	无	无	有（write_todos）
子任务委派	无	手动实现	有（内置 task 工具）

5.6 Skill 版本：从教程到可加载能力

上面的 5.4 节是”在代码中手动写工具和提示词”的传统方式。DeepAgents 的 Skills 系统 让你可以将整个 Text2SQL 工作流封装为一个可复用的技能包，Agent 按需加载即可获得完整能力。

两种方式对比

维度	传统方式（5.4 节）	Skill 方式
工具定义	在代码中逐个定义	打包在 `tools.py` 中
工作流指引	写在 system_prompt 里	写在 `SKILL.md` 里
SQL 模式参考	无（靠 LLM 自行判断）	`patterns.md` 提供模式库
示例查询	无	`examples/` 目录提供场景示例
加载方式	手动传 tools 参数	`skills=["/skills/text2sql/"]`
复用性	每次重写	一次打包，到处加载
渐进式披露	全量加载	启动时只读 frontmatter，匹配后才加载完整内容

Skill 目录结构

skills/text2sql/
├── SKILL.md               # 技能元数据 + 工作流指令
├── tools.py               # 工具函数（list_tables, get_table_schema 等）
├── patterns.md            # SQL 模式参考与常见陷阱
├── usage_example.py       # 使用示例
└── examples/
    ├── ecommerce.md       # 电商场景示例（Chinook 数据库）
    └── analytics.md       # 分析场景示例（留存、RFM 等）

使用 Skill 创建 Text2SQL Agent

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

# 一行配置，Agent 自动获得 Text2SQL 全部能力
agent = create_deep_agent(
    model="openai:gpt-4o",
    skills=["/skills/text2sql/"],  # 🎯 只需指定技能路径
    checkpointer=InMemorySaver(),
)

# 直接使用，无需手动定义 tools 和 system_prompt
result = agent.invoke(
    {"messages": [{"role": "user", "content": "哪个音乐类型的歌曲平均时长最长？"}]},
    config={"configurable": {"thread_id": "skill-demo"}},
)

渐进式披露的运行机制

flowchart TD
A["🚀 Agent 启动"] --> B["读取 SKILL.md frontmatter<br/>(name + description)"]
B --> C["只消耗 ~50 tokens"]
C --> D["👤 用户提问"]
D --> E{"技能 description 匹配？"}
E -->|是| F["加载完整 SKILL.md + patterns.md"]
E -->|否| G["不加载，节省 tokens"]
F --> H["Agent 按工作流执行<br/>1. 列表 → 2. Schema → 3. SQL → 4. 检查 → 5. 执行"]
H --> I["返回结果"]

style B fill:#e3f2fd
style E fill:#fff9c4
style F fill:#e8f5e9
style G fill:#ffebee

完整 Skill 包 位于 skills/text2sql/ 目录，可直接在 DeepAgents 项目中使用。