10장: 실전 프로젝트 — Knowledge Graph + AI 시스템

from dataclasses import dataclass, field
 
@dataclass
class Neo4jConfig:
    uri: str = "bolt://localhost:7687"
    user: str = "neo4j"
    password: str = "password"
    database: str = "neo4j"
 
@dataclass
class LLMConfig:
    extraction_model: str = "claude-sonnet-4-20250514"
    generation_model: str = "claude-sonnet-4-20250514"
    embedding_model: str = "text-embedding-3-small"
    embedding_dimensions: int = 1536
 
@dataclass
class PipelineConfig:
    chunk_size: int = 2000
    chunk_overlap: int = 200
    similarity_threshold: float = 0.85
    vector_top_k: int = 5
    graph_depth: int = 2
 
@dataclass
class AppConfig:
    neo4j: Neo4jConfig = field(default_factory=Neo4jConfig)
    llm: LLMConfig = field(default_factory=LLMConfig)
    pipeline: PipelineConfig = field(default_factory=PipelineConfig)

# 기술 문서 도메인의 엔티티/관계 스키마
 
ENTITY_TYPES = {
    "Technology": {
        "description": "프로그래밍 언어, 프레임워크, 라이브러리, 도구, 데이터베이스",
        "properties": ["name", "category", "version", "description"]
    },
    "Concept": {
        "description": "기술 개념, 알고리즘, 디자인 패턴, 방법론",
        "properties": ["name", "description", "domain"]
    },
    "Organization": {
        "description": "회사, 오픈소스 프로젝트, 연구 기관",
        "properties": ["name", "type", "url"]
    },
    "UseCase": {
        "description": "기술의 적용 분야, 활용 사례",
        "properties": ["name", "domain", "description"]
    }
}
 
RELATION_TYPES = {
    "DEPENDS_ON": {"source": "Technology", "target": "Technology"},
    "IMPLEMENTS": {"source": "Technology", "target": "Concept"},
    "DEVELOPED_BY": {"source": "Technology", "target": "Organization"},
    "USED_FOR": {"source": "Technology", "target": "UseCase"},
    "RELATED_TO": {"source": "Concept", "target": "Concept"},
    "ALTERNATIVE_TO": {"source": "Technology", "target": "Technology"},
    "EXTENDS": {"source": "Technology", "target": "Technology"},
}

from neo4j import GraphDatabase
 
class GraphInitializer:
    """Neo4j 스키마를 초기화합니다."""
 
    def __init__(self, config):
        self.driver = GraphDatabase.driver(
            config.neo4j.uri,
            auth=(config.neo4j.user, config.neo4j.password)
        )
 
    def initialize_schema(self) -> None:
        """제약 조건, 인덱스, 벡터 인덱스를 생성합니다."""
        constraints = [
            "CREATE CONSTRAINT IF NOT EXISTS FOR (t:Technology) REQUIRE t.name IS UNIQUE",
            "CREATE CONSTRAINT IF NOT EXISTS FOR (c:Concept) REQUIRE c.name IS UNIQUE",
            "CREATE CONSTRAINT IF NOT EXISTS FOR (o:Organization) REQUIRE o.name IS UNIQUE",
            "CREATE CONSTRAINT IF NOT EXISTS FOR (u:UseCase) REQUIRE u.name IS UNIQUE",
            "CREATE CONSTRAINT IF NOT EXISTS FOR (d:Document) REQUIRE d.externalId IS UNIQUE",
        ]
 
        indexes = [
            # 전문 검색 인덱스
            """CREATE FULLTEXT INDEX IF NOT EXISTS ft_document_search
               FOR (d:Document) ON EACH [d.title, d.content]""",
            # 벡터 인덱스
            """CREATE VECTOR INDEX IF NOT EXISTS vec_document_embedding
               FOR (d:Document) ON (d.embedding)
               OPTIONS {indexConfig: {
                 `vector.dimensions`: 1536,
                 `vector.similarity_function`: 'cosine'
               }}""",
        ]
 
        for stmt in constraints + indexes:
            self.driver.execute_query(stmt)
 
        print("스키마 초기화 완료")

