실전 GraphRAG 파이프라인: PDF에서 답변까지

개념(왜 GraphRAG인가)과 문법(Cypher CRUD)을 모두 익혔으니, 이제 실제 문서 한 편으로 엔드투엔드 파이프라인을 만들어봅니다. 예제는 KB손해보험 펫보험 약관 PDF(16페이지)입니다. 시작은 벡터 RAG와 똑같이 Load→Split이지만, 그다음부터 길이 갈라집니다.

1) Load — PDF 파싱 + 전처리

랭체인의 PDFPlumberLoader로 페이지별 텍스트를 추출합니다. 약관 PDF에는 목차의 점선(......)이나 홀로 떠 있는 페이지 번호 같은 노이즈가 섞이므로, 간단한 정규식으로 정리한 뒤 결과를 parsed_docs.jsonl로 저장해 다음 단계에서 재사용합니다.

1_load.py — PDFPlumberLoader + 전처리

import re, json
from langchain_community.document_loaders import PDFPlumberLoader

loader = PDFPlumberLoader("kb_pet_insurance.pdf")
docs = loader.load()  # 페이지별 Document (16페이지)

def clean(text: str) -> str:
    text = re.sub(r"\.{4,}", " ", text)   # 목차 점선(......) 제거
    return text.strip()

with open("parsed_docs.jsonl", "w", encoding="utf-8") as f:
    for d in docs:
        rec = {"page": d.metadata.get("page", 0), "text": clean(d.page_content)}
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

2) Split — 청킹

긴 텍스트를 LLM 컨텍스트에 맞게 자릅니다. 이때 각 청크에 고유 ID를 붙여두는 것이 중요합니다. 뒤에서 MERGE로 그래프에 적재할 때, 이 ID가 없으면 같은 청크가 중복되거나 서로 충돌하기 때문입니다.

2_split.py — RecursiveCharacterTextSplitter + 고유 ID

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=80)
chunks = splitter.split_documents(docs)  # 16페이지 → 약 41개 청크

# 각 청크에 고유 id 부여 (뒤 단계 MERGE 충돌 방지)
for i, c in enumerate(chunks):
    c.metadata["chunk_id"] = f"chunk_{i}"
    c.metadata["index"] = i

print(len(chunks), "개 청크")  # 41

3) 렉시컬 그래프 적재(Ingest) — 문서 구조를 그래프로 (LLM 없이)

먼저 문서의 '뼈대'를 그래프로 옮깁니다. Document → Page → Chunk 노드와 그 사이 관계(HAS_PAGE, HAS_CHUNK, NEXT_CHUNK)를 만드는데, 이 정보는 이미 청크 메타데이터(페이지 번호·순서)에 들어 있습니다. 따라서 LLM을 호출할 필요 없이 MERGE로 곧장 적재할 수 있습니다.

제약조건 + MERGE로 구조 적재 (개념)

// 청크 id 중복을 원천 차단하는 제약조건 (최초 1회)
CREATE CONSTRAINT chunk_id IF NOT EXISTS
FOR (c:Chunk) REQUIRE c.id IS UNIQUE;

// 문서·페이지·청크와 그 관계를 안전하게 적재
MERGE (d:Document {id: 'kb_pet_insurance'})
MERGE (p:Page {id: 'kb_pet_insurance-p1'})
MERGE (d)-[:HAS_PAGE]->(p)
MERGE (c:Chunk {id: 'chunk_0'})
  SET c.text = '...'
MERGE (p)-[:HAS_CHUNK]->(c)
MERGE (c)-[:NEXT_CHUNK]->(:Chunk {id: 'chunk_1'});

청크끼리 NEXT_CHUNK로 이어두면, 다음 KG Build 단계에서 LLM이 앞뒤 청크의 맥락을 함께 참고해 엔티티·관계를 더 안정적으로 추출합니다. 청킹 때문에 끊긴 문맥을 그래프의 순서 관계로 보완해주는 셈입니다.

4) KG Build — LLM이 엔티티·관계 추출

이제 문서의 '의미'를 그래프로 옮깁니다. 여기서 처음으로 LLM이 등장하며, 네 단계로 진행합니다.

• 1) 스키마 정의 — 허용할 노드·관계 종류를 개발자가 미리 고정
• 2) LLM 추출 — 각 청크에서 구조화 출력으로 노드·관계를 뽑음
• 3) 프루닝 — 실존하지 않는 노드를 가리키는 엉뚱한 관계를 걸러냄
• 4) 저장 — MERGE로 그래프에 적재 (중복 노드 없이)
• (선택) 엔티티 리졸버 — '보안팀'과 '보안 팀'처럼 유사 개체를 하나로 통합

4a. 스키마 정의 — Pydantic + Literal로 고정

from typing import Literal, List
from pydantic import BaseModel, Field

