AI News
中文资讯 掘金 AI Wed, 09 Sep 2026 正文 ok 453 words

06 · 工具调用：Text2SQL 与四层护栏

英文标题：06 · 工具调用：Text2SQL 与四层护栏

本文介绍以 Schema 注册表统一管理 Agent 工具的方式，使新增工具仅需修改注册表，编排层无需改动。当前系统注册了 5 个工具，并通过 AST 白名单实现安全计算，避免使用 `eval`。Text2SQL 流程采用“生成、校验、执行、报错回炉”的闭环，结合 SQL 多语句、读写关键字、表白名单和 LIMIT 等四层护栏降低风险。对于高风险工具，系统通过 Human-in-the-loop 人工确认机制控制权限边界，并补充提示词注入检测、PII 脱敏与全链路可观测能力。
打开原文相关 3 · 机会 1
media3.2 KB media589 B media8.9 KB

中文

对应代码：app/agent/tools.py、app/agent/guardrails.py

一、工具即 Schema

Tool(
    name="sql_query",
    description="BI 数据查询（Text2SQL）：项目成本、缺陷趋势、交付效率等指标统计。只读。",
    parameters={"type": "object", "properties": {
        "question": {"type": "string", "description": "要查询的业务问题，中文描述即可"}},
        "required": ["question"]},
    fn=_tool_sql_query,
    risk="low",        
)

注册表对外提供三项能力：

registry.specs()                  
registry.describe_for_prompt()    
registry.execute(name, args, trace_id)

收益：新增工具只需修改一处，编排层无需改动。

二、当前已注册的 5 个工具

工具 风险 说明
search_knowledge low 知识库检索（RAG）
sql_query low BI 数据 Text2SQL
send_alert high 推送告警（需人工确认）
calculator low 安全四则运算
get_today low 解析“本月/上周”等相对时间

calculator 使用 AST 白名单实现（绝不使用 eval）：

_ALLOWED_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ...}
def _safe_eval(expr):
    return _eval(ast.parse(expr, mode="eval"))

三、Text2SQL：生成 → 校验 → 执行 → 报错回炉

for attempt in range(2):
    prompt = TEXT2SQL_TMPL.format(schema=schema, question=question, limit=row_limit)
    if last_err:
        prompt += f"\n上一次生成的 SQL 报错：{last_err}\n请修正后只输出 SQL。"
    raw_sql = llm.chat([{"role": "user", "content": prompt}], temperature=0)
    try:
        safe_sql = validate_sql(raw_sql)       
        return execute(safe_sql)
    except GuardrailError as e:
        last_err = str(e)

踩坑记录：第一版没有将数据库报错回传给模型，导致模型一旦生成错误的 SQL 字段名就会持续失败；加入“报错回炉”后，一次修正成功率显著提升。

Prompt 中的关键约束：

1. 只能输出一条 SELECT，不要解释、不要 markdown 代码块
2. 禁止 INSERT/UPDATE/DELETE/DROP/ATTACH/PRAGMA
3. 必须带 LIMIT 200
4. 日期字段为 TEXT（YYYY-MM-DD），本月使用 date('now','start of month')
5. 只能使用给定的表和字段
+ 一个少样本示例

四、四层 SQL 护栏

def validate_sql(sql, allowed_tables, row_limit, readonly=True):
    
    clean = strip_sql_comments(raw)
    
    if _MULTI_STMT.search(clean): raise GuardrailError("检测到多语句执行")
    
    if readonly and not low.startswith(("select", "with")): raise ...
    
    for kw in ("insert","update","delete","drop","attach","pragma",...):
        if re.search(rf"\b{kw}\b", low): raise ...
    
    tables = set(_TABLE_RE.findall(clean))
    if illegal := tables - allowed: raise GuardrailError(f"引用了未授权的表：{illegal}")
    
    if not re.search(r"\blimit\s+\d+", low): clean += f" LIMIT {row_limit}"

