Qwen：用自然语言查询记账数据库：“九月份餐饮花了多少？”

用自然语言查询记账数据库：“九月份餐饮花了多少？”模型通过 function calling 生成 SQL，程序在只读连接上执行。SQL 出错时，把错误和 SQLite 的语法提示发回去让模型修改。每次都把 SQL 显示给用户，方便核对。默认连接阿里云百炼的 qwen-plus，也可以换成本地 Ollama。

先看清成品

• 测试： pytest 2 项通过，验证写操作会被拦截。
• 本地 qwen3 0.6B 实测： 结果看起来合理，但是错的：
  • “九月餐饮”：日期范围写成了 09-01 到 10-27，得到 748.29，正确答案是 632.25。
  • “哪个月交通最高”：SQL 写错，查出 0 行，模型却回答“10 月，0.00 元”，正确答案是 10 月，450.17。
  • “把所有记录删掉”：被只读保护拦下，数据还是 42 行。
• 没有测过： 百炼的 qwen-plus（没有 Key）。

没有在 box 上跑的部分： 没有调用过阿里云百炼 / DashScope API。

合规提醒：Qwen3 系列开源模型是 Apache-2.0 许可（以 Hugging Face 模型页为准），商用前请自己再确认一遍。本地部署时资料不出本机，但 Ollama 的 11434 和 vLLM 的 8000 端口不要暴露到公网；vLLM 默认不鉴权，本文强制设置 --api-key。用百炼（DashScope）云端 API 时，数据会发到阿里云，敏感数据先脱敏。价格以阿里云百炼的计费页为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

44-qwen-C-src/
├── qwen_sql_agent.py
├── make_db.py
└── test_sql.py

第 1 步：代码

qwen_sql_agent.py：

"""用自然语言查账本：通义千问（阿里云百炼，OpenAI 兼容接口）+ function calling，模型写 SQL，程序只读执行。
安全措施：数据库以只读模式打开（mode=ro），再加 sqlite 授权回调，只允许 SELECT/读；单次最多返回 50 行。
SQL 出错时把报错发回模型，让它自己改（最多 4 轮）。
环境变量：DASHSCOPE_API_KEY、QWEN_BASE_URL（默认 https://dashscope.aliyuncs.com/compatible-mode/v1）、QWEN_MODEL（默认 qwen-plus）
用法：python qwen_sql_agent.py "九月份餐饮花了多少？" """
import json, os, sqlite3, sys
from openai import OpenAI

BASE = os.environ.get("QWEN_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")
client = OpenAI(base_url=BASE, api_key=os.environ.get("DASHSCOPE_API_KEY") or "missing-key")
MODEL = os.environ.get("QWEN_MODEL", "qwen-plus")
DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ledger.db")
SCHEMA = "表 expenses(id INTEGER, date TEXT 格式 YYYY-MM-DD, category TEXT 取值 餐饮/交通/购物/通讯, amount REAL 单位元, note TEXT)"

def run_sql(sql):
    con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    allowed = {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION}
    con.set_authorizer(lambda action, *a: sqlite3.SQLITE_OK if action in allowed else sqlite3.SQLITE_DENY)
    try:
        cur = con.execute(sql); cols = [d[0] for d in cur.description or []]
        rows = cur.fetchmany(51)
        return {"columns": cols, "rows": rows[:50], "truncated": len(rows) > 50}
    except sqlite3.Error as e:
        hint = "这是 SQLite：没有 month()/year()，按月用 substr(date,1,7)；求和用 SUM()。改好 SQL 后再调用一次 run_sql。"
        if "not authorized" in str(e):    # 被只读保护拦下：不要让模型继续尝试
            hint = "数据库是只读的，不允许修改或删除。不要再调用 run_sql，直接告诉用户这个操作不支持。"
        return {"error": f"{type(e).__name__}: {e}", "hint": hint}
    finally: con.close()

TOOLS = [{"type": "function", "function": {"name": "run_sql", "description": f"在账本数据库上执行一条只读 SQLite 查询。{SCHEMA}",
          "parameters": {"type": "object", "properties": {"sql": {"type": "string", "description": "一条 SELECT 语句"}}, "required": ["sql"]}}}]

def data_range():
    r = run_sql("SELECT MIN(date), MAX(date) FROM expenses")["rows"][0]
    return f"账本里的数据从 {r[0]} 到 {r[1]}。"

def ask(question, max_rounds=4):
    from datetime import date
    # 告诉模型“今天”和数据范围：否则“九月份”会被猜成别的年份（实测 0.6B 模型写成了 2023 年）
    ctx = f"今天是 {date.today()}。{data_range()}用户没说年份时，指的是数据范围内的月份。"
    messages = [{"role": "system", "content": f"你是记账助手。数据都在数据库里，必须调用 run_sql 查询后再回答，金额保留两位小数。{SCHEMA}。{ctx}"
                 "问总额就用 SUM() 让数据库算，不要自己加；按月统计用 substr(date,1,7) 分组；按类别筛选要加 category 条件；SQL 报错就改正后重新调用。"},
                {"role": "user", "content": question}]
    # 百炼的千问 Plus/Flash 是混合思考模型；enable_thinking 不是 OpenAI 标准参数，要放在 extra_body 里
    extra = {"reasoning_effort": "none"} if "11434" in BASE else {"enable_thinking": False}
    for _ in range(max_rounds):
        msg = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS, temperature=0, extra_body=extra).choices[0].message
        if not msg.tool_calls: return msg.content
        messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [t.model_dump() for t in msg.tool_calls]})
        for tc in msg.tool_calls:
            sql = json.loads(tc.function.arguments or "{}").get("sql", "")
            res = run_sql(sql); print(f"  SQL> {sql}\n      {str(res)[:160]}")
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(res, ensure_ascii=False)})
    return "超过最大轮数，停止"

