RAG locale: un agente AI che legge i tuoi documenti

Architettura RAG Locale

┌─────────────────────────────────────────────────────────────────────────────┐
│                           RAG PIPELINE LOCALE                                │
│                                                                              │
│  ┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌──────────────┐   │
│  │ Documenti│───►│  Chunking    │───►│  Embedding  │───►│  ChromaDB    │   │
│  │ (PDF,    │    │  (Recursive, │    │  (nomic-    │    │  (Vector     │   │
│  │  MD, TXT,│    │   Semantic,  │    │   embed-    │    │   Store      │   │
│  │  HTML)   │    │   Markdown)  │    │   text)     │    │   Persistente)│   │
│  └──────────┘    └──────────────┘    └─────────────┘    └──────┬───────┘   │
│                                                                   │         │
│                              ┌──────────────────────────────────┘         │
│                              ▼                                                │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────────────────────┐  │
│  │   Query      │───►│  Retrieval   │───►│  Generation (LLM + Context)  │  │
│  │   Utente     │    │  (Similarity │    │  ConversationalRetrievalChain│  │
│  │              │    │   / MMR)     │    │  + Source Citations          │  │
│  └──────────────┘    └──────────────┘    └──────────────────────────────┘  │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Dipendenze

# requirements.txt (aggiungi a Cap. 4)
chromadb>=0.4.0
langchain-chroma>=0.1.0
langchain-experimental>=0.0.60  # SemanticChunker
pypdf>=3.17.0
python-docx>=1.1.0
unstructured>=0.11.0  # per HTML, PDF complessi
markdown>=3.5.0

pip install -r requirements.txt

1. Embedding Locale: nomic-embed-text via Ollama

# embeddings.py
from langchain_ollama import OllamaEmbeddings
from langchain_core.embeddings import Embeddings
from config import config  # Cap. 3


def create_embeddings(
    model: str = "nomic-embed-text:latest",
    base_url: str = None,
) -> Embeddings:
    """
    Crea embedding model locale via Ollama.
    
    nomic-embed-text: 768 dimensioni, context 8192, ottimo per RAG.
    Alternative: mxbai-embed-large (1024 dim), bge-m3 (1024 dim, multilingue).
    """
    return OllamaEmbeddings(
        model=model,
        base_url=base_url or config.base_url,
    )


# Test veloce
if __name__ == "__main__":
    emb = create_embeddings()
    vectors = emb.embed_documents([
        "Il gatto siede sul tappeto",
        "Il cane corre nel parco",
    ])
    query_vec = emb.embed_query("Animali domestici")
    print(f"Dim documenti: {len(vectors[0])}")  # 768
    print(f"Dim query: {len(query_vec)}")       # 768

Perché nomic-embed-text? Produce vettori da 768 dimensioni, supporta un contesto fino a 8192 token (in Ollama il valore predefinito di num_ctx è più basso: alzalo se indicizzi chunk lunghi), ha licenza Apache 2.0 e circa 137 milioni di parametri. Gira anche su CPU.

2. ChromaDB: Vector Store Persistente

# vectorstore.py
import chromadb
from chromadb.config import Settings
from langchain_chroma import Chroma
from langchain_core.vectorstores import VectorStore
from embeddings import create_embeddings
import os


CHROMA_PERSIST_DIR = "./chroma_db"


def create_chroma_client(
    persist_dir: str = CHROMA_PERSIST_DIR,
    anonymized_telemetry: bool = False,
) -> chromadb.Client:
    """Client ChromaDB con persistenza su disco."""
    os.makedirs(persist_dir, exist_ok=True)
    return chromadb.PersistentClient(
        path=persist_dir,
        settings=Settings(
            anonymized_telemetry=anonymized_telemetry,
            allow_reset=True,
        ),
    )


