Memory

Memory gives your LangGraph agent the ability to recall past interactions, user preferences, and learned facts. There are two distinct kinds: short-term memory scoped to a single thread (conversation), and long-term memory that persists across threads using the LangGraph Store API. injectAgent() surfaces graph state through Angular Signals, so a memory panel needs no manual state wiring. The running example is a chat that learns facts about you as you talk, and this guide walks the three files that make it work.

What the demo does

The Run tab shows the prebuilt <chat> composition beside a sidebar titled Learned Facts, which starts out saying that no facts have been learned yet. Tell the agent something about yourself — your name, where you live, the framework you work in — and the reply streams in first, then the sidebar fills with the facts the agent decided were worth keeping, one key and value per row.

Two things are worth trying. Keep going for a few turns and watch the list grow rather than reset, because each pass merges new facts into the ones already known. Then ask the agent what it remembers about you: the facts in the sidebar are injected into the system prompt of the next reply, so it answers from the same list you are looking at.

The memory in this demo lives in the thread's state, which makes it thread-scoped. Memory that has to follow a user from one conversation to the next is the Store API, further down this page.

How it is built

Three files carry the whole feature: a graph that answers and then extracts, an application config that registers the agent, and a component that derives a signal from the state the graph returns. Open the Code tab to read them in place.

The state the graph carries

The state is the mechanism. Alongside the message list the graph keeps a plain memory dict, and because it is a state field it is checkpointed with the thread and streamed to the browser on every update.

class MemoryState(TypedDict):
    # add_messages appends returned messages instead of replacing the list,
    # so the transcript accumulates across turns.
    messages: Annotated[list[BaseMessage], add_messages]
    memory: dict  # {"user_name": "Alice", "preferences": {...}, ...}

The values are ordinary JSON, which is what lets the Angular side render them without knowing anything about their shape.

Answering with what is already known

The first node reads the current memory dict and folds it into the system prompt as a bulleted list of facts before it calls the model. That is the entire recall mechanism: no retrieval step, no vector index, just state read back into the prompt.

async def generate(state: MemoryState) -> dict:
    """Generate a response using current messages and known memory."""
    system_prompt = (PROMPTS_DIR / "memory.md").read_text()
 
    # Inject known memory facts into the system prompt
    memory = state.get("memory", {})
    if memory:
        facts = "\n".join(f"- {k}: {v}" for k, v in memory.items())
        system_prompt = f"{system_prompt}\n\n## What you remember about the user:\n{facts}"
 
    messages = [SystemMessage(content=system_prompt)] + list(state["messages"])
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

The model is constructed with streaming=True, so the answer arrives token by token while the extraction pass has not yet run.

Extracting new facts

The second node runs after the reply. It builds a transcript from the last few messages, asks a non-streaming model to return only a JSON object of new or updated facts, parses that defensively, and merges the result over what was already known.

    async def extract_memory(state: MemoryState) -> dict:
        """
        Scan the latest exchange for facts worth remembering.
 
        Uses a structured prompt to extract key-value pairs from the
        conversation. Returns an updated `memory` dict merged with any
        facts already known.
        """
        messages = state.get("messages", [])
        current_memory = state.get("memory", {})
 
        # Build a summary of the last few exchanges for extraction.
        recent = list(messages[-4:])
        # Guarantee the newest human turn is in the excerpt, even when the
        # tail of the transcript happens to be assistant output only.
        if not any(getattr(m, "type", None) == "human" for m in recent):
            latest_human = next(
                (m for m in reversed(messages) if getattr(m, "type", None) == "human"),
                None,
            )
            if latest_human is not None:
                recent = [latest_human] + recent
        transcript = "\n".join(
            f"{m.type.upper()}: {m.content}"
            for m in recent
            if hasattr(m, "content") and isinstance(m.content, str)
        )
 
        extraction_prompt = f"""Review this conversation excerpt and extract any facts about the user worth remembering.
Return ONLY a JSON object with string keys and string values. Keys should be snake_case identifiers
(e.g. "user_name", "location", "favorite_color"). Return an empty object {{}} if nothing new is learned.
 
Conversation:
{transcript}
 
Current known facts (do not repeat these unless the value changed):
{json.dumps(current_memory)}
 
Respond with ONLY the JSON object of NEW or UPDATED facts:"""
 
        response = await extractor_llm.ainvoke([SystemMessage(content=extraction_prompt)])
        raw = response.content.strip()
 
        # Safely parse the JSON response
        try:
            # Strip markdown code fences if present
            match = re.search(r"\{.*\}", raw, re.DOTALL)
            new_facts = json.loads(match.group()) if match else {}
        except (json.JSONDecodeError, AttributeError):
            new_facts = {}
 
        if not new_facts:
            return {}
 
        updated_memory = {**current_memory, **new_facts}
        return {"memory": updated_memory}

