Text2Cypher QA 시스템

자연어 질의를 Cypher로 자동 변환하여 정확한 그래프 답변 생성

자연어 질문을 Cypher 쿼리로 자동 변환하는 Text2Cypher 시스템을 구현한다. GraphCypherQAChain의 기본 사용법부터 스키마 주입, Few-shot 예시 기반 정확도 향상, Cypher 검증, 오류 처리까지 실무에서 필요한 Text2Cypher 패턴을 다룬다.

AI
RAG
GraphRAG
Neo4j
저자

Kwangmin Kim

공개

2026년 03월 08일

1 Text2Cypher QA 시스템

1.1 Text2Cypher란

자연어 질문 → Cypher 쿼리 자동 생성 → Neo4j 실행 → 답변 생성

사용자: "일론 머스크가 설립한 회사는?"
   ↓ LLM + 그래프 스키마
Cypher: MATCH (p:Person {name: "Elon Musk"})-[:FOUNDED]->(c:Company)
        RETURN c.name

   ↓ Neo4j 실행
결과: [{"c.name": "Tesla"}, {"c.name": "SpaceX"}]

   ↓ LLM
답변: "일론 머스크가 설립한 회사는 Tesla와 SpaceX입니다."

1.2 기본 GraphCypherQAChain

from langchain_neo4j import Neo4jGraph, GraphCypherQAChain
from langchain_openai import ChatOpenAI

graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password",
)

# 스키마를 LLM에 자동 주입
graph.refresh_schema()
print(graph.schema)
# Node properties:
# Person {name: STRING, born: INTEGER, pagerank: FLOAT}
# Company {name: STRING, founded: INTEGER, sector: STRING}
# Location {name: STRING, state: STRING}
# Relationship properties:
# FOUNDED {year: INTEGER}
# The relationships:
# (:Person)-[:FOUNDED]->(:Company)
# (:Company)-[:LOCATED_IN]->(:Location)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True,
    validate_cypher=True,   # 실행 전 Cypher 문법 검증
    top_k=10,
    return_intermediate_steps=True,  # 생성된 Cypher도 반환
)

result = chain.invoke("Who founded Tesla?")
print("답변:", result["result"])
print("생성된 Cypher:", result["intermediate_steps"][0]["query"])

1.3 스키마 최적화: 필요한 정보만 노출

스키마가 너무 복잡하면 LLM이 잘못된 Cypher를 생성한다. GraphRAG에 필요한 부분만 선택적으로 노출한다.

# 커스텀 스키마 정의
custom_schema = """
Node properties:
- Person: {name: STRING, born: INTEGER}
- Company: {name: STRING, founded: INTEGER, sector: STRING}
- Location: {name: STRING}

Relationships:
- (Person)-[:FOUNDED {year: INTEGER}]->(Company)
- (Person)-[:WORKS_AT]->(Company)
- (Company)-[:LOCATED_IN]->(Location)
- (Company)-[:ACQUIRED]->(Company)
"""

from langchain_neo4j import GraphCypherQAChain
from langchain_core.prompts import PromptTemplate

CYPHER_GENERATION_PROMPT = PromptTemplate.from_template("""
Task: Generate a Cypher statement to query Neo4j.

Schema:
{schema}

Instructions:
- Use only the provided node labels and relationship types.
- Do not use properties or relationships not listed in the schema.
- Use MERGE instead of CREATE to avoid duplicates.
- Always use parameters instead of literal values when possible.

Question: {question}

Cypher Query:
""")

chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    cypher_prompt=CYPHER_GENERATION_PROMPT,
    schema=custom_schema,   # 커스텀 스키마 주입
    verbose=True,
)

1.4 Few-shot 예시로 정확도 향상

도메인 특화 예시를 제공하면 Cypher 생성 품질이 크게 향상된다.

few_shot_examples = """
# 예시 1: 설립자 조회
질문: "Tesla를 만든 사람은?"
Cypher: MATCH (p:Person)-[:FOUNDED]->(c:Company {{name: 'Tesla'}}) RETURN p.name

# 예시 2: 다중 회사 설립자
질문: "두 개 이상의 회사를 설립한 사람은?"
Cypher: MATCH (p:Person)-[:FOUNDED]->(c:Company) WITH p, count(c) AS num WHERE num >= 2 RETURN p.name, num ORDER BY num DESC

# 예시 3: 위치 기반 조회
질문: "텍사스에 있는 회사는?"
Cypher: MATCH (c:Company)-[:LOCATED_IN]->(l:Location {{state: 'Texas'}}) RETURN c.name

# 예시 4: 2-hop 탐색
질문: "일론 머스크의 회사들이 위치한 도시는?"
Cypher: MATCH (p:Person {{name: 'Elon Musk'}})-[:FOUNDED]->(c:Company)-[:LOCATED_IN]->(l:Location) RETURN DISTINCT l.name
"""