def get_vectorstore(
    collection_name: str = "documents",
    persist_dir: str = CHROMA_PERSIST_DIR,
    embedding_model: str = "nomic-embed-text:latest",
) -> VectorStore:
    """
    Restituisce LangChain Chroma vector store pronto all'uso.
    Crea collection se non esiste.
    """
    client = create_chroma_client(persist_dir)
    embeddings = create_embeddings(embedding_model)
    
    vectorstore = Chroma(
        client=client,
        collection_name=collection_name,
        embedding_function=embeddings,
        persist_directory=persist_dir,
    )
    return vectorstore


def list_collections(persist_dir: str = CHROMA_PERSIST_DIR) -> list[str]:
    client = create_chroma_client(persist_dir)
    # Nelle versioni recenti di ChromaDB list_collections() restituisce già i nomi
    return [getattr(c, "name", c) for c in client.list_collections()]


def delete_collection(collection_name: str, persist_dir: str = CHROMA_PERSIST_DIR):
    client = create_chroma_client(persist_dir)
    client.delete_collection(collection_name)


# Uso
if __name__ == "__main__":
    vs = get_vectorstore("my_docs")
    print(f"Collection: {vs._collection.name}")
    print(f"Documenti: {vs._collection.count()}")

3. Document Loading & Chunking Strategie

# document_processing.py
from langchain_core.documents import Document
from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    UnstructuredHTMLLoader,
    DirectoryLoader,
)
from embeddings import create_embeddings
from typing import List, Optional
import os


# === Text Splitters ===

def get_recursive_splitter(
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
    separators: Optional[List[str]] = None,
) -> RecursiveCharacterTextSplitter:
    """Splitter generico: prova separatori in ordine, ricorsivo."""
    return RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=separators or ["\n\n", "\n", ". ", " ", ""],
        length_function=len,
        is_separator_regex=False,
    )


def get_markdown_splitter(
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
) -> MarkdownHeaderTextSplitter:
    """Splitter per Markdown: preserva struttura header (H1, H2, H3)."""
    headers_to_split_on = [
        ("#", "header_1"),
        ("##", "header_2"),
        ("###", "header_3"),
    ]
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on,
    )
    # Poi applica recursive sui chunk grandi
    return splitter


def get_semantic_splitter(
    breakpoint_threshold_type: str = "percentile",
    breakpoint_threshold_amount: float = 95.0,
    min_chunk_size: int = 100,
) -> SemanticChunker:
    """Splitter semantico: usa embedding per trovare boundary naturali."""
    embeddings = create_embeddings()
    return SemanticChunker(
        embeddings=embeddings,
        breakpoint_threshold_type=breakpoint_threshold_type,
        breakpoint_threshold_amount=breakpoint_threshold_amount,
        min_chunk_size=min_chunk_size,
    )


# === Document Loaders ===

