document-graph-pipeline

name	document-graph-pipeline
description	Excel/PDF → Table Transformer → Document AST → Neo4j / GraphRAG パイプラインを構築するスキル。 ドキュメントからテーブルを抽出し、AST化してNeo4jに取り込み、GraphRAGクエリを可能にする。 Use when: the user wants to parse Excel/PDF documents into a graph database for RAG.

Document Graph Pipeline Skill

概要

Excel / PDF
    ↓  [Step 1] 入力パーサー
Table Transformer
    ↓  [Step 2] テーブル抽出・構造化
Document AST
    ↓  [Step 3] グラフモデル変換
Neo4j / GraphRAG
    ↓  [Step 4] クエリ・RAG

Workflow

TodoWriteで全タスクをリストアップしてから着手する。

---

Step 0: 環境調査

既存のファイルと設定を確認する。

# 入力ファイルの確認
ls -la *.xlsx *.pdf 2>/dev/null || find . -name "*.xlsx" -o -name "*.pdf" | head -20

# Neo4j接続情報の確認
echo "NEO4J_URI=${NEO4J_URI:-bolt://localhost:7687}"
echo "NEO4J_USER=${NEO4J_USER:-neo4j}"
echo "NEO4J_PASSWORD=${NEO4J_PASSWORD:-(未設定)}"

# Python環境
python3 --version && pip list 2>/dev/null | grep -E "neo4j|transformers|pdfplumber|openpyxl|pandas|sentence"

不足しているものを把握してから次へ進む。

---

Step 1: 依存ライブラリのインストール

pip install \
  pdfplumber \
  pymupdf \
  openpyxl \
  pandas \
  pillow \
  transformers \
  timm \
  torch \
  torchvision \
  neo4j \
  sentence-transformers \
  anthropic

ライブラリ選定の理由:

• pdfplumber — テキストベースPDFからテーブルを直接抽出
• pymupdf (fitz) — PDF→画像変換（Table Transformer用）
• transformers + timm — microsoft/table-transformer-detection モデル
• openpyxl / pandas — Excelパース
• neo4j — Neo4j Python ドライバ
• sentence-transformers — ベクトル埋め込み（GraphRAG用）

---

Step 2: ディレクトリ構成の作成

pipeline/
├── __init__.py
├── parsers/
│   ├── __init__.py
│   ├── excel_parser.py      # Excel → raw tables
│   └── pdf_parser.py        # PDF → raw tables (pdfplumber + Table Transformer)
├── ast_builder.py           # raw tables → Document AST
├── neo4j_ingester.py        # Document AST → Neo4j
├── graphrag.py              # Neo4j + embedding → RAG query
├── models.py                # AST dataclass定義
└── main.py                  # エントリーポイント

mkdir -p pipeline/parsers
touch pipeline/__init__.py pipeline/parsers/__init__.py

---

Step 3: データモデル (models.py)

Document ASTのノード型を定義する。

# pipeline/models.py
from dataclasses import dataclass, field
from typing import List, Optional, Any
from enum import Enum
import uuid


class NodeType(str, Enum):
    DOCUMENT = "Document"
    SECTION = "Section"
    TABLE = "Table"
    ROW = "Row"
    CELL = "Cell"
    HEADER = "Header"


@dataclass
class ASTNode:
    node_type: NodeType
    node_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content: str = ""
    children: List["ASTNode"] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)
    embedding: Optional[List[float]] = None

    def to_dict(self) -> dict:
        return {
            "node_id": self.node_id,
            "node_type": self.node_type.value,
            "content": self.content,
            "metadata": .metadata,
        }

グラフ関係の設計:

(Document)-[:CONTAINS]->(Section)
(Section)-[:CONTAINS]->(Table)
(Table)-[:HAS_HEADER]->(Header)
(Table)-[:CONTAINS]->(Row)
(Row)-[:CONTAINS]->(Cell)
(Row)-[:NEXT]->(Row)         # 行間の順序
(Cell)-[:SAME_COLUMN]->(Cell) # 列方向の参照

---

