Build semantic memory with LangGraph

Introduction

When you wire an agent to a single "remember-everything" function, extraction, deduplication, and recall collapse into one tangled call — and the first time a retrieval bug corrupts what gets stored, you cannot tell which step failed. LangGraph fixes this by turning each memory operation into a typed node in a directed graph, so extract, store, and retrieve become separately testable, separately swappable, and separately observable. By the end of this lesson you will be able to model a semantic-memory pipeline as a StateGraph with an explicit MemoryState, wire extraction and storage into a write-path graph, wire retrieval into a read-path graph, and reason about which graph to invoke at which point in the agent loop.

Key Terminology

• Semantic memory — the agent's long-term store of facts, preferences, and entities distilled from prior conversations; distinct from short-term scratchpad/working memory because it persists across sessions and is queried by meaning rather than recency.
• StateGraph — LangGraph's typed directed graph primitive; you declare a TypedDict state schema once, then every node reads and returns partial updates against that schema, which is what makes the memory pipeline composable instead of monolithic.
• Memory node — a single Python callable registered on the graph (extract, store, retrieve) that owns one responsibility and returns a partial state dict; isolating each step here is what lets you unit-test extraction without touching the store.
• Write-path graph — a compiled graph that runs after a turn to learn (extract → store); kept separate from the read path so a slow LLM extraction call never blocks the user-visible response.
• Read-path graph — a compiled graph that runs before a turn to recall (retrieve only); intentionally lightweight so it can sit on the latency-critical path without dragging in extraction cost.

Concepts

Modeling memory as a typed state graph

LangGraph's contract is that nodes communicate exclusively through a typed state object. For semantic memory that means defining a single MemoryState TypedDict whose fields cover everything any node might read or write: the raw conversation going in, the structured extracted_memories produced by the extractor, the stored_count reported by the writer, the retrieved_context produced by the reader, and a stage marker for observability. Each node returns a partial dict; LangGraph merges it into state. This is what lets you add a new node (say, a re-ranker between retrieve and END) without changing any existing node signature — every concept here is implemented in Code Walkthrough.

Splitting the write path from the read path

A common mistake is to wire extract → store → retrieve → END as one graph and invoke it every turn. That couples two operations that have different latency budgets and different invocation cadences: extraction is expensive and only needs to run after the assistant has replied, while retrieval is cheap and needs to run before the assistant generates anything. The fix is two compiled graphs sharing one MemoryState schema — a write-path graph (extract → store) you fire post-turn, and a read-path graph (retrieve) you fire pre-turn. The diagram below shows both paths explicitly; note that no single graph chains all three nodes.

Node responsibilities, in one sentence each

• extract_memories_node — prompt an LLM with the conversation, parse out {fact, type, entities, importance} items, return {"extracted_memories": [...], "stage": "extract"}.
• store_memories_node — iterate the extracted items, dedupe against the existing store, persist the new ones, return {"stored_count": n, "stage": "store"}.
• retrieve_memories_node — pull query entities from the current conversation, search the store, format hits as a context string, return {"retrieved_context": "...", "stage": "retrieve"}.

Because each node only reads the fields it needs and only returns the fields it produces, you can swap (for example) a vector-store-backed store_memories_node for a SQL-backed one without touching the extractor or the retriever.

Code Walkthrough

The snippet below demonstrates all three concepts from the previous section at once: the shared MemoryState schema, the three single-responsibility nodes, and the two separately compiled graphs for the write and read paths.

Code snippetpython
1from langgraph.graph import StateGraph, START, END
2from typing import TypedDict, List
3
4class MemoryState(TypedDict):
5    conversation: str
6    extracted_memories: List[dict]
7    stored_count: int
8    retrieved_context: str
9    stage: str
10
11def extract_memories_node(state: MemoryState) -> dict:
12    prompt = f"""Extract important memories from this conversation.
13For each memory return: key fact, type (fact|preference|event),
14entities mentioned, importance (1-10).
15
16Conversation: {state["conversation"]}"""
17    memories = parse_extraction_response(llm_call(prompt))
18    return {"extracted_memories": memories, "stage": "extract"}
19
20def store_memories_node(state: MemoryState) -> dict:
21    stored = 0
22    for memory in state["extracted_memories"]:
23        if not is_duplicate(memory):
24            save_to_store(memory)
25            stored += 1
26    return {"stored_count": stored, "stage": "store"}
27
28def retrieve_memories_node(state: MemoryState) -> dict:
29    entities = extract_query_entities(state["conversation"])
30    relevant = search_memories(entities)
31    return {
32        "retrieved_context": format_memory_context(relevant),
33        "stage": "retrieve",
34    }
35
36def build_write_graph():
37    g = StateGraph(MemoryState)
38    g.add_node("extract", extract_memories_node)
39    g.add_node("store", store_memories_node)
40    g.add_edge(START, "extract")
41    g.add_edge("extract", "store")
42    g.add_edge("store", END)
43    return g.compile()
44
45def build_read_graph():
46    g = StateGraph(MemoryState)
47    g.add_node("retrieve", retrieve_memories_node)
48    g.add_edge(START, "retrieve")
49    g.add_edge("retrieve", END)
50    return g.compile()
51
52write_graph = build_write_graph()
53read_graph = build_read_graph()
54
55# Pre-turn: recall.
56ctx = read_graph.invoke({"conversation": user_msg})["retrieved_context"]
57# ... generate the assistant reply using `ctx` ...
58# Post-turn: learn.
59write_graph.invoke({"conversation": user_msg + "\n" + assistant_reply})

You'll know it works when a single user turn produces two graph invocations — one before the LLM call returning a non-empty retrieved_context, and one after returning a stored_count >= 0 — and each invocation only touches the nodes registered on its graph (verify with langgraph tracing or by logging state["stage"] inside every node).

Do's and Don'ts

Having walked through building semantic memory with LangGraph above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do define one MemoryState TypedDict and reuse it across both graphs — a single schema is what keeps the write path and read path swappable without rewriting nodes.
2. ✓Do compile separate write-path and read-path graphs — extraction is expensive and runs post-turn; retrieval is cheap and runs pre-turn; one graph can't serve both budgets.
3. ✓Do return partial state dicts from every node — only emit the keys you produced (stage plus your output) so LangGraph's merge semantics stay predictable.

Don'ts

1. ✗Don't chain extract → store → retrieve in a single graph — it forces extraction onto the latency-critical read path and couples failure modes that should be isolated.
2. ✗Don't put deduplication logic inside the extractor — keep it in store_memories_node so you can swap stores (vector, SQL, hybrid) without rewriting extraction prompts.
3. ✗Don't skip the stage field — it's the cheapest observability you'll ever add, and you'll need it the first time a node silently no-ops in production.