Understanding and implementing Long-Term Memory in Chatbots and AI Agents.

Humans can speak because we have the physical mechanisms required for speech — our mouth, tongue, vocal cords, and auditory system. But meaningful conversation requires much more than the ability to produce and understand words. That’s where we require memory.

This also applies to Large Language Models which have made enormous progress in generating and understanding language. They can reason over context, answer questions, write code. Thus, researchers thought that LLMs had now learned to “speak”, but to make them truly capable of having conversations, they required memory.

This is where the concept of memory came into LLM applications. Two types of memory were introduced — Short-Term Memory and Long-Term Memory. Similar to what we have, whenever we communicate with any person, we simultaneously use both: the context of what we are talking about in this particular conversation- Short Term Memory and the important points from conversations that we had earlier in the past, that’s Long Term Memory. These memories are stored separately for each person (user).

There are different ways to implement short-term memory. One common approach is to take the previous messages from the conversation and feed them back to the LLM along with the current user message. This makes the LLM aware of the previous context. However, there is a problem with this approach: longer conversations become difficult to maintain as they can exceed the context window, increase the number of input tokens, and consequently increase the cost of inference.

To overcome this problem we usually truncate the chat history or feed the summary of the chat history.

But short-term memory alone is not enough if we want an AI agent to have conversations over a long period of time. The moment a conversation ends, the context associated with that conversation is no longer available to the LLM in a new conversation. If the user comes back tomorrow and opens a new chat and asks something related to a previous conversation, the agent would have no knowledge of what was discussed earlier unless we explicitly provide that information again.

This is where Long-Term Memory comes into the picture.

Instead of keeping the entire conversation history and sending it back to the LLM every time, we can identify and store the information that is actually important. This could include a user’s preferences, personal information, goals, past decisions, interests, or facts learned during previous interactions.

For example, if a user tells an agent, “I prefer Python over Java,” there is no reason to repeatedly send the entire conversation containing this statement. We can extract this information as a memory and store it separately for that particular user.

Later, when the same user asks, “Which programming language should I use for this project?”, the agent can retrieve the relevant memory and use it along with the current conversation to generate a more contextual response.

This gives us a simple distinction:

Short-Term Memory remembers the current conversation. Long-Term Memory remembers what matters across conversations.

The challenge, however, is no longer simply storing information. We also need to decide what is worth remembering, how to represent it, where to store it, and how to retrieve the right memories when they are needed.

Now that we understand why long-term memory is required, the next question is: how do we actually implement it?

A long-term memory system needs to solve three problems. First, it needs to identify what information from a conversation is worth remembering. Second, it needs to store that information in a form that can be efficiently searched. Finally, when the user returns with a new query, it needs to retrieve only the memories that are relevant to the current conversation.

How Long Term Memory Works.

The Architecture

Before we get into code, let’s look at how the pieces fit together. There are four components, each handling one part of the pipeline.

1. An embedding model

Converts text into vectors, so we can measure how similar two pieces of text are. “I love Python” and “Python is my favorite language” end up close together in vector space even though the wording is different. I’m using Qwen3 Embedding, running locally.

2. A vector database

Where memories actually live. Each entry stores the vector (for search) plus the payload: the memory text, user id, categories, and a timestamp. I’m using Qdrant Cloud.

3. A response agent

Talks to the user directly. It gets the conversation transcript and the latest question, and has a tool available to search memory if it needs more context. It decides on its own whether a lookup is worth doing and which categories to search, rather than querying memory on every message.

4. A memory update agent

Runs after each turn, in the background. It checks whether anything from the latest exchange is worth remembering, and if so, whether to add a new memory, update an existing one, delete an outdated one, or do nothing.

Let’s Implement Long Term Memory:

1. Generating Embeddings:

#generate_embeddings.py

from sentence_transformers import SentenceTransformer

import numpy as np

model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")

def generate_embeddings(strings: list[str]):
    # print(strings)

    embeddings = model.encode(
        strings,
        convert_to_numpy=True
    )

    # print(embeddings.shape)
    # print(embeddings)

    return embeddings

if __name__=="__main__":
    texts=[
        "Hello how are you",
        "I like machine learning"
    ]
    embeddings = generate_embeddings(texts)

