Python 构建法律DeepSeek RAG

法律条文如图显示：
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) * unit
                total += current
                current = 0
                temp = 0
            else:
                # 处理十/百/千等小单位
                if temp == 0:
                    temp = 1  # 处理「十」的特殊情况：十万 → 10 * 10000
                current += temp * unit
                temp = 0
    
    # 添加最后未处理的数字[2](@ref)
    return total + current + temp

def 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 = 0
    for 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 += 1
            
    print(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 + 1
        actual_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 articles

with 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,  # 自动生成ID
        description="民法典条文向量库",  # 集合描述
        # 字段定义
        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 requests
     print("当前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)
            
            # 插入到Milvus
            res = 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)}")
            continue
    
    print(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.content

answer = 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, DataType


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) * unit
                total += current
                current = 0
                temp = 0
            else:
                # 处理十/百/千等小单位
                if temp == 0:
                    temp = 1  # 处理「十」的特殊情况：十万 → 10 * 10000
                current += temp * unit
                temp = 0
    
    # 添加最后未处理的数字[2](@ref)
    return total + current + temp

def 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 = 0
    for 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 += 1
            
    print(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 + 1
        actual_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 articles

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,  # 自动生成ID
        description="民法典条文向量库",  # 集合描述
        
        # 字段定义
        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)
            
            # 插入到Milvus
            res = 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)}")
            continue
    
    print(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 requests
        print("当前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)