网站字体选择今日热搜新闻头条

首先叠甲，本文只是用于学习目的，不用于其他用途

---

前言

---

项目目标

1. 学会 RAG 项目的构建流程： 从数据获取到最终部署的完整端到端流程。
2. 学习影响 RAG 项目性能的因素： 深入理解数据质量、嵌入模型、检索策略、重排序以及 Prompt 工程等关键因素。
3. 学习 RAG 和 LoRA 微调的结合场景： 理解何时以及如何将这两种技术结合，以实现最佳效果。

技术栈：

• RAG 框架： LlamaIndex
• 向量数据库： ChromaDB
• 知识库数据： 网上爬取的劳动合同法相关数据
• 大语言模型 (LLM)： Qwen/Qwen1.5-1.8B-Chat
• 嵌入模型： sungw111/text2vec-base-chinese-sentence
• 重排序模型： BAAI/bge-reranker-large

• 如果对于第 RAG 和微调不太清晰，我们已经在另一篇文章 RAG 和 微调如何抉择进行了介绍
• 如果对于 RAG 流程不太清晰，我们已经在RAG 流程进行了介绍
• 我们使用LlamaIndex进行构建 RAG 项目，并已经将LlamaIndex学习笔记分享了出来
• 对于向量数据库我们已经在Chroma 向量数据库学习笔记分享了出来

---

方案设计

---

项目阶段 1: 知识库构建

此阶段目标是将原始、非结构化的法律文本转化为可供向量检索的结构化数据，并进行持久化存储。

1.1 数据爬取

• 目标： 获取劳动合同法相关法律条文、司法解释、典型案例、权威解读等文本。
• 方法对比：

方法	优点	缺点
手动收集/下载	精度高，数据质量可控。	效率极低，数据量受限，难以更新。
通用爬虫工具	灵活性强，可定制化，能处理复杂网站结构。	需要编程技能，维护成本较高。
无代码/低代码平台	入门简单，操作直观。	灵活性有限，难以处理反爬机制或复杂数据提取。

• 最适合的方法及理由：
  • 推荐： 通用爬虫工具 (如 Python 的 Scrapy 或 BeautifulSoup)。
  • 理由： 劳动合同法资料多分布在政府官网、法院判例库、法律网站等，这些网站结构相对固定但可能存在反爬机制。通用爬虫工具能提供足够的灵活性和效率来批量获取数据，同时满足开源免费的要求。对于无法通过爬虫获取的少量权威资料，可少量手动下载。

本文爬取地址

1.2 数据处理 (清洗与预处理)

本小节为可选

• 目标： 清除爬取数据中的噪音，统一格式，提取必要信息。
• 方法对比：

方法	优点	缺点
正则表达式/字符串操作	直观，效率高，适合简单、规则化清洗。	难以处理复杂、不规则文本。
基于规则的解析器	专门用于解析结构化数据 (如 HTML/XML)。	仅限于结构化数据，对纯文本无效。
开源 NLP 库	能进行深层次语义理解和结构化。	复杂性高，需要 NLP 知识。

• 最适合的方法及理由：
  • 推荐： 使用正则表达式进行数据处理。
  • 理由： 法律文本往往具有一定的结构，如下表示，他总是由第 N 条来表示一条法律条文，因此采用正则表达式高效且有效处理即可。

<p>　　第一条　为了保护劳动者的合法权益，调整劳动关系，建立和维护适应社会主义市场经济的劳动制度，促进经济发展和社会进步，根据宪法，制定本法。</p>
<p>　　第二条　在中华人民共和国境内的企业、个体经济组织（以下统称用人单位）和与之形成劳动关系的劳动者，适用本法。</p>

1.3 数据向量化并持久化

• 目标： 将处理后的文本转化为嵌入向量，并可靠地存储在向量数据库中。

采用嵌入模型进行词嵌入，然后调用向量数据库存储即可，对于嵌入模型和向量数据库的选择见后文，此处暂时不做过多分析

项目阶段 2: 知识库处理（分块策略对比）

此阶段目标是优化文本分块，以提高检索时的语义准确性和 LLM 处理效率。

• 目标： 确定最适合劳动合同法文本的分块策略。
• 方法对比：

分块策略	优点	缺点	适用场景
固定大小分块	实现简单，可预测。	易切断语义，影响上下文连贯性。	快速原型，文本无强结构。
按句子/段落分块	保持基本语义单位。	段落仍可能长，句子过于碎片化。	通用文本，语义单元明确。
递归分块	尝试保留语义结构，灵活。	仍可能切断复杂逻辑，需精心选择分隔符。	结构化与非结构化混杂文本。
基于结构化信息分块	最能保持法律文本的逻辑完整性。	需要复杂的前期数据结构识别。	强结构化法律法规、判例。
小块-大上下文策略 (SWR/AMR)	索引小块精准，传递大上下文给 LLM。	实现稍复杂，需要额外处理逻辑。	兼顾检索效率与生成质量。

• 最适合的方法及理由：
  • 推荐：

    1. 基于结构化信息分块（真正使用） ：利用法律文本的明确结构（如“第X条”、“第Y款”）进行分块，确保每个块都是一个完整的法律概念。这是法律文本特有的优势。

    2. 递归分块：对于结构不明确的解读性文本或长段落，使用 LlamaIndex 的 RecursiveCharacterTextSplitter 进行二次分割，以确保块大小适中。

    3. 小块-大上下文策略 (高级优化)：索引时使用较小的语义单位（如句子或短条款），但在检索到这些小单位后，动态地扩展到包含更多上下文的大块，提供给 LLM。这能平衡检索效率（小块嵌入更精准）和生成质量（大上下文信息更丰富）。

  • 理由： 法律文本的特点是严谨的逻辑结构。简单分块容易破坏这种结构，导致 LLM 无法获取完整的法律背景。

对于2 和 3 在该项目中不再使用，理由：法律条文每条都比较短，因此只需要使用结构化信息分块，保证条文的完整性即可

项目阶段 3: 模型选择

此阶段目标是选择并配置 RAG 流程中的核心开源 AI 模型

注意这里我们采用的都是开源免费大模型，并且不会选择很大的模型，实操会做更详细的说明

3.1 嵌入模型选择

• 目标： 获得高质量的文本嵌入，准确捕捉法律文本的语义相似性。
• 方法对比：