The function takes a list of strings and returns a list of vectors, one per string. That’s it. We load Qwen3-Embedding-0.6B once at import time so it doesn't reload on every call, and model.encode() handles the batching internally, so passing multiple strings at once is more efficient than calling it in a loop. Every time a user provides a message, it is converted into an embedding.

The vector size matters downstream. This model outputs 1024-dimensional vectors, which is why the Qdrant collection was created with VectorParams(size=1024, ...) in vector_db.py. If you swap the embedding model, you have to update that number too, or Qdrant will reject the inserts.

2. The Storage Schema.

Before looking at how memories get saved or searched, it helps to define what a memory actually looks like. This file uses two separate models for that, and the split is intentional.

class EmbeddedMemory(BaseModel):
    user_id: int
    memory_text: str
    categories: list[str]
    date: str
    embedding: list[float]


class RetrievedMemory(BaseModel):
    point_id: str
    user_id: int
    memory_text: str
    categories: list[str]
    date: str
    score: float

EmbeddedMemory is what you build before writing to the database. It doesn't need a point_id since Qdrant assigns that on insert, and it doesn't have a score since that only makes sense in the context of a search result.

RetrievedMemory is what comes back when you search. It has a point_id so later operations can update or delete that exact record, and a score representing how close it was to the search query.

3. Storing, inserting and searching of Memories.

async def create_memory_collection():
    if not (await client.collection_exists("memories")):
        await client.create_collection(
            collection_name="memories",
            vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
        )
        await client.create_payload_index(
            collection_name="memories",
            field_name="user_id",
            field_schema=models.PayloadSchemaType.INTEGER
        )
        await client.create_payload_index(
            collection_name="memories",
            field_name="categories",
            field_schema=models.PayloadSchemaType.KEYWORD
        )

The vector size (1024) has to match whatever the embedding model outputs, which is why this number is tied directly to the choice made in Step 1. We use cosine distance since it’s the standard choice for text embeddings.

Get Rajattsharma’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The two payload indexes matter for performance. Every search in this system is scoped to a specific user, and often scoped to specific categories too. Without an index on user_id and categories, Qdrant would have to scan every vector in the collection and filter afterward. With the index, it can filter first.

async def insert_memories(
        memories: list[EmbeddedMemory]
):
    await client.upsert(
        collection_name = COLLECTION_NAME,
        points = [
            models.PointStruct(
                id = uuid4().hex,
                payload={
                    "user_id":memory.user_id,
                    "categories":memory.categories,
                    "memory_text":memory.memory_text,
                    "date":memory.date
                },
                vector = memory.embedding
            )
            for memory in memories
        ]
    )

Insertion is straightforward, each memory becomes a point with a random UUID as its ID, the embedding as the vector, and everything else as payload.

async def search_memories(search_vector, user_id, categories=None):
    must_condition = [
        models.FieldCondition(key="user_id", match=models.MatchValue(value=user_id))
    ]
    if categories is not None and len(categories) > 0:
        must_condition.append(
            models.FieldCondition(key="categories", match=models.MatchAny(any=categories))
        )

    outs = await client.query_points(
        collection_name=COLLECTION_NAME,
        query=search_vector,
        with_payload=True,
        query_filter=Filter(must=must_condition),
        score_threshold=0.1,
        limit=2
    )
    return [convert_retrieved_records(p) for p in outs.points if p is not None]

Search always filters by user_id, since memories are private to each user. Category filtering is optional, if the caller passes categories, we narrow the search to just those, otherwise we search across everything the user has. The score_threshold and limit here are tunable, a low threshold favors recall over precision, and you'd raise it if you start seeing irrelevant memories get pulled in.

The one clever bit: listing categories without hardcoding them

async def get_all_categories(user_id):
    facet_filter = Filter(
        must=[models.FieldCondition(key="user_id", match=models.MatchValue(value=user_id))]
    )
    facet_result = await client.facet(
        collection_name=COLLECTION_NAME,
        key="categories",
        facet_filter=facet_filter,
        limit=1000
    )
    return [hit.value for hit in facet_result.hits]

