LangGraph - FalkorDB Docs

LangGraph is an open-source framework for building stateful, multi-actor agentic applications using LLMs. It allows you to design complex single- and multi-agent workflows as directed graphs, where nodes represent tasks and edges define the information flow.

Resources

Installation

Install LangGraph with required dependencies. The langchain-falkordb package provides the FalkorDB integration, including a LangGraph checkpointer (FalkorDBSaver) via the langgraph extra:

pip install langgraph "langchain-falkordb[langgraph]" langchain-openai

Quick Start

1. Define Your Graph Workflow

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_falkordb import FalkorDBGraph
from langchain_openai import ChatOpenAI

# Define state
class GraphState(TypedDict):
    question: str
    cypher_query: str
    graph_data: str
    answer: str

# Initialize FalkorDB
graph_db = FalkorDBGraph(
    "knowledge_graph",
    host="localhost",
    port=6379,
)

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

2. Create Agent Nodes

def generate_cypher(state: GraphState) -> GraphState:
    """Generate Cypher query from natural language question"""
    question = state["question"]
    schema = graph_db.get_schema

    prompt = f"""
    Given the graph schema:
    {schema}

    Generate a Cypher query to answer: {question}
    """

    response = llm.invoke(prompt)
    state["cypher_query"] = response.content
    return state

def execute_query(state: GraphState) -> GraphState:
    """Execute Cypher query on FalkorDB"""
    cypher = state["cypher_query"]
    result = graph_db.query(cypher)
    state["graph_data"] = str(result)
    return state

def generate_answer(state: GraphState) -> GraphState:
    """Generate natural language answer from graph data"""
    question = state["question"]
    data = state["graph_data"]

    prompt = f"""
    Question: {question}
    Graph Data: {data}

    Provide a clear, natural language answer based on the data.
    """

    response = llm.invoke(prompt)
    state["answer"] = response.content
    return state

3. Build the Workflow

# Create workflow
workflow = StateGraph(GraphState)

# Add nodes
workflow.add_node("generate_cypher", generate_cypher)
workflow.add_node("execute_query", execute_query)
workflow.add_node("generate_answer", generate_answer)

# Define edges
workflow.set_entry_point("generate_cypher")
workflow.add_edge("generate_cypher", "execute_query")
workflow.add_edge("execute_query", "generate_answer")
workflow.add_edge("generate_answer", END)

# Compile the graph
app = workflow.compile()

4. Run the Agent

# Execute the workflow
result = app.invoke({
    "question": "Who are the top 5 customers by purchase amount?",
    "cypher_query": "",
    "graph_data": "",
    "answer": "",
})

print(result["answer"])

Advanced Usage

Multi-Agent GraphRAG System

from langgraph.graph import StateGraph, END

class RAGState(TypedDict):
    question: str
    context: str
    entities: list[str]
    relationships: list[str]
    answer: str
    next_step: str

def extract_entities(state: RAGState) -> RAGState:
    """Extract entities from the question"""
    question = state["question"]

    prompt = f"Extract key entities from: {question}"
    response = llm.invoke(prompt)

    state["entities"] = response.content.split(",")
    state["next_step"] = "find_relationships"
    return state

def find_relationships(state: RAGState) -> RAGState:
    """Find relationships between entities in the graph"""
    entities = state["entities"]

    # Query graph for relationships
    cypher = f"""
    MATCH (a)-[r]->(b)
    WHERE a.name IN {entities} OR b.name IN {entities}
    RETURN a, r, b
    LIMIT 20
    """

    result = graph_db.query(cypher)
    state["relationships"] = result
    state["next_step"] = "retrieve_context"
    return state

def retrieve_context(state: RAGState) -> RAGState:
    """Retrieve relevant context from the graph"""
    entities = state["entities"]

    cypher = f"""
    MATCH (n)
    WHERE n.name IN {entities}
    RETURN n
    """

    result = graph_db.query(cypher)
    state["context"] = str(result)
    state["next_step"] = "generate_answer"
    return state

def generate_final_answer(state: RAGState) -> RAGState:
    """Generate the final answer using all context"""
    question = state["question"]
    context = state["context"]
    relationships = state["relationships"]

    prompt = f"""
    Question: {question}
    Context: {context}
    Relationships: {relationships}

    Provide a comprehensive answer using the graph context.
    """

    response = llm.invoke(prompt)
    state["answer"] = response.content
    state["next_step"] = END
    return state

# Build RAG workflow
rag_workflow = StateGraph(RAGState)

rag_workflow.add_node("extract_entities", extract_entities)
rag_workflow.add_node("find_relationships", find_relationships)
rag_workflow.add_node("retrieve_context", retrieve_context)
rag_workflow.add_node("generate_answer", generate_final_answer)

