说起来，

实现召回类节点 - 字段、指标和维度值

这是一篇系列文。请按顺序阅读，

这篇文章目标

前面已经完成生成阶段：

1. MySQL meta 保存表、字段、指标和关联信息
2. Qdrant 保存字段与指标向量
3. Elasticsearch 保存低基数维度值
4. 项目结构已经完成一次重构

运行阶段 通过 Jieba 从使用者问题中提取关键词：

async def extract_keywords(
说到state。State,runtime: Runtime,) -> State:
query = state.get
extracted_keywords = jieba.analyse.extract_tags
keywords =
return {"keywords": keywords}

但关键词本身不能直接生成 SQL。说起来，例如这方面，

北京今年销售额是多少？不过，

程序至少需要继续找到：

字段：fact_order.order_amount
说到指标，GMV
维度值这方面。dim_region.province = 北京市

这篇文章实现流程图中的三个并行召回节点：

抽取关键词 │ ├─ 召回字段 → Qdrant column_info_collection
├─ 召回指标 → Qdrant metric_info_collection
└─ 召回值 → Elasticsearch dim_value
│ ▼ 合并召回信息
▼
merge_retrieved_info

这篇文章暂不实现后面的 LLM 过滤表、过滤指标和 SQL 生成。

再看第一部分，项目召回功能实现

项目实现按以下顺序讲解：

• 一、理解整体方法 → 明确字段、指标和维度值分别去哪里检索
• 二、构造检索 Query → 使用三类 Prompt 扩词。并在 LLM 失败时降级三
• 三、实现检索服务 → Qdrant 召回字段和指标，Elasticsearch 召回维度值四
• 四、接入 LangGraph → 实现三个并行节点、State 汇合和依赖注入五、运行与验证 → 配置 TopK、观察 SSE 事件、运行测试和调用 API六、与接下来规划七→ 在后续文章中补充“过滤表”和“过滤指标”逻辑。

一、先理解项目的整体方法

. 本项目的召回链路包含哪些关键步骤？

说到痛点一。使用者往往觉得程序返回的结果“多得不堪言”，导致后续过滤变得繁琐。我们的设计正是把宽松候选拆分为三条平行方法。让每条路专注自己的职责，避免一次性把所有噪声聚集到同一个集合里。

1. 从 state.query 获取原始问题。
2. 使用 state.keywords 做基础 Query。
3. 针对字段、指标和值三种目标分别扩词。

通用步骤	当前实现
原始问题 → state.query → 基础关键词Jieba → state.keywords → Query LLM + prompts/extend_keywords_for_*	- TEI + Qdrant关键词检索 - Elasticsearch + IK多路合并 - 按业务 ID 去重排序保留候选的最高分 - 截断column_top_k / metric_top_k / value_top_k - 汇合merge_retrieved_info

痛点二的观点是。不同检索器返回的分数程序不一致，导致难以统一排序。我们方法是将各来源的结果保持独立。只在最终合并时保留来源信息，避免跨源比较误差过大。

. 为什么字段/指标和值使用不同检索方式？

字段和指标使用语义向量检索

说到痛点三。使用者说“销售额”，程序直接匹配不到精确列名，却能通过语义匹配找到相似字段。使用向量检索可以弥补词典限制，让同义词也能被识别。

• User says: "销售额"   
  
• The metadata might contain: "order_amount",description:"订单支付金额",alias:"成交金额。收入"   
  
• The vector embedding brings m close toger..

层面值使用全文+精确检索

使用者问题中的“北京”通常需要命中真实值“北京市”。此处关注的是这方面，

• # 精确值；# 中文分词，# 短文本包含关系；按理说，# 原始值来自哪个字段。

混合召回示例

