RAG：使用 ChromaDB 和 Mistral

ollama pull ministral-3:8b
ollama pull bge-m3

mkdir mon-rag && cd mon-rag
python3 -m venv venv
source venv/bin/activate   # .\venv\Scripts\activate sous Windows
pip install chromadb pypdf requests

from pathlib import Path
import chromadb
from chromadb.utils.embedding_functions.ollama_embedding_function import OllamaEmbeddingFunction
from pypdf import PdfReader

ef = OllamaEmbeddingFunction(url="http://localhost:11434", model_name="bge-m3")
coll = chromadb.PersistentClient(path="./chroma_db").get_or_create_collection("mes_docs", embedding_function=ef)

def lire(path: Path) -> str:
    if path.suffix.lower() == ".pdf":
        return "\n\n".join(p.extract_text() or "" for p in PdfReader(str(path)).pages)
    return path.read_text(encoding="utf-8", errors="ignore")

def decouper(texte: str, max_chars=1800):
    """Regroupe des paragraphes entiers jusqu'à max_chars ; un paragraphe trop long est coupé."""
    chunks, courant = [], ""
    for para in (p.strip() for p in texte.split("\n\n")):
        if not para:
            continue
        while len(para) > max_chars:
            if courant:
                chunks.append(courant); courant = ""
            chunks.append(para[:max_chars]); para = para[max_chars:]
        if len(courant) + len(para) + 2 > max_chars and courant:
            chunks.append(courant); courant = ""
        courant = (courant + "\n\n" + para).strip()
    if courant:
        chunks.append(courant)
    return chunks

n = 0
for path in sorted(Path("docs").rglob("*")):
    if path.suffix.lower() not in {".pdf", ".md", ".txt"}:
        continue
    chunks = decouper(lire(path))
    if not chunks:
        print(f"  ! {path.name} : aucun texte extrait (PDF scanné ?)")
        continue
    for i in range(0, len(chunks), 32):  # par lots de 32
        lot = chunks[i:i + 32]
        coll.upsert(
            ids=[f"{path.name}-{i + j}" for j in range(len(lot))],
            documents=lot,
            metadatas=[{"source": path.name}] * len(lot),
        )
    n += len(chunks)
    print(f"  + {path.name} : {len(chunks)} passages")
print(f"Terminé : {n} passages indexés")

import sys, requests
import chromadb
from chromadb.utils.embedding_functions.ollama_embedding_function import OllamaEmbeddingFunction

MODELE = "ministral-3:8b"
ef = OllamaEmbeddingFunction(url="http://localhost:11434", model_name="bge-m3")
coll = chromadb.PersistentClient(path="./chroma_db").get_collection("mes_docs", embedding_function=ef)

SYSTEME = (
    "Tu réponds en français, uniquement à partir des passages fournis. "
    "Si la réponse n'y figure pas, dis-le clairement au lieu de deviner. "
    "Termine chaque affirmation par le nom du fichier source entre crochets."
)

def repondre(question: str, k: int = 5):
    res = coll.query(query_texts=[question], n_results=k)
    passages = list(zip(res["documents"][0], res["metadatas"][0]))
    contexte = "\n\n---\n\n".join(f"[{m['source']}]\n{p}" for p, m in passages)
    r = requests.post("http://localhost:11434/api/chat", json={
        "model": MODELE,
        "stream": False,
        "options": {"temperature": 0.2, "num_ctx": 8192},
        "messages": [
            {"role": "system", "content": SYSTEME},
            {"role": "user", "content": f"PASSAGES :\n{contexte}\n\nQUESTION : {question}"},
        ],
    }, timeout=300)
    r.raise_for_status()
    return r.json()["message"]["content"], sorted({m["source"] for _, m in passages})

if __name__ == "__main__":
    q = " ".join(sys.argv[1:]) or input("Question : ")
    reponse, sources = repondre(q)
    print("\n" + reponse)
    print("\nSources consultées :", ", ".join(sources))

python index.py
python ask.py "Quel est le délai de préavis prévu au contrat ?"

import sys
import chromadb
from chromadb.utils.embedding_functions.ollama_embedding_function import OllamaEmbeddingFunction

ef = OllamaEmbeddingFunction(url="http://localhost:11434", model_name="bge-m3")
coll = chromadb.PersistentClient(path="./chroma_db").get_collection("mes_docs", embedding_function=ef)
res = coll.query(query_texts=[" ".join(sys.argv[1:])], n_results=8)
for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]):
    print(f"{dist:.3f}  {meta['source']}  {doc[:120]!r}")

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