嵌入模型系列	优点	缺点	推荐模型示例
BGE (BAAI General Embeddings)	MTEB 排行榜领先，性能优异；支持多种语言，有专门针对中英文的模型；开源免费，可本地部署。	需要一定的硬件资源（GPU）进行推理。	BAAI/bge-large-zh-v1.5 (中文)BAAI/bge-large-en-v1.5 (英文)
E5 (Embed Everything)	性能优秀，通用性强，有多语言版本；开源免费，可本地部署。	性能略低于 BGE，但在某些任务上表现出色。	intfloat/multilingual-e5-large
MiniLM (Sentence Transformers)	模型体积小，推理速度快，内存占用低；适合资源有限的环境。	性能上可能略逊于 BGE 和 E5 的大模型。	sentence-transformers/all-MiniLM-L6-v2
Cosine Sentence	专门针对中文句子训练，模型相较于前面模型较为轻量	长度限制，最大序列长度为 256	sungw111/text2vec-base-chinese-sentence

• 最适合的方法及理由：
  • 推荐： sungw111/text2vec-base-chinese-sentence
  • 理由： 通常嵌入模型对项目的影响较小，主要因素是由于语言的限制，该项目法律条文为中文，因此采用该模型最为合适。

3.2 问答模型选择 (LLM)

• 目标： 选择一个能理解法律上下文、遵守指令并生成准确、专业答案的开源 LLM。
• 方法对比：

开源 LLM 系列	优点	缺点	推荐模型示例
Llama 3 (Meta)	强大的通用推理能力，指令遵循能力好，社区活跃，模型版本多样。	需要较高的硬件资源（GPU），在某些特定领域可能需微调。	meta-llama/Llama-3-8B-Instruct (8B版本)<br>meta-llama/Llama-3-70B-Instruct (性能更强，资源要求更高)
Qwen (阿里)	强大的中文能力，在中文法律语料上有较好表现；模型版本丰富。	英文能力可能略逊于 Llama 系列；资源要求较高。	Qwen/Qwen2-7B-Instruct 或Qwen/Qwen1.5-1.8B-Chat
Mistral (Mistral AI)	轻量级但高性能，推理速度快，在英文任务上表现出色。	中文能力相对较弱（有微调版本）；资源要求中等。	mistralai/Mistral-7B-Instruct-v0.2
Baichuan2 (百川智能)	优秀的中文能力，在中文通用领域表现好。	更新迭代不如 Llama 3 频繁；在英文能力上相对弱势。	baichuan-inc/Baichuan2-7B-Chat

• 最适合的方法及理由：
  • 推荐： Qwen/Qwen2-7B-Instruct 或Qwen/Qwen1.5-1.8B-Chat (取决于硬件条件)。
  • 理由： Qwen2 是目前开源 LLM 中性能最顶尖的系列之一，它们在通用推理、指令遵循和多轮对话方面表现出色，是构建 RAG 助手的坚实基础。Qwen2 在中文处理上优势明显。选择7B或 1.5-1.8B-Chat版本取决于显存条件。

3.3 重排序模型选择 (Re-ranker)

• 目标： 对初步检索结果进行二次精细排序，提高相关性。
• 方法对比：

重排序模型类型	优点	缺点	推荐模型示例
开源交叉编码器	性能显著优于仅向量相似度；免费，可本地部署；能捕捉查询和文档之间更复杂的交互。	计算成本较高（需要为每个查询-文档对单独推理），推理速度相对较慢。	cross-encoder/ms-marco-MiniLM-L-6-v2<br>BAAI/bge-reranker-base
无重排序	实现简单，计算成本最低。	精度相对较低，可能将不那么相关的文档排在前面。	N/A

• 最适合的方法及理由：
  • 推荐： cross-encoder/ms-marco-MiniLM-L-6-v2 或 BAAI/bge-reranker-base。
  • 理由： 对于学习项目，免费且高性能的开源交叉编码器是理想选择。它们能够显著提升检索结果的准确性，帮助理解重排序的价值。MiniLM 模型相对轻量，适合初步实验。BGE Re-ranker 则提供更强大的性能。

项目阶段 4: 项目其他技术选择

此阶段巩固 RAG 框架和向量数据库的选择。

4.1 RAG 框架选择

• 目标： 选择一个功能全面、易于使用且适合 RAG 流程开发的开源框架。
• 方法对比：

RAG 框架	优点	缺点
LlamaIndex	专注于数据摄取、索引和检索；提供丰富的 Data Loaders、Node Parser、各种索引类型、Retriever 和 Response Synthesizer；社区活跃，文档丰富。	相对 LangChain，在 Agent 和 Chain 方面的抽象可能稍弱（但持续改进）。
LangChain	功能非常全面，不仅包括 RAG，还涵盖 Agent、Chain、Memory 等，生态系统庞大；Agent 方面的能力更强。	抽象层级较多，学习曲线可能稍陡峭；早期版本稳定性问题较多。

• 最适合的方法及理由：
  • 推荐： LlamaIndex。
  • 理由： LlamaIndex 的设计理念更侧重于数据与 LLM 的连接，其在数据索引、检索和上下文管理方面的抽象非常清晰和强大，这正是 RAG 项目的核心。对于“学会 RAG 项目的构建流程”这一目标，LlamaIndex 能提供更直接和聚焦的学习体验，且完全开源免费。

对于LlamaIndex和LangChain更详细的对比请参见

4.2 向量数据库选择

• 目标： 选择一个可靠、易于使用的开源向量数据库进行知识库存储。
• 方法对比：

向量数据库	优点	缺点	适用场景
ChromaDB	嵌入式，极易上手，轻量级；支持持久化；社区活跃，与 LlamaIndex/LangChain 集成良好。	不适合大规模分布式部署和高并发场景；性能不如专业分布式数据库。	学习项目，原型开发，小规模应用。
FAISS	免费，高性能，纯 C++ 实现，绑定了 Python；适合内存内高性能检索。	不支持持久化（需手动保存/加载索引）；仅支持向量存储，无元数据过滤等功能。	内存内高性能检索，不关注持久化和元数据。
Milvus/Zilliz	开源，高可用，弹性伸缩，支持大规模向量检索和分布式部署；功能丰富（过滤、标量字段索引）。	部署和运维复杂，资源消耗大；对学习项目来说初期投入较大。	大规模、生产级 RAG 系统，高并发、高可用。
Weaviate	开源，原生支持向量存储和元数据索引，语义搜索能力强；提供 GraphQL API。	部署相对复杂，资源消耗较高。	需要复杂元数据过滤，关注语义搜索能力的生产级应用。

