Právní vyhledávač s vektorovými vložkami
Právník hledající judikaturní precedenty týkající se „náhrady škod za vadné výrobky“ může chybět relevantní rozhodnutí, která používají formulaci „odpovědnost výrobce za neřesti dobra“. Tradiční fulltextové vyhledávání založené na přesné shodě klíčových slov, systematicky selhává v oblasti, kde lze popsat stejnou normu nebo případ s různou terminologií v různých dobách nebo jurisdikcích.
I Vektorové vložení e la Sémantické vyhledávání tento problém řeší ke kořenu: místo srovnávání slov srovnávají význam. Dotaz o „neplatnost smlouvy z důvodu vady souhlasu“ automaticky najde věty o „zrušitelnosti právní transakce kvůli zásadní chybě“, protože tyto dva pojmy žijí v podobných regionech vektorový prostor. V tomto článku vytváříme vyhledávač judikatury od nuly produkční připravenost s Pythonem, specializované modely vkládání pro právo a vektorová databáze.
Co se naučíte
- Architektura sémantického vyhledávače jurisprudence
- Specializované modely vkládání pro právní doménu (legal-BERT, ModernBERT)
- Efektivní indexování pomocí FAISS a Pinecone
- Hybridní vyhledávání: BM25 + vektorová podobnost pro maximální přesnost
- Přehodnocení pomocí křížového kodéru pro konečné výsledky
- REST API s FastAPI pro integraci do aplikací LegalTech
Architektura systému
Moderní právní vyhledávač se skládá ze tří hlavních vrstev:
- Potrubí příjmu: stahuje, normalizuje a zpracovává věty ze zdrojů oficiální (EUR-Lex, ECLI API, DeJure, Cassation). Vytváří rozdělené dokumenty připravené k použití vkládání.
- Indexovací stroj: generuje vektorové vložení pro každý blok vět a indexuje je ve vektorovém úložišti (FAISS pro self-hosted, Pinecone pro spravované).
- Query Engine: zpracovává uživatelské dotazy, převádí je na vložení, provede vektorové vyhledávání, použije hybridní přehodnocení a vrátí výsledky s ověřitelné citace.
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)Výběr modelu vkládání
Výběr modelu vkládání je rozhodující pro kvalitu právního výzkumu.
Univerzální modely jako např text-embedding-3-large od OpenAI
dosahují dobrých výsledků, ale modely předem vyškolené na právních korpusech dosahují lepších výsledků
výrazně všeobecný pro specializované právní úkoly.
| Model | Rozměry | Specializace | NDCG@10 legální | Nasazení |
|---|---|---|---|---|
| text-embedding-3-velký | 3072 | Generál | 0,71 | API (OpenAI) |
| nlpaueb/legal-bert-base | 768 | právní (EN) | 0,79 | Objímání tváře |
| Free-Law-Project/modernbert | 768 | judikatura (EN) | 0,83 | Objímání tváře |
| dbmdz/bert-base-italian | 768 | italština (obecně) | 0,74 | Objímání tváře |
| Plavební zákon-2 | 1024 | Právní (EN+vícejazyčné) | 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]Indexování pomocí FAISS
FAISS (Facebook AI Similarity Search) je referenční knihovna pro vyhledávání vektorů vysoký výkon na velkých souborech dat. Za soubor 10 milionů vět a IVF index (Inverted File Index) s kvantizací produktu (PQ) umožňuje udržovat doby odezvy méně než 100 ms na komoditních CPU.
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']Hybridní vyhledávání: BM25 + vektorová podobnost
Čistě sémantické vyhledávání vyniká v hledání souvisejících pojmů, ale může minout přesné shody s přesnými regulačními odkazy (např. „článek 1453 c.c.“, „legislativní vyhláška 231/2001“). Vyhledávání BM25 (založené na klíčových slovech) je vynikající pro přesné shody, ale slepé k sémantice. Přístup hybridní kombinuje obojí pro maximalizaci vybavování a přesnosti.
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]Přehodnocení křížového kodéru
Po hybridním hledání krok přeřazení s křížovým kodérem dále zlepšuje přesnost. Křížové kodéry zpracují pár (dotaz, dokument) společně vytvářejí mnohem přesnější skóre relevance než bi-kodéry, ale s vyšší výpočetní náklady — to je důvod, proč se následně používají pouze u nejlepších K kandidátů počáteční vyhledávání.
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]REST API s FastAPI
Vyhledávač vystavujeme jako REST mikroslužbu s FastAPI, připravenou k integraci v jakékoli aplikaci LegalTech.
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}Úvahy o ECLI a evropských standardech
L'Identifikátor evropské judikatury (ECLI) a evropská norma pro
jedinečná identifikace vět. ECLI má tvar:
ECLI:{country}:{court}:{year}:{number}
- například ECLI:IT:CASS:2024:12345 pro rozsudek italské kasace z roku 2024.
Oficiální zdroje pro indexování
- EUR-Lex: EU věty s SPARQL API a hromadné stahování
- ItálieJuries: Italská judikatura (Cassation, TAR, Státní rada)
- DeJure (Giuffre): komerční databáze s API
- Free Law Project (CourtListener): Americká jurisprudence, open source
- Soudní dvůr EU: API curia.europa.eu s výstupem XML/JSON
Osvědčené postupy a anti-vzorce
Anti-vzory, kterým je třeba se vyhnout
- Vložení celého textu věty: dlouhé věty musí být rozděleny podle sekcí (maximum, skutečnost, zákon, zařízení). Vložení 10 000 slov „ředí“ sémantický význam.
- Hranice skóre je příliš nízká: vrátit všechny výsledky se skóre > 0,3 zahrnuje příliš mnoho šumu. Začněte s 0,65 a kalibrujte na základě zpětné vazby od uživatelů.
- Ignorujte datum podání: rozsudek z roku 1990 o legislativě zrušena v roce 2005 a pro současný výzkum irelevantní. Vždy časový filtr.
- Vložení bez předpony pro modely E5: Modely E5 vyžadují různé předpony pro "dotaz" vs "průchod". Jeho ignorování snižuje výkon o ~15 %.
Závěry
Právní vyhledávač založený na vektorovém vkládání předčí fulltextové vyhledávání tradiční ve všech metrikách, které jsou důležité pro právníka: odvolání z relevantní precedenty, odolnost vůči terminologickým variacím a schopnost najít pojmové analogie mezi různými případy.
Kompletní potrubí — specializované vestavby + FAISS + hybrid BM25 + křížový kodér — dosahuje produkčního výkonu na datových sadách milionů vět s latencí méně než 200 ms. Kód uvedený v tomto článku je výchozím bodem ideální pro budování srdce moderní platformy LegalTech.
Série LegalTech a AI
- NLP pro analýzu smluv: od OCR k porozumění
- Architektura platformy e-Discovery
- Automatizace shody s dynamickými pravidly
- Chytrá smlouva pro právní dohody: Solidita a Vyper
- Shrnutí právních dokumentů s generativní AI
- Zákon o vyhledávačích: Vkládání vektorů (tento článek)
- Digitální podpis a ověřování dokumentů ve společnosti Scala
- Ochrana osobních údajů a systémy dodržování GDPR
- Vytvoření právního asistenta AI (právní kopilot)
- Vzor integrace dat LegalTech