[PYTHON/LANGCHAIN] NEO4J 데이터베이스 결과에 기반한 답변 생성하기

■ NEO4J 데이터베이스 결과에 기반한 답변을 생성하는 방법을 보여준다.

※ NEO4J_URI, NEO4J_USERNAME, NEO4J_PASSWORD 환경 변수 값은 .env 파일에 정의한다.
※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

▶ main.py


from dotenv                                       import load_dotenv
from langchain_core.prompts                       import ChatPromptTemplate
from langchain_openai                             import ChatOpenAI
from pydantic                                     import BaseModel
from typing                                       import List
from pydantic                                     import Field
from langchain_neo4j                              import Neo4jGraph
from typing                                       import Optional
from langchain_core.runnables                     import RunnablePassthrough
from langchain_core.output_parsers                import StrOutputParser
from langchain_neo4j.chains.graph_qa.cypher_utils import Schema
from langchain_neo4j.chains.graph_qa.cypher_utils import CypherQueryCorrector

load_dotenv()

chatPromptTemplate1 = ChatPromptTemplate.from_messages(
    [
        ("system", "You are extracting person and movies from the text."),
        ("human" , "Use the given format to extract information from the following input : {question}")
    ]
)

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

class EntityData(BaseModel):
    """Identifying information about entities."""

    nameList : List[str] = Field(..., description = "All the person or movies appearing in the text")

runnableSequence1 = chatPromptTemplate1 | chatOpenAI.with_structured_output(EntityData)

selectDataQuery = """MATCH (p:Person|Movie)
WHERE p.name CONTAINS $value OR p.title CONTAINS $value
RETURN coalesce(p.name, p.title) AS result, labels(p)[0] AS type
LIMIT 1
"""

neo4jGraph = Neo4jGraph()

def getMappingStringFromDatabase(entityData : EntityData) -> Optional[str]:
    mappingString = ""
    for entityName in entityData.nameList:
        responseDictionaryList = neo4jGraph.query(selectDataQuery, {"value" : entityName})
        try:
            mappingString += f"{entityName} maps to {responseDictionaryList[0]['result']} {responseDictionaryList[0]['type']} in database\n"
        except IndexError:
            pass
    return mappingString

humanMessage2 = """Based on the Neo4j graph schema below, write a Cypher query that would answer the user's question :
{schema}
Entities in the question map to the following database values :
{entity_list}
Question : {question}
Cypher query :"""

chatPromptTemplate2 = ChatPromptTemplate.from_messages(
    [
        ("system", "Given an input question, convert it to a Cypher query. No pre-amble."),
        ("human" , humanMessage2)
    ]
)

runnableSequence2 = (
    RunnablePassthrough.assign(entity_data = runnableSequence1)
    | RunnablePassthrough.assign(entity_list = lambda x : getMappingStringFromDatabase(x["entity_data"]), schema = lambda _ : neo4jGraph.get_schema)
    | chatPromptTemplate2
    | chatOpenAI.bind(stop = ["\nCypherResult:"])
    | StrOutputParser()
)

schemaList = [Schema(elementDictionary["start"], elementDictionary["type"], elementDictionary["end"]) for elementDictionary in neo4jGraph.structured_schema.get("relationships")]

cypherQueryCorrector = CypherQueryCorrector(schemaList)

humanMessage3 = """Based on the the question, Cypher query, and Cypher response, write a natural language response:
Question : {question}
Cypher query : {query}
Cypher Response : {response}"""

chatPromptTemplate3 = ChatPromptTemplate.from_messages(
    [
        ("system", "Given an input question and Cypher response, convert it to a natural language answer. No pre-amble."),
        ("human" , humanMessage3)
    ]
)

runnableSequence3 = (
    RunnablePassthrough.assign(query = runnableSequence2)
    | RunnablePassthrough.assign(response = lambda x : neo4jGraph.query(cypherQueryCorrector(x["query"])))
    | chatPromptTemplate3
    | chatOpenAI
    | StrOutputParser()
)

responseString = runnableSequence3.invoke({"question" : "Who played in Casino movie?"})

print(responseString)

"""
The actors who played in the movie "Casino" are Robert De Niro, Joe Pesci, Sharon Stone, and James Woods.
"""

100

101

102

103

104

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from pydantic import BaseModel

from typing import List

from pydantic import Field

from langchain_neo4j import Neo4jGraph

from typing import Optional

from langchain_core.runnables import RunnablePassthrough

from langchain_core.output_parsers import StrOutputParser

from langchain_neo4j.chains.graph_qa.cypher_utils import Schema

from langchain_neo4j.chains.graph_qa.cypher_utils import CypherQueryCorrector

load_dotenv()

chatPromptTemplate1 = ChatPromptTemplate.from_messages(

[

("system", "You are extracting person and movies from the text."),

("human" , "Use the given format to extract information from the following input : {question}")

]

)

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

class EntityData(BaseModel):

"""Identifying information about entities."""