def load_documents_from_directory(
    directory: str,
    glob_pattern: str = "**/*",
    chunk_strategy: str = "recursive",  # "recursive" | "markdown" | "semantic"
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
) -> List[Document]:
    """
    Carica tutti i documenti da directory, applica chunking, restituisce chunk pronti per indexing.
    
    Supporta: .pdf, .md, .txt, .html (via unstructured per Markdown e HTML)
    """
    # Loader per estensione
    loader_mapping = {
        ".pdf": PyPDFLoader,
        ".md": UnstructuredMarkdownLoader,
        ".txt": TextLoader,
        ".html": UnstructuredHTMLLoader,
        ".htm": UnstructuredHTMLLoader,
    }
    
    all_docs = []
    
    for root, dirs, files in os.walk(directory):
        for file in files:
            ext = os.path.splitext(file)[1].lower()
            if ext not in loader_mapping:
                continue
            
            file_path = os.path.join(root, file)
            try:
                loader = loader_mapping[ext](file_path)
                docs = loader.load()
                
                # Aggiungi metadata source
                for doc in docs:
                    doc.metadata.update({
                        "source": file_path,
                        "filename": file,
                        "extension": ext,
                        "rel_path": os.path.relpath(file_path, directory),
                    })
                
                all_docs.extend(docs)
            except Exception as e:
                print(f"Errore caricamento {file_path}: {e}")
    
    # Chunking
    if chunk_strategy == "markdown":
        splitter = get_markdown_splitter()
        # MarkdownHeaderTextSplitter restituisce già chunk per header
        # Applica recursive secondario per chunk > chunk_size
        recursive = get_recursive_splitter(chunk_size, chunk_overlap)
        chunked = []
        for doc in all_docs:
            # split_text() restituisce Document: l'header resta nei metadata
            for md_chunk in splitter.split_text(doc.page_content):
                metadata = {**doc.metadata, **md_chunk.metadata}
                if len(md_chunk.page_content) > chunk_size:
                    for sc in recursive.split_text(md_chunk.page_content):
                        chunked.append(Document(page_content=sc, metadata=metadata.copy()))
                else:
                    chunked.append(Document(page_content=md_chunk.page_content, metadata=metadata))
        return chunked
    
    elif chunk_strategy == "semantic":
        splitter = get_semantic_splitter()
        return splitter.split_documents(all_docs)
    
    else:  # recursive (default)
        splitter = get_recursive_splitter(chunk_size, chunk_overlap)
        return splitter.split_documents(all_docs)


# === Metadata Enrichment ===

def enrich_metadata(docs: List[Document], **extra_metadata) -> List[Document]:
    """Aggiunge metadata comuni a tutti i chunk."""
    for doc in docs:
        doc.metadata.update(extra_metadata)
    return docs

4. Indexing Pipeline Completa

# indexing.py
from langchain_core.vectorstores import VectorStore
from langchain_core.documents import Document
from vectorstore import get_vectorstore, create_chroma_client
from document_processing import load_documents_from_directory, enrich_metadata
from typing import List, Optional
import hashlib
import time


def compute_doc_hash(content: str) -> str:
    """Hash per deduplicazione."""
    return hashlib.sha256(content.encode()).hexdigest()[:16]


