当前位置： 首页 > news >正文

Python 构建法律DeepSeek RAG

news 来源：原创 2025/6/7 6:43:45

法律条文如图显示：
Typora 显示：
在这里插入图片描述
vscode显示：
在这里插入图片描述

1. 读取和分割法律文件

把整个法律文件的内容分割成一条一条的法律条文，并检验分割的数量是否正确。

def chinese_to_number(chinese_num):"""将中文数字（支持到万亿级别）转换为阿拉伯数字支持格式：一百二十三万四千五百六十七"""# 中文数字到阿拉伯数字的映射char_map = {'零': 0, '一': 1, '二': 2, '三': 3, '四': 4,'五': 5, '六': 6, '七': 7, '八': 8, '九': 9,'两': 2  # 特殊处理「两」}# 中文单位到乘数的映射unit_map = {'十': 10,'百': 100,'千': 1000,'万': 10000,'亿': 100000000}total = 0        # 最终结果current = 0      # 当前分段值（处理万/亿等大单位）temp = 0         # 临时存储当前数字# 遍历每个字符[3,4](@ref)for char in chinese_num:if char in char_map:# 遇到数字，存储到临时变量temp = char_map[char]elif char in unit_map:unit = unit_map[char]if unit >= 10000:# 处理万/亿等大单位[4](@ref)current = (current + temp) * unittotal += currentcurrent = 0temp = 0else:# 处理十/百/千等小单位if temp == 0:temp = 1  # 处理「十」的特殊情况：十万 → 10 * 10000current += temp * unittemp = 0# 添加最后未处理的数字[2](@ref)return total + current + tempdef split_legal_documents(text):# 匹配格式：**第X条** + 内容（支持多款）pattern = r"\*\*(第[\u4e00-\u9fa5]{2,}条)\*\*\s*([^#]+?)(?=\*\*第|$|###|####|#####)"articles = []# 用于检查重复和缺失的法条found_articles = []processed_articles = set()# 使用正则表达式查找所有匹配的条文matches = re.finditer(pattern, text, re.DOTALL)num = 0for match in matches:article_num = match.group(1)  # 条文编号（如"第二百三十四条"）content = match.group(2).strip()arabic_num = chinese_to_number(match.group(1).replace("第", "").replace("条", ""))# 记录找到的法条编号found_articles.append(arabic_num)# 检查是否已经处理过这个法条编号if article_num not in processed_articles:articles.append({"article_num": article_num,"content": content,"full_text": f"**{article_num}** {content}"})processed_articles.add(article_num)num += 1print(f"找到 {num} 条法条")    # 验证条文连续性if found_articles:sorted_nums = sorted(found_articles)min_num = sorted_nums[0]max_num = sorted_nums[-1]# print(f"条文连续性检查: "#       f"最小编号={min_num} | "#       f"最大编号={max_num}")expected_count = max_num - min_num + 1actual_count = len(set(found_articles))  # 去重后条数# 查找缺失条号full_range = set(range(min_num, max_num + 1))missing_nums = sorted(full_range - set(found_articles))if actual_count != expected_count or missing_nums:print(f"⚠️ 条文连续性警告: "f"实际分割条数={actual_count} | "f"预期连续条数={expected_count} | "f"缺失条号={missing_nums}")else:print(f"✅ 条文连续性验证: "f"实际分割条数={actual_count} | "f"预期连续条数={expected_count} | "f"缺失条号={missing_nums}")    return articleswith open("./mfd.md", "r") as file:file_text = file.read()
articles = split_legal_documents(file_text)
print("Total lines:", len(articles))

输出：

找到 386 条法条
✅ 条文连续性验证: 实际分割条数=386 | 预期连续条数=386 | 缺失条号=[]
Total lines: 386

2. 设置Milvus数据的配置

Milvus数据库使用时需要配置一些参数，以便存储不同类型数据，以及检索。

