MaxKB是一个基于大语言模型和RAG（检索增强生成）的开源知识库问答系统。该项目具有以下主要特点：

1. 技术架构：

• 前端：Vue.js
• 后端：Python/Django
• 大模型框架：LangChain
• 数据库：PostgreSQL + pgvector（用于向量存储）

• 核心功能模块：

• 知识库管理（dataset应用）：负责文档管理、导入和处理
• 向量嵌入（embedding应用）：处理文本向量化和检索
• 函数库与工作流（function_lib应用）：提供AI工作流编排功能
• 应用系统（application应用）：管理聊天和应用流程
• 用户管理（users应用）：处理用户认证和权限

• 主要特点：

• 支持多种大语言模型，包括本地私有模型和各种公共模型
• 文档自动拆分和向量化
• 灵活的工作流引擎
• 支持无缝嵌入到第三方业务系统

• 部署方式：

• Docker容器化部署（默认端口8080）
• 支持离线安装
• 通过1Panel应用商店快速部署

• 版本划分：

• 社区版（开源）
• 专业版（付费，具有更多高级功能）

项目结构清晰，采用了Django的多应用模块化设计，前后端分离架构，使用了现代化的AI技术栈。

---

问题1：它有一个很特别的功能，就是为导入后的文档的分片，加上相关的问题，从而提高召回率，我想知道涉及到了哪些代码？并且想知道它这个功能具体是怎么是实现的，比如【问题】和【分片】之间构造了怎样的关联关系，在查询阶段又是怎么处理的？是使用【用户问题】与维护好了的【分片】所对应的问题做语义计算，还是怎么招了。。。

我将为您查找相关的代码并分析这个特别的功能实现方式。

首先，让我找一下与文档分片和问题生成相关的代码：

Search files... 让我找一下与问题和分片关联相关的代码：

Search files... 让我看一下查询阶段的代码，特别是如何处理分片问题：

Search files... 让我们找一下问题生成的代码：

Search files... 现在让我看看如何将问题和分片关联起来以及在搜索中使用的部分：

Search files... 让我查看一下用于通过问题与分片进行匹配的SQL语句：

Read file: apps/dataset/task/tools.py 让我查看ProblemParagraphMapping的模型和相关操作：

Search files...

Search files... 根据我对代码的分析，我现在能够为您解释MaxKB中为导入后的文档分片加上相关问题从而提高召回率的功能实现方式：

文档分片与问题生成的实现

1. 基本概念与数据模型

MaxKB使用以下几个关键实体模型来实现这个功能：

1. Paragraph（分片）：文档被拆分成的小段落，每个段落包含内容和标题。
2. Problem（问题）：为段落生成的相关问题，这些问题更贴近用户可能的提问方式。
3. ProblemParagraphMapping：问题和段落之间的关联关系，建立了多对多的映射。
4. Embedding：向量化存储，包含了问题和段落的embedding向量，用于相似度搜索。

2. 问题生成过程

当文档被导入系统后，MaxKB会执行以下流程来为分片生成问题：

1. 首先在dataset/task/generate.py中，通过generate_related_by_document_id函数启动分片问题生成任务。
2. 系统会使用LLM模型（如OpenAI、DeepSeek等）来为每个分片生成相关问题。
3. 生成问题的提示词模板定义在ui/src/locales/lang/zh-CN/views/document.ts中，基本格式是：

   内容：{data}
   请总结上面的内容，并根据内容总结生成5个问题。
   回答要求：
   - 请只输出问题；
   - 请将每个问题放置<question>标签中。
   

4. 生成的问题通过save_problem函数在dataset/task/tools.py中被保存，它会使用正则表达式提取标签中的问题内容，并创建Problem实体。
5. 同时创建ProblemParagraphMapping对象，建立问题和段落之间的关联关系。

3. 向量化处理

为提高检索效率，系统会对问题和段落都进行向量化处理：

1. 在common/event/listener_manage.py中，embedding_by_problem和embedding_by_paragraph函数分别处理问题和段落的向量化。
2. 向量化的数据被存储在Embedding表中，包括了来源ID（source_id）、来源类型（source_type，区分问题和段落）、段落ID、文档ID等信息。
3. 向量化数据通过embedding.sql中定义的SQL语句写入PostgreSQL的pgvector扩展中。

4. 查询处理机制

当用户提问时，MaxKB通过以下方式利用预生成的问题提高召回率：

1. 在apps/application/flow/step_node/search_dataset_node/impl/base_search_dataset_node.py中的execute方法处理查询请求。

2. 系统支持三种搜索模式（在embedding/models/embedding.py中定义的SearchMode）：

• embedding：纯向量相似度搜索（通过embedding_search.sql实现）
• keywords：基于关键词的搜索（通过keywords_search.sql实现）
• blend：混合搜索模式，同时利用向量相似度和关键词（通过blend_search.sql实现）

• 关键机制：当用户提问时，系统会：

