LangGraph 与 Redis：使用内存和持久性构建更智能的 AI 代理

from typing import Literal

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.redis import RedisSaver

# Define a simple tool
@tool
def get_weather(city: Literal["nyc", "sf"]):
    """Use this to get weather information."""
    if city == "nyc":
        return "It might be cloudy in nyc"
    elif city == "sf":
        return "It's always sunny in sf"
    else:
        raise AssertionError("Unknown city")

# Set up model and tools
tools = [get_weather]
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Create Redis persistence
REDIS_URI = "redis://localhost:6379"
with RedisSaver.from_conn_string(REDIS_URI) as checkpointer:
    # Initialize Redis indices (only needed once)
    checkpointer.setup()
    
    # Create agent with memory
    graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
    
    # Use the agent with a specific thread ID to maintain conversation state
    config = {"configurable": {"thread_id": "user123"}}
    res = graph.invoke({"messages": [("human", "what's the weather in sf")]}, config)

import uuid

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import RunnableConfig

from langgraph.checkpoint.redis import RedisSaver
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.store.redis import RedisStore
from langgraph.store.base import BaseStore

# Set up model
model = ChatAnthropic(model="claude-3-5-sonnet-20240620")

# Function that uses store to access and save user memories
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)
    
    # Retrieve relevant memories for this user
    memories = store.search(namespace, query=str(state["messages"][-1].content))
    info = "\n".join([d.value["data"] for d in memories])
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
    
    # Store new memories if the user asks to remember something
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        memory = "User name is Bob"
        store.put(namespace, str(uuid.uuid4()), {"data": memory})
    
    # Generate response
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}

# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

# Initialize Redis persistence and store
REDIS_URI = "redis://localhost:6379"
with RedisSaver.from_conn_string(REDIS_URI) as checkpointer:
    checkpointer.setup()
    
    with RedisStore.from_conn_string(REDIS_URI) as store:
        store.setup()
        
        # Compile graph with both checkpointer and store
        graph = builder.compile(checkpointer=checkpointer, store=store)

# First conversation - tell the agent to remember something
config = {"configurable": {"thread_id": "convo1", "user_id": "user123"}}
response = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
    config
)

# Second conversation - different thread but same user
new_config = {"configurable": {"thread_id": "convo2", "user_id": "user123"}}
response = graph.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    new_config
)
# Agent will respond with "Your name is Bob"

# Synchronous API
with RedisSaver.from_conn_string(REDIS_URI) as checkpointer:
    checkpointer.setup()
    # ... use synchronously ...

# Asynchronous API
async with AsyncRedisSaver.from_conn_string(REDIS_URI) as checkpointer:
    await checkpointer.asetup()
    # ... use asynchronously ...

pip install langgraph-checkpoint-redis