数据截至 2026-08-18(上游 commit 69155611c59f)

QueryWeaver — 架构与原理

30 秒导读: QueryWeaver 是一个开源 Text2SQL 工具:你用大白话问「上个月有多少人注册」,它把你数据库的 schema 事先装进一个图数据库,先用「向量检索 + 图上多跳」精准挑出跟问题相关的几张表,再让一串 LLM agent 依次判断能不能答、生成 SQL、执行、出错就自愈、最后用人话复述结果。核心卖点是用图来理解 schema,而不是把整个 schema 一股脑塞给 LLM。

1. 这是什么(零基础也能懂)

一句话定义

QueryWeaver = 自然语言 → SQL → 查询结果 → 人话回答 的一条流水线,它对 schema 的理解建立在一张 FalkorDB 图(表是节点、列是节点、外键是边)之上。

解决什么问题 / 给谁用

想象你有一个几百张表的业务库。你想知道「上季度华东区退货率最高的品类」,但你:

不记得退货表叫 returns 还是 order_returns;
不知道品类字段在哪张表、要 join 几层才连得上订单。

把整个 schema 贴给 ChatGPT?几百张表的 DDL 会撑爆上下文,而且模型会在一堆无关表里迷路、编造不存在的列。

QueryWeaver 的做法是:先只挑出真正相关的那几张表,再让 LLM 在这个小得多、干净得多的子 schema 上生成 SQL。用户是数据分析师、运营、任何想问库但不想手写 SQL 的人,以及想在自己应用里嵌 Text2SQL 的开发者(有 Python SDK)。

它能做什么(功能)

连接 PostgreSQL / MySQL(Snowflake 需 [server] 额外依赖),自动把 schema 抽成图。
把自然语言问题翻成 SQL 并执行,返回结果 + 人话解读。
多轮对话:记住上文,「他们最近的订单呢」这种指代能接得住。
SQL 跑挂了自动自愈(把报错喂回 LLM 让它改)。
INSERT/UPDATE/DELETE 等破坏性操作先弹确认。
三种用法:REST API(流式)、MCP server、Python SDK(无需起服务)。
多 LLM 供应商:OpenAI / Gemini / Anthropic / Cohere / Azure / Ollama,靠环境变量自动选。

用起来什么样

最直观的是 Python SDK(README.md:261-283):

import asyncio
from queryweaver import QueryWeaver

async def main():
    qw = QueryWeaver(falkordb_url="redis://localhost:6379")
    # 连库并把 schema 抽进图(只需一次)
    conn = await qw.connect_database("postgresql://user:pass@host:5432/mydb")
    # 问问题:自然语言进,SQL + 结果 + 人话出
    result = await qw.query(conn.database_id, "Show me all customers from NYC")
    print(result.sql_query)   # SELECT * FROM customers WHERE city = 'NYC'
    print(result.results)     # [{"id": 1, "name": "Alice", ...}, ...]
    print(result.ai_response) # "Found 42 customers from NYC..."
    await qw.close()

asyncio.run(main())

一句话直觉 / 类比

把 schema 想成一张地铁图:表是站点,外键是铁轨。传统做法是把整张地铁图背给 LLM。QueryWeaver 的做法是:先按你的问题「你要去哪」在图上点亮几个站(向量检索),再顺着铁轨扩散几站(图多跳)找到换乘必经的中转站,只把这一小片地铁图交给 LLM。地图小了,不容易迷路,也不会编造不存在的站。

2. 顶层全景(它大概怎么转)

整个系统分两大阶段:离线建图(连库时做一次)和在线问答(每次提问都走)。

顶层图

怎么读这张图:上半是「连库时」把 schema 灌进 FalkorDB;下半是「提问时」每一步经过谁,从左到右基本就是时间顺序。

=== 连库阶段(connect_database,做一次)===

  真实数据库          Loader                       FalkorDB 图
  Postgres/MySQL  →  抽表/列/外键/采样值  →  Table节点·Column节点·REFERENCES边
                     LLM 生成每表/每库描述       + Table/Column 向量索引