nameList : List[str] = Field(..., description = "All the person or movies appearing in the text")

runnableSequence1 = chatPromptTemplate1 | chatOpenAI.with_structured_output(EntityData)

selectDataQuery = """MATCH (p:Person|Movie)

WHERE p.name CONTAINS $value OR p.title CONTAINS $value

RETURN coalesce(p.name, p.title) AS result, labels(p)[0] AS type

LIMIT 1

"""

neo4jGraph = Neo4jGraph()

def getMappingStringFromDatabase(entityData : EntityData) -> Optional[str]:

mappingString = ""

for entityName in entityData.nameList:

responseDictionaryList = neo4jGraph.query(selectDataQuery, {"value" : entityName})

try:

mappingString += f"{entityName} maps to {responseDictionaryList[0]['result']} {responseDictionaryList[0]['type']} in database\n"

except IndexError:

pass

return mappingString

humanMessage2 = """Based on the Neo4j graph schema below, write a Cypher query that would answer the user's question :

{schema}

Entities in the question map to the following database values :

{entity_list}

Question : {question}

Cypher query :"""

chatPromptTemplate2 = ChatPromptTemplate.from_messages(

[

("system", "Given an input question, convert it to a Cypher query. No pre-amble."),

("human" , humanMessage2)

]

)

runnableSequence2 = (

RunnablePassthrough.assign(entity_data = runnableSequence1)

| RunnablePassthrough.assign(entity_list = lambda x : getMappingStringFromDatabase(x["entity_data"]), schema = lambda _ : neo4jGraph.get_schema)

| chatPromptTemplate2

| chatOpenAI.bind(stop = ["\nCypherResult:"])

| StrOutputParser()

)

schemaList = [Schema(elementDictionary["start"], elementDictionary["type"], elementDictionary["end"]) for elementDictionary in neo4jGraph.structured_schema.get("relationships")]

cypherQueryCorrector = CypherQueryCorrector(schemaList)

humanMessage3 = """Based on the the question, Cypher query, and Cypher response, write a natural language response:

Question : {question}

Cypher query : {query}

Cypher Response : {response}"""

chatPromptTemplate3 = ChatPromptTemplate.from_messages(

[

("system", "Given an input question and Cypher response, convert it to a natural language answer. No pre-amble."),

("human" , humanMessage3)

]

)

runnableSequence3 = (

RunnablePassthrough.assign(query = runnableSequence2)

| RunnablePassthrough.assign(response = lambda x : neo4jGraph.query(cypherQueryCorrector(x["query"])))

| chatPromptTemplate3

| chatOpenAI

| StrOutputParser()

)

responseString = runnableSequence3.invoke({"question" : "Who played in Casino movie?"})

print(responseString)

"""

The actors who played in the movie "Casino" are Robert De Niro, Joe Pesci, Sharon Stone, and James Woods.

"""

▶ requirements.txt


aiohappyeyeballs==2.4.4
aiohttp==3.11.10
aiosignal==1.3.2
annotated-types==0.7.0
anyio==4.7.0
attrs==24.3.0
certifi==2024.12.14
charset-normalizer==3.4.0
colorama==0.4.6
distro==1.9.0
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
idna==3.10
jiter==0.8.2
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.12
langchain-core==0.3.25
langchain-neo4j==0.1.1
langchain-openai==0.2.12
langchain-text-splitters==0.3.3
langsmith==0.2.3
multidict==6.1.0
neo4j==5.27.0
numpy==2.2.0
openai==1.57.4
orjson==3.10.12
packaging==24.2
propcache==0.2.1
pydantic==2.10.3
pydantic_core==2.27.1
python-dotenv==1.0.1
pytz==2024.2
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.36
tenacity==9.0.0
tiktoken==0.8.0
tqdm==4.67.1
typing_extensions==4.12.2
urllib3==2.2.3
yarl==1.18.3

aiohappyeyeballs==2.4.4

aiohttp==3.11.10

aiosignal==1.3.2

annotated-types==0.7.0

anyio==4.7.0

attrs==24.3.0

certifi==2024.12.14

charset-normalizer==3.4.0

colorama==0.4.6

distro==1.9.0

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.28.1

idna==3.10

jiter==0.8.2

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.12

langchain-core==0.3.25

langchain-neo4j==0.1.1

langchain-openai==0.2.12

langchain-text-splitters==0.3.3

langsmith==0.2.3

multidict==6.1.0

neo4j==5.27.0

numpy==2.2.0

openai==1.57.4

orjson==3.10.12

packaging==24.2

propcache==0.2.1

pydantic==2.10.3

pydantic_core==2.27.1

python-dotenv==1.0.1

pytz==2024.2

PyYAML==6.0.2

regex==2024.11.6

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.36

tenacity==9.0.0

tiktoken==0.8.0

tqdm==4.67.1

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.18.3

※ pip install python-dotenv langchain-neo4j langchain-openai 명령을 실행했다.