from anthropic import Anthropic
from src.schema import ENTITY_TYPES, RELATION_TYPES
import json
 
class EntityExtractor:
    """LLM을 사용하여 텍스트에서 엔티티와 관계를 추출합니다."""
 
    def __init__(self, config):
        self.client = Anthropic()
        self.model = config.llm.extraction_model
        self.system_prompt = self._build_system_prompt()
 
    def _build_system_prompt(self) -> str:
        """스키마 정보를 포함한 시스템 프롬프트를 생성합니다."""
        entity_desc = "\n".join(
            f"- {name}: {info['description']}"
            for name, info in ENTITY_TYPES.items()
        )
        relation_desc = "\n".join(
            f"- {name}: ({info['source']}) -> ({info['target']})"
            for name, info in RELATION_TYPES.items()
        )
 
        json_format = '{"entities": [{"id": "소문자-하이픈", "type": "타입", "name": "이름", "properties": {}}], "relationships": [{"source": "소스id", "target": "타겟id", "type": "관계타입", "properties": {}}]}'
        return f"""텍스트에서 엔티티와 관계를 추출하세요.
 
## 엔티티 타입
{entity_desc}
 
## 관계 타입
{relation_desc}
 
## 출력 형식
JSON으로 응답하세요:
{json_format}
 
규칙: 텍스트에 명시된 정보만 추출하세요. 추론하지 마세요."""
 
    def extract(self, text: str) -> dict:
        """텍스트에서 엔티티와 관계를 추출합니다."""
        response = self.client.messages.create(
            model=self.model,
            max_tokens=4096,
            system=self.system_prompt,
            messages=[{"role": "user", "content": f"텍스트:\n\n{text}"}]
        )
        content = response.content[0].text
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0]
        return json.loads(content.strip())

from pathlib import Path
from src.config import AppConfig
from src.chunker import TextChunker
from src.extractor import EntityExtractor
from src.resolver import EntityResolver
from src.loader import GraphInitializer, KnowledgeGraphLoader
 
class TechDocsKGPipeline:
    """기술 문서에서 Knowledge Graph를 구축하는 전체 파이프라인입니다."""
 
    def __init__(self, config: AppConfig):
        self.config = config
        self.chunker = TextChunker(config.pipeline.chunk_size, config.pipeline.chunk_overlap)
        self.extractor = EntityExtractor(config)
        self.resolver = EntityResolver(config.pipeline.similarity_threshold)
        self.loader = KnowledgeGraphLoader(config)
        self.initializer = GraphInitializer(config)
 
    def setup(self) -> None:
        """스키마 초기화를 수행합니다."""
        self.initializer.initialize_schema()
 
    def ingest_document(self, doc_path: Path) -> dict:
        """단일 문서를 처리하여 KG에 적재합니다."""
        text = doc_path.read_text(encoding="utf-8")
        doc_id = doc_path.stem
 
        # 1. 청킹
        chunks = self.chunker.chunk(text)
        print(f"[{doc_id}] {len(chunks)}개 청크 생성")
 
        all_entities = []
        all_relationships = []
 
        # 2. 각 청크에서 추출
        for i, chunk in enumerate(chunks):
            try:
                result = self.extractor.extract(chunk)
                all_entities.extend(result.get("entities", []))
                all_relationships.extend(result.get("relationships", []))
            except Exception as err:
                print(f"  청크 {i + 1} 추출 실패: {err}")
 
        # 3. 엔티티 해소
        resolved = self.resolver.resolve_batch(all_entities)
        unique_count = len(set(e["name"] for e in resolved))
        print(f"  엔티티: {len(all_entities)} -> {unique_count} (해소 후)")
 
        # 4. 적재
        self.loader.load_document(doc_id, text, resolved, all_relationships)
 
        # 5. 임베딩 생성 및 저장
        self.loader.create_embeddings(doc_id, chunks)
 
        return {
            "doc_id": doc_id,
            "chunks": len(chunks),
            "entities": unique_count,
            "relationships": len(all_relationships)
        }
 
    def ingest_directory(self, dir_path: Path) -> list[dict]:
        """디렉토리의 모든 문서를 처리합니다."""
        results = []
        files = sorted(dir_path.glob("*.md")) + sorted(dir_path.glob("*.txt"))
 
        for f in files:
            print(f"\n처리 중: {f.name}")
            result = self.ingest_document(f)
            results.append(result)
 
        # 통계 출력
        total_entities = sum(r["entities"] for r in results)
        total_rels = sum(r["relationships"] for r in results)
        print(f"\n총 {len(results)}개 문서, {total_entities}개 엔티티, {total_rels}개 관계")
 
        return results