This uses Qdrant’s facet operation to return every distinct category a given user's memories fall under, without us having to maintain a separate list of categories somewhere. This is what lets the response agent see what categories exist and choose to search within specific ones, rather than always searching the entire database. As new categories get created over time, this function picks them up automatically.

4. Retrieval and Response

This is the part of the system the user actually talks to. It’s built using dspy.ReAct, which lets the model decide for itself whether it needs to call a tool before answering, rather than us hardcoding a "always search memory first" step.

class ResponseGenerator(dspy.Signature):
    """
    You will be given a past conversation transcript between user and an AI agent. Also the latest question by the user.

    You have the option to look up the past memories from a vector database to fetch relevant context if required. If you can't find the answer to user's question from transcript or from your own internal knowledge, use the provided search tool calls to search for information.

    You are also provided a list of existing categories in the memory database to quickly search across categories. You can select multiple categories as a list to do your searches. If you select no categories (keep it empty). If you keep categories as empty, we will simply search across the entire database - that is fine too.

    The retrieved information may or may not contain the information user wants. React accordingly.

    You must output the final response, and also decide the latest interaction needs to be recorded into the memory database. New memories are meant to store new information that the user provides.

    While responding, you must be aware that you are continuously learning new memories about the user, so if retrieved memories do not directly address the user's question, mention what you know, acknowledge the gaps in your knowledge, and ask the user for information.

    If you retrieved records using the search tools, and the information was already present, no need to save a new memory. Only save memory if the new information is richer than what you retrieved or didn't find.

    New memories should be made when the USER provides new info. It is not to save information about the the AI or the assistant.
    """

    transcript: list[dict] = dspy.InputField()
    existing_categories: list[str] = dspy.InputField()
    question: str = dspy.InputField()
    response: str = dspy.OutputField()
    save_memory: bool = dspy.OutputField(
        description="True if a new memory record needs to be created for the latest interaction"
    )

A dspy.Signature defines the shape of the task: what goes in, what comes out. Here, the inputs are the conversation so far, the list of categories that exist in the database, and the current question. The outputs are the actual reply, and a boolean flag saying whether this interaction is worth remembering.

async def fetch_similar_memories(search_text: str, categories: list[str]):
    """
    Search the user's stored memories for information semantically similar
    to the given search text.

    Args:
        search_text: The information or context to search for in the user's
            memories. Use a concise description of what you want to retrieve.
        categories: Optional list of memory categories to restrict the search.
            Pass an empty list [] when no specific categories apply; this
            searches across all memory categories.

    Returns:
        A dictionary containing the retrieved memories as formatted strings.
    """

    search_vector = generate_embeddings([search_text])[0]
    memories = await search_memories(
        search_vector,
        user_id=user_id,
        categories=None if len(categories) == 0 else categories,
    )
    memories_str = [stringify_retrieved_point(m_) for m_ in memories]
    return {"memories": memories_str}

This function is passed to the agent as a tool, not called directly by us. The docstring inside it is what the model reads to decide when and how to use it, that’s why it explicitly explains what search_text and categories mean. If the model decides no categories apply, it passes an empty list, and search_memories treats that as "search everything."

response_generator = dspy.ReAct(
    ResponseGenerator, tools=[fetch_similar_memories], max_iters=2
)

max_iters=2 caps how many times the agent can call the tool before it has to produce a final answer. This matters in practice, without a cap, a ReAct agent can occasionally loop searching for something that isn't there. Capping it at 2 keeps latency and cost predictable.

The main loop

out = await response_generator.acall(
    transcript=past_messages,
    question=question,
    existing_categories=existing_categories,
)
response = out.response

past_messages.extend([
    {"role": "user", "content": question},
    {"role": "assistant", "content": response},
])

if out.save_memory:
    update_result = await update_memories(
        user_id=user_id,
        messages=past_messages[-6:],
    )
    existing_categories = await get_all_categories(user_id=user_id)

Each turn appends to past_messages, so the transcript grows across the conversation. When save_memory comes back true, we call the update agent with just the last few messages, not the entire transcript, since only the most recent exchange is likely to contain anything new. After an update, we refresh existing_categories, so if a new category was just created, the response agent can see it and use it on the very next turn.

5. The Memory Update Agent.

a. Deciding What to Remember

This is the part that keeps the memory store from turning into an ever-growing pile of duplicate and contradictory facts. It’s a separate dspy.ReAct agent whose only job is to look at a new exchange, compare it against similar existing memories, and decide what to do.

class UpdateMemorySignature(dspy.Signature):
    """
    You will be given the conversation between user and assistant and some
    similar memories from the database. Your goal is to decide how to combine
    the new memories into the database with the existing memories.

    Actions meaning:
    - ADD: add new memories into the database as a new memory
    - UPDATE: update an existing memory with richer information.
    - DELETE: remove memory items from the database that aren't required anymore
    - NOOP: No need to take any action
    """
    messages: list[dict] = dspy.InputField()
    existing_memories: list[MemoryWithIds] = dspy.InputField()
    summary: str = dspy.OutputField(description="Very short summary of what you did")

Notice there’s no response output here, unlike the response agent. This agent doesn't talk to the user at all, it just acts, using tools, and reports back a short summary of what it did.

async def add_memory(memory_text: str, categories: list[str]) -> str:
    embeddings = generate_embeddings([memory_text])
    await insert_memories(memories=[EmbeddedMemory(
        user_id=user_id, memory_text=memory_text, categories=categories,
        date=datetime.now().strftime("%Y-%m-%d %H:%m"), embedding=embeddings[0],
    )])
    return f"Memory: '{memory_text}' was added to DB"

add_memory handles genuinely new information, embed the text, store it.

async def update(memory_id: int, updated_memory_text: str, categories: list[str]):
    point_id = get_point_id_from_memory_id(memory_id)
    await delete_records([point_id])

    embeddings = generate_embeddings([updated_memory_text])
    await insert_memories(memories=[EmbeddedMemory(
        user_id=user_id, memory_text=updated_memory_text, categories=categories,
        date=datetime.now().strftime("%Y-%m-%d %H:%m"), embedding=embeddings[0],
    )])
    return f"Memory {memory_id} has been updated to: '{updated_memory_text}'"

update is really delete-then-reinsert under the hood, Qdrant doesn't support editing a vector in place cleanly here, so the old point is removed and a new one takes its place with the revised text and a fresh embedding. This is the action that fires when the user says something that refines or corrects an existing memory, for example going from "interested in Japan" to "booked a trip to Tokyo in March."

async def delete(memory_ids: list[int]):
    point_ids = [get_point_id_from_memory_id(mid) for mid in memory_ids]
    await delete_records(point_ids)
    return f"Memory {memory_ids} deleted"

delete removes memories that are now false or irrelevant, the "switched from Python to Rust" case from earlier.

async def noop():
    return "No action done"

noop exists so the agent has an explicit way to say nothing needs to change, rather than being forced to take some action just because it was called.

Why memory_id here is a local index, not a database ID?

memory_ids = [
    MemoryWithIds(memory_id=idx, memory_text=m.memory_text, memory_categories=m.categories)
    for idx, m in enumerate(existing_memories)
]

The agent reasons about memories using small integer indexes (0, 1, 2…) instead of the actual Qdrant point IDs, which are long UUID strings. This is a small but useful detail, it’s easier and less error-prone for a model to reference “memory 2” than to copy a UUID exactly. The get_point_id_from_memory_id helper is what translates that index back into the real point ID when a tool actually needs to touch the database.

async def update_memories(user_id: int, messages: list[dict]):
    latest_user_message = [x["content"] for x in messages if x["role"] == "user"][-1]
    embedding = generate_embeddings([latest_user_message])[0]
    retrieved_memories = await search_memories(search_vector=embedding, user_id=user_id)
    response = await update_memories_agent(
        user_id=user_id, existing_memories=retrieved_memories, messages=messages
    )
    return response

Before the agent decides anything, we first search for memories similar to the latest user message. This search isn’t scoped by category, it’s a broad check across everything the user has, since the goal here is specifically to catch near-duplicates or outdated facts, not to answer a question.

This closes the loop: the response agent decides if something’s worth remembering, and this agent decides what to actually do about it.

Putting It All Together.

Here’s a real run of the system, terminal output included.