def setup_milvus_collection(collection_name,dimension,metric_type):"""创建并配置Milvus数据库"""# 1. 初始化Milvus客户端，连接到本地数据库文件milvus_client = MilvusClient(uri="./milvus_mfd.db")# 2. 检查并删除已存在的同名集合（避免冲突）if milvus_client.has_collection(collection_name):milvus_client.drop_collection(collection_name)# 3. 创建新的集合并进行详细配置milvus_client.create_collection(collection_name=collection_name,  # 集合名称dimension=dimension,  # 向量维度（如1024）metric_type=metric_type,  # 相似度计算方式（如COSINE）auto_id=True,  # 自动生成IDdescription="民法典条文向量库",  # 集合描述# 字段定义field_params=[{"name": "id", "type": DataType.INT64, "is_primary": True, "auto_id": True},  # 主键ID{"name": "vector", "type": DataType.FLOAT_VECTOR, "dim": 1024},  # 向量字段 VARCHAR 是 SQL 标准中的可变长度字符串类型{"name": "article_num", "type": DataType.VARCHAR, "max_length": 20},  # 条文编号{"name": "content", "type": DataType.VARCHAR, "max_length": 5000},  # 条文内容{"name": "full_text", "type": DataType.VARCHAR, "max_length": 5000}  # 完整文本],# 索引配置#先通过 IVF 快速定位到最近的几个聚类区域,然后在区域内用 FLAT 方式精确计算余弦相似度,返回最相似的结果index_params={"index_type": "IVF_FLAT",  # 使用IVF_FLAT索引类型（精确搜索）"metric_type": metric_type,  # 使用余弦相似度"params": {"nlist": 2048}  # 设置2048个聚类中心})return milvus_client  # 返回配置好的客户端实例milvus_client = setup_milvus_collection("PRCCivilCode", 1024, "COSINE")

3. 下载并加载嵌入模型

嵌入模型将text 转变成向量，嵌入模型一般在 huggingface 仓库里，使用先打开代理