def index_documents(
    source_dir: str,
    collection_name: str = "documents",
    chunk_strategy: str = "recursive",
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
    metadata: Optional[dict] = None,
    batch_size: int = 100,
) -> dict:
    """
    Pipeline completa: load → chunk → embed → store in ChromaDB.
    Restituisce statistiche.
    """
    start = time.time()
    
    # 1. Load & chunk
    print(f"Caricamento documenti da {source_dir}...")
    docs = load_documents_from_directory(
        source_dir,
        chunk_strategy=chunk_strategy,
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    print(f"  {len(docs)} chunk creati")
    
    # 2. Enrich metadata
    base_metadata = {
        "indexed_at": time.time(),
        "chunk_strategy": chunk_strategy,
        "chunk_size": chunk_size,
        "chunk_overlap": chunk_overlap,
    }
    if metadata:
        base_metadata.update(metadata)
    docs = enrich_metadata(docs, **base_metadata)
    
    # 3. Deduplicazione (opzionale)
    seen_hashes = set()
    unique_docs = []
    for doc in docs:
        h = compute_doc_hash(doc.page_content)
        if h not in seen_hashes:
            seen_hashes.add(h)
            doc.metadata["content_hash"] = h
            unique_docs.append(doc)
    print(f"  {len(unique_docs)} chunk unici (rimossi {len(docs) - len(unique_docs)} duplicati)")
    
    # 4. Vector store
    vs = get_vectorstore(collection_name)
    
    # 5. Batch insert con progress
    total = len(unique_docs)
    for i in range(0, total, batch_size):
        batch = unique_docs[i:i+batch_size]
        vs.add_documents(batch)
        print(f"  Indicizzati {min(i+batch_size, total)}/{total}")
    
    # 6. Persist (Chroma auto-persist su PersistentClient)
    elapsed = time.time() - start
    
    stats = {
        "collection": collection_name,
        "total_chunks": len(docs),
        "unique_chunks": total,
        "duplicates_removed": len(docs) - total,
        "elapsed_seconds": round(elapsed, 2),
        "source_dir": source_dir,
    }
    print(f"Indicizzazione completata in {elapsed:.1f}s")
    return stats


# Uso
if __name__ == "__main__":
    stats = index_documents(
        source_dir="./data/docs",
        collection_name="knowledge_base",
        chunk_strategy="recursive",
        chunk_size=1000,
        chunk_overlap=200,
        metadata={"project": "agente-digitale", "version": "1.0"},
    )
    print(stats)

5. Retrieval: Similarity, MMR, Metadata Filtering

# retrieval.py
from langchain_core.vectorstores import VectorStore
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from vectorstore import get_vectorstore
from typing import List, Optional, Dict, Any
from pydantic import Field


class FilteredRetriever(BaseRetriever):
    """Retriever con metadata filtering built-in."""
    
    vectorstore: VectorStore
    search_type: str = "similarity"  # "similarity" | "mmr" | "similarity_score_threshold"
    k: int = 4
    fetch_k: int = 20  # per MMR
    lambda_mult: float = 0.5  # diversity per MMR
    score_threshold: float = 0.0  # per similarity_score_threshold
    filter_metadata: Optional[Dict[str, Any]] = None
    
    def _get_relevant_documents(self, query: str) -> List[Document]:
        search_kwargs = {"k": self.k}
        
        if self.search_type == "mmr":
            search_kwargs.update({
                "fetch_k": self.fetch_k,
                "lambda_mult": self.lambda_mult,
            })
        elif self.search_type == "similarity_score_threshold":
            search_kwargs["score_threshold"] = self.score_threshold
        
        if self.filter_metadata:
            search_kwargs["filter"] = self.filter_metadata
        
        return self.vectorstore.as_retriever(
            search_type=self.search_type,
            search_kwargs=search_kwargs,
        ).invoke(query)


def create_retriever(
    collection_name: str = "documents",
    search_type: str = "mmr",  # MMR default: più diversità, meno ridondanza
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    filter_metadata: Optional[Dict[str, Any]] = None,
) -> BaseRetriever:
    """Factory retriever configurato."""
    vs = get_vectorstore(collection_name)
    return FilteredRetriever(
        vectorstore=vs,
        search_type=search_type,
        k=k,
        fetch_k=fetch_k,
        lambda_mult=lambda_mult,
        filter_metadata=filter_metadata,
    )


# Esempi filtri metadata
FILTER_EXAMPLES = {
    "by_extension": {"extension": ".pdf"},
    # Sui metadata Chroma supporta $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $and, $or.
    # $regex vale solo per i filtri sul testo (where_document), non per i metadata.
    "by_folder": {"rel_path": {"$in": ["guide/intro.md", "guide/setup.md"]}},
    "by_tag": {"tag": {"$in": ["api", "reference"]}},  # un valore scalare per chiave
    "by_date_range": {"indexed_at": {"$gte": 1704067200, "$lte": 1735689600}},  # 2024
    "combined": {"$and": [{"extension": ".md"}, {"project": "agente-digitale"}]},
}

6. Generation: RetrievalQA & ConversationalRetrievalChain

# generation.py
from langchain.chains import RetrievalQA, ConversationalRetrievalChain
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.messages import BaseMessage
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from llm_setup import get_llm
from retrieval import create_retriever
from typing import List, Dict, Any, Optional


# === Prompt Templates ===

# QA semplice (single-turn)
QA_PROMPT = ChatPromptTemplate.from_template("""Sei un assistente AI che risponde basandosi SOLO sul contesto fornito.
Se l'informazione non è nel contesto, rispondi: "Non trovo questa informazione nei documenti forniti."
Cita sempre le fonti usando il formato [source: filename] alla fine di ogni affermazione.

Contesto:
{context}

Domanda: {question}

Risposta:""")

# Conversazionale (multi-turn con history)
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template("""Data la conversazione seguente e una domanda di follow-up, 
riformula la domanda in una domanda autonoma che incorpori il contesto necessario.

Cronologia della conversazione:
{chat_history}

Follow-up: {question}

Domanda autonoma:""")

ANSWER_PROMPT = ChatPromptTemplate.from_template("""Sei un assistente AI che risponde basandosi SOLO sul contesto fornito.
Se l'informazione non è nel contesto, rispondi: "Non trovo questa informazione nei documenti forniti."
Cita sempre le fonti usando il formato [source: filename] alla fine di ogni affermazione.
Mantieni un tono professionale e tecnico.

Contesto:
{context}

Domanda: {question}

Risposta:""")


# === Chain Factory ===

def create_qa_chain(
    collection_name: str = "documents",
    search_type: str = "mmr",
    k: int = 4,
    return_source_documents: bool = True,
) -> RetrievalQA:
    """Chain QA single-turn con citazioni fonti."""
    
    retriever = create_retriever(
        collection_name=collection_name,
        search_type=search_type,
        k=k,
    )
    
    chain = RetrievalQA.from_chain_type(
        llm=get_llm(),
        chain_type="stuff",  # Tutto context in un prompt
        retriever=retriever,
        return_source_documents=return_source_documents,
        chain_type_kwargs={
            "prompt": QA_PROMPT,
        },
        verbose=True,
    )
    return chain


def create_conversational_qa_chain(
    collection_name: str = "documents",
    search_type: str = "mmr",
    k: int = 4,
    return_source_documents: bool = True,
    return_generated_question: bool = True,
) -> ConversationalRetrievalChain:
    """Chain conversazionale multi-turn con history e citazioni."""
    
    retriever = create_retriever(
        collection_name=collection_name,
        search_type=search_type,
        k=k,
    )
    
    chain = ConversationalRetrievalChain.from_llm(
        llm=get_llm(),
        retriever=retriever,
        return_source_documents=return_source_documents,
        return_generated_question=return_generated_question,
        condense_question_prompt=CONDENSE_QUESTION_PROMPT,
        combine_docs_chain_kwargs={
            "prompt": ANSWER_PROMPT,
        },
        verbose=True,
    )
    return chain


# === LCEL Modern Chain (LangChain Expression Language) ===

def create_lcel_rag_chain(
    collection_name: str = "documents",
    k: int = 4,
) -> RunnablePassthrough:
    """Chain RAG moderna usando LCEL (componibile, streaming-ready)."""
    
    retriever = create_retriever(collection_name=collection_name, k=k)
    
    def format_docs(docs: List[Document]) -> str:
        formatted = []
        for i, doc in enumerate(docs):
            source = doc.metadata.get("filename", "unknown")
            formatted.append(f"[Doc {i+1} - source: {source}]\n{doc.page_content}")
        return "\n\n".join(formatted)
    
    rag_chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | QA_PROMPT
        | get_llm()
        | StrOutputParser()
    )
    
    return rag_chain


