Vector Database e Embeddings

Esempio intuitivo:

"Disdetta locazione commerciale"
→ vettore: [0.23, -0.87, 0.45, 0.12, ...]

"Rescindere contratto di affitto del negozio"
→ vettore: [0.21, -0.85, 0.43, 0.14, ...]

"Ricetta per la torta di mele"
→ vettore: [-0.67, 0.34, -0.22, 0.89, ...]

I primi due vettori sono vicinissimi perché il significato
è simile, anche se le parole sono diverse.
Il terzo è lontano: parla di tutt'altro.

La ricerca tradizionale per parole chiave non troverebbe
la corrispondenza tra "disdetta locazione" e "rescindere
contratto affitto". La ricerca semantica sì.

import openai
import numpy as np

# Inizializza il client
client = openai.OpenAI()

# Tre frasi mediche dalla knowledge base della Dott.ssa Bianchi
frasi_mediche = [
    "Protocollo per il trattamento dell'ipertensione arteriosa "
    "nel paziente diabetico di tipo 2",
    "Gestione della pressione alta in soggetti con diabete mellito",
    "Linee guida per la vaccinazione antinfluenzale negli anziani"
]

# Genera gli embeddings
risposta = client.embeddings.create(
    model="text-embedding-3-small",
    input=frasi_mediche
)

# Estrai i vettori
vettori = [np.array(item.embedding) for item in risposta.data]

print(f"Dimensioni di ogni vettore: {len(vettori[0])}")
# Output: Dimensioni di ogni vettore: 1536

print(f"Prime 5 componenti del primo vettore: "
      f"{vettori[0][:5].round(4)}")
# Output: Prime 5 componenti del primo vettore:
# [0.0234, -0.0871, 0.0452, 0.0123, -0.0567]

import numpy as np

def similarita_coseno(vettore_a: np.ndarray,
                       vettore_b: np.ndarray) -> float:
    """Calcola la similarità coseno tra due vettori."""
    prodotto_scalare = np.dot(vettore_a, vettore_b)
    norma_a = np.linalg.norm(vettore_a)
    norma_b = np.linalg.norm(vettore_b)
    return prodotto_scalare / (norma_a * norma_b)

# Confronta le tre frasi mediche (usando i vettori generati prima)
# Frase 0: "Protocollo ipertensione nel diabetico"
# Frase 1: "Gestione pressione alta nel diabete"
# Frase 2: "Vaccinazione antinfluenzale anziani"

sim_01 = similarita_coseno(vettori[0], vettori[1])
sim_02 = similarita_coseno(vettori[0], vettori[2])
sim_12 = similarita_coseno(vettori[1], vettori[2])

print(f"Ipertensione diabetico vs Pressione alta diabete: "
      f"{sim_01:.4f}")
# Output: 0.9234 — molto simili (stesso argomento)

print(f"Ipertensione diabetico vs Vaccinazione anziani: "
      f"{sim_02:.4f}")
# Output: 0.4521 — poco simili (argomenti diversi)

print(f"Pressione alta diabete vs Vaccinazione anziani: "
      f"{sim_12:.4f}")
# Output: 0.4387 — poco simili (argomenti diversi)

import chromadb

# Crea il client e una collection per le linee guida
client_db = chromadb.PersistentClient(path="./db_linee_guida")
collection = client_db.get_or_create_collection(
    name="linee_guida_cliniche",
    metadata={"hnsw:space": "cosine"}  # usa similarità coseno
)

# Aggiungi documenti con metadata
collection.add(
    ids=["lg_001", "lg_002", "lg_003", "lg_004"],
    documents=[
        "Il trattamento dell'ipertensione nel paziente diabetico "
        "prevede ACE-inibitori come prima scelta. Target pressorio "
        "inferiore a 130/80 mmHg.",
        "La metformina resta il farmaco di prima linea per il "
        "diabete di tipo 2. Iniziare con 500mg e titolare.",
        "Protocollo vaccinazione antinfluenzale: somministrare "
        "annualmente a tutti i pazienti over 65.",
        "In caso di crisi ipertensiva nel diabetico, somministrare "
        "captopril sublinguale e monitorare ogni 15 minuti."
    ],
    metadatas=[
        {"categoria": "cardiologia", "anno": 2024,
         "tipo": "protocollo"},
        {"categoria": "endocrinologia", "anno": 2024,
         "tipo": "raccomandazione"},
        {"categoria": "prevenzione", "anno": 2023,
         "tipo": "protocollo"},
        {"categoria": "cardiologia", "anno": 2024,
         "tipo": "emergenza"}
    ]
)