CYPHER_PROMPT_WITH_EXAMPLES = PromptTemplate.from_template("""
Task: Generate a Cypher query for Neo4j based on the schema and examples.

Schema:
{schema}

Examples:
""" + few_shot_examples + """

Question: {question}
Cypher Query:
""")

chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    cypher_prompt=CYPHER_PROMPT_WITH_EXAMPLES,
    verbose=True,
)

1.5 오류 처리 및 재시도

from langchain_neo4j import GraphCypherQAChain
from langchain_openai import ChatOpenAI

CYPHER_FIX_PROMPT = PromptTemplate.from_template("""
The following Cypher query failed with an error.

Original Question: {question}
Failed Cypher: {cypher}
Error: {error}

Please generate a corrected Cypher query:
""")

def safe_text2cypher(question: str, max_retries: int = 2) -> str:
    """오류 발생 시 자동 재시도."""

    for attempt in range(max_retries + 1):
        try:
            result = chain.invoke(question)
            return result["result"]

        except Exception as e:
            if attempt < max_retries:
                print(f"시도 {attempt+1} 실패: {e}")

                # 오류 메시지를 포함하여 수정된 Cypher 재생성
                fix_prompt = CYPHER_FIX_PROMPT.format(
                    question=question,
                    cypher=str(e).split("query:")[-1] if "query:" in str(e) else "",
                    error=str(e),
                )
                fixed_cypher = llm.invoke(fix_prompt).content

                # 수정된 Cypher로 직접 실행
                try:
                    result = graph.query(fixed_cypher)
                    return str(result)
                except Exception as e2:
                    print(f"수정 시도도 실패: {e2}")
            else:
                return f"질문을 처리할 수 없습니다: {e}"

    return "최대 재시도 횟수 초과"

1.6 neo4j-graphrag의 Text2CypherRetriever

공식 Neo4j GraphRAG 패키지 사용 시:

import neo4j
from neo4j_graphrag.retrievers import Text2CypherRetriever
from neo4j_graphrag.llm import OpenAILLM

driver = neo4j.GraphDatabase.driver(
    "bolt://localhost:7687", auth=("neo4j", "password")
)
llm = OpenAILLM(model_name="gpt-4o", model_params={"temperature": 0})

# 스키마 직접 제공
neo4j_schema = """
Node properties:
Person {name: STRING, born: INTEGER}
Company {name: STRING, founded: INTEGER}
Relationship properties:
FOUNDED {year: INTEGER}
The relationships:
(:Person)-[:FOUNDED]->(:Company)
"""

retriever = Text2CypherRetriever(
    driver=driver,
    llm=llm,
    neo4j_schema=neo4j_schema,
)

result = retriever.search(query_text="Who founded Tesla?")
for item in result.items:
    print(item.content)

1.7 전체 QA 파이프라인

Text2Cypher + 벡터 검색 결합:

from langchain_core.runnables import RunnablePassthrough

ANSWER_PROMPT = PromptTemplate.from_template("""
다음 정보를 바탕으로 질문에 정확하게 답하세요.

질문: {question}

그래프 검색 결과:
{graph_result}

벡터 검색 결과:
{vector_result}

답변:
""")

def comprehensive_qa(question: str) -> str:
    """Text2Cypher + 벡터 검색 결합 QA."""

    # 그래프 검색 (Text2Cypher)
    try:
        graph_result = chain.invoke(question)["result"]
    except Exception:
        graph_result = "그래프에서 직접적인 정보를 찾을 수 없음"

    # 벡터 검색
    vector_results = vector_store.similarity_search(question, k=3)
    vector_result = "\n".join(doc.page_content for doc in vector_results)

    # LLM으로 통합 답변
    return llm.invoke(
        ANSWER_PROMPT.format(
            question=question,
            graph_result=graph_result,
            vector_result=vector_result,
        )
    ).content

# 실행
answer = comprehensive_qa("일론 머스크가 설립한 회사들과 각각의 특징은?")
print(answer)

1.8 정리

Text2Cypher 흐름:
  질문 → [LLM + 스키마] → Cypher 생성 → Neo4j 실행 → 결과 → [LLM] → 답변

품질 향상 방법:
  1. 스키마 최적화: 필요한 부분만 노출
  2. Few-shot 예시: 도메인 특화 Cypher 예시 제공
  3. 오류 처리: 실패 시 자동 Cypher 수정 재시도

선택지:
  GraphCypherQAChain    ← LangChain 표준 (간편)
  Text2CypherRetriever  ← neo4j-graphrag 공식 (정교)
  수동 구현             ← 완전한 제어 필요 시

다음 파일에서는 Neo4j GraphRAG와 메타데이터 기반 방식을 정량적으로 비교한다.

Subscribe

Enjoy this blog? Get notified of new posts by email: