文章

RAG知识库多跳查询方案

发表于
Preview Image
作者 handsomestwei
40 分钟阅读
RAG知识库多跳查询方案

RAG知识库多跳查询方案

1. 方案解决的问题

在传统的RAG（检索增强生成）知识库问答场景中，当用户提问涉及文档附件内容时，存在以下问题：

  • 附件内容检索不精准：附件文档通常包含多个章节，在知识库中作为单个文档存储时，检索容易带出无关章节内容，影响答案准确性
  • 附件引用信息缺失：大模型回答中虽然可能提到”详见附件X”，但无法直接展示附件具体内容，用户体验不完整
  • 多跳查询效率低：如果需要通过引用关系进行二次检索，传统方式需要再次调用大模型，消耗大量token和计算资源
  • 知识库检索范围过大：在包含大量文档的知识库中检索附件，容易产生噪音，影响检索精度

本方案通过多跳查询和知识库治理相结合的方式，在传统知识库问答工作流之后，补充一个轻量级的附件内容检索流程，实现精准、高效、低成本的附件内容补充输出。

2. 主要思路

本方案采用两阶段查询的设计思路：

  1. 第一阶段：传统的知识库问答流程
    • 用户提问 → 知识库检索 → 大模型生成回答
    • 在提示词中强制要求大模型输出格式化的”相关附件”章节，包含附件引用信息
  2. 第二阶段：附件内容补充查询流程（本方案核心）
    • 从大模型回答中提取附件引用信息
    • 在专门的附件文档知识库集合中进行精准检索
    • 通过API批量获取附件完整内容
    • 格式化输出附件内容，补充到最终回答中

关键设计点：

  • 附件文档独立治理，每个附件章节拆分为独立文档，提升检索精准度
  • 使用关键词匹配而非大模型推理，降低token消耗
  • 通过代码节点和批量处理节点组合，实现高效的多跳查询流程

3. 一阶段对话提示词要求

在传统知识库问答的提示词中，需要强制要求大模型在回答末尾输出格式化的”相关附件”章节。

输出格式要求

大模型必须在回答中包含以下格式的章节：

1
2
3
4
5

**相关附件**

附件11-上级、地方及政府特服应急电话
附件12-公司组织机构及内部应急联系电话
附件13-外部应急救援资源公共机构及联系电话

关键要求

  1. 章节标题：必须使用 **相关附件** 作为章节标题
  2. 附件格式：每行一个附件，格式为 附件XX-附件名称
    • 附件XX：附件编号，必须是数字
    • 附件名称：完整的附件名称，用于后续匹配
  3. 文档名-章节名规范：
    • 附件名称应遵循”文档名-章节名”的命名规范
    • 文档名用于定位和区分关联的附件
    • 例如：附件16-社会主要应急物资和装备清单，其中”社会主要应急物资和装备清单”是文档名

提示词示例片段

1
2
3
4
5
6
7
8
9
10
11

在回答末尾，必须输出"相关附件"章节，格式如下：

**相关附件**

附件X-附件名称
附件Y-附件名称

注意：
- 如果回答中提到了"详见附件X"或"参考附件X"，必须在"相关附件"章节中列出
- 附件名称必须完整准确，用于后续检索匹配
- 附件编号必须与文档中的实际编号一致

4. 附件文档知识治理

治理原则

核心原则：将单文档的每个章节附件，拆分成独立的多文档。

具体操作

  1. 文档拆分：
    • 不要将所有附件章节都放到一个文档里
    • 每个附件章节应作为独立的文档上传到知识库
    • 例如：原文档”应急预案.docx”包含”附件11”、”附件12”等多个附件，应拆分为：
      • 附件11-上级、地方及政府特服应急电话.docx
      • 附件12-公司组织机构及内部应急联系电话.docx
  2. 独立知识库集合：
    • 创建专门的”附件文档”知识库集合
    • 将所有拆分后的附件文档上传到此集合
    • 在二阶段查询时，指定在此集合中检索，减少检索范围
  3. 文件命名规范：
    • 文件名称格式：附件XX-附件名称.docx
    • 与提示词中要求的输出格式保持一致
    • 便于后续通过文件名进行匹配

治理效果

  • 精准检索：每个附件作为独立文档，检索时不会带出其他附件内容
  • 减少噪音：独立的附件文档知识库集合，避免与主文档库混合检索
  • 提升效率：缩小检索范围，提升检索速度和准确度