• 最适合的方法及理由：
  • 推荐： ChromaDB。
  • 理由： ChromaDB 的嵌入式特性使其成为学习和原型开发的最佳选择。它无需复杂部署和管理，直接作为 Python 库即可使用，能够大幅降低项目初期门槛，完全符合“学习 RAG 项目构建流程”的需求。同时，它支持数据持久化，保证了知识库的可靠存储，且完全免费开源。

项目阶段 5: 项目评估

• 目标： 量化 RAG 系统的性能，并为持续优化提供依据。
• 评估指标：
  • 检索准确性 (Retrieval Precision/Recall)： 召回的文档是否与查询相关，最相关的文档是否被召回。
  • 答案忠实度 (Faithfulness)： LLM 生成的答案是否完全基于检索到的上下文，没有编造信息（幻觉）。
  • 答案相关性 (Answer Relevance)： LLM 生成的答案是否直接、完整且流畅地回答了用户问题。
  • 效率指标： 延迟 (Latency)，token 消耗量。
• 评估工具 (开源免费)：
  • RAGAS： 专门用于 RAG 系统评估的开源框架，可评估答案忠实度、答案相关性、上下文精度、上下文召回率等。
  • LlamaIndex 内置评估工具： 提供 ResponseEvaluator, FaithfulnessEvaluator 等，可配合自定义评估数据进行使用。
• 评估流程：
  1. 构建测试集： 收集一组具有代表性的劳动合同法问题，并人工标注其预期答案和/或相关文档。
  2. 自动化评估： 使用 RAGAS 或 LlamaIndex 评估工具进行自动化评估。
  3. 人工审核： 对于关键指标（如幻觉率、答案准确性），进行严格的人工抽样审核，尤其是在法律领域，这至关重要。
  4. 反馈循环： 根据评估结果，识别性能瓶颈，并反馈到数据处理、分块策略、模型选择等阶段进行优化。

---

整体结构图 (Mermaid)

LoRA 微调的介入时机： 当发现使用开源 LLM 结合 RAG，其在 法律领域的专业术语、回答风格、特定任务（如案例总结、法条解释的结构化输出）方面 仍有欠缺时，可以考虑对问答模型 (LLM) 进行 LoRA 微调。

注意，真实项目中我们首先应该考虑的是模型是不是使用更好的模型

通过这种方式，RAG 负责提供最新的事实知识，而 LoRA 微调则使 LLM 能够以更专业的“法律助手”身份来消化和呈现这些知识，达到协同增效的效果。

---

项目实现

---

1. 知识库构建

• 爬虫并保存数据为json文件

def fetch_and_parse(url):# 请求网页response = requests.get(url)# 设置网页编码格式response.encoding = 'utf-8'# 解析网页内容soup = BeautifulSoup(response.text, 'html.parser')# 提取正文内容content = soup.find_all('p')# 初始化存储数据data = []# 提取文本并格式化for para in content:text = para.get_text(strip=True)if text:  # 只处理非空文本# 根据需求格式化内容data.append(text)# 将data列表转换为字符串data_str = '\n'.join(data)return data_strdef extract_law_articles(data_str):# 正则表达式，匹配每个条款号及其内容pattern = re.compile(r'第([一二三四五六七八九十零百]+)条.*?(?=\n第|$)', re.DOTALL)# 初始化字典来存储条款号和内容lawarticles = {}# 搜索所有匹配项for match in pattern.finditer(data_str):articlenumber = match.group(1)articlecontent = match.group(0).replace('第' + articlenumber + '条', '').strip()lawarticles[f"中华人民共和国劳动法 第{articlenumber}条"] = articlecontent# 转换字典为JSON字符串jsonstr = json.dumps(lawarticles, ensure_ascii=False, indent=4)return jsonstrif __name__ == '__main__':# 请求页面url = "https://www.gov.cn/banshi/2005-05/25/content_905.htm"data_str = fetch_and_parse(url)json_str = extract_law_articles(data_str)print(json_str)

保存的json数据结果为：

[{"中华人民共和国劳动合同法 第一条": "为了完善劳动合同制度...",},{"中华人民共和国劳动法 第一条": "为了...",}]

• 加载json数据

def load_and_validate_json_files(data_dir: str) -> List[Dict]:"""加载并验证JSON法律文件"""json_files = list(Path(data_dir).glob("*.json"))assert json_files, f"未找到JSON文件于 {data_dir}"all_data = []for json_file in json_files:with open(json_file, 'r', encoding='utf-8') as f:try:data = json.load(f)# 验证数据结构if not isinstance(data, list):raise ValueError(f"文件 {json_file.name} 根元素应为列表")for item in data:if not isinstance(item, dict):raise ValueError(f"文件 {json_file.name} 包含非字典元素")for k, v in item.items():if not isinstance(v, str):raise ValueError(f"文件 {json_file.name} 中键 '{k}' 的值不是字符串")all_data.extend({"content": item,"metadata": {"source": json_file.name}} for item in data)except Exception as e:raise RuntimeError(f"加载文件 {json_file} 失败: {str(e)}")print(f"成功加载 {len(all_data)} 个法律文件条目")return all_data

• 数据向量化并持久化

def init_vector_store(nodes: List[TextNode]) -> VectorStoreIndex:chroma_client = chromadb.PersistentClient(path=Config.VECTOR_DB_DIR)chroma_collection = chroma_client.get_or_create_collection(name=Config.COLLECTION_NAME,metadata={"hnsw:space": "cosine"})# 确保存储上下文正确初始化storage_context = StorageContext.from_defaults(vector_store=ChromaVectorStore(chroma_collection=chroma_collection))# 判断是否需要新建索引if chroma_collection.count() == 0 and nodes is not None:print(f"创建新索引（{len(nodes)}个节点）...")# 显式将节点添加到存储上下文storage_context.docstore.add_documents(nodes)index = VectorStoreIndex(nodes,storage_context=storage_context,show_progress=True)# 双重持久化保障storage_context.persist(persist_dir=Config.PERSIST_DIR)index.storage_context.persist(persist_dir=Config.PERSIST_DIR)  # <-- 新增else:print("加载已有索引...")storage_context = StorageContext.from_defaults(persist_dir=Config.PERSIST_DIR,vector_store=ChromaVectorStore(chroma_collection=chroma_collection))index = VectorStoreIndex.from_vector_store(storage_context.vector_store,storage_context=storage_context,embed_model=Settings.embed_model)# 安全验证print("\n存储验证结果：")doc_count = len(storage_context.docstore.docs)print(f"DocStore记录数：{doc_count}")if doc_count > 0:sample_key = next(iter(storage_context.docstore.docs.keys()))print(f"示例节点ID：{sample_key}")else:print("警告：文档存储为空，请检查节点添加逻辑！")return index

