知识库查询 —— 向量检索节点

本文档详细介绍向量检索节点（search_embedding）的设计与实现，该节点是多路并行检索中的核心检索通道，负责将用户查询向量化并在 Milvus 中执行混合搜索。

学习理念：向量检索是知识库的"搜索入口"——SearchEmbeddingNode 是多路并行检索中最核心的通道。它用 BGE-M3 将用户问题转为稠密+稀疏双向量，在 Milvus 混合检索 + 商品名过滤，返回最相关的切片。混合检索 = 语义理解 + 关键词精确匹配。
海外对标：SearchEmbeddingNode 的"BGE-M3 混合嵌入 → Milvus hybrid_search → WeightedRanker 融合"流程对标 Pinecone 的 query API 的 sparse_dense 参数，以及 Weaviate 的 hybrid 查询。AnnSearchRequest + WeightedRanker 是 Milvus 2.4+ 混合检索的标准 API。

本节 AI 替代率：~85% | 人工干预率：~15%

角色	能力范围
🤖 AI 擅长	BGE-M3 调用、混合搜索请求构建、过滤表达式代码、execute_hybrid_search、测试代码
👤 人类需理解	混合检索权重调优（稠密:稀疏=0.5:0.5 是否最优）、过滤表达式（item_name in 的精度与召回权衡）

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习目标明确
🟢	§2 核心概念	~90%	~10%	混合检索 / BGE-M3 / AnnSearchRequest / WeightedRanker
🟡	§3 整体流程	~90%	~10%	理解数据流转即可
🔴	§4 分步实现	~80%	~20%	6 步流程中 Step4(混合搜索请求) + Step5(执行检索) 是核心
🔴	§4.4 主代码	~75%	~25%	SearchEmbeddingNode + build_hybrid_search_requests + execute_hybrid_search
🟢	§5 测试运行	~95%	~5%	看预期输出即可
🟡	§6 总结	~90%	~10%	设计要点回顾

技术栈健康度标签体系

技术	健康度	建议
BGE-M3	🔥 巅峰	多语言混合嵌入模型的事实标准。`generate_hybrid_embeddings` + `normalize_sparse_vector` 是标准工具链。
Milvus hybrid_search	🔥 巅峰	Milvus 2.4+ 原生混合检索 API。`AnnSearchRequest` × 2 + `WeightedRanker` 是标准模式。
WeightedRanker	🔥 巅峰	稠密+稀疏加权融合器。`norm_score=True` 时先归一化再加权。
过滤表达式 (expr)	🟢 稳定	Milvus 的标量过滤语法，`item_name in [...]` 是精准过滤的标准写法。

体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。

中英文对照表

English	中文	本质
Hybrid Search	混合检索	同时用稠密向量（语义）+ 稀疏向量（关键词）的检索方式
AnnSearchRequest	ANN 搜索请求	Milvus 中封装一路向量搜索的参数对象
WeightedRanker	加权融合器	将多路检索结果按权重融合排序的组件
Filter Expression	过滤表达式	Milvus 中标量字段的过滤语法，类似 SQL WHERE
CSR Matrix	压缩稀疏行矩阵	BGE-M3 输出稀疏向量的格式，indptr + indices + data
TopK	前 K 条	检索返回的最相关结果数量
Metric Type	度量类型	向量相似度的计算方式（IP = 内积，COSINE = 余弦）

💡 程序员比喻
SearchEmbeddingNode 就像 grep -r "query" /knowledge + sort -k2 -n——先用向量（语义 grep）搜出候选，再用 WeightedRanker 排序（sort by score）。
混合检索 就像 git log --grep + git log -S 的结合——前者模糊语义（grep），后者精确关键词（pickaxe）。
过滤表达式 就像 SELECT * FROM chunks WHERE item_name IN ("...")——先缩小范围再检索，性能更好。
WeightedRanker(0.5, 0.5) 就像搜索引擎的 BM25 + TF-IDF 加权——两个信号各占一半，综合排名。
CSR 稀疏矩阵提取 就像 git diff --numstat——只看变更的文件（非零元素），不关心没变动的（零维）。

1. 任务目标

1.1 本章目标

通过本章学习，你将掌握：

理解混合向量检索原理：掌握稠密向量 + 稀疏向量的混合检索策略
学会 BGE-M3 模型的使用：生成同时包含稠密和稀疏表示的嵌入向量
掌握 Milvus 混合搜索 API：使用 AnnSearchRequest 和 WeightedRanker
理解过滤表达式构建：基于商品名称精准过滤检索结果
实现可测试的检索节点：通过 if __name__ == "__main__" 验证节点功能