from neo4j import GraphDatabase
from openai import OpenAI
from dataclasses import dataclass
 
@dataclass
class RetrievalResult:
    content: str
    score: float
    source: str
    metadata: dict
 
class HybridRetriever:
    """벡터 + 그래프 + 키워드 하이브리드 검색을 수행합니다."""
 
    def __init__(self, config):
        self.driver = GraphDatabase.driver(
            config.neo4j.uri,
            auth=(config.neo4j.user, config.neo4j.password)
        )
        self.openai = OpenAI()
        self.config = config
 
    def search(self, query: str, entities: list[str] = None) -> list[RetrievalResult]:
        """하이브리드 검색을 수행합니다."""
        results = {}
 
        # 벡터 검색
        vector_results = self._vector_search(query)
        for r in vector_results:
            key = r.metadata.get("title", "")
            results[key] = r
            results[key].score *= 0.4
 
        # 그래프 순회
        if entities:
            graph_results = self._graph_search(entities)
            for r in graph_results:
                key = r.metadata.get("title", "")
                if key in results:
                    results[key].score += r.score * 0.4
                else:
                    r.score *= 0.4
                    results[key] = r
 
        # 키워드 검색
        keyword_results = self._keyword_search(query)
        for r in keyword_results:
            key = r.metadata.get("title", "")
            if key in results:
                results[key].score += r.score * 0.2
            else:
                r.score *= 0.2
                results[key] = r
 
        return sorted(results.values(), key=lambda x: x.score, reverse=True)
 
    def _vector_search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
        """벡터 유사도 검색을 수행합니다."""
        embedding = self.openai.embeddings.create(
            model=self.config.llm.embedding_model,
            input=query
        ).data[0].embedding
 
        records, _, _ = self.driver.execute_query("""
            CALL db.index.vector.queryNodes("vec_document_embedding", $topK, $embedding)
            YIELD node, score
            OPTIONAL MATCH (node)-[:COVERS]->(tech:Technology)
            RETURN node.title AS title, node.content AS content, score,
                   collect(DISTINCT tech.name) AS technologies
        """, topK=top_k, embedding=embedding)
 
        return [
            RetrievalResult(
                content=r["content"] or "", score=r["score"],
                source="vector",
                metadata={"title": r["title"], "technologies": r["technologies"]}
            ) for r in records
        ]
 
    def _graph_search(self, entities: list[str]) -> list[RetrievalResult]:
        """엔티티 기반 그래프 순회를 수행합니다."""
        records, _, _ = self.driver.execute_query("""
            UNWIND $entities AS name
            MATCH (e {name: name})
            MATCH (e)-[*1..2]-(related)
            WHERE related:Document OR related:Technology OR related:Concept
            WITH DISTINCT related, e.name AS startEntity
            OPTIONAL MATCH (related)-[:COVERS]->(tech:Technology)
            RETURN related.name AS title,
                   related.content AS content,
                   0.8 AS score,
                   collect(DISTINCT tech.name) AS technologies
            LIMIT 10
        """, entities=entities)
 
        return [
            RetrievalResult(
                content=r["content"] or r["title"] or "",
                score=r["score"], source="graph",
                metadata={"title": r["title"], "technologies": r.get("technologies", [])}
            ) for r in records
        ]
 
    def _keyword_search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
        """전문 검색을 수행합니다."""
        records, _, _ = self.driver.execute_query("""
            CALL db.index.fulltext.queryNodes("ft_document_search", $query)
            YIELD node, score
            RETURN node.title AS title, node.content AS content, score
            LIMIT $topK
        """, query=query, topK=top_k)
 
        return [
            RetrievalResult(
                content=r["content"] or "", score=r["score"],
                source="keyword", metadata={"title": r["title"]}
            ) for r in records
        ]

