知识库查询 —— 骨架代码与周边模块

本文档详细介绍知识库查询流程的骨架代码设计与实现，包括配置管理、异常处理、状态定义、节点基类和流程图构建。

学习理念：查询流程的骨架是知识库的"大脑"——它编排多路并行检索（向量/HyDE/KG/Web）、RRF 融合、Rerank 精排、答案生成。与导入流程的顺序执行不同，查询流程的核心是多路并发 + 条件路由 + 结果融合。
海外对标：查询骨架的"多路检索 → RRF 融合 → Rerank 精排"架构对标 Google 搜索的混合检索（语义+关键词+知识图谱）和 Bing Copilot 的 Multi-Source Answer Generation。LangGraph 的并行边 + 条件路由设计对标 Amazon Step Functions 的 Parallel + Choice 状态机。

本节 AI 替代率：~90% | 人工干预率：~10%

角色	能力范围
🤖 AI 擅长	配置类、异常层级、状态定义、基类模板、LangGraph 主图拼接、并行边设置
👤 人类需理解	条件路由的工程决策（何时跳过检索直接回答）、虚节点设计（multi_search/join 的分发汇合模式）

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习目标明确
🟢	§2 核心概念	~95%	~5%	多路检索/HyDE/RRF/Rerank 概念
🟡	§3 整体流程	~90%	~10%	5 个模块协作关系
🟠	§4.1 配置管理	~85%	~15%	检索参数 + RRF/Rerank/KG 配置
🟢	§4.2 异常处理	~95%	~5%	异常层级模式固定
🔴	§4.3 状态定义	~80%	~20%	多路结果字段设计是关键
🔴	§4.4 节点基类	~75%	~25%	流式输出追踪 + 任务完成标记
🟡	§4.5 主图构建	~85%	~15%	并行边 + 条件路由 + 虚节点
🟢	§5 测试入口	~95%	~5%	看预期输出和图结构
🟡	§6 总结	~90%	~10%	设计模式回顾

技术栈健康度标签体系

技术	健康度	建议
LangGraph	⏳ 成长期	查询流程的核心编排框架。并行边 + 条件路由 + 虚节点模式成熟，但 API 仍在迭代。
RRF	🟢 稳定	经典排名融合算法。源自 TREC 2009，被 Google / Elasticsearch / Milvus 广泛采用。
HyDE	⏳ 成长期	假设文档嵌入增强检索。2024-2026 年 RAG 领域热门技术，效果依赖 LLM 生成质量。
BGE-Reranker	🔥 巅峰	交叉编码器精排模型，对 RRF 融合结果二次排序。
MCP (Model Context Protocol)	⏳ 成长期	Anthropic 提出的 AI 模型工具调用标准协议。本项目用于 Web 搜索集成。

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

中英文对照表

English	中文	本质
Multi-Route Retrieval	多路检索	同时用多种方式检索以提高召回率
HyDE (Hypothetical Document Embeddings)	假设文档嵌入	先生成假设答案再检索，提升语义匹配
RRF (Reciprocal Rank Fusion)	倒数排名融合	基于排名位置而非原始分数的多路融合
Rerank	重排序	用精排模型对召回结果重新打分
Conditional Edge	条件边	根据状态决定是否跳过检索
Virtual Node	虚节点	不执行实际逻辑，仅用于分发/汇合语义
Stream Output	流式输出	逐 token 返回 LLM 生成结果
MCP	模型上下文协议	AI 模型调用外部工具的标准协议

💡 程序员比喻
查询骨架 就像后端微服务的 API Gateway——先鉴权（item_name_confirm），再路由到不同的 upstream service（多路检索），最后聚合结果返回。
并行边 就像 Promise.all()——四条检索路径同时执行，谁先返回不重要，等所有都完成才进下一步。
条件路由 就像 if-else 短路求值——有缓存（answer）就直接返回，不需要查数据库。
虚节点 (multi_search/join) 就像 fork() / join()——创建四个子进程，等它们都结束。
RRF + Rerank 就像 Code Review 的两轮评审——RRF = 广度评审（多人都通过的 PR 排在前面），Rerank = 深度评审（找专家逐条精排）。

1. 任务目标

1.1 本章目标

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

理解查询流程架构：掌握多路并行检索、RRF 融合、Rerank 重排序的核心概念
设计可扩展的查询骨架：学会使用基类、配置、异常等模式构建健壮的查询流程
实现查询流程主图：构建完整的知识库查询工作流
编写可测试的代码：通过 if __name__ == "__main__" 验证流程

1.2 涉及文件

knowledge/processor/query_process/
├── config.py        # 配置管理模块
├── exceptions.py    # 自定义异常类
├── state.py         # 图状态类型定义
├── base.py          # 节点基类
├── main_graph.py    # 主图定义与构建
└── __init__.py      # 模块导出

1.3 文件依赖关系

config.py ─────────────────────────┐
                                   │
exceptions.py ─────────────────────┼──> base.py ──> main_graph.py
                                   │
state.py ──────────────────────────┘

1.4 查询流程 vs 导入流程

对比项	导入流程	查询流程
目标	将文档处理并存入向量库	根据用户问题检索并生成答案
流程特点	顺序执行，单一路径	多路并行检索 + 融合
关键技术	PDF 转换、切片、向量化	多路检索、RRF、Rerank
状态复杂度	中等	较高（多路结果合并）