2. LamaIndex设置

• 创建LlamaIndex的文档Node

def create_nodes(raw_data: List[Dict]) -> List[TextNode]:"""添加ID稳定性保障"""nodes = []for entry in raw_data:law_dict = entry["content"]source_file = entry["metadata"]["source"]for full_title, content in law_dict.items():# 生成稳定ID（避免重复）node_id = f"{source_file}::{full_title}"parts = full_title.split(" ", 1)law_name = parts[0] if len(parts) > 0 else "未知法律"article = parts[1] if len(parts) > 1 else "未知条款"node = TextNode(text=content,id_=node_id,  # 显式设置稳定IDmetadata={"law_name": law_name,"article": article,"full_title": full_title,"source_file": source_file,"content_type": "legal_article"})nodes.append(node)print(f"生成 {len(nodes)} 个文本节点（ID示例：{nodes[0].id_}）")return nodes

• 模型基础信息定义

class Config:EMBED_MODEL_PATH = r"../embedding_model/sungw111/text2vec-base-chinese-sentence"RERANK_MODEL_PATH = r"../rerank_model/BAAI/bge-reranker-large"  # 新增重排序模型路径LLM_MODEL_PATH = r"../Qwen/Qwen1___5-1___8B-Chat"# 数据路径DATA_DIR = "./data"VECTOR_DB_DIR = "./chroma_db"PERSIST_DIR = "./storage"# chromadb的集合名COLLECTION_NAME = "labor_laws"TOP_K = 10  # 扩大初始检索数量RERANK_TOP_K = 3  # 重排序后保留数量MIN_RERANK_SCORE = 0.4

TOP_K设置的稍微大一些，类似粗处理，然后重排序的RERANK_TOP_K设置稍微小一些，以防过于不相关的影响大模型判断

• 模型初始化

def init_models():"""初始化模型并验证"""# Embedding模型embed_model = HuggingFaceEmbedding(model_name=Config.EMBED_MODEL_PATH,)# LLMllm = HuggingFaceLLM(model_name=Config.LLM_MODEL_PATH,tokenizer_name=Config.LLM_MODEL_PATH,model_kwargs={"trust_remote_code": True,},tokenizer_kwargs={"trust_remote_code": True},generate_kwargs={"temperature": 0.3})# 初始化重排序器reranker = SentenceTransformerRerank(model=Config.RERANK_MODEL_PATH,top_n=Config.RERANK_TOP_K)Settings.embed_model = embed_modelSettings.llm = llmreturn embed_model, llm, reranker  # 返回重排序器

其中问答模型使用HuggingFaceLLM推理引擎，表现稍微慢一些，并且在实际测试的时候确实存在一些奇怪的bug，比如模型回复总是说一半就中断，因此在实际使用的时候应该优先考虑LMDeploy作为推理引擎，或者使用Vllm，使用方法如下：

llm = OpenAILike(model="../Qwen/Qwen1.5-1.8B-Chat-law",api_base="http://localhost:23333/v1",api_key="fake",context_window=4096,is_chat_model=True,is_function_calling_model=False,)

3. 主程序逻辑

• 准备工作（即下面的 1、2、3）

# 1.获取三个模型
embed_model, llm, reranker = init_models()# 2. 创建 Node并获取可查询的索引inde
# 仅当需要更新数据时执行
if not Path(Config.VECTOR_DB_DIR).exists():print("\n初始化数据...")raw_data = load_and_validate_json_files(Config.DATA_DIR)nodes = create_nodes(raw_data)
else:nodes = Noneprint("\n初始化向量存储...")
start_time = time.time()
index = init_vector_store(nodes)
print(f"索引加载耗时：{time.time()-start_time:.2f}s")# 3. 创建检索器和响应合成器
retriever = index.as_retriever(similarity_top_k=Config.TOP_K  # 扩大初始检索数量
)
response_synthesizer = get_response_synthesizer(# text_qa_template=response_template,verbose=True
)

• 用户输入检索

为了可以运行之后反复询问大模型，下面的逻辑在while(true)逻辑中，在用户输入q时退出循环

 question = input("\n请输入劳动法相关问题（输入q退出）: ")if question.lower() == 'q':breakstart_time = time.time()# 初始检索initial_nodes = retriever.retrieve(question)retrieval_time = time.time() - start_time

• 检索之后进行重排序

reranked_nodes = reranker.postprocess_nodes(initial_nodes, query_str=question)
rerank_time = time.time() - start_time - retrieval_time# 执行过滤
filtered_nodes = [node for node in reranked_nodes if node.score > Config.MIN_RERANK_SCORE
]
# for node in reranked_nodes:
#     print(node.score)
#一般对模型的回复做限制就从filtered_nodes的返回值下手
print("原始分数样例：",[node.score for node in reranked_nodes[:3]])
print("重排序过滤后的结果：",filtered_nodes)
# 空结果处理
if not filtered_nodes:print("你的问题未匹配到相关资料！")continue

设置阈值MIN_RERANK_SCORE，在重排序后，节点的相似度分数小于该值则丢掉该节点

• 回复生成

