09 - Knowledge Graphs per AI: Strutturare Conoscenza

KNOWLEDGE GRAPH: esempio dominio aziendale

ENTITA (Nodi):
  Person: "Luca Rossi" (name, role="CEO", birthYear=1975)
  Company: "TechCorp" (name, founded=2010, sector="AI")
  Product: "AIAnalytics" (name, version="2.0", category="software")
  Technology: "Python" (name, type="language")

RELAZIONI (Archi):
  (Luca Rossi) --[WORKS_AT]--> (TechCorp)
  (Luca Rossi) --[FOUNDED]--> (TechCorp)
  (TechCorp) --[DEVELOPS]--> (AIAnalytics)
  (AIAnalytics) --[USES_TECHNOLOGY]--> (Python)
  (TechCorp) --[COMPETES_WITH]--> (AICorp)

TRIPLE RDF:
  ("TechCorp", "rdf:type", "Company")
  ("TechCorp", "schema:foundingDate", "2010")
  ("Luca Rossi", "schema:worksFor", "TechCorp")

PROPERTY GRAPH (Neo4j):
  (:Person {name: "Luca Rossi", role: "CEO"})
    -[:WORKS_AT {since: 2010, equity: true}]->
  (:Company {name: "TechCorp", sector: "AI"})

VANTAGGI dei Knowledge Graph rispetto a tabelle relazionali:
1. Flessibilità: aggiungi relazioni senza modificare lo schema
2. Navigabilita: traversal multi-hop naturale
3. Reasoning: inferenza di nuove relazioni
4. Semantica: relazioni hanno significato esplicito
      

from neo4j import GraphDatabase
from typing import List, Dict, Any, Optional
import os

class Neo4jKnowledgeGraph:
    """Interfaccia Python per Neo4j knowledge graph"""

    def __init__(
        self,
        uri: str = "bolt://localhost:7687",
        user: str = "neo4j",
        password: str = "password"
    ):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self.driver.close()

    def execute_query(self, query: str, parameters: dict = None) -> List[Dict]:
        """Esegui una query Cypher e ritorna i risultati"""
        with self.driver.session() as session:
            result = session.run(query, parameters or {})
            return [record.data() for record in result]

    def create_entity(self, label: str, properties: Dict) -> str:
        """Crea un nodo con label e proprietà"""
        props_str = ", ".join(f"{k}: ${k}" for k in properties.keys())
        query = f"CREATE (n:{label} {{{props_str}}}) RETURN id(n) as id"
        result = self.execute_query(query, properties)
        return result[0]["id"] if result else None

    def create_relationship(
        self,
        from_label: str, from_props: Dict,
        rel_type: str, rel_props: Dict,
        to_label: str, to_props: Dict
    ):
        """Crea una relazione tra due nodi"""
        from_match = " AND ".join(f"a.{k} = $from_{k}" for k in from_props)
        to_match = " AND ".join(f"b.{k} = $to_{k}" for k in to_props)
        rel_props_str = ", ".join(f"{k}: $rel_{k}" for k in rel_props) if rel_props else ""

        params = {
            **{f"from_{k}": v for k, v in from_props.items()},
            **{f"to_{k}": v for k, v in to_props.items()},
            **{f"rel_{k}": v for k, v in rel_props.items()}
        }

        query = f"""
MATCH (a:{from_label}) WHERE {from_match}
MATCH (b:{to_label}) WHERE {to_match}
MERGE (a)-[r:{rel_type} {{{rel_props_str}}}]->(b)
RETURN type(r) as rel_type"""

        return self.execute_query(query, params)

    def upsert_entity(self, label: str, match_props: Dict, set_props: Dict = None):
        """Upsert: crea se non esiste, aggiorna se esiste"""
        match_str = ", ".join(f"{k}: ${k}" for k in match_props)
        query = f"MERGE (n:{label} {{{match_str}}})"

        params = dict(match_props)
        if set_props:
            set_str = ", ".join(f"n.{k} = $set_{k}" for k in set_props)
            query += f" ON CREATE SET {set_str} ON MATCH SET {set_str}"
            params.update({f"set_{k}": v for k, v in set_props.items()})

        query += " RETURN n"
        return self.execute_query(query, params)