1.2 涉及文件

knowledge/processor/query_process/
├── nodes/
│   └── search_embedding.py    # 向量检索节点（本章重点）
└── ...

knowledge/tools/
├── embedding_utils.py         # 向量嵌入工具（BGE-M3）
├── milvus_utils.py            # Milvus 向量数据库工具
└── normalize_sparse_vector.py # 稀疏向量归一化

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 为什么需要混合检索？

传统的向量检索方式各有优缺点：

检索方式	优点	缺点
稠密向量	语义理解强，支持同义词匹配	对专业术语、型号等精确匹配较弱
稀疏向量	精确匹配强，对关键词敏感	语义理解弱，无法处理同义词
混合检索	兼具两者优点	需要额外的融合策略

示例对比：

查询: "万用表怎么测电压"

稠密向量检索:
  ✓ "数字万用表电压测量方法"     (语义相似)
  ✓ "使用多用电表测量电势差"     (同义词匹配)
  ✗ "RS-12型万用表"             (缺少语义关联)

稀疏向量检索:
  ✓ "万用表测量电压的步骤"       (关键词匹配)
  ✓ "RS-12万用表电压档位说明"   (包含关键词)
  ✗ "数字仪表电势测定"           (无共同词汇)

混合检索:
  ✓ 同时召回以上所有相关结果
  ✓ 通过加权融合得到最优排序

2.2 BGE-M3 模型

BGE-M3 是北京智源人工智能研究院（BAAI）开发的多功能嵌入模型：

特性	说明
Multi-Linguality	支持 100+ 语言
Multi-Functionality	同时生成稠密和稀疏向量
Multi-Granularity	支持短文本和长文档

输出结构：

🟡 【P1 看注释就行】 Step1 获取查询参数——使用 rewritten_query（已商品名替换）而非 original_query。

python

embeddings = bge_m3_model(["查询文本"])
# embeddings = {
#     "dense": [[0.123, 0.456, ...]],      # 1024 维稠密向量
#     "sparse": <CSR 稀疏矩阵>              # 词汇表 ID → 权重
# }

2.3 稀疏向量表示

稀疏向量使用字典格式存储非零元素：

🟡 【P1 看注释就行】 Step2 查询向量化——generate_hybrid_embeddings([query]) 生成稠密+稀疏双向量。

python

sparse_vector = {
    12345: 0.85,   # 词汇表中第 12345 个词，权重 0.85
    67890: 0.62,   # 词汇表中第 67890 个词，权重 0.62
    ...
}

特点：

大部分维度为 0，只存储非零项
词汇表通常有 25 万+ 个词
需要归一化以保证检索效果

2.4 Milvus 混合搜索 API

Milvus 提供了原生的混合搜索支持：

🟡 【P1 看注释就行】 稀疏向量提取——CSR 三数组：indptr[i] 到 indptr[i+1] 确定行范围，indices=词汇 ID，data=权重。normalize_sparse_vector 做 L2 归一化。

python

from pymilvus import AnnSearchRequest, WeightedRanker

# 1. 构建两路搜索请求
dense_req = AnnSearchRequest(
    data=[dense_vector],
    anns_field="dense_vector",
    param={"metric_type": "IP"},
    limit=10,
)

sparse_req = AnnSearchRequest(
    data=[sparse_vector],
    anns_field="sparse_vector",
    param={"metric_type": "IP"},
    limit=10,
)

# 2. 使用加权融合器
ranker = WeightedRanker(0.5, 0.5)  # 稠密:稀疏 = 0.5:0.5

# 3. 执行混合搜索
results = client.hybrid_search(
    collection_name="chunks",
    reqs=[dense_req, sparse_req],
    ranker=ranker,
    limit=5,
)

2.5 过滤表达式

Milvus 支持在向量检索时添加标量过滤条件：

🟡 【P1 看注释就行】 Step3 过滤表达式——item_name in ["a", "b"] 格式，空列表时返回 None（不过滤）。

python

# 单值过滤
filter_expr = 'item_name == "万用表RS-12"'

# 多值过滤（IN 语法）
filter_expr = 'item_name in ["万用表RS-12", "示波器DS-100"]'

# 组合过滤
filter_expr = 'item_name == "万用表RS-12" and category == "电子仪器"'

3. 向量检索业务处理流程（总）

3.1 整体流程图

3.2 数据流转

4. 向量检索业务处理流程（分）

4.1 目标

实现一个能够：

将用户查询转换为混合向量表示（稠密 + 稀疏）
根据已确认的商品名称构建过滤条件
在 Milvus 中执行高效的混合检索
返回相关度最高的文档切片

4.2 需求分析

4.2.1 功能需求

查询向量化：使用 BGE-M3 模型生成稠密和稀疏向量
过滤表达式构建：将商品名称列表转换为 Milvus 过滤语法
混合搜索请求构建：创建稠密和稀疏两路 AnnSearchRequest
混合检索执行：调用 Milvus hybrid_search API
结果返回：返回包含 chunk_id、content、item_name 的结果列表

4.2.2 技术依赖

依赖	用途
BGE-M3	生成混合向量（稠密 + 稀疏）
Milvus	向量数据库，支持混合检索
pymilvus	Milvus Python SDK

4.2.3 配置参数

🔥 【P0 必须要学】 Step4 混合搜索请求——AnnSearchRequest 是混合检索的核心构建块。注意：(1) 稠密和稀疏各创建一个 request (2) anns_field 指定向量字段 (3) expr 传入过滤表达式 (4) limit = 每路返回数。两个 request 就是"两条腿走路"。

python

SEARCH_TOP_K = 10           # 每路检索候选数
RERANK_TOP_K = 5            # 融合后返回数
RANKER_WEIGHTS = (0.5, 0.5) # 稠密:稀疏权重
OUTPUT_FIELDS = ["chunk_id", "content", "item_name"]

4.3 实现流程

4.3.1 实现流程图

4.3.2 具体实现步骤

Step 1: 获取查询参数

目的： 从图状态中提取查询文本和商品名称过滤条件。

实现逻辑：

从 state 中获取 rewritten_query（已改写的查询）
从 state 中获取 item_names（已确认的商品名称列表）
确定目标集合名称

代码片段：

🔥 【P0 必须要学】 Step5 执行混合检索——execute_hybrid_search 是核心执行器。注意：(1) WeightedRanker(0.5, 0.5) 的权重配置 (2) norm_score=True 先归一化再融合 (3) 融合公式：final = w1 * score1 + w2 * score2。混合检索的理解重点：两路独立搜索 → 加权融合排序。

python

def process(self, state: QueryGraphState) -> QueryGraphState:
    # 获取查询文本
    query = state.get("rewritten_query", "")

    # 获取商品名称（用于过滤）
    item_names = state.get("item_names")

    # 目标集合
    collection_name = "chunks_test"

说明：

使用 rewritten_query 而非 original_query，因为前者已经过商品名确认节点的改写，包含更完整的信息
item_names 用于精准过滤，确保只检索指定商品的相关内容

Step 2: 查询向量化

目的： 使用 BGE-M3 模型将查询文本转换为稠密向量和稀疏向量。

实现逻辑：

调用 generate_hybrid_embeddings([query]) 生成混合嵌入
内部流程：
- 获取 BGE-M3 模型单例
- 调用模型推理，得到原始输出
- 提取稠密向量（直接转换为 list）
- 提取稀疏向量（从 CSR 矩阵按行解析为字典）
- 对稀疏向量进行归一化

代码片段：

🟡 【P1 看注释就行】 Step6 返回结果——只返回 {"embedding_chunks": chunks}，LangGraph 自动合并到完整状态。

python

from knowledge.tools.embedding_utils import generate_hybrid_embeddings

# 记录日志
self.log_step("step_1", f"查询向量化: {query}")

# 生成混合嵌入
embeddings = generate_hybrid_embeddings([query])

# embeddings 结构:
# {
#     "dense": [[0.12, 0.34, ...]],  # 1024 维稠密向量
#     "sparse": [{12345: 0.85, ...}]  # 稀疏向量字典
# }

工具函数详解：

🔥 【P0 必须要学】 SearchEmbeddingNode 完整代码——5 步流程：获取参数 → 向量化 → 过滤表达式 → 构建请求 → 执行检索。重点理解 build_hybrid_search_requests + execute_hybrid_search 两个工具函数的配合。_build_filter_expr 的防御性处理（空列表 → None）避免语法错误。

python

# embedding_utils.py

def generate_hybrid_embeddings(texts: List[str]) -> Dict[str, list]:
    """为一组文本生成混合嵌入（稠密向量 + 稀疏向量）。"""

    # 1. 获取 BGE-M3 模型
    model = get_bge_m3_model()

    # 2. 调用模型推理
    raw_embeddings = model(texts)
    # raw_embeddings["dense"]: numpy 数组
    # raw_embeddings["sparse"]: CSR 稀疏矩阵

    # 3. 转换稠密向量
    dense_vectors = [emb.tolist() for emb in raw_embeddings["dense"]]

    # 4. 提取并归一化稀疏向量
    sparse_vectors = _extract_sparse_vectors(raw_embeddings, len(texts))

    return {
        "dense": dense_vectors,
        "sparse": sparse_vectors,
    }

稀疏向量提取详解：

🟢 【P2 后面可以查】 测试代码——有/无过滤条件两种场景。看预期输出中的混合检索分数即可。

python

def _extract_sparse_vectors(raw_embeddings, text_count):
    """从 CSR 稀疏矩阵中提取稀疏向量。"""

    sparse_matrix = raw_embeddings["sparse"]
    sparse_vectors = []

    for i in range(text_count):
        # CSR 矩阵按行存储，通过 indptr 定位每行的起止位置
        row_start = sparse_matrix.indptr[i]
        row_end = sparse_matrix.indptr[i + 1]

        # 提取该行的非零元素：{token_id: weight}
        sparse_dict = dict(zip(
            sparse_matrix.indices[row_start:row_end].tolist(),  # 词汇 ID
            sparse_matrix.data[row_start:row_end].tolist(),     # 权重值
        ))

        # 归一化
        sparse_vectors.append(normalize_sparse_vector(sparse_dict))

    return sparse_vectors

Step 3: 构建过滤表达式

目的： 将商品名称列表转换为 Milvus 的过滤表达式语法。

实现逻辑：

检查 item_names 是否为空
如果有值，构建 item_name in ["xxx", "yyy"] 格式的表达式
如果为空，返回 None（不过滤）

代码片段：

python

@staticmethod
def _build_filter_expr(item_names: Optional[List[str]]) -> Optional[str]:
    """将商品名称列表转换为 Milvus 过滤表达式。"""

    # 空列表不过滤
    if not item_names:
        return None

    # 构建 IN 表达式
    quoted = ", ".join(f'"{v}"' for v in item_names)
    return f"item_name in [{quoted}]"

# 示例:
# item_names = ["万用表RS-12", "示波器DS-100"]
# 结果: 'item_name in ["万用表RS-12", "示波器DS-100"]'

使用场景：

python

filter_expr = self._build_filter_expr(item_names)
self.logger.debug(f"过滤表达式: {filter_expr}")

# 输出: 过滤表达式: item_name in ["万用表RS-12"]

Step 4: 构建混合搜索请求

目的： 为稠密向量和稀疏向量分别创建 AnnSearchRequest 对象。

实现逻辑：

创建稠密向量搜索请求：
- 指定 anns_field="dense_vector"
- 使用内积（IP）作为相似度度量
- 添加过滤表达式
- 设置 TopK
创建稀疏向量搜索请求：
- 指定 anns_field="sparse_vector"
- 同样使用内积（IP）
- 相同的过滤条件和 TopK

代码片段：

python

from knowledge.tools.milvus_utils import build_hybrid_search_requests

reqs = build_hybrid_search_requests(
    dense_vector=embeddings["dense"][0],   # 稠密向量
    sparse_vector=embeddings["sparse"][0], # 稀疏向量
    dense_search_params={"metric_type": "IP"},
    sparse_search_params={"metric_type": "IP"},
    filter_expr=filter_expr,
    top_k=self.SEARCH_TOP_K,  # 10
)

工具函数详解：

python

# milvus_utils.py

def build_hybrid_search_requests(
    dense_vector: List[float],
    sparse_vector: Dict[int, float],
    *,
    dense_search_params: Optional[Dict] = None,
    sparse_search_params: Optional[Dict] = None,
    filter_expr: Optional[str] = None,
    top_k: int = 5,
) -> List[AnnSearchRequest]:
    """构建混合检索的 ANN 搜索请求。"""

    # 默认参数
    if dense_search_params is None:
        dense_search_params = {"metric_type": "IP"}
    if sparse_search_params is None:
        sparse_search_params = {"metric_type": "IP"}

    # 稠密向量搜索请求
    dense_request = AnnSearchRequest(
        data=[dense_vector],
        anns_field="dense_vector",
        param=dense_search_params,
        expr=filter_expr,
        limit=top_k,
    )

    # 稀疏向量搜索请求
    sparse_request = AnnSearchRequest(
        data=[sparse_vector],
        anns_field="sparse_vector",
        param=sparse_search_params,
        expr=filter_expr,
        limit=top_k,
    )

    return [dense_request, sparse_request]

Step 5: 执行混合检索

