优化需求类 Markdown 文件以提高 RAG 系统查询准确度

标题：优化需求类 Markdown 文件以提高 RAG 系统查询准确度

概述

处理需求类的 Markdown 文件时，由于标题如“需求名称”、“功能描述”等过于通用，直接用于检索难以准确定位有价值的内容。因此需要对文件进行预处理，通过丰富标题语义、清洗文本、生成摘要、文本切分等步骤提高 RAG（Retrieval-Augmented Generation）系统的查询准确度。

示例 Markdown 文件

文件名：requirement.md

# 需求名称用户管理系统# 需求描述本系统需要实现对用户的增删改查操作，支持批量导入和导出用户数据。# 功能描述## 用户添加管理员可以添加新用户，填写用户名、密码、角色等信息。## 用户删除管理员可以删除现有用户，支持单个和批量删除。![用户删除流程图](https://example.com/images/delete_user_flowchart.png)# 非功能需求系统需要具备高可用性和安全性，支持权限控制和日志记录。

处理步骤

合并标题与内容：合并 Markdown 文件中每个标题及其下属内容，形成完整的章节。
丰富标题语义：根据章节内容，对通用标题（如“需求名称”）进行语义丰富。
清洗文本：去除不必要的 Markdown 标记、图片链接等，保留核心内容。
生成摘要：为每个章节生成简要摘要，方便快速理解内容。
语义文本切分：将长文本切分为语义完整的片段，适合模型处理。

处理步骤代码详解

合并标题与内容

合并每个标题及其下属内容。

def merge_titles_with_content(markdown_text):import relines = markdown_text.split('\n')merged_sections = []current_section = {'title': '', 'content': ''}title_pattern = re.compile(r'^(#{1,6})\s*(.*)')for line in lines:title_match = title_pattern.match(line)if title_match:if current_section['title'] or current_section['content']:merged_sections.append(current_section)current_section = {'title': title_match.group(2), 'content': ''}else:current_section['content'] += line + '\n'if current_section['title'] or current_section['content']:merged_sections.append(current_section)return merged_sections

丰富标题语义

通过章节内容的第一行来增强标题语义。

def enrich_title(section):generic_titles = ['需求名称', '需求描述', '功能描述', '非功能需求']title = section['title']content_lines = section['content'].strip().split('\n')if title in generic_titles and content_lines:enriched_title = f"{title}: {content_lines[0]}"else:enriched_title = titlereturn enriched_title

清洗文本

去除不必要的 Markdown 标记、图片链接等，使文本更简洁易于处理。

def clean_text(text):import retext = re.sub(r'#{1,6}\s*', '', text)text = re.sub(r'!\[([^\]]*)\]\([^\)]*\)', r'\1', text)text = re.sub(r'\s+', ' ', text).strip()return text

生成摘要

生成每个章节的摘要。

def generate_summary(text, max_length=50):from gensim.summarization import summarizetry:summary = summarize(text, word_count=max_length)if not summary:summary = text[:max_length]except ValueError:summary = text[:max_length]return summary

语义文本切分

将文本切分为语义完整的片段，适合向量化处理。

def semantic_text_split(text, max_length=100):import nltknltk.download('punkt', quiet=True)sentences = nltk.sent_tokenize(text)chunks = []current_chunk = ''for sentence in sentences:if len(current_chunk) + len(sentence) <= max_length:current_chunk += ' ' + sentenceelse:chunks.append(current_chunk.strip())current_chunk = sentenceif current_chunk:chunks.append(current_chunk.strip())return chunks

综合处理

将上述所有步骤整合在一起。

def process_requirement_markdown(markdown_text):sections = merge_titles_with_content(markdown_text)processed_data = []for section in sections:enriched_title = enrich_title(section)content_clean = clean_text(section['content'])summary = generate_summary(content_clean)chunks = semantic_text_split(content_clean)for chunk in chunks:combined_text = f"{enriched_title}\n{chunk}"processed_data.append({'text': combined_text,'metadata': {'title': enriched_title,'summary': summary}})return processed_data

应用：

with open('requirement.md', 'r', encoding='utf-8') as file:markdown_text = file.read()processed_data = process_requirement_markdown(markdown_text)
for item in processed_data:print("Processed Text:", item['text'])print("Metadata:", item['metadata'])print("---")

向量化与存储

对处理后的文本进行向量化和存储，以供后续检索使用。

向量化文本

from sentence_transformers import SentenceTransformerdef vectorize_texts(texts):model = SentenceTransformer('distiluse-base-multilingual-cased-v1')embeddings = model.encode(texts)return embeddings

应用：

texts = [item['text'] for item in processed_data]
embeddings = vectorize_texts(texts)

存储向量和元数据

import faiss
import numpy as npdef store_embeddings(embeddings, processed_data):dimension = embeddings.shape[1]index = faiss.IndexFlatL2(dimension)index.add(embeddings)faiss.write_index(index, 'requirement_vectors.index')metadata_list = [item['metadata'] for item in processed_data]np.save('requirement_metadata.npy', metadata_list)with open('requirement_texts.txt', 'w', encoding='utf-8') as f:for item in processed_data:f.write(item['text'] + '\n')

应用：

store_embeddings(embeddings, processed_data)

检索示例

通过向量化检索以实现精准查询。

def search_requirements(query, k=3):index = faiss.read_index('requirement_vectors.index')metadata_list = np.load('requirement_metadata.npy', allow_pickle=True)with open('requirement_texts.txt', 'r', encoding='utf-8') as f:texts = f.read().splitlines()query_embedding = vectorize_texts([query])distances, indices = index.search(query_embedding, k)results = []for idx in indices[0]:result = {'text': texts[idx],'metadata': metadata_list[idx]}results.append(result)return resultsquery = "如何删除用户？"
results = search_requirements(query)
for result in results:print("Result Text:", result['text'])print("Metadata:", result['metadata'])print("---")