# === Streaming con citazioni ===

def stream_rag_with_citations(
    query: str,
    collection_name: str = "documents",
    k: int = 4,
):
    """Streaming response con citazioni inline."""
    chain = create_lcel_rag_chain(collection_name, k)
    
    # Per avere fonti, dobbiamo fare retrieval separato
    retriever = create_retriever(collection_name, k=k)
    docs = retriever.invoke(query)
    
    # Stampa fonti prima
    print("Fonti recuperate:")
    for i, doc in enumerate(docs):
        source = doc.metadata.get("filename", "unknown")
        page = doc.metadata.get("page", "N/A")
        print(f"  [{i+1}] {source} (pagina {page})")
    print()
    
    # Stream risposta
    print("Risposta: ", end="", flush=True)
    for chunk in chain.stream(query):
        print(chunk, end="", flush=True)
    print()


# Demo
if __name__ == "__main__":
    # Single-turn
    qa = create_qa_chain()
    result = qa.invoke({"query": "Cos'è un Agente AI secondo i documenti?"})
    print(f"Risposta: {result['result']}")
    print(f"Fonti: {[d.metadata.get('filename') for d in result['source_documents']]}")
    
    # Conversazionale
    conv_qa = create_conversational_qa_chain()
    chat_history = []
    
    result = conv_qa.invoke({
        "question": "Cos'è un Agente AI?",
        "chat_history": chat_history,
    })
    chat_history.append(("Cos'è un Agente AI?", result["answer"]))
    print(f"Turn 1: {result['answer']}")
    
    result = conv_qa.invoke({
        "question": "Quali modelli consigliate per iniziare?",
        "chat_history": chat_history,
    })
    print(f"Turn 2: {result['answer']}")