5. 二阶段补充查询流程

流程图

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

大模型回答输出
    ↓
[代码节点1] 附件文件名提取
    ↓
附件名称数组: ["附件11-...", "附件12-..."]
    ↓
[批量执行节点] 循环知识库检索
    ├─ 循环项1: 检索"附件11-..."
    ├─ 循环项2: 检索"附件12-..."
    └─ ...
    ↓
知识库引用数组（二维数组）
    ↓
[代码节点2] 附件文档Id提取
    ↓
collection_ids: [["id1"], ["id2"], ...]
collection_attachment_map: {"id1": "附件11-...", ...}
    ↓
[批量执行节点] 循环HTTP请求
    ├─ 循环项1: 请求collectionId="id1"的数据块
    ├─ 循环项2: 请求collectionId="id2"的数据块
    └─ ...
    ↓
HTTP响应数组
    ↓
[代码节点3] 附件内容合并和格式化
    ↓
格式化后的附件内容字符串
    ↓
[指定输出节点] 输出最终内容

技术细节

（1）代码节点：附件文件名提取

功能：从大模型输出对话里，提取约定的”相关附件”章节内容。

输入变量：

  • answer_content（类型：string）- 大模型的回答内容

输出变量：

  • attachment_info（类型：array）- 附件信息数组，格式为 `["附件11-上级、地方及政府特服应急电话", "附件12-公司组织机构及内部应急联系电话"]`

实现逻辑：

  • 使用正则表达式匹配 **相关附件** 章节
  • 提取每行的 附件XX-附件名称 格式内容
  • 自动去重，保持顺序
