Curious builder0 XP earned · 300 to level 2

0 daysFinish a lesson to begin

Badge collection0 of 6 unlocked

18 small wins to finish your path

Support assistant that remembers customers

The final project combines an agent that searches memory while it answers, a background store manager that writes memories after each chat, and tests.

An agent that searches, a manager that saves

Exampleassistant.py

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langmem import ReflectionExecutor, create_memory_store_manager, create_search_memory_tool

from memory_model import MemoryModel, embed

NAMESPACE = ("memories", "{user_id}")


def build(model):
    store = InMemoryStore(index={"dims": 64, "embed": embed})
    agent = create_agent(
        model,
        tools=[create_search_memory_tool(namespace=NAMESPACE)],
        store=store,
        checkpointer=InMemorySaver(),
    )
    reflector = ReflectionExecutor(create_memory_store_manager(model, namespace=NAMESPACE, store=store), store=store)
    return agent, reflector, store


def chat(agent, reflector, user_id, thread_id, text, delay=0):
    config = {"configurable": {"user_id": user_id, "thread_id": thread_id}}
    result = agent.invoke({"messages": [{"role": "user", "content": text}]}, config=config)
    saved = reflector.submit({"messages": result["messages"]}, config=config, after_seconds=delay)
    return result["messages"][-1].content, saved

The agent only gets search_memory. Deciding what to save is the background manager's job, so answers are not slowed down by saving.
build takes the model, so tests and the real app can pass different ones.
chat answers, then submits the whole thread's messages for reflection and returns the future with the reply. In production delay would be minutes and the app would not wait for the future; the tests wait, so they can check the store.

Monday's chat, Friday's question

Examplemain.py

from assistant import build, chat
from memory_model import MemoryModel

agent, reflector, store = build(MemoryModel())
with reflector:
    for text in ["My name is Asha. Please email me.", "Order A-1001 arrived broken."]:
        reply, saved = chat(agent, reflector, "asha", "monday", text, delay=1)
        print("assistant:", reply)
    saved.result()
    for memory in sorted(item.value["content"]["content"] for item in store.search(("memories", "asha"))):
        print("remembered:", memory)

    reply, saved = chat(agent, reflector, "asha", "friday", "What do you know about how to contact me?")
    print("assistant:", reply)

Example

python main.py

On Monday the agent had nothing to search and gave its fallback reply to both messages. Both chats were in the thread monday, submitted with a one-second delay, so the second submit replaced the first: one extraction ran, over the whole thread, and found three memories. saved.result() waited for it. On Friday, in a new thread, the question was answered by searching the store.

Answer now, remember afterwards

Hover or tap a piece to see what it is and which lesson built it.

Follow a chat

Testing memory writes and namespaces

Exampletest_assistant.py

from assistant import build, chat
from memory_model import MemoryModel


def memories(store, user_id):
    return sorted(item.value["content"]["content"] for item in store.search(("memories", user_id)))


def test_memories_are_extracted_after_the_chat():
    agent, reflector, store = build(MemoryModel())
    with reflector:
        _, saved = chat(agent, reflector, "asha", "t1", "My name is Asha. Please email me.")
        saved.result()
    assert memories(store, "asha") == ["The customer's name is Asha", "Wants us to email them"]


def test_a_new_preference_replaces_the_old_one():
    agent, reflector, store = build(MemoryModel())
    with reflector:
        chat(agent, reflector, "asha", "t1", "Please email me.")[1].result()
        chat(agent, reflector, "asha", "t2", "Actually, text me.")[1].result()
    assert memories(store, "asha") == ["Wants us to text them"]


def test_customers_do_not_share_memories():
    agent, reflector, store = build(MemoryModel())
    with reflector:
        chat(agent, reflector, "asha", "t1", "Please email me.")[1].result()
        chat(agent, reflector, "ravi", "t2", "Please call me.")[1].result()
    assert memories(store, "asha") == ["Wants us to email them"]
    assert memories(store, "ravi") == ["Wants us to call them"]

Example

pytest -q -p no:warnings

-p no:warnings hides deprecation warnings that LangMem's dependencies print under Python 3.14. The tests check the parts a model change could break silently: memories are written after a chat, a new preference replaces the old one instead of piling up, and namespaces keep customers apart. Each test waits on the futures, then leaves the with block, which stops the worker thread.

Using Claude or GPT for extraction

Pass init_chat_model("anthropic:claude-sonnet-4-5") to build and use a real embedding model in the store's index. Keep the stand-in in the tests: they check your wiring, and stay fast and free. Measure what a real model extracts with a small set of conversations and expected memories, the way the DeepEval course measures answers.

Memory features for later

Topic	What it is for
Episodic memory	Schemas that record how a past interaction was handled, as examples for next time.
create_memory_searcher	A runnable that writes a search query from a conversation and searches the store.
Multi-prompt optimizer	Improving several prompts of a multi-agent system from shared feedback.
PostgresStore	A persistent store with the same API.
SummarizationNode	The summary as a node inside a LangGraph graph.
LangGraph Platform	Hosted stores and background runs for memory.

PreviousPrompt optimization: learning from feedback

You understood something today that you didn't yesterday.