Motor de căutare jurisprudențial cu încorporare vectorială
Un avocat în căutarea precedentelor jurisprudențiale privind „despăgubirea daunelor pentru produse defecte” poate lipsi hotărârile relevante care folosesc expresia „responsabilitatea producătorului pentru vicii ale binelui”. Căutare tradițională de text integral, bazată pe potrivirea exactă a cuvintelor cheie, eșuează sistematic într-un domeniu în care poate fi descrisă aceeași normă sau caz cu terminologii diferite în vremuri sau jurisdicţii diferite.
I Înglobări vectoriale iar cel Căutare semantică ei rezolvă această problemă la rădăcină: în loc să compare cuvintele, ei compară sens. O întrebare despre „invaliditatea contractului din cauza viciului de consimțământ” găsește automat propoziții despre „anulabilitate a tranzacţiei juridice din cauza erorii esenţiale” deoarece cele două concepte trăiesc în regiuni similare ale spațiu vectorial. În acest articol construim un motor de căutare a jurisprudenței de la zero gata de producție cu Python, modele specializate de încorporare pentru drept și o bază de date vectorială.
Ce vei învăța
- Arhitectura unui motor de căutare semantică pentru jurisprudență
- Modele de încorporare specializate pentru domeniul juridic (legal-BERT, ModernBERT)
- Indexare eficientă cu FAISS și Pinecone
- Căutare hibridă: BM25 + similaritate vectorială pentru precizie maximă
- Re-clasificare cu cross-encoder pentru rezultatele finale
- REST API cu FastAPI pentru integrare în aplicațiile LegalTech
Arhitectura sistemului
Un motor de căutare jurisprudențial modern este alcătuit din trei straturi principale:
- Conducta de ingestie: descarcă, normalizează și procesează propoziții din surse oficial (EUR-Lex, ECLI API, DeJure, Cassation). Produce documente fragmentate gata de utilizare încorporarea.
- Motor de indexare: generează înglobări vectoriale pentru fiecare fragment de propoziție și le indexează într-un magazin de vectori (FAISS pentru self-hosted, Pinecone pentru gestionat).
- Motor de interogări: procesează interogările utilizatorilor, le transformă în înglobări, efectuează căutare vectorială, aplică reclasificare hibridă și returnează rezultatele cu citate verificabile.
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)Alegerea modelului de încorporare
Alegerea modelului de încorporare este esențială pentru calitatea cercetării juridice.
Modele de uz general precum text-embedding-3-large de OpenAI
produc rezultate bune, dar modelele pregătite în prealabil pe corpus juridic depășesc
semnificativ cu scop general pe sarcini juridice specializate.
| Model | Dimensiuni | Specializare | NDCG@10 legal | Desfăşurare |
|---|---|---|---|---|
| text-incorporare-3-mare | 3072 | General | 0,71 | API (OpenAI) |
| nlpaueb/legal-bert-base | 768 | Legal (EN) | 0,79 | HuggingFace |
| Free-Lew-Project/modernbert | 768 | Jurisprudență (EN) | 0,83 | HuggingFace |
| dbmdz/bert-base-italian | 768 | italiană (general) | 0,74 | HuggingFace |
| Legea-călătorii-2 | 1024 | Legal (EN+multilingv) | 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]Indexarea cu FAISS
FAISS (Facebook AI Similarity Search) este biblioteca de referință pentru căutarea vectorială performanță ridicată pe seturi mari de date. Pentru o colectare de 10 milioane de sentințe, a Indexul IVF (Inverted File Index) cu cuantizarea produsului (PQ) vă permite să mențineți timpi de răspuns mai mici de 100 ms pe procesoarele de bază.
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']Căutare hibridă: BM25 + Similitudine vectorială
Căutarea pur semantică excelează în găsirea de concepte înrudite, dar poate rata potriviri exacte pe referințe precise de reglementare (ex. „art. 1453 c.c.”, „Decretul legislativ 231/2001”). Căutarea BM25 (bazată pe cuvinte cheie) este excelentă pentru potriviri exacte, dar oarbă la semantică. Abordarea hibrid combină ambele pentru a maximiza amintirea și precizia.
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]Re-clasificare încrucișată
După căutarea hibridă, un pas de re-clasificare cu cross-encoder îmbunătățește în continuare precizia. Codificatorii încrucișați procesează perechea (interogare, document) împreună, producând un scor de relevanță mult mai precis decât bi-encodere, dar cu cost de calcul mai mare - de aceea sunt utilizate numai pe candidații de top-K ulterior căutarea inițială.
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 cu FastAPI
Expunem motorul de căutare ca un microserviciu REST cu FastAPI, gata de integrare în orice aplicație 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}Considerații privind ECLI și standardele europene
L'Identificator european de jurisprudență (ECLI) și standardul european pentru
identificarea unică a propoziţiilor. Un ECLI are forma:
ECLI:{country}:{court}:{year}:{number}
- de exemplu ECLI:IT:CASS:2024:12345 pentru o sentință a Casației italiene din 2024.
Surse oficiale pentru indexare
- EUR-Lex: Propoziții UE cu API SPARQL și descărcare în bloc
- Juriile din Italia: Jurisprudența italiană (Casație, TAR, Consiliul de Stat)
- DeJure (Giuffre): bază de date comercială cu API
- Proiect de drept gratuit (CourtListener): Jurisprudența SUA, sursă deschisă
- Curtea de Justiție a UE: curia.europa.eu API cu ieșire XML/JSON
Cele mai bune practici și anti-modele
Anti-modele de evitat
- Încorporarea întregului text al propoziției: frazele lungi trebuie să fie împărțite în funcție de secțiune (maxim, fapt, lege, dispozitiv). O încorporare a 10.000 de cuvinte „diluează” sensul semantic.
- Pragul de scor prea mic: returnează toate rezultatele cu scor > 0,3 include prea mult zgomot. Începeți cu 0,65 și calibrați cu feedback-ul utilizatorului.
- Ignorați data de depunere: o hotărâre din 1990 privind legislația abrogat în 2005 și irelevant pentru cercetările actuale. Întotdeauna filtru de timp.
- Încorporare fără prefix pentru modelele E5: Modelele E5 necesită diferite prefixe pentru „interogare” vs „pasare”. Ignorarea acesteia degradează performanța cu ~15%.
Concluzii
Un motor de căutare jurisprudențial bazat pe înglobări vectoriale depășește căutarea full-text tradițional cu privire la toate valorile care contează pentru un profesionist juridic: rechemarea precedente relevante, robustețe la variațiile terminologice și capacitatea de a găsi analogii conceptuale între diferite cazuri.
Conducta completă — încorporare specializate + FAISS + hibrid BM25 + codificator încrucișat — realizează performanțe de producție pe seturi de date de milioane de propoziții cu latențe mai puțin de 200 ms. Codul prezentat în acest articol este punctul de plecare ideal pentru construirea inimii unei platforme LegalTech moderne.
Seria LegalTech și AI
- NLP pentru analiza contractelor: de la OCR la înțelegere
- Arhitectura platformei e-Discovery
- Automatizarea conformității cu Dynamic Rules Engine
- Contract inteligent pentru acorduri juridice: Solidity și Vyper
- Rezumat documente legale cu IA generativă
- Legea motoarelor de căutare: încorporare vectorială (acest articol)
- Semnătura digitală și autentificarea documentelor la Scala
- Confidențialitatea datelor și sisteme de conformitate GDPR
- Crearea unui asistent legal AI (copilot juridic)
- Model de integrare a datelor LegalTech