ナレッジ グラフと AI: 構造化された知識を LLM に統合する
I 大規模言語モデル 流暢なテキストを生成することに優れており、 しかし、彼らは根本的な限界に苦しんでいます。それは、そこに含まれる知識と 暗黙、モデルパラメータに分散されており、更新が困難 また、構造化クエリを使用してクエリを実行することは不可能です。 「その人たち全員を私にください」 「2020年以降に設立されたAI企業で働く」という知識は些細なものです グラフ、LLM で保証することは不可能です。
I ナレッジグラフ (KG) 知識をグラフとして表す エンティティと関係: 明示的、クエリ可能、更新可能、検証可能な構造。 KG と LLM の統合 - パラダイム グラフRAG - システムを生成する 従来の RAG では提供できない構造化された推論が可能です。 この記事では、Neo4j を使用して GraphRAG システムを構築し、マイニングを探索します LLM を使用してテキストからグラフを自動生成し、ナレッジ グラフをクエリする方法を見てみましょう RAG システムを強化します。
何を学ぶか
- ナレッジ グラフの基礎: ノード、リレーション、プロパティ、RDF、プロパティ グラフ
- Neo4j: モデル、Cypher クエリ言語、LangChain の統合
- LLM を使用した非構造化テキストからの自動 KG 抽出
- GraphRAG: グラフ検索とベクトル検索の組み合わせ
- RAG の KG: エンティティ関係でチャンクを強化する
- ナレッジグラフでのマルチホップ推論
- 公開ウィキデータとナレッジグラフによる強化
- 本番環境で保守可能な KG を構築するためのベスト プラクティス
1. ナレッジグラフの基礎
Un ナレッジグラフ は知識の表現です 私がいるグラフ 結び目 エンティティ（人、組織、 コンセプト、イベント）と アーチ それらの間の関係を表します。 各トリプル (主語、述語、目的語) は事実をエンコードします。
ナレッジグラフの構造
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
1.1 RDF とプロパティのグラフ
ナレッジ グラフ モデルには主に 2 つあり、それぞれに異なるトレードオフがあります。
RDF とプロパティのグラフ
| サイズ | RDF/スパークル | プロパティグラフ (Neo4j) |
|---|---|---|
| モデル | トリプル（S、P、O）を標準化 | 任意のプロパティを持つノードとエッジ |
| 標準 | W3C 標準、相互運用可能 | 独自仕様だがより柔軟 |
| クエリ言語 | SPARQL (複合) | サイファー (より読みやすく) |
| 関係よりもプロパティ | 複雑（具体化） | ネイティブかつシンプル |
| AIエコシステム | ウィキデータ、DBpedia、Schema.org | Neo4j (LangChain 統合) |
| いつ使用するか | オープンデータ、相互運用性 | AIアプリケーション、GraphRAG |
2. Neo4j: セットアップと暗号クエリ言語
Neo4j AI アプリケーション用の最も人気のあるグラフ データベースです。 LangChainとの優れた統合を備えています。言語 サイファー は、直感的な ASCII アート構文を使用してグラフ パターンを表現します。
Neo4j のセットアップとクエリ サイファー ベース
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"""
}
3. テキストからのナレッジグラフの自動抽出
ナレッジ グラフを手動で構築するにはコストがかかります。最新の LLM を使用すると、次のことが可能になります。 非構造化テキストからエンティティと関係を自動的に抽出し、データを入力します。 半自動でグラフを作成します。
LLM と LangChain を使用した KG マイニング
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")
4. GraphRAG: グラフとベクトル取得の組み合わせ
グラフRAG 従来のセマンティック検索を組み合わせたパラダイムです (ベクトル検索) ナレッジ グラフ トラバーサルを使用します。必要な質問については、 エンティティ間の関係に関する推論では、GraphRAG は従来の RAG を大幅に上回ります。
LangChain と Neo4j を使用した GraphRAG
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?"
]
5. エンリッチメント RAG のナレッジ グラフ
完全な GraphRAG を実行しなくても、ナレッジ グラフを充実させることができます 従来の RAG システムと大きく異なる: エンティティを使用したクエリの拡張 関連、関連する関係についてドキュメントをフィルタリングする、またはコンテキストを追加する 復元されたチャンクに構造化されます。
KG-Enhanced RAG: グラフを使用したクエリ拡張
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)
6. ベストプラクティスとアンチパターン
AI のベスト プラクティス ナレッジ グラフ
- 小さく反復的に始めてください。 完璧な KG を最初から構築しないでください。ユースケースにとって最も重要なエンティティと関係から始めて、拡張していきます。
- 明確なオントロジーを定義します。 始める前に、ノードと関係のタイプを定義します。不適切なオントロジーは、グラフにデータを入力した後に変更するのが困難です。
- 自動抽出を検証します。 LLM は抽出時にエラーを起こします。特に最初に、重要なデータに対して人間による検証プロセスを実装します。
- CREATE ではなく MERGE を使用します。 Neo4j では、重複を避けるためにエンティティには常に MERGE を使用してください。 CREATE は、ノードが既に存在する場合でも、常に新しいノードを作成します。
- 検索プロパティのインデックス: WHERE クエリで使用されるプロパティ (名前、日付など) に Neo4j インデックスを作成します。インデックスがないと、大きなグラフに対するクエリが遅くなります。
避けるべきアンチパターン
- RAG の完全な代替品としての KG: GraphRAG は強力ですが、セットアップコストが高くなります。純粋に意味的な質問の場合、多くの場合、従来の RAG の方が優れており、安価です。
- 一般的すぎる関係: 「RELATED_TO」関係には情報がありません。関係は意味的に正確である必要があります (FOUNDED、WORKS_AT、COMPETES_WITH)。
- 更新戦略なし: 静的な KG はすぐに時代遅れになります。グラフを更新する方法と頻度を最初から定義します。
- 検証なしで生成された暗号クエリ: LLM によって生成された Cypher クエリは危険な可能性があります (誤った削除、パフォーマンスの問題)。可能な場合は、パラメータ化されたテンプレート クエリを使用します。
結論
ナレッジ グラフは、従来の RAG システムでは実現できないものをもたらします。 構造化され、関係性があり、検証可能な知識。 GraphRAG の組み合わせ 両方の長所: セマンティック検索の柔軟性と精度 グラフに関する構造化された推論。自動抽出機能である Neo4j を調べました LLM を使用した KG、LangChain を使用した GraphRAG、およびグラフを使用した従来の RAG エンリッチメント。
重要なポイント:
- KG は知識をエンティティおよび関係として表現します: 明示的でクエリ可能な構造
- LLM を使用すると、非構造化テキストから KG を自動的に抽出できます。
- GraphRAG は、リレーショナル クエリおよびマルチホップ クエリに関して従来の RAG よりも優れたパフォーマンスを発揮します
- KG によるクエリ拡張により、従来の RAG の再現率が向上します
- 単純なオントロジーから開始し、反復的に拡張します
これがシリーズの完結編です AIエンジニアリングと高度なRAG。 RAG の基礎から埋め込み、ベクトルに至るまで、スタック全体をカバーしました。 データベースからハイブリッド検索、マルチエージェントなどのより高度なシステムまで 知識グラフ。この分野は急速に進化しています - ブログをフォローし続けてください アップデート用に。
AIエンジニアリングおよびAdvanced RAGシリーズ全体
- 記事 1: RAG の説明 - 基礎
- 第 2 条: 埋め込みとセマンティック検索
- 第3条：ベクターデータベース
- 第 4 条: ハイブリッド検索
- 第 5 条: 本番環境における RAG
- 第 6 条: RAG 用の LangChain
- 第 7 条: コンテキスト ウィンドウの管理
- 第 8 条: マルチエージェント システム
- 第 9 条: 生産における迅速なエンジニアリング
- 第 10 条: AI のナレッジ グラフ (現行)
関連シリーズを続けてください: PostgreSQL 上の RAG 用の pgvector e BERT と最新の NLP.