2. 核心概念扫盲

2.1 查询流程核心概念

概念	说明	作用
多路检索	同时使用多种检索方式（向量、HyDE、知识图谱、网页）	提高召回率
HyDE	Hypothetical Document Embeddings，先生成假设答案再检索	提升语义匹配
RRF	Reciprocal Rank Fusion，倒数排名融合	合并多路结果
Rerank	重排序，使用精排模型对召回结果重新打分	提高精度
商品名确认	从用户问题中识别具体产品名称	精准定位知识

2.2 LangGraph 并行执行

LangGraph 支持多路并行执行节点：

实现方式：

🟡 【P1 看注释就行】 config.py 与导入流程类似，新增检索参数（embedding_search_limit / hyde_search_limit）和 RRF 参数。注意 reset_config() 用于测试重置。

python

# 分发：一个节点连接到多个下游节点
workflow.add_edge("multi_search", "search_embedding")
workflow.add_edge("multi_search", "search_embedding_hyde")
workflow.add_edge("multi_search", "query_kg")
workflow.add_edge("multi_search", "web_search_mcp")

# 汇合：多个节点连接到同一个下游节点
workflow.add_edge("search_embedding", "join")
workflow.add_edge("search_embedding_hyde", "join")
workflow.add_edge("query_kg", "join")
workflow.add_edge("web_search_mcp", "join")

2.3 RRF 融合算法

Reciprocal Rank Fusion 是一种简单有效的结果融合算法：

RRF_score(d) = Σ (1 / (k + rank_i(d)))

d：文档
k：平滑常数（通常取 60）
rank_i(d)：文档在第 i 路检索中的排名

直观理解：

排名越靠前，得分越高
多路都排名靠前的文档，最终得分更高

2.4 条件路由

查询流程中，如果商品名确认节点直接给出了答案（如闲聊回复），则跳过后续检索：

🟡 【P1 看注释就行】 异常层级与导入流程类似，新增 SearchError / EntityAlignmentError / RerankError / ItemNameConfirmError。

python

def route_after_item_confirm(state: QueryGraphState) -> bool:
    """根据是否已有答案决定路由"""
    if state.get("answer"):
        return True   # 直接到答案输出
    return False      # 继续检索流程

3. 知识库查询业务处理流程（总）

3.1 整体流程图

3.2 骨架模块职责

模块	职责	重要性
config.py	管理检索参数、LLM 配置、数据库连接等	配置与代码分离
exceptions.py	定义查询相关异常（搜索、重排序、LLM 等）	错误可追踪
state.py	定义查询状态结构，多路检索结果存储	数据契约
base.py	定义查询节点基类，统一执行逻辑	代码复用
main_graph.py	构建查询工作流，编排并行检索和融合	流程编排

3.3 数据流向

4. 知识库查询业务处理流程（分）

4.1 配置管理模块

4.1.1 目标

集中管理查询流程的所有配置项
支持环境变量覆盖
提供配置验证
实现单例模式

4.1.2 需求分析

配置分类：

类别	配置项	说明
检索配置	`embedding_search_limit`	向量搜索返回数量
	`hyde_search_limit`	HyDE 搜索返回数量
Rerank 配置	`rerank_max_topk`	重排序最大 TopK
	`rerank_gap_ratio`	断崖检测阈值
RRF 配置	`rrf_k`	RRF 平滑常数
	`rrf_kg_weight`	知识图谱权重
LLM 配置	`openai_api_base`	API 地址
	`default_model`	默认模型名
数据库配置	`milvus_url`	Milvus 地址
	`neo4j_uri`	Neo4j 地址

4.1.3 实现流程

4.1.4 代码实现

🔥 【P0 必须要学】 QueryGraphState 定义了多路检索结果字段（embedding_chunks / hyde_embedding_chunks / kg_chunks / web_search_docs）和融合结果字段（rrf_chunks / reranked_docs）。理解这些字段的命名规则（谁写入、谁读取）是理解查询流程数据流的关键。

python

# knowledge/processor/query_process/config.py

"""查询流程配置管理模块

集中管理所有配置项，支持环境变量覆盖。
"""

from dataclasses import dataclass, field
from typing import Optional
import os