目的： 调用 Milvus 的 hybrid_search API，融合两路检索结果。

实现逻辑：

获取 Milvus 客户端单例
创建加权融合器 WeightedRanker
调用 hybrid_search 执行检索
返回融合排序后的结果

代码片段：

python

from knowledge.tools.milvus_utils import get_milvus_client, execute_hybrid_search

self.log_step("step_2", "执行混合搜索")

res = execute_hybrid_search(
    client=get_milvus_client(),
    collection_name=collection_name,
    search_requests=reqs,
    ranker_weights=self.RANKER_WEIGHTS,  # (0.5, 0.5)
    normalize_score=True,
    top_k=self.RERANK_TOP_K,  # 5
    output_fields=self.OUTPUT_FIELDS,
)

工具函数详解：

python

# milvus_utils.py

def execute_hybrid_search(
    client: MilvusClient,
    collection_name: str,
    search_requests: List[AnnSearchRequest],
    *,
    ranker_weights: Tuple[float, float] = (0.5, 0.5),
    normalize_score: bool = False,
    top_k: int = 5,
    output_fields: Optional[List[str]] = None,
) -> Optional[List]:
    """执行混合检索（稠密 + 稀疏加权融合）。"""

    if output_fields is None:
        output_fields = ["item_name"]

    try:
        # 创建加权融合器
        ranker = WeightedRanker(
            ranker_weights[0],  # 稠密权重
            ranker_weights[1],  # 稀疏权重
            norm_score=normalize_score,
        )

        # 执行混合搜索
        results = client.hybrid_search(
            collection_name=collection_name,
            reqs=search_requests,
            ranker=ranker,
            limit=top_k,
            output_fields=output_fields,
        )

        hit_count = len(results[0]) if results else 0
        logger.info(f"混合检索完成: collection={collection_name}, 命中={hit_count}")

        return results

    except Exception as e:
        logger.error(f"混合检索执行失败: {e}")
        return None

WeightedRanker 工作原理：

稠密检索结果:          稀疏检索结果:
Doc A: 0.9             Doc A: 0.7
Doc B: 0.8             Doc C: 0.9
Doc C: 0.6             Doc B: 0.5

融合公式（normalize_score=True 时先归一化）:
final_score = 0.5 * dense_score + 0.5 * sparse_score

融合结果:
Doc A: 0.5 * 0.9 + 0.5 * 0.7 = 0.80
Doc C: 0.5 * 0.6 + 0.5 * 0.9 = 0.75
Doc B: 0.5 * 0.8 + 0.5 * 0.5 = 0.65

最终排序: Doc A > Doc C > Doc B

Step 6: 返回结果

目的： 将检索结果写入图状态，供后续节点使用。

实现逻辑：

从返回结果中提取第一个列表（对应第一个查询）
记录日志
返回包含 embedding_chunks 的字典

代码片段：

python

# 提取结果
chunks = res[0] if res else []

# 记录日志
self.log_step("step_3", f"搜索完成，返回 {len(chunks)} 条结果")

# 返回结果（只返回需要更新的字段）
return {"embedding_chunks": chunks}

返回数据结构：

python

{
    "embedding_chunks": [
        {
            "entity": {
                "chunk_id": 12345,
                "content": "万用表测量电压时，首先将旋钮转到...",
                "item_name": "万用表RS-12"
            },
            "distance": 0.9234
        },
        {
            "entity": {
                "chunk_id": 12346,
                "content": "使用直流电压档测量时，需要注意...",
                "item_name": "万用表RS-12"
            },
            "distance": 0.8756
        },
        # ...
    ]
}

4.4 代码实现

以下是完整的节点实现代码：

python

# knowledge/processor/query_process/nodes/search_embedding.py

"""向量搜索节点

对用户查询进行向量化，在 Milvus 中执行混合搜索（稠密 + 稀疏），返回相关切片。
"""

import os
from typing import List, Optional

from knowledge.processor.query_process.base import BaseNode, setup_logging
from knowledge.processor.query_process.state import QueryGraphState