rag_workflow.set_entry_point("extract_entities")
rag_workflow.add_edge("extract_entities", "find_relationships")
rag_workflow.add_edge("find_relationships", "retrieve_context")
rag_workflow.add_edge("retrieve_context", "generate_answer")
rag_workflow.add_edge("generate_answer", END)

rag_app = rag_workflow.compile()

# Execute
result = rag_app.invoke({
    "question": "What is the relationship between Company A and Person B?",
    "context": "",
    "entities": [],
    "relationships": [],
    "answer": "",
    "next_step": "",
})

print(result["answer"])

Conditional Routing

def should_continue(state: GraphState) -> str:
    """Decide next step based on state"""
    if state.get("cypher_query") and "ERROR" in state["cypher_query"]:
        return "regenerate"
    elif state.get("graph_data") == "[]":
        return "no_results"
    else:
        return "continue"

# Add conditional edges
workflow.add_conditional_edges(
    "execute_query",
    should_continue,
    {
        "regenerate": "generate_cypher",
        "no_results": "handle_no_results",
        "continue": "generate_answer",
    }
)

Memory Integration

FalkorDBSaver (from langchain-falkordb) is a LangGraph checkpointer that persists agent state in FalkorDB, so threads survive restarts and can be shared between processes:

from langchain_falkordb.checkpoint import FalkorDBSaver

# Add persistence backed by FalkorDB
memory = FalkorDBSaver(host="localhost", port=6379)

# Compile with checkpointing
app = workflow.compile(checkpointer=memory)

# Use with thread ID for conversation memory
config = {"configurable": {"thread_id": "user_123"}}

result1 = app.invoke({
    "question": "Who is the CEO?",
    "cypher_query": "",
    "graph_data": "",
    "answer": "",
}, config)

result2 = app.invoke({
    "question": "What companies do they lead?",
    "cypher_query": "",
    "graph_data": "",
    "answer": "",
}, config)

Use Cases

Complex Query Decomposition: Break down complex questions into multiple graph queries
Multi-Step Reasoning: Chain multiple graph operations for advanced analytics
Agentic RAG: Combine retrieval, reasoning, and generation in a graph-based workflow
Error Recovery: Implement retry logic and fallback mechanisms in graph queries
Conversational GraphRAG: Maintain context across multiple turns of conversation

Best Practices

State Management: Keep state minimal and well-typed
Error Handling: Add nodes for handling query errors and edge cases
Conditional Logic: Use conditional edges for dynamic workflow routing
Checkpointing: Enable persistence for long-running workflows
Observability: Add logging at each node for debugging

Resources

Frequently Asked Questions

What is LangGraph and how does it work with FalkorDB?

LangGraph is an open-source framework for building stateful, multi-actor agentic applications using LLMs. With FalkorDB, you can create graph-based workflows where nodes represent tasks (like Cypher generation and query execution) and edges define information flow between them.

How does LangGraph differ from LangChain for FalkorDB?

LangChain provides pre-built chains like FalkorDBQAChain for simple query-answer flows. LangGraph gives you fine-grained control to build custom directed-graph workflows with conditional routing, error recovery, multi-step reasoning, and checkpointed state.

Can I add error recovery to my graph queries?

Yes. Use conditional edges to route failed queries back to a regeneration node. For example, if execute_query returns an error, a should_continue function can route back to generate_cypher for retry with adjusted parameters.

Does LangGraph support conversation memory with FalkorDB?

Yes. Use FalkorDBSaver from langchain-falkordb (install with pip install langchain-falkordb[langgraph]) as a LangGraph checkpointer with a thread_id config to persist conversation state in FalkorDB across multiple invocations and restarts. Agents can then handle follow-up questions that reference previous answers.

What are good use cases for LangGraph with FalkorDB?

Key use cases include complex query decomposition (breaking questions into multiple graph queries), multi-step reasoning, agentic RAG with retrieval-reasoning-generation pipelines, error recovery workflows, and conversational GraphRAG with multi-turn context.

​Resources

​Installation

​Quick Start

​1. Define Your Graph Workflow

​2. Create Agent Nodes

​3. Build the Workflow

​4. Run the Agent

​Advanced Usage

​Multi-Agent GraphRAG System

​Conditional Routing

​Memory Integration

​Use Cases

​Best Practices

​Resources

​Frequently Asked Questions

Resources

Installation

Quick Start

1. Define Your Graph Workflow

2. Create Agent Nodes

3. Build the Workflow

4. Run the Agent

Advanced Usage

Multi-Agent GraphRAG System

Conditional Routing

Memory Integration

Use Cases

Best Practices

Resources

Frequently Asked Questions