neo4j-document-import-skill

Neo4j ドキュメントインポートスキル

使用する場面

• PDF、HTML、プレーンテキスト、Markdown を Neo4j にナレッジグラフとして取り込む場合
• ドキュメントをチャンクに分割し、埋め込みを含む :Chunk ノードを保存する場合
• LLM でテキストからエンティティとリレーションシップを抽出する場合
• SimpleKGPipeline（neo4j-graphrag）をプログラムから使用する場合
• Neo4j LLM Graph Builder（ノーコードのウェブ UI）を使用する場合
• apoc.load.json で半構造化 JSON を読み込む場合
• LangChain または LlamaIndex のドキュメントローダーを Neo4j に接続する場合

使用しない場面

• 構造化 CSV／リレーショナルデータ → neo4j-import-skill
• 取り込み後の GraphRAG 検索 → neo4j-graphrag-skill
• ベクトルインデックスの作成 → neo4j-vector-search-skill
• Cypher クエリの作成 → neo4j-cypher-skill

---

アプローチ選択表

状況	アプローチ
コード不要で、ドラッグ＆ドロップの UX が必要	LLM Graph Builder ウェブ UI
プログラムによるパイプライン、PDF／テキスト	SimpleKGPipeline（neo4j-graphrag）
JSON／REST API レスポンス	apoc.load.json または Python + UNWIND
すでに LangChain をスタックで使用している	Neo4jGraph + ドキュメントローダー
すでに LlamaIndex をスタックで使用している	Neo4jQueryEngine／Neo4jVectorStore
チャンクのみ（エンティティ抽出なし）	手動チャンク分割 + MERGE パターン

---

インストール

pip install neo4j-graphrag                   # includes SimpleKGPipeline
pip install neo4j-graphrag[openai]           # + OpenAI LLM/embedder
pip install neo4j-graphrag[anthropic]        # + Anthropic Claude
pip install neo4j-graphrag[google]           # + Vertex AI / Gemini
pip install neo4j-graphrag[bedrock]          # + Amazon Bedrock (boto3) — added v1.15.0
pip install neo4j-graphrag[ollama]           # + Ollama (local)
pip install neo4j-graphrag[mistralai]        # + MistralAI
pip install neo4j-graphrag[fuzzy-matching]   # + FuzzyMatchResolver (rapidfuzz)
# spaCy entity resolver (Python <= 3.13 only — unsupported on 3.14+):
pip install neo4j-graphrag[nlp]

必要条件：neo4j>=5.17.0（ドライバー 6.x に対応）、Python>=3.10、Neo4j>=5.18.1（Aura>=5.18.0）。

---

ステップ 1 — グラフスキーマを定義する

スキーマによって、LLM が抽出する内容が決まります。パイプラインの構築前に定義してください。

# Option A — Simple string lists (LLM infers descriptions)
entities = ["Person", "Organization", "Location", "Product", "Event"]
relations = ["WORKS_AT", "LOCATED_IN", "KNOWS", "MENTIONS", "PART_OF"]
patterns = [
    ("Person", "WORKS_AT", "Organization"),
    ("Organization", "LOCATED_IN", "Location"),
    ("Person", "KNOWS", "Person"),
    ("Article", "MENTIONS", "Organization"),
]

# Option B — Rich GraphSchema (production; best extraction quality)
from neo4j_graphrag.components.schema import (
    GraphSchema, NodeType, RelationshipType, PropertyType, ConstraintType
)
schema = GraphSchema(
    node_types=[
        NodeType(
            label="Person",
            description="A human individual",
            properties=[
                PropertyType(name="name", type="STRING"),
                PropertyType(name="role", type="STRING"),
            ],
        ),
        NodeType(
            label="Organization",
            description="A company or institution",
            properties=[
                PropertyType(name="name", type="STRING"),
                PropertyType(name="industry", type="STRING"),
            ],
        ),
    ],
    relationship_types=[
        RelationshipType(label="WORKS_AT", description="Employment relationship"),
    ],
    patterns=[("Person", "WORKS_AT", "Organization")],
    # Optional: constraints emitted to ParquetWriter metadata (v1.15.0+)
    constraints=[
        ConstraintType(label="Person", property_name="name", type="UNIQUENESS"),
        ConstraintType(label="Organization", property_name="name", type="KEY"),
    ],
)

# Option C — Auto-extract schema from text (no constraints)
schema = "EXTRACTED"   # LLM infers types; noisier output
schema = "FREE"        # No schema guidance; most noise

本番環境ではオプション B、プロトタイピングではオプション A を使用してください。"EXTRACTED" は探索のみに使用してください。

---

ステップ 2 — SimpleKGPipeline をセットアップする

import asyncio
from neo4j import GraphDatabase
from neo4j_graphrag.experimental.pipeline.kg_builder import SimpleKGPipeline
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.embeddings import OpenAIEmbeddings

driver = GraphDatabase.driver(
    "neo4j+s://xxxx.databases.neo4j.io",
    auth=("neo4j", "password")
)

llm = OpenAILLM(
    model_name="gpt-4.1",
    model_params={"temperature": 0},
    # Note: SimpleKGPipeline auto-enables structured output for OpenAI/VertexAI LLMs (v1.14.0+)
    # Do NOT set response_format manually — it is managed by the pipeline
)
embedder = OpenAIEmbeddings()   # OPENAI_API_KEY from env

pipeline = SimpleKGPipeline(
    llm=llm,
    driver=driver,
    embedder=embedder,
    schema=schema,              # GraphSchema, dict, "FREE", or "EXTRACTED"
    from_file=True,             # False → pass text= instead of file_path=
    on_error="IGNORE",          # RAISE to surface extraction failures
    perform_entity_resolution=True,
    neo4j_database="neo4j",     # omit to use default
)

LLM の代替候補（同じインターフェース）：

