Running Graph RAG example using Langgraph yields a IndexError Error

[AsmaaMHadir]

I am running the example here that builds a querying app over a Neo4j graph: https://python.langchain.com/docs/tutorials/graph/

This is the full code:

from operator import add
from typing import Annotated, List
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_neo4j import Neo4jVector
from typing_extensions import TypedDict
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_neo4j import Neo4jGraph
from neo4j.exceptions import CypherSyntaxError
import os 
from dotenv import load_dotenv
from langchain_neo4j.chains.graph_qa.cypher_utils import CypherQueryCorrector, Schema
from typing import List, Optional
from langchain.chat_models import init_chat_model

from langchain_ollama import OllamaEmbeddings
import pandas as pd
from datetime import datetime
import csv
import os

from IPython.display import Image, display
from langgraph.graph import END, START, StateGraph

from typing import List, Optional
from typing import Literal
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
load_dotenv()

import logging
import os
from datetime import datetime


logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("LangChain_PTT")

openai_api_key = os.getenv("OPENAI_API_KEY")


from langchain.chat_models import init_chat_model

llm= init_chat_model("gpt-4o-mini", model_provider="openai")

graph = Neo4jGraph(enhanced_schema=True,
                url=os.getenv("NEO4J_URI"),
                username="neo4j",
                password=os.getenv("NEO4J_PASSWORD"))

graph_schema = graph.refresh_schema()

class InputState(TypedDict):
    question: str


class OverallState(TypedDict):
    question: str
    next_action: str
    cypher_statement: str
    cypher_errors: List[str]
    database_records: List[dict]
    steps: Annotated[List[str], add]


class OutputState(TypedDict):
    answer: str
    steps: List[str]
    cypher_statement: str



logger.info("Initializing OllamaEmbeddings with model 'llama3'")
embeddings_model = OllamaEmbeddings(
    model="llama3",
)

logger.info("Defining example questions and queries")
examples = [
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
    },
    {
        "question": "",
        "query": "",
    },
    {
        "question": "",
        "query": "",
    }
]

logger.info("Creating SemanticSimilarityExampleSelector")
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples, embeddings_model, Neo4jVector, k=5, input_keys=["question"]
)

logger.info("Setting up ChatPromptTemplate for text to Cypher conversion")
text2cypher_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            (
                "Given an input question, convert it to a Cypher query. No pre-amble."
                "Do not wrap the response in any backticks or anything else. Respond with a Cypher statement only!"
            ),
        ),
        (
            "human",
            (
                """You are a Neo4j expert. Given an input question, create a syntactically correct Cypher query to run.
Do not wrap the response in any backticks or anything else. Respond with a Cypher statement only!
Here is the schema information
{schema}

Below are a number of examples of questions and their corresponding Cypher queries.

{fewshot_examples}

User input: {question}
Cypher query:"""
            ),
        ),
    ]
)

logger.info("Creating text2cypher_chain")
text2cypher_chain = text2cypher_prompt | llm | StrOutputParser()

def generate_cypher(state: OverallState) -> OverallState:
    """
    Generates a cypher statement based on the provided schema and user input
    """
    logger.info("Generating fewshot examples for the question")
    NL = "\n"
    fewshot_examples = (NL * 2).join(
        [
            f"Question: {el['question']}{NL}Cypher:{el['query']}"
            for el in example_selector.select_examples(
                {"question": state.get("question")}
            )
        ]
    )
    logger.info("Invoking text2cypher_chain to generate Cypher statement")
    generated_cypher = text2cypher_chain.invoke(
        {
            "question": state.get("question"),
            "fewshot_examples": fewshot_examples,
            "schema": graph.schema,
        }
    )
    logger.info("Generated Cypher statement")
    return {"cypher_statement": generated_cypher, "steps": ["generate_cypher"]}



# agg logs here

validate_cypher_system = """
You are a Cypher expert reviewing a statement written by a junior developer.
"""