@dataclass
class QueryConfig:
    """查询流程配置。

    Attributes:
        max_context_chars: 上下文最大字符数限制。
        rerank_max_topk: 重排序动态 TopK 硬上限。
        rerank_min_topk: 重排序最小 TopK。
        rerank_gap_ratio: 重排序断崖阈值（相对）。
        rerank_gap_abs: 重排序断崖阈值（绝对）。
        rrf_k: RRF 算法常数。
        rrf_kg_weight: RRF 融合中知识图谱权重。
        ...
    """

    # ==================== 文本处理配置 ====================
    max_context_chars: int = 12000  # 上下文最大字符数

    # ==================== Rerank 配置 ====================
    rerank_max_topk: int = 10       # 重排序最大返回数
    rerank_min_topk: int = 3        # 重排序最小返回数
    rerank_gap_ratio: float = 0.25  # 断崖检测阈值（相对）
    rerank_gap_abs: float = 0.5     # 断崖检测阈值（绝对）

    # ==================== RRF 配置 ====================
    rrf_k: int = 60                 # RRF 平滑常数
    rrf_kg_weight: float = field(
        default_factory=lambda: float(os.getenv("RRF_KG_WEIGHT", "0.7"))
    )
    rrf_max_results: int = 10       # RRF 最大返回结果数

    # ==================== 检索配置 ====================
    embedding_search_limit: int = 10  # 向量搜索返回数量
    hyde_search_limit: int = 5        # HyDE 搜索返回数量

    # ==================== 知识图谱配置 ====================
    kg_entity_align_min_score: Optional[float] = field(
        default_factory=lambda: (
            float(os.getenv("KG_ENTITY_ALIGN_MIN_SCORE"))
            if os.getenv("KG_ENTITY_ALIGN_MIN_SCORE")
            else None
        )
    )
    kg_max_seed_candidates: int = 3   # 每个实体最大种子候选数
    kg_max_total_seeds: int = 30      # 总种子节点上限
    kg_max_triples_per_seed: int = 50 # 每个种子最大三元组数
    kg_max_total_triples: int = 200   # 总三元组上限
    kg_max_total_chunks: int = 200    # 总切片上限

    # ==================== LLM 配置 ====================
    openai_api_base: str = field(
        default_factory=lambda: os.getenv("OPENAI_API_BASE", "")
    )
    openai_api_key: str = field(
        default_factory=lambda: os.getenv("OPENAI_API_KEY", "")
    )
    default_model: str = field(
        default_factory=lambda: os.getenv("MODEL", "")
    )
    item_model: str = field(
        default_factory=lambda: os.getenv("ITEM_MODEL", "")
    )

    # ==================== Milvus 配置 ====================
    milvus_url: str = field(
        default_factory=lambda: os.getenv("MILVUS_URL", "")
    )
    chunks_collection: str = field(
        default_factory=lambda: os.getenv("CHUNKS_COLLECTION", "")
    )
    item_name_collection: str = field(
        default_factory=lambda: os.getenv("ITEM_NAME_COLLECTION", "")
    )
    entity_name_collection: str = field(
        default_factory=lambda: os.getenv("ENTITY_NAME_COLLECTION", "")
    )

    # ==================== Neo4j 配置 ====================
    neo4j_uri: str = field(
        default_factory=lambda: os.getenv("NEO4J_URI", "")
    )
    neo4j_username: str = field(
        default_factory=lambda: os.getenv("NEO4J_USERNAME", "")
    )
    neo4j_password: str = field(
        default_factory=lambda: os.getenv("NEO4J_PASSWORD", "")
    )
    neo4j_database: str = field(
        default_factory=lambda: os.getenv("NEO4J_DATABASE", "neo4j")
    )

    # ==================== MCP 配置 ====================
    mcp_dashscope_base_url: str = field(
        default_factory=lambda: os.getenv("MCP_DASHSCOPE_BASE_URL", "")
    )

    @classmethod
    def from_env(cls) -> "QueryConfig":
        """从环境变量加载配置。"""
        return cls()

    def validate(self, strict: bool = False) -> None:
        """验证配置是否完整。

        Args:
            strict: 是否严格模式，严格模式下缺少配置会抛出异常。
        """
        required_fields = ["milvus_url", "chunks_collection"]
        missing = [k for k in required_fields if not getattr(self, k)]

        if missing:
            msg = f"缺少必要配置: {missing}"
            if strict:
                raise ValueError(msg)
            else:
                print(f"警告: {msg}")


# ==================== 全局单例 ====================
_config: Optional[QueryConfig] = None


def get_config() -> QueryConfig:
    """获取配置单例。"""
    global _config
    if _config is None:
        _config = QueryConfig.from_env()
    return _config


def reset_config() -> None:
    """重置配置（用于测试）。"""
    global _config
    _config = None

与导入流程配置对比：

特性	导入流程	查询流程
检索参数	无	`rerank_`, `rrf_`, `embedding_search_limit`
KG 参数	无	`kg_max_*`
MCP 配置	无	`mcp_dashscope_base_url`
重置功能	无	`reset_config()` 用于测试

4.2 异常处理模块

4.2.1 目标

定义查询流程特有的异常层级
提供清晰的错误分类
支持错误溯源

4.2.2 需求分析

异常层级结构：

QueryProcessError (基础异常)
├── ConfigurationError      # 配置错误
├── SearchError             # 搜索错误（向量/混合/网页）
├── EmbeddingError          # 向量化错误
├── LLMError                # LLM 调用错误
├── StorageError            # 存储错误
│   ├── MilvusError         # Milvus 错误
│   ├── Neo4jError          # Neo4j 错误
│   └── MongoDBError        # MongoDB 错误（历史记录）
├── ValidationError         # 数据验证错误
├── EntityAlignmentError    # 实体对齐错误
├── RerankError             # 重排序错误
└── ItemNameConfirmError    # 商品名确认错误

4.2.3 实现流程

4.2.4 代码实现

🔥 【P0 必须要学】 查询流程 BaseNode 与导入流程的区别：新增 add_done_task 完成标记 + is_stream 流式支持。__call__ 在执行完 process 后立即标记任务完成。

python

# knowledge/processor/query_process/exceptions.py

"""查询流程自定义异常类

统一错误处理，提供更清晰的错误信息。
"""


