8장: 지식 그래프 쿼리와 추론

// 1~5 홉 이내의 모든 경로 탐색
MATCH path = (start:Technology {name: "GraphRAG"})
             -[:DEPENDS_ON*1..5]->
             (end:Technology)
RETURN path, length(path) AS depth
ORDER BY depth
 
// 최단 경로 탐색
MATCH path = shortestPath(
  (a:Technology {name: "GraphRAG"})-[:DEPENDS_ON|USES*]-(b:Technology {name: "Java"})
)
RETURN path, length(path) AS hops
 
// 모든 최단 경로 (여러 개 존재할 수 있음)
MATCH path = allShortestPaths(
  (a:Technology {name: "GraphRAG"})-[*]-(b:Technology {name: "Java"})
)
RETURN path, length(path) AS hops

// 기술과 관련 문서 조회 (문서가 없어도 기술은 반환)
MATCH (t:Technology)
WHERE t.category = "Database"
OPTIONAL MATCH (t)<-[:COVERS]-(d:Document)
OPTIONAL MATCH (t)-[:DEPENDS_ON]->(dep:Technology)
RETURN t.name,
       collect(DISTINCT d.title) AS documents,
       collect(DISTINCT dep.name) AS dependencies
 
// 존재하지 않는 관계 필터링
MATCH (t:Technology)
WHERE NOT EXISTS {
  MATCH (t)<-[:COVERS]-(d:Document)
}
RETURN t.name AS undocumented_tech

// 각 카테고리별 상위 3개 기술 조회
MATCH (cat:Category)
CALL (cat) {
  MATCH (t:Technology)-[:BELONGS_TO]->(cat)
  OPTIONAL MATCH (t)<-[:COVERS]-(d:Document)
  WITH t, count(d) AS docCount
  ORDER BY docCount DESC
  LIMIT 3
  RETURN t.name AS techName, docCount
}
RETURN cat.name AS category, techName, docCount
 
// UNION을 서브쿼리로 대체
MATCH (p:Person {name: "김개발"})
CALL (p) {
  MATCH (p)-[:AUTHORED]->(d:Document)
  RETURN d.title AS item, "authored" AS type
  UNION
  MATCH (p)-[:EXPERT_IN]->(t:Technology)
  RETURN t.name AS item, "expertise" AS type
}
RETURN item, type

// 날짜 기반 시간대 분석
MATCH (d:Document)
WITH d, apoc.date.format(d.createdAt.epochMillis, 'ms', 'yyyy-MM') AS yearMonth
RETURN yearMonth, count(d) AS docCount
ORDER BY yearMonth DESC
 
// JSON 데이터 임포트
CALL apoc.load.json("https://api.example.com/technologies")
YIELD value
MERGE (t:Technology {name: value.name})
SET t.category = value.category, t.version = value.version
 
// 동적 관계 생성 (관계 타입이 변수일 때)
WITH "DEPENDS_ON" AS relType
MATCH (a:Technology {name: "GraphRAG"})
MATCH (b:Technology {name: "Neo4j"})
CALL apoc.create.relationship(a, relType, {since: 2024}, b) YIELD rel
RETURN rel
 
// 가상 그래프 (결과를 그래프로 시각화)
MATCH path = (t:Technology)-[:DEPENDS_ON*1..3]->(dep:Technology)
WHERE t.name = "GraphRAG"
RETURN path

// 프로젝션 생성
CALL gds.graph.project(
  'dependencyGraph',
  'Technology',
  'DEPENDS_ON'
)
 
// PageRank 실행 및 결과 저장
CALL gds.pageRank.write('dependencyGraph', {
  writeProperty: 'pageRank',
  maxIterations: 20,
  dampingFactor: 0.85
})
 
// 결과 조회: 핵심 기술 상위 10개
MATCH (t:Technology)
WHERE t.pageRank IS NOT NULL
RETURN t.name, round(t.pageRank, 4) AS importance
ORDER BY importance DESC
LIMIT 10
 
// 정리
CALL gds.graph.drop('dependencyGraph')