Step 4: Excelパーサー (parsers/excel_parser.py)

# pipeline/parsers/excel_parser.py
import openpyxl
import pandas as pd
from pathlib import Path
from typing import List, Dict, Any


def parse_excel(file_path: str) -> List[Dict[str, Any]]:
    """
    Excelファイルの全シートからテーブルを抽出する。
    Returns: [{"sheet_name": str, "headers": [...], "rows": [[...], ...]}, ...]
    """
    path = Path(file_path)
    result = []

    wb = openpyxl.load_workbook(path, data_only=True)
    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
        df = pd.DataFrame(ws.values)

        if df.empty:
            continue

        # 最初の非空行をヘッダーとして検出
        header_row_idx = _find_header_row(df)
        if header_row_idx is None:
            continue

        headers = [str(c) if c is not None else "" for c in df.iloc[header_row_idx]]
        rows = []
        for _, row in df.iloc[header_row_idx + 1:].iterrows():
            row_data = [str(v) if v is not None    v  row]
             (v.strip()  v  row_data):  
                rows.append(row_data)

        result.append({
            : (path),
            : sheet_name,
            : headers,
            : rows,
        })

     result


 () ->  | :
     i, row  df.iterrows():
        non_null = [v  v  row  v     (v).strip()]
         (non_null) >= :
             i
     

---

Step 5: PDFパーサー (parsers/pdf_parser.py)

2段階戦略: pdfplumber（テキストPDF）→ 失敗時に Table Transformer（画像PDF）

# pipeline/parsers/pdf_parser.py
import pdfplumber
import fitz  # pymupdf
from pathlib import Path
from typing import List, Dict, Any
from PIL import Image
import io


def parse_pdf(file_path: str, use_table_transformer: bool = False) -> List[Dict[str, Any]]:
    """
    PDFからテーブルを抽出する。
    use_table_transformer=True でOCR/画像ベースPDFに対応。
    """
    path = Path(file_path)
    result = []

    with pdfplumber.open(path) as pdf:
        for page_num, page in enumerate(pdf.pages, 1):
            tables = page.extract_tables()
            if tables:
                for tbl_idx, table in enumerate(tables):
                    if not table or len(table) < 2:
                        continue
                    headers = [str(h) if h else "" for h in table[0]]
                    rows = [[str(c)  c    c  row]  row  table[:]]
                    result.append({
                        : (path),
                        : page_num,
                        : tbl_idx,
                        : headers,
                        : rows,
                        : ,
                    })

      result  use_table_transformer:
        result = _extract_with_table_transformer(path)

     result


 () -> [[, ]]:
    
     transformers  AutoImageProcessor, TableTransformerForObjectDetection
     torch

    detection_model_name = 
    structure_model_name = 

    det_processor = AutoImageProcessor.from_pretrained(detection_model_name)
    det_model = TableTransformerForObjectDetection.from_pretrained(detection_model_name)
    str_processor = AutoImageProcessor.from_pretrained(structure_model_name)
    str_model = TableTransformerForObjectDetection.from_pretrained(structure_model_name)

    doc = fitz.((path))
    result = []

     page_num, page  (doc, ):
        pix = page.get_pixmap(dpi=)
        img = Image.(io.BytesIO(pix.tobytes())).convert()

        
        inputs = det_processor(images=img, return_tensors=)
         torch.no_grad():
            outputs = det_model(**inputs)

        target_sizes = torch.tensor([img.size[::-]])
        detections = det_processor.post_process_object_detection(
            outputs, threshold=, target_sizes=target_sizes
        )[]

         box_idx, (score, label, box)  (
            (detections[], detections[], detections[])
        ):
             label.item() == :  
                x0, y0, x1, y1 = [(v)  v  box.tolist()]
                table_img = img.crop((x0, y0, x1, y1))
                table_data = _recognize_table_structure(
                    table_img, str_processor, str_model
                )
                 table_data:
                    result.append({
                        : (path),
                        : page_num,
                        : box_idx,
                        : table_data.get(, []),
                        : table_data.get(, []),
                        : ,
                        : [x0, y0, x1, y1],
                    })

     result


 () -> [, ]:
    
     torch

    inputs = processor(images=table_img, return_tensors=)
     torch.no_grad():
        outputs = model(**inputs)

    target_sizes = torch.tensor([table_img.size[::-]])
    results = processor.post_process_object_detection(
        outputs, threshold=, target_sizes=target_sizes
    )[]

    
    
    label_map = {v: k  k, v  model.config.label2id.items()}
    rows, headers = [], []

     score, label, box  (
        results[], results[], results[]
    ):
        label_name = label_map.get(label.item(), )
         label_name == :
            rows.append(box.tolist())
         label_name == :
            headers.append(box.tolist())

    
    
     {
        : [  i  ((headers))],
        : [[]  _  rows],
        : rows,
        : headers,
    }

