포스트

2026-06-11 강의 정리

LangChain 활용 그래프 데이터베이스 검색

게시
By kimseokryu
12 분읽는 시간
2026-06-11 강의 정리

목차

  1. 문제와 접근
  2. corpus chunking
  3. chunk별 엔티티/관계 추출
  4. extract_all_chunks
  5. merge_chunk_results
  6. 엔티티 정규화
  7. edge/node weight
  8. 그래프 시각화
  9. Leiden 커뮤니티 탐지
  10. 커뮤니티 요약
  11. index 저장/로드
  12. local_search
  13. global_search
  14. 핵심 개념 총정리
  15. 자주 나오는 실수/주의사항
  16. 인덱스의 생성 과정까지 추적하기

1. 문제와 접근

Mini GraphRAG를 직접 구성하면 추출·병합·커뮤니티·검색 사이의 데이터 흐름이 드러난다. 각 단계가 어떤 정보를 추가하고 어떤 정보를 줄이는지 확인하면서 결과를 저장하고 다시 읽는 구조까지 연결한다. 그래프 요약과 검색 답변의 근거가 원본 청크로 이어지는지를 기준으로 본다.

1
2
3
4
5
6
7
8
9
10
11
12
13

corpus
  ↓ chunking
chunk별 entity/relation 추출
  ↓
graph merge
  ↓
entity resolve + weight
  ↓
community detection
  ↓
community summary
  ↓
local/global search

2. corpus chunking

1
2
3
4
5
6
7
8
9
10
11
12
13

import json
import re
import pickle
from pathlib import Path
from collections import defaultdict

import networkx as nx
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_text_splitters import RecursiveCharacterTextSplitter

llm = ChatOpenAI(model="gpt-4o-mini")
corpus = Path("./data/council.txt").read_text(encoding="utf-8")

1
2
3
4
5
6
7

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " ", ""],
)

chunks = splitter.split_text(corpus)

3. chunk별 엔티티/관계 추출

1
2
3
4
5
6
7
8
9

EXTRACTION_PROMPT = """You are an information extraction system.
Extract entities and relations from the text as JSON.

Schema:
{
  "entities":  [{"name": "...", "type": "Person|Organisation|Product|Concept"}],
  "relations": [{"source": "...", "target": "...", "label": "..."}]
}
Output ONLY valid JSON. No commentary."""

1
2
3
4
5
6

def strip_fence(t):
    t = t.strip()
    if t.startswith("```"):
        t = re.sub(r"^```(?:json)?\s*", "", t)
        t = re.sub(r"\s*```$", "", t)
    return t

1
2
3
4
5
6
7
8
9
10

def extract_one_chunk(chunk) -> dict:
    msgs = [
        SystemMessage(content=EXTRACTION_PROMPT),
        HumanMessage(content=f'Text:\n"""{chunk}"""'),
    ]
    raw = llm.invoke(msgs).content
    try:
        return json.loads(strip_fence(raw))
    except json.JSONDecodeError:
        return {"entities": [], "relations": []}

4. extract_all_chunks

모든 chunk를 순회하면서 추출 결과에 chunk index를 붙인다.

1
2
3
4
5
6
7
8

def extract_all_chunks(chunks) -> list[dict]:
    results = []
    for i, c in enumerate(chunks):
        r = extract_one_chunk(c)
        r["chunk_index"] = i
        results.append(r)
        print(f"chunk {i}: {len(r['entities'])} entities, {len(r['relations'])} relations")
    return results

5. merge_chunk_results

여러 청크에서 같은 관계를 얻었을 때 단순히 하나로 합치면 출처를 잃을 수 있다. 관계 설명과 등장 횟수를 어떻게 합칠지 정하고, 원문 청크 ID를 유지해 병합 결과를 다시 확인할 수 있는 형태를 검토한다.

chunk별 추출 결과를 하나의 방향 그래프로 합친다.

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
27
28
29
30

