(1/3) 문서에서 지식 그래프 만들기 — 한 번 뽑고 끝이 아니라 계속 관리되게
- 문서에서 지식 그래프 만들기 — 한 번 뽑고 끝이 아니라 계속 관리되게 ← 지금 글
- GraphRAG 제대로 얹기 — 벡터 검색과 그래프 탐색을 한 파이프라인에서 섞기
- GraphRAG, 진짜 나아졌나 — 평가 세트로 일반 RAG 와 수치 비교하기
Summary
ChromaDB RAG 파이프라인 글에서 문서를 벡터로 잘라 넣고 질문과 비슷한 조각을 찾아 답을 만드는 구조를 정리했어요. 이 방식은 “정산 시스템 장애 대응 절차가 뭐야?” 같은 질문엔 강한데, “정산 시스템이 죽으면 어떤 시스템이 연달아 영향을 받아?” 같은 질문엔 약합니다. 답이 한 문서에 안 들어있고, 여러 문서에 흩어진 관계를 이어야 나오거든요. 이럴 때 필요한 게 지식 그래프(knowledge graph)예요.
그런데 지식 그래프를 만들어본 분들이 공통으로 하는 말이 있어요. 만드는 건 하루, 썩는 건 일주일. LLM 으로 문서에서 트리플(주어-관계-목적어)을 뽑는 것 자체는 이제 어렵지 않은데, 문서가 갱신되고 새 문서가 들어올 때마다 그래프를 같이 맞는 상태로 유지하는 게 진짜 문제입니다. 이 글은 그 유지보수를 처음부터 설계에 넣은 파이프라인을 Python 으로 짭니다.
💡 이 글에서 다루는 것
- 지식 그래프가 금방 썩는 4가지 이유 와 각각에 대응하는 설계 원칙
- 저장 구조 — SQLite 트리플 스토어 + 모든 엣지에 출처(provenance) 남기기
- 스키마를 코드로 — 엔티티/관계 타입 화이트리스트로 LLM 출력 거르기
- LLM 트리플 추출 — structured output 으로 JSON 파싱 걱정 없애기
- 엔티티 정규화 — “정산 시스템”과 “정산시스템”을 한 노드로 합치기
- 문서 단위 멱등 업서트 — 문서가 바뀌면 그 문서 출신 엣지만 갈아끼우기
- NetworkX 다단계 질의 + 그래프를 RAG 컨텍스트로 쓰는 법
예시에 나오는 시스템/팀 이름은 전부 가상의 값이에요.
1. 지식 그래프는 왜 금방 썩나
먼저 실패 패턴부터 보겠습니다. 문서 → LLM → 그래프를 순진하게 이으면 보통 이렇게 무너져요.
| 썩는 이유 | 증상 | 대응 원칙 |
|---|---|---|
| 문서는 계속 바뀜 | 조직 개편 후에도 그래프엔 옛 팀이 시스템을 소유 중 | 문서 단위 멱등 업서트 |
| 엣지의 근거를 모름 | “이 관계 맞아?”에 아무도 답 못 함 | 모든 엣지에 출처 기록 |
| LLM 출력이 매번 다름 | 같은 문서인데 돌릴 때마다 관계 이름이 제각각 | 스키마를 코드로 고정 |
| 같은 것이 여러 노드로 | “정산 시스템”, “정산시스템”, “settlement” 이 3개 노드 | 엔티티 정규화 + alias |
핵심은 네 가지 대응 원칙이 전부 쓰기 경로(ingest) 에 들어간다는 점이에요. 그래프가 만들어진 뒤에 청소하는 게 아니라, 들어올 때부터 썩을 수 없는 구조로 들어오게 만듭니다. 이건 PDF 적재 파이프라인 글에서 “검증은 적재 전에” 라고 했던 것과 같은 철학이에요.
전체 흐름을 그림으로 요약하면 다음과 같아요.
문서 텍스트
│
▼
[추출] LLM + structured output ── 트리플 후보 뽑기
│
▼
[검증] 스키마 화이트리스트 ── 모르는 타입/관계는 버리고 로그
│
▼
[정규화] normalize + alias ── 표기가 달라도 같은 노드로
│
▼
[업서트] 문서 단위 delete → insert ── 출처(doc_id) 달아서 멱등하게
2. 저장 구조 — 엣지에 출처를 박은 트리플 스토어
시작부터 Neo4j 같은 그래프 DB 를 세우는 것도 방법이지만, 저는 첫 버전은 SQLite 트리플 스토어 를 추천드려요. 파일 하나라 운영 부담이 없고, 스키마가 단순해서 나중에 그래프 DB 로 이관할 때도 테이블 덤프만 옮기면 됩니다. 유지보수 관점에서 중요한 건 어떤 DB 냐가 아니라 엣지마다 doc_id 가 붙어있느냐 예요.
import sqlite3
def init_db(path: str = "kg.db") -> sqlite3.Connection:
conn = sqlite3.connect(path)
conn.executescript("""
CREATE TABLE IF NOT EXISTS documents (
doc_id TEXT PRIMARY KEY, -- 예: "wiki/정산시스템_운영가이드"
doc_hash TEXT NOT NULL, -- 본문 해시 (변경 감지용)
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS nodes (
node_key TEXT PRIMARY KEY, -- 정규화된 키: "system:정산시스템"
node_type TEXT NOT NULL,
label TEXT NOT NULL -- 사람이 읽는 원래 표기
);
CREATE TABLE IF NOT EXISTS edges (
src_key TEXT NOT NULL,
relation TEXT NOT NULL,
dst_key TEXT NOT NULL,
doc_id TEXT NOT NULL, -- 이 엣지가 어느 문서에서 왔는지
evidence TEXT, -- 근거가 된 문장
PRIMARY KEY (src_key, relation, dst_key, doc_id)
);
CREATE TABLE IF NOT EXISTS aliases (
alias TEXT PRIMARY KEY, -- "settlement" 같은 별칭
node_key TEXT NOT NULL
);
""")
return conn
포인트 세 개만 짚을게요.
edges.doc_id— 같은 관계라도 문서마다 별도 행으로 쌓여요. 그래서 “이 관계의 근거 문서가 2개” 같은 신뢰도 판단이 가능하고, 특정 문서가 갱신되면 그 문서 출신 엣지만 지울 수 있어요. 이 컬럼 하나가 이 글의 절반입니다.edges.evidence— 근거 문장을 같이 저장해요. 나중에 “이 관계 진짜야?” 라는 질문에 그래프가 스스로 답할 수 있게 됩니다.documents.doc_hash— 본문 해시로 변경을 감지해서, 안 바뀐 문서는 LLM 호출 자체를 건너뜁니다. 추출 비용이 문서 수가 아니라 변경된 문서 수 에 비례하게 돼요.
3. 스키마를 코드로 — LLM 이 아니라 코드가 어휘를 정한다
LLM 에게 자유롭게 관계를 뽑게 하면 같은 의미가 운영한다, 담당한다, manages, operates 로 흩어져요. 오늘 뽑은 그래프와 다음 주에 뽑은 그래프가 서로 다른 언어를 쓰는 셈이라, 질의가 불가능해집니다. 그래서 어휘(온톨로지)는 코드에 고정 하고, LLM 은 그 어휘 안에서만 고르게 해요.
ENTITY_TYPES = {"system", "team", "person", "database", "document"}
# 관계 이름 → (허용되는 주어 타입, 허용되는 목적어 타입)
RELATION_TYPES = {
"owns": ("team", "system"), # 팀이 시스템을 소유
"operates": ("person", "system"), # 사람이 시스템을 운영
"depends_on": ("system", "system"), # 시스템이 시스템에 의존
"stores_in": ("system", "database"), # 시스템이 DB 에 저장
}
def validate_triple(src_type: str, relation: str, dst_type: str) -> tuple[bool, str]:
if src_type not in ENTITY_TYPES or dst_type not in ENTITY_TYPES:
return False, f"unknown entity type: {src_type} / {dst_type}"
if relation not in RELATION_TYPES:
return False, f"unknown relation: {relation}"
want_src, want_dst = RELATION_TYPES[relation]
if (src_type, dst_type) != (want_src, want_dst):
return False, f"{relation} expects ({want_src} -> {want_dst}), got ({src_type} -> {dst_type})"
return True, "ok"
실제로 넣어보면 이렇게 동작해요.
print(validate_triple("team", "owns", "system"))
print(validate_triple("person", "owns", "system")) # 사람이 소유? 스키마 위반
print(validate_triple("system", "manages", "team")) # 모르는 관계
(True, 'ok')
(False, 'owns expects (team -> system), got (person -> system)')
(False, 'unknown relation: manages')
검증에서 떨어진 트리플은 버리되 로그로 남기는 것 을 추천드려요. 같은 관계 이름이 로그에 반복해서 쌓이면, 그건 LLM 의 실수가 아니라 스키마에 그 관계가 필요하다는 신호거든요. 스키마 확장은 이 로그를 보고 사람이 결정합니다. 이렇게 하면 어휘가 흘러내리는(drift) 걸 막으면서도 필요한 확장은 놓치지 않아요.
⚠️ 스키마에 관계를 추가하는 건 쉽지만 빼거나 이름을 바꾸는 건 마이그레이션 이에요.
UPDATE edges SET relation = ...으로 기존 데이터를 같이 옮겨야 합니다. 그래서 처음엔 관계를 5~10개로 최소하게 시작하는 게 좋아요.
4. LLM 트리플 추출 — structured output 으로 어휘 강제
이제 문서에서 트리플을 뽑습니다. 여기서 두 가지 장치를 써요. 첫째, 프롬프트에 스키마를 그대로 넣어서 어휘를 알려주고, 둘째, structured output(JSON 스키마 강제) 으로 응답 형식을 못 벗어나게 합니다. enum 에 허용 타입/관계를 박아두면 형식이 어긋난 JSON 을 파싱하다 터지는 일 자체가 없어져요.
import json
import anthropic
client = anthropic.Anthropic()
TRIPLE_SCHEMA = {
"type": "object",
"properties": {
"triples": {
"type": "array",
"items": {
"type": "object",
"properties": {
"src": {"type": "string"},
"src_type": {"type": "string", "enum": sorted(ENTITY_TYPES)},
"relation": {"type": "string", "enum": sorted(RELATION_TYPES)},
"dst": {"type": "string"},
"dst_type": {"type": "string", "enum": sorted(ENTITY_TYPES)},
"evidence": {"type": "string"},
},
"required": ["src", "src_type", "relation", "dst", "dst_type", "evidence"],
"additionalProperties": False,
},
}
},
"required": ["triples"],
"additionalProperties": False,
}
def extract_triples(doc_text: str) -> list[dict]:
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=4096,
system=(
"너는 사내 문서에서 지식 그래프 트리플을 추출한다. "
"정의된 엔티티 타입과 관계 안에서만 추출하고, "
"문서에 명시된 사실만 뽑는다. 추측은 금지. "
"각 트리플에는 근거가 된 문장을 evidence 로 그대로 인용한다."
),
output_config={"format": {"type": "json_schema", "schema": TRIPLE_SCHEMA}},
messages=[{"role": "user", "content": doc_text}],
)
return json.loads(response.content[0].text)["triples"]
가상의 운영 문서 한 조각을 넣어볼게요.
doc = """
정산 시스템은 결제플랫폼팀이 소유한다. 일 배치는 김운영 매니저가 담당한다.
정산 시스템은 집계가 끝나면 결과를 정산DB 에 적재하며,
원천 데이터는 주문 시스템에 의존한다.
"""
triples = extract_triples(doc)
print(json.dumps(triples, ensure_ascii=False, indent=2))
[
{"src": "결제플랫폼팀", "src_type": "team", "relation": "owns",
"dst": "정산 시스템", "dst_type": "system",
"evidence": "정산 시스템은 결제플랫폼팀이 소유한다."},
{"src": "김운영", "src_type": "person", "relation": "operates",
"dst": "정산 시스템", "dst_type": "system",
"evidence": "일 배치는 김운영 매니저가 담당한다."},
{"src": "정산 시스템", "src_type": "system", "relation": "stores_in",
"dst": "정산DB", "dst_type": "database",
"evidence": "집계가 끝나면 결과를 정산DB 에 적재하며"},
{"src": "정산 시스템", "src_type": "system", "relation": "depends_on",
"dst": "주문 시스템", "dst_type": "system",
"evidence": "원천 데이터는 주문 시스템에 의존한다."}
]
스키마가 enum 으로 강제되어 있어도, 방어선은 이중으로 둡니다. 추출 결과를 3장의 validate_triple 에 한 번 더 통과시켜요. 모델이 바뀌거나 스키마 파일과 프롬프트가 어긋나는 날이 반드시 오는데, 그때 조용히 오염되는 것보다 검증에서 시끄럽게 떨어지는 쪽이 훨씬 낫습니다.
💡 “문서에 명시된 사실만, 추측 금지” 한 줄이 생각보다 중요해요. 이게 없으면 LLM 이 상식으로 그럴듯한 관계를 만들어 넣는데, 그래프에서 추측과 사실이 섞이는 순간 전체 신뢰도가 무너집니다. evidence 를 필수로 받는 것도 같은 목적이에요 — 인용할 문장이 없으면 뽑을 수 없게 만드는 겁니다.
5. 엔티티 정규화 — 표기가 달라도 같은 노드로
문서마다 같은 대상을 다르게 불러요. “정산 시스템”, “정산시스템”, “settlement 시스템”이 각각 노드가 되면 그래프가 세 조각으로 갈라지고, 다단계 질의는 그 시점에 끝납니다. 그래서 노드 키를 만들 때 정규화를 거쳐요.
import re
# 사람이 관리하는 별칭 사전 — 자동화가 아니라 운영 데이터
ALIASES = {
"settlement": "정산시스템",
"settlement시스템": "정산시스템",
"정산배치": "정산시스템",
}
def normalize_key(node_type: str, label: str) -> str:
slug = re.sub(r"[\s\-_]+", "", label).lower() # 공백/하이픈 제거 + 소문자
slug = ALIASES.get(slug, slug) # 별칭이면 대표 표기로
return f"{node_type}:{slug}"
print(normalize_key("system", "정산 시스템"))
print(normalize_key("system", "정산시스템"))
print(normalize_key("system", "Settlement"))
system:정산시스템
system:정산시스템
system:정산시스템
세 표기가 전부 같은 키로 떨어졌어요. 여기서 운영 요령이 하나 있는데, 별칭 사전은 완벽하게 만들려고 하지 않는 것 입니다. 처음엔 비워두고, 그래프를 쓰다가 갈라진 노드를 발견할 때마다 한 줄씩 추가하세요. 별칭 사전은 2장의 aliases 테이블에 같이 저장해두면, 나중에 파이프라인을 다시 돌려도 같은 정규화가 재현됩니다.
노드 키에 타입을 접두어로 붙이는 이유도 있어요. document:정산시스템 (정산 시스템 운영 가이드 문서)과 system:정산시스템 (시스템 그 자체)은 다른 대상인데, 라벨만으로 합치면 이 둘이 붙어버리거든요.
6. 문서 단위 멱등 업서트 — 이 글의 핵심
이제 유지보수의 심장부입니다. 규칙은 한 문장이에요.
✅ 문서가 바뀌면, 그 문서에서 나온 엣지를 전부 지우고 새로 뽑은 걸 넣는다.
엣지 하나하나를 비교해서 갱신하려고 하면 지옥이 열립니다. “이 엣지가 저 엣지의 수정본인가?” 를 판단할 방법이 없거든요. 문서 단위로 통째로 갈아끼우면 그런 판단이 필요 없고, 몇 번을 다시 돌려도 결과가 같은 멱등 파이프라인이 됩니다.
import hashlib
from datetime import datetime, timezone
def upsert_document(conn, doc_id: str, doc_text: str, triples: list[dict]) -> str:
doc_hash = hashlib.sha256(doc_text.encode()).hexdigest()[:16]
row = conn.execute(
"SELECT doc_hash FROM documents WHERE doc_id = ?", (doc_id,)
).fetchone()
if row and row[0] == doc_hash:
return "skip" # 내용 동일 → LLM 도 DB 도 건드리지 않음
# 1) 이 문서 출신 엣지만 제거 (다른 문서가 만든 엣지는 무사)
conn.execute("DELETE FROM edges WHERE doc_id = ?", (doc_id,))
# 2) 새로 뽑은 트리플 삽입
for t in triples:
src = normalize_key(t["src_type"], t["src"])
dst = normalize_key(t["dst_type"], t["dst"])
conn.execute("INSERT OR IGNORE INTO nodes VALUES (?, ?, ?)",
(src, t["src_type"], t["src"]))
conn.execute("INSERT OR IGNORE INTO nodes VALUES (?, ?, ?)",
(dst, t["dst_type"], t["dst"]))
conn.execute("INSERT OR REPLACE INTO edges VALUES (?, ?, ?, ?, ?)",
(src, t["relation"], dst, doc_id, t["evidence"]))
# 3) 어떤 엣지에도 등장하지 않는 고아 노드 정리
conn.execute("""
DELETE FROM nodes WHERE node_key NOT IN (
SELECT src_key FROM edges UNION SELECT dst_key FROM edges
)
""")
conn.execute("INSERT OR REPLACE INTO documents VALUES (?, ?, ?)",
(doc_id, doc_hash, datetime.now(timezone.utc).isoformat()))
conn.commit()
return "reindexed" if row else "created"
동작을 확인해볼게요. 같은 문서를 두 번 넣고, 내용을 바꿔서 한 번 더 넣습니다.
conn = init_db()
print(upsert_document(conn, "wiki/정산가이드", doc, triples))
print(upsert_document(conn, "wiki/정산가이드", doc, triples)) # 그대로 재실행
doc_v2 = doc.replace("김운영", "박신입") # 담당자 변경
print(upsert_document(conn, "wiki/정산가이드", doc_v2, extract_triples(doc_v2)))
for row in conn.execute("SELECT src_key, relation, dst_key FROM edges"):
print(row)
created
skip
reindexed
('team:결제플랫폼팀', 'owns', 'system:정산시스템')
('person:박신입', 'operates', 'system:정산시스템')
('system:정산시스템', 'stores_in', 'database:정산db')
('system:정산시스템', 'depends_on', 'system:주문시스템')
담당자가 바뀐 문서를 다시 넣었더니 person:김운영 의 엣지가 사라지고 person:박신입 이 들어왔어요. 그리고 어떤 엣지에도 남지 않은 김운영 노드는 고아 정리 단계에서 같이 사라집니다. 누구도 그래프를 수동으로 고치지 않았다 는 게 포인트예요. 문서가 진실이고, 그래프는 문서를 따라옵니다.
🚨 고아 노드 정리는 취향이 갈릴 수 있어요. 퇴사자/폐기 시스템의 이력을 남기고 싶다면 DELETE 대신
archived플래그를 세우는 방식으로 바꾸세요. 다만 기본값은 “그래프는 현재 상태만 반영”으로 두는 게 질의를 단순하게 유지해 줍니다.
7. 운영 루틴 — 주기 작업으로 돌리기
여기까지 만들면 운영은 단순해져요. 할 일은 세 가지뿐입니다.
| 루틴 | 주기 | 하는 일 |
|---|---|---|
| 증분 재색인 | 매일 | 문서 목록 순회 → 해시 다르면 추출 + 업서트 |
| 스키마 위반 리뷰 | 매주 | 검증 탈락 로그 확인 → 필요하면 관계 추가 |
| 노드 분열 리뷰 | 매주 | 비슷한 라벨의 노드 쌍 확인 → 별칭 사전에 추가 |
증분 재색인은 Airflow 개인정보 게이트 글에서 쓰던 것과 같은 모양의 @task 하나로 충분해요.
@task
def reindex_knowledge_graph():
conn = init_db("/data/kg.db")
stats = {"created": 0, "reindexed": 0, "skip": 0}
for doc_id, doc_text in iter_documents(): # 위키/문서 저장소 순회
if changed(conn, doc_id, doc_text): # 해시 비교 먼저
triples = [t for t in extract_triples(doc_text)
if validate_triple(t["src_type"], t["relation"], t["dst_type"])[0]]
else:
triples = []
stats[upsert_document(conn, doc_id, doc_text, triples)] += 1
return stats # 예: {'created': 3, 'reindexed': 7, 'skip': 190}
마지막에 리턴하는 통계가 중요합니다. skip 이 대부분이고 reindexed 가 소수인 게 정상 상태예요. 어느 날 갑자기 reindexed 가 폭증하면 문서 저장소 쪽에 무슨 일이 생긴 것이고, 노드/엣지 수가 갑자기 줄면 추출이 망가진 겁니다. 이 숫자를 Slack 리포트로 흘려두면 그래프가 조용히 썩는 일은 없어요.
8. 활용 — 다단계 질의와 GraphRAG
이제 처음의 질문으로 돌아갑니다. “정산 시스템이 죽으면 어떤 시스템이 영향을 받아?” 는 그래프에선 탐색 한 줄이에요.
import networkx as nx
G = nx.DiGraph()
for src, rel, dst in conn.execute(
"SELECT src_key, relation, dst_key FROM edges WHERE relation = 'depends_on'"):
G.add_edge(src, dst) # A → B : A 가 B 에 의존
# 정산시스템에 (직접이든 몇 단계를 거치든) 의존하는 모든 시스템
impacted = nx.ancestors(G, "system:정산시스템")
print(impacted)
{'system:재무리포트', 'system:세금계산서발행'}
재무리포트가 정산에 직접 의존하고, 세금계산서 발행이 재무리포트에 의존하는 상황이라면, 벡터 검색으로는 절대 한 번에 안 나오는 답이 두 홉 탐색으로 나옵니다.
그리고 이 그래프는 기존 RAG 와 경쟁하는 게 아니라 합쳐집니다. 질문에서 엔티티를 찾고, 그 주변 서브그래프를 문장으로 풀어서 LLM 컨텍스트에 벡터 검색 결과와 같이 넣는 방식이에요. 흔히 GraphRAG 라고 부르는 구성의 가장 작은 버전입니다.
def graph_context(conn, node_key: str) -> str:
rows = conn.execute("""
SELECT src_key, relation, dst_key, evidence FROM edges
WHERE src_key = ? OR dst_key = ?
""", (node_key, node_key)).fetchall()
return "\n".join(f"- {s} --{r}--> {d} (근거: {e})" for s, r, d, e in rows)
print(graph_context(conn, "system:정산시스템"))
- team:결제플랫폼팀 --owns--> system:정산시스템 (근거: 정산 시스템은 결제플랫폼팀이 소유한다.)
- person:박신입 --operates--> system:정산시스템 (근거: 일 배치는 박신입 매니저가 담당한다.)
- system:정산시스템 --stores_in--> database:정산db (근거: 집계가 끝나면 결과를 정산DB 에 적재하며)
- system:정산시스템 --depends_on--> system:주문시스템 (근거: 원천 데이터는 주문 시스템에 의존한다.)
엣지마다 evidence 를 저장해둔 게 여기서 빛나요. LLM 에게 관계만 주는 게 아니라 근거 문장까지 같이 주니까, 답변이 “그래프에 그렇게 되어 있음”이 아니라 원문 인용으로 나옵니다.
정리하면, 유지보수하기 편한 지식 그래프의 조건은 결국 이거예요.
- 모든 엣지에 출처와 근거 문장이 붙어 있다
- 어휘는 코드(스키마)가 정하고, LLM 은 그 안에서만 고른다
- 표기가 달라도 정규화 + 별칭 사전으로 한 노드가 된다
- 문서가 바뀌면 그 문서 출신 엣지만 통째로 교체된다 (멱등)
- 스키마 위반 로그와 재색인 통계가 사람에게 흘러온다
일단 오늘은 여기까지….. 다음 글에서는 이 그래프 위에 GraphRAG 를 제대로 얹어서, 벡터 검색과 그래프 탐색을 한 파이프라인에서 섞는 구성을 정리해볼게요.