Two details are load-bearing. The extraction prompt is given the current facts and told not to repeat them, so each pass returns a small delta. And the node returns {} when it learns nothing, which leaves the existing memory value untouched.

Wiring the two nodes

The graph is a straight line: generate, then extract, then end. Putting extraction after the reply keeps it off the path the user waits on.

graph = StateGraph(MemoryState)
graph.add_node("generate", generate)
graph.add_node("extract_memory", extract_memory)
graph.set_entry_point("generate")
graph.add_edge("generate", "extract_memory")
graph.add_edge("extract_memory", END)
return graph.compile()

compile() is called with no checkpointer, because the LangGraph API server provides persistence and either rejects or ignores a graph that brings its own, depending on whether it is langgraph dev or a deployment. The Persistence guide covers the checkpointer choices for a graph you embed in your own process.

The agent provider

provideAgent() registers the agent once for the whole application, and it is the only provider <chat> requires. The example passes a factory because it resolves its connection details at runtime from the host that serves the demo.

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/langgraph';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'langgraph') {
        throw new Error('incompatible runtime');
      }
      return {
        apiUrl: connection.apiUrl,
        assistantId: connection.assistantId,
        clientOptions: connection.clientOptions,
      };
    }),
  ],
};

Your own application does not need the factory. Pass the two values directly:

provideAgent({
  apiUrl: 'https://your-deployment.langgraph.app',
  assistantId: 'memory',
});

assistantId must match the graph name in langgraph.json.

Reading memory with value()

injectAgent() returns the agent registered above, and value() is the full graph state as a signal — messages, memory, and any other field the state schema declares. The component derives the sidebar rows from it with computed(), casting the state to the shape it expects and returning an empty list until a memory object exists.

/**
 * The streaming resource connected to the memory graph.
 *
 * The graph returns a `memory` dict alongside messages in its state.
 * We expose it via `stream.value()` and derive a reactive signal for display.
 */
protected readonly agent = injectAgent();
 
/**
 * Reactive list of [key, value] memory entries derived from the graph state.
 *
 * The Python graph stores learned facts in `state.memory` as a plain dict.
 * This signal re-computes whenever the stream state changes.
 */
protected readonly memoryEntries = computed(() => {
  const val = this.agent.value() as Record<string, unknown>;
  const mem = val?.['memory'];
  if (!mem || typeof mem !== 'object') return [];
  return Object.entries(mem as Record<string, string>);
});

No extra request is involved. The memory dict rides along with the state updates the run is already streaming, so memoryEntries recomputes the moment the extraction node commits.

Rendering the facts

Beside the prebuilt <chat> in the layout's main slot, the sidebar slot holds the fact list. It reads memoryEntries() directly, so nothing subscribes and nothing is torn down by hand.

<div sidebar class="panel">
  <h3 class="cap">Learned Facts</h3>
  @if (memoryEntries().length === 0) {
    <p class="empty">No facts learned yet</p>
  }
  @for (entry of memoryEntries(); track entry[0]) {
    <div class="fact">
      <span class="fact__key">{{ entry[0] }}:</span>
      <span class="fact__value"> {{ entry[1] }}</span>
    </div>
  }
</div>

Designing the state schema

The example keeps one memory field because one is enough to show the mechanism. A larger agent usually splits what it remembers into fields with different lifetimes: a rolling summary that is rewritten every turn, preferences that accumulate, topics that append.

from operator import add
from typing import Annotated, TypedDict
 
from langchain_core.messages import BaseMessage
 
 
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    user_preferences: dict           # Merged by the node that writes it
    conversation_summary: str        # Replaced on every write
    mentioned_topics: Annotated[list[str], add]  # Appended by the reducer

The Annotated[..., add] fields append on every write; the plain fields are replaced by whatever a node returns for them. Choosing the reducer per field is how you decide, once, whether a node adds to memory or overwrites it.

Short-term memory (thread-scoped)