validate_cypher_user = """You must check the following:
* Are there any syntax errors in the Cypher statement?
* Are there any missing or undefined variables in the Cypher statement?
* Are any node labels missing from the schema?
* Are any relationship types missing from the schema?
* Are any of the properties not included in the schema?
* Does the Cypher statement include enough information to answer the question?

Examples of good errors:
* Label (:Foo) does not exist, did you mean (:Bar)?
* Property bar does not exist for label Foo, did you mean baz?
* Relationship FOO does not exist, did you mean FOO_BAR?

Schema:
{schema}

The question is:
{question}

The Cypher statement is:
{cypher}

Make sure you don't make any mistakes!"""

validate_cypher_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            validate_cypher_system,
        ),
        (
            "human",
            (validate_cypher_user),
        ),
    ]
)


class Property(BaseModel):
    """
    Represents a filter condition based on a specific node property in a graph in a Cypher statement.
    """

    node_label: str = Field(
        description="The label of the node to which this property belongs."
    )
    property_key: str = Field(description="The key of the property being filtered.")
    property_value: str = Field(
        description="The value that the property is being matched against."
    )


class ValidateCypherOutput(BaseModel):
    """
    Represents the validation result of a Cypher query's output,
    including any errors and applied filters.
    """

    errors: Optional[List[str]] = Field(
        description="A list of syntax or semantical errors in the Cypher statement. Always explain the discrepancy between schema and Cypher statement"
    )
    filters: Optional[List[Property]] = Field(
        description="A list of property-based filters applied in the Cypher statement."
    )


validate_cypher_chain = validate_cypher_prompt | llm.with_structured_output(
    ValidateCypherOutput
)



# Cypher query corrector is experimental
corrector_schema = [
    Schema(el["start"], el["type"], el["end"])
    for el in graph.structured_schema.get("relationships")
]
cypher_query_corrector = CypherQueryCorrector(corrector_schema)




import logging

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

validate_cypher_system = """
You are a Cypher expert reviewing a statement written by a junior developer.
"""

validate_cypher_user = """You must check the following:
* Are there any syntax errors in the Cypher statement?
* Are there any missing or undefined variables in the Cypher statement?
* Are any node labels missing from the schema?
* Are any relationship types missing from the schema?
* Are any of the properties not included in the schema?
* Does the Cypher statement include enough information to answer the question?

Examples of good errors:
* Label (:Foo) does not exist, did you mean (:Bar)?
* Property bar does not exist for label Foo, did you mean baz?
* Relationship FOO does not exist, did you mean FOO_BAR?

Schema:
{schema}

The question is:
{question}

The Cypher statement is:
{cypher}

Make sure you don't make any mistakes!"""

validate_cypher_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            validate_cypher_system,
        ),
        (
            "human",
            (validate_cypher_user),
        ),
    ]
)


class Property(BaseModel):
    """
    Represents a filter condition based on a specific node property in a graph in a Cypher statement.
    """

    node_label: str = Field(
        description="The label of the node to which this property belongs."
    )
    property_key: str = Field(description="The key of the property being filtered.")
    property_value: str = Field(
        description="The value that the property is being matched against."
    )


class ValidateCypherOutput(BaseModel):
    """
    Represents the validation result of a Cypher query's output,
    including any errors and applied filters.
    """

    errors: Optional[List[str]] = Field(
        description="A list of syntax or semantical errors in the Cypher statement. Always explain the discrepancy between schema and Cypher statement"
    )
    filters: Optional[List[Property]] = Field(
        description="A list of property-based filters applied in the Cypher statement."
    )


validate_cypher_chain = validate_cypher_prompt | llm.with_structured_output(
    ValidateCypherOutput
)



