Prompt实战之SQL问答系统3）从链到智能体 —— 构建 SQL Agent

在前两章中，我们通过 LangGraph 实现了 SQL 问答系统，并引入了人工审查流程。但这些流程仍是“线性”的，系统每一步都依赖于固定节点，缺乏灵活策略与自我规划能力。

为了解决这些局限，我们将在本章中构建一个具备 自主思考能力的 SQL Agent。它不仅能生成查询，还能决定是否需要用户澄清、进行多轮对话，甚至请求外部工具（如数据库接口）执行任务。

---

3.1 为什么需要 Agent？

相比于链式结构，Agent 更像是一个“智能体”，它在一个“行动-观察-反思”的循环中做出决策：

graph TD
    A[用户提问] --> B[Agent 分析]
    B --> C{需要工具吗？}
    C -- 是 --> D[调用工具]
    D --> E[观察结果]
    E --> B
    C -- 否 --> F[输出最终答案]

---

3.2 使用 LangChain Agent 构建 SQL 智能体

我们将使用 LangChain 提供的 Agent 架构，结合工具机制完成智能体的搭建。

步骤一：定义 SQL 工具

from langchain.tools import Tool

def run_sql_tool(query: str) -> str:
    # 假设你有一个 query_database 函数
    return query_database(query)

sql_tool = Tool(
    name="SQL Executor",
    func=run_sql_tool,
    description="用于执行 SQL 查询，并返回结果"
)

---

步骤二：定义 Agent 的 Prompt 模板

from langchain.agents import initialize_agent, AgentType

agent = initialize_agent(
    tools=[sql_tool],
    llm=chat_openai,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

通过 ZERO_SHOT_REACT_DESCRIPTION 类型的 Agent，系统会自动推理是否需要使用工具。Agent 会生成如下格式的思考链：

Thought: 我需要从数据库查询信息
Action: SQL Executor
Action Input: SELECT * FROM employees;
Observation: 返回了5条记录...
Final Answer: 员工列表如下：...

---

步骤三：调用 Agent 回答问题

agent.run("列出所有销售部门的员工")

Agent 会判断是否需要调用工具，然后生成 SQL、执行查询、处理结果并最终返回答案。

---

3.3 整合进 LangGraph 流程中

我们可以将 Agent 作为 LangGraph 的一个节点，并保留前面章节中构建的审批逻辑。

def agent_node(state: State) -> dict:
    question = state["question"]
    answer = agent.run(question)
    return {"answer": answer}

然后将其作为流程中的执行节点。

---

3.4 Agent 的优势与挑战

优势	挑战
自主判断是否使用工具	推理步骤更难调试
具备“思考 - 执行 - 观察”能力	对 Prompt 和工具定义更敏感
可轻松集成多个工具	增加响应延迟，需要优化性能
适合复杂推理、多轮对话任务	风控更难，需加强控制与审计

---

小贴士：让你的 SQL Agent 更强大

• ✅ 为 Agent 添加多个工具（如自然语言转 SQL、数据库执行器、表结构查看器）；
• ✅ 将中间 Thought 和 Observation 日志化，用于调试或训练；
• ✅ 使用 OpenAI Functions Agent 支持更复杂结构化输入/输出；
• ✅ 配合 LangGraph 的分支控制，实现“Agent 失败兜底逻辑”或“转人工审批”。

3.5 多智能体协作：SQL 生成器 + 审查官

在现实场景中，SQL 执行往往需要更高的准确性和安全性。为此，我们可以采用 两个 Agent 协作 的设计思路：

• 🧠 SQL 生成 Agent：负责理解用户意图并生成 SQL 查询。
• 🧐 SQL 审查 Agent：检查生成的 SQL 是否合理、安全，并提供反馈或修改建议。

这是一种“写 + 审”模式，能显著提高查询质量和系统可控性。

---

🤖 架构流程图

graph TD
    A[用户提问] --> B[SQL 生成 Agent]
    B --> C[SQL 审查 Agent]
    C -->|通过| D[SQL 执行工具]
    C -->|不通过| E[生成反馈，返回用户]
    D --> F[结果返回用户]

---

步骤一：定义两个 Agent

from langchain.agents import initialize_agent, Tool, AgentType
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI()

# Agent 1: SQL 生成器
def generate_sql(question: str) -> str:
    return llm.predict(f"请为以下问题生成 SQL 查询：{question}")

sql_generator_tool = Tool(
    name="SQL Generator",
    func=generate_sql,
    description="根据问题生成 SQL 查询语句"
)

# Agent 2: SQL 审查官
def review_sql(sql: str) -> str:
    return llm.predict(f"请检查以下 SQL 是否合理，若不合理请说明原因，否则回复 '通过'：{sql}")

sql_reviewer_tool = Tool(
    name="SQL Reviewer",
    func=review_sql,
    description="检查 SQL 查询语句是否合理"
)

---

步骤二：多 Agent 运行逻辑

def multi_agent_chain(user_question: str) -> str:
    sql = generate_sql(user_question)
    review = review_sql(sql)

    if "通过" in review:
        result = run_sql_tool(sql)
        return f"✅ 查询结果：{result}"
    else:
        return f"⚠️ 审查未通过：{review}"

---

步骤三：集成进 LangGraph

将整个多智能体结构包装成一个节点，插入到 LangGraph 的控制流中：

def multi_agent_node(state: State) -> dict:
    question = state["question"]
    answer = multi_agent_chain(question)
    return {"answer": answer}

---

小贴士：多智能体设计建议

• ✅ 为每个 Agent 明确任务边界：如“生成 SQL” vs “判断风险”；
• ✅ 审查 Agent 可使用专用 Prompt 加强安全检测（如注入风险、权限判断）；
• ✅ 可以扩展为三智能体：生成器 + 审查官 + 优化器；
• ✅ 审查未通过时引入“Human-in-the-loop”机制请求人工介入；

---

多智能体对比单 Agent

特性	单 Agent	多 Agent 协作
模型调用次数	一次	多次
安全性	一般	高（可审查、阻止非法 SQL）
开发复杂度	低	中等（需控制协调逻辑）
适用于	快速原型、低风险查询	企业系统、权限敏感数据访问

---

本章小结

通过本章内容，我们完成了从链式结构向智能体的跃迁，掌握了：

• ✔️ 使用 LangChain 构建具有工具调用能力的 Agent；
• ✔️ 实现 Agent 的 SQL 查询执行与结果生成；
• ✔️ 将 Agent 集成进 LangGraph 流程中；
• ✔️ 探索 Agent 的优缺点及可扩展性。

---

练习区

1. 为 Agent 增加一个新工具：解释 SQL 语句含义；
2. 修改 Prompt，使 Agent 在返回最终答案前总是展示“查询的 SQL”；
3. 结合前一章内容，在 Agent 调用工具前自动触发一次“人工审批”。
4. 将 SQL 审查 Agent 拓展为两个阶段：结构审查 + 权限审查；
5. 使用 LangGraph 的分支功能，在“审查未通过”时自动跳转至用户澄清节点；
6. 增加一个“SQL 优化 Agent”，尝试将冗余查询转换为更高效语句。