langgraph-checkpoint-redis

import os from typing import Annotated from langgraph.graph import StateGraph, START from langgraph.checkpoint.redis import RedisSaver from langchain_core.messages import BaseMessage, HumanMessage, AIMessage # Set up Redis connection string, e.g., 'redis://localhost:6379/0' REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0') # Define your graph state class AgentState: messages: Annotated[list[BaseMessage], lambda x, y: x + y] # Define a simple node def simple_agent_node(state: AgentState) -> dict: new_message = AIMessage(content=f"Echo: {state.messages[-1].content}") return {"messages": [new_message]} # Create the Redis Checkpoint Saver # Make sure Redis is running and accessible at REDIS_URL # For production, ensure RedisJSON and RediSearch modules are enabled (Redis 8.0+ includes them) with RedisSaver.from_conn_string(REDIS_URL) as checkpointer: # Important: Call setup() to initialize required RediSearch indices checkpointer.setup() # Build the graph workflow = StateGraph(AgentState) workflow.add_node("agent", simple_agent_node) workflow.set_entry_point(START) workflow.set_finish_point("agent") # Compile the graph with the checkpointer app = workflow.compile(checkpointer=checkpointer) # Example usage with a specific thread_id for persistence thread_id = "test_conversation_123" config = {"configurable": {"thread_id": thread_id}} print(f"\n--- Invoking agent for thread: {thread_id} ---") inputs = {"messages": [HumanMessage(content="Hello LangGraph with Redis!")]} result = app.invoke(inputs, config) print("Result 1:", result) print(f"\n--- Invoking agent again for the same thread: {thread_id} ---") inputs = {"messages": [HumanMessage(content="How are you doing?")]} result = app.invoke(inputs, config) print("Result 2:", result) # List checkpoints (optional) print(f"\n--- Checkpoints for thread {thread_id} ---") for checkpoint in checkpointer.list(config): print(checkpoint)