from anthropic import Anthropic
from src.config import AppConfig
from src.retriever import HybridRetriever
 
class TechDocsQA:
    """기술 문서 QA 시스템의 메인 인터페이스입니다."""
 
    def __init__(self, config: AppConfig):
        self.retriever = HybridRetriever(config)
        self.llm = Anthropic()
        self.config = config
 
    def ask(self, question: str) -> dict:
        """자연어 질문에 대한 답변을 생성합니다."""
 
        # 1. 질문에서 엔티티 추출
        entities = self._extract_query_entities(question)
        print(f"감지된 엔티티: {entities}")
 
        # 2. 하이브리드 검색
        results = self.retriever.search(question, entities)
        print(f"검색 결과: {len(results)}건")
 
        # 3. 컨텍스트 구성
        context_parts = []
        for r in results[:5]:
            source_tag = f"[{r.source}]"
            techs = ", ".join(r.metadata.get("technologies", []))
            header = f"{source_tag} {r.metadata.get('title', 'N/A')}"
            if techs:
                header += f" (관련 기술: {techs})"
            context_parts.append(f"{header}\n{r.content}")
 
        context = "\n\n---\n\n".join(context_parts)
 
        # 4. LLM 응답 생성
        response = self.llm.messages.create(
            model=self.config.llm.generation_model,
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""다음 기술 문서 컨텍스트를 기반으로 질문에 정확하게 답변하세요.
컨텍스트에 없는 정보는 "해당 정보를 찾을 수 없습니다"라고 답하세요.
 
컨텍스트:
{context}
 
질문: {question}
 
답변:"""
            }]
        )
 
        return {
            "question": question,
            "answer": response.content[0].text,
            "sources": [
                {"title": r.metadata.get("title"), "source": r.source, "score": r.score}
                for r in results[:5]
            ],
            "entities_detected": entities
        }
 
    def _extract_query_entities(self, question: str) -> list[str]:
        """질문에서 핵심 엔티티를 추출합니다."""
        response = self.llm.messages.create(
            model="claude-haiku-4-20250414",
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": f"""다음 질문에서 기술 엔티티(기술명, 도구명, 개념명)만 추출하세요.
JSON 배열로 응답: ["엔티티1", "엔티티2"]
 
질문: {question}"""
            }]
        )
        import json
        try:
            return json.loads(response.content[0].text)
        except (json.JSONDecodeError, IndexError):
            return []
 
# CLI 인터페이스
def main():
    config = AppConfig()
    qa = TechDocsQA(config)
 
    print("기술 문서 QA 시스템 (종료: quit)")
    print("-" * 50)
 
    while True:
        question = input("\n질문: ").strip()
        if question.lower() in ("quit", "exit", "q"):
            break
        if not question:
            continue
 
        result = qa.ask(question)
        print(f"\n답변: {result['answer']}")
        print(f"\n출처:")
        for s in result["sources"]:
            print(f"  - [{s['source']}] {s['title']} (점수: {s['score']:.3f})")
 
if __name__ == "__main__":
    main()

from dataclasses import dataclass
 
@dataclass
class EvalResult:
    question: str
    expected: str
    vector_answer: str
    graphrag_answer: str
    vector_score: float
    graphrag_score: float
 
class RAGBenchmark:
    """벡터 전용 RAG와 GraphRAG의 성능을 비교합니다."""
 
    def __init__(self, config: AppConfig):
        self.vector_retriever = VectorOnlyRetriever(config)
        self.hybrid_retriever = HybridRetriever(config)
        self.llm = Anthropic()
 
    def evaluate(self, test_cases: list[dict]) -> list[EvalResult]:
        """테스트 케이스를 실행하고 결과를 비교합니다."""
        results = []
 
        for case in test_cases:
            question = case["question"]
            expected = case["expected_answer"]
 
            # 벡터 전용 RAG
            vector_results = self.vector_retriever.search(question)
            vector_answer = self._generate_answer(question, vector_results)
 
            # GraphRAG (하이브리드)
            entities = self._extract_entities(question)
            hybrid_results = self.hybrid_retriever.search(question, entities)
            graphrag_answer = self._generate_answer(question, hybrid_results)
 
            # LLM 기반 평가
            vector_score = self._evaluate_answer(question, expected, vector_answer)
            graphrag_score = self._evaluate_answer(question, expected, graphrag_answer)
 
            results.append(EvalResult(
                question=question,
                expected=expected,
                vector_answer=vector_answer,
                graphrag_answer=graphrag_answer,
                vector_score=vector_score,
                graphrag_score=graphrag_score
            ))
 
        return results
 
    def _evaluate_answer(self, question: str, expected: str, actual: str) -> float:
        """LLM을 사용하여 답변 품질을 0~1 점수로 평가합니다."""
        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=64,
            messages=[{
                "role": "user",
                "content": f"""답변의 정확성을 0.0~1.0 점수로 평가하세요. 숫자만 응답하세요.
 
질문: {question}
기대 답변: {expected}
실제 답변: {actual}
 
점수:"""
            }]
        )
        try:
            return float(response.content[0].text.strip())
        except ValueError:
            return 0.0
 
    def print_report(self, results: list[EvalResult]) -> None:
        """평가 결과 보고서를 출력합니다."""
        vector_avg = sum(r.vector_score for r in results) / len(results)
        graphrag_avg = sum(r.graphrag_score for r in results) / len(results)
        improvement = ((graphrag_avg - vector_avg) / vector_avg) * 100
 
        print("\n" + "=" * 60)
        print("성능 비교 보고서")
        print("=" * 60)
        print(f"테스트 케이스 수: {len(results)}")
        print(f"벡터 전용 RAG 평균 점수: {vector_avg:.3f}")
        print(f"GraphRAG 평균 점수:      {graphrag_avg:.3f}")
        print(f"개선율:                   {improvement:+.1f}%")
        print("=" * 60)
 
        # 질문 유형별 분석
        print("\n질문별 상세:")
        for r in results:
            delta = r.graphrag_score - r.vector_score
            indicator = "+" if delta > 0 else ""
            print(f"  Q: {r.question}")
            print(f"    Vector: {r.vector_score:.2f} | GraphRAG: {r.graphrag_score:.2f} ({indicator}{delta:.2f})")

TEST_CASES = [
    # 유형 1: 단순 사실 질문 (벡터와 큰 차이 없을 것)
    {
        "question": "Neo4j에서 벡터 인덱스를 생성하는 Cypher 구문은?",
        "expected_answer": "CREATE VECTOR INDEX 구문을 사용하며, 차원과 유사도 함수를 지정합니다.",
        "type": "factual"
    },
    # 유형 2: 관계 질문 (GraphRAG가 우수할 것)
    {
        "question": "GraphRAG가 의존하는 기술과 그 기술들의 대안은?",
        "expected_answer": "GraphRAG는 Neo4j, Python, LLM에 의존하며, Neo4j의 대안으로 Neptune, TigerGraph 등이 있습니다.",
        "type": "relational"
    },
    # 유형 3: 다중 홉 질문 (GraphRAG가 크게 우수할 것)
    {
        "question": "Python으로 개발된 프레임워크 중 그래프 데이터베이스를 사용하는 것은?",
        "expected_answer": "neo4j-graphrag-python, LangChain (Neo4j 통합), Graphiti 등이 있습니다.",
        "type": "multi_hop"
    },
    # 유형 4: 전역 질문 (GraphRAG만 답변 가능)
    {
        "question": "기술 문서 전체에서 가장 핵심적인 기술은 무엇이며, 기술 간 주요 커뮤니티는?",
        "expected_answer": "Neo4j, Python이 핵심 기술이며, 그래프 DB, RAG 프레임워크, 임베딩 등의 커뮤니티가 존재합니다.",
        "type": "global"
    },
]