7. Source Attribution: Citazioni Automatiche

# citations.py
from langchain_core.documents import Document
from typing import List, Dict, Any
import re


CITATION_PATTERN = re.compile(r'\[source:\s*([^\]]+)\]')


def extract_citations(text: str) -> List[str]:
    """Estrae tutte le citazioni [source: ...] dal testo."""
    return CITATION_PATTERN.findall(text)


def format_citations(
    source_docs: List[Document],
    style: str = "inline",  # "inline" | "footnotes" | "brackets"
) -> str:
    """Formatta citazioni per display."""
    if not source_docs:
        return ""
    
    unique_sources = []
    seen = set()
    for doc in source_docs:
        source = doc.metadata.get("filename", "unknown")
        page = doc.metadata.get("page", "")
        key = f"{source}:{page}"
        if key not in seen:
            seen.add(key)
            unique_sources.append((source, page))
    
    if style == "inline":
        return " | ".join(f"[{s} p.{p}]" if p else f"[{s}]" for s, p in unique_sources)
    elif style == "footnotes":
        lines = []
        for i, (s, p) in enumerate(unique_sources, 1):
            lines.append(f"[{i}] {s}" + (f" (p. {p})" if p else ""))
        return "\n".join(lines)
    elif style == "brackets":
        return " ".join(f"[{i}]" for i in range(1, len(unique_sources)+1))
    
    return ""


def verify_citations(response: str, source_docs: List[Document]) -> Dict[str, Any]:
    """Verifica che le citazioni nella risposta corrispondano a fonti reali."""
    cited = extract_citations(response)
    available = [d.metadata.get("filename", "unknown") for d in source_docs]
    
    valid = [c for c in cited if c in available]
    invalid = [c for c in cited if c not in available]
    
    return {
        "total_citations": len(cited),
        "valid_citations": len(valid),
        "invalid_citations": invalid,
        "coverage": len(valid) / len(cited) if cited else 1.0,
    }

Tabella Parametri RAG: Riferimento Rapido

| Componente | Parametro | Default | Range | Impatto | Tuning | | --- | --- | --- | --- | --- | --- | | Chunking | chunk_size | 1000 | 500-2000 | Più grande = più contesto, meno chunk | 800-1200 per technical docs | | | chunk_overlap | 200 | 100-400 | Overlap previene taglio info | 15-25% di chunk_size | | | strategy | recursive | recursive/markdown/semantic | Semantic = boundary naturali | Markdown per .md, recursive altrimenti | | Embedding | model | nomic-embed-text | nomic/mxbai/bge | Dim, qualità, velocità | nomic per default, bge-m3 per multilingue | | Vector Store | collection | documents | - | Namespace logico | Una per progetto/dominio | | Retrieval | search_type | mmr | similarity/mmr/score_threshold | MMR = diversità, similarity = pertinenza | MMR per RAG generico | | | k | 4 | 3-10 | Più k = più contesto, più rumore | 3-5 per focus, 8-10 per panoramica | | | fetch_k | 20 | 10-50 | Solo MMR: pool iniziale | 3-5x k | | | lambda_mult | 0.5 | 0.0-1.0 | Solo MMR: 0=diversità, 1=similarità | 0.3-0.7 | | | filter | None | metadata dict | Filtra per cartella/tag/data | Essenziale per multi-tenant | | Generation | chain_type | stuff | stuff/map_reduce/refine | Stuff = tutto in un prompt | Stuff per k≤5, map_reduce per k>10 | | | temperature | 0.1 | 0.0-0.3 | Bassa = fedele al contesto | 0.0 per extractive, 0.2 per sintesi |