class SearchEmbeddingNode(BaseNode):
    """向量搜索节点。

    流程: 查询向量化 → 构建混合搜索请求 → 执行检索 → 返回结果
    """

    name = "search_embedding"

    # 检索参数
    SEARCH_TOP_K = 10        # 每路检索候选数
    RERANK_TOP_K = 5         # 融合后返回数
    RANKER_WEIGHTS = (0.5, 0.5)  # 稠密:稀疏权重
    OUTPUT_FIELDS = ["chunk_id", "content", "item_name"]

    def process(self, state: QueryGraphState) -> QueryGraphState:
        """执行向量搜索。

        Args:
            state: 需包含 rewritten_query 和 item_names。

        Returns:
            {"embedding_chunks": [...]} 搜索结果列表。
        """
        from knowledge.tools.embedding_utils import generate_hybrid_embeddings
        from knowledge.tools.milvus_utils import (
            get_milvus_client,
            build_hybrid_search_requests,
            execute_hybrid_search,
        )

        # 获取查询参数
        query = state.get("rewritten_query", "")
        item_names = state.get("item_names")
        collection_name = "chunks_test"

        # Step 1: 向量化
        self.log_step("step_1", f"查询向量化: {query}")
        embeddings = generate_hybrid_embeddings([query])

        # Step 2: 构建过滤表达式
        filter_expr = self._build_filter_expr(item_names)
        self.logger.debug(f"过滤表达式: {filter_expr}")

        # Step 3: 构建混合搜索请求
        reqs = build_hybrid_search_requests(
            dense_vector=embeddings["dense"][0],
            sparse_vector=embeddings["sparse"][0],
            dense_search_params={"metric_type": "IP"},
            sparse_search_params={"metric_type": "IP"},
            filter_expr=filter_expr,
            top_k=self.SEARCH_TOP_K,
        )

        # Step 4: 执行混合检索
        self.log_step("step_2", "执行混合搜索")
        res = execute_hybrid_search(
            client=get_milvus_client(),
            collection_name=collection_name,
            search_requests=reqs,
            ranker_weights=self.RANKER_WEIGHTS,
            normalize_score=True,
            top_k=self.RERANK_TOP_K,
            output_fields=self.OUTPUT_FIELDS,
        )

        # Step 5: 返回结果
        chunks = res[0] if res else []
        self.log_step("step_3", f"搜索完成，返回 {len(chunks)} 条结果")

        return {"embedding_chunks": chunks}

    @staticmethod
    def _build_filter_expr(item_names: Optional[List[str]]) -> Optional[str]:
        """将商品名称列表转换为 Milvus 过滤表达式。

        Args:
            item_names: 商品名称列表。

        Returns:
            如 `item_name in ["a", "b"]`；列表为空则返回 None。
        """
        if not item_names:
            return None
        quoted = ", ".join(f'"{v}"' for v in item_names)
        return f"item_name in [{quoted}]"


# ================================================================== #
#                        兼容 & 测试                                   #
# ================================================================== #

_node_instance = SearchEmbeddingNode()


def node_search_embedding(state: QueryGraphState) -> QueryGraphState:
    """兼容原有调用方式的入口函数。"""
    return _node_instance(state)

5. 测试入口

5.1 测试代码

在 search_embedding.py 文件末尾添加测试代码：

python

if __name__ == "__main__":
    import json
    from dotenv import load_dotenv

    load_dotenv()
    setup_logging()

    print("=" * 60)
    print("向量检索节点测试")
    print("=" * 60)

    # 1. 准备测试状态
    test_state = {
        "session_id": "test_001",
        "rewritten_query": "如何使用万用表测量电压？",
        "item_names": ["RS-12数字万用表"],
        "embedding_chunks": [],
    }

    print(f"\n输入状态:")
    print(f"  rewritten_query: {test_state['rewritten_query']}")
    print(f"  item_names: {test_state['item_names']}")
    print("-" * 60)

    # 2. 执行节点
    try:
        result = node_search_embedding(test_state)
        chunks = result.get("embedding_chunks", [])

        print(f"\n检索到 {len(chunks)} 条结果:")
        print("-" * 60)

        for i, chunk in enumerate(chunks, 1):
            # 兼容不同的返回格式
            entity = chunk.get("entity", chunk) if isinstance(chunk, dict) else {}
            content = entity.get("content", "")
            item_name = entity.get("item_name", "未知")
            chunk_id = entity.get("chunk_id", "N/A")
            score = chunk.get("distance", 0)

            print(f"[{i}] 商品: {item_name}")
            print(f"    ID: {chunk_id}")
            print(f"    分数: {score:.4f}")
            print(f"    内容: {content[:80]}...")
            print()

    except Exception as e:
        print(f"\n执行失败: {e}")
        import traceback
        traceback.print_exc()

    # 3. 测试无过滤条件的场景
    print("=" * 60)
    print("测试无商品名过滤")
    print("=" * 60)

    test_state_no_filter = {
        "session_id": "test_002",
        "rewritten_query": "如何测量电压？",
        "item_names": [],  # 无商品名
        "embedding_chunks": [],
    }

    print(f"\n输入状态:")
    print(f"  rewritten_query: {test_state_no_filter['rewritten_query']}")
    print(f"  item_names: {test_state_no_filter['item_names']} (无过滤)")
    print("-" * 60)

    try:
        result2 = node_search_embedding(test_state_no_filter)
        chunks2 = result2.get("embedding_chunks", [])
        print(f"\n检索到 {len(chunks2)} 条结果（无过滤）")

        # 打印前 3 条
        for i, chunk in enumerate(chunks2[:3], 1):
            entity = chunk.get("entity", chunk) if isinstance(chunk, dict) else {}
            print(f"[{i}] {entity.get('item_name', '?')} | score={chunk.get('distance', 0):.4f}")

    except Exception as e:
        print(f"\n执行失败: {e}")