---

Step 6: Document AST ビルダー (ast_builder.py)

# pipeline/ast_builder.py
from typing import List, Dict, Any
from pathlib import Path
from .models import ASTNode, NodeType


def build_ast(tables: List[Dict[str, Any]], source_name: str = "") -> ASTNode:
    """
    抽出されたテーブルリストからDocument ASTを構築する。
    """
    doc_node = ASTNode(
        node_type=NodeType.DOCUMENT,
        content=source_name,
        metadata={"source": source_name, "table_count": len(tables)},
    )

    # シートまたはページ単位でSectionを作成
    sections: Dict[str, ASTNode] = {}

    for table_data in tables:
        section_key = _get_section_key(table_data)
        if section_key not in sections:
            section = ASTNode(
                node_type=NodeType.SECTION,
                content=section_key,
                metadata=_extract_section_meta(table_data),
            )
            sections[section_key] = section
            doc_node.children.append(section)

        table_node = _build_table_node(table_data)
        sections[section_key].children.append(table_node)

    return doc_node


def _get_section_key(table_data: Dict[str, Any]) -> str:
    if "sheet_name" in table_data:
        return 
     


 () -> :
    meta = {: table_data.get(, )}
       table_data:
        meta[] = table_data[]
       table_data:
        meta[] = table_data[]
     meta


 () -> ASTNode:
    headers = table_data.get(, [])
    rows = table_data.get(, [])

    table_node = ASTNode(
        node_type=NodeType.TABLE,
        content=,
        metadata={
            : (headers),
            : (rows),
            : table_data.get(, ),
            : table_data.get(, ),
        },
    )

    
     headers:
        header_node = ASTNode(
            node_type=NodeType.HEADER,
            content=.join(headers),
            metadata={: headers},
        )
         col_name  headers:
            header_node.children.append(
                ASTNode(node_type=NodeType.CELL, content=col_name,
                        metadata={: })
            )
        table_node.children.append(header_node)

    
     row_idx, row_values  (rows):
        row_node = ASTNode(
            node_type=NodeType.ROW,
            content=.join(row_values),
            metadata={: row_idx},
        )
         col_idx, cell_val  (row_values):
            col_name = headers[col_idx]  col_idx < (headers)  
            row_node.children.append(
                ASTNode(
                    node_type=NodeType.CELL,
                    content=cell_val,
                    metadata={: col_name, : col_idx, : row_idx},
                )
            )
        table_node.children.append(row_node)

     table_node

---

Step 7: Neo4j インジェスター (neo4j_ingester.py)

# pipeline/neo4j_ingester.py
import os
from neo4j import GraphDatabase
from typing import Optional
from .models import ASTNode, NodeType


NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password")


def get_driver():
    return GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))


def setup_schema(driver):
    """インデックスと制約を作成する。"""
    with driver.session() as session:
        session.run(
            "CREATE CONSTRAINT doc_id IF NOT EXISTS "
            "FOR (n:Document) REQUIRE n.node_id IS UNIQUE"
        )
        for label in ["Section", "Table", "Header", "Row", "Cell"]:
            session.run(
                f"CREATE INDEX {label.lower()}_id IF NOT EXISTS "
                f"FOR (n:{label}) ON (n.node_id)"
            )
        # ベクトルインデックス（Neo4j 5.x+）
        session.run("""
            CREATE VECTOR INDEX cell_embedding IF NOT EXISTS
            FOR (n:Cell) ON (n.embedding)
            OPTIONS {indexConfig: {
                `vector.dimensions`: 384,
                `vector.similarity_function`: 'cosine'
            }}
        """)