response = response_synthesizer.synthesize(question, nodes=filtered_nodes  # 使用过滤后的节点)
synthesis_time = time.time() - start_time - retrieval_time - rerank_time

通过用户的问题输入和过滤后的节点(即检索到的知识库)合并交给大模型进行生成回复

• 结果打印

# 显示结果（修改显示逻辑）
print(f"\n智能助手回答：\n{response.response}")
print("\n支持依据：")
for idx, node in enumerate(reranked_nodes, 1):# 兼容新版API的分数获取方式initial_score = node.metadata.get('initial_score', node.score)  # 获取初始分数rerank_score = node.score  # 重排序后的分数meta = node.node.metadataprint(f"\n[{idx}] {meta['full_title']}")print(f"  来源文件：{meta['source_file']}")print(f"  法律名称：{meta['law_name']}")print(f"  初始相关度：{node.node.metadata.get('initial_score', 0):.4f}")  # 安全访问print(f"  重排序得分：{getattr(node, 'score', 0):.4f}")  # 兼容属性访问print(f"  条款内容：{node.node.text[:100]}...")print(f"\n[性能分析] 检索: {retrieval_time:.2f}s | 重排序: {rerank_time:.2f}s | 合成: {synthesis_time:.2f}s")

4. 项目评估

4.1 评估数据准备

• 召回率评估

RETRIEVAL_BENCHMARK = [# 劳动合同解除类{"question": "劳动者可以立即解除劳动合同的情形有哪些？","relevant_ids": ["中华人民共和国劳动合同法 第三十八条"],"confusing_ids": ["中华人民共和国劳动合同法 第三十九条", "中华人民共和国劳动法 第三十二条"]},{"question": "用人单位单方解除劳动合同需要提前多久通知？","relevant_ids": ["中华人民共和国劳动合同法 第四十条"],"confusing_ids": ["中华人民共和国劳动合同法 第三十七条", "中华人民共和国劳动法 第二十六条"]},# 工资与补偿类{"question": "经济补偿金的计算标准是什么？","relevant_ids": ["中华人民共和国劳动合同法 第四十七条"],"confusing_ids": ["中华人民共和国劳动合同法 第八十七条", "中华人民共和国劳动法 第二十八条"]},{"question": "试用期工资最低标准是多少？","relevant_ids": ["中华人民共和国劳动合同法 第二十条"],"confusing_ids": ["中华人民共和国劳动合同法 第十九条", "中华人民共和国劳动法 第四十八条"]},# 工伤与福利类{"question": "工伤认定需要哪些材料？","relevant_ids": ["中华人民共和国劳动合同法 第三十条"],"confusing_ids": ["中华人民共和国劳动法 第七十三条", "中华人民共和国劳动合同法 第十七条"]},{"question": "女职工产假有多少天？","relevant_ids": ["中华人民共和国劳动法 第六十二条"],"confusing_ids": ["中华人民共和国劳动合同法 第四十二条", "中华人民共和国劳动法 第六十一条"]},# 劳动合同订立类{"question": "无固定期限劳动合同的订立条件是什么？","relevant_ids": ["中华人民共和国劳动合同法 第十四条"],"confusing_ids": ["中华人民共和国劳动合同法 第十三条", "中华人民共和国劳动法 第二十条"]},{"question": "劳动合同必须包含哪些条款？","relevant_ids": ["中华人民共和国劳动合同法 第十七条"],"confusing_ids": ["中华人民共和国劳动法 第十九条", "中华人民共和国劳动合同法 第十条"]},# 特殊用工类{"question": "劳务派遣岗位的限制条件是什么？","relevant_ids": ["中华人民共和国劳动合同法 第六十六条"],"confusing_ids": ["中华人民共和国劳动合同法 第五十八条", "中华人民共和国劳动法 第二十条"]},{"question": "非全日制用工的每日工作时间上限？","relevant_ids": ["中华人民共和国劳动合同法 第六十八条"],"confusing_ids": ["中华人民共和国劳动法 第三十六条", "中华人民共和国劳动合同法 第三十八条"]},# 劳动合同终止类{"question": "劳动合同终止的法定情形有哪些？","relevant_ids": ["中华人民共和国劳动合同法 第四十四条"],"confusing_ids": ["中华人民共和国劳动合同法 第四十六条", "中华人民共和国劳动法 第二十三条"]},{"question": "劳动合同期满后必须续签的情形？","relevant_ids": ["中华人民共和国劳动合同法 第四十五条"],"confusing_ids": ["中华人民共和国劳动合同法 第十四条", "中华人民共和国劳动法 第二十条"]},# 劳动保护类{"question": "女职工哺乳期工作时间限制？","relevant_ids": ["中华人民共和国劳动法 第六十三条"],"confusing_ids": ["中华人民共和国劳动合同法 第四十二条", "中华人民共和国劳动法 第六十一条"]},{"question": "未成年工禁止从事的劳动类型？","relevant_ids": ["中华人民共和国劳动法 第六十四条"],"confusing_ids": ["中华人民共和国劳动法 第五十九条", "中华人民共和国劳动合同法 第六十六条"]},{"question": "工伤保险待遇包含哪些项目？","relevant_ids": ["中华人民共和国劳动法 第七十三条"],"confusing_ids": ["中华人民共和国劳动合同法 第三十条", "中华人民共和国劳动法 第四十四条"]},# 劳动争议类{"question": "劳动争议仲裁时效是多久？","relevant_ids": ["中华人民共和国劳动法 第八十二条"],"confusing_ids": ["中华人民共和国劳动合同法 第六十条", "中华人民共和国劳动法 第七十九条"]},{"question": "集体合同的法律效力如何？","relevant_ids": ["中华人民共和国劳动法 第三十五条"],"confusing_ids": ["中华人民共和国劳动合同法 第五十五条", "中华人民共和国劳动法 第三十三条"]},# 特殊条款类{"question": "服务期违约金的上限规定？","relevant_ids": ["中华人民共和国劳动合同法 第二十二条"],"confusing_ids": ["中华人民共和国劳动合同法 第二十三条", "中华人民共和国劳动法 第一百零二条"]},{"question": "无效劳动合同的认定标准？","relevant_ids": ["中华人民共和国劳动合同法 第二十六条"],"confusing_ids": ["中华人民共和国劳动法 第十八条", "中华人民共和国劳动合同法 第三十九条"]}
]

• 端到端评估

E2E_BENCHMARK = [# 案例1：劳动合同解除{"question": "用人单位在哪些情况下不得解除劳动合同？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第四十二条"],"标准答案": "根据《劳动合同法》第四十二条，用人单位不得解除劳动合同的情形包括：\n1. 从事接触职业病危害作业的劳动者未进行离岗前职业健康检查\n2. 在本单位患职业病或者因工负伤并被确认丧失/部分丧失劳动能力\n3. 患病或非因工负伤在规定的医疗期内\n4. 女职工在孕期、产期、哺乳期\n5. 连续工作满15年且距退休不足5年\n6. 法律、行政法规规定的其他情形\n违法解除需按第八十七条支付二倍经济补偿金","必备条件": ["职业病危害作业未检查", "孕期女职工", "连续工作满15年"]}},# 案例2：工资支付{"question": "拖欠工资劳动者可以采取哪些措施？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第三十条", "中华人民共和国劳动法 第五十条"],"标准答案": "劳动者可采取以下救济措施：\n1. 根据劳动合同法第三十条向法院申请支付令\n2. 依据劳动合同法第三十八条解除合同并要求经济补偿\n3. 向劳动行政部门投诉\n逾期未支付的，用人单位需按应付金额50%-100%加付赔偿金（劳动合同法第八十五条）","必备条件": ["支付令申请", "解除劳动合同", "行政投诉"]}},# 案例3：竞业限制{"question": "竞业限制的最长期限是多久？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第二十四条"],"标准答案": "根据劳动合同法第二十四条：\n- 竞业限制期限不得超过二年\n- 适用人员限于高管/高级技术人员/其他保密人员\n- 需按月支付经济补偿\n注意区分服务期约定（第二十二条）","限制条件": ["期限≤2年", "按月支付补偿"]}},# 案例4：劳务派遣{"question": "劳务派遣用工的比例限制是多少？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第六十六条"],"标准答案": "劳务派遣用工限制：\n- 临时性岗位不超过6个月\n- 辅助性岗位≤用工总量10%\n- 违法派遣按每人1000-5000元罚款（第九十二条）\n派遣协议需包含岗位/期限/报酬等条款（第五十九条）","限制条件": ["临时性≤6月", "辅助性≤10%"]}},# 案例5：非全日制用工{"question": "非全日制用工的工资支付周期要求？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第七十二条"],"标准答案": "非全日制用工支付规则：\n- 工资结算周期≤15日\n- 小时工资≥当地最低标准\n- 终止用工不支付经济补偿（第七十一条）\n区别于全日制月薪制（第三十条）","支付规则": ["周期≤15天", "小时工资≥最低标准"]}},# 案例6：劳动合同无效{"question": "劳动合同被确认无效后的工资支付标准？","standard_answer": {"条款": ["中华人民共和国劳动合同法 第二十八条"],"标准答案": "无效劳动合同的工资支付：\n1. 参照本单位相同岗位工资支付\n2. 无相同岗位的按市场价\n3. 已付报酬不足的需补差\n过错方需承担赔偿责任（第八十六条）","支付规则": ["参照同岗位", "市场价补差"]}}
]

4.2 评估代码实现

召回率评估器

class RecallEvaluator:def __init__(self, retriever, reranker):self.retriever = retrieverself.reranker = rerankerdef calculate_recall(self, retrieved_nodes, relevant_ids):retrieved_ids = [n.node.metadata["full_title"] for n in retrieved_nodes]hit = len(set(retrieved_ids) & set(relevant_ids))return hit / len(relevant_ids) if relevant_ids else 0.0def evaluate(self, benchmark):results = []for case in benchmark:# 初始检索initial_nodes = self.retriever.retrieve(case["question"])# 重排序reranked_nodes = self.reranker.postprocess_nodes(initial_nodes, query_str=case["question"])# 计算召回率recall = self.calculate_recall(reranked_nodes, case["relevant_ids"])results.append(recall)print(f"问题：{case['question']}")print(f"初始检索结果：{[n.node.metadata['full_title'] for n in initial_nodes]}")print(f"重排序后结果：{[n.node.metadata['full_title'] for n in reranked_nodes]}")print(f"召回条款：{[n.node.metadata['full_title'] for n in reranked_nodes[:3]]}")print(f"目标条款：{case['relevant_ids']}")print(f"召回率：{recall:.1%}\n")avg_recall = np.mean(results)print(f"平均召回率：{avg_recall:.1%}")return avg_recall

calculate_recall：该方法计算给定检索结果的召回率Recall。

• 输入：retrieved_nodes（检索到的节点，通常是一个文档列表）和 relevant_ids（目标相关条款的 ID 列表）。
• retrieved_ids 提取检索结果中每个节点的 full_title 元数据。
• hit 计算检索结果中的相关条款数量，即在检索结果中存在于目标相关条款中的数量。
• 最后返回召回率，计算公式是：召回率 = 找到的相关条款数量 / 目标条款总数。如果没有目标条款（relevant_ids为空），返回 0。

benchmark 是一个包含若干查询和其相关条款的字典列表，每个字典通常包含：查询的问题 (question)、相关条款的 ID 列表 (relevant_ids) 等信息。

• evaluate 方法是该类的核心评估逻辑，接受一个基准数据集（benchmark）作为输入，执行评估任务。

• 在每次循环中：

  • 初始检索： 调用 retriever.retrieve(case["question"]) 执行检索操作，根据问题获取初步检索结果 initial_nodes。
  • 重排序： 使用 reranker.postprocess_nodes() 对初步结果进行重排序，获得优化后的检索结果 reranked_nodes。此时会传入查询字符串（query_str）以指导重排序。
  • 召回率计算： 调用 calculate_recall() 方法，计算基于重排序结果与目标条款的召回率。

端到端评估器

评估给定查询的返回结果，判断它们是否命中了预期的条款，并计算命中率

class E2EEvaluator:def __init__(self, query_engine):self.query_engine = query_enginedef evaluate_case(self, response, standard):try:# 获取实际命中的条款retrieved_clauses = [node.node.metadata["full_title"] for node in response.source_nodes]# 获取标准答案要求的条款required_clauses = standard["standard_answer"]["条款"]# 计算命中情况hit_clauses = list(set(retrieved_clauses) & set(required_clauses))missed_clauses = list(set(required_clauses) - set(retrieved_clauses))# 计算命中率clause_hit = len(hit_clauses) / len(required_clauses) if required_clauses else 0.0return {"clause_score": clause_hit,"hit_clauses": hit_clauses,"missed_clauses": missed_clauses}except Exception as e:print(f"评估失败：{str(e)}")return None

• 功能： 该方法负责评估单个查询的结果。

• 输入：

  • response: 查询引擎返回的结果，通常包含了与查询相关的条款。
  • standard: 该查询对应的标准答案，包含期望的条款。

• 处理流程：

  • retrieved_clauses: 从 response.source_nodes 提取实际检索到的条款，假设 source_nodes 是一个包含节点（文档）的对象，其中每个节点都有一个 metadata 属性，其中包含条款的详细信息。
  • required_clauses: 从标准答案中提取期望的条款。
  • 计算实际命中的条款 hit_clauses 和缺失的条款 missed_clauses，分别通过集合的交集和差集来实现。
  • 计算条款命中率 clause_hit，即命中的条款数除以期望条款的总数。如果没有期望条款，返回 0.0。

• 输出： 返回一个字典，包含：

  • "clause_score": 条款命中率（命中条款数除以标准条款数）。
  • "hit_clauses": 命中的条款列表。
  • "missed_clauses": 缺失的条款列表。

  如果在评估过程中发生错误（如无法访问 response 或 standard 中的某些属性），会打印错误信息并返回 None。

    def evaluate(self, benchmark):results = []for case in benchmark:try:response = self.query_engine.query(case["question"])case_result = self.evaluate_case(response, case)if case_result:print(f"\n问题：{case['question']}")print(f"命中条款：{case_result['hit_clauses']}")print(f"缺失条款：{case_result['missed_clauses']}")print(f"条款命中率：{case_result['clause_score']:.1%}")results.append(case_result)else:results.append(None)except Exception as e:print(f"查询失败：{str(e)}")results.append(None)# 计算统计数据valid_results = [r for r in results if r is not None]avg_hit = np.mean([r["clause_score"] for r in valid_results]) if valid_results else 0print("\n=== 最终评估报告 ===")print(f"有效评估案例：{len(valid_results)}/{len(benchmark)}")print(f"平均条款命中率：{avg_hit:.1%}")# 输出详细错误分析for i, result in enumerate(results):if result is None:print(f"案例{i+1}：{benchmark[i]['question']} 评估失败")return results

• 功能： 该方法负责对整个基准数据集（benchmark）进行评估，计算所有查询的整体效果。

• 输入： benchmark 是一个包含多个查询案例的列表。每个案例包括查询问题（question）和标准答案（standard_answer）。

• 处理流程：

  • 循环遍历 benchmark 中的每个查询案例（case）。
  • 对每个查询，调用 query_engine.query() 执行查询，获取响应。
  • 使用 evaluate_case() 方法评估查询的结果，得到命中条款、缺失条款和条款命中率。
  • 如果评估成功，打印查询、命中条款、缺失条款以及命中率等信息，并将结果添加到 results 列表中。
  • 如果评估失败，记录 None。

  统计部分：

  • 计算所有有效结果的平均命中率（avg_hit）。
  • 输出最终的评估报告，包括：
    • 有效评估案例的数量。
    • 平均条款命中率。
  • 输出详细的错误分析，标记哪些案例的评估失败。

• 输出： 返回 results 列表，包含每个查询的评估结果（包含命中条款、缺失条款、命中率等信息）。如果查询失败，结果为 None。

主程序逻辑

print("\n=== 开始评估 ===")# 召回率评估
recall_evaluator = RecallEvaluator(retriever, reranker)
recall_result = recall_evaluator.evaluate(RETRIEVAL_BENCHMARK)# 端到端评估
e2e_evaluator = E2EEvaluator(query_engine)
e2e_results = e2e_evaluator.evaluate(E2E_BENCHMARK)# 生成报告
print("\n=== 最终评估报告 ===")
print(f"重排序召回率：{recall_result:.1%}")
print(f"端到端条款命中率：{np.mean([r['clause_score'] for r in e2e_results]):.1%}")

5. 可视化页面（可选）

我们采用Streamlit，Streamlit 是一个开源的 Python 库，用于快速构建和共享数据应用程序。它可以让开发者用简洁的代码将机器学习模型、数据可视化、分析工具等快速构建为交互式的 Web 应用

页面设置

st.set_page_config(page_title="智能劳动法咨询助手",page_icon="⚖️",layout="centered",initial_sidebar_state="auto"
)def disable_streamlit_watcher():"""Patch Streamlit to disable file watcher"""def _on_script_changed(_):returnfrom streamlit import runtimeruntime.get_instance()._on_script_changed = _on_script_changed

为了让页面变化更顺畅，用户体验好一些，在上述代码中修改如下

@st.cache_resource(show_spinner="初始化模型中...")
def init_models():
...@st.cache_resource(show_spinner="加载知识库中...")
def init_vector_store(_nodes):
...

页面组件设置

def init_chat_interface():if "messages" not in st.session_state:st.session_state.messages = []for msg in st.session_state.messages:role = msg["role"]content = msg.get("cleaned", msg["content"])  # 优先使用清理后的内容with st.chat_message(role):st.markdown(content)# 如果是助手消息且包含思维链if role == "assistant" and msg.get("think"):with st.expander("📝 模型思考过程（历史对话）"):for think_content in msg["think"]:st.markdown(f'<span style="color: #808080">{think_content.strip()}</span>',unsafe_allow_html=True)# 如果是助手消息且有参考依据（需要保持原有参考依据逻辑）if role == "assistant" and "reference_nodes" in msg:show_reference_details(msg["reference_nodes"])def show_reference_details(nodes):with st.expander("查看支持依据"):for idx, node in enumerate(nodes, 1):meta = node.node.metadatast.markdown(f"**[{idx}] {meta['full_title']}**")st.caption(f"来源文件：{meta['source_file']} | 法律名称：{meta['law_name']}")st.markdown(f"相关度：`{node.score:.4f}`")# st.info(f"{node.node.text[:300]}...")st.info(f"{node.node.text}")

重新修改后的主程序

def main():# 禁用 Streamlit 文件热重载disable_streamlit_watcher()st.title("⚖️ 智能劳动法咨询助手")st.markdown("欢迎使用劳动法智能咨询系统，请输入您的问题，我们将基于最新劳动法律法规为您解答。")# 初始化会话状态if "history" not in st.session_state:st.session_state.history = []# 加载模型和索引embed_model, llm, reranker = init_models()# 初始化数据if not Path(Config.VECTOR_DB_DIR).exists():with st.spinner("正在构建知识库..."):raw_data = load_and_validate_json_files(Config.DATA_DIR)nodes = create_nodes(raw_data)else:nodes = Noneindex = init_vector_store(nodes)retriever = index.as_retriever(similarity_top_k=Config.TOP_K,vector_store_query_mode="hybrid",alpha=0.5)response_synthesizer = get_response_synthesizer(verbose=True)# 聊天界面init_chat_interface()if prompt := st.chat_input("请输入劳动法相关问题"):# 添加用户消息到历史st.session_state.messages.append({"role": "user", "content": prompt})with st.chat_message("user"):st.markdown(prompt)# 处理查询with st.spinner("正在分析问题..."):start_time = time.time()# 检索流程initial_nodes = retriever.retrieve(prompt)reranked_nodes = reranker.postprocess_nodes(initial_nodes, query_str=prompt)# 过滤节点MIN_RERANK_SCORE = 0.4filtered_nodes = [node for node in reranked_nodes if node.score > MIN_RERANK_SCORE]if not filtered_nodes:response_text = "⚠️ 未找到相关法律条文，请尝试调整问题描述或咨询专业律师。"else:# 生成回答response = response_synthesizer.synthesize(prompt, nodes=filtered_nodes)response_text = response.response# 显示回答with st.chat_message("assistant"):# 提取思维链内容并清理响应文本think_contents = re.findall(r'<think>(.*?)</think>', response_text, re.DOTALL)cleaned_response = re.sub(r'<think>.*?</think>', '', response_text, flags=re.DOTALL).strip()# 显示清理后的回答st.markdown(cleaned_response)# 如果有思维链内容则显示if think_contents:with st.expander("📝 模型思考过程（点击展开）"):for content in think_contents:st.markdown(f'<span style="color: #808080">{content.strip()}</span>', unsafe_allow_html=True)# 显示参考依据（保持原有逻辑）show_reference_details(filtered_nodes[:3])# 添加助手消息到历史（需要存储原始响应）st.session_state.messages.append({"role": "assistant","content": response_text,  # 保留原始响应"cleaned": cleaned_response,  # 存储清理后的文本"think": think_contents  # 存储思维链内容})

运行的页面大致如下

---

总结收尾

---

到这里其实就已经完成整个 RAG 项目了，我们做一下总结

通过这个项目，我们不仅构建了一个基于开源技术栈的劳动合同法法律助手，更重要的是，它为我们深入理解 RAG 系统的原理、构建流程以及性能优化提供了宝贵的实践经验。

1. 项目成果回顾：

我们成功地从零开始，搭建了一个具备以下核心功能的 RAG 法律助手：

• 数据驱动的知识库： 通过网络爬取和数据清洗，构建了劳动合同法相关的专属知识库。
• 高效的向量检索： 利用嵌入模型和向量数据库，实现了对法律条文的快速语义检索。
• 精准的检索增强： 通过结合结构化分块、递归分块及小块-大上下文策略，优化了知识库的组织，并利用 BGE Re-ranker 对检索结果进行精细重排序，确保提供给 LLM 的上下文高度相关。
• 智能的问答能力： 集成Qwen2 开源 LLM，使助手能够根据检索到的法律依据，生成专业、准确的答案。
• 可评估的性能： 明确了评估指标（检索准确性、答案忠实度、答案相关性等），并计划使用 RAGAS 等工具进行量化评估，为持续优化奠定基础。
• 未来可扩展性： 方案中预留了 LoRA 微调的路径，当面临需要个性化的需求提供可能。

2. 核心技术选择的思考：

• LlamaIndex 与 ChromaDB 的组合： 这种选择在学习和原型开发阶段表现出极高的效率。LlamaIndex 专注于 RAG 核心流程的清晰抽象，使得数据摄取、索引、查询引擎的构建变得直观。ChromaDB 的嵌入式特性则极大降低了部署和管理门槛，让开发者能将更多精力放在 RAG 逻辑本身而非基础设施。这种组合特别适合个人项目或资源有限的团队进行快速迭代。
• BGE 嵌入与重排序模型： BGE 系列模型在开源社区中表现出色。实践中，我们会深刻体会到，高质量的嵌入是 RAG 的基石（当然嵌入模型主要是对语言的考量），而重排序则是提升用户体验的关键一步。
• Qwen2 等开源 LLM：我们选择模型都是选择的开源免费且参数量较小的模型，但是这会带来一些问题，因此该项目实现只用于学习。

3. 影响 RAG 性能的关键因素再思考：

• “垃圾进，垃圾出”法则： 数据质量是压倒一切的。法律领域的严谨性使得数据清洗、去重和标准化变得尤为重要。不准确、过时或有偏见的数据将直接导致助手提供错误或误导性的法律建议，这在法律场景下是绝对不能接受的。
• 分块艺术： 分块不仅仅是技术细节，更是对“上下文”的深刻理解。根据数据特征选择合适的分块策略至关重要。
• 嵌入模型的能力： 虽然该模型对性能影响一般不大，但是也是需要关注是否是他影响到知识库的检索，从而影响后续问答模型的生成回复质量。
• 检索与生成的平衡： RAG 的核心是检索增强生成。检索的“召回率”和“准确率”与生成的“忠实度”和“相关性”是相互影响的。重排序模型就是在召回和准确率之间寻求平衡的关键工具。
• **模型能力的优先考虑：**在用户输入一些问题回复不佳时，优先考虑换一下更好的模型，而不是优先考虑微调，例如可以考虑换一下32B 以上的模型就可以

4. RAG 与 LoRA 微调的协同效应思考：

这个项目设计中包含了 LoRA 微调的探索，这揭示了 RAG 并非 LLM 微调的替代品，而是互补的策略

• RAG 解决“知识鲜度”和“事实准确性”： 法律法规更新频繁，RAG 能够动态加载最新知识，避免模型“知识滞后”的问题。
• LoRA 解决“领域专业性”和“行为约束”： RAG 很难教会通用 LLM 以律师的思维方式、法律的严谨措辞来表达。LoRA 微调则能让 LLM 学习到法律领域的特定风格、术语使用、推理模式，以及在特定场景下的输出格式要求。例如，如何规范地引用法条、如何分析案例等。
• 效率与效果的平衡： RAG 避免了全量微调的巨大成本，而 LoRA 则在不牺牲过多效率的前提下，提升了 LLM 在特定领域的“软技能”。这种结合使得我们既能拥有最新的法律知识，又能确保助手回答的专业性和可信度。