LangChain + Neo4j 통합

GraphCypherQAChain: 자연어 → Cypher 자동 변환

from langchain_neo4j import Neo4jGraph
from langchain_neo4j import GraphCypherQAChain
from langchain_openai import ChatOpenAI

# Neo4j 그래프 연결
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="manufacturing2026"
)

# 스키마 확인 (LLM에게 전달됨)
print(graph.schema)
# Node properties:
#   Equipment {name: STRING, model: STRING, status: STRING, ...}
#   Part {partNumber: STRING, name: STRING, lifespan: INTEGER, ...}
#   Fault {alarmCode: STRING, name: STRING, severity: STRING, ...}
# ...

# LLM 설정
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# GraphCypherQAChain 생성
chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True,  # Cypher 쿼리 출력
    allow_dangerous_requests=True,
    return_intermediate_steps=True,
)

# 자연어로 질문!
result = chain.invoke({
    "query": "알람 E-201의 원인과 조치 방법을 확률 순으로 알려줘"
})

print(result["result"])
# 자동 생성된 Cypher:
# MATCH (f:Fault {alarmCode: "E-201"})
#       -[:CAUSED_BY]->(cause:RootCause)
#       -[:RESOLVED_BY]->(action:MaintenanceAction)
# RETURN cause.name, cause.probability, action.name
# ORDER BY cause.probability DESC

Neo4jVector: 그래프 + 벡터 검색

from langchain_neo4j import Neo4jVector
from langchain_openai import OpenAIEmbeddings

# Neo4j에 벡터 인덱스 생성 + 문서 저장
vector_store = Neo4jVector.from_texts(
    texts=[
        "CNC밀링 3호기의 스핀들 과열 시 먼저 냉각수를 확인합니다.",
        "베어링 교체는 스핀들 정비 자격이 있는 기술자만 수행합니다.",
        "볼스크류 백래시는 다이얼 게이지로 측정 후 프리로드를 조정합니다.",
        "정기 점검은 500시간마다 실시하며, 윤활유 교체를 포함합니다.",
    ],
    embedding=OpenAIEmbeddings(),
    url="bolt://localhost:7687",
    username="neo4j",
    password="manufacturing2026",
    index_name="mfg_docs",
    node_label="Document",
    text_node_property="text",
)

# 유사도 검색
docs = vector_store.similarity_search("스핀들 과열 조치", k=3)
for doc in docs:
    print(doc.page_content)

하이브리드 검색: Graph + Vector

class HybridManufacturingRAG:
    """Graph + Vector 하이브리드 RAG"""

    def __init__(self, graph, vector_store, llm):
        self.graph_chain = GraphCypherQAChain.from_llm(
            llm=llm, graph=graph,
            allow_dangerous_requests=True
        )
        self.vector_store = vector_store
        self.llm = llm

    def ask(self, question: str) -> str:
        # 1. Graph RAG: 구조화된 지식 탐색
        try:
            graph_result = self.graph_chain.invoke({"query": question})
            graph_context = graph_result.get("result", "")
        except Exception:
            graph_context = ""

        # 2. Vector RAG: 관련 문서 검색
        docs = self.vector_store.similarity_search(question, k=3)
        doc_context = "\n".join(d.page_content for d in docs)

        # 3. 하이브리드 맥락으로 답변 생성
        prompt = f"""당신은 제조 현장 기술 지원 AI입니다.

## Knowledge Graph 정보:
{graph_context}

## 관련 기술 문서:
{doc_context}

## 질문:
{question}

위 정보를 종합하여 실용적인 답변을 제공하세요.
"""
        from langchain_core.messages import HumanMessage
        response = self.llm.invoke([HumanMessage(content=prompt)])
        return response.content

주의사항

[GraphCypherQAChain 사용 시 주의]

1. allow_dangerous_requests=True 필수 (쓰기 쿼리 방지는 별도 처리)
2. 스키마가 복잡하면 LLM이 잘못된 Cypher를 생성할 수 있음
   → validate_cypher=True 옵션 사용
3. 한국어 질문에서 영어 속성명으로의 변환이 핵심
   → 프롬프트에 스키마 설명 추가
4. 비용 관리: 매 질문마다 LLM 호출 발생
   → 자주 묻는 질문은 Cypher 캐싱 권장