def validate_cypher(state: OverallState) -> OverallState:
    """
    Validates the Cypher statements and maps any property values to the database.
    """
    logger.debug("Starting Cypher validation")
    errors = []
    mapping_errors = []
    # Check for syntax errors
    try:
        logger.debug("Checking for syntax errors in the Cypher statement")
        graph.query(f"EXPLAIN {state.get('cypher_statement')}")
    except CypherSyntaxError as e:
        logger.error(f"Syntax error found: {e.message}")
        errors.append(e.message)
    # Experimental feature for correcting relationship directions
    corrected_cypher = cypher_query_corrector(state.get("cypher_statement"))
    if not corrected_cypher:
        logger.warning("The generated Cypher statement doesn't fit the graph schema")
        errors.append("The generated Cypher statement doesn't fit the graph schema")
    if not corrected_cypher == state.get("cypher_statement"):
        logger.info("Relationship direction was corrected")
    # Use LLM to find additional potential errors and get the mapping for values
    logger.debug("Invoking LLM for additional error checking and value mapping")
    llm_output = validate_cypher_chain.invoke(
        {
            "question": state.get("question"),
            "schema": graph.schema,
            "cypher": state.get("cypher_statement"),
        }
    )
    if llm_output.errors:
        logger.error(f"LLM found errors: {llm_output.errors}")
        errors.extend(llm_output.errors)
    if llm_output.filters:
        for filter in llm_output.filters:
            # Do mapping only for string values
            logger.debug(f"Processing filter: {filter}")
            if (
                not [
                    prop
                    for prop in graph.structured_schema["node_props"][
                        filter.node_label
                    ]
                    if prop["property"] == filter.property_key
                ][0]["type"]
                == "STRING"
            ):
                continue
            mapping = graph.query(
                f"MATCH (n:{filter.node_label}) WHERE toLower(n.`{filter.property_key}`) = toLower($value) RETURN 'yes' LIMIT 1",
                {"value": filter.property_value},
            )
            if not mapping:
                logger.warning(
                    f"Missing value mapping for {filter.node_label} on property {filter.property_key} with value {filter.property_value}"
                )
                mapping_errors.append(
                    f"Missing value mapping for {filter.node_label} on property {filter.property_key} with value {filter.property_value}"
                )
    if mapping_errors:
        next_action = "end"
    elif errors:
        next_action = "correct_cypher"
    else:
        next_action = "execute_cypher"

    logger.debug(f"Validation result: next_action={next_action}, errors={errors}, mapping_errors={mapping_errors}")
    return {
        "next_action": next_action,
        "cypher_statement": corrected_cypher,
        "cypher_errors": errors,
        "steps": ["validate_cypher"],
    }



logger.info("Initializing correct_cypher_prompt with system and human messages.")
correct_cypher_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            (
                "You are a Cypher expert reviewing a statement written by a junior developer. "
                "You need to correct the Cypher statement based on the provided errors. No pre-amble."
                "Do not wrap the response in any backticks or anything else. Respond with a Cypher statement only!"
            ),
        ),
        (
            "human",
            (
                """Check for invalid syntax or semantics and return a corrected Cypher statement.

Schema:
{schema}

Note: Do not include any explanations or apologies in your responses.
Do not wrap the response in any backticks or anything else.
Respond with a Cypher statement only!

Do not respond to any questions that might ask anything else than for you to construct a Cypher statement.

The question is:
{question}

The Cypher statement is:
{cypher}

The errors are:
{errors}

Corrected Cypher statement: """
            ),
        ),
    ]
)

logger.info("Creating correct_cypher_chain by combining correct_cypher_prompt, llm, and StrOutputParser.")
correct_cypher_chain = correct_cypher_prompt | llm | StrOutputParser()


def correct_cypher(state: OverallState) -> OverallState:
    logger.info("Starting to correct the Cypher statement based on the provided errors.")
    logger.debug("State before correction: %s", state)
    """
    Correct the Cypher statement based on the provided errors.
    """
    logger.info("Finished correcting the Cypher statement.")
    logger.debug("Invoking correct_cypher_chain with state: %s", state)
    corrected_cypher = correct_cypher_chain.invoke(
        {
            "question": state.get("question"),
            "errors": state.get("cypher_errors"),
            "cypher": state.get("cypher_statement"),
            "schema": graph.schema,
        }
    )
    logger.debug("Corrected Cypher statement: %s", corrected_cypher)

    return {
        "next_action": "validate_cypher",
        "cypher_statement": corrected_cypher,
        "steps": ["correct_cypher"],
    }