# Esempi di query Cypher avanzate
CYPHER_EXAMPLES = {
    # Trova tutte le aziende AI fondate dopo il 2020
    "aziende_recenti": """
MATCH (c:Company {sector: 'AI'})
WHERE c.founded > 2020
RETURN c.name, c.founded
ORDER BY c.founded DESC""",

    # Trova percorso tra due persone (degree di separazione)
    "percorso_sociale": """
MATCH path = shortestPath(
  (p1:Person {name: $person1})-[*..6]-(p2:Person {name: $person2})
)
RETURN path, length(path) as degrees""",

    # Trova comunita di entità correlate (community detection)
    "entita_correlate": """
MATCH (n:Company)-[r]-(related)
WHERE n.name = $company_name
RETURN related, type(r), n
LIMIT 50""",

    # Multi-hop: prodotti usati da aziende che competono con X
    "prodotti_competitor": """
MATCH (c1:Company)-[:COMPETES_WITH]->(c2:Company)
WHERE c1.name = $company_name
MATCH (c2)-[:DEVELOPS]->(p:Product)
RETURN DISTINCT p.name, p.category, c2.name as developed_by"""
}
      

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List, Optional


# Schema per l'estrazione strutturata
class Entity(BaseModel):
    """Un'entità estratta dal testo"""
    name: str = Field(description="Nome dell'entità")
    entity_type: str = Field(description="Tipo: Person, Company, Product, Technology, Location, Event, Concept")
    description: Optional[str] = Field(description="Breve descrizione dell'entità", default=None)
    properties: dict = Field(description="Proprietà aggiuntive (es. founded_year, role)", default_factory=dict)


class Relationship(BaseModel):
    """Una relazione tra due entità"""
    source: str = Field(description="Nome dell'entità sorgente")
    target: str = Field(description="Nome dell'entità destinazione")
    relationship_type: str = Field(description="Tipo di relazione (es. WORKS_AT, FOUNDED, COMPETES_WITH)")
    properties: dict = Field(description="Proprietà della relazione (es. since_year)", default_factory=dict)


class KnowledgeGraphExtraction(BaseModel):
    """Risultato dell'estrazione di un knowledge graph da testo"""
    entities: List[Entity] = Field(description="Entità estratte dal testo")
    relationships: List[Relationship] = Field(description="Relazioni tra entità")


class LLMKnowledgeGraphExtractor:
    """Estrae knowledge graph da testo usando LLM"""

    def __init__(self, model: str = "gpt-4o-mini"):
        llm = ChatOpenAI(model=model, temperature=0)
        self.structured_llm = llm.with_structured_output(KnowledgeGraphExtraction)

        self.extraction_prompt = ChatPromptTemplate.from_template("""
Estrai le entità e le relazioni dal seguente testo per costruire un knowledge graph.

Tipi di entità da estrarre: Person, Company, Product, Technology, Location, Event, Concept
Tipi di relazioni comuni: WORKS_AT, FOUNDED, DEVELOPS, USES, COMPETES_WITH, PART_OF,
  LOCATED_IN, ACQUIRED_BY, INVESTED_IN, AUTHORED_BY

Testo da analizzare:
{text}

Estrai TUTTE le entità e relazioni menzionate, anche quelle implicite.
Per le proprietà, estrai solo quelle esplicitamente menzionate nel testo.""")

    def extract(self, text: str) -> KnowledgeGraphExtraction:
        """Estrai entità e relazioni da un testo"""
        return self.structured_llm.invoke(
            self.extraction_prompt.format_messages(text=text)
        )

    def extract_and_store(
        self,
        text: str,
        neo4j_kg: Neo4jKnowledgeGraph,
        source_metadata: Dict = None
    ) -> dict:
        """Estrai dal testo e memorizza direttamente in Neo4j"""
        extraction = self.extract(text)

        stored_entities = 0
        stored_relationships = 0

        # Memorizza entità
        for entity in extraction.entities:
            props = {
                "name": entity.name,
                **(entity.properties or {}),
            }
            if entity.description:
                props["description"] = entity.description
            if source_metadata:
                props["source"] = source_metadata.get("source", "")

            neo4j_kg.upsert_entity(
                label=entity.entity_type,
                match_props={"name": entity.name},
                set_props=props
            )
            stored_entities += 1

        # Memorizza relazioni
        for rel in extraction.relationships:
            # Verifica che le entità esistano prima di creare la relazione
            source_entity = next(
                (e for e in extraction.entities if e.name == rel.source), None
            )
            target_entity = next(
                (e for e in extraction.entities if e.name == rel.target), None
            )

            if source_entity and target_entity:
                neo4j_kg.create_relationship(
                    from_label=source_entity.entity_type,
                    from_props={"name": rel.source},
                    rel_type=rel.relationship_type,
                    rel_props=rel.properties or {},
                    to_label=target_entity.entity_type,
                    to_props={"name": rel.target}
                )
                stored_relationships += 1

        return {
            "entities_found": len(extraction.entities),
            "relationships_found": len(extraction.relationships),
            "entities_stored": stored_entities,
            "relationships_stored": stored_relationships
        }


