ベクトル埋め込みを使用した法学検索エンジン
「不良品の損害賠償」に関する判例を探す弁護士 「生産者の責任」という文言を使用した関連判決を見逃す可能性がある。 善の悪徳」。キーワードの完全一致に基づく従来の全文検索、 同じ規範またはケースを記述できる領域では体系的に失敗します。 時代や管轄区域が異なると用語も異なります。
I ベクトルの埋め込み そして セマンティック検索 彼らはこの問題を解決します ルートまで: 単語を比較するのではなく、 意味。についての問い合わせ 「同意の瑕疵による契約の無効」は、「取消可能性」に関する文章を自動的に検索します。 なぜなら、この 2 つの概念は、法的取引の類似した領域に存在するからです。 ベクトル空間。この記事では、判例検索エンジンをゼロから構築します。 Python、法律に特化した埋め込みモデル、ベクトル データベースを使用して本番環境に対応します。
何を学ぶか
- 法学用のセマンティック検索エンジンのアーキテクチャ
- 法的領域に特化した埋め込みモデル (legal-BERT、ModernBERT)
- FAISS と Pinecone による効率的なインデックス作成
- ハイブリッド検索: BM25 + ベクトル類似性により精度を最大化
- 最終結果を得るためにクロスエンコーダーを使用して再ランキングする
- LegalTech アプリケーションに統合するための FastAPI を使用した REST API
システムアーキテクチャ
最新の法学検索エンジンは、次の 3 つの主要な層で構成されています。
- 取り込みパイプライン: ソースから文をダウンロード、正規化、処理します 公式 (EUR-Lex、ECLI API、DeJure、Cassation)。すぐに使用できるチャンク化されたドキュメントを生成します 埋め込み。
- インデックス作成エンジン: 各文チャンクのベクトル埋め込みを生成します そして、それらをベクター ストアにインデックス付けします (セルフホスト型の場合は FAISS、マネージド型の場合は Pinecone)。
- クエリエンジン: ユーザーのクエリを処理し、埋め込みに変換します。 ベクトル検索を実行し、ハイブリッド再ランキングを適用して、結果を返します。 検証可能な引用。
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import date
from enum import Enum
class JurisdictionType(Enum):
CASSAZIONE = "cassazione"
CORTE_APPELLO = "corte_appello"
TRIBUNALE = "tribunale"
CORTE_COSTITUZIONALE = "corte_costituzionale"
CORTE_GIUSTIZIA_UE = "corte_giustizia_ue"
CEDU = "cedu"
@dataclass
class CourtDecision:
"""Rappresenta una sentenza indicizzata nel sistema."""
ecli: str # European Case Law Identifier (es. ECLI:IT:CASS:2024:1234)
court: JurisdictionType
date: date
number: str # numero sentenza
subject_matter: str # materia (civile, penale, amministrativo...)
keywords: List[str] # parole chiave ufficiali
headnotes: str # massima/principio di diritto
full_text: str # testo integrale
citations: List[str] # sentenze citate
cited_by: List[str] = field(default_factory=list) # sentenze che citano questa
@dataclass
class ChunkedDecision:
"""Chunk di sentenza pronto per l'embedding."""
chunk_id: str
ecli: str
chunk_type: str # "headnote", "facts", "reasoning", "decision"
content: str
embedding: Optional[List[float]] = None
metadata: dict = field(default_factory=dict)埋め込みモデルの選択
埋め込みモデルの選択は、法的調査の品質にとって非常に重要です。
などの汎用モデル text-embedding-3-large OpenAIによる
良い結果が得られますが、法的コーパスで事前トレーニングされたモデルの方が優れたパフォーマンスを発揮します
専門的な法律業務に関しては非常に汎用的です。
| モデル | 寸法 | 専門分野 | NDCG@10 合法 | 導入 |
|---|---|---|---|---|
| テキスト埋め込み-3-大 | 3072 | 一般的な | 0.71 | API（OpenAI） |
| nlpaueb/legal-bert-base | 768 | 法的 (英語) | 0.79 | ハグ顔 |
| 自由法プロジェクト/modernbert | 768 | 判例法 (英語) | 0.83 | ハグ顔 |
| dbmdz/bert-base-イタリア語 | 768 | イタリア語（一般） | 0.74 | ハグ顔 |
| 航海の法則-2 | 1024 | 法的 (EN+多言語) | 0.86 | API（Voyage AI） |
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import torch
import numpy as np
from typing import List
class LegalEmbeddingService:
"""
Servizio di embedding specializzato per testi giuridici.
Supporta modelli locali (HuggingFace) e API remoti.
"""
def __init__(self, model_name: str = "nlpaueb/legal-bert-base-uncased"):
self.model_name = model_name
self.device = "cuda" if torch.cuda.is_available() else "cpu"
# Usa SentenceTransformer per modelli ottimizzati per similarity
self.model = SentenceTransformer(model_name, device=self.device)
self.embedding_dim = self.model.get_sentence_embedding_dimension()
print(f"Modello caricato: {model_name} | Dim: {self.embedding_dim} | Device: {self.device}")
def encode_texts(
self,
texts: List[str],
batch_size: int = 32,
normalize: bool = True
) -> np.ndarray:
"""
Genera embedding per una lista di testi.
Normalizzazione L2 per cosine similarity via dot product.
"""
embeddings = self.model.encode(
texts,
batch_size=batch_size,
normalize_embeddings=normalize,
show_progress_bar=len(texts) > 100,
convert_to_numpy=True
)
return embeddings
def encode_query(self, query: str) -> np.ndarray:
"""
Encode di una singola query utente.
Per alcuni modelli (es. E5) si usa il prefisso "query: "
"""
# E5 e INSTRUCTOR richiedono prefissi per le query
if "e5" in self.model_name.lower():
query = f"query: {query}"
elif "instructor" in self.model_name.lower():
query = f"Represent the legal question for retrieval: {query}"
return self.model.encode(
[query],
normalize_embeddings=True,
convert_to_numpy=True
)[0]FAISS によるインデックス作成
FAISS (Facebook AI類似性検索) は、ベクトル検索のリファレンス ライブラリです。 大規模なデータセットでの高いパフォーマンス。 1,000 万文のコレクションの場合、 プロダクト量子化 (PQ) を備えた IVF インデックス (逆ファイル インデックス) により、 コモディティ CPU では応答時間は 100 ミリ秒未満です。
import faiss
import numpy as np
import pickle
import os
from typing import Tuple
class FAISSCaseLawIndex:
"""
Indice FAISS ottimizzato per ricerca giurisprudenziale.
Supporta indici flat (piccoli dataset) e IVF+PQ (milioni di sentenze).
"""
def __init__(self, embedding_dim: int, index_type: str = "ivf"):
self.embedding_dim = embedding_dim
self.index_type = index_type
self.index = None
self.id_to_metadata = {} # mapping interno_id -> metadati chunk
self.next_id = 0
def build_index(self, embeddings: np.ndarray, num_clusters: int = 1024):
"""
Costruisce l'indice FAISS.
- 'flat': ricerca esatta (fino a ~500K vettori)
- 'ivf': ricerca approssimata (milioni di vettori, ~5-10x più veloce)
"""
n_vectors = embeddings.shape[0]
print(f"Building {self.index_type} index per {n_vectors} vettori...")
if self.index_type == "flat":
# Inner product = cosine similarity se vettori normalizzati
self.index = faiss.IndexFlatIP(self.embedding_dim)
elif self.index_type == "ivf":
# IVF con quantizzazione per grandi dataset
quantizer = faiss.IndexFlatIP(self.embedding_dim)
# PQ: 8 sottospazi, 8 bit = compressione 32x con loss minima
pq_segments = min(self.embedding_dim, 8)
self.index = faiss.IndexIVFPQ(
quantizer,
self.embedding_dim,
num_clusters,
pq_segments, # numero di segmenti PQ
8 # bit per centroide
)
# Training obbligatorio per IVF
print("Training IVF index...")
self.index.train(embeddings)
# nprobe: quanti cluster esaminare. Tradeoff recall/speed
self.index.nprobe = 64
# Aggiunta dei vettori
self.index.add(embeddings)
print(f"Index costruito: {self.index.ntotal} vettori")
def add_with_metadata(self, embeddings: np.ndarray, chunks: List[dict]):
"""Aggiunge embedding con metadati associati."""
start_id = self.next_id
self.index.add(embeddings)
for i, chunk in enumerate(chunks):
self.id_to_metadata[start_id + i] = chunk
self.next_id += len(chunks)
def search(
self,
query_embedding: np.ndarray,
k: int = 20,
score_threshold: float = 0.6
) -> List[dict]:
"""
Ricerca per similarity con filtro score minimo.
"""
query = query_embedding.reshape(1, -1).astype(np.float32)
scores, indices = self.index.search(query, k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx == -1: # FAISS usa -1 per risultati invalidi
continue
if score >= score_threshold:
result = {**self.id_to_metadata.get(idx, {}), 'score': float(score)}
results.append(result)
return results
def save(self, path: str):
"""Salva indice e metadati su disco."""
faiss.write_index(self.index, f"{path}/index.faiss")
with open(f"{path}/metadata.pkl", "wb") as f:
pickle.dump({'id_to_metadata': self.id_to_metadata, 'next_id': self.next_id}, f)
def load(self, path: str):
"""Carica indice da disco."""
self.index = faiss.read_index(f"{path}/index.faiss")
with open(f"{path}/metadata.pkl", "rb") as f:
data = pickle.load(f)
self.id_to_metadata = data['id_to_metadata']
self.next_id = data['next_id']ハイブリッド検索: BM25 + ベクトル類似性
純粋なセマンティック検索は、関連する概念を見つけることに優れていますが、見逃す可能性があります。 正確な規制参照との完全一致 (例: 「art. 1453 c.c.」、「Legislative Decree 231/2001」)。 BM25 (キーワードベース) 検索は完全一致には優れていますが、セマンティクスは考慮されません。 アプローチ ハイブリッド 両方を組み合わせて再現率と精度を最大化します。
from rank_bm25 import BM25Okapi
import re
from typing import List, Tuple
class HybridCaseLawSearch:
"""
Motore di ricerca ibrido BM25 + Vector Similarity.
Usa Reciprocal Rank Fusion (RRF) per combinare i ranking.
"""
def __init__(self, embedding_service, faiss_index, corpus: List[dict]):
self.embedding_service = embedding_service
self.faiss_index = faiss_index
self.corpus = corpus
# Inizializza BM25 su tutti i testi del corpus
tokenized_corpus = [self._tokenize_legal(doc['content']) for doc in corpus]
self.bm25 = BM25Okapi(tokenized_corpus)
print(f"BM25 inizializzato su {len(corpus)} documenti")
def _tokenize_legal(self, text: str) -> List[str]:
"""
Tokenizzazione specializzata per testi legali italiani.
Preserva riferimenti normativi come "art.1453", "D.Lgs.231/2001".
"""
# Normalizza riferimenti normativi
text = re.sub(r'art\.\s*(\d+)', r'art_\1', text, flags=re.IGNORECASE)
text = re.sub(r'D\.Lgs\.\s*(\d+/\d+)', r'dlgs_\1', text, flags=re.IGNORECASE)
# Tokenizza
tokens = re.findall(r'\b[a-zA-Z_àèìòùÀÈÌÒÙ][a-zA-Z0-9_àèìòùÀÈÌÒÙ]*\b', text.lower())
# Rimuovi stopwords legali italiane comuni
stopwords = {'il', 'la', 'i', 'le', 'di', 'del', 'della', 'dei', 'delle',
'in', 'con', 'per', 'su', 'da', 'al', 'allo', 'alle', 'che', 'si'}
return [t for t in tokens if t not in stopwords and len(t) > 2]
def _reciprocal_rank_fusion(
self,
vector_results: List[dict],
bm25_results: List[Tuple[int, float]],
k: int = 60,
vector_weight: float = 0.6,
bm25_weight: float = 0.4
) -> List[dict]:
"""
Reciprocal Rank Fusion (RRF) per combinare ranking eterogenei.
Formula: RRF(d) = sum(weight_i / (k + rank_i(d)))
"""
rrf_scores = {}
# Score vector results
for rank, result in enumerate(vector_results):
doc_id = result['chunk_id']
if doc_id not in rrf_scores:
rrf_scores[doc_id] = {'score': 0, 'data': result}
rrf_scores[doc_id]['score'] += vector_weight / (k + rank + 1)
# Score BM25 results
for rank, (doc_idx, _) in enumerate(bm25_results):
doc_id = self.corpus[doc_idx]['chunk_id']
if doc_id not in rrf_scores:
rrf_scores[doc_id] = {'score': 0, 'data': self.corpus[doc_idx]}
rrf_scores[doc_id]['score'] += bm25_weight / (k + rank + 1)
# Sort per RRF score
sorted_results = sorted(rrf_scores.values(), key=lambda x: x['score'], reverse=True)
return [{**r['data'], 'rrf_score': r['score']} for r in sorted_results]
def search(self, query: str, top_k: int = 10) -> List[dict]:
"""
Ricerca ibrida con fusione RRF.
"""
# Vector search
query_embedding = self.embedding_service.encode_query(query)
vector_results = self.faiss_index.search(query_embedding, k=50)
# BM25 search
tokenized_query = self._tokenize_legal(query)
bm25_scores = self.bm25.get_scores(tokenized_query)
top_bm25_indices = np.argsort(bm25_scores)[::-1][:50]
bm25_results = [(idx, bm25_scores[idx]) for idx in top_bm25_indices]
# Fusione RRF
fused = self._reciprocal_rank_fusion(vector_results, bm25_results)
return fused[:top_k]クロスエンコーダの再ランキング
ハイブリッド検索の後、次のステップが実行されます。 再ランキング クロスエンコーダー付き さらに精度が向上します。クロスエンコーダはペア (クエリ、ドキュメント) を処理します。 を組み合わせて、バイエンコーダーよりもはるかに正確な関連性スコアを生成しますが、 計算コストが高くなります。これが、後で上位 K 個の候補にのみ使用される理由です。 最初の検索。
from sentence_transformers import CrossEncoder
class LegalReranker:
"""
Cross-encoder per re-ranking di risultati giurisprudenziali.
"""
def __init__(
self,
model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"
):
self.model = CrossEncoder(model_name, max_length=512)
def rerank(
self,
query: str,
candidates: List[dict],
top_k: int = 5
) -> List[dict]:
"""
Re-ranking con cross-encoder su lista di candidati.
"""
if not candidates:
return []
# Costruisce coppie (query, documento) per il cross-encoder
pairs = [(query, candidate['content'][:400]) for candidate in candidates]
# Score di rilevanza (singolo float per ogni coppia)
scores = self.model.predict(pairs)
# Associa score e ordina
for candidate, score in zip(candidates, scores):
candidate['rerank_score'] = float(score)
reranked = sorted(candidates, key=lambda x: x['rerank_score'], reverse=True)
return reranked[:top_k]
# Pipeline completa
class CaseLawSearchEngine:
"""
Search engine giurisprudenziale con pipeline completa:
Hybrid Search -> Cross-Encoder Re-ranking -> Format results
"""
def __init__(self, hybrid_searcher, reranker):
self.hybrid_searcher = hybrid_searcher
self.reranker = reranker
def search(self, query: str, top_k: int = 5) -> List[dict]:
# Step 1: Hybrid retrieval (candidate generation)
candidates = self.hybrid_searcher.search(query, top_k=20)
# Step 2: Cross-encoder re-ranking (precision optimization)
results = self.reranker.rerank(query, candidates, top_k=top_k)
# Step 3: Format con citazioni ECLI
return [{
'ecli': r.get('ecli', 'N/A'),
'court': r.get('court', 'N/A'),
'date': r.get('date', 'N/A'),
'headnote': r.get('headnote', ''),
'excerpt': r['content'][:300] + "...",
'relevance_score': r['rerank_score'],
'source_url': f"https://www.italgiure.giustizia.it/{r.get('ecli', '')}"
} for r in results]FastAPIを使用したREST API
検索エンジンを FastAPI を備えた REST マイクロサービスとして公開し、すぐに統合できます あらゆるリーガルテック アプリケーションで。
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import List, Optional
import time
app = FastAPI(
title="Case Law Search API",
description="Motore di ricerca giurisprudenziale con vector embeddings",
version="1.0.0"
)
class SearchRequest(BaseModel):
query: str = Field(..., min_length=10, max_length=500)
top_k: int = Field(default=5, ge=1, le=20)
jurisdiction: Optional[str] = Field(None, description="Filtra per giurisdizione")
date_from: Optional[str] = Field(None, description="Data minima (YYYY-MM-DD)")
date_to: Optional[str] = Field(None, description="Data massima (YYYY-MM-DD)")
class SearchResult(BaseModel):
ecli: str
court: str
date: str
headnote: str
excerpt: str
relevance_score: float
source_url: str
class SearchResponse(BaseModel):
query: str
results: List[SearchResult]
total_results: int
processing_time_ms: float
@app.post("/api/v1/search", response_model=SearchResponse)
async def search_case_law(request: SearchRequest):
"""
Ricerca semantica nella giurisprudenza italiana ed europea.
"""
start_time = time.time()
try:
results = search_engine.search(
query=request.query,
top_k=request.top_k
)
processing_time = (time.time() - start_time) * 1000
return SearchResponse(
query=request.query,
results=results,
total_results=len(results),
processing_time_ms=round(processing_time, 2)
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Errore nella ricerca: {str(e)}")
@app.get("/api/v1/health")
async def health_check():
return {"status": "ok", "index_size": search_engine.hybrid_searcher.faiss_index.index.ntotal}ECLI と欧州規格に関する考慮事項
L'欧州判例法識別子 (ECLI) そして欧州規格では、
文の一意の識別。 ECLI の形式は次のとおりです。
ECLI:{country}:{court}:{year}:{number}
- 例えば ECLI:IT:CASS:2024:12345 2024年のイタリアの破毀院の判決のために。
インデックス作成のための公式ソース
- ユーロレックス: SPARQL API と一括ダウンロードを使用した EU 文章
- イタリア陪審員: イタリア法学 (破毀院、TAR、国務院)
- デジュール (ジュフレ): APIを備えた商用データベース
- 無料法律プロジェクト (CourtListener): 米国の法学、オープンソース
- EU司法裁判所: XML/JSON 出力を備えた curia.europa.eu API
ベストプラクティスとアンチパターン
避けるべきアンチパターン
- 文全体の埋め込み: 長い文章は必ず セクション (最大値、事実、法律、デバイス) ごとにチャンク化されます。の埋め込み 10,000 語を使用すると、意味上の意味が「薄まります」。
- スコアのしきい値が低すぎます: すべての結果をスコア付きで返す > 0.3 にはノイズが多すぎます。 0.65 から始めて、ユーザーのフィードバックをもとに調整します。
- 出願日を無視します。 1990年の法律に関する判決 2005 年に廃止され、現在の研究とは無関係です。常に時間フィルター。
- E5 モデルのプレフィックスなしの埋め込み: E5 モデルには次のものが必要です 「クエリ」と「パッセージ」では接頭辞が異なります。これを無視すると、パフォーマンスが最大 15% 低下します。
結論
ベクトル埋め込みに基づく法学検索エンジンは全文検索よりも優れたパフォーマンスを発揮します 法律専門家にとって重要なすべての指標に関する伝統的な手法: 関連する先例、用語のバリエーションに対する堅牢性、および検索能力 異なるケース間の概念的な類似性。
完全なパイプライン — 特殊なエンベディング + FAISS + BM25 ハイブリッド + クロスエンコーダー — レイテンシのある数百万文のデータセットで本番環境のパフォーマンスを実現 200ミリ秒未満。この記事で紹介されているコードは出発点です 最新の LegalTech プラットフォームの中核を構築するのに最適です。
リーガルテックとAIシリーズ
- 契約分析のための NLP: OCR から理解まで
- 電子証拠開示プラットフォームのアーキテクチャ
- 動的ルールエンジンによるコンプライアンスの自動化
- 法的合意のためのスマートコントラクト: Solidity と Vyper
- 生成 AI による法的文書の要約
- 検索エンジンの法則: ベクトル埋め込み (この記事)
- Scala でのデジタル署名と文書認証
- データプライバシーとGDPRコンプライアンスシステム
- 法務 AI アシスタント (法務副操縦士) の構築
- LegalTech データ統合パターン