Memory
Give agents conversation memory with pluggable backends — in-memory, SQL, vector stores, or the Dynamiq platform — plus save modes and retrieval strategies.
dynamiq.memory.Memory stores and retrieves conversation messages for agents. Attach a Memory instance to an agent and pass user_id / session_id at run time — the agent loads prior messages scoped to those ids before reasoning and persists the new turn afterwards.
Attach memory to an agent
from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.flows import Flow
from dynamiq.memory import Memory
from dynamiq.memory.backends.in_memory import InMemory
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI
llm = OpenAI(connection=OpenAIConnection(), model="gpt-4o")
agent = Agent(
name="assistant",
llm=llm,
role="Helpful assistant that remembers prior turns.",
memory=Memory(backend=InMemory()),
)
wf = Workflow(flow=Flow(nodes=[agent]))
ids = {"user_id": "demo-user", "session_id": "demo-session"}
wf.run(input_data={"input": "My name is Alex and I work at TechCorp.", **ids})
result = wf.run(input_data={"input": "Where do I work?", **ids})
print(result.output[agent.id]["output"]["content"]) # mentions TechCorpThe agent-side knobs:
memoryMemorymemory_limitintmemory_retrieval_strategyMemoryRetrievalStrategyMemory configuration
backendMemoryBackendmessage_limitintfiltersdictsave_modeMemorySaveModefrom dynamiq.memory import Memory, MemorySaveMode
memory = Memory(
backend=InMemory(),
save_mode=MemorySaveMode.INPUT_OUTPUT, # clean multi-turn chat, no tool traces
)Use MemorySaveMode.FULL when you need replay/debug fidelity; use INPUT_OUTPUT for lean chat history that keeps the next turn's prompt small.
Retrieval strategies
MemoryRetrievalStrategy controls which messages are loaded into the agent's prompt:
| Strategy | Behavior |
|---|---|
ALL (default) | The most recent messages, up to the limit |
RELEVANT | Messages semantically relevant to the current input |
BOTH | Recent messages merged with relevant ones |
from dynamiq.memory import MemoryRetrievalStrategy
agent = Agent(
name="assistant",
llm=llm,
memory=memory,
memory_limit=50,
memory_retrieval_strategy=MemoryRetrievalStrategy.BOTH,
)RELEVANT and BOTH need a backend that can rank by similarity: vector-store backends use embeddings, while InMemory ranks with BM25 keyword scoring.
Backends
All backends live in dynamiq.memory.backends: InMemory, SQLite, PostgreSQL, DynamoDB, Pinecone, Qdrant, Weaviate, and Dynamiq.
InMemory
No setup; messages vanish with the process. Best for tests and notebooks.
from dynamiq.memory.backends import InMemory
memory = Memory(backend=InMemory())SQLite
from dynamiq.memory.backends import SQLite
memory = Memory(backend=SQLite(db_path="conversations.db", index_name="conversations"))PostgreSQL
from dynamiq.connections import PostgreSQL as PostgreSQLConnection
from dynamiq.memory.backends import PostgreSQL
memory = Memory(
backend=PostgreSQL(
connection=PostgreSQLConnection(), # host/port/user/password from env
table_name="conversations",
create_if_not_exist=True,
)
)DynamoDB
from dynamiq.connections import AWS
from dynamiq.memory.backends import DynamoDB
memory = Memory(
backend=DynamoDB(
connection=AWS(),
table_name="conversations",
create_if_not_exist=True,
)
)Vector-store backends (Pinecone, Qdrant, Weaviate)
These store each message as an embedded vector, enabling RELEVANT retrieval. They require an embedder:
from dynamiq.connections import Pinecone as PineconeConnection
from dynamiq.memory.backends import Pinecone
from dynamiq.nodes.embedders import OpenAIDocumentEmbedder
from dynamiq.storages.vector.pinecone.pinecone import PineconeIndexType
memory = Memory(
backend=Pinecone(
connection=PineconeConnection(),
embedder=OpenAIDocumentEmbedder(model="text-embedding-3-small"),
index_name="conversations",
index_type=PineconeIndexType.SERVERLESS,
cloud="aws",
region="us-east-1",
create_if_not_exist=True,
)
)Qdrant takes connection, embedder, index_name, and dimension; Weaviate takes connection, embedder, and collection_name, plus an alpha parameter for hybrid keyword/vector search.
Dynamiq platform backend
The Dynamiq backend persists messages to a memory resource hosted on the Dynamiq platform, so SDK agents and deployed Apps can share the same conversation store:
import os
from dynamiq.connections import Dynamiq as DynamiqConnection
from dynamiq.memory import Memory, MemorySaveMode
from dynamiq.memory.backends.dynamiq import Dynamiq as DynamiqBackend
backend = DynamiqBackend(
connection=DynamiqConnection(
url=os.getenv("DYNAMIQ_URL", "https://api.getdynamiq.ai"),
api_key=os.environ["DYNAMIQ_API_KEY"],
),
memory_id=os.environ["DYNAMIQ_MEMORY_ID"], # id of the remote memory resource
)
memory = Memory(backend=backend, save_mode=MemorySaveMode.FULL)user_id and session_id from the agent's run input become metadata filters on the remote store, so multiple users and sessions share one memory resource without leaking into each other.
Working with memory directly
Memory is also usable without an agent:
from dynamiq.prompts import MessageRole
memory.add(role=MessageRole.USER, content="I need help with billing.",
metadata={"user_id": "u-1", "session_id": "s-1"})
recent = memory.get_all(limit=20)
relevant = memory.search(query="billing", filters={"user_id": "u-1"}, limit=10)
conversation = memory.get_agent_conversation(filters={"user_id": "u-1", "session_id": "s-1"})
memory.delete(session_id="s-1", user_id="u-1") # scoped cleanup