# 개발자가 허용 노드·관계를 고정 → LLM이 일관되게 추출
NodeType = Literal["제품", "특약", "조항", "금액", "기간"]
RelType = Literal["보장한다", "제외된다", "포함한다", "지급한다"]

class Node(BaseModel):
    id: str
    type: NodeType

class Relationship(BaseModel):
    source: str
    target: str
    type: RelType

class KnowledgeGraph(BaseModel):
    nodes: List[Node] = Field(default_factory=list)
    relationships: List[Relationship] = Field(default_factory=list)

4b. 추출 + 프루닝 — with_structured_output

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
extractor = llm.with_structured_output(KnowledgeGraph)

all_nodes, all_rels = {}, []
for c in chunks:
    kg = extractor.invoke(
        f"다음 약관 청크에서 노드와 관계를 추출하세요.\n\n{c.page_content}"
    )
    for n in kg.nodes:
        all_nodes[n.id] = n          # id 기준으로 중복 제거
    all_rels.extend(kg.relationships)

# 프루닝: 실존 노드를 잇는 관계만 남긴다
valid = set(all_nodes.keys())
pruned = [r for r in all_rels if r.source in valid and r.target in valid]

# 이후 graph.query(...) 로 MERGE 하며 저장
print(len(all_nodes), len(pruned))  # 예: 177 497

품질의 핵심은 '스키마 고정'입니다. 허용 노드·관계를 Literal로 못 박아 두면 LLM이 제멋대로 라벨을 만들지 못해 그래프가 일관됩니다. 이 16페이지 약관에서도 스키마를 고정한 덕에 약 177개 노드·497개 관계가 뒤죽박죽 없이 추출됐습니다.

5) Retrieve — 질문 → Cypher → 답변

그래프가 완성됐으니 질문할 차례입니다. GraphCypherQAChain이 자연어 질문을 Cypher로 번역해 그래프를 조회하고, 그 결과를 LLM에 넣어 최종 답변을 만듭니다.

5a. 기본 질의

from langchain_neo4j import Neo4jGraph, GraphCypherQAChain
from langchain_openai import ChatOpenAI

chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    graph=graph,
    verbose=True,
    allow_dangerous_requests=True,  # 로컬 학습용. 운영은 읽기 전용 계정 권장
)
print(chain.invoke({"query": "보험금은 언제 지급되나요?"})["result"])

5b. 커스텀 Cypher 프롬프트로 품질·안전 높이기

from langchain_core.prompts import PromptTemplate
from langchain_neo4j import GraphCypherQAChain
from langchain_openai import ChatOpenAI

CYPHER_PROMPT = PromptTemplate(
    input_variables=["schema", "question"],
    template=(
        "당신은 Neo4j Cypher 전문가입니다.\n"
        "아래 스키마만 사용해 질문에 답할 Cypher를 작성하세요.\n"
        "CREATE/MERGE/DELETE 같은 쓰기 구문은 절대 쓰지 마세요.\n\n"
        "스키마:\n{schema}\n\n질문: {question}\nCypher:"
    ),
)

chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    graph=graph,
    cypher_prompt=CYPHER_PROMPT,
    top_k=10,                 # 조회 결과 상위 몇 개를 LLM에 넘길지
    verbose=True,
    allow_dangerous_requests=True,
)

실제로 KB 펫보험 그래프에 물어보면, "보험금은 언제 지급되나요?" 질문에는 LLM이 Cypher를 생성·조회해 "지급 사유가 결정되면 7일 이내 지급"으로, "손해배상금은 어디에 규정돼 있나요?" 질문에는 "제1조 제2항 제1호"처럼 근거 조항까지 답합니다. 다만 근거가 그래프에 담기지 않은 질문에는 "알 수 없음"이라고 정직하게 답하기도 합니다 — 이 한계를 아는 것도 중요합니다.

지금 만든 것은 관계에만 의존하는 '나이브 GraphRAG'입니다. 여기에 벡터 검색과 키워드(BM25) 검색을 더한 '하이브리드 GraphRAG'로 확장하면, 관계형 질문과 의미형 질문을 한 시스템에서 모두 커버할 수 있습니다.

LLM이 추출한 그래프에는 오류·중복 노드가 섞일 수 있으니 스키마 고정과 사람의 검수가 필요하고, KG Build·Retrieve처럼 LLM·임베딩을 호출하는 단계마다 API 비용이 듭니다. 특히 allow_dangerous_requests=True는 LLM이 생성한 Cypher를 검증 없이 그대로 실행하므로, 운영에서는 읽기 전용 계정과 쿼리 검증 장치를 반드시 두세요.

로드 → 분할 → 적재 → 그래프 구축 → 검색까지, 실제 약관 문서 한 편으로 GraphRAG 전 과정을 완주했습니다. 여기에 벡터·키워드 검색을 얹으면 하이브리드 GraphRAG로 한 걸음 더 나아갈 수 있습니다.