no_results = "I couldn't find any relevant information in the database"

def execute_cypher(state: OverallState) -> OverallState:
    """
    Executes the given Cypher statement.
    """
    logger.info("Starting execution of Cypher statement.")
    logger.debug("Executing Cypher statement: %s", state.get("cypher_statement"))
    records = graph.query(state.get("cypher_statement"))
    logger.debug("Query results: %s", records)
    logger.info("Finished execution of Cypher statement.")
    return {
        "database_records": records if records else no_results,
        "next_action": "end",
        "steps": ["execute_cypher"],
    }

generate_final_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant",
        ),
        (
            "human",
            (
                """Use the following results retrieved from a database to provide
a succinct, definitive answer to the user's question.

Respond as if you are answering the question directly.

Results: {results}
Question: {question}"""
            ),
        ),
    ]
)

generate_final_chain = generate_final_prompt | llm | StrOutputParser()

def generate_final_answer(state: OverallState) -> OutputState:
    """
    Decides if the question is related to SAP ERP data.
    """
    logger.info("Starting to generate the final answer.")
    logger.debug("Generating final answer with question: %s and results: %s", state.get("question"), state.get("database_records"))
    final_answer = generate_final_chain.invoke(
        {"question": state.get("question"), "results": state.get("database_records")}
    )
    logger.debug("Final answer generated: %s", final_answer)
    logger.info("Finished generating the final answer.")
    return {"answer": final_answer, "steps": ["generate_final_answer"]}


def guardrails_condition(
    state: OverallState,
) -> Literal["generate_cypher", "generate_final_answer"]:
    logger.debug("Evaluating guardrails_condition with state: %s", state)
    if state.get("next_action") == "end":
        logger.info("Next action is 'end', returning 'generate_final_answer'")
        return "generate_final_answer"
    elif state.get("next_action") == "erp":
        logger.info("Next action is 'erp', returning 'generate_cypher'")
        return "generate_cypher"


def validate_cypher_condition(
    state: OverallState,
) -> Literal["generate_final_answer", "correct_cypher", "execute_cypher"]:
    logger.debug("Evaluating validate_cypher_condition with state: %s", state)
    if state.get("next_action") == "end":
        logger.info("Next action is 'end', returning 'generate_final_answer'")
        return "generate_final_answer"
    elif state.get("next_action") == "correct_cypher":
        logger.info("Next action is 'correct_cypher', returning 'correct_cypher'")
        return "correct_cypher"
    elif state.get("next_action") == "execute_cypher":
        logger.info("Next action is 'execute_cypher', returning 'execute_cypher'")
        return "execute_cypher"
    


logger.info("Initializing StateGraph with OverallState, InputState, and OutputState.")
langgraph = StateGraph(OverallState, input=InputState, output=OutputState)

logger.info("Adding nodes to the StateGraph.")
langgraph.add_node(generate_cypher)
langgraph.add_node(validate_cypher)
langgraph.add_node(correct_cypher)
langgraph.add_node(execute_cypher)
langgraph.add_node(generate_final_answer)

logger.info("Adding edges to the StateGraph.")
langgraph.add_edge(START, "generate_cypher")
langgraph.add_edge("generate_cypher", "validate_cypher")
langgraph.add_conditional_edges(
    "validate_cypher",
    validate_cypher_condition,
)
langgraph.add_edge("execute_cypher", "generate_final_answer")
langgraph.add_edge("correct_cypher", "validate_cypher")
langgraph.add_edge("generate_final_answer", END)

logger.info("Compiling the StateGraph.")
langgraph = langgraph.compile()

logger.info("Displaying the StateGraph as a Mermaid diagram.")

# display(Image(langgraph.get_graph().draw_mermaid_png()))


