> ## Documentation Index
> Fetch the complete documentation index at: https://new.docs.falkordb.com/llms.txt
> Use this file to discover all available pages before exploring further.

# LangGraph

> Build stateful, multi-actor agentic applications with LangGraph and FalkorDB.

[LangGraph](https://www.langgraph.dev/) 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

* 📓 [Blog: Implementing GraphRAG with FalkorDB, LangChain & LangGraph](https://www.falkordb.com/blog/graphrag-workflow-falkordb-langchain/)
* 🔗 [LangGraph Documentation](https://www.langgraph.dev/)

## Installation

Install LangGraph with required dependencies. The [`langchain-falkordb`](https://pypi.org/project/langchain-falkordb/) package provides the FalkorDB integration, including a LangGraph checkpointer (`FalkorDBSaver`) via the `langgraph` extra:

```bash theme={null}
pip install langgraph "langchain-falkordb[langgraph]" langchain-openai
```

## Quick Start

### 1. Define Your Graph Workflow

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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

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

## Resources

* [LangGraph Documentation](https://langchain-ai.github.io/langgraph/)
* [langchain-falkordb on PyPI](https://pypi.org/project/langchain-falkordb/)
* [langchain-falkordb Repository](https://github.com/FalkorDB/langchain-falkordb)
* [Blog: Implementing GraphRAG with FalkorDB, LangChain & LangGraph](https://www.falkordb.com/blog/graphrag-workflow-falkordb-langchain/)
* [LangGraph GitHub Repository](https://github.com/langchain-ai/langgraph)

## Frequently Asked Questions

<AccordionGroup>
  <Accordion title="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.
  </Accordion>

  <Accordion title="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.
  </Accordion>

  <Accordion title="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.
  </Accordion>

  <Accordion title="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.
  </Accordion>

  <Accordion title="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.
  </Accordion>
</AccordionGroup>