• AnthropicLLM(model_name="claude-3-5-sonnet-20241022")
• VertexAILLM(model_name="gemini-2.0-flash")
• OllamaLLM(model_name="llama3") — ローカルで動作し、API キーは不要です
• BedrockLLM(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0") — Amazon Bedrock（v1.15.0+）

---

ステップ 3 — パイプラインを実行する

# From PDF file:
result = asyncio.run(pipeline.run_async(
    file_path="report.pdf",        # auto-dispatches to PdfLoader
    document_metadata={"source": "Q4 report", "year": 2025},
))

# From Markdown file (v1.15.0+):
result = asyncio.run(pipeline.run_async(
    file_path="notes.md",          # auto-dispatches to MarkdownLoader
    document_metadata={"source": "meeting notes"},
))

# Note: old `from_pdf=True` parameter is DEPRECATED since v1.15.0; use `from_file=True` instead
# pipeline = SimpleKGPipeline(..., from_file=True)   ← correct
# pipeline = SimpleKGPipeline(..., from_pdf=True)    ← deprecated

# From raw text:
result = asyncio.run(pipeline.run_async(
    text=document_text,
))

# Batch — process multiple files:
async def ingest_all(paths):
    for p in paths:
        await pipeline.run_async(file_path=str(p))

asyncio.run(ingest_all(list(pdf_dir.glob("*.pdf"))))

document_metadata の dict は、:Document ノードのプロパティとして保存されます。

---

ステップ 4 — チャンク分割を設定する

デフォルトのスプリッター：FixedSizeSplitter(chunk_size=300, chunk_overlap=50)。

from neo4j_graphrag.components.text_splitters.fixed_size_splitter import FixedSizeSplitter

splitter = FixedSizeSplitter(
    chunk_size=512,       # tokens; 300–512 typical for GPT-4o
    chunk_overlap=50,     # ~10% of chunk_size; preserves boundary context
    approximate=True,     # respect sentence/word boundaries when possible
)

pipeline = SimpleKGPipeline(
    ...,
    text_splitter=splitter,
)

チャンク分割の指針：

ドキュメントの種類	chunk_size	chunk_overlap
密度の高い技術文書	256–512	50–80
物語形式／ニュース記事	512–1024	80–128
法務／財務文書	256–384	40–64

ルール：チャンクは、抽出時の LLM コンテキストに収まり、かつ埋め込みモデルの上限以内である必要があります。GPT-4o：コンテキスト 128k、text-embedding-3-small：8191 トークン。chunk_size を 2048 より大きく設定しないでください。

---

ステップ 5 — エンティティを解決する

パイプラインの実行後に、抽出された重複エンティティをマージします。

from neo4j_graphrag.components.resolver import (
    SinglePropertyExactMatchResolver,   # identical name → merge
    FuzzyMatchResolver,                  # Levenshtein similarity; needs rapidfuzz
    SpaCySemanticMatchResolver,          # cosine similarity; needs neo4j-graphrag[nlp]
)

# Exact match (fastest; good baseline)
resolver = SinglePropertyExactMatchResolver(driver)
asyncio.run(resolver.run())

# Fuzzy match (handles typos / alternate spellings)
from neo4j_graphrag.components.resolver import FuzzyMatchResolver
resolver = FuzzyMatchResolver(driver, threshold=0.9)
asyncio.run(resolver.run())

# Scope resolution to specific labels only:
resolver = SinglePropertyExactMatchResolver(
    driver,
    filter_query="WHERE n:Organization OR n:Person",
)
asyncio.run(resolver.run())

リゾルバーは取り込み後に実行し、インラインでは実行しないでください。一括マージのほうが高速です。

---

生成されるグラフ構造

パイプラインは常に、次の語彙グラフレイヤーを生成します。

(:Document {id, fileName, status, ...metadata})
    -[:HAS_CHUNK]->
(:Chunk {id, text, index, embedding, ...})
    -[:NEXT_CHUNK]->          ← linked list for ordered traversal
(:Chunk {...})

(:Chunk)-[:FROM_DOCUMENT]->(:Document)   ← back-pointer

エンティティ抽出によって、次の要素が追加されます。

(:Chunk)-[:MENTIONS]->(:Person {name, ...})
(:Chunk)-[:MENTIONS]->(:Organization {name, ...})
(:Person)-[:WORKS_AT]->(:Organization)

取り込み後に確認します。

CYPHER 25
MATCH (d:Document)-[:HAS_CHUNK]->(c:Chunk)
RETURN d.fileName, count(c) AS chunks LIMIT 10;

MATCH (c:Chunk)-[:MENTIONS]->(e)
RETURN labels(e)[0] AS type, count(*) AS cnt ORDER BY cnt DESC LIMIT 20;

---

LLM Graph Builder（ノーコード UI）

使用する場面：開発者以外がドキュメントを取り込む必要がある場合、迅速なプロトタイピング、Python 環境がない場合。

ホステッド版： https://llm-graph-builder.neo4jlabs.com/

ローカル版（Docker）：

git clone https://github.com/neo4j-labs/llm-graph-builder
cd llm-graph-builder
# Set OPENAI_API_KEY (or other provider keys) in .env
docker-compose up
# Opens at http://localhost:8080

対応ソース：PDF、プレーンテキスト、Markdown、画像、ウェブページ、YouTube の文字起こし、S3／GCS バケットへのアップロード。

LLM プロバイダー：OpenAI、Gemini、Claude、Llama3、Diffbot、Qwen。

制限事項：長文の英語テキストに最適です。表形式のデータには不向きです（CSV／Excel には neo4j-import-skill を使用してください）。視覚的な図表は抽出されません。

---

APOC による JSON 取り込み（半構造化）

ソースが REST API、S3、またはファイルエクスポートの JSON の場合に使用します。

CYPHER 25
CALL apoc.load.json("https://example.com/articles.json") YIELD value
UNWIND value.articles AS article
CALL (article) {
  MERGE (d:Document {id: article.id})
  SET d.title = article.title, d.url = article.url, d.publishedAt = article.publishedAt
  FOREACH (tag IN article.tags |
    MERGE (t:Tag {name: tag})
    MERGE (d)-[:HAS_TAG]->(t)
  )
} IN TRANSACTIONS OF 1000 ROWS

ローカルファイルの場合：apoc.load.json("file:///import/data.json")。ファイルは $NEO4J_HOME/import/ 内に配置するか、APOC の allowlist を設定する必要があります。

APOC が利用可能か確認するには、RETURN apoc.version() を実行します。APOC はすべての Aura ティアに含まれています。

---

LangChain との統合パターン

from langchain_community.graphs import Neo4jGraph
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from neo4j import GraphDatabase

graph = Neo4jGraph(
    url="neo4j+s://xxxx.databases.neo4j.io",
    username="neo4j",
    password="password",
)

loader = PyPDFLoader("report.pdf")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
chunks = splitter.split_documents(docs)

embedder = OpenAIEmbeddings()
driver = GraphDatabase.driver(url, auth=("neo4j", "password"))

for i, chunk in enumerate(chunks):
    emb = embedder.embed_query(chunk.page_content)
    driver.execute_query(
        """
        MERGE (doc:Document {id: $doc_id})
        SET doc.source = $source
        CREATE (c:Chunk {id: $chunk_id, text: $text, embedding: $emb, index: $idx})
        CREATE (doc)-[:HAS_CHUNK]->(c)
        """,
        doc_id=chunk.metadata.get("source", "unknown"),
        source=chunk.metadata.get("source"),
        chunk_id=f"chunk-{i}",
        text=chunk.page_content,
        emb=emb,
        idx=i,
    )

LangChain でエンティティを抽出する場合は、LLMGraphTransformer（langchain_experimental.graph_transformers から）を使用してください。同じ :Document／:Chunk／エンティティのパターンを生成します。

---

制約とインデックス（取り込みの前に実行）

CYPHER 25
// Prevent duplicate documents
CREATE CONSTRAINT doc_id_unique IF NOT EXISTS
  FOR (d:Document) REQUIRE d.id IS UNIQUE;

// Prevent duplicate chunks
CREATE CONSTRAINT chunk_id_unique IF NOT EXISTS
  FOR (c:Chunk) REQUIRE c.id IS UNIQUE;

// Entity deduplication
CREATE CONSTRAINT person_name_unique IF NOT EXISTS
  FOR (p:Person) REQUIRE p.name IS UNIQUE;
CREATE CONSTRAINT org_name_unique IF NOT EXISTS
  FOR (o:Organization) REQUIRE o.name IS UNIQUE;

// Vector index for chunk embeddings (adjust dims for your model)
CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
  FOR (c:Chunk) ON c.embedding
  OPTIONS {indexConfig: {`vector.dimensions`: 1536, `vector.similarity_function`: 'cosine'}};

// Poll until index ONLINE:
// SHOW INDEXES YIELD name, state WHERE state <> 'ONLINE'

すべてのインデックスが ONLINE になるまで、取り込みを開始しないでください。

SHOW INDEXES YIELD name, state WHERE state <> 'ONLINE';

行が返された場合は、待ってから再実行してください。ONLINE であれば、安全に取り込めます。

---

よくあるエラー

エラー	原因	対処方法
LLM がスキーマにないノードタイプを抽出する	スキーマの制約が緩すぎる、または "EXTRACTED" モードになっている	entities と patterns を明示的に定義し、オプション B のスキーマを使用します
MissingEmbedderError	embedder= が省略されている	ベクトル検索を行わない場合でも、必ず embedder= を指定してください。パイプラインは Chunk ノードに埋め込みを保存します
エンティティがまったく抽出されない	LLM のコンテキストがあふれている	chunk_size を小さくするか、より大きなコンテキストに対応するモデルに切り替えます
取り込み後にエンティティノードが重複する	エンティティ解決が実行されていない	一括取り込み後に SinglePropertyExactMatchResolver を実行します
apoc.load.json のアクセスが拒否される	APOC の許可リストが設定されていない	URL を apoc.import.file.enabled=true と dbms.security.allow_csv_import_from_file_urls=true に追加します
チャンク分割で文の途中が欠落する	approximate=False（デフォルト）では正確なトークン数で切り分けられる	FixedSizeSplitter で approximate=True を設定します
chunk_size が大きすぎて LLM がタイムアウトする	抽出プロンプトとチャンクの合計がコンテキストを超えている	GPT-4o による抽出では chunk_size を 512 以下にします。絶対上限は 2048 です
Python 3.14 で SpaCySemanticMatchResolver が失敗する	spaCy は 3.14 以降に対応していない	FuzzyMatchResolver を使用するか、Python 3.13 にダウングレードします
neo4j-driver パッケージが見つからない	6.0 以降、パッケージ名が非推奨になっている	neo4j パッケージを使用します：pip install neo4j>=5.17.0
プロパティのない NodeType で ValidationError が発生する	v1.13.0 以降、NodeType には 1 つ以上のプロパティが必要	少なくとも PropertyType(name="name", type="STRING") を追加します。文字列リストのラベルには自動的に追加されます
from_pdf の非推奨警告が表示される	v1.15.0 で from_pdf=True が削除された	代わりに from_file=True を使用します
model_params の response_format が無視される	SimpleKGPipeline は OpenAI／VertexAI に対して構造化出力を自動的に有効にする（v1.14.0 以降）	model_params から response_format を削除します。パイプラインが管理します

---

検証チェックリスト

• 取り込み開始前に制約を作成し、ONLINE になっている
• 埋め込みを保存する前にベクトルインデックスを作成している
• chunk_size が埋め込みモデルの上限以内である（2048 以下、抽出では 512 以下）
• chunk_overlap を chunk_size の 10～15% に設定している
• Document→HAS_CHUNK→Chunk のパターンを使用している（検索時のグラフ走査が可能になります）
• document_metadata にソース識別子を設定している
• 一括取り込み後にエンティティリゾルバーを実行している
• apoc.load.json を使用する場合、apoc.version() を確認している
• .env に API キーがあり、.env が .gitignore に含まれている
• 構造を検証する：MATCH (d:Document)-[:HAS_CHUNK]->(c:Chunk) RETURN count(c)
• エンティティを検証する：MATCH (c:Chunk)-[:MENTIONS]->(e) RETURN labels(e)[0], count(*)

---

GraphSchema — 現行 API（≥1.8.0）

entities／relations／potential_schema は非推奨です。schema=GraphSchema(...) を使用してください。

from neo4j_graphrag.components.schema import (
    GraphSchema, NodeType, RelationshipType, PropertyType,
    ConstraintType, GraphConstraintType,
)

schema = GraphSchema(
    node_types=[
        NodeType(label="Person", properties=[PropertyType(name="name", type="STRING")]),
        NodeType(label="Organization", properties=[PropertyType(name="name", type="STRING")]),
    ],
    relationship_types=[RelationshipType(label="WORKS_AT")],
    patterns=[("Person", "WORKS_AT", "Organization")],
    # Constraints (v1.15.0+) — emitted to ParquetWriter metadata and enforce schema
    constraints=[
        # UNIQUENESS — property must be unique across nodes of that label
        ConstraintType(label="Person", property_name="name", type=GraphConstraintType.UNIQUENESS),
        # KEY — uniqueness + existence (null not allowed)
        ConstraintType(label="Organization", property_name="name", type=GraphConstraintType.KEY),
        # EXISTENCE — property must be non-null (replaces deprecated PropertyType.required)
        ConstraintType(label="Event", property_name="date", type=GraphConstraintType.EXISTENCE),
        # Composite KEY (v1.15.0+)
        ConstraintType(
            label="Person",
            property_names=("first_name", "last_name"),
            type=GraphConstraintType.KEY,
        ),
    ],
)
pipeline = SimpleKGPipeline(llm=llm, driver=driver, embedder=embedder, schema=schema)

schema="FREE"（ガイダンスなし）または schema="EXTRACTED"（LLM がタイプを推論）は、探索用途に限ってください。出力にノイズが増えます。

テキストからスキーマを自動抽出する（v1.15.0 以降）

SimpleKGPipeline に schema が渡されない場合、SchemaFromTextExtractor が自動的に実行されます。 明示的に実行するには、次のようにします。

from neo4j_graphrag.components.graph_schema_extraction import (
    SchemaFromTextExtractor,
    SchemaFromExistingGraphExtractor,
)

# Infer schema from sample text
extractor = SchemaFromTextExtractor(llm=llm, use_structured_output=True)
schema = asyncio.run(extractor.run(text=sample_text))

# Derive schema from an existing Neo4j graph
extractor = SchemaFromExistingGraphExtractor(driver=driver)
schema = asyncio.run(extractor.run())

Parquet エクスポート（実験的、v1.14.0 以降）

from neo4j_graphrag.components.parquet_output import ParquetWriter

# Use ParquetWriter instead of KGWriter inside a Pipeline to export to Parquet files
writer = ParquetWriter(output_dir="/data/kg_export/")
# Produces one Parquet file per node label and per relationship type
# Metadata includes UNIQUENESS, EXISTENCE, and KEY constraints (v1.15.0/1.16.0)

---

LexicalGraphConfig — ラベルのカスタマイズ

デフォルトの語彙レイヤーのラベルを上書きします（既存のグラフと統合する場合を除き、デフォルトのままにしてください）。

from neo4j_graphrag.components.types import LexicalGraphConfig
# All fields have sensible defaults — only override what differs from your graph's conventions
config = LexicalGraphConfig(
    document_node_label="Article",             # default: "Document"
    chunk_node_label="Passage",                # default: "Chunk"
    node_to_chunk_relationship_type="HAS_ENTITY",  # default: "MENTIONS"
    chunk_text_property="content",             # default: "text"
)
pipeline = SimpleKGPipeline(..., lexical_graph_config=config)

---

カスタムドキュメントローダー

デフォルトの file_loader は、拡張子（.pdf→PdfLoader、.md→MarkdownLoader）に応じて自動的に振り分けます。 fsspec URI（s3://、gcs://）に対応しています。HTML／Web／カスタム形式には、DataLoader をサブクラス化して使用します。

from neo4j_graphrag.components.data_loader import DataLoader
from neo4j_graphrag.components.types import DocumentInfo, LoadedDocument

class WebPageLoader(DataLoader):
    async def run(self, filepath, metadata=None):
        import httpx
        text = httpx.get(filepath).text   # strip HTML in real impl
        return LoadedDocument(text=text,
            document_info=DocumentInfo(path=filepath, metadata=metadata))

pipeline = SimpleKGPipeline(..., file_loader=WebPageLoader(), from_file=True)

用途別のチャンク分割戦略とリゾルバーの全設定については、references/kg-construction.md を参照してください。

---

参考資料

必要に応じて読み込んでください：

• neo4j-graphrag KG Builder ガイド
• neo4j-graphrag ライブラリの概要
• LLM Graph Builder（ホスト版）
• LLM Graph Builder GitHub
• APOC の読み込みプロシージャ
• GraphAcademy：LLM を使用したナレッジグラフの構築
• LangChain Neo4j 統合
• LlamaIndex Neo4jQueryEngine
• KG 構築の詳細なリファレンス