5.2 运行测试

bash

# 进入项目目录
cd knowledge

# 激活虚拟环境
source .venv/bin/activate  # Linux/Mac
# 或
.venv\Scripts\activate     # Windows

# 运行测试
python -m knowledge.processor.query_process.nodes.search_embedding

5.3 预期输出

============================================================
向量检索节点测试
============================================================

输入状态:
  rewritten_query: 如何使用万用表测量电压？
  item_names: ['RS-12数字万用表']
------------------------------------------------------------
2024-01-15 10:30:00 - query.search_embedding - INFO - --- search_embedding 开始 ---
2024-01-15 10:30:00 - query.search_embedding - INFO - [step_1] 查询向量化: 如何使用万用表测量电压？
2024-01-15 10:30:01 - query.search_embedding - INFO - [step_2] 执行混合搜索
2024-01-15 10:30:01 - milvus_utils - INFO - 混合检索完成: collection=chunks_test, 命中=5
2024-01-15 10:30:01 - query.search_embedding - INFO - [step_3] 搜索完成，返回 5 条结果
2024-01-15 10:30:01 - query.search_embedding - INFO - --- search_embedding 完成 ---

检索到 5 条结果:
------------------------------------------------------------
[1] 商品: RS-12数字万用表
    ID: 12345
    分数: 0.9234
    内容: 万用表测量电压时，首先将旋钮转到直流电压（V-）或交流电压（V~）档位，根据被测电压...

[2] 商品: RS-12数字万用表
    ID: 12346
    分数: 0.8756
    内容: 使用直流电压档测量时，需要注意表笔的极性：红表笔接正极，黑表笔接负极。如果接反了...

...

============================================================
测试无商品名过滤
============================================================

输入状态:
  rewritten_query: 如何测量电压？
  item_names: [] (无过滤)
------------------------------------------------------------

检索到 5 条结果（无过滤）
[1] 万用表RS-12 | score=0.8901
[2] 示波器DS-100 | score=0.7823
[3] 万用表MT-200 | score=0.7654

6. 总结

6.1 节点功能概览

功能	说明
查询向量化	使用 BGE-M3 生成稠密向量（1024 维）和稀疏向量
过滤表达式构建	将商品名称列表转换为 Milvus IN 语法
混合搜索请求	同时构建稠密和稀疏两路 AnnSearchRequest
加权融合	使用 WeightedRanker 按 0.5:0.5 融合两路结果
结果返回	返回 chunk_id、content、item_name、distance

6.2 节点设计要点

1. 混合检索的优势

稠密向量: 语义理解强，捕捉同义词、上下文关系
稀疏向量: 精确匹配强，捕捉关键词、专业术语

混合检索 = 语义召回 + 关键词召回 = 更高的召回率和精度

2. 单例模式的资源管理

python

# BGE-M3 模型单例
_bge_m3_model: Optional[BGEM3EmbeddingFunction] = None

def get_bge_m3_model():
    global _bge_m3_model
    if _bge_m3_model is None:
        _bge_m3_model = BGEM3EmbeddingFunction(...)
    return _bge_m3_model

避免重复加载模型（模型加载耗时约 10-30 秒）
全局共享，节省内存

3. 延迟导入避免启动开销

python

def process(self, state):
    from knowledge.tools.embedding_utils import generate_hybrid_embeddings
    from knowledge.tools.milvus_utils import ...

模块导入时不加载重型依赖
首次调用时才加载，加速启动

4. 过滤表达式的防御性处理

python

@staticmethod
def _build_filter_expr(item_names):
    if not item_names:
        return None  # 空列表返回 None，不过滤
    ...

空列表时不添加过滤，避免语法错误
使用静态方法，无需访问实例状态

5. 返回值的精简设计