Troubleshooting RAG

| Problema | Causa | Soluzione | | --- | --- | --- | | Risposte "Non trovo info" ma info c'è | Chunking taglia contesto chiave / retrieval sbagliato | Aumenta chunk_overlap, usa markdown splitter, alza k, prova similarity_score_threshold=0.5 | | Citazioni inventate / file inesistenti | L'LLM inventa le fonti | Prompt più stretto, temperature=0.0, verifica post-generation con verify_citations | | Lento su query | Chroma non indicizzato / troppi documenti | Verifica il numero di chunk con collection.count(), riduci k e usa filter per restringere la ricerca | | Memoria Chroma cresce | Batch di inserimento troppo grandi | Usa batch_size=100 in index_documents; client.reset() cancella tutti i dati, usalo solo se vuoi ripartire da zero | | Embedding OOM | nomic-embed-text su CPU con batch grandi | Riduci batch_size in index_documents e chiudi gli altri processi pesanti | | Risultati irrilevanti | Metadata filter sbagliato / query ambigua | Logga retriever.invoke(query) per debug, aggiungi query expansion (Cap. 6 tool use) |

Come fare, passo per passo: dai documenti alla risposta con fonti

Questa sequenza usa i file creati nel capitolo. Servono Ollama in esecuzione e l'ambiente Python dei capitoli precedenti.

1. Scarica il modello di embedding. Dal terminale esegui ollama pull nomic-embed-text. Il download avviene una volta sola.
2. Installa le dipendenze. Aggiungi le righe di questo capitolo a requirements.txt ed esegui pip install -r requirements.txt.
3. Prepara la cartella dei documenti. Metti PDF, Markdown, testi e HTML in ./data/docs, divisi per argomento. I nomi dei file finiscono nelle citazioni, quindi scegli nomi leggibili.
4. Verifica gli embedding. Lancia python embeddings.py: devi vedere 768 per entrambe le dimensioni. Se Ollama non risponde, controlla che il servizio sia attivo.
5. Indicizza. Esegui python indexing.py. Lo script carica i file, li divide in chunk, scarta i duplicati e salva tutto in ./chroma_db.
6. Interroga. Esegui python generation.py: la demo fa una domanda singola e una conversazione a due turni.
7. Controlla le citazioni. Confronta i nomi file citati con verify_citations: ogni fonte nella risposta deve comparire tra i documenti recuperati.

Alla prima indicizzazione conviene partire da dieci o venti file: se la qualità delle risposte non convince, correggere il chunking costa poco, mentre reindicizzare migliaia di pagine richiede ore.

Checklist: la pipeline è pronta?

• Ollama è attivo e il modello nomic-embed-text risulta tra quelli installati.
• Gli stessi embedding sono usati sia per indicizzare sia per interrogare.
• I documenti sono stati caricati senza errori nel log di indicizzazione.
• Ogni chunk ha nei metadata filename, rel_path e, per i PDF, la pagina.
• Il prompt impone di rispondere solo con il contesto fornito.
• La temperatura del modello è bassa (0.0-0.2).
• La cartella chroma_db è inclusa nei backup e fuori dal repository Git.
• Hai provato domande a cui il documento non risponde, e l'agente dice di non trovare l'informazione.

Errori da evitare

