벡터 임베딩을 사용한 법학 검색 엔진
"불량제품에 대한 손해배상"에 관한 판례를 찾는 변호사 "생산자의 책임"이라는 문구를 사용한 관련 판결을 놓칠 수 있습니다. 좋은 악덕". 정확한 키워드 일치를 기반으로 하는 기존의 전체 텍스트 검색, 동일한 규범이나 사례가 설명될 수 있는 영역에서는 체계적으로 실패합니다. 시대나 관할권에 따라 다른 용어를 사용합니다.
I 벡터 임베딩 그리고 의미 검색 그들은 이 문제를 해결해 뿌리까지: 단어를 비교하는 대신, 그들은 단어를 비교합니다. 의미. 에 대한 쿼리 "동의의 결함으로 인한 계약 무효"는 "무효"에 대한 문장을 자동으로 찾습니다. 본질적 오류로 인한 법적 거래"라는 두 개념이 유사한 영역에 있기 때문입니다. 벡터 공간. 이 기사에서는 판례 검색 엔진을 처음부터 구축합니다. Python, 법률용 특수 임베딩 모델 및 벡터 데이터베이스를 사용하여 프로덕션 준비가 완료되었습니다.
무엇을 배울 것인가
- 법학을 위한 의미 검색 엔진의 아키텍처
- 법적 도메인을 위한 전문 임베딩 모델(legal-BERT, ModernBERT)
- FAISS와 Pinecone을 이용한 효율적인 인덱싱
- 하이브리드 검색: 최대 정밀도를 위한 BM25 + 벡터 유사성
- 최종 결과를 위해 크로스 인코더로 다시 순위 지정
- LegalTech 애플리케이션에 통합하기 위한 FastAPI가 포함된 REST API
시스템 아키텍처
현대 법학 검색 엔진은 세 가지 주요 계층으로 구성됩니다.
- 수집 파이프라인: 소스로부터 문장을 다운로드, 정규화 및 처리합니다. 공식 (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 오픈AI로
좋은 결과를 생성하지만 법적 말뭉치에 대해 사전 훈련된 모델이 더 나은 성능을 발휘합니다.
전문적인 법률 업무에 매우 범용적입니다.
| 모델 | 치수 | 전문화 | NDCG@10 법률 | 전개 |
|---|---|---|---|---|
| 텍스트 삽입-3-대형 | 3072 | 일반적인 | 0.71 | API(오픈AI) |
| nlpaueb/법률-버트-베이스 | 768 | 법률(EN) | 0.79 | 포옹얼굴 |
| 자유법 프로젝트/modernbert | 768 | 판례법(EN) | 0.83 | 포옹얼굴 |
| dbmdz/bert-base-이탈리아어 | 768 | 이탈리아어(일반) | 0.74 | 포옹얼굴 |
| 항해법-2 | 1024 | 법률(EN+다국어) | 0.86 | API(항해 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의 응답 시간은 100ms 미만입니다.
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 + 벡터 유사성
순수 의미 검색은 관련 개념을 찾는 데 탁월하지만 놓칠 수 있음 정확한 규제 참조와 정확히 일치합니다(예: "1453조 c.c.", "입법령 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 마이크로서비스로 노출합니다. 모든 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}ECLI 및 유럽 표준에 대한 고려 사항
L'유럽 판례 식별자(ECLI) 그리고 유럽 표준
문장의 고유 식별. ECLI의 형식은 다음과 같습니다.
ECLI:{country}:{court}:{year}:{number}
- 예를 들어 ECLI:IT:CASS:2024:12345 2024년 이탈리아 판결의 문장입니다.
색인 생성 공식 소스
- EUR-렉스: SPARQL API 및 대량 다운로드가 포함된 EU 문장
- 이탈리아심사위원단: 이탈리아 법학(Cassation, TAR, Council of State)
- DeJure(주프레): API를 갖춘 상용 데이터베이스
- 무료 법률 프로젝트(CourtListener): 미국 법학, 오픈 소스
- EU 법원: XML/JSON 출력을 포함하는 curia.europa.eu API
모범 사례 및 안티 패턴
피해야 할 안티패턴
- 문장 전체 텍스트 삽입: 긴 문장은 반드시 섹션(최대값, 사실, 법률, 장치)별로 청크됩니다. 임베딩 10,000 단어는 의미론적 의미를 "희석"시킵니다.
- 점수 임계값이 너무 낮음: 모든 결과를 점수와 함께 반환 > 0.3에는 노이즈가 너무 많습니다. 0.65부터 시작하여 사용자 피드백을 통해 보정하세요.
- 제출 날짜를 무시하십시오. 1990년 입법 판결 2005년에 폐지되었으며 현재 연구와 관련이 없습니다. 항상 시간 필터.
- E5 모델에 접두사 없이 포함: E5 모델에는 다음이 필요합니다. "쿼리"와 "통로"에 대한 접두사가 다릅니다. 이를 무시하면 성능이 최대 15%까지 저하됩니다.
결론
벡터 임베딩을 기반으로 한 법학 검색 엔진은 전체 텍스트 검색보다 성능이 뛰어납니다. 법률 전문가에게 중요한 모든 측정 항목에 대한 전통적인 방식: 관련 선례, 용어 변형에 대한 견고성, 검색 능력 다양한 사례 간의 개념적 유추.
완전한 파이프라인 — 전문 임베딩 + FAISS + BM25 하이브리드 + 크로스 인코더 — 대기 시간이 있는 수백만 개의 문장으로 구성된 데이터 세트에서 생산 성능을 달성합니다. 200ms 미만. 이 글에 제시된 코드가 출발점이다 현대 LegalTech 플랫폼의 핵심을 구축하는 데 이상적입니다.
LegalTech 및 AI 시리즈
- 계약 분석을 위한 NLP: OCR에서 이해까지
- e-Discovery 플랫폼 아키텍처
- 동적 규칙 엔진을 통한 규정 준수 자동화
- 법적 계약을 위한 스마트 계약: Solidity 및 Vyper
- Generative AI를 사용한 법률 문서 요약
- 검색 엔진 법칙: 벡터 임베딩(이 문서)
- Scala의 디지털 서명 및 문서 인증
- 데이터 개인정보 보호 및 GDPR 규정 준수 시스템
- 법률 AI 보조원 구축(Legal Copilot)
- LegalTech 데이터 통합 패턴