Chatbot RAG local peste documentația internă si ChromaDB

Situatie

Documentația tehnică internă (proceduri, manuale, contracte, runbook-uri) este răspândită în zeci de fișiere PDF și Word, greu de căutat și de consultat rapid. Angajații pierd timp căutând informații în documente sau întrebând colegi.

Această soluție indexează local toate documentele, creează o bază de cunoștințe vectorială (ChromaDB) și permite interogarea lor în limbaj natural printr-un model AI local (Ollama) — răspunsurile sunt generate exclusiv din documentele tale, fără ca nicio dată să iasă din rețea.

Solutie

Pasi de urmat

Pasul 1 — Instalare Python și dependențe

powershell

# Verifica Python (minim 3.10)
python --version

# Instaleaza dependentele necesare
pip install chromadb pymupdf python-docx ollama requests colorama

Pasul 2 — Descărcare modele în Ollama

powershell

# Model pentru generare raspunsuri
ollama pull llama3

# Model pentru embeddings (transforma textul in vectori)
ollama pull nomic-embed-text

nomic-embed-text este optimizat pentru embeddings și consumă mult mai puțin RAM decât llama3.

Pasul 3 — Script de indexare documente (indexer.py)

Salvează fișierul ca C:\RAG\indexer.py:

python:

# ============================================================
# RAG Indexer - Indexare PDF si Word cu ChromaDB + Ollama
# Autor: Claudiu Ceausescu
# ============================================================

import os
import sys
import fitz          # pymupdf - pentru PDF
import chromadb
import ollama
from docx import Document
from colorama import Fore, Style, init

init(autoreset=True)

DOCS_PATH  = r"C:\RAG\Documente"   # folderul cu fisierele sursă
DB_PATH    = r"C:\RAG\db"          # unde se salveaza baza vectoriala
EMBED_MODEL = "nomic-embed-text"
CHUNK_SIZE  = 500                  # caractere per fragment


def extract_pdf(path):
    doc = fitz.open(path)
    return "\n".join(page.get_text() for page in doc)


def extract_word(path):
    doc = Document(path)
    return "\n".join(p.text for p in doc.paragraphs if p.text.strip())


def chunk_text(text, filename, size=CHUNK_SIZE):
    chunks, docs, ids = [], [], []
    words = text.split()
    current, idx = [], 0
    for word in words:
        current.append(word)
        if len(" ".join(current)) >= size:
            chunk = " ".join(current)
            chunks.append(chunk)
            docs.append({"source": filename})
            ids.append(f"{filename}_{idx}")
            current = []
            idx += 1
    if current:
        chunks.append(" ".join(current))
        docs.append({"source": filename})
        ids.append(f"{filename}_{idx}")
    return chunks, docs, ids


def get_embedding(text):
    response = ollama.embeddings(model=EMBED_MODEL, prompt=text)
    return response["embedding"]


def main():
    os.makedirs(DOCS_PATH, exist_ok=True)
    client     = chromadb.PersistentClient(path=DB_PATH)
    collection = client.get_or_create_collection("documente_interne")

    files = [f for f in os.listdir(DOCS_PATH)
             if f.endswith((".pdf", ".docx", ".doc"))]

    if not files:
        print(Fore.YELLOW + f"Nu s-au gasit documente in {DOCS_PATH}")
        sys.exit(0)

    print(Fore.CYAN + f"Gasit {len(files)} documente. Incep indexarea...\n")

    for i, filename in enumerate(files, 1):
        filepath = os.path.join(DOCS_PATH, filename)
        print(Fore.CYAN + f"[{i}/{len(files)}] Indexez: {filename}")

        try:
            if filename.endswith(".pdf"):
                text = extract_pdf(filepath)
            else:
                text = extract_word(filepath)

            if len(text.strip()) < 50:
                print(Fore.YELLOW + "   (document gol - skip)")
                continue

            chunks, metadatas, ids = chunk_text(text, filename)

            # Genereaza embeddings si salveaza in ChromaDB
            embeddings = [get_embedding(c) for c in chunks]
            collection.add(
                documents=embeddings,
                metadatas=metadatas,
                ids=ids
            )

            print(Fore.GREEN + f"   -> {len(chunks)} fragmente indexate")

        except Exception as e:
            print(Fore.RED + f"   EROARE: {e}")

    total = collection.count()
    print(Fore.GREEN + f"\n[DONE] Indexare completa. Total fragmente in baza: {total}")


