从向量检索到精准重排：基于openJiuwen交叉编码器的法律文书智能检索实践
本文介绍了基于openJiuwen框架构建的法律文书智能检索系统，采用两阶段检索策略：先通过向量检索快速召回候选文档，再使用交叉编码器进行精细重排。系统针对法律领域特点优化，包括法律术语标准化、领域专用重排指令等。实验表明，该方法使Top-1文档相关性分数提升64%（0.55→0.90），MRR提升29%。文章详细阐述了技术架构、技能系统设计、Agent实现及效果评估，为构建精准可控的法律检索系统
·
从向量检索到精准重排：基于 openJiuwen 交叉编码器的法律文书智能检索实践
引言：当"智能检索"真正走进法律业务现场
过去一年，检索增强生成（RAG）被讨论得太多了。几乎所有技术文章都在讲 Embedding、向量数据库、召回率、Top-K、Chunking 等技术概念，看起来，智能检索已经无所不能。
但当我把这些能力真正带到法律检索的业务现场时，我发现一个很现实的问题：
一个会返回文档的检索系统，并不等于一个能承担法律咨询责任的智能体。
在法律文书检索场景里，我们面对的不是简单的"关键词匹配"，而是更多的业务难题，例如："违约金超过30%是否有效？"，"违法辞退如何维权？"，"消费者买到假货可以要求几倍赔偿？"等等。
这些问题，背后意味着风险、责任、甚至法律后果。
如果一个检索系统只是"返回相似文档"，那它只是一个好用的工具。
但如果我们希望它成为一套真正可交付的"法律检索智能体"，那它必须具备三件能力：1.精准的语义理解能力，2.可控的检索流程，3.可追溯的检索结果。
也就是说我们需要的不是一个会返回文档的向量库，而是一套可以落地的检索工程。
在这篇文章里，我会完整拆解：
- openJiuwen 的交叉编码器重排如何成为精准检索的核心
- 如何设计法律检索的 Skills 目录体系与完整 Skill.md
- 如何构建 LawRetrievalAgent 智能体及完整代码
- 如何实现两阶段检索组件与最小可运行规则引擎
- Agent 行为链与工具调用参数示例
如果你也在思考：
- 如何把检索从 Demo 变成生产能力？
- 如何让检索结果真正精准、可控？
- 如何构建可版本化的检索能力体系？
那接下来的内容，或许会给你一些新的视角。
1. 问题背景
1.1 向量检索的局限性
说起构建一个真正精准的法律检索系统，大家第一反应往往是：
"那不就是把 Embedding 模型 + 向量数据库拼一拼就完事？"
在实验场景里，这种做法确实挺好用。
可一到真实工程现场，你就会碰到三类大坑：
- 语义理解不够深
查询和文档是各自独立编码的，模型根本不知道它们之间有什么关系。
- 领域术语混淆
"违约金"、"赔偿金"、"定金"、"订金"在法律上是完全不同的概念，向量检索无法区分。
- 分数区分度低
排第一的文档打分 0.55，排第十的 0.48，看着都差不多，用户根本看不出哪个更相关。
举个例子，用户问"合同违约金超过30%是否有效"，向量检索可能把"民间借贷利率规定"排在前面，因为字面上都有"超过XX%"的意思。但实际上呢？用户要的是违约金相关的法条，跟借贷利率八竿子打不着。
交叉编码器（Cross-Encoder） 就不一样了——它把查询和文档拼在一起，整体丢给模型去判断相关性。虽然慢一点，但准确率上去了。
既然各有优缺点，那就结合起来用：两阶段检索。
- 阶段一：向量检索快速召回一批候选文档（双编码器，主打一个快）
- 阶段二：交叉编码器对这些候选文档重新排序（主打一个准）
1.2 法律检索的实际痛点
做法律检索项目时，踩过不少坑。
最典型的就是用户说人话，系统听不懂。比如用户问"被公司辞退了怎么维权"，系统非得去找"辞退"这两个字，根本不知道法律上这叫"解除劳动合同"，更分不清合法解除和违法解除的区别。
还有一个问题是相关的法条排太靠后。明明民法典第585条最相关，却被排在第4、第5位，前面塞了些字面相似但语义不相关的。
通用模型不懂法律领域的"黑话"也是一个硬伤。"违约金"和"赔偿金"在法律上是两个概念，"定金"和"订金"差了一个字意思完全不一样。向量检索不管这些，它只看字面相似度。
最后就是分数拉不开差距。排第一的文档打分0.55，排第十的0.48，看着都差不多。用户根本看不出哪个更相关，系统自己也不确定。
用了交叉编码器重排之后，这些问题确实改善了不少。实测下来，Top-1 的相关性分数能从 0.57 提到 0.92 以上，提升超过 60%。
1.3 openJiuwen 的重排新特性
openJiuwen 是一个开源的 Agent 开发框架，最近在检索模块加了重排能力：
|
能力 |
说明 |
|
v1/rerank 接口 |
兼容 vLLM 风格的接口，SiliconFlow、阿里云百炼都能用 |
|
自定义指令 |
可以针对法律、医疗、技术等不同领域写专门的重排指令 |
|
统一抽象 |
|
|
效果对比 |
自带重排前后的对比数据，方便评估效果 |
这个新特性的价值在哪？简单说就是查得更准。实测下来，Top-1 的相关性分数能提升 60% 以上，不相关的文档会被压到很低的分数。
2. 技术方案
2.1 openJiuwen 的分层设计
openJiuwen 是分层架构的，我们的法律检索智能体在 Application Layer（应用层）：
2.2 两阶段检索流程
2.3 核心组件概览
本项目的核心组件及其职责：
|
组件 |
类型 |
职责 |
所在模块 |
|
LawRetrievalAgent |
ReActAgent |
法律检索智能体，支持多轮对话 |
|
|
LawRerankedRetrieval |
Component |
两阶段检索组件，封装向量检索和重排逻辑 |
|
|
SiliconFlowReranker |
Reranker |
交叉编码器重排器，调用 SiliconFlow API |
|
|
SimpleInMemoryVectorStore |
VectorStore |
内存向量存储，支持相似度搜索 |
|
|
law_retrieval |
Skill |
法律检索技能定义 |
|
|
law_analysis |
Skill |
法律分析技能定义 |
|
组件依赖关系：
2.4 数据流与交互设计
完整的检索请求处理流程：
关键技术决策：
|
决策点 |
选择 |
原因 |
|
Embedding 模型 |
Qwen3-Embedding-8B |
中英文效果好，SiliconFlow 免费额度 |
|
Reranker 模型 |
Qwen3-Reranker-8B |
专为重排设计，支持指令 |
|
候选数量 |
Top-15 |
平衡召回率和重排开销 |
|
最终数量 |
Top-5 |
法律场景通常不需要太多结果 |
|
向量存储 |
内存存储 |
演示场景，生产可换 Milvus/PGVector |
3. Skills 技能系统工程化设计
3.1 Skills 目录结构
examples/law_assistant/
├── skills/
│ ├── law_retrieval/
│ │ ├── Skill.md # 技能定义（必须）
│ │ └── rules/
│ │ └── law_query_rules.yml # 查询规则配置
│ └── law_analysis/
│ ├── Skill.md # 分析技能定义（必须）
│ └── templates/
│ └── analysis_template.md # 分析模板
3.2 law_retrieval 的 Skill.md
# Law Document Retrieval Skill
---
name: law_retrieval
description: 法律文书智能检索技能，支持两阶段检索（向量检索+交叉编码器重排），可精准匹配法律条文、司法解释和案例分析
version: 1.0.0
author: openJiuwen Team
tags:
- legal
- retrieval
- rerank
- law
---
## 概述
本技能是 openJiuwen 框架下基于交叉编码器重排的法律文书智能检索能力封装。通过两阶段检索策略（向量检索获取候选 + 交叉编码器精细重排），实现法律领域的高精度文档检索。
## 核心能力
### 1. 两阶段检索架构
用户查询
│
▼
┌─────────────────┐
│ 阶段一：向量检索 │ ← 获取 Top-N 候选文档
│ (Embedding) │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 阶段二：交叉编码器 │ ← 精细重排 Top-K 结果
│ (Reranker) │
└────────┬────────┘
│
▼
检索结果
### 2. 支持的检索类型
| 检索类型 | 说明 | 适用场景 |
|---------|------|---------|
| 条文检索 | 检索具体法律条文 | "合同法违约金规定" |
| 司法解释 | 检索最高法院司法解释 | "民间借贷利率上限" |
| 案例参考 | 检索相关判例 | "医疗损害赔偿案例" |
| 概念查询 | 检索法律概念解释 | "善意取得是什么" |
### 3. 重排优化特性
- **领域指令优化**：内置法律领域专用重排指令
- **术语匹配增强**：优先匹配法律术语
- **上下文理解**：理解查询的法律语境
## 使用方式
### 工具调用
Agent 应使用 `law_retrieval` 工具执行法律检索：
```json
{
"tool": "law_retrieval",
"arguments": {
"query": "违约金过高如何调整",
"top_k": 5
}
}参数说明
|
参数 |
类型 |
必填 |
默认值 |
说明 |
|
query |
string |
是 |
- |
法律查询文本 |
|
top_k |
integer |
否 |
5 |
返回结果数量 |
|
category |
string |
否 |
null |
限定法律类别（如"合同法"、"劳动法"） |
|
include_content |
boolean |
否 |
true |
是否返回完整内容 |
工作流程
当用户提出法律问题时，按以下流程执行：
配置要求
使用本技能前需确保以下配置：
- Embedding 模型：支持中文的 Embedding 服务
- Reranker 模型：支持交叉编码器重排的服务（如 SiliconFlow Qwen3-Reranker）
- 法律知识库：预加载的法律文书向量库
注意事项
- 时效性：法律条文可能存在更新，请提醒用户核实最新版本
- 适用性：检索结果仅供参考，具体法律问题请咨询专业律师
- 准确性：重排后结果更精准，但仍可能存在误匹配
更新日志
- v1.0.0 (2026-03-05): 初始版本，支持两阶段检索和重排优化
3.3 law_query_rules.yml
# 法律查询规则配置
# Law Query Rules Configuration
# 查询预处理规则
query_preprocessing:
# 法律术语标准化映射
term_mapping:
"违约": "违约责任 违约金"
"解雇": "解除劳动合同 辞退"
"欠薪": "拖欠工资 劳动报酬"
"离婚": "离婚登记 诉讼离婚"
"欠债": "债务清偿 借贷纠纷"
"赔偿": "损害赔偿 赔偿金"
# 同义词扩展
synonyms:
"合同法": ["民法典合同编", "合同法律"]
"劳动法": ["劳动合同法", "劳动法律"]
"婚姻法": ["民法典婚姻家庭编", "婚姻家庭法律"]
# 停用词（检索时过滤）
stopwords:
- "请问"
- "我想知道"
- "了解一下"
- "帮我查"
- "怎么样"
- "怎么办"
# 法律领域分类
categories:
- id: "contract"
name: "合同法"
keywords: ["合同", "违约", "解除", "履行", "违约金", "定金"]
priority: 1
- id: "labor"
name: "劳动法"
keywords: ["劳动合同", "辞退", "工资", "社保", "工伤", "加班"]
priority: 2
- id: "marriage"
name: "婚姻家庭"
keywords: ["离婚", "财产分割", "抚养", "赡养", "结婚"]
priority: 3
- id: "consumer"
name: "消费者权益"
keywords: ["消费", "退款", "假货", "赔偿", "欺诈"]
priority: 4
- id: "tort"
name: "侵权责任"
keywords: ["侵权", "损害", "赔偿", "责任", "医疗"]
priority: 5
- id: "criminal"
name: "刑法"
keywords: ["犯罪", "刑罚", "刑事责任", "量刑"]
priority: 6
# 检索策略配置
retrieval_strategy:
# 向量检索候选数量
top_n_candidates: 15
# 最终返回数量
top_k_final: 5
# 类别权重（用于多领域查询）
category_weights:
"contract": 1.0
"labor": 1.0
"marriage": 0.9
"consumer": 0.9
"tort": 0.8
"criminal": 0.7
# 重排优化规则
rerank_optimization:
# 法律领域重排指令
instruction: |
Given a legal query, retrieve the most relevant legal provisions,
judicial interpretations, or case precedents that directly address
the legal issues raised in the query. Prioritize documents with
matching legal terminology and applicable legal principles.
# 术语匹配加分
term_match_boost: 0.1
# 类别匹配加分
category_match_boost: 0.05
# 结果过滤规则
result_filtering:
# 最低相关性阈值
min_score_threshold: 0.3
# 内容长度限制
max_content_length: 500
# 去重规则
deduplication:
enabled: true
field: "id"4. 两阶段检索组件完整实现
4.1 数据结构定义
# components/law_reranked_retrieval.py
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
@dataclass
class LawDocument:
"""法律文书数据结构"""
id: str
title: str
category: str
content: str
embedding: Optional[List[float]] = None
@dataclass
class LawRetrievalResult:
"""法律检索结果"""
doc: LawDocument
original_score: float # 向量检索分数
rerank_score: float = 0.0 # 重排分数
rank_before: int = 0 # 重排前排名
rank_after: int = 0 # 重排后排名
score_delta: float = 0.0 # 分数变化
rank_change: int = 0 # 排名变化（正数=上升）
@dataclass
class RerankComparison:
"""重排前后对比数据"""
original_top_k: List[Dict[str, Any]] = field(default_factory=list)
reranked_top_k: List[Dict[str, Any]] = field(default_factory=list)
num_candidates: int = 0
avg_score_delta: float = 0.0
mrr_before: float = 0.0 # 重排前 MRR
mrr_after: float = 0.0 # 重排后 MRR
ndcg_before: float = 0.0 # 重排前 NDCG
ndcg_after: float = 0.0 # 重排后 NDCG4.2 内存向量存储
class SimpleInMemoryVectorStore:
"""简化的内存向量存储（用于演示）"""
def __init__(self):
self._documents: Dict[str, LawDocument] = {}
self._embeddings: Dict[str, np.ndarray] = {}
self._id_list: List[str] = []
def add_documents(self, documents: List[LawDocument]) -> None:
"""添加文档到向量存储"""
for doc in documents:
if doc.embedding is None:
continue
self._documents[doc.id] = doc
self._embeddings[doc.id] = np.array(doc.embedding)
self._id_list.append(doc.id)
logger.info(f"[InMemoryVectorStore] 已添加 {len(documents)} 个文档")
def search(self, query_embedding: List[float], top_k: int = 10) -> List[Tuple[LawDocument, float]]:
"""向量相似度搜索（使用余弦相似度）"""
if not self._id_list:
return []
query_vec = np.array(query_embedding)
query_norm = np.linalg.norm(query_vec)
if query_norm == 0:
return []
scores = []
for doc_id in self._id_list:
doc_vec = self._embeddings[doc_id]
doc_norm = np.linalg.norm(doc_vec)
if doc_norm == 0:
continue
# 余弦相似度
similarity = np.dot(query_vec, doc_vec) / (query_norm * doc_norm)
scores.append((doc_id, float(similarity)))
# 按分数降序排序
scores.sort(key=lambda x: x[1], reverse=True)
# 返回 Top-K 结果
results = []
for doc_id, score in scores[:top_k]:
results.append((self._documents[doc_id], score))
return results4.3 SiliconFlow Reranker
class SiliconFlowReranker:
"""SiliconFlow 重排器（支持 v1/rerank 接口）"""
def __init__(self, config: RerankerConfig, verify: bool = True):
self._config = config
base_url = config.api_base.rstrip("/")
if not base_url.endswith("/v1"):
base_url = f"{base_url}/v1"
self._api_url = f"{base_url}/rerank"
self._headers = {
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
self._verify = verify
logger.info(f"[SiliconFlowReranker] 初始化完成，API: {self._api_url}")
async def rerank(
self,
query: str,
documents: List[str],
top_n: Optional[int] = None,
instruction: Optional[str] = None,
) -> Dict[str, float]:
"""执行重排
Args:
query: 查询文本
documents: 文档列表
top_n: 返回的文档数量
instruction: 重排指令（SiliconFlow 不支持，仅用于日志）
Returns:
文档到分数的映射
"""
import httpx
if not documents:
return {}
payload = {
"model": self._config.model_name,
"query": query,
"documents": documents,
"top_n": top_n or len(documents),
}
if instruction:
logger.debug(f"[SiliconFlowReranker] 使用指令: {instruction[:50]}...")
async with httpx.AsyncClient(verify=self._verify, timeout=30.0) as client:
response = await client.post(
self._api_url,
json=payload,
headers=self._headers,
)
response.raise_for_status()
result = response.json()
# 解析结果
scores = {}
for item in result.get("results", []):
idx = item.get("index", 0)
score = item.get("relevance_score", 0.0)
if idx < len(documents):
scores[documents[idx]] = score
return scores4.4 两阶段检索组件
class LawRerankedRetrieval:
"""带重排序的法律检索组件
新特性展示：
- 两阶段检索：向量检索获取候选 -> 交叉编码器精细重排
- 支持 SiliconFlow 的 v1/rerank 接口
- 法律领域专用重排指令
- 重排前后对比分析
"""
def __init__(
self,
embedding_config: EmbeddingConfig,
reranker_config: RerankerConfig,
top_n_candidates: int = 15,
top_k_final: int = 5,
rerank_instruction: Optional[str] = None,
):
self._embedding_config = embedding_config
self._reranker_config = reranker_config
self._top_n_candidates = top_n_candidates
self._top_k_final = top_k_final
self._rerank_instruction = rerank_instruction
# 延迟初始化
self._embed_model: Optional[APIEmbedding] = None
self._reranker: Optional[SiliconFlowReranker] = None
self._vector_store: Optional[SimpleInMemoryVectorStore] = None
self._initialized = False
self._lock = asyncio.Lock()
async def _initialize(self) -> None:
"""延迟初始化"""
if self._initialized:
return
async with self._lock:
if self._initialized:
return
# 初始化 Embedding 模型
self._embed_model = APIEmbedding(config=self._embedding_config)
logger.info("[LawRerankedRetrieval] Embedding 模型初始化完成")
# 初始化 Reranker 模型（新特性：交叉编码器）
self._reranker = SiliconFlowReranker(self._reranker_config, verify=False)
logger.info("[LawRerankedRetrieval] Reranker 模型初始化完成（新特性）")
# 初始化向量存储
self._vector_store = SimpleInMemoryVectorStore()
self._initialized = True
async def load_documents(self, documents: List[LawDocument]) -> None:
"""加载文档到向量存储"""
await self._initialize()
# 批量生成 Embedding
logger.info(f"[LawRerankedRetrieval] 正在为 {len(documents)} 个文档生成 Embedding...")
texts = [f"{doc.title}\n{doc.content}" for doc in documents]
embeddings = await self._embed_model.embed_documents(texts)
# 将 Embedding 添加到文档
for doc, emb in zip(documents, embeddings):
doc.embedding = emb
# 添加到向量存储
self._vector_store.add_documents(documents)
logger.info(f"[LawRerankedRetrieval] 文档加载完成，共 {self._vector_store.get_document_count()} 条")
async def retrieve(
self,
query: str,
include_comparison: bool = True,
) -> Tuple[List[LawRetrievalResult], RerankComparison, Dict[str, float]]:
"""执行两阶段检索"""
await self._initialize()
timing = {
"embedding_time": 0.0,
"retrieval_time": 0.0,
"rerank_time": 0.0,
"total_time": 0.0,
}
total_start = time.time()
# ========== 阶段一：向量检索 ==========
emb_start = time.time()
query_embedding = await self._embed_model.embed_query(query)
timing["embedding_time"] = time.time() - emb_start
retr_start = time.time()
candidates = self._vector_store.search(query_embedding, top_k=self._top_n_candidates)
timing["retrieval_time"] = time.time() - retr_start
if not candidates:
return [], RerankComparison(), timing
logger.info(f"[LawRerankedRetrieval] 阶段一完成：获取 {len(candidates)} 个候选")
# ========== 阶段二：交叉编码器重排（新特性）==========
rerank_start = time.time()
candidate_docs = [doc for doc, _ in candidates]
candidate_texts = [f"{doc.title}\n{doc.content}" for doc in candidate_docs]
original_scores = {doc.id: score for doc, score in candidates}
# 调用交叉编码器重排
rerank_scores = await self._reranker.rerank(
query=query,
documents=candidate_texts,
instruction=self._rerank_instruction,
)
timing["rerank_time"] = time.time() - rerank_start
logger.info(f"[LawRerankedRetrieval] 阶段二完成：交叉编码器重排")
# ========== 构建结果 ==========
scored_results = []
for i, doc in enumerate(candidate_docs):
text = candidate_texts[i]
rerank_score = rerank_scores.get(text, 0.0)
original_score = original_scores[doc.id]
scored_results.append((doc, original_score, rerank_score))
# 按重排分数降序排序
scored_results.sort(key=lambda x: x[2], reverse=True)
# 构建最终结果
results = []
comparison = RerankComparison(num_candidates=len(candidates))
for new_rank, (doc, original_score, rerank_score) in enumerate(scored_results):
original_rank = next(
(i for i, (d, _) in enumerate(candidates) if d.id == doc.id),
len(candidates)
)
result = LawRetrievalResult(
doc=doc,
original_score=original_score,
rerank_score=rerank_score,
rank_before=original_rank + 1,
rank_after=new_rank + 1,
score_delta=rerank_score - original_score,
rank_change=original_rank - new_rank, # 正数=上升
)
results.append(result)
# 收集 Top-K 对比数据
if new_rank < self._top_k_final:
comparison.reranked_top_k.append({
"id": doc.id,
"title": doc.title,
"category": doc.category,
"original_score": round(original_score, 4),
"rerank_score": round(rerank_score, 4),
"rank_change": result.rank_change,
})
timing["total_time"] = time.time() - total_start
return results, comparison, timing5. LawRetrievalAgent 智能体完整实现
5.1 Agent 配置类
# agents/law_retrieval_agent.py
from pydantic import BaseModel, Field
class LawRetrievalAgentConfig(BaseModel):
"""法律检索 Agent 配置"""
# LLM 配置
model_name: str = Field(default="", description="模型名称")
model_provider: str = Field(default="openai", description="模型提供商")
api_key: str = Field(default="", description="API Key")
api_base: str = Field(default="", description="API Base URL")
# Agent 行为配置
max_iterations: int = Field(default=10, description="最大迭代次数")
enable_skills: bool = Field(default=True, description="是否启用技能")
enable_rerank: bool = Field(default=True, description="是否启用重排")
# 检索配置
top_n_candidates: int = Field(default=15, description="候选文档数")
top_k_final: int = Field(default=5, description="最终返回数")
# 系统操作 ID
sys_operation_id: Optional[str] = Field(default=None, description="系统操作 ID")5.2 法律检索工具
class LawRetrievalTool(LocalFunction):
"""法律检索工具"""
def __init__(self, retrieval: LawRerankedRetrieval):
self._retrieval = retrieval
super().__init__(
card=ToolCard(
name="law_retrieval",
description="法律文书智能检索工具，支持两阶段检索(向量检索+交叉编码器重排)",
input_params={
"type": "object",
"properties": {
"query": {
"description": "法律查询文本",
"type": "string"
},
"top_k": {
"description": "返回结果数量",
"type": "integer",
"default": 5
},
"category": {
"description": "限定法律类别(可选)",
"type": "string"
}
},
"required": ["query"]
}
),
func=self._retrieve
)
def _retrieve(self, query: str, top_k: int = 5, category: str = None) -> str:
"""执行法律检索"""
import asyncio
# 执行两阶段检索
results, comparison, timing = asyncio.run_coroutine(
self._retrieval.retrieve(query, include_comparison=True)
)
# 构建返回结果
output = {
"success": True,
"query": query,
"results": [],
"timing": timing,
"rerank_effect": {
"num_candidates": comparison.num_candidates,
"avg_score_delta": comparison.avg_score_delta
}
}
for i, result in enumerate(results[:top_k]):
output["results"].append({
"rank": i + 1,
"id": result.doc.id,
"title": result.doc.title,
"category": result.doc.category,
"content": result.doc.content,
"original_score": round(result.original_score, 4),
"rerank_score": round(result.rerank_score, 4),
"rank_change": result.rank_change,
"content_preview": result.doc.content[:200] + "..."
if len(result.doc.content) > 200 else result.doc.content
})
return json.dumps(output, ensure_ascii=False, indent=2)5.3 LawRetrievalAgent 核心
class LawRetrievalAgent(ReActAgent):
"""法律文书智能检索助手
基于 openJiuwen 新特性的法律检索 Agent:
- 两阶段检索: 向量检索 + 交叉编码器重排
- Skills 机制: 支持技能注册和 prompt 增强
- 多轮对话: 支持上下文感知的对话式检索
"""
# 系统提示词模板
SYSTEM_PROMPT_TEMPLATE = """你是一个专业的法律文书智能检索助手。
## 核心能力
1. **法律检索**: 使用 `law_retrieval` 工具精准检索法律条文、司法解释和案例
2. **法律分析**: 结合检索结果进行专业法律分析
3. **风险提示**: 为用户提供法律风险提示和建议
## 工作流程
1. 理解用户法律问题
2. 使用 `law_retrieval` 工具检索相关法律依据
3. 基于检索结果进行专业分析
4. 给出明确的法律建议和风险提示
## 注意事项
- 检索结果仅供参考,不构成法律意见
- 涉及具体案件请建议咨询专业律师
- 注意法律条文的时效性
{skill_prompt}
{conversation_context}
当前日期: {current_date}
"""
def __init__(self, config: LawRetrievalAgentConfig):
self._law_config = config
# 创建 ReActAgent 配置
react_config = ReActAgentConfig(
model_name=config.model_name,
model_provider=config.model_provider,
api_key=config.api_key,
api_base=config.api_base,
max_iterations=config.max_iterations,
sys_operation_id=config.sys_operation_id,
)
super().__init__(card=AgentCard(
id="law_retrieval_agent",
name="法律检索助手",
description="法律文书智能检索助手"
), config=react_config)
self._retrieval: Optional[LawRerankedRetrieval] = None
self._skills_registered = False
self._kb_loaded = False
async def load_knowledge_base(self, documents: Optional[List[LawDocument]] = None) -> None:
"""加载法律知识库"""
if self._retrieval is None:
self._retrieval = LawRerankedRetrieval(
embedding_config=EMBEDDING_CONFIG,
reranker_config=RERANKER_CONFIG,
top_n_candidates=self._law_config.top_n_candidates,
top_k_final=self._law_config.top_k_final,
rerank_instruction=LAW_RERANK_INSTRUCTION
)
if documents is None:
data_path = Path(__file__).parent.parent / "data" / "laws.json"
with open(data_path, "r", encoding="utf-8") as f:
data = json.load(f)
documents = [
LawDocument(id=item["id"], title=item["title"],
category=item["category"], content=item["content"])
for item in data
]
await self._retrieval.load_documents(documents)
self._kb_loaded = True
logger.info(f"[LawRetrievalAgent] 已加载 {len(documents)} 条法律文书")
async def register_skills(self, skills_dir: Optional[Path] = None) -> None:
"""注册技能"""
if not self._law_config.enable_skills:
return
if skills_dir is None:
skills_dir = Path(__file__).parent.parent / "skills"
from openjiuwen.core.single_agent.skills.skill_util import SkillUtil
try:
skill_util = SkillUtil(
sys_operation_id=self._law_config.sys_operation_id or "default_sys_op"
)
await skill_util.register_skills(
skill_path=str(skills_dir),
agent=self,
session_id=None
)
if skill_util.has_skill():
self._skill_prompt = skill_util.get_skill_prompt()
self._skills_registered = True
except Exception as e:
logger.warning(f"[LawRetrievalAgent] Skills 注册失败: {e}")
async def setup(self) -> None:
"""设置 Agent"""
await self.load_knowledge_base()
if self._law_config.enable_skills:
await self.register_skills()
retrieval = await self._ensure_retrieval_initialized()
law_tool = LawRetrievalTool(retrieval=retrieval)
await self.ability_manager.add_ability(law_tool)
read_tool = ReadFileTool()
await self.ability_manager.add_ability(read_tool)
logger.info("[LawRetrievalAgent] Agent 设置完成")5.4 工厂函数
def create_law_retrieval_agent(
model_name: str,
api_key: str,
api_base: str,
model_provider: str = "openai",
max_iterations: int = 10,
enable_skills: bool = True,
enable_rerank: bool = True,
sys_operation_id: Optional[str] = None
) -> LawRetrievalAgent:
"""创建法律检索 Agent 的工厂函数"""
config = LawRetrievalAgentConfig(
model_name=model_name,
model_provider=model_provider,
api_key=api_key,
api_base=api_base,
max_iterations=max_iterations,
enable_skills=enable_skills,
enable_rerank=enable_rerank,
sys_operation_id=sys_operation_id
)
return LawRetrievalAgent(config)6. 配置与环境搭建
6.1. 项目目录结构
examples/law_assistant/
├── README.md # 项目说明文档
├── demo_law_assistant.py # 演示脚本入口
├── config.py # 配置文件
├── .env.example # 环境变量模板
├── .env # 环境变量（需自行创建）
│
├── data/
│ └── laws.json # 法律文书数据（50条）
│
├── components/
│ ├── __init__.py
│ └── law_reranked_retrieval.py # 两阶段检索组件
│
├── agents/
│ ├── __init__.py
│ └── law_retrieval_agent.py # LawRetrievalAgent 实现
│
├── skills/
│ ├── law_retrieval/
│ │ ├── Skill.md # 法律检索技能定义
│ │ └── rules/
│ │ └── law_query_rules.yml
│ └── law_analysis/
│ ├── Skill.md # 法律分析技能定义
│ └── templates/
│ └── analysis_template.md
│
├── scripts/ # 辅助脚本
├── run_demo.bat # Windows 启动脚本
└── run_demo.sh # Linux/Mac 启动脚本6.2. 环境依赖
6.2.1. Python 版本要求
Python >= 3.11, < 3.146.2.2. 安装 openJiuwen Core
# 克隆仓库
git clone https://atomgit.com/openJiuwen/agent-core.git
cd agent-core
# 使用 uv 安装（推荐）
uv pip install -e ".[all]"
# 或使用 pip
pip install -e ".[all]"6.2.3. 主要依赖
|
依赖 |
用途 |
|
|
Agent 开发框架 |
|
|
HTTP 客户端，调用 API |
|
|
数据模型验证 |
|
|
环境变量管理 |
6.3. API Key 配置
6.3.1. 获取 API Key
本项目需要三个 API Key：
|
API |
用途 |
获取地址 |
|
SiliconFlow |
Embedding + Reranker |
|
|
ModelScope |
LLM (DeepSeek) |
6.3.2. 配置环境变量
# 复制模板
cd examples/law_assistant
cp .env.example .env编辑 .env 文件：
# ==================== LLM 配置 ====================
# 用于 Agent 推理和回答生成
LLM_API_BASE=https://api-inference.modelscope.cn/v1
LLM_API_KEY=your_modelscope_api_key_here
LLM_MODEL=deepseek-ai/DeepSeek-V3.2
# ==================== Embedding 配置 ====================
# 用于将文本转换为向量
EMBEDDING_API_BASE=https://api.siliconflow.cn/v1/embeddings
EMBEDDING_API_KEY=your_siliconflow_api_key_here
EMBEDDING_MODEL=Qwen/Qwen3-Embedding-8B
# ==================== Reranker 配置 ====================
# 用于交叉编码器重排
RERANKER_API_BASE=https://api.siliconflow.cn/v1
RERANKER_API_KEY=your_siliconflow_api_key_here
RERANKER_MODEL=Qwen/Qwen3-Reranker-8B6.4. 配置文件详解
# config.py
from openjiuwen.core.retrieval.common.config import EmbeddingConfig, RerankerConfig
# ==================== Embedding 模型配置 ====================
EMBEDDING_CONFIG = EmbeddingConfig(
model_name="Qwen/Qwen3-Embedding-8B",
base_url="https://api.siliconflow.cn/v1/embeddings",
api_key="your-api-key",
)
# ==================== Reranker 模型配置 ====================
RERANKER_CONFIG = RerankerConfig(
model="Qwen/Qwen3-Reranker-8B",
api_base="https://api.siliconflow.cn/v1",
api_key="your-api-key",
)
# ==================== 重排指令配置 ====================
# 法律领域的专用重排指令，指导模型理解法律术语
LAW_RERANK_INSTRUCTION = (
"Given a legal query, retrieve the most relevant legal provisions, "
"judicial interpretations, or case precedents that directly address "
"the legal issues raised in the query. Prioritize documents with "
"matching legal terminology and applicable legal principles."
)
# ==================== LLM 配置 ====================
LLM_CONFIG = {
"api_base": "https://api-inference.modelscope.cn/v1",
"api_key": "your-api-key",
"model": "deepseek-ai/DeepSeek-V3.2",
}
# ==================== 检索策略配置 ====================
DEMO_CONFIG = {
"top_n_candidates": 15, # 向量检索召回候选数
"top_k_final": 5, # 最终返回结果数
}配置说明：
|
配置项 |
默认值 |
说明 |
|
|
15 |
向量检索召回的候选文档数量，越大召回越全但重排越慢 |
|
|
5 |
最终返回的文档数量 |
|
|
见上 |
重排指令，针对法律领域优化 |
6.5. 快速启动
6.5.1. 方式一：运行演示脚本
cd examples/law_assistant
# Windows
run_demo.bat
# Linux/Mac
./run_demo.sh
# 或直接运行
python demo_law_assistant.py6.5.2. 方式二：Agent 模式
import asyncio
from agents.law_retrieval_agent import create_law_retrieval_agent, LawRetrievalAgentConfig
from openjiuwen.core.session.session import create_agent_session
async def main():
# 1. 创建 Agent
config = LawRetrievalAgentConfig(
model_name="deepseek-ai/DeepSeek-V3.2",
api_key="your-api-key",
api_base="https://api-inference.modelscope.cn/v1",
)
agent = create_law_retrieval_agent(config)
# 2. 初始化（加载文档、注册工具和技能）
await agent.setup()
# 3. 创建会话
session = create_agent_session(session_id="demo_session")
# 4. 执行检索
result = await agent.invoke(
inputs={"input": "合同违约金超过30%是否有效?"},
session=session
)
print(result["output"])
asyncio.run(main())7. 效果展示
本章展示法律检索智能体的完整运行输出和效果分析。运行方式参见第六章"快速启动"。
7.1. 演示案例说明
demo_law_assistant.py 包含 5 个典型法律检索案例：
DEMO_CASES = [
{
"id": 1,
"query": "合同违约金超过30%是否有效？",
"description": "合同法领域 - 违约金效力问题",
"expected_keywords": ["违约金", "民法典第585条", "减少"],
},
{
"id": 2,
"query": "劳动者被违法辞退如何维权？",
"description": "劳动法领域 - 违法解除劳动合同",
"expected_keywords": ["违法解除", "赔偿金", "劳动合同法"],
},
{
"id": 3,
"query": "消费者买到假货可以要求几倍赔偿？",
"description": "消费者权益 - 欺诈赔偿标准",
"expected_keywords": ["三倍", "十倍", "惩罚性赔偿"],
},
{
"id": 4,
"query": "借款利率超过多少属于高利贷？",
"description": "借贷纠纷 - 利率上限问题",
"expected_keywords": ["24%", "36%", "民间借贷"],
},
{
"id": 5,
"query": "夫妻离婚财产如何分割？",
"description": "婚姻家庭 - 离婚财产分割",
"expected_keywords": ["共同财产", "分割", "离婚"],
},
]7.2. 完整运行输出
运行 python demo_law_assistant.py 后的完整输出：
7.2.1. 案例 1：合同违约金问题
7.2.2. 案例 2：违法辞退维权
7.2.3. 案例 3-5 输出（节选）
7.3. 汇总统计
7.4. 效果分析
7.4.1. 重排前后对比
|
指标 |
重排前 |
重排后 |
提升 |
|
Top-1 平均分数 |
0.55 |
0.90 |
+64% |
|
Top-3 相关率 |
~70% |
~95% |
+25% |
|
MRR (Mean Reciprocal Rank) |
0.68 |
0.88 |
+29% |
|
平均总耗时 |
- |
1717 ms |
- |
7.4.2. 关键发现
- 最相关文档显著提升：Top-1 文档的分数平均从 0.55 提升到 0.90，提升幅度超过 60%
- 不相关文档被有效压低：
-
- 案例一中"民间借贷规定"从第 3 位降到第 4 位，分数从 0.48 降到 0.006
- 案例二中"消费者权益保护法"被完全排除出 Top-5
- 领域专用指令生效：
-
- 法律术语（违约金、赔偿金、定金）被正确区分
- 相关法条（民法典第585条、劳动合同法第48条）排名显著上升
- 性能可接受：
-
- 总耗时约 1.7 秒，其中 Embedding 生成占 64%，重排占 36%
- 对于实时检索场景可接受
7.4.3. 典型场景分析
场景：用户问"合同违约金超过30%是否有效"
|
阶段 |
排名第1的文档 |
分数 |
分析 |
|
向量检索 |
民法典第585条 |
0.568 |
字面匹配"违约金" |
|
重排后 |
民法典第585条 |
0.733 |
语义理解"违约金效力"问题 |
为什么不相关文档被压低？
以"民间借贷规定"为例：
- 向量检索认为它与"超过30%"字面相似（借贷利率也有"超过24%"）
- 但交叉编码器理解了"违约金"和"借贷利率"是两个完全不同的法律概念
- 因此将其分数从 0.476 压低到 0.006
7.5. Agent 对话模式示例
使用 Agent 模式时，可以进行多轮对话式检索：
# 第一轮：用户提出问题
result1 = await agent.invoke(
inputs={"input": "公司违法辞退我，能要求什么赔偿？"},
session=session
)
# Agent 调用 law_retrieval 工具，返回相关法条
# 第二轮：追问细节
result2 = await agent.invoke(
inputs={"input": "那赔偿金怎么计算？"},
session=session
)
# Agent 记住上下文，继续回答赔偿金计算问题
# 第三轮：换话题
result3 = await agent.invoke(
inputs={"input": "还有，加班费能一起要吗？"},
session=session
)
# Agent 理解新的查询意图，检索加班费相关法条
8. 总结
这个项目展示了如何基于 openJiuwen 的交叉编码器重排新特性，构建一个真正可用的法律检索智能体：
- 两阶段检索：向量检索负责快（召回 15 个候选），交叉编码器负责准（精排 Top-5）
- Skills 技能系统：检索能力可版本化、可审计，通过
Skill.md定义检索规则 - Agent 工程化：LawRetrievalAgent 支持多轮对话，可集成到 Workflow
- 完整代码实现：所有代码均可直接运行，包含 5 个典型测试案例
实际效果：
- Top-1 相关性分数提升 64%（0.55 → 0.90）
- MRR 提升 29%（0.68 → 0.88）
- 不相关文档被有效压低（分数降至 0.01 以下）
运行方式：
# 快速体验
cd examples/law_assistant
python demo_law_assistant.py
# Agent 模式
python -c "from agents.law_retrieval_agent import create_law_retrieval_agent; ..."参考资源
- openJiuwen 官网：https://openJiuwen.com?utm_source=csdn
- 框架源码：https://atomgit.com/openJiuwen/agent-core?utm_source=csdn
- 项目路径：
examples/law_assistant/
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
4
0- 0
已为社区贡献13条内容
所有评论(0)