print(f"Documenti nella collection: {collection.count()}")
# Output: Documenti nella collection: 4

import os

def carica_e_indicizza(cartella_pdf: str,
                        collection,
                        dimensione_chunk: int = 500):
    """Pipeline completa: leggi PDF, spezza in chunk, indicizza."""

    # Simulazione: in produzione si userebbe PyPDF2 o pdfplumber
    documenti_esempio = {
        "linee_guida_diabete.pdf": [
            "Sezione 1: Diagnosi. Il diabete di tipo 2 si diagnostica "
            "con glicemia a digiuno superiore a 126 mg/dL confermata "
            "in due occasioni separate.",
            "Sezione 2: Terapia. La metformina è il farmaco di prima "
            "scelta. Iniziare con 500mg una volta al giorno, aumentare "
            "gradualmente fino a 2000mg.",
            "Sezione 3: Complicanze. Monitorare funzionalità renale "
            "ogni 6 mesi, fondo oculare annuale, piede diabetico "
            "a ogni visita."
        ],
        "protocollo_ipertensione.pdf": [
            "Il target pressorio nel paziente diabetico è inferiore "
            "a 130/80 mmHg. Prima scelta terapeutica: ACE-inibitori "
            "o sartani.",
            "In caso di mancato raggiungimento del target con "
            "monoterapia, aggiungere calcio-antagonista o diuretico "
            "tiazidico a basso dosaggio."
        ]
    }

    contatore = 0
    for nome_file, chunks in documenti_esempio.items():
        for i, chunk in enumerate(chunks):
            doc_id = f"doc_{contatore:04d}"
            collection.add(
                ids=[doc_id],
                documents=[chunk],
                metadatas=[{
                    "fonte": nome_file,
                    "sezione": i + 1,
                    "categoria": "endocrinologia"
                        if "diabete" in nome_file
                        else "cardiologia"
                }]
            )
            contatore += 1

    print(f"Indicizzati {contatore} chunk da "
          f"{len(documenti_esempio)} documenti")

# Esegui la pipeline
carica_e_indicizza("./pdf_linee_guida", collection)
# Output: Indicizzati 5 chunk da 2 documenti

# Ricerca semantica nella knowledge base
risultati = collection.query(
    query_texts=[
        "trattamento ipertensione paziente diabetico"
    ],
    n_results=3
)

print("Query: trattamento ipertensione paziente diabetico\n")

for i, (doc, meta, distanza) in enumerate(zip(
    risultati["documents"][0],
    risultati["metadatas"][0],
    risultati["distances"][0]
)):
    print(f"Risultato {i + 1} (distanza: {distanza:.4f}):")
    print(f"  Fonte: {meta.get('fonte', 'N/A')}")
    print(f"  Categoria: {meta.get('categoria', 'N/A')}")
    print(f"  Testo: {doc[:120]}...")
    print()

# Output tipico:
# Risultato 1 (distanza: 0.1234):
#   Fonte: protocollo_ipertensione.pdf
#   Categoria: cardiologia
#   Testo: Il target pressorio nel paziente diabetico è inferiore
#   a 130/80 mmHg. Prima scelta terapeutica: ACE-inibitori...

# Ricerca filtrata: solo protocolli di cardiologia del 2024
risultati_filtrati = collection.query(
    query_texts=["gestione crisi ipertensiva"],
    n_results=3,
    where={
        "$and": [
            {"categoria": {"$eq": "cardiologia"}},
            {"anno": {"$gte": 2024}}
        ]
    }
)

print("Query filtrata: crisi ipertensiva "
      "(solo cardiologia, dal 2024)\n")

for doc, meta in zip(
    risultati_filtrati["documents"][0],
    risultati_filtrati["metadatas"][0]
):
    print(f"Fonte: {meta.get('fonte', 'N/A')}")
    print(f"Tipo: {meta.get('tipo', 'N/A')}")
    print(f"Testo: {doc[:150]}...")
    print()

# Filtro per tipo di documento
risultati_emergenza = collection.query(
    query_texts=["emergenza pressione alta"],
    n_results=2,
    where={"tipo": {"$eq": "emergenza"}}
)

print("Solo documenti di emergenza:")
for doc in risultati_emergenza["documents"][0]:
    print(f"  {doc[:100]}...")