某中型法律咨询公司（以下简称"L律所"）主营业务包括企业法律顾问、劳动纠纷、合同审核等。公司有3名资深律师和8名律师助理，每天需要处理大量法律文书：
在AI知识库上线前，律师助理的工作流程是：
核心痛点总结：
| 指标 | 现状 | 目标 |
|---|---|---|
| 单次咨询准备时间 | 45分钟 | ≤10分钟 |
| 检索准确率 | （人工，无法量化） | ≥90% |
| 相关判例召回率 | 依赖经验，约60% | ≥85% |
| 法规引用准确率 | 100%（人工） | 100% |
| 系统可用时间 | - | 7×24小时 |
┌─────────────────────────────────────────────────────────────────────┐
│ 用户端（律师/助理） │
│ 浏览器 + 企业内部系统 │
└────────────────────────────┬──────────────────────────────────────┘
│ HTTPS
▼
┌─────────────────────────────────────────────────────────────────────┐
│ Dify（RAG应用编排平台） │
│ 负责：对话界面、检索流程编排、LLM调用、结果生成 │
│ 访问地址：http://内网IP:5001 │
└────────────┬──────────────────────────┬───────────────────────────────┘
│ │
▼ ▼
┌──────────────────────┐ ┌──────────────────────┐
│ Ollama（LLM推理） │ │ Milvus（向量数据库） │
│ qwen2.5-14b-instruct│ │ 5000+份文档向量 │
│ 处理检索结果 │ │ 检索相关文档片段 │
│ 生成回答 │ │ │
└──────────────────────┘ └──────────────────────┘
▲ ▲
│ │
▼ ▼
┌──────────────────────┐ ┌──────────────────────┐
│ GPU服务器 │ │ 文档处理服务 │
│ NVIDIA A4000 16GB │ │ PDF解析/分块/Embedding│
│ ollama serve │ │ m3e-base模型 │
└──────────────────────┘ └──────────────────────┘在技术选型时，我们对比了Milvus和Qdrant，最终选择Milvus的原因是：
# Step 1: 建立文档目录结构
mkdir -p /data/legal_kb/{raw_documents,processed,chunks}
mkdir -p /data/legal_kb/raw_documents/{judgments,regulations,contracts}
# Step 2: 使用Python脚本批量解析PDF文档
# 安装依赖：pip install pymupdf python-docx
import os
import pymupdf # fitz - 用于解析PDF
from docx import Document
from typing import List
def extract_pdf_text(pdf_path: str) -> str:
"""
从PDF中提取文本内容
处理嵌套表格、多栏布局等复杂PDF
"""
text_parts = []
with pymupdf.open(pdf_path) as doc:
for page_num, page in enumerate(doc):
# 获取页面文本（保留基本结构）
text = page.get_text("text")
# 过滤掉页眉页脚（法律文书通常有固定的页眉格式）
lines = text.split('\n')
filtered_lines = []
for line in lines:
# 过滤空行和页码
if line.strip() and not line.strip().isdigit():
filtered_lines.append(line.strip())
page_text = '\n'.join(filtered_lines)
text_parts.append(page_text)
return '\n\n'.join(text_parts)
def extract_docx_text(docx_path: str) -> str:
"""从Word文档中提取文本"""
doc = Document(docx_path)
paragraphs = [p.text.strip() for p in doc.paragraphs if p.text.strip()]
return '\n\n'.join(paragraphs)
# Step 3: 批量处理所有文档
def process_all_documents(source_dir: str, output_dir: str) -> int:
"""遍历目录，处理所有PDF和DOCX文件"""
processed_count = 0
for root, dirs, files in os.walk(source_dir):
for filename in files:
filepath = os.path.join(root, filename)
try:
if filename.lower().endswith('.pdf'):
text = extract_pdf_text(filepath)
elif filename.lower().endswith(('.docx', '.doc')):
text = extract_docx_text(filepath)
else:
continue
# 保存处理后的文本文件
safe_name = filename.replace('.pdf', '.txt').replace('.docx', '.txt').replace('.doc', '.txt')
output_path = os.path.join(output_dir, safe_name)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(f"# 源文件：{filename}\n\n")
f.write(text)
processed_count += 1
print(f"✓ 已处理：{filename}")
except Exception as e:
print(f"✗ 处理失败：{filename} - {str(e)}")
return processed_count
# 执行处理
count = process_all_documents('/data/legal_kb/raw_documents', '/data/legal_kb/processed')
print(f"\n总计处理文档：{count}份")文档分块（Chunking）是RAG系统中最重要的环节之一。分块太小会丢失上下文，分块太大会引入过多噪声。
# 分块策略：使用递归字符分割 + 法律文书优化
import re
from typing import List, Dict
class LegalDocumentChunker:
"""
法律文书专用分块器
策略：
1. 先按章节/段落分割（法律文书有固定的章节结构）
2. 如果章节过长，再按固定长度分割
3. 保留章节标题作为元数据
"""
def __init__(self,
chunk_size: int = 512, # 每块目标token数（按中文估算）
chunk_overlap: int = 64, # 相邻块的重叠token数
min_chunk_size: int = 128): # 最小块大小
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.min_chunk_size = min_chunk_size
def split_by_legal_sections(self, text: str) -> List[Dict[str, str]]:
"""
按法律文书的标准结构分割
法律文书通常包含：案件基本信息、事实陈述、法院认定、判决结果
"""
sections = []
# 法律文书常见的章节标题模式
section_patterns = [
r'^第[一二三四五六七八九十百]+条', # 法律条文：第X条
r'^（[一二三四五六七八九十百]+）', # 条款：（一）（二）
r'^【[^】]+】', # 方框标题：【案件基本信息】
r'^一[\u4e00-\u9fa5]{1,10}$', # 章节：一、关于争议焦点
r'^（案由[^）]+）', # 案由：（案由机动车交通事故责任纠纷）
]
combined_pattern = '|'.join(section_patterns)
lines = text.split('\n')
current_section = ""
current_content = []
for line in lines:
# 检查是否是新的章节标题
if re.match(combined_pattern, line.strip()):
# 保存上一个章节
if current_content:
sections.append({
"title": current_section or "开头",
"content": '\n'.join(current_content)
})
current_section = line.strip()
current_content = []
else:
if line.strip():
current_content.append(line.strip())
# 保存最后一个章节
if current_content:
sections.append({
"title": current_section or "结尾",
"content": '\n'.join(current_content)
})
return sections
def chunk_text(self, text: str, metadata: dict) -> List[Dict[str, any]]:
"""
对文本进行分块
Args:
text: 原始文本
metadata: 元数据（如文件名、文档类型等）
Returns:
分块列表，每个块包含 text, chunk_id, metadata
"""
chunks = []
chunk_id = 0
# 尝试按章节分割
sections = self.split_by_legal_sections(text)
for section in sections:
section_text = section["content"]
section_title = section["title"]
# 如果章节文本已经不长，直接作为chunk
if len(section_text) <= self.chunk_size * 2:
chunks.append({
"chunk_id": f"chunk_{chunk_id}",
"text": f"【{section_title}】\n{section_text}",
"metadata": {
**metadata,
"section": section_title,
"source": "section_split"
}
})
chunk_id += 1
continue
# 章节过长，按段落继续分割
paragraphs = [p.strip() for p in section_text.split('\n\n') if p.strip()]
current_chunk = ""
current_tokens = 0
for para in paragraphs:
para_tokens = len(para) // 2 # 粗略估算中文字符token数
# 如果加上这个段落会超过chunk大小，保存当前chunk并开始新的
if current_tokens + para_tokens > self.chunk_size:
if current_tokens >= self.min_chunk_size:
chunks.append({
"chunk_id": f"chunk_{chunk_id}",
"text": current_chunk,
"metadata": {
**metadata,
"section": section_title,
"source": "recursive_split"
}
})
chunk_id += 1
# 重叠部分：从上一个chunk末尾取一些内容
overlap_text = current_chunk[-self.chunk_overlap * 2:]
current_chunk = overlap_text
current_tokens = len(overlap_text) // 2
current_chunk += '\n' + para
current_tokens += para_tokens
# 保存最后一个chunk
if current_tokens >= self.min_chunk_size:
chunks.append({
"chunk_id": f"chunk_{chunk_id}",
"text": current_chunk,
"metadata": {
**metadata,
"section": section_title,
"source": "final_chunk"
}
})
return chunks
# 使用分块器
chunker = LegalDocumentChunker(chunk_size=512, chunk_overlap=64)
all_chunks = []
for filename in os.listdir('/data/legal_kb/processed'):
if not filename.endswith('.txt'):
continue
with open(f'/data/legal_kb/processed/{filename}', 'r', encoding='utf-8') as f:
content = f.read()
# 从第一行提取原始文件名（格式：# 源文件：xxx.pdf）
lines = content.split('\n', 1)
raw_filename = lines[0].replace('# 源文件：', '').strip()
text = lines[1] if len(lines) > 1 else ""
# 确定文档类型
if '裁判' in raw_filename or '案' in raw_filename:
doc_type = 'judgments'
elif '法规' in raw_filename or '法律' in raw_filename:
doc_type = 'regulations'
else:
doc_type = 'contracts'
metadata = {
"source_file": raw_filename,
"doc_type": doc_type,
"filename": filename
}
chunks = chunker.chunk_text(text, metadata)
all_chunks.extend(chunks)
print(f"✓ {filename} → 生成了 {len(chunks)} 个chunk")
print(f"\n总计生成chunk：{len(all_chunks)} 个")# 使用Ollama的Embedding模型生成向量
import requests
import json
def get_embedding(text: str, model: str = "m3e-base") -> list:
"""
调用Ollama API获取文本的Embedding向量
Args:
text: 待向量化的文本
model: Embedding模型名称（默认使用m3e-base）
Returns:
向量列表（1024维）
"""
try:
response = requests.post(
"http://localhost:11434/api/embeddings",
json={"model": model, "prompt": text},
timeout=30
)
result = response.json()
return result["embedding"]
except Exception as e:
print(f"Embedding生成失败：{text[:50]}... - {e}")
return [0.0] * 1024 # 返回全零向量作为fallback
# 批量生成Embedding（带进度显示）
def generate_embeddings_batch(chunks: List[Dict], batch_size: int = 32) -> List[Dict]:
"""批量生成Embedding"""
embeddings = []
total = len(chunks)
for i in range(0, total, batch_size):
batch = chunks[i:i+batch_size]
batch_embeddings = []
for chunk in batch:
# 控制文本长度，超过4000字符的截断（m3e-base有输入长度限制）
text = chunk["text"][:4000]
embedding = get_embedding(text)
batch_embeddings.append({
"embedding": embedding,
"text": chunk["text"],
"metadata": chunk["metadata"],
"chunk_id": chunk["chunk_id"]
})
embeddings.extend(batch_embeddings)
progress = min(i + batch_size, total)
print(f" 进度：{progress}/{total} ({100*progress//total}%)")
return embeddings
print("开始生成Embedding...")
chunks_with_embeddings = generate_embeddings_batch(all_chunks)
print(f"✓ Embedding生成完成，共{len(chunks_with_embeddings)}条")然后将Embedding导入Milvus：
# 安装Milvus Python客户端：pip install pymilvus
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
# ============ 连接Milvus ============
MILVUS_HOST = "localhost"
MILVUS_PORT = "19530"
connections.connect(
alias="default",
host=MILVUS_HOST,
port=MILVUS_PORT,
timeout=30
)
print("✓ Milvus连接成功")
# ============ 创建Collection ============
COLLECTION_NAME = "legal_knowledge_base"
# 先检查是否存在，存在则删除重建
if utility.has_collection(COLLECTION_NAME):
utility.drop_collection(COLLECTION_NAME)
print(f"已删除旧Collection：{COLLECTION_NAME}")
# 定义Collection Schema
# 向量字段：1024维（对应m3e-base的输出）
fields = [
FieldSchema(name="chunk_id", dtype=DataType.VARCHAR, max_length=64, is_primary=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=8192),
FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=32),
FieldSchema(name="source_file", dtype=DataType.VARCHAR, max_length=256),
FieldSchema(name="section", dtype=DataType.VARCHAR, max_length=256),
]
schema = CollectionSchema(fields=fields, description="法律知识库RAG系统")
collection = Collection(name=COLLECTION_NAME, schema=schema)
# 创建索引（HNSW索引，检索精度高）
index_params = {
"metric_type": "COSINE", # 余弦相似度
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 128}
}
collection.create_index(
field_name="embedding",
index_params=index_params
)
# 创建分区（按文档类型分区，方便过滤检索）
collection.create_partition("judgments")
collection.create_partition("regulations")
collection.create_partition("contracts")
print("✓ Collection和分区创建成功")
# ============ 批量导入数据 ============
from pymilvus import Partition
# 分批次导入（Milvus单次导入有数量限制）
BATCH_SIZE = 1000
def import_to_milvus(chunks_with_embeddings: List[Dict], collection_name: str):
"""批量导入向量到Milvus"""
collection = Collection(collection_name)
collection.load() # 先加载collection，使其可用于检索
total = len(chunks_with_embeddings)
for i in range(0, total, BATCH_SIZE):
batch = chunks_with_embeddings[i:i+BATCH_SIZE]
entities = [
[chunk["chunk_id"] for chunk in batch],
[chunk["embedding"] for chunk in batch],
[chunk["text"] for chunk in batch],
[chunk["metadata"].get("doc_type", "unknown") for chunk in batch],
[chunk["metadata"].get("source_file", "") for chunk in batch],
[chunk["metadata"].get("section", "") for chunk in batch],
]
# 根据doc_type选择分区
first_doc_type = batch[0]["metadata"].get("doc_type", "unknown")
try:
partition = Partition(collection, first_doc_type)
partition.insert(entities)
except Exception:
# 如果分区不存在，插入到默认分区
mr = collection.insert(entities)
progress = min(i + BATCH_SIZE, total)
print(f" 导入进度：{progress}/{total}")
# 刷新，使数据立即可检索
collection.flush()
print(f"✓ 导入完成，共{total}条向量")
print("开始导入Milvus...")
import_to_milvus(chunks_with_embeddings, COLLECTION_NAME)
# 验证数据量
collection = Collection(COLLECTION_NAME)
print(f"✓ Collection统计：{collection.num_entities} 条向量")# 在Dify中通过API配置知识库和应用
DIFY_HOST = "http://localhost:5001"
# Step 1: 创建数据集（知识库）
def create_dataset(name: str, description: str):
response = requests.post(
f"{DIFY_HOST}/api/v1/datasets",
headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
json={
"name": name,
"description": description,
"indexing_technique": "high_quality",
"embedding_model": "m3e-base",
"permission": "only_me"
}
)
return response.json()
dataset = create_dataset(
name="法律知识库",
description="L律所法律文书知识库，包含判例、法规、合同模板"
)
dataset_id = dataset["id"]
print(f"✓ 数据集创建成功，ID：{dataset_id}")
# Step 2: 配置向量数据库（使用我们搭建的Milvus）
# Dify默认支持多种向量数据库，通过环境变量配置
# 由于Dify对Milvus支持有限，我们使用其内置向量库功能
# 这里记录如何通过API配置外部向量库（以Qdrant为例）
# Step 3: 创建RAG应用
def create_rag_app(name: str, dataset_id: str):
response = requests.post(
f"{DIFY_HOST}/api/v1/apps",
headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
json={
"name": name,
"description": "基于法律知识库的智能问答助手",
"app_mode": "chatbot",
"icon": "⚖️",
"model_config": {
"provider": "ollama",
"model_id": "qwen2.5-14b-instruct",
"temperature": 0.3, # 法律场景建议低temperature，减少幻觉
"max_tokens": 2048,
"top_p": 0.95
},
"dataset_configs": {
"datasets": [
{
"dataset_id": dataset_id,
"rules": {
"top_k": 8, # 召回8个相关片段
"score_threshold": 0.65, # 相似度阈值
"reranking_model": None
}
}
],
"retrieval_method": "semantic_search",
"reranking_mode": "Reranking"
}
}
)
return response.json()
app = create_rag_app("法律问答助手", dataset_id)
app_id = app["id"]
print(f"✓ 应用创建成功，ID：{app_id}")# 问题描述：
# 部分裁判文书PDF解析后出现大量乱码，如"民初字第1号"变成"呡初字第1呡"
# 原因是PDF使用了特殊字体，PyMuPDF无法正确识别
# 原因分析：
# 法律文书常用"方正小标宋"、"仿宋"等特殊字体
# 这些字体在PDF中以Type0/CJK字形形式嵌入
# PyMuPDF的get_text()默认无法正确解析这些字形
# 解决方案1：使用pdfplumber（对表格支持更好）
import pdfplumber
def extract_with_pdfplumber(pdf_path: str) -> str:
"""使用pdfplumber提取文本，对CJK字体有特殊处理"""
text_parts = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
# try_vert=True 会尝试提取垂直文字
page_text = page.extract_text(x_tolerance=3, y_tolerance=3)
if page_text:
text_parts.append(page_text)
return '\n\n'.join(text_parts)
# 解决方案2：如果仍然乱码，使用OCR（最可靠但最慢）
# from PIL import Image
# import pytesseract
#
# def extract_with_ocr(pdf_path: str) -> str:
# """使用Tesseract OCR提取文本"""
# with pymupdf.open(pdf_path) as doc:
# for page in doc:
# # 将页面渲染为高分辨率图像
# mat = pymupdf.Matrix(3, 3) # 3x分辨率，提升OCR准确率
# pix = page.get_pixmap(matrix=mat)
# img = Image.open(io.BytesIO(pix.tobytes("png")))
# text = pytesseract.image_to_string(img, lang='chi_sim+eng')
# text_parts.append(text)
# return '\n\n'.join(text_parts)# 问题描述：
# 客户问："公司要开除一个怀孕的女员工合法吗？"
# 但判例中用的是："用人单位违法解除孕期女职工劳动合同"
# 这两句话语义相同但词汇差异巨大，Embedding相似度很低
# 解决方案：构建法律术语同义词扩展表
LEGAL_TERM_EXPANSIONS = {
# 客户常用词 → 法律术语
"开除": ["解除劳动合同", "解除劳动关系", "违法辞退", "单方解除"],
"怀孕": ["孕期", "妊娠期", "三期", "孕期女职工", "孕期劳动者"],
"员工": ["劳动者", "职工", "雇员", "用人单元员工"],
"公司": ["用人单位", "用人单位", "企业"],
"合法": ["符合法律规定", "具有法律依据", "依法解除", "符合劳动合同法"],
"工资": ["劳动报酬", "薪资", "薪酬"],
"赔偿": ["经济补偿", "赔偿金", "损害赔偿", "违约金"],
"合同": ["劳动合同", "劳动关系", "劳动合同期限"],
"加班": ["延长工作时间", "加班加点", "超时工作"],
"退休": ["退休", "基本养老保险", "养老金"],
}
def expand_query_with_legal_terms(query: str) -> str:
"""
将客户查询扩展为包含法律术语的形式
这是改进检索质量的关键技巧
"""
expanded = query
for common_term, legal_terms in LEGAL_TERM_EXPANSIONS.items():
if common_term in query:
# 用OR逻辑替换
pattern = common_term
replacement = f"({common_term} OR {' OR '.join(legal_terms)})"
# 这里只是记录，实际使用时需要用查询重写(Query Rewriting)技术
print(f"扩展：{pattern} → {replacement}")
return expanded
# 在检索前，对查询进行重写
def rewrite_legal_query(query: str) -> str:
"""
使用LLM将客户口语转换为法律专业表述
"""
rewrite_prompt = f"""
你是一个法律专家。请将用户的问题改写为法律专业表述。
要求：
1. 保留原问题的核心法律问题
2. 将口语化词汇替换为法律专业术语
3. 改写后的查询要能被法律文书检索系统正确匹配
用户问题：{query}
改写后的查询（直接输出查询语句，不要解释）：
"""
response = requests.post(
"http://localhost:11434/api/generate",
json={
"model": "qwen2.5-14b-instruct",
"prompt": rewrite_prompt,
"stream": False
}
)
rewritten = response.json()["response"].strip()
return rewritten
# 测试
query = "公司要开除一个怀孕的女员工合法吗？"
rewritten = rewrite_legal_query(query)
print(f"原始查询：{query}")
print(f"改写后：{rewritten}")# 问题描述：初期检索准确率只有65%左右，很多查询召回的文档不相关
# 根因分析：我们做了以下优化
# ========== 优化1：使用重排（Reranking）模型 ==========
# 初始方案：直接用向量检索top_k=5
# 优化后：先用向量检索top_k=20，然后用重排模型重新排序，选top_k=5
# 安装重排模型（需要Ollama支持）
# ollama pull bge-reranker-large
def rerank_search(query: str, search_results: list, top_k: int = 5) -> list:
"""
使用重排模型对检索结果进行二次排序
Args:
query: 用户查询
search_results: 初始向量检索结果 [(metadata, score), ...]
top_k: 最终返回的数量
Returns:
重排后的top_k个结果
"""
# 调用LLM评估每个结果与查询的相关性
rerank_prompt = f"""
请评估以下法律文档片段与用户问题的相关性。
用户问题：{query}
文档内容：
{search_results[0][0]['text'] if search_results else '无'}
请输出1-10的相关性评分（1=完全不相关，10=完全相关），直接输出数字：
"""
reranked = []
for metadata, original_score in search_results:
# 生成重排评分（这里用简化的LLM评分，实际可用bge-reranker-large）
doc_text = metadata['text'][:500] # 取前500字符
# 这里简化为使用向量相似度的扩展，结合LLM评分
# 实际项目中建议使用专门的Reranking模型
reranked.append({
"metadata": metadata,
"original_score": original_score,
# 综合评分 = 0.3 * 原始相似度 + 0.7 * LLM相关性评分
"final_score": 0.3 * original_score + 0.7 * (original_score * 10)
})
# 按最终评分排序
reranked.sort(key=lambda x: x["final_score"], reverse=True)
return reranked[:top_k]
# ========== 优化2：多路召回（Hybrid Retrieval）============
# 不要只依赖向量检索，使用多路召回增加召回率
def hybrid_search(query: str, collection, top_k: int = 10) -> list:
"""
混合检索：向量检索 + 关键词检索 + 同义词扩展检索
三路召回后取Union，然后统一重排
"""
results = {}
# 1. 向量检索
query_embedding = get_embedding(query)
vector_results = collection.search(
data=[query_embedding],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"ef": 128}},
limit=top_k * 2,
output_fields=["text", "doc_type", "source_file"]
)
for result in vector_results[0]:
doc_id = result.id
if doc_id not in results:
results[doc_id] = {
"metadata": result.payload,
"vector_score": result.score,
"keyword_score": 0.0
}
# 2. 关键词检索（通过元数据过滤实现）
# 法律文书使用精确关键词检索很重要
keywords = extract_legal_keywords(query)
for keyword in keywords:
filtered = collection.query(
expr=f'source_file like "%{keyword}%"',
output_fields=["text", "doc_type", "source_file"]
)
for item in filtered:
doc_id = item["chunk_id"]
if doc_id in results:
results[doc_id]["keyword_score"] += 1.0 # 每命中一个关键词加1分
else:
results[doc_id] = {
"metadata": item,
"vector_score": 0.0,
"keyword_score": 1.0
}
# 3. 综合评分排序
combined_results = []
for doc_id, scores in results.items():
combined_score = 0.7 * scores["vector_score"] + 0.3 * (scores["keyword_score"] / max(len(keywords), 1))
combined_results.append(({"chunk_id": doc_id, **scores["metadata"]}, combined_score))
combined_results.sort(key=lambda x: x[1], reverse=True)
return combined_results[:top_k]
def extract_legal_keywords(query: str) -> list:
"""从查询中提取法律关键词"""
# 简单实现：基于我们的术语表提取
all_terms = []
for terms in LEGAL_TERM_EXPANSIONS.values():
all_terms.extend(terms)
all_terms.extend(list(LEGAL_TERM_EXPANSIONS.keys()))
keywords = [term for term in all_terms if term in query]
return keywords
# 应用优化后，准确率从65%提升到90%以上# 问题描述：
# LLM生成的回答中出现了不存在的法律条文，如"根据《劳动法》第38条规定..."
# 实际上《劳动法》没有第38条，或者条文内容与真实不符
# 解决方案：强制使用RAG模式，并在Prompt中做严格约束
STRICT_RAG_PROMPT = """
你是一个法律咨询助手。请严格根据提供的法律文书内容回答问题。
【核心规则】
1. 只能基于检索到的法律文书内容进行回答
2. 引用法律条文时，必须同时说明：
- 条文出自哪个文件（第X条）
- 条文的具体内容是什么
- 该条文在本案/本文中的适用结论
3. 如果检索到的内容无法回答问题，必须明确说明：
"根据现有的法律文书，我无法找到直接相关的依据。建议您查阅XXX法规或咨询专业律师。"
4. 禁止编造法律条文编号、内容或案例细节
5. 所有法律依据必须附带来源文件名称
【输出格式】
回答应包含以下部分：
1. 【问题分析】- 简述问题的法律性质
2. 【相关依据】- 引用的法律条文（必须来自检索到的文档）
3. 【实务建议】- 基于法律的实际操作建议
4. 【风险提示】- 可能存在的法律风险
请开始回答：
"""
def generate_strict_answer(query: str, retrieved_docs: list) -> str:
"""
使用严格RAG模式生成回答
确保LLM不会凭空编造法律条文
"""
# 构建上下文：将检索到的文档内容拼接
context_parts = []
for i, doc in enumerate(retrieved_docs, 1):
context_parts.append(f"【文档{i}】\n来源：{doc['metadata']['source_file']}\n内容：{doc['metadata']['text'][:800]}")
context = "\n\n".join(context_parts)
full_prompt = f"""【上下文】
{context}
【用户问题】
{query}
{STRICT_RAG_PROMPT}"""
response = requests.post(
"http://localhost:11434/api/generate",
json={
"model": "qwen2.5-14b-instruct",
"prompt": full_prompt,
"stream": False,
"options": {
"temperature": 0.3, # 低温减少幻觉
"top_k": 20,
"repeat_penalty": 1.1 # 轻微惩罚重复，增加多样性
}
}
)
return response.json()["response"]# 问题描述：
# 某些判例文书很长（如100页），分块后丢失了案件的整体逻辑
# 导致检索到的片段是孤立的，缺乏上下文背景
# 解决方案1：保留足够大的重叠（overlap）
# 我们将overlap从32 tokens提升到64 tokens，相邻块有50%重叠
# 解决方案2：为每个chunk添加元数据关联
# 在chunk中添加：所属判例的完整案件名称、案号、审理法院
def add_contextual_metadata(chunk: dict, doc_metadata: dict) -> dict:
"""为chunk添加上下文元数据"""
enhanced_chunk = chunk.copy()
# 添加文档级别的元数据
enhanced_chunk["metadata"]["case_name"] = doc_metadata.get("case_name", "")
enhanced_chunk["metadata"]["case_number"] = doc_metadata.get("case_number", "")
enhanced_chunk["metadata"]["court"] = doc_metadata.get("court", "")
enhanced_chunk["metadata"]["case_date"] = doc_metadata.get("case_date", "")
# 在text开头添加上下文说明
context_header = f"""【案件】{doc_metadata.get('case_name', '未知')}
【案号】{doc_metadata.get('case_number', '未知')}
【审理法院】{doc_metadata.get('court', '未知')}
【案件日期】{doc_metadata.get('case_date', '未知')}
---
"""
enhanced_chunk["text"] = context_header + chunk["text"]
return enhanced_chunk
# 解决方案3：在检索结果中补充完整案例
# 当某个chunk被命中时，自动补充同一案件的其他相关chunk
def supplement_case_chunks(initial_results: list, collection, max_chunks_per_case: int = 3) -> list:
"""补充同一案件的其他相关chunk"""
case_chunks = {} # case_name → [chunks]
for result in initial_results:
case_name = result["metadata"].get("case_name", "")
if case_name and len(case_chunks.get(case_name, [])) < max_chunks_per_case:
if case_name not in case_chunks:
# 查询同一案件的其他chunk
additional = collection.query(
expr=f'case_name == "{case_name}"',
limit=max_chunks_per_case,
output_fields=["text", "doc_type", "section"]
)
case_chunks[case_name] = additional
# 去重后补充
existing_ids = {r["metadata"].get("chunk_id") for r in initial_results}
for chunk in case_chunks[case_name]:
if chunk["chunk_id"] not in existing_ids:
initial_results.append({
"metadata": chunk,
"source": "case_supplement",
"relevance": result.get("relevance", 0.8) * 0.95
})
return initial_results# 问题描述：初期系统响应慢，最慢时超过30秒
# 根因分析：
# 1. Ollama推理速度慢（没有GPU加速，或者模型参数太大）
# 2. 向量检索没有使用索引（全量扫描）
# 3. Embedding生成没有批量处理
# 优化措施1：启用Ollama GPU加速
# 确认Docker中的Ollama使用了GPU
# docker-compose.yml中必须有：
# deploy:
# resources:
# reservations:
# devices:
# - driver: nvidia
# count: all
# capabilities: [gpu]
# 验证GPU是否被使用：
# docker exec -it ollama nvidia-smi
# 如果能看到显存被使用，说明GPU加速生效
# 优化措施2：确保Milvus索引已创建
collection = Collection("legal_knowledge_base")
collection.reload()
# 检查索引状态
indexes = collection.indexes
print("索引状态：")
for index in indexes:
print(f" {index.field_name}: {index.params}")
# 如果没有索引，创建HNSW索引（提升10-50倍检索速度）
if not indexes or len(indexes) == 0:
index_params = {
"metric_type": "COSINE",
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 128}
}
collection.create_index(
field_name="embedding",
index_params=index_params
)
collection.flush()
print("✓ HNSW索引创建完成")
# 优化措施3：Embedding批量处理
# 原来单条处理，改为批量32条并行
import concurrent.futures
def batch_get_embeddings(texts: list, model: str = "m3e-base", batch_size: int = 32) -> list:
"""批量获取Embedding"""
results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
futures = {
executor.submit(get_embedding, text, model): text
for text in batch
}
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
return results
# 优化后，平均响应时间从30秒降至3-5秒| 指标 | 上线前（人工） | 上线后（AI辅助） | 提升幅度 |
|---|---|---|---|
| 单次咨询准备时间 | 45分钟 | 6分钟 | ↓ 87% |
| 相关判例召回率 | 约60% | 91% | ↑ 52% |
| 法规引用准确率 | 100%（人工） | 97% | ↓ 3% |
| 客户满意度 | 未量化 | 4.6/5.0 | 基准建立 |
| 日均处理咨询量 | 15次 | 35次 | ↑ 133% |
| 指标 | 数值 | 说明 |
|---|---|---|
| 向量数据库规模 | 103,284条向量 | 5000份文档分块后的大小 |
| 向量维度 | 1024维 | m3e-base模型输出 |
| 平均检索耗时 | 127ms | 含网络延迟和Milvus查询 |
| 平均回答生成耗时 | 2.8秒 | qwen2.5-14b-instruct在A4000上 |
| 检索准确率（Top-5） | 91.3% | 基于100个测试Query的人工评估 |
| 幻觉率 | <3% | 回答中错误引用法条的比例 |
这个案例的核心经验总结：
RAG系统建设不是一蹴而就的，需要持续优化文档处理流程、检索策略和回答生成质量。