Stateful Node Patterns

Stateful Node Design Patterns¶

Stateful nodes are essential for enabling context-aware decision-making and dynamic workflow evolution in agentic systems. By persisting and leveraging state across interactions, nodes can adapt to user intent, system context, and evolving requirements. This section explores core design patterns for implementing stateful nodes in LangGraph workflows.

1. Memory-Driven State Stores¶

Use a structured memory store (e.g., dictionaries, databases) to persist and retrieve state across node invocations. This pattern is ideal for chatbots, recommendation systems, or task management workflows.

Example: Chatbot Context Preservation¶

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class ChatState(TypedDict):
    history: list[str]
    user_intent: str

def respond_to_user(state: ChatState):
    # Update state with new user input
    state["history"].append("User: Hello")
    state["user_intent"] = "greeting"
    return {"response": "Hi there!"}

def route_to_next_step(state: ChatState):
    if state["user_intent"] == "greeting":
        return "respond_to_user"
    else:
        return END

workflow = StateGraph(ChatState)
workflow.add_node("respond_to_user", respond_to_user)
workflow.add_node("route_to_next_step", route_to_next_step)
workflow.set_entry_point("respond_to_user")
workflow.add_edge("respond_to_user", "route_to_next_step")
workflow.add_edge("route_to_next,step", END)

app = workflow.compile()

Diagram¶

[START] --> "respond_to_user" --> "route_to_next_step" --> [END]

2. Context-Aware Decision Trees¶

Embed conditional logic within nodes to alter behavior based on stored state. This enables workflows to "learn" from past interactions.

Example: Customer Support Routing¶

def handle_support_request(state: dict):
    if state.get("issue_type") == "billing":
        return {"next_node": "billing_team"}
    elif state.get("issue_type") == "technical":
        return {"next_node": "tech_support"}
    else:
        return {"next_node": "default_team"}

Diagram¶

[START] --> "handle_support_request" --> 
    /               \
  billing_team   tech_support   \
                     \          \
                   default_team

3. Dynamic Workflow Evolution¶

Allow nodes to modify the workflow graph at runtime based on state. This is useful for adaptive systems like personalized recommendation engines.

Example: Adaptive Recommendation Engine¶

def update_recommendations(state: dict):
    if state.get("user_preference") == "sports":
        return {"next_node": "sports_news_feed"}
    else:
        return {"next_node": "general_news_feed"}

Diagram¶

[START] --> "update_recommendations" --> 
    /                        \
sports_news_feed       general_news_feed

4. State Aggregation with External Stores¶

Integrate with vector databases (e.g., FAISS, Pinecone) or MLflow to persist state for long-term use. This is critical for systems requiring historical data access.

Example: RAG System with Vector Store¶

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

def retrieve_context(state: dict):
    embeddings = OpenAIEmbeddings()
    vector_store = FAISS.load_local("vector_store", embeddings)
    docs = vector_store.similarity_search(state["query"])
    return {"context": docs}

Key takeaways¶

Memory stores (dictionaries, databases) enable persistent state across interactions.
Conditional routing allows nodes to adapt behavior based on stored context.
Dynamic workflow evolution lets systems modify their structure at runtime.
External integration with vector databases or MLflow ensures scalable state management.
Stateful patterns are foundational for building intelligent, context-aware agentic workflows.