基于判例法的 RAG Légifrance

mkdir -p dila/CASS && cd dila/CASS
curl -O https://echanges.dila.gouv.fr/OPENDATA/CASS/Freemium_cass_global_20250713-140000.tar.gz
tar xzf Freemium_cass_global_20250713-140000.tar.gz

# Puis chaque archive incrémentale, dans l'ordre chronologique
curl -s https://echanges.dila.gouv.fr/OPENDATA/CASS/ | grep -o 'CASS_2026[0-9-]*\.tar\.gz' | sort -u > maj.txt
for f in $(cat maj.txt); do curl -sO https://echanges.dila.gouv.fr/OPENDATA/CASS/$f && tar xzf $f; done

from lxml import etree
from pathlib import Path
import re

def parser(chemin):
    racine = etree.parse(str(chemin)).getroot()

    def val(xp):
        e = racine.find(xp)
        return (e.text or '').strip() if e is not None else None

    contenu = racine.find('.//TEXTE/BLOC_TEXTUEL/CONTENU')
    if contenu is None:
        return None
    for br in contenu.iter('br'):
        br.tail = '\n' + (br.tail or '')
    texte = ''.join(contenu.itertext())
    texte = re.sub(r'[ \t]+', ' ', re.sub(r'\n\s*\n+', '\n', texte)).strip()
    return {
        'id': val('.//META_COMMUN/ID'),
        'titre': val('.//META_JURI/TITRE'),
        'date': val('.//META_JURI/DATE_DEC'),
        'juridiction': val('.//META_JURI/JURIDICTION'),
        'solution': val('.//META_JURI/SOLUTION'),
        'pourvoi': val('.//META_JURI_JUDI/NUMEROS_AFFAIRES/NUMERO_AFFAIRE'),
        'formation': val('.//META_JURI_JUDI/FORMATION'),
        'ecli': val('.//META_JURI_JUDI/ECLI'),
        'texte': texte,
    }

def decisions(dossier):
    for chemin in Path(dossier).rglob('*.xml'):
        try:
            d = parser(chemin)
            if d:
                yield d
        except etree.XMLSyntaxError as e:
            print('ignoré', chemin, e)

import re

COUPURE = re.compile(r"\n(?=(?:Faits et procédure|Examen des moyens|Sur le |Énoncé du moyen|Enoncé du moyen|Réponse de la Cour|PAR CES MOTIFS))")

def extraits(d, max_car=2200):
    en_tete = f"{d['titre']} ({d['ecli']})\n"
    sortie, tampon = [], ''
    for bloc in COUPURE.split(d['texte']):
        for para in bloc.split('\n'):
            if len(tampon) + len(para) > max_car and tampon:
                sortie.append(en_tete + tampon)
                tampon = ''
            tampon += para + '\n'
    if tampon.strip():
        sortie.append(en_tete + tampon)
    return sortie

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient, models

emb = SentenceTransformer('BAAI/bge-m3', device='cuda')
emb.max_seq_length = 1024
client = QdrantClient(host='localhost', port=6333)
if not client.collection_exists('cass'):
    client.create_collection('cass', vectors_config=models.VectorParams(
        size=1024, distance=models.Distance.COSINE))

def indexer(dossier, taille_lot=64):
    n, lot = 0, []
    def vider():
        vecs = emb.encode([x[0] for x in lot], batch_size=32, normalize_embeddings=True)
        client.upsert('cass', points=[models.PointStruct(id=x[2], vector=v.tolist(), payload=x[1])
                                       for x, v in zip(lot, vecs)])
        lot.clear()
    for d in decisions(dossier):
        for texte in extraits(d):
            n += 1
            payload = {k: d[k] for k in ('titre', 'ecli', 'pourvoi', 'juridiction', 'formation', 'solution')}
            payload['date_int'] = int(d['date'].replace('-', ''))
            payload['texte'] = texte
            lot.append((texte, payload, n))
            if len(lot) >= taille_lot:
                vider()
    if lot:
        vider()

def chercher(question, depuis=None, formation=None, k=8):
    conds = []
    if depuis:
        conds.append(models.FieldCondition(key='date_int', range=models.Range(gte=depuis)))
    if formation:
        conds.append(models.FieldCondition(key='formation', match=models.MatchValue(value=formation)))
    vec = emb.encode(question, normalize_embeddings=True).tolist()
    res = client.query_points('cass', query=vec, limit=k,
                              query_filter=models.Filter(must=conds) if conds else None)
    return res.points

for p in chercher('rupture anticipée du CDD par l employeur', depuis=20230101):
    print(p.payload['titre'], p.payload['pourvoi'], round(p.score, 3))

Tu es un assistant de recherche en jurisprudence. Tu réponds uniquement à partir des
extraits fournis. Pour chaque affirmation, cite entre crochets le numéro de pourvoi
et l'ECLI de la décision. Si les extraits ne permettent pas de répondre, écris :
aucune décision retrouvée. Tu ne donnes pas d'avis juridique.

有反馈、发现了错误，或想补充说明？请告诉我们，让这份指南对每个人都更有帮助。