• 将用户问题向量化
• 在Embedding表中搜索相似向量，这里的搜索包括对问题向量和段落向量的搜索
• 通过SQL查询（如embedding_search.sql）计算相似度分数
• 根据相似度排序并返回最相关的段落

这种机制的精妙之处在于：用户的问题不仅会与段落内容进行匹配，还会与预先生成的问题进行匹配。当用户问题与某个预生成问题相似时，系统会通过ProblemParagraphMapping关联找到对应的段落。

5. 为什么这能提高召回率？

这种设计之所以能提高召回率，主要基于以下原因：

1. 问题表达和内容表达的差异桥接：用户提问的方式（如"如何解决xxx问题？"）通常与内容的表达方式（如陈述句"xxx可以通过yyy解决"）有差异。预生成的问题作为"桥梁"，使得即使用户的提问方式与内容表达方式不同，也能找到相关段落。

2. 语义扩展：为每个段落生成多个相关问题，本质上是对段落内容的语义扩展，增加了匹配的可能性。

3. 匹配路径多样化：查询时，系统提供了多种搜索路径：

• 用户问题 → 段落内容（直接匹配）
• 用户问题 → 预生成问题 → 段落内容（间接匹配）

通过SQL查询（如embedding_search.sql）可以看出，系统会对同一段落的不同匹配结果（如直接匹配段落或匹配预生成问题）取其中相似度最高的一个，确保结果不会重复。

总结来说，MaxKB通过为分片生成相关问题并建立问题-分片映射关系，在查询时同时搜索问题和分片的向量表示，有效地解决了用户提问方式与内容表达方式之间的差异，从而显著提高了知识库的召回率。这是一种结合了大模型问题生成能力和向量检索技术的巧妙解决方案。

---

问题2：你刚才说了，系统会对同一段落的不同匹配结果（如直接匹配段落或匹配预生成问题）取其中相似度最高的一个，确保结果不会重复。OK，那请把对应的核心逻辑的代码直接摘抄出来，我看一下它怎么细节上是怎么实现的

我来为您找出实现这个核心逻辑的代码。这部分逻辑主要在SQL查询和查询结果处理中实现：

Read file: apps/embedding/sql/embedding_search.sql 这里的SQL查询是核心逻辑的关键部分。让我来分析一下这个embedding_search.sql文件的关键部分：

1. 内部查询部分:

   ( SELECT *, ( 1 - ( embedding.embedding <=>  %s ) ) AS similarity FROM embedding ${embedding_query}) TEMP
   

   这部分计算用户问题向量与embedding表中的所有向量（包括问题和段落）的相似度，使用<=> 是PostgreSQL的pgvector扩展提供的向量相似度计算算符。

2. DISTINCT ON子句部分:

   SELECT DISTINCT ON ("paragraph_id") ( similarity ),* ,similarity AS comprehensive_score
   FROM ... TEMP
   ORDER BY paragraph_id, similarity DESC
   

   这是关键部分，DISTINCT ON ("paragraph_id") 确保每个段落(paragraph_id)只会出现一次。ORDER BY paragraph_id, similarity DESC 确保对于每个段落，取相似度最高的那个匹配结果。

通过这个SQL查询，系统确保了每个段落只出现一次，且是相似度最高的那个匹配结果，无论这个匹配是直接通过段落内容匹配的，还是通过该段落关联的预生成问题匹配的。

让我们看一下这个SQL是如何在代码中被调用的：

Read file: apps/embedding/vector/pg_vector.py 现在让我们看下搜索查询方法的实现：

Read file: apps/embedding/vector/pg_vector.py 让我们看看数据如何被存储和准备的，特别是问题和段落之间的关联：

Read file: apps/common/sql/list_embedding_text.sql 让我们看一下搜索结果处理部分，这是实现从向量库检索结果到最终判断的核心环节：

Read file: apps/application/flow/step_node/search_dataset_node/impl/base_search_dataset_node.py 让我来看看一下段落匹配和相似度处理的代码：

Read file: apps/application/flow/step_node/search_dataset_node/impl/base_search_dataset_node.py 根据我对代码的分析，现在我可以更详细地解释MaxKB如何实现在查询阶段对同一段落的不同匹配结果取最高相似度的核心逻辑：

核心SQL部分实现

最关键的逻辑在apps/embedding/sql/embedding_search.sql中：

SELECT
    paragraph_id,
    comprehensive_score,
    comprehensive_score as similarity
FROM
    (
    SELECT DISTINCT ON
        ("paragraph_id") ( similarity ),* ,similarity AS comprehensive_score
    FROM
        ( SELECT *, ( 1 - ( embedding.embedding <=>  %s ) ) AS similarity FROM embedding ${embedding_query}) TEMP
    ORDER BY
        paragraph_id,
        similarity DESC
    ) DISTINCT_TEMP
WHERE comprehensive_score>%s
ORDER BY comprehensive_score DESC
LIMIT %s

这个SQL查询的核心逻辑是：

1. 最内层子查询：计算用户查询向量与所有向量的相似度

   SELECT *, ( 1 - ( embedding.embedding <=>  %s ) ) AS similarity FROM embedding ${embedding_query}
   

• <=> 是PostgreSQL的pgvector扩展提供的余弦距离计算操作符
• 1 - (向量距离) 转换为相似度分数（1为完全相似，0为完全不相似）

• 中间层查询：对每个段落ID选取最高相似度的记录

  SELECT DISTINCT ON ("paragraph_id") ( similarity ),* ,similarity AS comprehensive_score
  FROM ... TEMP
  ORDER BY paragraph_id, similarity DESC
  

  这就是系统如何处理同一段落可能存在多个匹配来源（直接匹配段落内容或通过关联问题匹配）的关键机制。

• DISTINCT ON ("paragraph_id") 是关键语法，确保每个段落ID只保留一条记录
• ORDER BY paragraph_id, similarity DESC 保证在对段落ID去重时，保留相似度最高的那条记录

• 最外层查询：过滤低相似度结果并排序

  WHERE comprehensive_score>%s
  ORDER BY comprehensive_score DESC
  LIMIT %s
  

• 只保留相似度高于阈值的结果
• 按相似度降序排列
• 限制返回结果数量

段落和问题的向量化与存储

在apps/common/sql/list_embedding_text.sql中可以看到，系统如何准备用于向量化的文本数据：

-- 第一部分：准备问题的文本数据
SELECT
    problem_paragraph_mapping."id" AS "source_id",
    paragraph.document_id AS document_id,
    paragraph."id" AS paragraph_id,
    problem.dataset_id AS dataset_id,
    0 AS source_type,  -- 0表示问题类型
    problem."content" AS "text",
    paragraph.is_active AS is_active
FROM
    problem problem
    LEFT JOIN problem_paragraph_mapping problem_paragraph_mapping ON problem_paragraph_mapping.problem_id=problem."id"
    LEFT JOIN paragraph paragraph ON paragraph."id" = problem_paragraph_mapping.paragraph_id
 ${problem}

UNION

-- 第二部分：准备段落的文本数据
SELECT
    paragraph."id" AS "source_id",
    paragraph.document_id AS document_id,
    paragraph."id" AS paragraph_id,
    paragraph.dataset_id AS dataset_id,
    1 AS source_type,  -- 1表示段落类型
    concat_ws(E'\n',paragraph.title,paragraph."content") AS "text",
    paragraph.is_active AS is_active
FROM
    paragraph paragraph
 ${paragraph}

这个SQL查询通过UNION合并了两种数据：

1. 问题数据（source_type=0）：从问题表和问题-段落映射表获取
2. 段落数据（source_type=1）：直接从段落表获取

重要的是，在这个合并查询中，每个问题记录也被关联到了相应的段落ID（paragraph_id），这是实现通过问题匹配找到相关段落的关键。

搜索结果处理

在apps/application/flow/step_node/search_dataset_node/impl/base_search_dataset_node.py中的execute方法处理了搜索过程和结果：

def execute(self, dataset_id_list, dataset_setting, question,
            exclude_paragraph_id_list=None,
            **kwargs) -> NodeResult:
    # ...
    embedding_value = embedding_model.embed_query(question) 
    # ...
    embedding_list = vector.query(question, embedding_value, dataset_id_list, exclude_document_id_list,
                                  exclude_paragraph_id_list, True, dataset_setting.get('top_n'),
                                  dataset_setting.get('similarity'), SearchMode(dataset_setting.get('search_mode')))
    # ...
    paragraph_list = self.list_paragraph(embedding_list, vector)
    result = [self.reset_paragraph(paragraph, embedding_list) for paragraph in paragraph_list]
    result = sorted(result, key=lambda p: p.get('similarity'), reverse=True)
    # ...

这里：

1. 先将用户问题转换为向量表示
2. 通过vector.query方法执行向量相似度搜索
3. 通过list_paragraph函数获取匹配的段落详细信息
4. 使用reset_paragraph重新格式化段落信息，添加相似度分数
5. 最后按相似度降序排序结果

特别是，reset_paragraph方法将搜索结果中的相似度分数附加到段落对象上：

@staticmethod
def reset_paragraph(paragraph: Dict, embedding_list: List):
    filter_embedding_list = [embedding for embedding in embedding_list if
                             str(embedding.get('paragraph_id')) == str(paragraph.get('id'))]
    if filter_embedding_list is not None and len(filter_embedding_list) > 0:
        find_embedding = filter_embedding_list[-1]
        return {
            **paragraph,
            'similarity': find_embedding.get('similarity'),
            # ...其他属性
        }

总结来说，MaxKB通过PostgreSQL的DISTINCT ON语法结合精心设计的SQL查询，巧妙地实现了在搜索阶段对同一段落的不同匹配结果（可能来自段落本身或与段落关联的预生成问题）取相似度最高的一个。这样既避免了结果重复，又保证了能选择最佳匹配路径（直接匹配段落或间接通过问题匹配段落）。