// 비방향 그래프 프로젝션 (커뮤니티 감지에 적합)
CALL gds.graph.project(
  'techCommunities',
  'Technology',
  {
    DEPENDS_ON: {orientation: 'UNDIRECTED'},
    RELATED_TO: {orientation: 'UNDIRECTED'}
  }
)
 
// Louvain 커뮤니티 감지
CALL gds.louvain.stream('techCommunities')
YIELD nodeId, communityId
WITH gds.util.asNode(nodeId) AS tech, communityId
WITH communityId, collect(tech.name) AS members, count(*) AS size
WHERE size >= 2
RETURN communityId, members, size
ORDER BY size DESC
 
// Leiden 알고리즘 (Louvain의 개선 버전, GraphRAG에서 사용)
CALL gds.leiden.stream('techCommunities', {
  gamma: 1.0,
  theta: 0.01
})
YIELD nodeId, communityId
WITH gds.util.asNode(nodeId) AS tech, communityId
RETURN communityId, collect(tech.name) AS members
 
CALL gds.graph.drop('techCommunities')

CALL gds.graph.project(
  'bridgeGraph',
  ['Technology', 'Concept'],
  {
    DEPENDS_ON: {orientation: 'UNDIRECTED'},
    IMPLEMENTS: {orientation: 'UNDIRECTED'}
  }
)
 
CALL gds.betweenness.stream('bridgeGraph')
YIELD nodeId, score
WITH gds.util.asNode(nodeId) AS node, score
WHERE score > 0
RETURN node.name, labels(node) AS type, round(score, 2) AS betweenness
ORDER BY betweenness DESC
LIMIT 10
 
CALL gds.graph.drop('bridgeGraph')

from neo4j import GraphDatabase
from anthropic import Anthropic
 
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
llm = Anthropic()
 
def answer_with_graph_context(question: str, entity: str) -> str:
    """그래프 컨텍스트를 활용하여 질문에 답변합니다."""
 
    # 1. 엔티티 주변 그래프 정보 수집
    records, _, _ = driver.execute_query("""
        MATCH (e {name: $entity})
        OPTIONAL MATCH (e)-[r]->(target)
        WITH e, collect({type: type(r), target: target.name, props: properties(r)}) AS outgoing
        OPTIONAL MATCH (source)-[r2]->(e)
        WITH e, outgoing, collect({type: type(r2), source: source.name}) AS incoming
        RETURN e.name AS name,
               labels(e) AS types,
               properties(e) AS props,
               outgoing,
               incoming
    """, entity=entity)
 
    if not records:
        return "해당 엔티티를 찾을 수 없습니다."
 
    record = records[0]
 
    # 2. 그래프 정보를 자연어 컨텍스트로 변환
    context_parts = [f"[{record['name']}]은(는) {', '.join(record['types'])} 타입입니다."]
 
    for rel in record["outgoing"]:
        if rel["target"]:
            context_parts.append(f"- {record['name']} --{rel['type']}--> {rel['target']}")
 
    for rel in record["incoming"]:
        if rel["source"]:
            context_parts.append(f"- {rel['source']} --{rel['type']}--> {record['name']}")
 
    graph_context = "\n".join(context_parts)
 
    # 3. LLM에 그래프 컨텍스트와 함께 질문
    response = llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""다음 지식 그래프 정보를 바탕으로 질문에 답변하세요.
 
지식 그래프 컨텍스트:
{graph_context}
 
질문: {question}
 
답변:"""
        }]
    )
    return response.content[0].text

def multi_hop_reasoning(question: str) -> str:
    """다중 홉 추론을 수행합니다."""
 
    # 1. LLM에게 질문 분해 요청
    decomposition = llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""다음 질문을 단계별 하위 질문으로 분해하세요.
각 단계는 이전 단계의 결과를 활용합니다.
 
질문: {question}
 
JSON 형식으로 응답:
{{"steps": ["하위 질문 1", "하위 질문 2", ...]}}"""
        }]
    )
 
    # 2. 각 단계를 순차적으로 실행
    steps = parse_json(decomposition.content[0].text)["steps"]
    accumulated_context = []
 
    for i, step in enumerate(steps):
        # 각 하위 질문을 Cypher로 변환하여 실행
        cypher = text_to_cypher(step, accumulated_context)
        result = execute_cypher(cypher)
        accumulated_context.append(f"단계 {i + 1}: {step}\n결과: {result}")
 
    # 3. 최종 답변 생성
    final_response = llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""다음 추론 과정을 바탕으로 원래 질문에 답변하세요.
 
원래 질문: {question}
 
추론 과정:
{chr(10).join(accumulated_context)}
 
최종 답변:"""
        }]
    )
    return final_response.content[0].text

TEXT2CYPHER_PROMPT = """당신은 자연어를 Neo4j Cypher 쿼리로 변환하는 전문가입니다.
 
## 그래프 스키마
노드:
- Person (name, role, affiliation)
- Technology (name, category, version)
- Document (title, content, publishedAt)
- Concept (name, description)
- Category (name)
 
관계:
- (Person)-[:AUTHORED]->(Document)
- (Person)-[:EXPERT_IN]->(Technology)
- (Document)-[:COVERS]->(Technology)
- (Technology)-[:BELONGS_TO]->(Category)
- (Technology)-[:DEPENDS_ON]->(Technology)
- (Technology)-[:IMPLEMENTS]->(Concept)
 
## 규칙
1. 유효한 Cypher만 생성하세요
2. 주입 공격 방지를 위해 파라미터($param)를 사용하세요
3. 결과 수를 LIMIT으로 제한하세요
4. 설명 없이 Cypher 쿼리만 반환하세요
"""
 
def text_to_cypher(question: str, context: list[str] = None) -> str:
    """자연어 질문을 Cypher 쿼리로 변환합니다."""
    messages = [{"role": "user", "content": f"질문: {question}"}]
 
    if context:
        messages[0]["content"] += f"\n\n이전 컨텍스트:\n{chr(10).join(context)}"
 
    response = llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=512,
        system=TEXT2CYPHER_PROMPT,
        messages=messages
    )
 
    cypher = response.content[0].text.strip()
    # 마크다운 코드 블록 제거
    if cypher.startswith("```"):
        cypher = cypher.split("```")[1]
        if cypher.startswith("cypher"):
            cypher = cypher[6:]
    return cypher.strip()

class Text2CypherPipeline:
    """자연어 질의를 Cypher로 변환하고 실행하는 파이프라인입니다."""
 
    def __init__(self, driver, llm_client):
        self.driver = driver
        self.llm = llm_client
 
    def query(self, question: str) -> dict:
        """자연어 질문에 대한 답변을 생성합니다."""
 
        # 1. 자연어 -> Cypher
        cypher = text_to_cypher(question)
        print(f"생성된 Cypher: {cypher}")
 
        # 2. Cypher 실행
        try:
            records, _, _ = self.driver.execute_query(cypher)
            results = [record.data() for record in records]
        except Exception as err:
            # 쿼리 실행 실패 시 재시도
            print(f"쿼리 실행 실패: {err}")
            cypher = self._retry_with_error(question, cypher, str(err))
            records, _, _ = self.driver.execute_query(cypher)
            results = [record.data() for record in records]
 
        # 3. 결과를 자연어로 변환
        answer = self._generate_answer(question, results)
 
        return {
            "question": question,
            "cypher": cypher,
            "raw_results": results,
            "answer": answer
        }
 
    def _retry_with_error(self, question: str, failed_cypher: str,
                          error: str) -> str:
        """실패한 쿼리를 오류 정보와 함께 재생성합니다."""
        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            system=TEXT2CYPHER_PROMPT,
            messages=[{
                "role": "user",
                "content": f"""이전에 생성한 Cypher가 실패했습니다. 수정해 주세요.
 
질문: {question}
실패한 쿼리: {failed_cypher}
오류: {error}
 
수정된 Cypher:"""
            }]
        )
        return response.content[0].text.strip()
 
    def _generate_answer(self, question: str, results: list[dict]) -> str:
        """쿼리 결과를 자연어 답변으로 변환합니다."""
        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": f"""다음 데이터를 바탕으로 질문에 자연스럽게 답변하세요.
 
질문: {question}
데이터: {results}
 
답변:"""
            }]
        )
        return response.content[0].text