Short-term memory is the kind the demo uses: state fields accumulated inside one thread. State updates are checkpointed, so reconnecting with the same threadId restores the fields the graph wrote, the demo's memory dict among them. The demo declares messages with the add_messages reducer, which is what makes the transcript accumulate the same way rather than being replaced on every turn.

# Same thread, two invocations — the second one sees what the first learned
result = graph.invoke(
    {"messages": [{"role": "user", "content": "I prefer dark mode"}]},
    config={"configurable": {"thread_id": "user_42_session"}},
)
 
result = graph.invoke(
    {"messages": [{"role": "user", "content": "What theme do I like?"}]},
    config={"configurable": {"thread_id": "user_42_session"}},
)

On the Angular side this takes no extra code beyond naming the thread. Pass a threadId to provideAgent(), as a signal when it changes at runtime:

const threadId = signal<string | null>(localStorage.getItem('memory-thread'));
 
provideAgent({
  apiUrl: 'https://your-deployment.langgraph.app',
  assistantId: 'memory',
  threadId,
  onThreadId: (id) => localStorage.setItem('memory-thread', id),
});

agent.messages() restores the full history on reconnect and agent.value() restores every custom state field with it. The Persistence guide covers thread switching and checkpointer configuration in full.

Long-term memory (cross-thread) with the Store API

Thread state disappears when you start a new thread. For knowledge that should outlive a conversation — preferences, learned facts, project context — use the LangGraph Store API. The Store is a key-value layer that any node can read from and write to, independent of the current thread, and a node opts into it by declaring a store parameter.

import json
from uuid import uuid4
 
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.store.base import BaseStore
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-5-mini")
 
 
def recall_memories(state: MessagesState, *, store: BaseStore, config) -> dict:
    """Load long-term memories for this user before responding."""
    user_id = config["configurable"]["user_id"]
 
    # Fetch every memory in this user's namespace
    memories = store.search(("memories", user_id))
    memory_text = "\n".join(f"- {m.value['content']}" for m in memories)
 
    system = (
        "You are a helpful assistant with long-term memory.\n\n"
        f"What you remember about this user:\n{memory_text}"
    )
    response = llm.invoke([
        {"role": "system", "content": system},
        *state["messages"],
    ])
    return {"messages": [response]}
 
 
def save_memories(state: MessagesState, *, store: BaseStore, config) -> dict:
    """Extract and persist new facts to the Store."""
    user_id = config["configurable"]["user_id"]
 
    extraction = llm.invoke([
        {"role": "system", "content": (
            "Extract new facts about the user from the latest "
            "exchange. Return a JSON list of strings. "
            "Return [] if nothing new."
        )},
        *state["messages"][-4:],
    ])
 
    for fact in json.loads(extraction.content):
        store.put(("memories", user_id), key=str(uuid4()), value={"content": fact})
 
    return {}
 
 
builder = StateGraph(MessagesState)
builder.add_node("recall", recall_memories)
builder.add_node("save", save_memories)
builder.add_edge(START, "recall")
builder.add_edge("recall", "save")
builder.add_edge("save", END)
 
graph = builder.compile()

The identity the nodes namespace by has to come from the client. Pass it per run through submit(), which forwards a config to LangGraph:

agent.submit(
  { message: text },
  { config: { configurable: { user_id: 'user_42' } } },
);

Semantic memory with vector search

For agents that accumulate hundreds or thousands of memories, keyword matching is not enough. The Store API supports semantic search with embeddings, so your agent can retrieve the most relevant memories for any given context.

def recall_relevant(state: MessagesState, *, store: BaseStore, config) -> dict:
    """Retrieve memories semantically related to the current question."""
    user_id = config["configurable"]["user_id"]
    query = state["messages"][-1].content
 
    # Vector search — returns memories ranked by cosine similarity
    results = store.search(("memories", user_id), query=query, limit=5)
    memory_text = "\n".join(f"- [{r.score:.2f}] {r.value['content']}" for r in results)
 
    response = llm.invoke([
        {"role": "system", "content": (
            "Relevant memories (similarity score in brackets):\n"
            f"{memory_text}\n\n"
            "Use these memories to personalize your response."
        )},
        *state["messages"],
    ])
    return {"messages": [response]}

The store.search() call accepts a query string and returns results ranked by vector similarity. You control how many results to retrieve with the limit parameter. Each result includes a score field indicating how relevant the memory is to the query.

Memory best practices

What's Next