查看完整代码 ```python """ FastGPT智能体平台代码节点：提取"相关附件"章节中的附件信息 使用说明： 1. 在FastGPT中创建代码节点 2. 输入变量名：answer_content（类型：string） 3. 输出变量名：attachment_info（类型：array） 功能： - 从回答内容的"相关附件"章节中提取附件信息 - 格式：附件XX-附件名称 - 自动去重 - 返回字符串数组 """ import re from typing import List # FastGPT代码节点入口函数 # 注意：FastGPT会自动将输入变量作为函数参数传入 def main(answer_content: str = "") -> dict: """ FastGPT代码节点主函数 输入参数（FastGPT会自动传入）： answer_content: str - 回答内容文本 返回值： dict - 包含以下字段： attachment_info: List[str] - 附件信息数组（去重后），格式为"附件XX-附件名称" """ # 输入验证 if not answer_content: return { "attachment_info": [] } if not isinstance(answer_content, str): # 如果不是字符串，尝试转换 answer_content = str(answer_content) try: # 首先提取"相关附件"章节的内容 # 匹配"相关附件"章节，从"**相关附件**"开始到下一个"**"标题或文件结束 attachment_section_pattern = r'\*\*相关附件\*\*[：:]?\s*\n(.*?)(?=\n\*\*|$)' attachment_section_match = re.search(attachment_section_pattern, answer_content, re.DOTALL | re.IGNORECASE) if not attachment_section_match: # 如果没有找到"相关附件"章节，返回空数组 return { "attachment_info": [] } attachment_section = attachment_section_match.group(1) # 匹配格式：附件XX-附件名称（可能后面有引用编号，但不提取） # 正则表达式说明： # ^\s* - 行首可能有空格 # 附件(\d+) - 匹配"附件"后跟数字 # \s*-\s* - 匹配分隔符"-"（前后可能有空格） # (.+?) - 匹配附件名称（所有字符，非贪婪，直到遇到 -[ 或行尾） # 可选：后面可能有 -[引用编号](CITE)，但不捕获 # 使用前瞻断言，匹配到 -[ 之前或行尾 pattern = r'^\s*附件(\d+)\s*-\s*(.+?)(?=\s*-\s*\[|$)' # 按行处理 lines = attachment_section.split('\n') attachment_info_list = [] for line in lines: line = line.strip() if not line: continue match = re.search(pattern, line, re.IGNORECASE) if match: attachment_num = match.group(1) attachment_name = match.group(2).strip() # 只输出附件编号和名称，不包含引用编号 attachment_info = f"附件{attachment_num}-{attachment_name}" attachment_info_list.append(attachment_info) # 去重并保持顺序 unique_info = list(dict.fromkeys(attachment_info_list)) # 返回结果 # FastGPT会自动将返回值中的字段映射到输出变量 return { "attachment_info": unique_info } except Exception as e: # 错误处理：返回空数组 return { "attachment_info": [] } ``` </details> #### （2）批量执行节点：循环知识库检索 **功能**：循环执行知识库检索节点，获取对应的文件引用。 **配置要点**： 1. **循环变量**： - 使用 `attachment_info` 数组作为循环源 - 每次循环处理一个附件名称 2. **知识库检索节点配置**： - **指定知识库集合**：选择"附件文档"知识库集合（治理后的独立集合） - **查询方式**：关键字检索 - **查询内容**：使用变量引用，引用循环中的附件名称（如 ``） - **检索策略**：可以先使用"文档名-章节名"方式测试命中效果 3. **输出**： - 输出知识库引用数组（二维数组） - 每个元素对应一个附件的检索结果 - 检索结果中包含治理后的真实附件文档Id（`collectionId`） **注意事项**： - 用户问题字段应使用变量引用数组变量，而不是原始用户问题 - 确保检索的知识库集合是专门的附件文档集合，避免在主文档库中检索 #### （3）代码节点：附件文档Id提取 **功能**：从知识库引用数组中提取 `collectionId`，构建文档集合数组。 **输入变量**： - `knowledge_base_refs`（类型：array<array>）- 知识库引用内容（循环节点多次检索结果的合集） - `attachment_list`（类型：array）- 附件名称数组列表 **输出变量**： - `collection_ids`（类型：array<array>）- collectionId数组的数组 - `collection_attachment_map`（类型：object）- collectionId和附件文件名称的映射关系 **实现逻辑**： - 扁平化知识库引用数组（二维转一维） - 根据附件名称匹配知识库引用中的文档 - 支持两种匹配模式：全相等匹配（`exact`）或包含匹配（`contains`，默认） - 提取匹配文档的 `collectionId`，构建映射关系
查看完整代码 ```python """ FastGPT智能体平台代码节点：根据附件名称匹配知识库引用内容并提取collectionId数组 使用说明： 1. 在FastGPT中创建代码节点 2. 输入变量名1：knowledge_base_refs（类型：array<array>）- 知识库引用内容（循环节点多次检索结果的合集） 3. 输入变量名2：attachment_list（类型：array）- 附件名称数组列表 4. 输出变量名1：collection_ids（类型：array<array>）- collectionId数组的数组，每个元素是一个附件的collectionIds列表 5. 输出变量名2：collection_attachment_map（类型：object）- collectionId和附件文件名称的映射关系 功能： - 根据附件名称在知识库引用内容中匹配对应的附件 - 支持两种匹配模式：全相等匹配或包含匹配（默认） - 从匹配的分块中提取collectionId，构建文档集合数组（去重） - 输出collectionId数组的数组，每个元素对应一个附件的collectionIds列表 输出格式： { "collection_ids": [ ["69548f1002a61be1b5011994"], ... ], "collection_attachment_map": { "69548f1002a61be1b5011994": "附件16-社会主要应急物资和装备清单", ... } } """ import re import json from typing import List, Dict, Any, Union # 匹配模式配置 # 可选值："exact"（全相等匹配）或 "contains"（包含匹配，默认） MATCH_MODE = "contains" # 默认使用包含匹配模式 # FastGPT代码节点入口函数 # 注意：FastGPT会自动将输入变量作为函数参数传入 def main(knowledge_base_refs: Union[List, str] = None, attachment_list: List[str] = None) -> dict: """ FastGPT代码节点主函数 输入参数（FastGPT会自动传入）： knowledge_base_refs: Union[List, str] - 知识库引用内容（二维数组或JSON字符串） attachment_list: List[str] - 附件名称数组列表，格式如：["附件11-上级、地方及政府特服应急电话", "附件12-公司组织机构及内部应急联系电话"] 匹配模式： 通过代码中的 MATCH_MODE 常量配置，可选值："exact"（全相等匹配）或 "contains"（包含匹配，默认） 返回值： dict - 包含以下字段： collection_ids: List[List[str]] - collectionId数组的数组，每个元素是一个附件的collectionIds列表 collection_attachment_map: Dict[str, str] - collectionId和附件文件名称的映射关系 """ # 输入验证 if not attachment_list or not isinstance(attachment_list, list) or len(attachment_list) == 0: return { "collection_ids": [], "collection_attachment_map": {} } # 获取匹配模式（从代码常量中读取） match_mode = MATCH_MODE if match_mode not in ["exact", "contains"]: match_mode = "contains" # 默认使用包含匹配 # 处理知识库引用内容 if not knowledge_base_refs: return { "collection_ids": [], "collection_attachment_map": {} } # 如果knowledge_base_refs是字符串，尝试解析为JSON if isinstance(knowledge_base_refs, str): try: knowledge_base_refs = json.loads(knowledge_base_refs) except: return { "collection_ids": [], "collection_attachment_map": {} } # 验证knowledge_base_refs是否为二维数组 if not isinstance(knowledge_base_refs, list): return { "collection_ids": [], "collection_attachment_map": {} } try: # 扁平化知识库引用内容（将二维数组转换为一维数组） flattened_refs = [] for ref_group in knowledge_base_refs: if isinstance(ref_group, list): flattened_refs.extend(ref_group) elif isinstance(ref_group, dict): flattened_refs.append(ref_group) if not flattened_refs: return { "collection_ids": [], "collection_attachment_map": {} } # 用于存储每个附件对应的collectionId数组 collection_ids_array = [] # 用于存储collectionId和附件文件名称的映射关系 collection_attachment_map = {} # 遍历附件列表，在知识库引用内容中匹配 for attachment_name in attachment_list: if not attachment_name or not isinstance(attachment_name, str): continue # 提取附件编号和名称 # 格式：附件XX-附件名称 attachment_match = re.match(r'^附件(\d+)\s*-\s*(.+)$', attachment_name.strip()) if not attachment_match: continue attachment_num = attachment_match.group(1) attachment_name_part = attachment_match.group(2).strip() attachment_identifier = f"附件{attachment_num}-{attachment_name_part}" # 收集所有匹配的附件分块（同一个附件可能有多个chunkIndex） matched_chunks = [] # 在知识库引用内容中查找匹配的附件 for ref in flattened_refs: if not isinstance(ref, dict): continue source_name = ref.get("sourceName", "") if not source_name: continue # 去掉文件扩展名（如.docx）后匹配 source_name_without_ext = re.sub(r'\.\w+$', '', source_name) # 从sourceName中提取附件编号和名称部分 source_match = re.match(r'^附件(\d+)\s*-\s*(.+)$', source_name_without_ext) if source_match: # sourceName格式符合"附件XX-名称"格式 source_num = source_match.group(1) source_name_part = source_match.group(2).strip() # 首先检查附件编号是否匹配 if attachment_num != source_num: continue # 根据匹配模式进行名称匹配 if match_mode == "exact": # 全相等匹配：附件名称部分必须完全相等 is_match = (attachment_name_part == source_name_part) else: # 包含匹配（默认）：附件名称部分相互包含即可 is_match = (attachment_name_part in source_name_part or source_name_part in attachment_name_part) else: # 如果sourceName格式不符合"附件XX-名称"格式，使用完整标识匹配 if match_mode == "exact": # 全相等匹配：检查完整标识是否相等 is_match = (attachment_identifier == source_name_without_ext) else: # 包含匹配：检查完整标识是否相互包含 is_match = (attachment_identifier in source_name_without_ext or source_name_without_ext in attachment_identifier) # 如果匹配成功，收集该分块 if is_match: matched_chunks.append(ref) # 如果找到匹配的分块，提取collectionId和doc_id if matched_chunks: # 从匹配的分块中提取collectionId，构建文档集合数组（去重） collection_ids = set() doc_id = None for ref in matched_chunks: collection_id = ref.get("collectionId", "") if collection_id: collection_ids.add(collection_id) # 获取文档块ID（使用第一个分块的ID） if doc_id is None: doc_id = ref.get("id", "") # 添加到数组 if collection_ids: collection_ids_array.append(list(collection_ids)) # 记录collectionId和附件名称的映射关系 for coll_id in collection_ids: if coll_id not in collection_attachment_map: collection_attachment_map[coll_id] = attachment_identifier return { "collection_ids": collection_ids_array, "collection_attachment_map": collection_attachment_map } except Exception as e: # 错误处理：返回空数组 return { "collection_ids": [], "collection_attachment_map": {} } ``` </details> #### （4）批量执行节点：循环HTTP请求 **功能**：内嵌HTTP请求节点，调用智能体平台的API，使用文档Id（`collectionId`）批量查询附件内容。 **配置要点**： 1. **循环变量**： - 使用 `collection_ids` 数组作为循环源 - 每次循环处理一个附件的 `collectionId` 列表（可能包含多个 `collectionId`） 2. **HTTP请求节点配置**： - **API地址**：FastGPT智能体平台的数据块查询API - **请求方法**：POST 或 GET（根据API文档） - **请求参数**：使用 `collectionId` 作为查询参数 - **认证信息**：配置API的Authorization头 3. **输出格式**： - 每个HTTP响应包含一个文档的所有数据块列表 - 响应格式示例： ```json { "code": 200, "statusText": "", "message": "", "data": { "list": [ { "_id": "69548f4002a61be1b5011c5d", "collectionId": "69548f1002a61be1b5011994", "q": "附件内容...", "chunkIndex": 0 }, ... ] } } ``` **注意事项**： - **不要在代码节点里发起HTTP请求**，容易被sandbox沙盒环境拦截 - 使用FastGPT工作流中的HTTP请求节点，避免环境限制 - 确保API调用有适当的错误处理和重试机制 #### （5）代码节点：附件内容合并和格式化 **功能**：解析HTTP响应数组，合并附件内容，格式化输出。 **输入变量**： - `collection_attachment_map`（类型：object）- collectionId和附件文件名称的映射关系 - `http_response_array`（类型：array）- HTTP调用节点的批量处理结果数组 **输出变量**： - `formatted_content`（类型：string）- 格式化后的附件内容字符串 **实现逻辑**： - 从HTTP响应数组中提取数据块列表 - 按 `collectionId` 组织数据块 - 对于每个附件，收集所有相关的数据块 - 按 `chunkIndex` 排序，拼接所有 `q` 字段内容 - 格式化输出，包含附件标题和引用信息 **输出格式**： ```markdown **相关附件内容：** **附件11：上级、地方及政府特服应急电话**[文档块ID](CITE) 附件内容... **附件12：公司组织机构及内部应急联系电话**[文档块ID](CITE) 附件内容... ```
查看完整代码 ```python """ FastGPT智能体平台代码节点：解析HTTP响应数组并合并附件内容 使用说明： 1. 在FastGPT中创建代码节点 2. 输入变量名1：collection_attachment_map（类型：object）- collectionId和附件文件名称的映射关系（来自文件1的输出） 格式如： { "69548f1002a61be1b5011994": "附件16-社会主要应急物资和装备清单", ... } 3. 输入变量名2：http_response_array（类型：array）- HTTP调用节点的批量处理结果数组 每个元素是一个HTTP响应对象，格式如： [ { "code": 200, "data": { "list": [ {"_id": "...", "collectionId": "...", "q": "...", "chunkIndex": 0}, ... ] } }, ... ] 5. 输出变量名：formatted_content（类型：string）- 格式化后的附件内容字符串 功能： - 从HTTP响应数组中提取数据块列表 - 根据collectionId匹配附件 - 按chunkIndex排序，拼接所有q字段内容 - 按照指定格式输出附件内容，包含引用信息 """ import re import json from typing import List, Dict, Any, Union # 匹配模式配置 # 可选值："exact"（全相等匹配）或 "contains"（包含匹配，默认） MATCH_MODE = "contains" # 默认使用包含匹配模式 # FastGPT代码节点入口函数 # 注意：FastGPT会自动将输入变量作为函数参数传入 def main(collection_attachment_map: Union[Dict, str] = None, http_response_array: Union[List, str] = None) -> dict: """ FastGPT代码节点主函数 输入参数（FastGPT会自动传入）： collection_attachment_map: Union[Dict, str] - collectionId和附件文件名称的映射关系（字典或JSON字符串） http_response_array: Union[List, str] - HTTP响应数组，每个元素是一个HTTP响应对象 返回值： dict - 包含以下字段： formatted_content: str - 格式化后的附件内容字符串 """ # 输入验证 if not collection_attachment_map: return { "formatted_content": "\n**相关附件内容：**\n\n输入的collection_attachment_map为空或无效，未进行检索。" } # 如果collection_attachment_map是字符串，尝试解析为JSON if isinstance(collection_attachment_map, str): try: collection_attachment_map = json.loads(collection_attachment_map) except: return { "formatted_content": "\n**相关附件内容：**\n\ncollection_attachment_map格式错误，未进行检索。" } # 验证collection_attachment_map是否为字典 if not isinstance(collection_attachment_map, dict): return { "formatted_content": "\n**相关附件内容：**\n\ncollection_attachment_map格式错误，未进行检索。" } if len(collection_attachment_map) == 0: return { "formatted_content": "\n**相关附件内容：**\n\ncollection_attachment_map为空，未进行检索。" } # 处理http_response_array if not http_response_array: return { "formatted_content": "\n**相关附件内容：**\n\nHTTP响应数组为空，未检索到附件内容。" } # 如果http_response_array是字符串，尝试解析为JSON if isinstance(http_response_array, str): try: http_response_array = json.loads(http_response_array) except: return { "formatted_content": "\n**相关附件内容：**\n\nHTTP响应数组格式错误，未进行检索。" } # 验证http_response_array是否为数组 if not isinstance(http_response_array, list): return { "formatted_content": "\n**相关附件内容：**\n\nHTTP响应数组格式错误，未进行检索。" } try: # 从collection_attachment_map中提取附件列表（去重） attachment_set = set(collection_attachment_map.values()) attachment_list = sorted(list(attachment_set)) # 排序以保证输出顺序一致 # 从HTTP响应数组中提取数据块，按collectionId组织 # collection_id -> [chunk1, chunk2, ...] collection_chunks_map = {} for response in http_response_array: if not isinstance(response, dict): continue # 检查响应是否成功 if response.get("code") != 200: continue # 提取data.list data = response.get("data", {}) chunks_list = data.get("list", []) if not isinstance(chunks_list, list): continue # 按collectionId分组 for chunk in chunks_list: if not isinstance(chunk, dict): continue collection_id = chunk.get("collectionId", "") if collection_id: if collection_id not in collection_chunks_map: collection_chunks_map[collection_id] = [] collection_chunks_map[collection_id].append(chunk) # 用于存储匹配到的附件内容 matched_attachments = {} # 遍历附件列表，匹配并合并内容 for attachment_name in attachment_list: if not attachment_name or not isinstance(attachment_name, str): continue # 提取附件编号和名称 attachment_match = re.match(r'^附件(\d+)\s*-\s*(.+)$', attachment_name.strip()) if not attachment_match: continue attachment_num = attachment_match.group(1) attachment_name_part = attachment_match.group(2).strip() attachment_identifier = f"附件{attachment_num}-{attachment_name_part}" # 从collection_attachment_map中查找该附件对应的collectionId # 可能有多个collectionId对应同一个附件 collection_ids = [coll_id for coll_id, att_name in collection_attachment_map.items() if att_name == attachment_identifier] if not collection_ids: continue # 收集所有数据块 all_chunks = [] doc_id = None # 遍历该附件的所有collectionId，从collection_chunks_map中获取数据块 for collection_id in collection_ids: if collection_id in collection_chunks_map: chunks = collection_chunks_map[collection_id] for chunk in chunks: if isinstance(chunk, dict): q_content = chunk.get("q", "") chunk_index = chunk.get("chunkIndex", 0) if q_content: all_chunks.append({ "q": q_content, "chunkIndex": chunk_index if isinstance(chunk_index, int) else 0 }) # 获取文档块ID（使用第一个数据块的_id） if doc_id is None: doc_id = chunk.get("_id", "") # 按chunkIndex排序 all_chunks.sort(key=lambda x: x.get("chunkIndex", 0) if isinstance(x.get("chunkIndex"), int) else 0) # 拼接所有q字段的内容 q_contents = [] for chunk in all_chunks: q_content = chunk.get("q", "") if q_content and q_content.strip(): q_contents.append(q_content.strip()) # 生成最终合并内容（使用双换行分隔） final_content = "\n\n".join(q_contents) if q_contents else "" # 如果找到匹配的附件，保存 if doc_id and final_content: matched_attachments[attachment_identifier] = { "id": doc_id, "content": final_content, "num": attachment_num, "name": attachment_name_part } # 按照输入顺序生成输出内容 output_lines = ["", "**相关附件内容：**", ""] for attachment_name in attachment_list: if not attachment_name or not isinstance(attachment_name, str): continue # 提取附件编号 attachment_match = re.match(r'^附件(\d+)\s*-\s*(.+)$', attachment_name.strip()) if not attachment_match: continue attachment_num = attachment_match.group(1) attachment_name_part = attachment_match.group(2).strip() attachment_identifier = f"附件{attachment_num}-{attachment_name_part}" # 查找匹配的附件 if attachment_identifier in matched_attachments: attachment_data = matched_attachments[attachment_identifier] # 格式化输出：**附件X：附件名称**[文档块ID](CITE) output_lines.append(f"**附件{attachment_data['num']}：{attachment_data['name']}**[{attachment_data['id']}](CITE)") output_lines.append(attachment_data['content']) output_lines.append("") # 空行分隔 else: # 未匹配到 output_lines.append(f"附件{attachment_num}内容未检索到确证") output_lines.append("") # 如果没有任何匹配结果 if not matched_attachments: output_lines = ["", "**相关附件内容：**", ""] for attachment_name in attachment_list: if attachment_name and isinstance(attachment_name, str): attachment_match = re.match(r'^附件(\d+)\s*-\s*(.+)$', attachment_name.strip()) if attachment_match: attachment_num = attachment_match.group(1) output_lines.append(f"附件{attachment_num}内容未检索到确证") output_lines.append("") # 生成最终输出字符串 formatted_content = "\n".join(output_lines).strip() # 如果输出为空，返回默认消息 if not formatted_content or formatted_content == "**相关附件内容：**": formatted_content = "\n**相关附件内容：**\n\n知识库引用内容中未匹配到任何附件。" return { "formatted_content": formatted_content } except Exception as e: # 错误处理：返回错误信息 return { "formatted_content": f"\n**相关附件内容：**\n\n处理过程中发生错误：{str(e)}" } ``` </details> #### （6）指定输出节点 **功能**：直接输出编辑后的附件内容。 **配置要点**： - 使用代码节点3输出的 `formatted_content` 作为输出内容 - 可以与大模型的第一阶段回答内容合并，形成完整的最终回答 - 或者作为独立的"相关附件内容"章节输出 ## 6. 方案优点 1. **流程快速**：二次查询不需要经过大模型对话总结输出，直接通过代码节点和API调用完成，响应速度快 2. **减少token消耗**：附件内容检索和合并过程不调用大模型，仅在代码节点中进行字符串处理和格式化，大幅减少token消耗 3. **减少资源性能占用**： - 不需要额外的模型推理计算 - 代码节点执行效率高，资源占用低 - HTTP API调用相比模型推理，资源消耗更小 4. **精准检索**： - 通过附件文档知识库治理，每个附件作为独立文档，检索精准度高 - 独立的附件文档知识库集合，避免与主文档库混合，减少噪音 5. **可扩展性强**： - 工作流节点化设计，易于调整和优化 - 代码节点逻辑清晰，便于维护和升级 ## 7. 方案缺点 1. **关键词匹配局限性**： - 使用关键词匹配方式，容易误判和缺漏 - 如果附件名称提取不准确，或知识库中文件名格式不一致，可能导致匹配失败 - 对于语义相似但名称不同的附件，可能无法正确匹配 2. **事后补充的体验问题**： - 输出内容的流程只能在大模型对话之后做补充输出，属于事后补充 - 用户需要等待两个阶段的处理完成，才能看到完整的回答 - 如果第一阶段回答中没有正确输出"相关附件"章节，第二阶段无法触发 3. **依赖格式规范**： - 方案高度依赖大模型输出格式的规范性 - 如果提示词要求不够严格，或大模型输出格式不符合预期，会导致后续流程失败 4. **知识库治理成本**： - 需要将附件文档拆分为独立文档，增加了知识库治理的工作量 - 需要维护独立的附件文档知识库集合，增加了管理复杂度 ## 8. 后续优化方向 ### 知识图谱增强 后续可以外挂知识图谱，在对话之前就先查询出图谱里关联的附件文档节点，获取附件内容。这样可以： - **提前获取附件信息**：在用户提问阶段，通过知识图谱预判可能需要的附件 - **提升匹配准确度**：通过图谱关系，而非关键词匹配，提升附件关联的准确性 - **优化用户体验**：附件内容可以在第一阶段就整合到回答中，而非事后补充 ### 其他优化方向 1. **语义匹配优化**：使用向量检索或语义匹配，替代关键词匹配，提升匹配准确度 2. **缓存机制**：对常用的附件内容进行缓存，减少重复查询 3. **异步处理**：将附件内容查询改为异步处理，提升用户体验 4. **错误恢复**：增加更完善的错误处理和重试机制，提升系统稳定性
AI, 智能体
AI 智能体 RAG
本文由作者按照 CC BY 4.0 进行授权