MODEL_CHOICES = {"text2vec": "GanymedeNil/text2vec-large-chinese",  # 中文专用"bge": "BAAI/bge-large-zh-v1.5",  # 中文语义检索最佳"multilingual": "sentence-transformers/paraphrase-multilingual-mpnet-base-v2"  # 多语言支持}os.environ["HTTP_PROXY"] = "socks5h://localhost:1080"  # HTTP 代理os.environ["HTTPS_PROXY"] = "socks5h://localhost:1080"  # HTTPS 代理# 验证代理是否生效（可选）
try:import requestsprint("当前IP:", requests.get("https://ipinfo.io/json", timeout=5).json()['ip'])
except Exception as e:print("代理验证失败:", e)embedding_model = SentenceTransformer(MODEL_CHOICES["bge"])

4. 把数据插入向量库

def prepare_batch_data(batch, embedding_model):"""准备批量数据，包括生成嵌入向量"""contents = [art["content"] for art in batch]embeddings = embedding_model.encode(contents).tolist()return [{"vector": emb,"article_num": art["article_num"],"content": art["content"],"full_text": art["full_text"]} for emb, art in zip(embeddings, batch)]def process_and_insert_articles(articles, milvus_client, embedding_model, batch_size=64):"""处理文章并批量插入到Milvus"""total_articles = len(articles)print(f"开始处理 {total_articles} 条法律条文...")for i in range(0, total_articles, batch_size):batch = articles[i:i+batch_size]try:# 准备批量数据data = prepare_batch_data(batch, embedding_model)# 插入到Milvusres = milvus_client.insert("PRCCivilCode", data)print(f"进度: {min(i + batch_size, total_articles)}/{total_articles} | "f"批次 {i//batch_size}: 插入 {len(res['ids'])} 条记录")except Exception as e:print(f"⚠️ 批次 {i//batch_size} 插入失败: {str(e)}")continueprint(f"✅ 完成! 共处理 {total_articles} 条记录")process_and_insert_articles(articles, milvus_client, embedding_model)

5. 查询问题

查询的问题先变成向量，在向量数据库中查找与问题最相似的向量

def query_legal_question(question, milvus_client, embedding_model, limit=5):"""查询法律问题"""search_res = milvus_client.search(collection_name="PRCCivilCode",data=embedding_model.encode([question]),limit=limit,search_params={"metric_type": "COSINE", "params": {}},output_fields=["content", "article_num", "full_text"])return [(res["entity"]["content"], res["entity"]["article_num"], res["entity"]["full_text"], res["distance"]) for res in search_res[0]]question = "丢失动物,如何处理？"
retrieved_lines = query_legal_question(question, milvus_client, embedding_model)
context = "\n".join([line[0] for line in retrieved_lines])

6.生成回答

问题与搜索得到的答案一起作为prompt 发送给大模型。大模型整理后返回答案

def generate_legal_response(question, context):"""生成法律回答"""SYSTEM_PROMPT = """Human: 你是一个经验丰富的法律工作者。你能够从提供的上下文段落片段中找到问题的答案，并给出简洁明了的总结。如果你无法从上下文中找到答案，请明确说明。"""USER_PROMPT = f"""请使用以下用 <context> 标签括起来的信息片段来回答用 <question> 标签括起来的问题,并说清楚法律来源,具体是第几条。；最后追加原始回答的英文翻译，并用 <translated>和</translated> 标签标注。<context>{context}</context><question>{question}</question><translated></translated>"""api_key = os.getenv("DEEPSEEK_API_KEY")deepseek_client = OpenAI(api_key=api_key,base_url="https://api.siliconflow.cn/v1",)response = deepseek_client.chat.completions.create(model="Pro/deepseek-ai/DeepSeek-V3",messages=[{"role": "system", "content": SYSTEM_PROMPT},{"role": "user", "content": USER_PROMPT},],)return response.choices[0].message.contentanswer = generate_legal_response(question, context)
print(answer)

7. 完整代码

import os
import re
import json
from openai import OpenAI
from huggingface_hub import snapshot_download
from sentence_transformers import SentenceTransformer
from pymilvus import MilvusClient, DataTypedef chinese_to_number(chinese_num):"""将中文数字（支持到万亿级别）转换为阿拉伯数字支持格式：一百二十三万四千五百六十七"""# 中文数字到阿拉伯数字的映射char_map = {'零': 0, '一': 1, '二': 2, '三': 3, '四': 4,'五': 5, '六': 6, '七': 7, '八': 8, '九': 9,'两': 2  # 特殊处理「两」}# 中文单位到乘数的映射unit_map = {'十': 10,'百': 100,'千': 1000,'万': 10000,'亿': 100000000}total = 0        # 最终结果current = 0      # 当前分段值（处理万/亿等大单位）temp = 0         # 临时存储当前数字# 遍历每个字符[3,4](@ref)for char in chinese_num:if char in char_map:# 遇到数字，存储到临时变量temp = char_map[char]elif char in unit_map:unit = unit_map[char]if unit >= 10000:# 处理万/亿等大单位[4](@ref)current = (current + temp) * unittotal += currentcurrent = 0temp = 0else:# 处理十/百/千等小单位if temp == 0:temp = 1  # 处理「十」的特殊情况：十万 → 10 * 10000current += temp * unittemp = 0# 添加最后未处理的数字[2](@ref)return total + current + tempdef split_legal_documents(text):"""优化版法律条文分割函数：1. 精确分割独立条文（含多款条文）2. 保留层级结构信息3. 自动识别条文中的多款内容"""# 匹配格式：**第X条** + 内容（支持多款）pattern = r"\*\*(第[\u4e00-\u9fa5]{2,}条)\*\*\s*([^#]+?)(?=\*\*第|$|###|####|#####)"articles = []# 用于检查重复和缺失的法条found_articles = []processed_articles = set()# 使用正则表达式查找所有匹配的条文matches = re.finditer(pattern, text, re.DOTALL)num = 0for match in matches:article_num = match.group(1)  # 条文编号（如"第二百三十四条"）content = match.group(2).strip()arabic_num = chinese_to_number(match.group(1).replace("第", "").replace("条", ""))# 记录找到的法条编号found_articles.append(arabic_num)# 检查是否已经处理过这个法条编号if article_num not in processed_articles:articles.append({"article_num": article_num,"content": content,"full_text": f"**{article_num}** {content}"})processed_articles.add(article_num)num += 1print(f"找到 {num} 条法条")    # 验证条文连续性if found_articles:sorted_nums = sorted(found_articles)min_num = sorted_nums[0]max_num = sorted_nums[-1]# print(f"条文连续性检查: "#       f"最小编号={min_num} | "#       f"最大编号={max_num}")expected_count = max_num - min_num + 1actual_count = len(set(found_articles))  # 去重后条数# 查找缺失条号full_range = set(range(min_num, max_num + 1))missing_nums = sorted(full_range - set(found_articles))if actual_count != expected_count or missing_nums:print(f"⚠️ 条文连续性警告: "f"实际分割条数={actual_count} | "f"预期连续条数={expected_count} | "f"缺失条号={missing_nums}")else:print(f"✅ 条文连续性验证: "f"实际分割条数={actual_count} | "f"预期连续条数={expected_count} | "f"缺失条号={missing_nums}")    return articlesdef setup_milvus_collection(collection_name,dimension,metric_type):"""创建并配置Milvus集合"""# 1. 初始化Milvus客户端，连接到本地数据库文件milvus_client = MilvusClient(uri="./milvus_mfd.db")# 2. 检查并删除已存在的同名集合（避免冲突）if milvus_client.has_collection(collection_name):milvus_client.drop_collection(collection_name)# 3. 创建新的集合并进行详细配置milvus_client.create_collection(collection_name=collection_name,  # 集合名称dimension=dimension,  # 向量维度（如1024）metric_type=metric_type,  # 相似度计算方式（如COSINE）auto_id=True,  # 自动生成IDdescription="民法典条文向量库",  # 集合描述# 字段定义field_params=[{"name": "id", "type": DataType.INT64, "is_primary": True, "auto_id": True},  # 主键ID{"name": "vector", "type": DataType.FLOAT_VECTOR, "dim": 1024},  # 向量字段 VARCHAR 是 SQL 标准中的可变长度字符串类型{"name": "article_num", "type": DataType.VARCHAR, "max_length": 20},  # 条文编号{"name": "content", "type": DataType.VARCHAR, "max_length": 5000},  # 条文内容{"name": "full_text", "type": DataType.VARCHAR, "max_length": 5000}  # 完整文本],# 索引配置#先通过 IVF 快速定位到最近的几个聚类区域,然后在区域内用 FLAT 方式精确计算余弦相似度,返回最相似的结果index_params={"index_type": "IVF_FLAT",  # 使用IVF_FLAT索引类型（精确搜索）"metric_type": metric_type,  # 使用余弦相似度"params": {"nlist": 2048}  # 设置2048个聚类中心})return milvus_client  # 返回配置好的客户端实例
def prepare_batch_data(batch, embedding_model):"""准备批量数据，包括生成嵌入向量"""contents = [art["content"] for art in batch]embeddings = embedding_model.encode(contents).tolist()return [{"vector": emb,"article_num": art["article_num"],"content": art["content"],"full_text": art["full_text"]} for emb, art in zip(embeddings, batch)]def process_and_insert_articles(articles, milvus_client, embedding_model, batch_size=64):"""处理文章并批量插入到Milvus"""total_articles = len(articles)print(f"开始处理 {total_articles} 条法律条文...")for i in range(0, total_articles, batch_size):batch = articles[i:i+batch_size]try:# 准备批量数据data = prepare_batch_data(batch, embedding_model)# 插入到Milvusres = milvus_client.insert("PRCCivilCode", data)print(f"进度: {min(i + batch_size, total_articles)}/{total_articles} | "f"批次 {i//batch_size}: 插入 {len(res['ids'])} 条记录")except Exception as e:print(f"⚠️ 批次 {i//batch_size} 插入失败: {str(e)}")continueprint(f"✅ 完成! 共处理 {total_articles} 条记录")def query_legal_question(question, milvus_client, embedding_model, limit=5):"""查询法律问题"""search_res = milvus_client.search(collection_name="PRCCivilCode",data=embedding_model.encode([question]),limit=limit,search_params={"metric_type": "COSINE", "params": {}},output_fields=["content", "article_num", "full_text"])return [(res["entity"]["content"], res["entity"]["article_num"], res["entity"]["full_text"], res["distance"]) for res in search_res[0]]def generate_legal_response(question, context):"""生成法律回答"""SYSTEM_PROMPT = """Human: 你是一个经验丰富的法律工作者。你能够从提供的上下文段落片段中找到问题的答案，并给出简洁明了的总结。如果你无法从上下文中找到答案，请明确说明。"""USER_PROMPT = f"""请使用以下用 <context> 标签括起来的信息片段来回答用 <question> 标签括起来的问题,并说清楚法律来源,具体是第几条。；最后追加原始回答的英文翻译，并用 <translated>和</translated> 标签标注。<context>{context}</context><question>{question}</question><translated></translated>"""api_key = os.getenv("DEEPSEEK_API_KEY")deepseek_client = OpenAI(api_key=api_key,base_url="https://api.siliconflow.cn/v1",)response = deepseek_client.chat.completions.create(model="Pro/deepseek-ai/DeepSeek-V3",messages=[{"role": "system", "content": SYSTEM_PROMPT},{"role": "user", "content": USER_PROMPT},],)return response.choices[0].message.content# 主流程
if __name__ == "__main__":# 1. 读取和分割法律文件with open("./mfd.md", "r") as file:file_text = file.read()articles = split_legal_documents(file_text)# 2. 设置Milvus集合milvus_client = setup_milvus_collection("PRCCivilCode", 1024, "COSINE")# 3. 加载模型MODEL_CHOICES = {"text2vec": "GanymedeNil/text2vec-large-chinese",  # 中文专用"bge": "BAAI/bge-large-zh-v1.5",  # 中文语义检索最佳"multilingual": "sentence-transformers/paraphrase-multilingual-mpnet-base-v2"  # 多语言支持}os.environ["HTTP_PROXY"] = "socks5h://localhost:1080"  # HTTP 代理os.environ["HTTPS_PROXY"] = "socks5h://localhost:1080"  # HTTPS 代理# 验证代理是否生效（可选）try:import requestsprint("当前IP:", requests.get("https://ipinfo.io/json", timeout=5).json()['ip'])except Exception as e:print("代理验证失败:", e)embedding_model = SentenceTransformer(MODEL_CHOICES["bge"])# 4. 处理并插入文章process_and_insert_articles(articles, milvus_client, embedding_model)# 5. 查询示例question = "丢失动物,如何处理？"retrieved_lines = query_legal_question(question, milvus_client, embedding_model)context = "\n".join([line[0] for line in retrieved_lines])# 6. 生成回答answer = generate_legal_response(question, context)print(answer)

相关文章：

  • YOLO训练保持原有识别能力
  • Spring 团队详解：AOT 缓存实践、JSpecify 空指针安全与支持策略升级
  • 宝塔think PHP8 安装使用FFmpeg 视频上传
  • 数据结构之常用排序算法（冒泡、选择等）
  • 以STM32H7微控制器为例，简要说明stm32h7xx_it.c的作用
  • 主流 AI IDE 之一的 Cursor 介绍
  • mysql+keepalived
  • C#中的CLR属性、依赖属性与附加属性
  • 《零基础读懂新能源汽车》—— 新能源汽车充电革命：从逆变器原理到800V超充实战，一篇全掌握！
  • Python训练营打卡Day46(2025.6.6）
  • 【React】React 18 并发特性
  • 我爱学算法之—— 前缀和（中）
  • Spring Cloud 2025.0.0 Gateway迁移全过程详解
  • 文件上传漏洞深度解析：检测与绕过技术矩阵
  • Ubuntu创建修改 Swap 文件分区的步骤——解决嵌入式开发板编译ROS2程序卡死问题
  • 力扣100-移动0
  • 学习STC51单片机29（芯片为STC89C52RCRC）
  • 6.6 day38
  • 2025年6月6日15:48:23
  • 数字孪生：解锁未来的“数字钥匙”
  • 东港区网站制作/互换链接的方法
  • wordpress菜单子页面/seo门户网价格是多少钱
  • 医疗保健 网站 备案/东莞网络排名优化
  • python不用框架做动态网站/seo外包大型公司
  • 深圳网站制作工作室/免费加精准客源
  • 网站公司不给ftp/it培训班出来现状