def ingest_ast():
    
    driver = driver  get_driver()
     driver.session()  session:
        _ingest_node(session, ast_root, parent_id=, embed_fn=embed_fn)


 ():
    props = node.to_dict()

     embed_fn  node.content.strip():
        node.embedding = embed_fn(node.content)
        props[] = node.embedding

    
    label = node.node_type.value
    session.run(
        ,
        node_id=node.node_id,
        props={k: v  k, v  props.items()  k != },
    )

    
     parent_id:
        rel_type = _get_rel_type(node.node_type)
        session.run(
            
            ,
            pid=parent_id, cid=node.node_id,
        )

    
     node.node_type == NodeType.ROW  prev_row_id:
        session.run(
            ,
            aid=prev_row_id, bid=node.node_id,
        )

    
    prev = 
     child  node.children:
        _ingest_node(session, child, parent_id=node.node_id, embed_fn=embed_fn,
                     prev_row_id=prev  child.node_type == NodeType.ROW  )
         child.node_type == NodeType.ROW:
            prev = child.node_id

    
     node.node_type == NodeType.TABLE:
        _link_same_column_cells(session, node)


 () -> :
     {
        NodeType.SECTION: ,
        NodeType.TABLE: ,
        NodeType.HEADER: ,
        NodeType.ROW: ,
        NodeType.CELL: ,
    }.get(node_type, )


 ():
    
    rows = [c  c  table_node.children  c.node_type == NodeType.ROW]
     (rows) < :
        

    col_count = ((r.children)  r  rows)
     col_idx  (col_count):
        col_cells = []
         row  rows:
            matching = [c  c  row.children
                         c.metadata.get() == col_idx]
             matching:
                col_cells.append(matching[].node_id)

         i  ((col_cells) - ):
            session.run(
                
                ,
                aid=col_cells[i], bid=col_cells[i + ],
            )

---

Step 8: GraphRAG クエリ (graphrag.py)

# pipeline/graphrag.py
import os
from neo4j import GraphDatabase
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Any

NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password")

_embed_model = None


def get_embed_model(model_name: str = "all-MiniLM-L6-v2") -> SentenceTransformer:
    global _embed_model
    if _embed_model is None:
        _embed_model = SentenceTransformer(model_name)
    return _embed_model


def embed(text: str) -> List[float]:
    return get_embed_model().encode(text).tolist()


def search(
    query: str,
    driver=None,
    top_k: int = 10,
    hop: int = 2,
) -> List[Dict[str, Any]]:
    """
    GraphRAGクエリ:
    1. クエリをベクトル化してCellの近傍検索（Vector RAG）
    2. ヒットしたCellから graph hop でコンテキスト収集（Graph RAG）
    """
    driver = driver  GraphDatabase.driver(
        NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD)
    )
    query_embedding = embed(query)

     driver.session()  session:
        
        vector_hits = session.run(
            ,
            top_k=top_k,
            embedding=query_embedding,
        ).data()

          vector_hits:
             []

        cell_ids = [h[]  h  vector_hits]

        
        graph_ctx = session.run(
            ,
            cell_ids=cell_ids,
        ).data()

         {
            : vector_hits,
            : graph_ctx,
        }


 () -> :
    
    lines = []
     hit  search_result.get(, []):
        meta = hit.get(, {})
        col = meta.get(, )
        lines.append()

    lines.append()
     node  search_result.get(, []):
        ntype = node.get(, )
        lines.append()

     .join(lines)

---

Step 9: エントリーポイント (main.py)

# pipeline/main.py
import argparse
import sys
from pathlib import Path

from .parsers.excel_parser import parse_excel
from .parsers.pdf_parser import parse_pdf
from .ast_builder import build_ast
from .neo4j_ingester import get_driver, setup_schema, ingest_ast
from .graphrag import embed, search, format_context_for_llm


