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.