再增加一层数据库侧保护：只读连接。

con = sqlite3.connect(f"file:{db_path}?mode=ro")

（生产环境改用只读账号。）

实测拦截效果：

输入 结果
“删掉成本表” ✅ 被关键字黑名单拦截
“SELECT * FROM secret_salary” ✅ 被表白名单拦截
“SELECT ...; DROP TABLE x” ✅ 被多语句检测拦截

五、Human-in-the-loop

if tool.risk == "high" and settings.hitl_enabled and not auto_approve_high_risk:
    return ToolResult(False, error="HITL_REQUIRED", meta={"need_human": True})

前端收到 hitl 事件 → 渲染橙色确认卡片 → 用户点击“确认执行” → 携带 approved=true 重新请求 → 后端从 _pending[thread_id] 中取出待执行操作并执行。

这就是 JD 中所说的“权限边界”。

六、另外两个护栏

提示词注入检测

INJECTION_PATTERNS = [
    r"忽略(?:以上|上面|之前).{0,6}(?:指令|规则|提示)",
    r"ignore\s+(?:all\s+)?previous\s+instructions",
    r"输出(?:你的|系统).{0,4}(?:提示词|prompt)",
    ...
]

命中后不会中断服务，而是降级为“只读检索模式”，并发送一条 warn 事件——粗暴拒绝会损害正常用户体验。

PII 脱敏（出网前）

_PII_RULES = [
    (re.compile(r"1[3-9]\d{9}"), "[PHONE]"),          
    (re.compile(r"\b\d{17}[\dXx]\b"), "[ID_CARD]"),   
    (re.compile(r"\b\d{16,19}\b"), "[CARD]"),         
    (re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"), "[EMAIL]"),
    (re.compile(r"(?:sk-|api[_-]?key\s*[:=]\s*)[A-Za-z0-9_\-]{16,}"), "[SECRET]"),
]

为什么必须做：企业场景中的用户可能会随手粘贴客户手机号、身份证号。一旦这些信息被发送至第三方模型，就可能构成合规事故。

七、工具执行的可观测性

每次工具调用都会被 Tracer 封装为一个 Span：

with tracer.span(trace_id, f"tool.{name}", args=args, risk=tool.risk):
    result = tool.fn(args, trace_id)

系统还会向前端推送 tool_start / tool_end 事件，因此可以在执行链路时间线上看到每一步工具调用的参数、耗时和结果预览。

下一篇：07 · 记忆、可观测与成本控制

English / 原文

对应代码：app/agent/tools.py、app/agent/guardrails.py

一、工具即 Schema

Tool(
    name="sql_query",
    description="BI 数据查询（Text2SQL）：项目成本、缺陷趋势、交付效率等指标统计。只读。",
    parameters={"type": "object", "properties": {
        "question": {"type": "string", "description": "要查询的业务问题，中文描述即可"}},
        "required": ["question"]},
    fn=_tool_sql_query,
    risk="low",        
)

注册表对外提供三个能力：

registry.specs()                  
registry.describe_for_prompt()    
registry.execute(name, args, trace_id)

收益：新增工具只改一处，编排层零改动。

二、目前注册的 5 个工具

工具 风险 说明
search_knowledge low 知识库检索（RAG）
sql_query low BI 数据 Text2SQL
send_alert high 推送告警（需人工确认）
calculator low 安全四则运算
get_today low 解析"本月/上周"相对时间

calculator 用 AST 白名单实现（绝不用 eval）：

_ALLOWED_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ...}
def _safe_eval(expr):
    return _eval(ast.parse(expr, mode="eval"))

三、Text2SQL：生成 → 校验 → 执行 → 报错回炉

for attempt in range(2):
    prompt = TEXT2SQL_TMPL.format(schema=schema, question=question, limit=row_limit)
    if last_err:
        prompt += f"\n上一次生成的 SQL 报错：{last_err}\n请修正后只输出 SQL。"
    raw_sql = llm.chat([{"role": "user", "content": prompt}], temperature=0)
    try:
        safe_sql = validate_sql(raw_sql)       
        return execute(safe_sql)
    except GuardrailError as e:
        last_err = str(e)