def merge_chunk_results(results: list[dict]) -> nx.DiGraph:
    G = nx.DiGraph()

    for r in results:
        ci = r["chunk_index"]

        for e in r["entities"]:
            name = e["name"]
            if name in G:
                G.nodes[name]["source_chunks"].append(ci)
            else:
                G.add_node(name, type=e.get("type", "unknown"), source_chunks=[ci])

        for rel in r["relations"]:
            s, t = rel["source"], rel["target"]
            label = rel.get("label", "")

            if s not in G:
                G.add_node(s, type="unknown", source_chunks=[ci])
            if t not in G:
                G.add_node(t, type="unknown", source_chunks=[ci])

            if G.has_edge(s, t):
                if label and label not in G[s][t]["relations"]:
                    G[s][t]["relations"].append(label)
                G[s][t]["source_chunks"].append(ci)
            else:
                G.add_edge(s, t, relations=[label] if label else [], source_chunks=[ci])

    return G

6. 엔티티 정규화

짧은 alias를 긴 이름으로 합친다.

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
27
28
29
30
31
32

def resolve_entities(G: nx.DiGraph) -> nx.DiGraph:
    nodes = list(G.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

    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

7. edge/node weight

같은 엔티티나 관계가 여러 chunk에서 나오면 weight를 높인다.

1
2
3
4
5
6

def add_edge_weights(G: nx.DiGraph) -> nx.DiGraph:
    for u, v, d in G.edges(data=True):
        d["weight"] = len(d.get("source_chunks", []))
    for n, d in G.nodes(data=True):
        d["weight"] = len(d.get("source_chunks", []))
    return G

8. 그래프 시각화

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

def draw(G, node_color, title="", figsize=(12, 8)):
    pos = nx.spring_layout(G, seed=42, k=1.2)
    plt.figure(figsize=figsize)
    nx.draw(
        G,
        pos,
        node_color=node_color,
        with_labels=True,
        node_size=1600,
        font_size=8,
        arrowsize=10,
        edge_color="#999",
        alpha=0.85,
    )
    plt.title(title)
    plt.axis("off")
    plt.show()

9. Leiden 커뮤니티 탐지

그래프를 커뮤니티 단위로 나눈다.

1
2
3
4
5
6
7
8
9
10
11
12

from graspologic.partition import leiden


def detect_comunities(G):
    partitions = leiden(G.to_undirected())
    communities = defaultdict(list)
    for node, cid in partitions.items():
        communities[cid].append(node)
    return dict(communities)


comms = detect_comunities(G)

10. 커뮤니티 요약

커뮤니티 요약은 검색에 사용할 또 하나의 가공된 문서다. 요약에 포함된 주장과 원본 관계의 연결을 확인하고, 요약에서 빠진 세부 사항이 필요한 질문에는 어떤 근거를 추가로 전달할지 생각해야 한다.

커뮤니티 내부 엔티티와 관계를 LLM에게 요약시킨다.

1
2
3
4
5
6
7
8

SUMMARY_PROMPT = """아래는 지식 그래프의 한 커뮤니티에 속한 엔티티와 관계 목록입니다.
이 커뮤니티의 주제와 주요 사실을 4~6문장의 한국어 단락으로 요약해 주세요.

엔티티:
{entities}

관계:
{relations}"""

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

def summarize_community(G: nx.DiGraph, nodes: list[str]) -> str:
    nodes_set = set(nodes)
    ent_lines = [f"- {n} ({G.nodes[n].get('type', '?')})" for n in nodes]
    rel_lines = []

    for u, v, d in G.edges(data=True):
        if u in nodes_set and v in nodes_set:
            for rel in (d.get("relations") or [""]):
                rel_lines.append(f"- {u} --[{rel}]--> {v}")

    prompt = SUMMARY_PROMPT.format(
        entities="\n".join(ent_lines) or "(없음)",
        relations="\n".join(rel_lines) or "(없음)",
    )
    return llm.invoke([HumanMessage(content=prompt)]).content

11. index 저장/로드

1
2
3
4
5
6
7
8

INDEX_PATH = Path("mini_graphrag_index_0611.pkl")

index = {
    "graph": G,
    "communities": comms,
    "summaries": summaries,
    "chunks": chunks,
}

1
2
3
4
5
6
7
8

def save_index(index, path=INDEX_PATH):
    with open(path, "wb") as f:
        pickle.dump(index, f)


def load_index(path=INDEX_PATH):
    with open(path, "rb") as f:
        return pickle.load(f)

질문에 들어 있는 엔티티를 찾아 주변 hop과 원문 chunk를 모아 답변한다.

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
27
28

def local_search(question, index, hops):
    G, chunks = index["graph"], index["chunks"]
    matched = [n for n in G.nodes if n in question]
    if not matched:
        return "질문에 해당되는 entity 없음. Global search를 시도해 보세요"

    nodes = set(matched)
    for _ in range(hops):
        new = set()
        for n in nodes:
            new |= set(G.successors(n)) | set(G.predecessors(n))
        nodes |= new

    chunk_ids = set()
    for n in nodes:
        chunk_ids |= set(G.nodes[n].get("source_chunks", []))

    chunk_text = "\n\n".join(f"[chunk {i}] {chunks[i]}" for i in sorted(chunk_ids))

    edge_lines = []
    for u, v, d in G.edges(data=True):
        if u in nodes and v in nodes:
            for rel in (d.get("relations") or [""]):
                edge_lines.append(f"({u}) --[{rel}]--> ({v})")

    sys = SystemMessage(content="주어진 컨텍스트만 사용해서 한국어로 답하세요. 관계 정보로 다중 홉 추론도 가능합니다.")
    user = HumanMessage(content=f"질문: {question}\n\n관계:\n{chr(10).join(edge_lines)}\n\n청크:\n{chunk_text}")
    return llm.invoke([sys, user]).content

전체 주제 질문은 커뮤니티 요약을 모두 종합해서 답한다.

1
2
3
4
5
6
7
8
9
10

def global_search(question, index) -> str:
    all_summary_text = "\n\n".join(
        f"[C{cid}]\n{s}" for cid, s in sorted(index["summaries"].items())
    )
    sys = SystemMessage(content=(
        "당신은 지식 그래프의 커뮤니티 요약을 종합해 답하는 전문가입니다. "
        "모든 요약을 비교/종합해 한국어로 답하세요."
    ))
    user = HumanMessage(content=f"질문 : {question}\n\n== 커뮤니티 요약 ==\n{all_summary_text}")
    return llm.invoke([sys, user]).content

14. 핵심 개념 총정리

단계역할
chunking긴 코퍼스를 추출 단위로 분할
extract_one_chunkchunk에서 엔티티/관계 추출
merge_chunk_resultschunk별 KG를 하나로 병합
source_chunks엔티티/관계의 원문 근거 위치
resolve_entitiesalias 정규화
weight여러 chunk에서 반복된 정도
Leiden그래프 커뮤니티 탐지
community summaryglobal search용 요약 단위
local_search특정 엔티티 주변 검색
global_search전체 코퍼스 주제 종합

15. 자주 나오는 실수/주의사항

실수 1: source chunk를 저장하지 않음

1

G.add_node(name, type=..., source_chunks=[ci])

GraphRAG 답변에는 원문 근거가 필요하다.

실수 2: alias 정규화 후 edge 정보 손실

1

같은 edge가 합쳐질 때 relations/source_chunks를 병합해야 한다.

실수 3: local_search에서 엔티티 매칭 실패

1

matched = [n for n in G.nodes if n in question]

표현이 다르면 매칭이 안 되므로 실제 시스템에서는 entity linker가 필요하다.

실수 4: community 요약만으로 세부 질문 답변

1

세부 질문은 local search, 전체 경향 질문은 global search가 적합하다.

16. 인덱스의 생성 과정까지 추적하기

저장한 인덱스에는 입력 코퍼스와 추출·정규화 설정을 연결해 둔다. 같은 코드로 다시 실행해도 원문이나 프롬프트가 달라졌다면 결과를 같은 버전으로 취급하기 어렵다.

Local과 Global의 결과를 비교할 때는 답변 문장뿐 아니라 사용한 관계와 요약을 함께 확인한다. 누락이 검색 단계에서 발생했는지 앞선 커뮤니티 요약에서 생겼는지 구분하면 재인덱싱이 필요한 범위도 좁힐 수 있다.

모두의연구소, AI/LLM 서비스 개발 과정
AI LLM
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.

인기 태그

AI LLM