목차
- 문제와 접근
- Naive RAG 한계 복습
- 수동 KG와 그래프 탐색
- LLM JSON 추출 프롬프트
- strip_code_fence와 extract_graph
- Pydantic 기반 ExtractedKG
- JSON을 NetworkX 그래프로 변환
- alias 정규화
- 확장 KG 구성
- answer_via_kg
- 관계 설명 함수
- 핵심 개념 총정리
- 자주 나오는 실수/주의사항
- 그래프의 연결을 근거와 함께 검토하기
1. 문제와 접근
지식 그래프 추출을 자동화하면 관계 탐색 앞에 정보 추출과 정규화 문제가 추가된다. JSON 형식을 맞추는 단계와 원문에 근거한 관계를 얻는 단계를 나누어 살펴본다. 이름을 합치거나 관계를 추가할 때 판단 근거를 다시 확인할 수 있도록 원문과의 연결을 유지한다.
1
2
3
4
5
6
7
8
9
10
11
| text chunk
↓
LLM entity/relation extraction
↓
JSON / Pydantic schema
↓
NetworkX graph
↓
alias resolve
↓
KG query
|
2. Naive RAG 한계 복습
1
2
3
4
5
6
| def naive_rag(question, vs, k=3):
chunks = retrieve(question, vs, k=k)
ctx = "\n\n".join(f"[{i+1}] {c}" for i, c in enumerate(chunks))
sys = SystemMessage(content="주어진 컨텍스트만으로 한국어로 답하세요")
user = HumanMessage(content=f"컨텍스트:\n{ctx}\n\n질문 : {question}")
return llm.invoke([sys, user]).content
|
1
2
3
4
5
6
| multi-hop 질문:
모리카이 센의 멘토가 쓴 논문은 무엇인가요?
필요 단서:
모리카이 센 <- mentor_of - 탈라미르 헤쉬
탈라미르 헤쉬 -- wrote --> 공명 프레임에 관하여
|
3. 수동 KG와 그래프 탐색
1
2
3
4
5
6
7
8
9
10
11
| G_demo = nx.DiGraph()
G_demo.add_node("모리카이 센", type="Person")
G_demo.add_node("콰빅스 연구소", type="Organization")
G_demo.add_node("브렌술 아카데미", type="Organization")
G_demo.add_node("탈라미르 헤쉬", type="Person")
G_demo.add_node("벨트란 사고 엔진", type="Product")
G_demo.add_edge("모리카이 센", "콰빅스 연구소", relation="founded", year=1987)
G_demo.add_edge("모리카이 센", "브렌술 아카데미", relation="postdoc_at")
G_demo.add_edge("탈라미르 헤쉬", "모리카이 센", relation="mentor_of")
G_demo.add_edge("콰빅스 연구소", "벨트란 사고 엔진", relation="develops")
|
4. LLM JSON 추출 프롬프트
LLM에게 엔티티와 관계를 정해진 JSON schema로 추출하게 한다.
1
2
3
4
5
6
7
8
9
10
11
| prompt_skel = """You are an information extraction system.
From the text, extract entities and relations as JSON.
Schema:
{
"entities": [{"name": "...", "type": "Person|Organisation|Product|Concept|Publication"}],
"relations": [{"source": "...", "target": "...", "label": "..."}]
}
Output ONLY valid JSON. No other text.
"""
|
논문명, 프로젝트명, 제품명처럼 따옴표 안 제목이 누락되기 쉬우므로 프롬프트에서 강조한다.
JSON 파싱에 성공해도 엔티티나 관계가 원문에 존재한다고 보장되지는 않는다. 추출 결과를 원문 구절과 대조하고, 필드 누락과 내용 오류를 구분해 어떤 실패에 재추출이 필요한지 판단한다.
LLM 출력에 코드블록이 섞이면 제거하고 JSON으로 파싱한다.
1
2
3
4
5
6
7
8
9
10
| import json
import re
def strip_code_fence(text):
text = text.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text)
return text
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def extract_graph(text):
sys = SystemMessage(content=(
"You are an information extraction system. "
"From the text, extract entities and relations as JSON. "
"Always extract titles in quotation marks as Concept/Publication entities."
))
user = HumanMessage(content=(
'Schema:\n'
'{"entities": [{"name": "...", "type": "Person|Organisation|Product|Concept|Publication"}],\n'
'"relations": [{"source": "...", "target": "...", "label": "..."}]}\n\n'
f'Text:\n"""{text}"""'
))
raw = llm.invoke([sys, user]).content
try:
return json.loads(strip_code_fence(raw))
except json.JSONDecodeError:
return {"entities": [], "relations": []}
|
구조화 출력을 쓰면 JSON 파싱 실패를 줄일 수 있다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| from pydantic import BaseModel
from typing import Literal
class Entity(BaseModel):
name: str
type: Literal["Person", "Organisation", "Product", "Concept", "Publication"]
class Relation(BaseModel):
source: str
target: str
label: str
class ExtractedKG(BaseModel):
entities: list[Entity]
relations: list[Relation]
|
1
2
3
4
5
| structured_llm = llm.with_structured_output(ExtractedKG)
demo_kg = structured_llm.invoke(
f"다음 텍스트에서 엔티티/관계를 추출하세요:\n{para1}"
)
|
7. JSON을 NetworkX 그래프로 변환
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def json_to_graph(data) -> nx.DiGraph:
G = nx.DiGraph()
for e in data.get("entities", []):
G.add_node(e["name"], type=e.get("type", "unknown"))
for r in data.get("relations", []):
if r["source"] not in G:
G.add_node(r["source"], type="unknown")
if r["target"] not in G:
G.add_node(r["target"], type="unknown")
G.add_edge(r["source"], r["target"], relation=r.get("label", ""))
return G
|
8. alias 정규화
별칭 정규화는 같은 대상을 합치는 작업이지만 동명이인이나 비슷한 조직명을 잘못 합칠 가능성도 있다. 이름만으로 확정하기 어려운 경우에는 원문 문맥과 식별 정보를 함께 확인하고, 병합 전 이름을 추적할 수 있게 남긴다.
센, 모리카이, 모리카이 센처럼 같은 엔티티가 여러 이름으로 나올 수 있다.
1
2
3
4
5
6
7
8
| def find_aliases(nodes):
aliases = {}
for short in nodes:
for long in nodes:
if short != long and len(short) < len(long) and short in long:
aliases[short] = long
break
return aliases
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| def resolve_entities(G: nx.DiGraph) -> nx.DiGraph:
nodes = list(G.nodes)
aliases = find_aliases(nodes)
def resolve(n):
seen = set()
while n in aliases and n not in seen:
seen.add(n)
n = aliases[n]
return n
G2 = nx.DiGraph()
for n in nodes:
c = resolve(n)
if c not in G2:
G2.add_node(c, **G.nodes[n])
for u, v, d in G.edges(data=True):
cu, cv = resolve(u), resolve(v)
if cu == cv:
continue
if G2.has_edge(cu, cv):
continue
G2.add_edge(cu, cv, **d)
return G2
|
9. 확장 KG 구성
수동으로 추가 관계를 넣어 multi-hop 질의가 가능하게 만든다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| G_full = G_demo.copy()
G_full.add_node("눌마리 연구 위원회", type="Organisation")
G_full.add_edge("눌마리 연구 위원회", "콰빅스 연구소", relation="funds")
G_full.add_node("공명 프레임에 관하여", type="Publication")
G_full.add_edge("탈라미르 헤쉬", "공명 프레임에 관하여", relation="wrote")
G_full.add_node("타즈렌 연구소", type="Organisation")
G_full.add_node("에빈 브로스", type="Person")
G_full.add_node("할렌 프로토콜", type="Publication")
G_full.add_edge("에빈 브로스", "타즈렌 연구소", relation="works_at")
G_full.add_edge("에빈 브로스", "할렌 프로토콜", relation="wrote")
G_full.add_edge("눌마리 연구 위원회", "할렌 프로토콜", relation="adopts")
|
10. answer_via_kg
출발 엔티티와 키워드가 들어간 목표 노드 사이의 경로를 찾는다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| def answer_via_kg(G, src, tgt_keyword, max_hops=4):
if src not in G:
return f"src '{src}' not in graph"
candidates = [n for n in G.nodes if tgt_keyword in n]
if not candidates:
return f"keyword '{tgt_keyword}' 포함한 노드 없음"
Gu = G.to_undirected()
best = None
for tgt in candidates:
try:
paths = list(nx.all_simple_paths(Gu, src, tgt, cutoff=max_hops))
if paths:
shortest = min(paths, key=len)
if best is None or len(shortest) < len(best):
best = shortest
except nx.NodeNotFound:
continue
if best is None:
return "경로 없음"
return best[-1]
|
11. 관계 설명 함수
두 엔티티 사이의 최단 경로를 관계 라벨과 함께 설명한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| def explain_relation(G, e1, e2, max_hops=4) -> str:
Gu = G.to_undirected()
if e1 not in Gu or e2 not in Gu:
missing = e1 if e1 not in Gu else e2
return f"'{missing}'을 그래프에서 못찾음"
try:
path = nx.shortest_path(Gu, e1, e2)
except nx.NetworkXNoPath:
return f"'{e1}' 과 '{e2}' 사이에 경로가 없음"
if len(path) > max_hops + 1:
return f"경로가 너무 김 ({len(path)-1}hops)"
parts = [path[0]]
for i in range(len(path) - 1):
u, v = path[i], path[i+1]
rel = G[u][v].get("relation") if G.has_edge(u, v) else G[v][u].get("relation", "?")
arrow = "->" if G.has_edge(u, v) else "<-"
parts.append(f"{arrow}[{rel}]{arrow}")
parts.append(v)
return " ".join(parts)
|
12. 핵심 개념 총정리
| 개념 | 역할 |
|---|
| extract_graph | 텍스트에서 엔티티/관계를 JSON으로 추출 |
| strip_code_fence | LLM 코드블록 제거 |
| ExtractedKG | Pydantic 구조화 추출 스키마 |
| json_to_graph | JSON 결과를 NetworkX 그래프로 변환 |
| resolve_entities | 짧은 alias를 canonical name으로 병합 |
| answer_via_kg | 그래프 경로로 답 후보 찾기 |
| explain_relation | 두 엔티티의 관계 경로 설명 |
13. 자주 나오는 실수/주의사항
실수 1: LLM JSON을 바로 json.loads
1
| json.loads(strip_code_fence(raw))
|
코드블록, 설명 문장, trailing text가 섞일 수 있다.
실수 2: alias 병합 기준 과신
1
| short in long 규칙은 간단하지만 동명이인/부분 문자열 오탐이 생길 수 있다.
|
실수 3: 방향 그래프를 무조건 undirected로 사용
경로 탐색에는 편하지만 관계 방향 해석은 별도로 확인해야 한다.
실수 4: 관계 라벨 누락
1
| G.add_edge(source, target, relation=label)
|
14. 그래프의 연결을 근거와 함께 검토하기
그래프 품질은 노드와 엣지가 많이 생성됐는지만으로 판단하지 않는다. 잘못된 관계가 하나 추가되면 이후 경로 탐색에 영향을 줄 수 있으므로, 질문에 사용되는 연결부터 원문과 대조한다.
정규화 규칙을 바꿀 때는 병합된 이름과 달라진 경로를 함께 확인한다. 답변에 사용한 관계를 원문으로 되돌릴 수 있어야 추출 오류와 탐색 오류를 나누어 수정할 수 있다.