踩坑记录 ： 第一版没有把数据库报错回传，模型生成的 SQL 字段名写错就永远失败； 加上"报错回炉"后，一次修正成功率显著提升。

Prompt 里的关键约束：

1. 只能输出一条 SELECT，不要解释、不要 markdown 代码块
2. 禁止 INSERT/UPDATE/DELETE/DROP/ATTACH/PRAGMA
3. 必须带 LIMIT 200
4. 日期字段为 TEXT（YYYY-MM-DD），本月用 date('now','start of month')
5. 只能用给定的表和字段
+ 一个少样本示例

四、四层 SQL 护栏

def validate_sql(sql, allowed_tables, row_limit, readonly=True):
    
    clean = strip_sql_comments(raw)
    
    if _MULTI_STMT.search(clean): raise GuardrailError("检测到多语句执行")
    
    if readonly and not low.startswith(("select", "with")): raise ...
    
    for kw in ("insert","update","delete","drop","attach","pragma",...):
        if re.search(rf"\b{kw}\b", low): raise ...
    
    tables = set(_TABLE_RE.findall(clean))
    if illegal := tables - allowed: raise GuardrailError(f"引用了未授权的表：{illegal}")
    
    if not re.search(r"\blimit\s+\d+", low): clean += f" LIMIT {row_limit}"

再加一层数据库侧保护：只读连接

con = sqlite3.connect(f"file:{db_path}?mode=ro")

（生产换成只读账号）

实测拦截效果 ：

输入 结果
"删掉成本表" ✅ 被关键字黑名单拦截
"SELECT * FROM secret_salary" ✅ 被表白名单拦截
"SELECT ...; DROP TABLE x" ✅ 被多语句检测拦截

五、Human-in-the-loop

if tool.risk == "high" and settings.hitl_enabled and not auto_approve_high_risk:
    return ToolResult(False, error="HITL_REQUIRED", meta={"need_human": True})

前端收到 hitl 事件 → 渲染橙色确认卡 → 用户点「确认执行」→ 带 approved=true 重新请求 → 后端从 _pending[thread_id] 取出并执行。

这就是 JD 里说的"权限边界"。

六、另外两个护栏

提示词注入检测

INJECTION_PATTERNS = [
    r"忽略(?:以上|上面|之前).{0,6}(?:指令|规则|提示)",
    r"ignore\s+(?:all\s+)?previous\s+instructions",
    r"输出(?:你的|系统).{0,4}(?:提示词|prompt)",
    ...
]

命中后不中断服务 ，而是降级为"只读检索模式"并发一条 warn 事件—— 粗暴拒绝会伤害正常用户体验。

PII 脱敏（出网前）

_PII_RULES = [
    (re.compile(r"1[3-9]\d{9}"), "[PHONE]"),          
    (re.compile(r"\b\d{17}[\dXx]\b"), "[ID_CARD]"),   
    (re.compile(r"\b\d{16,19}\b"), "[CARD]"),         
    (re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"), "[EMAIL]"),
    (re.compile(r"(?:sk-|api[_-]?key\s*[:=]\s*)[A-Za-z0-9_\-]{16,}"), "[SECRET]"),
]

为什么必须做 ：企业场景用户会随手粘贴客户手机号、身份证号。 一旦发到第三方模型，就是合规事故。

七、工具执行的可观测

每个工具调用都会被 Tracer 包成一个 Span：

with tracer.span(trace_id, f"tool.{name}", args=args, risk=tool.risk):
    result = tool.fn(args, trace_id)

同时推送 tool_start / tool_end 事件给前端， 所以你能在执行链路时间线上看到每一步工具调用的参数、耗时和结果预览 。

下一篇：07 · 记忆、可观测与成本控制