if __name__ == "__main__":
    main()

Pasul 4 — Script de interogare (query.py)

Salvează fișierul ca C:\RAG\query.py:

python

# ============================================================
# RAG Query - Interogare documentatie interna cu Ollama
# Autor: Claudiu Ceausescu
# ============================================================

import chromadb
import ollama
import sys
from colorama import Fore, Style, init

init(autoreset=True)

DB_PATH      = r"C:\RAG\db"
EMBED_MODEL  = "nomic-embed-text"
GEN_MODEL    = "llama3"
TOP_K        = 3   # numarul de fragmente relevante returnate


def get_embedding(text):
    response = ollama.embeddings(model=EMBED_MODEL, prompt=text)
    return response["embedding"]


def query_docs(question, collection):
    emb     = get_embedding(question)
    results = collection.query(
        query_embeddings=[emb],
        n_results=TOP_K,
        include=["documents", "metadatas"]
    )
    return results["documents"][0], results["metadatas"][0]


def generate_answer(question, context_chunks, sources):
    context = "\n\n".join(context_chunks)
    prompt  = f"""Esti un asistent tehnic care raspunde EXCLUSIV pe baza documentatiei interne furnizate.
Daca raspunsul nu se regaseste in documente, spune clar: "Nu am gasit aceasta informatie in documentatie."
Raspunde in limba romana, concis si tehnic.

DOCUMENTATIE:
{context}

INTREBARE:
{question}
"""
    response = ollama.chat(
        model=GEN_MODEL,
        messages=[{"role": "user", "content": prompt}]
    )
    return response["message"]["content"]


def main():
    client     = chromadb.PersistentClient(path=DB_PATH)
    collection = client.get_or_create_collection("documente_interne")

    if collection.count() == 0:
        print(Fore.RED + "Baza de cunostinte este goala. Ruleaza mai intai indexer.py")
        sys.exit(1)

    print(Fore.GREEN + "=== Chatbot RAG Local - Documentatie Interna ===")
    print(Fore.YELLOW + "Scrie 'exit' pentru a iesi.\n")

    while True:
        question = input(Fore.CYAN + "Intrebare: ").strip()
        if question.lower() == "exit":
            break
        if not question:
            continue

        print(Fore.YELLOW + "Caut in documentatie...")
        chunks, metadatas = query_docs(question, collection)

        print(Fore.YELLOW + "Generez raspuns...")
        answer  = generate_answer(question, chunks, metadatas)
        sources = list(set(m["source"] for m in metadatas))

        print(Fore.GREEN + "\nRASPUNS:")
        print(answer)
        print(Fore.CYAN + f"\nSurse consultate: {', '.join(sources)}\n")
        print("-" * 50)


if __name__ == "__main__":
    main()

Pasul 5 — Rulare indexare

powershell

# 1. Copiaza documentele PDF/Word in folderul sursa
Copy-Item "\\server\documente\*" "C:\RAG\Documente\" -Recurse

# 2. Ruleaza indexarea (o singura data sau la adaugare documente noi)
cd C:\RAG
python indexer.py

Output așteptat:

Gasit 12 documente. Incep indexarea...

[1/12] Indexez: Procedura_VPN.pdf
   -> 14 fragmente indexate
[2/12] Indexez: Manual_Active_Directory.docx
   -> 31 fragmente indexate
...
[DONE] Indexare completa. Total fragmente in baza: 187

Pasul 6 — Interogare chatbot

powershell

cd C:\RAG
python query.py

Exemplu sesiune:

=== Chatbot RAG Local - Documentatie Interna ===

Intrebare: Cum resetez parola unui utilizator in AD?
Caut in documentatie...
Generez raspuns...

RASPUNS:
Conform procedurii interne, pentru resetarea parolei unui utilizator
in Active Directory urmati pasii: 1. Deschideti ADUC...

Surse consultate: Procedura_Active_Directory.pdf

Pasul 7 — Automatizare re-indexare

powershell

# Adauga task programat pentru re-indexare zilnica la 02:00
$Action  = New-ScheduledTaskAction -Execute "python" `
           -Argument "C:\RAG\indexer.py"
$Trigger = New-ScheduledTaskTrigger -Daily -At "02:00"
Register-ScheduledTask -TaskName "RAG_Reindexare" `
    -Action $Action -Trigger $Trigger -RunLevel Highest

Tip solutie

Permanent

Follow Us