class QueryProcessError(Exception):
    """查询流程基础异常。

    Attributes:
        node_name: 发生异常的节点名称。
        cause: 原始异常对象。
    """

    def __init__(
        self,
        message: str,
        node_name: str = "",
        cause: Exception = None
    ):
        """初始化异常。

        Args:
            message: 错误信息。
            node_name: 节点名称。
            cause: 原始异常。
        """
        self.node_name = node_name
        self.cause = cause
        super().__init__(message)

    def __str__(self):
        """格式化异常信息。"""
        parts = []
        if self.node_name:
            parts.append(f"[{self.node_name}]")
        parts.append(super().__str__())
        if self.cause:
            parts.append(f"(原因: {self.cause})")
        return " ".join(parts)


class ConfigurationError(QueryProcessError):
    """配置错误：环境变量缺失或配置值无效。"""
    pass


class SearchError(QueryProcessError):
    """搜索错误：向量搜索、混合搜索或网络搜索失败。"""
    pass


class EmbeddingError(QueryProcessError):
    """向量化错误：模型调用失败、向量生成异常。"""
    pass


class LLMError(QueryProcessError):
    """LLM 调用错误：API 调用失败、响应解析失败。"""
    pass


class StorageError(QueryProcessError):
    """存储错误：数据库操作失败。"""
    pass


class MilvusError(StorageError):
    """Milvus 存储错误。"""
    pass


class Neo4jError(StorageError):
    """Neo4j 存储错误。"""
    pass


class MongoDBError(StorageError):
    """MongoDB 存储错误（用于历史记录）。"""
    pass


class ValidationError(QueryProcessError):
    """数据验证错误：输入数据不符合预期。"""
    pass


class EntityAlignmentError(QueryProcessError):
    """实体对齐错误：知识图谱实体对齐失败。"""
    pass


class RerankError(QueryProcessError):
    """重排序错误：文档重排序失败。"""
    pass


class ItemNameConfirmError(QueryProcessError):
    """商品名称确认错误：识别或确认失败。"""
    pass

使用示例：

🔥 【P0 必须要学】 main_graph.py 是查询流程的调度核心。重点理解：(1) route_after_item_confirm 的条件路由（有 answer 直接到 answer_output）(2) 并行边：multi_search → 4 个检索节点（LangGraph 自动并行）(3) 虚节点：multi_search（分发）+ join（汇合）不执行逻辑，仅用于图结构语义。

python

try:
    # 向量搜索操作
    results = milvus_client.search(query_vector)
except Exception as e:
    raise SearchError(
        message="向量搜索失败",
        node_name="search_embedding",
        cause=e
    )

输出效果：

SearchError: [search_embedding] 向量搜索失败 (原因: ConnectionError: ...)

4.3 状态定义模块

4.3.1 目标

定义查询流程的完整状态结构
支持多路检索结果存储
提供默认状态工厂函数

4.3.2 需求分析

状态字段分类：

类别	字段	说明
会话信息	`session_id`, `message_id`	会话追踪
输入数据	`original_query`, `item_names`, `history`	用户输入
检索结果	`embedding_chunks`, `hyde_embedding_chunks`, `kg_chunks`, `web_search_docs`	多路召回
融合结果	`rrf_chunks`, `reranked_docs`	融合排序
输出数据	`prompt`, `answer`	最终输出
控制标志	`is_stream`	流式输出

4.3.3 实现流程

4.3.4 代码实现

🟢 【P2 后面可以查】 测试代码量大但模式固定——mock 查询 → run_query → 查看多路检索结果数。重点看预期输出中的节点执行顺序（10 个节点）和图结构（多路并行分叉）。

python

# knowledge/processor/query_process/state.py

"""查询流程状态类型定义

定义完整的查询状态结构和辅助函数。
"""

from typing import TypedDict, List
import copy


class QueryGraphState(TypedDict):
    """查询流程图状态。

    包含整个查询流程中传递的所有数据。

    Attributes:
        session_id: 会话 ID，用于追踪多轮对话。
        message_id: 消息 ID，标识单次查询。
        original_query: 原始用户查询。
        embedding_chunks: 向量检索结果列表。
        hyde_embedding_chunks: HyDE 检索结果列表。
        rrf_chunks: RRF 融合后的切片列表。
        web_search_docs: 网页搜索结果列表。
        reranked_docs: 重排序后的文档列表。
        prompt: 构造的提示词。
        answer: 最终生成的答案。
        item_names: 识别的商品名称列表。
        rewritten_query: 重写后的查询。
        history: 历史对话列表。
        is_stream: 是否启用流式输出。
        kg_chunks: 知识图谱相关切片列表。
        kg_triples: 知识图谱三元组列表。
    """

    session_id: str
    message_id: str
    original_query: str
    embedding_chunks: list
    hyde_embedding_chunks: list
    rrf_chunks: list
    web_search_docs: list
    reranked_docs: list
    prompt: str
    answer: str
    item_names: List[str]
    rewritten_query: str
    history: list
    is_stream: bool
    kg_chunks: list
    kg_triples: list


# ==================== 默认状态 ====================

