Langchain 构建SQL Agent中遇到的坑

1.什么是langchain

LangChain 是一个开源的软件开发框架，旨在简化由大型语言模型（LLM）（如 GPT-4、Claude、Llama 等）驱动的应用程序的开发过程。
简单来说，如果把大语言模型比作一个“大脑”，那么 LangChain 就是给这个大脑装上了“四肢”（工具）、“耳朵”（数据输入）和“记事本”（记忆），让它能够真正地去执行复杂的任务，而不仅仅是陪你聊天。

2.如何利用langchain来快速构建自己的sql-agent

2.1 安装必要的库

需要安装 LangChain 的核心库、OpenAI 接口库（推荐使用 OpenAI，效果最好）以及社区工具包

uv add langchain langchain-community langchain-openai langgraph

2.2 核心代码实现

为了能直接运行，下面的代码包含了一个临时创建的 SQLite 内存数据库。在实际生产中，你只需要把数据库连接字符串换成你的 MySQL 或 PostgreSQL 地址即可。

import os
from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table, insert

# 1. 导入 LangChain 相关模块
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent
from langchain_openai import ChatOpenAI

# ==========================================
# 准备工作：设置 API Key (请替换为你自己的 Key)
# ==========================================
os.environ["OPENAI_API_KEY"] = "sk-..." 

# ==========================================
# A. 创建一个模拟数据库 (仅用于演示，实际项目中不需要这一步)
# ==========================================
# 创建内存数据库
engine = create_engine("sqlite:///:memory:")
metadata = MetaData()

# 创建一个 'users' 表
users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("age", Integer),
    Column("city", String),
)
metadata.create_all(engine)

# 插入一些测试数据
with engine.connect() as conn:
    conn.execute(insert(users), [
        {"name": "Alice", "age": 25, "city": "New York"},
        {"name": "Bob", "age": 30, "city": "San Francisco"},
        {"name": "Charlie", "age": 35, "city": "New York"},
        {"name": "David", "age": 40, "city": "London"},
    ])
    conn.commit()

# ==========================================
# B. 连接数据库
# ==========================================
# 在实际项目中，URI 可能是 "postgresql://user:pass@localhost:5432/mydb"
db = SQLDatabase(engine=engine)

# ==========================================
# C. 初始化 LLM (大模型)
# ==========================================
# 建议使用 gpt-4 或 gpt-3.5-turbo，温度设为 0 以保证 SQL 生成的准确性
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# ==========================================
# D. 创建 SQL Agent
# ==========================================
# agent_type="openai-tools" 是目前最高效的模式
agent_executor = create_sql_agent(
    llm=llm,
    db=db,
    agent_type="openai-tools",
    verbose=True  # 开启后可以看到 Agent 的思考过程 (生成的 SQL 等)
)

# ==========================================
# E. 运行查询
# ==========================================
query = "来自 New York 的用户平均年龄是多少？"
print(f"用户提问: {query}")

response = agent_executor.invoke({"input": query})

print(f"\n最终回答: {response['output']}")

3.内置方法的弊端

虽然 create_sql_agent 和 SQLDatabase 等内置方法非常适合快速构建原型（Demo），但在生产环境中直接使用它们，通常会遇到很多严重的瓶颈和风险。

以下是 LangChain 内置 SQL Agent 方法的主要弊端：

3.1 Token 消耗与上下文限制

这是最常见的问题。内置方法通常会将数据库的 表结构（Schema） 直接放入Prompt中发送给LLM。

• 弊端：如果你的数据库有 100 张表，或者表中有几百个字段，Prompt 会瞬间变得非常长，甚至超过 LLM 的上下文窗口限制（Context Limit）。

这就会造成

• 每次查询都要发送几千个 Token 的表结构，API 费用极高（自己部署）。
• LLM 面对过多的无关表信息，注意力会被分散，导致推理能力下降，选错表或字段

3.2 缺乏领域知识

LLM 懂 SQL 语法，但它不懂业务逻辑。

• 弊端：数据库字段名往往是缩写或晦涩的（如 t_stat_01, is_del）。内置 Agent 只能根据字段名“猜”意思。

这就会造成生成的 SQL 语法正确，但业务结果是错的。

3.3 幻觉与错误

• 弊端：LLM 可能会臆造不存在的列名，或者混淆相似的列名
• 后果：虽然内置 Agent 有“自动纠错机制”（执行报错 -> 把错误发回 LLM -> LLM 重写 SQL），但这会增加延迟和 Token 消耗，而且不一定能修好。

4.如何解决

4.1 RAG for Schema

把表名和列的描述存入向量数据库。当用户提问时，先检索出最相关的 3-5 张表，只把这几张表的结构发给 LLM

4.2 构建自己的database tools

原有SQLDatabaseToolkit一次性会获取完整的表格数据，但是实际需要的字段数据不多，因此会造成过多无关字段、敏感数据或消耗大量Token，因此可以构建自己Database tools，工具主要针对以下进行优化

• 白名单/黑名单机制：只向 LLM 暴露特定的表和字段。
• 简化 Schema 格式：不发送完整的 CREATE TABLE 语句，只发送简单的 列名: 类型 描述。
• 禁止获取样本数据：内置工具通常会查询前 3 行数据作为样本，自定义工具可以完全禁止这一行为。

@tool
def get_safe_table_schema(table_name: str):
    """
    获取指定表的结构信息。
    只返回允许访问的字段，不包含敏感字段或样本数据。
    """
    if table_name not in VISIBLE_SCHEMA:
        return f"Error: Table '{table_name}' does not exist or access is denied."

    # 使用 SQLAlchemy Inspector 获取真实的列信息
    inspector = inspect(engine)
    columns = inspector.get_columns(table_name)
    
    # 过滤字段
    allowed_columns = VISIBLE_SCHEMA[table_name]
    safe_columns_info = []
    
    for col in columns:
        if col['name'] in allowed_columns:
            # 拼接成简化的格式: "name (VARCHAR)"
            col_str = f"{col['name']} ({col['type']})"
            safe_columns_info.append(col_str)
            
    # 构造返回给 LLM 的文本
    schema_text = f"Table: {table_name}\nColumns:\n" + "\n".join(f"- {c}" for c in safe_columns_info)
    
    return schema_text

总结

LangChain 的 create_sql_agent 就像一把瑞士军刀，能让我们在 5 分钟内快速验证“Text-to-SQL”的可行性。然而，当我们试图将其推向生产环境时，“开箱即用”的便利性往往伴随着不可控的风险。

从 Demo 到 Production 的跨越，本质上是一场关于 Context（上下文）与Control（控制权） 的博弈：

拒绝暴力填充：不再将整个数据库 Schema 一股脑塞给 LLM，而是通过 RAG（检索增强生成） 按需加载相关表，解决 Token 爆炸和注意力分散问题。
精细化权限管理：放弃粗粒度的内置工具，转而构建自定义 Database Tools。通过白名单机制和 Schema 简化，既保护了敏感数据（如密码、薪资），又大幅降低了 API 成本。

注入业务语义：LLM 懂 SQL 语法但不懂业务逻辑。通过自定义工具层，我们可以将复杂的业务规则（如“毛利计算方式”）封装在工具内部，而不是依赖模型去“猜”。
一句话概括：在构建企业级 SQL Agent 时，不要迷信 LangChain 的“魔法”，只有通过自定义工具将数据库的解释权和访问权牢牢掌握在自己手中，才能构建出既安全又精准的智能应用。