• Cambiare modello di embedding senza reindicizzare. I vettori di modelli diversi non sono confrontabili e le dimensioni possono non coincidere: cambia modello e ricrea la collection.
• Chunk troppo piccoli o troppo grandi. Con pochi caratteri si perde il contesto, con molti il retrieval diventa impreciso. Parti da 800-1200 caratteri con overlap del 15-25%.
• Fidarsi delle citazioni senza controllarle. Un modello locale può citare un file che non è stato recuperato. Verifica sempre con verify_citations.
• Indicizzare dati riservati senza una regola. Tutto resta in locale, ma chiunque acceda alla cartella chroma_db legge anche i testi dei chunk, perché il database li conserva in chiaro.
• Reindicizzare a ogni avvio. Duplichi i chunk e rallenti le risposte. Controlla se la collection esiste già prima di indicizzare.
• Usare k molto alto per "essere sicuri". Più contesto significa più rumore e risposte più lente, soprattutto su CPU.

Consigli pratici

• Registra nei metadata la data di modifica del file: ti serve per aggiornare solo i documenti cambiati.
• Per documentazione tecnica in Markdown usa lo splitter per header, poi il recursive sui blocchi lunghi.
• Se i documenti sono in più lingue, prova bge-m3, un embedding multilingue, e confronta i risultati su dieci domande reali.
• Salva un piccolo set di domande di prova con la risposta attesa: dopo ogni modifica al chunking o al prompt rilanci il test e vedi se la qualità peggiora.
• Con dati personali, tieni la collection su un disco cifrato e riduci gli accessi alla cartella (è lo stesso rischio descritto per la Shadow AI, quando i dati finiscono dove nessuno li controlla); se vuoi anonimizzare i testi prima dell'indicizzazione, guarda lo strumento di anonimizzazione in locale.
• Per capire come l'agente tiene il filo della conversazione, rivedi il capitolo 4 sulla memoria; per capire dove impiegarlo in azienda, c'è la guida su cosa sono gli agenti AI e la pagina su chatbot e agenti.

Domande frequenti

Serve una GPU per il RAG locale? No. L'embedding con nomic-embed-text gira anche su CPU. Una GPU accelera la generazione della risposta, non è indispensabile per indicizzare.

Quanti documenti regge ChromaDB in locale? Dipende dalla memoria disponibile e dalla dimensione dei chunk. Per archivi personali o di un piccolo team, con migliaia di documenti, è una scelta adeguata; oltre questa scala conviene misurare i tempi di risposta prima di decidere.

Perché l'agente risponde "non trovo l'informazione" quando il documento c'è? Di solito il chunking ha spezzato il passaggio chiave o k è troppo basso. Aumenta l'overlap, prova lo splitter per header e rilancia la domanda.

Posso indicizzare file Word o fogli di calcolo? Non con i loader di questo capitolo. Convertili prima in PDF, Markdown o testo, oppure aggiungi un loader dedicato nella tabella loader_mapping.

I miei documenti escono dal computer? No, se usi Ollama e ChromaDB come mostrato: embedding, database e generazione girano sulla tua macchina e la telemetria di ChromaDB è disattivata nel codice.

Prossimo capitolo

Cap. 6: dagli strumenti all'agente (Tool Use) → LangChain Tools: @tool decorator, built-in (DuckDuckGoSearchRun, PythonREPL, Shell), custom tools (file I/O, HTTP, custom API), initialize_agent(agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION), function calling vs ReAct pattern, error handling in tools.

File completo

Modulo RAG production-ready: rag_pipeline.py — include indexing, retrieval, generation, citations, filtering.

Serie "Il Mio Primo Agente Digitale" — Testato LangChain 0.2+, ChromaDB 0.4+, Ollama 0.3+, nomic-embed-text, Python 3.11+. Nota: RetrievalQA e ConversationalRetrievalChain sono API legacy di LangChain; la versione LCEL del paragrafo 6 è quella da preferire per nuovo codice.