DEFAULT_STATE: QueryGraphState = {
    "session_id": "",               # 会话 ID
    "message_id": "",               # 消息 ID
    "original_query": "",           # 原始查询
    "embedding_chunks": [],         # 向量检索结果
    "hyde_embedding_chunks": [],    # HyDE 检索结果
    "rrf_chunks": [],               # RRF 融合后的切片
    "web_search_docs": [],          # 网页搜索结果
    "reranked_docs": [],            # 重排序后的文档
    "prompt": "",                   # 提示词
    "answer": "",                   # 答案
    "item_names": [],               # 商品名称
    "rewritten_query": "",          # 重写查询
    "history": [],                  # 历史对话
    "is_stream": False,             # 是否流式输出
    "kg_chunks": [],                # 知识图谱切片
    "kg_triples": []                # 知识图谱关系
}


def create_default_state(**overrides) -> QueryGraphState:
    """创建默认状态，支持字段覆盖。

    Args:
        **overrides: 要覆盖的字段键值对。

    Returns:
        新的状态实例，包含默认值和覆盖值。

    Examples:
        >>> state = create_default_state(
        ...     session_id="session_001",
        ...     original_query="万用表如何测量电压？"
        ... )
    """
    state = copy.deepcopy(DEFAULT_STATE)
    state.update(overrides)
    return state


def get_default_state() -> QueryGraphState:
    """获取默认状态副本。

    Returns:
        状态副本，避免修改全局默认值。
    """
    return copy.deepcopy(DEFAULT_STATE)


# 兼容旧版变量名
graph_default_state = DEFAULT_STATE

关键设计点：

多路结果字段
- embedding_chunks、hyde_embedding_chunks、kg_chunks、web_search_docs
- 每个检索节点写入独立字段，避免冲突
融合结果字段
- rrf_chunks：RRF 融合后
- reranked_docs：Rerank 后
兼容旧版
- graph_default_state = DEFAULT_STATE

4.4 节点基类模块

4.4.1 目标

定义统一的查询节点接口
提供通用功能（日志、任务追踪、异常处理）
支持流式输出追踪

4.4.2 需求分析

查询节点通用功能：

日志记录：节点开始/结束、步骤日志
任务追踪：注册当前节点（支持流式模式）
任务完成标记：标记节点处理完成
异常处理：统一包装异常，添加节点信息
配置注入：提供配置对象访问

4.4.3 实现流程

4.4.4 代码实现

🟡 【P1 看注释就行】 扩展指南——添加新检索通道的 5 步模板。理解 add_edge 的分发→汇合模式即可。

python

# knowledge/processor/query_process/base.py

"""查询流程节点基类

定义统一的节点接口规范，提供通用功能。
"""

from abc import ABC, abstractmethod
from typing import TypeVar, Optional
import logging

from knowledge.processor.query_process.config import QueryConfig, get_config
from knowledge.processor.query_process.exceptions import QueryProcessError
from knowledge.tools.task_utils import add_running_task, add_done_task

T = TypeVar("T")  # 泛型状态类型


class BaseNode(ABC):
    """查询流程节点基类。

    所有节点类都应继承此基类，实现 process 方法。
    基类提供统一的日志、任务追踪和错误处理。

    Attributes:
        name: 节点名称，子类应覆盖。
        config: 配置。
        logger: 日志记录器。

    Example:
        >>> class MyNode(BaseNode):
        ...     name = "my_node"
        ...
        ...     def process(self, state):
        ...         # 实现具体逻辑
        ...         return state
        ...
        >>> # 作为 LangGraph 节点使用
        >>> node = MyNode()
        >>> workflow.add_node("my_node", node)
    """

    name: str = "base_node"

    def __init__(self, config: Optional[QueryConfig] = None):
        """初始化节点。

        Args:
            config: 配置对象，默认使用全局配置。
        """
        self.config = config or get_config()
        self.logger = logging.getLogger(f"query.{self.name}")

    def __call__(self, state: T) -> T:
        """节点执行入口。

        LangGraph 调用节点时会调用此方法。
        提供统一的日志输出、任务追踪和异常处理。

        Args:
            state: 图状态字典。

        Returns:
            更新后的状态字典。

        Raises:
            QueryProcessError: 节点执行失败时抛出。
        """
        self.logger.info(f"--- {self.name} 开始 ---")

        # 注册任务追踪
        session_id = state.get("session_id", "") if isinstance(state, dict) else ""
        is_stream = state.get("is_stream", False) if isinstance(state, dict) else False

        if session_id:
            try:
                add_running_task(session_id, self.name, is_stream)
            except Exception as e:
                self.logger.warning(f"任务追踪注册失败: {e}")

        try:
            result = self.process(state)
            self.logger.info(f"--- {self.name} 完成 ---")

            # 标记任务完成
            if session_id:
                try:
                    add_done_task(session_id, self.name, is_stream)
                except Exception as e:
                    self.logger.warning(f"任务完成标记失败: {e}")

            return result
        except QueryProcessError:
            # 已经是自定义异常，直接抛出
            raise
        except Exception as e:
            self.logger.error(f"{self.name} 执行失败: {e}")
            raise QueryProcessError(
                message=str(e),
                node_name=self.name,
                cause=e
            )

    @abstractmethod
    def process(self, state: T) -> T:
        """节点核心处理逻辑。

        子类必须实现此方法。

        Args:
            state: 图状态字典。

        Returns:
            更新后的状态字典。
        """
        pass

    def log_step(self, step_name: str, message: str = ""):
        """记录步骤日志。

        Args:
            step_name: 步骤名称。
            message: 附加信息。
        """
        log_msg = f"[{step_name}]"
        if message:
            log_msg += f" {message}"
        self.logger.info(log_msg)


