LangGraph et la mémoire persistante : un état qui survit entre les exécutions (2026)

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

llm = ChatOpenAI(model="gpt-4o-mini")

def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Build graph
builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.set_entry_point("agent")
builder.add_edge("agent", END)

# MemorySaver: thread-level, in-process only
# All checkpoints vanish on process restart
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# Session 1
config = {"configurable": {"thread_id": "user-001-session-1"}}
graph.invoke({"messages": [{"role": "user", "content": "I prefer Python over TypeScript"}]}, config)

# Session 2 — new thread ID, different process restart
# Agent has NO memory of session 1
config2 = {"configurable": {"thread_id": "user-001-session-2"}}
result = graph.invoke({"messages": [{"role": "user", "content": "What language should I use?"}]}, config2)
# Agent cannot recall the Python preference from session 1

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated, Optional
from langchain_openai import ChatOpenAI
import operator
import httpx

KRONVEX_API_KEY = "kv-your-key"
KRONVEX_AGENT_ID = "your-agent-id"

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    memory_context: Optional[str]

llm = ChatOpenAI(model="gpt-4o-mini")

# ── Kronvex helpers ──────────────────────────────────────────────

def remember(content: str, memory_type: str = "fact"):
    httpx.post(
        f"https://api.kronvex.io/api/v1/agents/{KRONVEX_AGENT_ID}/remember",
        headers={"X-API-Key": KRONVEX_API_KEY},
        json={"content": content, "memory_type": memory_type}
    )

def recall(query: str, top_k: int = 5):
    r = httpx.post(
        f"https://api.kronvex.io/api/v1/agents/{KRONVEX_AGENT_ID}/recall",
        headers={"X-API-Key": KRONVEX_API_KEY},
        json={"query": query, "top_k": top_k, "threshold": 0.5}
    )
    return r.json()["results"]

# ── Graph nodes ──────────────────────────────────────────────────

def memory_recall_node(state: AgentState):
    """Recall relevant memories before the LLM responds."""
    latest_message = state["messages"][-1]["content"]
    memories = recall(latest_message)
    if memories:
        context = "\n".join(f"- {m['memory']['content']}" for m in memories)
        return {"memory_context": f"Relevant context from past sessions:\n{context}"}
    return {"memory_context": None}

def call_model(state: AgentState):
    """Call the LLM, injecting memory context into the system prompt."""
    messages = state["messages"].copy()
    if state.get("memory_context"):
        system = {
            "role": "system",
            "content": f"You are a helpful assistant.\n\n{state['memory_context']}"
        }
        messages = [system] + messages
    response = llm.invoke(messages)
    return {"messages": [response]}

def memory_store_node(state: AgentState):
    """Extract and store new facts from the assistant's last response."""
    last_response = state["messages"][-1].content
    # In production: use an LLM to extract structured facts before storing.
    # Here we store the full response as context (simplified example).
    remember(f"Assistant said: {last_response[:500]}", memory_type="event")
    return {}

# ── Build graph ──────────────────────────────────────────────────

builder = StateGraph(AgentState)
builder.add_node("recall", memory_recall_node)
builder.add_node("agent", call_model)
builder.add_node("store", memory_store_node)

builder.set_entry_point("recall")
builder.add_edge("recall", "agent")
builder.add_edge("agent", "store")
builder.add_edge("store", END)

graph = builder.compile(checkpointer=MemorySaver())

# Session 1
config = {"configurable": {"thread_id": "user-001-session-1"}}
graph.invoke({
    "messages": [{"role": "user", "content": "I prefer Python over TypeScript for all my projects"}],
    "memory_context": None
}, config)

# Session 2 — different thread, different process, different day
# Kronvex recalls the Python preference automatically
config2 = {"configurable": {"thread_id": "user-001-session-2"}}
result = graph.invoke({
    "messages": [{"role": "user", "content": "What language should I use for the new microservice?"}],
    "memory_context": None
}, config2)
# Agent now answers: "Based on your past preference for Python..."

import httpx

KRONVEX_API_KEY = "kv-your-key"
KRONVEX_AGENT_ID = "your-agent-id"

def inject_context(query: str, max_tokens: int = 800) -> str:
    """Return a formatted memory block ready for a system prompt."""
    r = httpx.post(
        f"https://api.kronvex.io/api/v1/agents/{KRONVEX_AGENT_ID}/inject-context",
        headers={"X-API-Key": KRONVEX_API_KEY},
        json={"query": query, "max_tokens": max_tokens, "threshold": 0.45}
    )
    data = r.json()
    return data.get("context", "")  # Returns "" if no relevant memories

def memory_recall_node(state: AgentState):
    """Use inject_context to get a fully formatted memory block."""
    latest_message = state["messages"][-1]["content"]
    context = inject_context(latest_message)
    return {"memory_context": context if context else None}

def call_model(state: AgentState):
    """System prompt now includes pre-formatted memory context."""
    messages = list(state["messages"])
    system_content = "You are a helpful assistant."
    if state.get("memory_context"):
        system_content += f"\n\n## Long-term memory\n{state['memory_context']}"
    messages = [{"role": "system", "content": system_content}] + messages
    response = llm.invoke(messages)
    return {"messages": [response]}