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

03 · 主管线 run_query

本章讲主循环 run_query(api/core/text2sql.py:308):把 01/02 章的部件串成一条端到端的 Text2SQL 流水线,并解释一个漂亮的设计——同一条生成器同时喂流式前端和无服务器 SDK。

3.1 一个生成器,两种消费者

它要解决的小问题: 网页要流式看到每一步进度(「正在生成 SQL…」「正在执行…」),而 Python SDK 只想要一个最终结果对象。若为两者各写一套管线,逻辑会漂移。

思路: 让 run_query 做成 async 生成器:

每一步 yield 一个面向前端的 wire dict({"type": "reasoning_step", "message": ...} 这类);
最后 yield 一个 _Final(QueryResult) 哨兵(api/core/text2sql.py:203),里面是结构化结果。

两种消费者各取所需:

流式路由 _serialize_pipeline (api/routes/graphs.py:37)
    每个 dict → json.dumps + MESSAGE_DELIMITER 推给前端;遇 _Final 停

SDK  collect_result (api/core/text2sql.py:209)
    丢掉所有 dict,只 return _Final.value(QueryResult)

MESSAGE_DELIMITER = "|||FALKORDB_MESSAGE_BOUNDARY|||"(api/core/pipeline.py:33)是前后端约定的切帧符,前端按它把粘在一起的 JSON 拆开。

3.2 主线各步(按执行顺序)

run_query(user_id, graph_id, chat_data)
  │
  1. 校验&截断对话历史(只留最近 5 轮 SHORT_MEMORY_LENGTH)
  │  并发起 MemoryTool 创建任务(若 use_memory)
  │
  2. 取库描述 db_description、库 URL、用户规则;定 db_type/loader
  │
  3. 并发:find(找表) ‖ RelevancyAgent(相关性门禁)
  │     ├─ off-topic → 取消 find,发 followup 事件,收尾(is_valid=False)
  │     └─ on-topic  → 等 find 拿到 tables;若开记忆则搜记忆上下文
  │
  4. AnalysisAgent.get_analysis → 一条 SQL + confidence + 缺失/歧义
  │     └─ is_sql_translatable=False → FollowUpAgent 反问用户,收尾
  │
  5. 自动给带特殊字符的表名加引号(auto_quote_sql_identifiers)
  │
  6. detect_destructive_operation:
  │     ├─ demo 库上破坏性 → 直接拒
  │     └─ 破坏性 → 发确认事件,收尾(requires_confirmation=True)
  │
  7. 执行 SQL
  │     └─ 抛错 → HealerAgent 自愈(见 04 章),仍失败则抛
  │
  8. schema 变更?→ 刷新图(_emit_schema_refresh)
  │
  9. ResponseFormatterAgent → 人话回答
  │
 10. 后台异步存记忆(save_memory_background)
  │
  └─ yield _Final(QueryResult)

3.3 几个值得停下来看的设计

判「能不能翻译」而不是硬生 SQL

AnalysisAgent 返回的不只是 SQL,还有 is_sql_translatable(api/agents/analysis_agent.py:12)。若为 False(比如「我的订单」但不知道用户是谁),管线不返回错误,而是转 FollowUpAgent 生成一句友好的反问(api/core/text2sql.py:426-452,prompt 见 api/agents/follow_up_agent.py:6),把球踢回给用户。这让「答不了」变成一次自然的多轮对话,而非报错。

自动加引号

LLM 生成的 SQL 里若把带 -、空格等特殊字符的表名不加引号,执行必挂。auto_quote_sql_identifiers(api/core/pipeline.py:332)用 02 章拿到的已知表名集合做交叉校验:只有「在 SQL 里出现、且确实是库里的表、且需要引号」的标识符才加(api/sql_utils/sql_sanitizer.py:107 auto_quote_identifiers),MySQL 用反引号、其余用双引号(DatabaseSpecificQuoter.get_quote_char)。用已知表名兜底,避免误伤别名或关键字。

响应用短记忆窗口

对话历史统一截到 SHORT_MEMORY_LENGTH = 5 轮(api/config.py:138、validate_and_truncate_chat api/core/pipeline.py:363),BaseAgent 把历史交错成 user/assistant 消息喂给每个 agent(api/agents/utils.py:33-42)。更长的上下文靠 04 章的 Graphiti 记忆,不靠塞进 prompt。

3.4 三种入口,一条管线

入口	怎么调	消费方式
REST 流式	`POST /graphs/{id}`(`api/routes/graphs.py:174`)	`_serialize_pipeline` 逐帧推
Python SDK	`QueryWeaver.query`(`queryweaver/client.py:125`)	`collect_result` 取最终
MCP tool	`query_database`(路由带 `tags=["mcp_tool"]`)	同 REST

SDK 路径里还有个细节:它用 contextvar 绑一个「后台任务槽」(_bind_task_sink,queryweaver/client.py:83),让 fire-and-forget 的记忆写入任务能在 close() 时被 await 干净(queryweaver/client.py:294-309)——服务器端事件循环长驻则不需要,任务自然排空。

3.5 确认路径 run_confirmed

破坏性 SQL 被弹确认后,用户回 CONFIRM,走 run_confirmed(api/core/text2sql.py:656)。它与 run_query 同样的「wire dict + _Final」形状,但跳过自愈:因为用户确认的是「这条」SQL,自愈改出来的变体不该未经确认就跑(api/core/text2sql.py:664-667)。demo 库上即使显式 CONFIRM 也拒绝写(api/core/text2sql.py:671-676)。

3.6 代码地图

主题	文件	符号
主管线生成器	`api/core/text2sql.py`	`run_query`
确认执行路径	`api/core/text2sql.py`	`run_confirmed`
最终结果哨兵	`api/core/text2sql.py`	`_Final` · `collect_result`
流式序列化 + 用量统计	`api/routes/graphs.py`	`_serialize_pipeline`
切帧分隔符	`api/core/pipeline.py`	`MESSAGE_DELIMITER`
相关性门禁	`api/agents/relevancy_agent.py`	`RelevancyAgent.get_answer`
分析生 SQL	`api/agents/analysis_agent.py`	`AnalysisAgent.get_analysis`
不可翻译时反问	`api/agents/follow_up_agent.py`	`FollowUpAgent.generate_follow_up_question`
自动加引号	`api/sql_utils/sql_sanitizer.py`	`SQLIdentifierQuoter.auto_quote_identifiers`
历史截断	`api/core/pipeline.py`	`validate_and_truncate_chat`
SDK 入口	`queryweaver/client.py`	`QueryWeaver.query`

3.1 一个生成器,两种消费者​

3.2 主线各步(按执行顺序)​

3.3 几个值得停下来看的设计​

相关性门禁与找表并发跑​

判「能不能翻译」而不是硬生 SQL​

自动加引号​

响应用短记忆窗口​

3.4 三种入口,一条管线​

3.5 确认路径 run_confirmed​

3.6 代码地图​