def setup_logging(level: int = logging.INFO):
    """配置查询流程日志。

    Args:
        level: 日志级别，默认 INFO。
    """
    logging.basicConfig(
        level=level,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

与导入流程基类对比：

特性	导入流程 BaseNode	查询流程 BaseNode
任务追踪	`add_running_task(task_id, name)`	`add_running_task(session_id, name, is_stream)`
完成标记	无	`add_done_task(session_id, name, is_stream)`
日志命名空间	`import.{name}`	`query.{name}`
异常类型	`ImportProcessError`	`QueryProcessError`

4.5 主图构建模块

4.5.1 目标

构建完整的查询工作流图
实现多路并行检索
支持条件路由（有答案时跳过检索）
提供便捷的测试入口

4.5.2 需求分析

流程结构：

入口节点：item_name_confirm 商品名确认
条件路由：有答案 → answer_output；无答案 → multi_search
并行检索：embedding、hyde、kg、web 四路并行
结果汇合：join 节点
融合排序：rrf → rerank
答案生成：answer_output
结束节点：END

4.5.3 实现流程

4.5.4 代码实现

python

# knowledge/processor/query_process/main_graph.py

"""查询流程主图

使用 LangGraph 构建知识库查询工作流。
"""

from langgraph.graph import StateGraph, END
from dotenv import load_dotenv

from knowledge.processor.query_process.state import (
    QueryGraphState,
    create_default_state
)
from knowledge.processor.query_process.nodes import (
    ItemNameConfirmNode,
    SearchEmbeddingNode,
    SearchEmbeddingHydeNode,
    QueryKgNode,
    WebSearchMcpNode,
    RrfNode,
    RerankNode,
    AnswerOutputNode,
)

# 加载环境变量
load_dotenv()


def route_after_item_confirm(state: QueryGraphState) -> bool:
    """商品名称确认后的路由逻辑。

    根据是否已有答案决定是否跳过搜索直接输出。

    Args:
        state: 查询图状态。

    Returns:
        True 表示已有答案需要跳过搜索，False 表示继续搜索流程。
    """
    if state.get("answer"):
        return True
    return False


def create_query_graph() -> StateGraph:
    """创建查询流程图。

    Returns:
        编译后的 StateGraph 实例。

    流程结构::

        item_name_confirm
              │
              ├── (有答案) ──────────────────────────> answer_output
              │                                              │
              └── (无答案) ──> multi_search ─────┬──────────>│
                                   │             │           │
                         ┌─────────┼─────────────┼───────┐   │
                         │         │             │       │   │
                         v         v             v       v   │
                   embedding  hyde_embedding  query_kg  web  │
                         │         │             │       │   │
                         └─────────┴─────────────┴───────┘   │
                                       │                     │
                                       v                     │
                                     join                    │
                                       │                     │
                                       v                     │
                                      rrf                    │
                                       │                     │
                                       v                     │
                                    rerank                   │
                                       │                     │
                                       v                     │
                               answer_output <───────────────┘
                                       │
                                       v
                                      END
    """
    # 1. 创建状态图
    workflow = StateGraph(QueryGraphState)

    # 2. 实例化节点
    nodes = {
        "item_name_confirm": ItemNameConfirmNode(),
        "multi_search": lambda x: x,              # 多路搜索分发（虚节点）
        "search_embedding": SearchEmbeddingNode(),
        "search_embedding_hyde": SearchEmbeddingHydeNode(),
        "query_kg": QueryKgNode(),
        "web_search_mcp": WebSearchMcpNode(),
        "join": lambda x: {},                     # 多路搜索汇合（虚节点）
        "rrf": RrfNode(),
        "rerank": RerankNode(),
        "answer_output": AnswerOutputNode(),
    }

    # 3. 添加节点
    for name, node in nodes.items():
        workflow.add_node(name, node)

    # 4. 设置入口点
    workflow.set_entry_point("item_name_confirm")

    # 5. 添加条件边：商品名称确认后根据是否有答案路由
    workflow.add_conditional_edges(
        "item_name_confirm",
        route_after_item_confirm,
        {
            False: "multi_search",   # 无答案，继续检索
            True: "answer_output"    # 有答案，直接输出
        }
    )

    # 6. 多路搜索分发（并行执行）
    workflow.add_edge("multi_search", "search_embedding")
    workflow.add_edge("multi_search", "search_embedding_hyde")
    workflow.add_edge("multi_search", "query_kg")
    workflow.add_edge("multi_search", "web_search_mcp")

    # 7. 多路搜索汇合
    workflow.add_edge("search_embedding", "join")
    workflow.add_edge("search_embedding_hyde", "join")
    workflow.add_edge("query_kg", "join")
    workflow.add_edge("web_search_mcp", "join")

    # 8. 顺序边
    workflow.add_edge("join", "rrf")
    workflow.add_edge("rrf", "rerank")
    workflow.add_edge("rerank", "answer_output")
    workflow.add_edge("answer_output", END)

    # 9. 编译并返回
    return workflow.compile()


# 创建全局图实例
query_app = create_query_graph()


def run_query(
    query: str,
    session_id: str = "",
    item_names: list = None,
    is_stream: bool = False
) -> dict:
    """便捷函数：运行查询流程。

    Args:
        query: 用户查询文本。
        session_id: 会话 ID。
        item_names: 已知的商品名称列表。
        is_stream: 是否启用流式输出。

    Returns:
        最终状态字典。
    """
    # 1. 创建初始状态
    initial_state = create_default_state(
        session_id=session_id or "default",
        original_query=query,
        item_names=item_names or [],
        is_stream=is_stream,
    )

    final_state = None

    # 2. 运行图的工作流节点
    for event in query_app.stream(initial_state):
        for key, value in event.items():
            print(f"节点: {key}")
            final_state = value

    # 3. 返回最终状态
    return final_state or initial_state

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

5. 测试入口

在 main_graph.py 文件末尾添加测试入口：

python

# ==================== 命令行入口 ====================

if __name__ == "__main__":
    import sys
    from knowledge.processor.query_process.base import setup_logging

    # 1. 配置日志
    setup_logging()

    print("=" * 60)
    print("知识库查询流程测试")
    print("=" * 60)

    # 2. 准备测试查询
    # 支持命令行参数或使用默认值
    if len(sys.argv) > 1:
        test_query = " ".join(sys.argv[1:])
    else:
        test_query = "万用表如何测量电压？"

    print(f"查询: {test_query}")
    print("-" * 60)

    # 3. 运行查询流程
    try:
        result = run_query(
            query=test_query,
            session_id="test_001",
            item_names=[],        # 初始不指定商品名
            is_stream=False       # 非流式输出
        )

        print("-" * 60)
        print("流程完成!")
        print("-" * 60)

        # 4. 输出结果摘要
        print(f"识别商品: {result.get('item_names', [])}")
        print(f"检索切片数: {len(result.get('embedding_chunks', []))}")
        print(f"HyDE 切片数: {len(result.get('hyde_embedding_chunks', []))}")
        print(f"KG 切片数: {len(result.get('kg_chunks', []))}")
        print(f"RRF 融合数: {len(result.get('rrf_chunks', []))}")
        print(f"Rerank 结果数: {len(result.get('reranked_docs', []))}")
        print("-" * 60)

        # 5. 输出答案（截取前 500 字符）
        answer = result.get('answer', 'N/A')
        print("答案:")
        print(answer[:500] + "..." if len(answer) > 500 else answer)

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

    # 6. 打印图结构（ASCII 可视化）
    print("-" * 60)
    print("图结构:")
    query_app.get_graph().print_ascii()

运行测试：

bash

# 进入项目目录
cd knowledge

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

# 使用默认查询运行
python -m knowledge.processor.query_process.main_graph

# 使用自定义查询运行
python -m knowledge.processor.query_process.main_graph "示波器怎么使用？"

预期输出：

============================================================
知识库查询流程测试
============================================================
查询: 万用表如何测量电压？
------------------------------------------------------------
节点: item_name_confirm
节点: multi_search
节点: search_embedding
节点: search_embedding_hyde
节点: query_kg
节点: web_search_mcp
节点: join
节点: rrf
节点: rerank
节点: answer_output
------------------------------------------------------------
流程完成!
------------------------------------------------------------
识别商品: ['万用表']
检索切片数: 10
HyDE 切片数: 5
KG 切片数: 3
RRF 融合数: 10
Rerank 结果数: 5
------------------------------------------------------------
答案:
万用表测量电压的步骤如下：

1. **选择量程**：将旋钮转到直流电压（V-）或交流电压（V~）档位...

------------------------------------------------------------
图结构:
              +-------------------+
              |     __start__     |
              +-------------------+
                       *
                       *
                       *
           +---------------------+
           | item_name_confirm   |
           +---------------------+
                  *        *
                 *          *
                *            *
    +---------------+   +--------------+
    | multi_search  |   |answer_output |
    +---------------+   +--------------+
     *   *   *   *              *
    *    *   *    *             *
   *     *   *     *            *
  +--+ +--+ +--+ +--+           *
  |E | |H | |K | |W |           *
  +--+ +--+ +--+ +--+           *
     *   *   *   *              *
      *  *   *  *               *
       * *   * *                *
        +-----+                 *
        | join|                 *
        +-----+                 *
            *                   *
            *                   *
            *                   *
         +-----+                *
         | rrf |                *
         +-----+                *
            *                   *
            *                   *
            *                   *
        +--------+              *
        | rerank |              *
        +--------+              *
            *                   *
             *                 *
              *               *
           +--------------+
           |answer_output |
           +--------------+
                  *
                  *
                  *
             +---------+
             | __end__ |
             +---------+

6. 总结

6.1 关键设计模式

模式	应用场景	优势
单例模式	`get_config()`	全局共享配置，避免重复创建
模板方法模式	`BaseNode.__call__` + `process`	统一执行流程，子类专注业务
工厂模式	`create_default_state()`	灵活创建状态实例
异常链	`QueryProcessError(cause=e)`	保留原始错误信息
虚节点	`multi_search`, `join`	实现分发/汇合语义

6.2 查询流程关键特性

特性	说明
多路并行检索	向量、HyDE、KG、Web 四路并行，提高召回率
条件路由	有直接答案时跳过检索，节省资源
RRF 融合	公平合并多路结果，避免单路主导
动态 Rerank	断崖检测，自动确定返回数量
流式支持	`is_stream` 标志支持流式输出

6.3 扩展指南

添加新的检索通道：

创建节点类，继承 BaseNode
在状态中添加对应的结果字段（如 new_search_chunks）
在 main_graph.py 中注册节点
添加分发边和汇合边
在 RRF 节点中处理新通道的结果

python

# 新检索节点示例
class NewSearchNode(BaseNode):
    name = "new_search"

    def process(self, state):
        query = state.get("original_query", "")

        # 执行检索逻辑
        results = self.search(query)

        # 更新状态
        state["new_search_chunks"] = results
        return state

企业痛点映射

痛点	传统方案	AI Agent 查询骨架方案	效率提升
单路检索召回率低	只用关键词或只用向量搜索	4 路并行检索（向量/HyDE/KG/Web）	召回率提升 ~50%（预估）
多路结果手动融合	人工看每路结果，逐一比对	RRF 自动融合 + setdefault 去重	融合时间从 10min 降至 ~1s
无答案时仍空跑全流程	不管有没有答案都跑整条链路	条件路由：有 answer 直接输出	无意义计算减少 ~70%
新检索通道接入成本高	改代码、改状态、改流程	虚节点分发 + 状态字段 + 注册节点 3 步	新通道接入从 1h 降至 ~10min

Remote & Agent 应用场景价值

Remote 场景价值：查询骨架的并行检索设计天然适配微服务架构——每条检索路径可以独立部署为远程微服务（向量搜索服务 / HyDE 服务 / KG 服务），通过 LangGraph 的条件边和并行边协同。multi_search 虚节点在分布式环境下可替换为消息队列分发。
Agent 落地场景：查询骨架本身就是"问答 Agent"的完整框架。item_name_confirm = Agent 的意图识别 + 工具选择，multi_search = Agent 的 Tool Calling 分发，rrf + rerank = Agent 的结果聚合，answer_output = Agent 的最终回复生成。

Git Commit 对应

本节查询流程骨架代码对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议在项目仓库中搜索 "query_process/" 相关提交>

bash

cd shopkeeper_brain
git log --oneline --all -- knowledge/processor/query_process/

知识库查询 —— 骨架代码与周边模块 ​

阅读指引 ​

技术栈健康度标签体系 ​

中英文对照表 ​

1. 任务目标 ​

1.1 本章目标 ​

1.2 涉及文件 ​

1.3 文件依赖关系 ​

1.4 查询流程 vs 导入流程 ​

2. 核心概念扫盲 ​

2.1 查询流程核心概念 ​

2.2 LangGraph 并行执行 ​

2.3 RRF 融合算法 ​

2.4 条件路由 ​

3. 知识库查询业务处理流程（总） ​

3.1 整体流程图 ​

3.2 骨架模块职责 ​

3.3 数据流向 ​

4. 知识库查询业务处理流程（分） ​

4.1 配置管理模块 ​

4.1.1 目标 ​

4.1.2 需求分析 ​

4.1.3 实现流程 ​

4.1.4 代码实现 ​

4.2 异常处理模块 ​

4.2.1 目标 ​

4.2.2 需求分析 ​

4.2.3 实现流程 ​

4.2.4 代码实现 ​

4.3 状态定义模块 ​

4.3.1 目标 ​

4.3.2 需求分析 ​

4.3.3 实现流程 ​

4.3.4 代码实现 ​

4.4 节点基类模块 ​

4.4.1 目标 ​

4.4.2 需求分析 ​

4.4.3 实现流程 ​

4.4.4 代码实现 ​

4.5 主图构建模块 ​

4.5.1 目标 ​

4.5.2 需求分析 ​

4.5.3 实现流程 ​

4.5.4 代码实现 ​

5. 测试入口 ​

6. 总结 ​

6.1 关键设计模式 ​

6.2 查询流程关键特性 ​

6.3 扩展指南 ​

企业痛点映射 ​

Remote & Agent 应用场景价值 ​

Git Commit 对应 ​

知识库查询 —— 骨架代码与周边模块

阅读指引

技术栈健康度标签体系

中英文对照表

1. 任务目标

1.1 本章目标

1.2 涉及文件

1.3 文件依赖关系

1.4 查询流程 vs 导入流程

2. 核心概念扫盲

2.1 查询流程核心概念

2.2 LangGraph 并行执行

2.3 RRF 融合算法

2.4 条件路由

3. 知识库查询业务处理流程（总）

3.1 整体流程图

3.2 骨架模块职责

3.3 数据流向

4. 知识库查询业务处理流程（分）

4.1 配置管理模块

4.1.1 目标

4.1.2 需求分析

4.1.3 实现流程

4.1.4 代码实现

4.2 异常处理模块

4.2.1 目标

4.2.2 需求分析

4.2.3 实现流程

4.2.4 代码实现

4.3 状态定义模块

4.3.1 目标

4.3.2 需求分析

4.3.3 实现流程

4.3.4 代码实现

4.4 节点基类模块

4.4.1 目标

4.4.2 需求分析

4.4.3 实现流程

4.4.4 代码实现

4.5 主图构建模块

4.5.1 目标

4.5.2 需求分析

4.5.3 实现流程

4.5.4 代码实现

5. 测试入口

6. 总结

6.1 关键设计模式

6.2 查询流程关键特性

6.3 扩展指南

企业痛点映射

Remote & Agent 应用场景价值

Git Commit 对应