python

return {"embedding_chunks": chunks}

只返回需要更新的字段
LangGraph 会自动合并到完整状态中
避免覆盖其他节点写入的字段

企业痛点映射

痛点	传统方案	AI Agent 向量检索方案	效率提升
纯语义搜索丢失关键词匹配	稠密向量搜不到"RS-12"型号	混合检索：稠密语义 + 稀疏精确匹配	型号召回率提升 ~60%
无商品名过滤导致跨产品混淆	搜到其他产品的无关内容	`item_name in [...]` 过滤表达式精准限定	噪声降低 ~80%
模型重复加载浪费 GPU 内存	每次查询重新加载 BGE-M3	单例模式 + 延迟导入	模型加载时间从 30s 降至 ~0s
多路检索结果排序不公平	分数不可比，稠密高分压制稀疏	WeightedRanker 归一化 + 加权融合	排序公平性提升 ~90%

Remote & Agent 应用场景价值

Remote 场景价值：BGE-M3 模型和 Milvus 都是远程服务（GPU 服务器 + 向量数据库集群），团队成员通过 .env 配置（MILVUS_URL / BGE_M3_PATH）即可调用。延迟导入确保本地开发时不加载模型以节省显存。
Agent 落地场景：SearchEmbeddingNode 可封装为"向量检索 Agent"——Agent 接收 rewritten_query + item_names → 调用 BGE-M3 向量化 → 构建 AnnSearchRequest → 执行混合检索 → 返回 chunks。RANKER_WEIGHTS 可作为 Agent 的动态参数，根据不同场景切换（如型号检索加重稀疏权重 = 0.7）。

Git Commit 对应

本节向量检索节点对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议搜索 "search_embedding.py" 相关提交>

bash

cd shopkeeper_brain
git log --oneline --all -- knowledge/processor/query_process/nodes/search_embedding.py

知识库查询 —— 向量检索节点 ​

阅读指引 ​

技术栈健康度标签体系 ​

中英文对照表 ​

1. 任务目标 ​

1.1 本章目标 ​

1.2 涉及文件 ​

1.3 节点在流程中的位置 ​

2. 核心概念扫盲 ​

2.1 为什么需要混合检索？ ​

2.2 BGE-M3 模型 ​

2.3 稀疏向量表示 ​

2.4 Milvus 混合搜索 API ​

2.5 过滤表达式 ​

3. 向量检索业务处理流程（总） ​

3.1 整体流程图 ​

3.2 数据流转 ​

4. 向量检索业务处理流程（分） ​

4.1 目标 ​

4.2 需求分析 ​

4.2.1 功能需求 ​

4.2.2 技术依赖 ​

4.2.3 配置参数 ​

4.3 实现流程 ​

4.3.1 实现流程图 ​

4.3.2 具体实现步骤 ​

Step 1: 获取查询参数 ​

Step 2: 查询向量化 ​

Step 3: 构建过滤表达式 ​

Step 4: 构建混合搜索请求 ​

Step 5: 执行混合检索 ​

Step 6: 返回结果 ​

4.4 代码实现 ​

5. 测试入口 ​

5.1 测试代码 ​

5.2 运行测试 ​

5.3 预期输出 ​

6. 总结 ​

6.1 节点功能概览 ​

6.2 节点设计要点 ​

企业痛点映射 ​

Remote & Agent 应用场景价值 ​

Git Commit 对应 ​

知识库查询 —— 向量检索节点

阅读指引

技术栈健康度标签体系

中英文对照表

1. 任务目标

1.1 本章目标

1.2 涉及文件

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 为什么需要混合检索？

2.2 BGE-M3 模型

2.3 稀疏向量表示

2.4 Milvus 混合搜索 API

2.5 过滤表达式

3. 向量检索业务处理流程（总）

3.1 整体流程图

3.2 数据流转

4. 向量检索业务处理流程（分）

4.1 目标

4.2 需求分析

4.2.1 功能需求

4.2.2 技术依赖

4.2.3 配置参数

4.3 实现流程

4.3.1 实现流程图

4.3.2 具体实现步骤

Step 1: 获取查询参数

Step 2: 查询向量化

Step 3: 构建过滤表达式

Step 4: 构建混合搜索请求

Step 5: 执行混合检索

Step 6: 返回结果

4.4 代码实现

5. 测试入口

5.1 测试代码

5.2 运行测试

5.3 预期输出

6. 总结

6.1 节点功能概览

6.2 节点设计要点

企业痛点映射

Remote & Agent 应用场景价值

Git Commit 对应