# Esempio utilizzo
extractor = LLMKnowledgeGraphExtractor()
kg = Neo4jKnowledgeGraph()

text = """
OpenAI, fondata da Sam Altman e Elon Musk nel 2015, ha sviluppato GPT-4 e ChatGPT.
L'azienda ha ricevuto un investimento da Microsoft di 10 miliardi di dollari nel 2023.
Anthropic, fondata da ex dipendenti OpenAI tra cui Dario Amodei, sviluppa Claude,
un modello che compete direttamente con ChatGPT.
"""

result = extractor.extract_and_store(text, kg, {"source": "news_article.txt"})
print(f"Estratte: {result['entities_found']} entità, {result['relationships_found']} relazioni")
      

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


class GraphRAGSystem:
    """
    Sistema GraphRAG che combina:
    1. Retrieval vettoriale per domande semantiche
    2. Cypher query su Neo4j per domande strutturate
    3. LLM per sintetizzare entrambe le fonti
    """

    def __init__(self, neo4j_url: str, username: str, password: str, vector_retriever):
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.vector_retriever = vector_retriever

        # Connessione Neo4j per LangChain
        self.graph = Neo4jGraph(
            url=neo4j_url,
            username=username,
            password=password
        )

        # Chain per generare e eseguire query Cypher automaticamente
        self.cypher_chain = GraphCypherQAChain.from_llm(
            cypher_llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
            qa_llm=self.llm,
            graph=self.graph,
            verbose=True,
            return_intermediate_steps=True,
            allow_dangerous_requests=True  # Necessario per query automatiche
        )

        # Router per decidere quale fonte usare
        self.router_chain = (
            ChatPromptTemplate.from_template("""
Analizza questa domanda e decidi la migliore strategia di retrieval.

Domanda: {question}

Scegli UNA strategia:
- "graph": la domanda richiede relazioni tra entità, conteggi, percorsi, o attributi specifici
- "vector": la domanda richiede spiegazioni, concetti, procedure o testo narrativo
- "hybrid": la domanda beneficia di entrambe le fonti

Rispondi SOLO con: graph, vector, o hybrid""")
            | self.llm
        )

    def _classify_query(self, question: str) -> str:
        """Classifica il tipo di query"""
        result = self.router_chain.invoke({"question": question})
        strategy = result.content.strip().lower()
        return strategy if strategy in ["graph", "vector", "hybrid"] else "vector"

    def query(self, question: str) -> dict:
        """Risponde alla domanda usando la strategia ottimale"""
        strategy = self._classify_query(question)
        print(f"Strategia selezionata: {strategy}")

        graph_context = ""
        vector_context = ""

        if strategy in ["graph", "hybrid"]:
            try:
                # Genera ed esegui query Cypher automaticamente
                graph_result = self.cypher_chain.invoke({"query": question})
                graph_context = str(graph_result.get("result", ""))
            except Exception as e:
                graph_context = f"Errore query grafo: {e}"

        if strategy in ["vector", "hybrid"]:
            docs = self.vector_retriever.invoke(question)
            vector_context = "\n".join(d.page_content for d in docs[:3])

        # Sintesi finale
        synthesis_prompt = f"""Domanda: {question}

{f"Dati dal knowledge graph:\n{graph_context}\n" if graph_context else ""}
{f"Documenti rilevanti:\n{vector_context}\n" if vector_context else ""}

Rispondi in modo completo basandoti sulle informazioni disponibili."""

        final_answer = self.llm.invoke(synthesis_prompt).content

        return {
            "answer": final_answer,
            "strategy": strategy,
            "graph_context": graph_context,
            "vector_context": vector_context[:200] if vector_context else ""
        }