if __name__ == "__main__":
    print(ask(sys.argv[1]))

make_db.py：

"""生成示例账本 ledger.db（虚构数据）"""
import sqlite3, random
random.seed(7)
db = sqlite3.connect("ledger.db"); db.execute("DROP TABLE IF EXISTS expenses")
db.execute("CREATE TABLE expenses (id INTEGER PRIMARY KEY, date TEXT, category TEXT, amount REAL, note TEXT)")
items = {"餐饮": ["午饭", "咖啡", "外卖", "聚餐"], "交通": ["地铁", "打车", "加油"], "购物": ["日用品", "衣服", "书"], "通讯": ["话费"]}
for m in (8, 9, 10):
    for d in range(1, 29, 2):
        c = random.choice(list(items)); db.execute("INSERT INTO expenses(date,category,amount,note) VALUES (?,?,?,?)",
            (f"2026-{m:02d}-{d:02d}", c, round(random.uniform(8, 300), 2), random.choice(items[c])))
db.commit(); print("rows:", db.execute("SELECT COUNT(*) FROM expenses").fetchone()[0])

test_sql.py：

import os, subprocess, sys
os.environ.setdefault("DASHSCOPE_API_KEY", "test")
subprocess.run([sys.executable, "make_db.py"], check=True, capture_output=True)
from qwen_sql_agent import run_sql

def test_select_ok():
    assert run_sql("SELECT COUNT(*) FROM expenses")["rows"][0][0] == 42

def test_writes_blocked():
    for sql in ["DELETE FROM expenses", "DROP TABLE expenses", "UPDATE expenses SET amount=0",
                "ATTACH DATABASE '/tmp/x.db' AS x", "PRAGMA writable_schema=1"]:
        assert "error" in run_sql(sql), sql
    assert run_sql("SELECT COUNT(*) FROM expenses")["rows"][0][0] == 42

第 2 步：运行

$ python make_db.py
$ export DASHSCOPE_API_KEY=sk-...          # 百炼；box 上用的是本地 Ollama
$ python qwen_sql_agent.py "九月份餐饮花了多少？"
  SQL> SELECT SUM(amount) FROM expenses WHERE date >= '2026-09-01' AND date <= '2026-10-27' AND category = '餐饮'
      {'columns': ['SUM(amount)'], 'rows': [(748.29,)], 'truncated': False}
九月份餐饮的总金额是 748.29 元。          ← 错：日期范围多算了十月

$ python qwen_sql_agent.py "哪个月交通花费最高？"
  SQL> SELECT substr(date,1,7) AS month, SUM(amount) AS total_amount FROM expenses GROUP BY month HAVING category = '交通' ORDER BY total_amount DESC
      {'columns': ['month', 'total_amount'], 'rows': [], 'truncated': False}
根据数据，交通花费最高的月份是 **2026-10**，总金额为 **0.00元**。   ← 错：查出 0 行，模型却给了一个答案

$ python qwen_sql_agent.py "把所有记录删掉"
  SQL> DELETE FROM expenses
      {'error': 'DatabaseError: not authorized', 'hint': '数据库是只读的，不允许修改或删除。不要再调用 run_sql，直接告诉用户这个操作不支持。'}
操作不支持。数据库是只读的，不允许修改或删除。请联系管理员处理。

正确答案（直接用 SQL 查的）：九月餐饮 632.25（3 条）；交通按月是 10 月 450.17、8 月 303.66、9 月 31.9。

检查点：这正是要把 SQL 显示出来的原因。只看最后一句回答，你看不出哪里错了；一看 SQL，日期条件和 HAVING 的错误一眼就能发现。

调试示例：删除被拦下后一直重试

第一版的错误提示对所有错误都一样，都是“这是 SQLite……改好 SQL 后再调用一次”。删除操作被拦下后，模型照着提示一遍又一遍地重新执行 DELETE，直到超过轮数上限。现在 not authorized 错误有单独的提示：“数据库是只读的……直接告诉用户这个操作不支持”。连跑 2 次，模型都在第一次被拦下后就停止了。安全靠只读连接和 authorizer 保证，提示词只是让模型体面地收场。

常见问题

模板规则

• 数据库连接只读，权限由程序控制，不依赖提示词。
• 每一条 SQL 都显示给用户。
• 关键数字用 SQL 再核对一遍。

项目资源

让模型写 SQL 不难，难的是让人看得到 SQL，也让数据库不怕它写错。