if __name__ == "__main__":
    sample_questions = ['q1','q2','q3']

    # Create a logs directory if it doesn't exist
    log_dir = "logs"
    os.makedirs(log_dir, exist_ok=True)

    # Create CSV file with timestamp
    csv_filename = os.path.join(log_dir, f"query_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv")

    # Initialize CSV with headers
    with open(csv_filename, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['timestamp', 'question', 'answer', 'status'])

    # Modified question processing loop with logging
    for question in sample_questions:
        print("-"*100)
        print(question)
        
        try:
            # Get response from langgraph
            response = langgraph.invoke({"question": question})
            
            # Log successful response
            with open(csv_filename, 'a', newline='') as f:
                writer = csv.writer(f)
                writer.writerow([
                    datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                    question,
                    response.get('answer', 'No answer provided'),
                    'success'
                ])
            
            print(response)
            
        except Exception as e:
            # Log error case
            logger.error(f"Error processing question: {question}\nError: {str(e)}", exc_info=True)
            
            with open(csv_filename, 'a', newline='') as f:
                writer = csv.writer(f)
                writer.writerow([
                    datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                    question,
                    'agentic error',
                    'error'
                ])
            
            print(f"Error processing question: {str(e)}")
        
        print("\n")

    # Optional: Load and display summary of results
    try:
        df = pd.read_csv(csv_filename)
        print(f"\nProcessing Summary:")
        print(f"Total questions processed: {len(df)}")
        print(f"Successful queries: {len(df[df['status'] == 'success'])}")
        print(f"Failed queries: {len(df[df['status'] == 'error'])}")
        print(f"Results saved to: {csv_filename}")
    except Exception as e:
        logger.error(f"Error generating summary: {str(e)}", exc_info=True)

When I run this script some questions succeed and some other ones yield this error:

response = langgraph.invoke({"question": question})
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/pregel/init.py", line 2669, in invoke
for chunk in self.stream(
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/pregel/init.py", line 2323, in stream
for _ in runner.tick(
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/pregel/runner.py", line 146, in tick
run_with_retry(
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/pregel/retry.py", line 40, in run_with_retry
return task.proc.invoke(task.input, config)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/utils/runnable.py", line 600, in invoke
input = step.invoke(input, config, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/asmaa/babdbenv/lib/python3.12/site-packages/langgraph/utils/runnable.py", line 365, in invoke
ret = context.run(self.func, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/asmaa/brightandbold/bedrock-data/src/querying/graph_querying.py", line 397, in validate_cypher
not [
^
IndexError: list index out of range
During task with name 'validate_cypher' and id 'd783ef79-17a3-2f0c-e2a9-0c8491c017fa'
Error processing question: list index out of range

what is causing this error? and how should I solve it?

[dosubot[bot]]

The IndexError: list index out of range error in the validate_cypher function occurs when the code attempts to access an element in a list that is empty. This typically happens when the code tries to retrieve properties from the graph schema, but the expected properties do not exist for the specified node label.

To resolve this issue, you should add a check to ensure that the list of properties is not empty before attempting to access its elements. Here's a suggested modification to the relevant part of your validate_cypher function:

for filter in llm_output.filters:
    logger.debug(f"Processing filter: {filter}")
    properties = [
        prop
        for prop in graph.structured_schema["node_props"].get(filter.node_label, [])
        if prop["property"] == filter.property_key
    ]
    if not properties:
        logger.warning(f"No properties found for label {filter.node_label} with key {filter.property_key}")
        continue  # Skip to the next filter if no properties are found
    if properties[0]["type"] != "STRING":
        continue
    mapping = graph.query(
        f"MATCH (n:{filter.node_label}) WHERE toLower(n.`{filter.property_key}`) = toLower($value) RETURN 'yes' LIMIT 1",
        {"value": filter.property_value},
    )
    if not mapping:
        logger.warning(
            f"Missing value mapping for {filter.node_label} on property {filter.property_key} with value {filter.property_value}"
        )
        mapping_errors.append(
            f"Missing value mapping for {filter.node_label} on property {filter.property_key} with value {filter.property_value}"
        )

This modification ensures that you only attempt to access the list if it contains elements. If the list is empty, it logs a warning and continues to the next iteration without trying to access an index. This should prevent the IndexError from occurring.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}