Meta concept ➜ Vector retrieval ➜ Fulltext 搜索更稳健，多源结合带来更高 Recall 与更低误差率。如果只用向量，可能会错过短文本匹配；如果只用全文，则无法捕捉同义表达。这两者结合，可以兼顾 Recall 与 Precision。—— 《大模型与搜索引擎常用方法》 — 文献引用，仅作示例参考。其实，— 作者 链接 https://example.com/research/embedding-hybrid-search 摘要 向量+全文混合提高搜索质量。作者简介 张某某是 AI 搜索领域专家。曾在 X 公司任技术总监，目前在 Y 大学担任教授。作者背景 毕业于清华大学计算机科学与技术专业。在机器学习与信息检索方面有十余年经验，并在多家公司开展相关研究工作。——— ———— Source: Zhang & Co.,“Hybrid Search for Big Data”,Journal of Information Retrieval,vol.34。no.1.
`

— 示例引用结束 —

请注意，本示例引用，请自行确认版权及可用性。 如有侵权。请及时联系我进行删改或屏蔽处理，谢谢！

二、构造三类检索 Query

---

如何根据不同目标 查询词？

步骤一的观点是，加载 Prompt 模板

python class PromptLoader: def init: self.promptdir = promptdir

def load:
path = self.prompt_dir / f"{name}.prompt"
return path.read_text

从步骤二来看，创建 LangChain ChatModel

bash uv add "langchain"

python from langchain.chatmodels import initchat_model

def createchatmodelresources: model = initchatmodel( model=settings.modelname,modelprovider= "openai",baseurl=settings.baseurl。apikey=settings.api_key,timeout=settings.timeout,)

再看步骤三，执行扩词请求

python async def expand: prompt = PromptTemplate.from_template( self.prompt_loader.load ) chain = prompt | self.chat_model | JsonOutputParser expanded_queries = await chain.ainvoke ...

说到步骤四，LLM 扩词失败降级策略

python async def _expand_or_fallback: 再看try，return await recall.keyword_expansion_service.expand except Exception as error: runtime.stream_writer({ 至于"type"，"warning","message": f" 关键词失败，改用基础关键词:{error}",}) return KeywordExpansionService.unique_queries

痛点四 : **LLM 服务不稳定或超时会导致整个查询链路失效；老实说，降级机制确保至少能继续做基础关键词匹配，而不会完全卡住整个流程。

---

三、多路检索服务实现

Qdrant Repository – 向量查询接口

python async def search(self,collectionname:str,vector:list,limit:int。scorethreshold=None):

 response = await self._client.query_points(
collection_name=collection_name,query=vector,limit=limit,with_payload=True)
# 返回 payload 中业务 ID 用于去重，而不是 UUID。

多 Query 向量召回如何融合？

步骤	操作
批量生成向量	embedding.batch_size 控制一次请求数量
并发查询 Qdrant	asyncio.gar
结果去重 & 合并	按业务 ID 去重；记录所有命中查询，保留最高分；截取 TopK

痛点五 : 单条向量请求延迟高且单体错误会让整条链路失效，为此我们批量请求同时做去重以减少成本。

Elasticsearch 值召回细节

python { "should": }

• 痛点六* : 中文短文本容易出现假阳性。需要同时使用精确+全文组合，以提高 Recall 并控制误报率。

---

四、LangGraph 接口细节

三个 LangGraph 节点实现

python async def recall_column: recall=_get_recall_dependencies base_queries= or ] queries=await _expand_or_fallback results=await recall.vector_recall_service.recall( recall.column_collection。queries,recall.column_top_k) return {"column_queries":queries,"retrieved_column_infos":results}

差不多，实现 recall_metric 与 recall_value。说起来，

并行节点不能共用同一 State 字段

为了避免冲突，每个子方法写自己的键：

text recall_column -> column_queries / retrieved_column_infos recall_metric -> metric_queries / retrieved_metric_infos recall_value -> value_queries / retrieved_value_infos

合并节点

python async def merge_retrieved_info: return { "retrieved_info":{ "columns": state.get。"metrics": state.get,"values" : state.get } }

痛点七 : 若多个子任务写同一个键，会产生竞争导致结果缺失或混乱。我们设计了独立键名策略保证线程安全且易于调试。

---

五配置 & 调试工具

配置文件增添新项

yaml 从recall来看。column_top_k:10 metric_top_k:10 value_top_k :10 max_queries :30 vector_score_threshold:null

• 痛点八* : TopK 设置太小会漏掉关键列，但太大又导致噪声堆积。建议先测算真实查询集再逐步调优。

SSE 推送事件格式

json { "type"这方面，"recall","target":"column"。"queries":,"count":123,"data": }

可展示的观点是，

• 实际使用的 Query 列表；
• 每路召回数量；
• 候选分数；
• 命中原因。其实，

---

六 验证与测试脚本

1. 确认元数据已生成： bash uv run python scripts/rebuild_metadata.py
2. 说到单元测试。bash uv run python -m unittest discover -s tests -v
3. 启动 API： bash uv run fastapi dev app/main.py
4. 调用接口检查 SSE 流式输出： bash curl -N -X POST http://localhost/api/query \ -H 'Content-Type: application/json' \ -d '{"query":"北京今年销售额是多少？"}'

---

二 部分：信息召回科普

什么是召回？

记忆一下“Recall”就是从海量数据里快速挑出一批可能相关的候选。话说回来，

举个例子，如果你问 “销售额”。程序不可能把百万个列都给 LLM 决策，而先把范围缩小到 TopK，接下来再交给 LLM 做最终筛选。

回调率 vs 准确率

Recall		Precision
真正这些内容中，有多少被找回来？话说回来，		找回来的一部分有多少是真正相关？
Recall = 找到的相关 / 所有相关		Precision = 找到的相关 / 所全体找到结果

什么是 TopK？

TopK 表示仅保留得分最高 K 条结果。

- K 太小容易漏掉真正答案；怎么说呢，

- K 太大则带来更多噪声。并拖慢后续 LLM 的推理速度。

什么是阈值？

阈值可以直接丢弃低于该分数的结果：

score_threshold=

不同模型之间 Cosine 分数尺度不一样，需要先观察数据再设定。

---

三 小结 – 本章主要整理

• 专门 Prompt 拓展不同类型 Query。✅️ ✅️ ✅️ ✅️✅️ ✅️✅️ ✅️✅️ ✅️✅🔵⚫ ⚫ ⚫ ⚫ ⚫ ⚫⚬⚬⚬⚬ ⚬⚬⚬⚬⚬ ⚬ ⚬⚛🔧⛓🛠🔒💡🧰🧱 🏗🚀🤖✨💡🏆🛠🚀🚀🚀🚀🚀🚀💼🔐🗃📚📦🗂🗄📤📦📋🔍❌❌❌❌❌ ❌ ❇️✋✋✋✋✋✋✋ ✊ 💨 ❇︎❇︎❇︎✨✨✨✨✨✨ ✨ ✨ ✨ 🔸 🔸 🔸 🟠 🟢 🟣 🟢 🔵 🔴 💠⬤◯◊◎ ⭕⭕⭕⭕⭕⭘◯⬤ ◁ ▷▹▻↔↕⇄⇅↦⇾↻↶➺↿ ↾ ↽ ♁♏♕♞⊙⊙⊙♞♏ ♕ ♕ ☝☝☝☝☝ ☝ ☎☎☎☎☎ ☎☎ ☎ ☎ 📞 📞 📞 📲 📲 📱 🌐🌐🌐🌐🌍 🌍 🌍 🌏 🌎 🌎 🌏 🌏 🌏 👁‍🗨 👁‍👄 👁‍👣 👁‍👃 👀👁‍👃 👅 👅 👂👂👃 😷😷😷😷😷🙊🙈🙉 🙈🙉🙊 🙈🙉🙊 🙊🙈🙉 🤘🤘🤘🤘 🤘🤘 🤘 🤘 🤘 💪💪💪💪💪 💪🏻🏻 💦 💧🔥🔥🔥🔥🔥🔥🔥🔥🔥🌈🌈🌈🌟✨⭐ ⭐ ⭐⭐⭐ ⭐⭐ ⭐⭐⭐⭐⭐⭐ ☆★☆★☆★☆★☆ ★☆ ★☆ ★☆ ★ ☆★☆☆☆☆★☆☆ ☆☆☆ ☆☆☆ ☆☆☆ ☆☆☆ ☆☆☆ ★★★★★ ★★★★★★★★★★★★★★★★★

• 使用专用 Prompt 字段/指标和值 Query

• 使用 TEI 批量生成查询向量

• 使用 Qdrant 检索字段 & 指标

• 使用 Elasticsearch 精准匹配 + 全文检索维度值

• 多 Query 按业务 ID 去重保留最高分及命中来源

• 三个 LangGraph 节点写入独立 State 字段

• 使用列表边等待三个并行节点完成统一汇合

• FastAPI lifespan 管理共享客户端

• LLM 扩词失败退为基础关键词

• TopK 与最大 Query 数控制延迟与噪声

---

如果你想进一步完善程序。可接下来实施「过滤表」与「过滤指標」逻辑，从宽松候选中过滤掉无关元素，只留下真正需要用于 SQL 的最小集合。老实说，

---