Prompt实战之SQL问答系统 2）引入人类审查流程（Human-in-the-Loop）

在第一章中，我们成功构建了一个完整的 SQL 问答系统。但在真实世界中，模型自动生成 SQL 的行为可能存在风险，如：

• 查询逻辑错误；
• 涉及敏感表或字段；
• 消耗过多计算资源；
• 用户输入被恶意构造为攻击型语句（SQL 注入）。

因此，在系统进入生产环境之前，我们应在关键步骤加入 人工审批（Human-in-the-Loop, HITL）机制。

---

2.1 审查流程结构

我们将在“生成 SQL 语句”与“执行查询”之间插入人工审查步骤。整体架构更新如下：

graph TD
    A[用户问题] --> B[生成 SQL 查询]
    B --> C{人工审查？}
    C -- 通过 --> D[执行查询]
    C -- 拒绝 --> E[返回提示信息]
    D --> F[生成回答]

---

2.2 实现人工审查节点

LangGraph 支持流程中的分支判断。我们用一个分支节点 check_query 判断是否需要人工审批：

from langgraph.graph import StateGraph, State, END

def check_query(state: State) -> str:
    query = state["query"].lower()

    # 简单规则判断是否需要审查
    if "drop " in query or "delete " in query or "update " in query:
        return "needs_approval"
    if "employee" in query:  # 模拟某些敏感表名
        return "needs_approval"
    
    return "no_approval"

---

2.3 人工审批节点（模拟）

我们可以暂时用控制台输入模拟人类行为。后续也可以替换为 Web 审批、Slack 通知等更复杂系统。

def approval_node(state: State):
    print("===== 审核环节 =====")
    print(f"原始问题: {state['question']}")
    print(f"生成 SQL: {state['query']}")
    decision = input("是否允许执行该查询？(y/n): ").strip().lower()

    if decision == "y":
        return {"approved": True}
    else:
        return {"approved": False}

如果审核被拒绝，我们返回提示：

def approval_failed(_: State):
    return {"answer": "您的问题触发了安全机制，无法继续执行查询。"}

---

2.4 更新 LangGraph 流程

我们将审批流程整合进原始流程图：

graph_builder = StateGraph(State)

# 原始步骤
graph_builder.add_node("write_query", write_query)
graph_builder.add_node("check_query", check_query)
graph_builder.add_node("approval", approval_node)
graph_builder.add_node("execute_query", execute_query)
graph_builder.add_node("generate_answer", generate_answer)
graph_builder.add_node("approval_failed", approval_failed)

# 节点连接
graph_builder.set_entry_point("write_query")
graph_builder.add_edge("write_query", "check_query")
graph_builder.add_conditional_edges(
    "check_query",
    {
        "no_approval": "execute_query",
        "needs_approval": "approval",
    },
)
graph_builder.add_conditional_edges(
    "approval",
    {
        "__else__": "approval_failed",
        "approved": "execute_query",
    }
)
graph_builder.add_edge("execute_query", "generate_answer")
graph_builder.add_edge("generate_answer", END)

graph = graph_builder.compile()

---

2.5 测试审查流程

for step in graph.stream({"question": "Show all employees"}, stream_mode="updates"):
    print(step)

当模型尝试查询 Employee 表时，系统会中断流程并进入人工审批节点。

---

小贴士：部署场景中的 HITL 扩展方案

• ✅ 审批请求发送到 Web 后台或审批系统；
• ✅ 审核流程延时可缓存用户状态（例如使用 Redis）；
• ✅ 可记录审批日志，供未来模型训练使用；
• ✅ 结合 LangSmith 实现可观测性和回溯。

---

本章小结

通过为关键流程加入人工判断机制，我们完成了更安全、健壮的 SQL 问答系统构建：

• ✔️ 使用条件分支插入审批节点；
• ✔️ 实现人工干预机制（人机协同）；
• ✔️ 完成对潜在风险查询的管控。

---

练习区

1. 修改 check_query，使其对所有 SELECT * 语句强制审批；
2. 将 approval_node 中的 input 替换为自动审批逻辑（例如对“safe”关键词自动放行）；
3. 扩展 approval_failed，返回对用户更有帮助的回复（例如建议问题改写）。