=== 问答阶段(run_query,每次提问)===

  用户问题
     │
     ├─→ [RelevancyAgent] 这问题该不该答?(门禁,off-topic 直接挡回)
     │
     ├─→ [find()] 在图上找相关表 ── 见 02 章
     │        LLM 生描述 → 向量检索命中表/列 → 图上三种扩散 → 去重
     │
     ├─→ [AnalysisAgent] 拿着「小 schema + 记忆 + 规则」生成一条 SQL
     │        (判定能不能翻译;不能则转 FollowUpAgent 反问用户)
     │
     ├─→ 自动给带特殊字符的表名加引号(SQLIdentifierQuoter)
     │
     ├─→ 破坏性?→ 弹确认,等 /confirm
     │
     ├─→ 执行 SQL ──失败──→ [HealerAgent] 把报错喂回 LLM,最多改 3 次
     │
     └─→ [ResponseFormatterAgent] 把结果翻成人话 → 后台异步存记忆

部件一句话职责

部件	干什么	在哪个文件
Loader(Postgres/MySQL/Snowflake)	连真实库,抽 schema,采样列值	`api/loaders/postgres_loader.py`
`load_to_graph`	把抽出来的表/列/外键写成 FalkorDB 图 + 建向量索引	`api/loaders/graph_loader.py:12`
`find`	问答时在图上找相关表(向量 + 多跳)	`api/graph.py:279`
RelevancyAgent	判断问题在不在库能答的范围	`api/agents/relevancy_agent.py:68`
AnalysisAgent	拿子 schema 生成一条 SQL + 判可翻译性	`api/agents/analysis_agent.py:7`
HealerAgent	SQL 执行失败时迭代修复	`api/agents/healer_agent.py:18`
FollowUpAgent	不能翻译时生成友好反问	`api/agents/follow_up_agent.py:31`
ResponseFormatterAgent	把 SQL 结果翻成人话	`api/agents/response_formatter_agent.py`
`run_query` / `run_confirmed`	把上面所有部件串成一条流式管线	`api/core/text2sql.py:308` / `:645`
MemoryTool(Graphiti)	跨会话记忆:存历史问答、相似查询、用户画像	`api/memory/graphiti_tool.py:49`
QueryWeaver(SDK)	不起服务器直接调管线的 Python 客户端	`queryweaver/client.py:41`

主线走一遍(高层)

连库:connect_database → Loader 抽 schema → load_to_graph 写图 + 建向量索引。这一步把「表叫什么、列是什么类型、谁引用谁」都变成图里的节点和边,并给每张表/每列算好一个语义向量。
提问:run_query 先让 RelevancyAgent 当门禁;通过后 find 在图上挑出相关表;AnalysisAgent 在这一小片 schema 上生成 SQL;自动加引号、破坏性确认;执行;失败则 HealerAgent 自愈;最后 ResponseFormatterAgent 复述。
流式输出:每一步都作为一个 JSON 事件推给前端(靠一个分隔符切帧),SDK 则把中间事件丢掉、只取最后的结构化结果。

一个关键设计:run_query 是一个 async 生成器,既 yield 面向前端的 wire 事件 dict,又以一个 _Final(QueryResult) 哨兵收尾。流式路由逐个序列化 dict;SDK 用 collect_result 只留最后那个哨兵。同一条管线服务两种调用方式,不重复实现(api/core/text2sql.py:187-220)。

3. 阅读地图(建议顺序)

本项目子系统多,拆成 4 章,由浅入深:

01-schema-graph.md —— 一切的地基。schema 怎么从真实库抽出来、变成 FalkorDB 的属性图、为什么每个表/列都要算向量。看完你会明白「图」到底长什么样。
02-table-finding.md —— 全项目最巧的一支:find()。一个问题如何变成向量、命中若干表、再顺着外键在图上三种方式扩散,把 join 必经的中间表也捞回来。这是「graph-powered」的实质。
03-agent-pipeline.md —— 主循环 run_query。相关性门禁、分析生 SQL、加引号、确认、执行、自愈、复述,以及「一个生成器服务流式与 SDK 两种消费者」的设计。
04-safety-and-memory.md —— 深水区:破坏性操作确认(含注释绕过防护)、prompt 注入的分层防御、SQL 自愈的对话式循环、Graphiti 跨会话记忆。

如果你只想抓精华:读 02(图检索)+ 03(管线)两章即可讲清「它是做什么、核心原理是什么」。

1. 这是什么(零基础也能懂)​

一句话定义​

解决什么问题 / 给谁用​

它能做什么(功能)​

用起来什么样​

一句话直觉 / 类比​

2. 顶层全景(它大概怎么转)​

顶层图​

部件一句话职责​

主线走一遍(高层)​

3. 阅读地图(建议顺序)​