def run_ingest(file_path: str, use_table_transformer: bool = False):
    path = Path(file_path)
    suffix = path.suffix.lower()

    print(f"[1/4] Parsing {suffix} file: {path.name}")
    if suffix in (".xlsx", ".xls"):
        tables = parse_excel(file_path)
    elif suffix == ".pdf":
        tables = parse_pdf(file_path, use_table_transformer=use_table_transformer)
    else:
        raise ValueError(f"Unsupported file type: {suffix}")

    print(f"  → {len(tables)} table(s) extracted")

    print("[2/4] Building Document AST")
    ast_root = build_ast(tables, source_name=path.name)
    total_nodes = _count_nodes(ast_root)
    print(f"  → {total_nodes} AST nodes")

    print("[3/4] Ingesting into Neo4j")
    driver = get_driver()
    setup_schema(driver)
    ingest_ast(ast_root, driver=driver, embed_fn=embed)
    ()

    ()
    ()


 ():
    ()
    result = search(query_text, top_k=top_k)
    context = format_context_for_llm(result)
    (context)
     context


 () -> :
      + (_count_nodes(c)  c  node.children)


 __name__ == :
    parser = argparse.ArgumentParser(description=)
    parser.add_argument(, metavar=, =)
    parser.add_argument(, metavar=, =)
    parser.add_argument(, =, default=)
    parser.add_argument(, action=,
                        =)
    args = parser.parse_args()

     args.ingest:
        run_ingest(args.ingest, use_table_transformer=args.table_transformer)
     args.query:
        run_query(args.query, top_k=args.top_k)
    :
        parser.print_help()

---

Step 10: 動作確認

# 環境変数設定
export NEO4J_URI="bolt://localhost:7687"
export NEO4J_USER="neo4j"
export NEO4J_PASSWORD="your_password"

# Excelファイルの投入
python -m pipeline.main --ingest sample.xlsx

# PDFの投入（テキストPDF）
python -m pipeline.main --ingest report.pdf

# PDFの投入（スキャン画像PDF → Table Transformer使用）
python -m pipeline.main --ingest scanned.pdf --table-transformer

# GraphRAGクエリ
python -m pipeline.main --query "売上が最も高い月は？"

Neo4j Browser で確認するクエリ:

// グラフ全体の俯瞰
MATCH (n) RETURN n LIMIT 200

// Document→Table→Row の経路確認
MATCH path=(d:Document)-[:CONTAINS*]->(r:Row)
RETURN path LIMIT 50

// 特定カラム列の縦断
MATCH (c:Cell {metadata: {column: '売上'}})
-[:SAME_COLUMN*]->(c2:Cell)
RETURN c.content, c2.content

---

Step 11: Claude API連携（オプション）

GraphRAGの文脈でClaude APIを使った回答生成:

import anthropic
from pipeline.graphrag import search, format_context_for_llm

def ask(question: str) -> str:
    result = search(question)
    context = format_context_for_llm(result)

    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="あなたは表形式データに精通したアナリストです。",
        messages=[{
            "role": "user",
            "content": f"以下のコンテキストを参照して質問に答えてください。\n\n{context}\n\n質問: {question}"
        }]
    )
    return response.content[0].text

---

拡張ポイント

課題	解決策
スキャンPDFのOCR	paddleocr または easyocr を _recognize_table_structure に統合
大規模ファイル	バッチ投入 + UNWIND でNeo4j書き込みを最適化
多言語埋め込み	paraphrase-multilingual-MiniLM-L12-v2 モデルを使用
リアルタイム更新	ファイル変更検知 (watchdog) → 差分投入
Cypher生成AI	LangChain GraphCypherQAChain または直接Cypherプロンプト

Wrap up

完了時に以下をレポートする:

• 投入したファイル数・テーブル数・ノード数
• Neo4j の接続URI
• 動作確認クエリの結果サンプル
• 未解決の問題（OCR精度、スキーマ設計の懸念など）