# Esempi che mostrano i vantaggi di GraphRAG
graph_rag_examples = [
    # Domanda strutturale: meglio con graph
    "Quante aziende AI sono state fondate dopo il 2020?",

    # Domanda relazionale: meglio con graph
    "Chi sono le persone che lavorano per aziende che competono con OpenAI?",

    # Domanda semantica: meglio con vector
    "Come funziona il meccanismo di attenzione nei transformer?",

    # Domanda ibrida: beneficia di entrambe
    "Qual è la strategia di sviluppo prodotti di Anthropic?"
]
      

class KGEnhancedRetriever:
    """
    Retriever che usa il knowledge graph per espandere le query
    con entità correlate prima del vector search.
    """

    def __init__(self, kg: Neo4jKnowledgeGraph, vector_retriever, llm):
        self.kg = kg
        self.retriever = vector_retriever
        self.llm = llm

    def extract_entities_from_query(self, query: str) -> List[str]:
        """Estrai entità dalla query usando NER"""
        prompt = f"""Estrai i nomi di entità (persone, organizzazioni, prodotti, tecnologie)
dalla seguente query. Restituisci solo i nomi, uno per riga.

Query: {query}"""

        result = self.llm.invoke(prompt).content
        entities = [e.strip() for e in result.split('\n') if e.strip()]
        return entities

    def get_related_entities(self, entity_name: str, max_hops: int = 2) -> List[str]:
        """Ottieni entità correlate nel grafo"""
        query = f"""
MATCH (n)-[*1..{max_hops}]-(related)
WHERE n.name CONTAINS $entity_name
RETURN DISTINCT related.name as name
LIMIT 20"""

        results = self.kg.execute_query(query, {"entity_name": entity_name})
        return [r["name"] for r in results if r["name"]]

    def enhanced_retrieve(self, query: str, top_k: int = 5) -> list:
        """
        Recupera documenti con espansione della query via KG.
        1. Estrai entità dalla query
        2. Trova entità correlate nel grafo
        3. Espandi la query con le entità correlate
        4. Fai vector search sulla query espansa
        """
        # Step 1: Estrai entità dalla query
        entities = self.extract_entities_from_query(query)
        print(f"Entità trovate: {entities}")

        # Step 2: Trova entità correlate
        all_related = set()
        for entity in entities[:3]:  # Limita a 3 entità
            related = self.get_related_entities(entity)
            all_related.update(related[:5])  # Massimo 5 correlate per entità

        # Step 3: Espandi la query
        if all_related:
            expansion = ", ".join(list(all_related)[:10])
            expanded_query = f"{query} [Entità correlate: {expansion}]"
            print(f"Query espansa con: {expansion}")
        else:
            expanded_query = query

        # Step 4: Vector search sulla query espansa
        docs = self.retriever.invoke(expanded_query)
        return docs[:top_k]

    def get_entity_context(self, entity_name: str) -> str:
        """Ottieni contesto strutturato di un'entità dal grafo"""
        query = """
MATCH (n {name: $name})
OPTIONAL MATCH (n)-[r]->(related)
RETURN n, type(r) as rel_type, related.name as related_name
LIMIT 20"""

        results = self.kg.execute_query(query, {"name": entity_name})
        if not results:
            return ""

        lines = [f"Entità: {entity_name}"]
        for r in results:
            if r.get("rel_type") and r.get("related_name"):
                lines.append(f"  -> {r['rel_type']}: {r['related_name']}")

        return "\n".join(lines)