Copyright 2026 Firefly Software Foundation. Licensed under the Apache License 2.0.
Unless an example explicitly compares models, it uses
FIREFLY_AGENTIC_DEFAULT_MODEL. Set that and your provider credentials using the
model configuration guide before running agent examples.
The Memory module provides conversation history, working memory, and pluggable storage backends for agents, multi-agent delegation, reasoning patterns, and pipelines. It ensures that LLM interactions are stateful and context-aware across turns, sessions, and pipeline steps.
Conversation history and working facts have different persistence paths:
flowchart TD
MANAGER["MemoryManager"] --> HISTORY["ConversationMemory<br/>process-local typed messages"]
MANAGER --> FACTS["WorkingMemory<br/>namespace per scope"]
HISTORY --> BUDGET["Token budget and summarizer"]
HISTORY <-->|export and import| SNAPSHOT["Serializable conversation snapshot"]
SNAPSHOT <-->|host reads and writes| ARCHIVE["Host persistence"]
FACTS --> STORE["MemoryStore protocol"]
STORE --> LOCAL["InMemoryStore, FileStore, SQLiteStore"]
STORE --> DATABASE["PostgreSQLStore, MongoDBStore"]
STORE --> CUSTOM["Custom backend"]
- ConversationMemory -- Token-aware, per-conversation chat history that
wraps pydantic-ai's
message_historymechanism. - WorkingMemory -- Scoped key-value scratchpad for session facts, entities, and intermediate state.
- MemoryStore -- Pluggable persistence for working facts, not conversation history. Three are stdlib-only
(
InMemoryStore,FileStore,SQLiteStore);PostgreSQLStoreandMongoDBStorelive behind optional dependency groups. Implement theMemoryStoreprotocol for any custom backend. - MemoryManager -- Facade that composes conversation and working memory behind a single API.
from fireflyframework_agentic.agents import FireflyAgent
from fireflyframework_agentic.memory import MemoryManager
# Create a memory manager
memory = MemoryManager(max_conversation_tokens=32_000)
# Attach to an agent
agent = FireflyAgent(
name="assistant",
memory=memory,
)
# Start a conversation
conv_id = memory.new_conversation()
# Each run automatically loads/stores conversation history
result1 = await agent.run("What is Python?", conversation_id=conv_id)
result2 = await agent.run("What about its type system?", conversation_id=conv_id)
# result2 has full context from result1ConversationMemory manages list[ModelMessage] per conversation ID. It
automatically enforces a token budget by dropping oldest turns (FIFO).
from fireflyframework_agentic.memory import ConversationMemory
conv_mem = ConversationMemory(max_tokens=16_000)
cid = conv_mem.new_conversation()
# After an agent run, store the turn
conv_mem.add_turn(
cid,
user_prompt="Hello",
assistant_response="Hi there!",
raw_messages=result.new_messages(), # pydantic-ai ModelMessage list
)
# Before the next run, get the trimmed history
history = conv_mem.get_message_history(cid)
# Pass to agent: agent.run("Next question", message_history=history)When FireflyAgent has memory attached and a conversation_id is supplied, it
loads history unless the caller provided message_history, then records the
completed turn. Explicit export/import is needed
for history to survive a process restart.
Both streaming modes retain typed messages, including tool results and provider
metadata, when the stream completes successfully. A cancelled, incomplete, or
approval-deferred turn is not added to conversation history. Supplying explicit
message_history takes precedence over automatic history loading.
sequenceDiagram
participant App
participant Agent as FireflyAgent
participant History as ConversationMemory
participant Model as Pydantic AI model
participant Archive as Host snapshot storage
App->>Agent: run_stream(prompt, conversation_id=cid)
Agent->>History: get_message_history(cid)
History-->>Agent: typed messages within token budget
Agent->>Model: prompt and history
Model-->>App: streamed output through agent context
alt Completed, non-deferred output
Agent->>History: add_turn with new_messages()
else Cancelled, incomplete or deferred
Note over Agent,History: No completed turn is recorded
end
App->>History: export_conversation(cid)
History-->>App: serializable snapshot
App->>Archive: persist snapshot
Note over App,Archive: After a restart
Archive-->>App: stored snapshot
App->>History: import_conversation(snapshot)
When the total token count exceeds max_tokens, older turns are evicted from
the front. The TokenEstimator (from the content module) is used for counting.
conv_mem = ConversationMemory(max_tokens=4_000)
# After many turns, only the most recent ones fitting in 4000 tokens are returnedWhen many turns are evicted, important context can be lost. The optional summarization feature compresses evicted turns into a summary that is retained for context enrichment.
summarize_threshold is a token count (default 3072). When the sum of
the per-turn token_estimate values exceeds it, the oldest turns are evicted
and (if a summarizer is configured) compressed into a summary. Pass a
summarizer callable that takes a list of ConversationTurn objects and
returns a summary string:
from fireflyframework_agentic.memory import ConversationMemory
def my_summarizer(turns):
texts = [f"{t.user_prompt} -> {t.assistant_response}" for t in turns]
return "Summary of earlier conversation: " + "; ".join(texts)
conv_mem = ConversationMemory(
max_tokens=4_000,
summarize_threshold=3_000,
summarizer=my_summarizer,
)Summarization only runs when a summarizer is supplied; with no summarizer,
turns are still evicted by the token budget but no summary is produced.
Retrieve the summary for a conversation:
summary = conv_mem.get_summary(conversation_id)
if summary:
print(f"Evicted turns summarised as: {summary}")In production, the summarizer can be an LLM call that condenses older turns into a concise paragraph, preserving key facts while staying within budget.
When the manager is built via MemoryManager.from_config(), the threshold
comes from memory_summarize_threshold (env
FIREFLY_AGENTIC_MEMORY_SUMMARIZE_THRESHOLD) and is applied as a token
count. The config default is 10, which is intentionally small; set a
realistic token budget for production:
export FIREFLY_AGENTIC_MEMORY_SUMMARIZE_THRESHOLD=3000WorkingMemory is a scoped key-value store for facts and intermediate state.
It is useful for passing context between pipeline steps, storing extraction
results for validation, or maintaining entities across reasoning iterations.
from fireflyframework_agentic.memory import WorkingMemory
wm = WorkingMemory(scope_id="idp-session-42")
wm.set("doc_type", "invoice")
wm.set("vendor", "Acme Corp", importance=0.9) # importance is 0.0-1.0, default 0.5
print(wm.get("doc_type")) # "invoice"
print(wm.has("vendor")) # True
print(wm.keys()) # ["doc_type", "vendor"]
print(wm.items()) # [("doc_type", "invoice"), ("vendor", "Acme Corp")]
print(wm.to_dict()) # {"doc_type": "invoice", "vendor": "Acme Corp"}
wm.delete("vendor") # remove a single key
# Render as a text block for prompt injection
print(wm.to_context_string())
# Working Memory:
# - doc_type: invoiceThe full key-value API is set(key, value, *, importance=0.5), get(key, default=None), has(key), delete(key), keys(), items(), to_dict(),
to_context_string(), and clear().
Multiple WorkingMemory instances can share the same store backend while
maintaining independent namespaces:
from fireflyframework_agentic.memory import InMemoryStore, WorkingMemory
store = InMemoryStore()
agent_a_mem = WorkingMemory(store=store, scope_id="agent_a")
agent_b_mem = WorkingMemory(store=store, scope_id="agent_b")
agent_a_mem.set("key", "from A")
agent_b_mem.set("key", "from B")
assert agent_a_mem.get("key") == "from A"
assert agent_b_mem.get("key") == "from B"The unit of storage is MemoryEntry, and every store honours its fields:
from fireflyframework_agentic.memory import MemoryEntry, MemoryScopescope— aMemoryScope(CONVERSATION,WORKING, orLONG_TERM) classifying the entry's lifetime.key— optional key for key-value lookups (used byWorkingMemory).content— the stored value (string, dict, list, etc.).importance— a0.0–1.0priority weight (default0.5).expires_at— optional TTL timestamp. Entries past their expiry reportentry.is_expiredand are filtered out ofload()/load_by_key()by every backend; the database backends additionally providecleanup_expired()to delete them.
Dict-backed, fast, non-persistent. Suitable for testing and short-lived sessions.
from fireflyframework_agentic.memory import InMemoryStore
store = InMemoryStore()JSON file persistence. Each namespace is a separate file.
from fireflyframework_agentic.memory import FileStore
store = FileStore(base_dir=".firefly_memory")FileStore also provides async wrappers (async_save, async_load,
async_load_by_key, async_delete, async_clear) that delegate blocking I/O
to asyncio.to_thread(), keeping the event loop non-blocking in async
applications:
await store.async_save("conversations", entry)
entries = await store.async_load("conversations")Stdlib-only SQLite persistence (sqlite3), with the same constructor-and-go
ergonomics as FileStore. Compared to FileStore, every operation is atomic
(a crash mid-save cannot corrupt the file) and namespace queries use a SQL
index instead of loading and parsing every entry. The file and its parent
directories are created automatically.
from fireflyframework_agentic.memory import SQLiteStore
store = SQLiteStore("data/firefly_memory.sqlite3")SQLiteStore exposes the same async wrappers as FileStore (async_save,
async_load, async_load_by_key, async_delete, async_clear). Choose
FileStore for human-readable JSON; choose SQLiteStore for crash safety and
faster operations on larger namespaces.
Production-grade PostgreSQL persistence with connection pooling. Requires
asyncpg (install via pip install fireflyframework-agentic[postgres]).
The constructor is PostgreSQLStore(url, *, pool_size=10, pool_min_size=2, timeout=30.0, schema_name="firefly_memory"). Call await store.initialize()
during startup to create the pool and schema. Initialization is idempotent;
store.initialize_sync() is the synchronous equivalent.
import os
from fireflyframework_agentic.memory import MemoryManager
from fireflyframework_agentic.memory.database_store import PostgreSQLStore
store = PostgreSQLStore(
url=os.environ["FIREFLY_AGENTIC_MEMORY_POSTGRES_URL"],
pool_size=10,
pool_min_size=2,
schema_name="firefly_memory",
)
await store.initialize()
memory = MemoryManager(store=store)
# During application shutdown, after all users have finished:
await memory.aclose()The pool belongs to one persistent event loop managed by the store. Both sync
save / load methods and async async_save / async_load methods dispatch to
that loop and initialize lazily when needed. Sync methods block the caller;
async methods await the worker without blocking the caller's event loop. Never
use the private pool directly from an application loop.
Use await store.close() or store.close_sync() to shut down a standalone store.
await store.cleanup_expired() deletes expired entries and returns their count.
Environment Configuration:
export FIREFLY_AGENTIC_MEMORY_BACKEND=postgres
export FIREFLY_AGENTIC_MEMORY_POSTGRES_URL="$DATABASE_URL"
export FIREFLY_AGENTIC_MEMORY_POSTGRES_POOL_SIZE=10
export FIREFLY_AGENTIC_MEMORY_POSTGRES_POOL_MIN_SIZE=2
export FIREFLY_AGENTIC_MEMORY_POSTGRES_SCHEMA=firefly_memoryMemoryManager.from_config() reads these and initializes the database backend
through initialize_sync() on its owning loop. Supply DATABASE_URL through the
host environment or secret store.
Schema (created automatically on initialize()):
CREATE TABLE IF NOT EXISTS firefly_memory.memory_entries (
entry_id TEXT PRIMARY KEY,
namespace TEXT NOT NULL,
scope TEXT NOT NULL,
key TEXT,
content JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
expires_at TIMESTAMPTZ,
importance FLOAT NOT NULL DEFAULT 0.5
CHECK (importance >= 0.0 AND importance <= 1.0)
);
CREATE INDEX idx_namespace ON firefly_memory.memory_entries(namespace);
CREATE INDEX idx_namespace_key ON firefly_memory.memory_entries(namespace, key)
WHERE key IS NOT NULL;
CREATE INDEX idx_expires_at ON firefly_memory.memory_entries(expires_at)
WHERE expires_at IS NOT NULL;The schema name (default firefly_memory) is configurable via schema_name.
Scalable MongoDB persistence with connection pooling. Requires motor and
pymongo (install via pip install fireflyframework-agentic[mongodb]).
The constructor is MongoDBStore(url, *, database="firefly_memory", collection="entries", pool_size=10). As with PostgreSQLStore, call
await store.initialize() before use (it connects and creates indexes) and
await store.close() on shutdown.
import os
from fireflyframework_agentic.memory import MemoryManager
from fireflyframework_agentic.memory.database_store import MongoDBStore
store = MongoDBStore(
url=os.environ["FIREFLY_AGENTIC_MEMORY_MONGODB_URL"],
database="firefly_memory",
collection="entries",
pool_size=10,
)
await store.initialize()
memory = MemoryManager(store=store)
# During application shutdown, after all users have finished:
await memory.aclose()MongoDB uses the same owning-loop lifecycle as PostgreSQL. Synchronous hosts can
use initialize_sync() / close_sync(); asynchronous hosts can await
initialize() / close(). Stored timestamps use timezone-aware BSON datetimes,
so expiry queries compare dates rather than strings.
MongoDBStore also exposes await store.cleanup_expired() -> int to purge
expired entries.
Environment Configuration:
export FIREFLY_AGENTIC_MEMORY_BACKEND=mongodb
export FIREFLY_AGENTIC_MEMORY_MONGODB_URL=mongodb://localhost:27017/
export FIREFLY_AGENTIC_MEMORY_MONGODB_DATABASE=firefly_memory
export FIREFLY_AGENTIC_MEMORY_MONGODB_COLLECTION=entries
export FIREFLY_AGENTIC_MEMORY_MONGODB_POOL_SIZE=10The store automatically creates the entries collection (in the
firefly_memory database by default) and these indexes:
// Each document is the serialized MemoryEntry plus namespace/scope fields:
{
entry_id: "abc123",
namespace: "working:session-42",
scope: "working",
key: "doc_type",
content: { ... },
metadata: { ... },
created_at: ISODate("..."),
expires_at: null,
importance: 0.5
}
// Indexes
db.entries.createIndex({ namespace: 1 })
db.entries.createIndex({ namespace: 1, key: 1 },
{ partialFilterExpression: { key: { $exists: true } } })
db.entries.createIndex({ expires_at: 1 },
{ partialFilterExpression: { expires_at: { $exists: true } } })
db.entries.createIndex({ entry_id: 1 }, { unique: true })Implement the MemoryStore protocol for a new backend or compose an existing one.
This complete adapter prefixes namespaces while delegating persistence and expiry
to its backend; use SQLiteStore or FileStore when persistence is required:
from fireflyframework_agentic.memory import MemoryStore, MemoryEntry
from fireflyframework_agentic.memory.store import InMemoryStore
class PrefixedStore:
def __init__(self, prefix: str, backend: MemoryStore) -> None:
self._prefix = prefix
self._backend = backend
def _namespace(self, namespace: str) -> str:
return f"{len(self._prefix)}:{self._prefix}:{namespace}"
def save(self, namespace: str, entry: MemoryEntry) -> None:
self._backend.save(self._namespace(namespace), entry)
def load(self, namespace: str) -> list[MemoryEntry]:
return self._backend.load(self._namespace(namespace))
def load_by_key(self, namespace: str, key: str) -> MemoryEntry | None:
return self._backend.load_by_key(self._namespace(namespace), key)
def delete(self, namespace: str, entry_id: str) -> None:
self._backend.delete(self._namespace(namespace), entry_id)
def clear(self, namespace: str) -> None:
self._backend.clear(self._namespace(namespace))
store = PrefixedStore("invoice-service", InMemoryStore())
store.save("working", MemoryEntry(key="vendor", content="Acme Corp"))
assert store.load_by_key("working", "vendor").content == "Acme Corp"ConversationMemory supports exporting and importing conversations as
JSON-serialisable dictionaries. This is useful for backup, migration,
debugging, and cross-service conversation transfer.
# Export a conversation
data = conv_mem.export_conversation(conversation_id)
# data is a dict with: conversation_id, turns, summary, total_tokens
# Import into another instance (or after restart)
new_id = conv_mem.import_conversation(data)
# Or override the conversation ID:
new_id = conv_mem.import_conversation(data, conversation_id="custom-id")Exported turns include turn_id, user_prompt, assistant_response,
token_estimate, and metadata. Note that raw_messages (pydantic-ai
ModelMessage objects) are not exported as they are not portable.
create_llm_summarizer() is a factory function that returns a callable
suitable for ConversationMemory's summarizer parameter. It uses an
ephemeral Pydantic AI agent to compress evicted turns into a concise summary.
from fireflyframework_agentic.memory.summarization import create_llm_summarizer
summarizer = create_llm_summarizer()
conv_mem = ConversationMemory(
max_tokens=4_000,
summarize_threshold=3_000,
summarizer=summarizer,
)Parameters:
model— Pydantic AI model string orModelinstance. WhenNone, uses the framework default model.prompt_template— Custom prompt template. Must contain a{turns_text}placeholder.max_summary_tokens— Soft limit on summary length (guidance, not enforced).
If the LLM call fails, the factory falls back to a non-LLM truncation strategy that extracts key sentences from the most recent turns.
MemoryManager is the single entry point for conversation and working memory.
It is the object you attach to agents, delegation routers, and pipelines. The
constructor takes keyword-only arguments:
from fireflyframework_agentic.memory import MemoryManager
mgr = MemoryManager(
store=None, # defaults to InMemoryStore
max_conversation_tokens=32_000,
summarize_threshold=10,
working_scope_id="main-session",
)
# Conversation
cid = mgr.new_conversation()
mgr.add_turn(cid, "hello", "hi", raw_messages)
history = mgr.get_message_history(cid)
# Working memory
mgr.set_fact("doc_type", "invoice")
mgr.get_fact("doc_type") # "invoice"
print(mgr.get_working_context()) # text block for prompt injectionMemoryManager.from_config() is the canonical env-driven wiring path. It reads
memory_backend, memory_max_conversation_tokens, and
memory_summarize_threshold from the framework config, selects the matching
backend (in_memory / file / postgres / mongodb), raises if a required
URL is missing, and initializes database backends for you. It is a synchronous
factory; use it during synchronous startup or offload initialization in an async
host:
import asyncio
from fireflyframework_agentic.memory import MemoryManager
mgr = await asyncio.to_thread(MemoryManager.from_config)The composed subsystems are reachable directly, and lifecycle helpers clear state:
mgr.conversation # the ConversationMemory instance
mgr.working # the WorkingMemory instance
mgr.store # the underlying MemoryStore
mgr.clear_conversation(cid) # clear one conversation
mgr.clear_working() # clear working memory
mgr.clear_all() # clear bothThe database backends keep asyncpg and Motor clients on a shared database worker loop. Sync methods block until that loop finishes the operation; async methods await it. The synchronous agent runner uses a separate loop, so synchronous memory hooks do not block the database loop that must serve them.
flowchart LR
ASYNC["Application event loop<br/>await agent.run"] -->|await DB method| DB["Database worker loop<br/>asyncpg or Motor client"]
SYNC["Synchronous caller"] --> AGENT["Agent worker loop<br/>agent.run_sync"]
AGENT -->|blocking memory method| DB
HOST["Application shutdown"] --> CLOSE["Close managers after all forks finish"]
CLOSE -->|await aclose or call close| DB
Provider clients used by await agent.run() remain on the application loop;
clients used by run_sync() stay on the agent worker loop. Keep each agent and its
loop-bound clients in one execution mode. These lifetimes are separate from the
database store's owning loop.
Call mgr.close() during synchronous shutdown or await mgr.aclose() during async
shutdown. These release supported database clients without deleting stored memory.
The shared database worker loop remains available until process exit. A synchronous application's lifetime
can use try/finally:
from fireflyframework_agentic.memory import MemoryManager
mgr = MemoryManager.from_config()
try:
mgr.set_fact("status", "ready")
print(mgr.get_fact("status"))
finally:
mgr.close()Forked managers share the backend. Close it only after every manager, fork, and
in-flight operation using that backend has finished; closing any manager closes
the shared database store. clear_all() removes state and is separate from closing.
When delegating to a sub-agent or branching a pipeline, fork() creates a
child manager that shares conversation memory but has independent working memory:
child = mgr.fork(working_scope_id="sub-agent-classify")
child.set_fact("classification", "invoice")
# parent's working memory is unaffectedagent = FireflyAgent(name="bot", memory=mgr)
result = await agent.run("Hi", conversation_id=cid)
# message_history is auto-injected, new messages auto-storedfrom fireflyframework_agentic.agents.delegation import DelegationRouter, RoundRobinStrategy
router = DelegationRouter([agent_a, agent_b], RoundRobinStrategy(), memory=mgr)
result = await router.route("Translate this text.")
# delegated agent receives a forked memory scopefrom fireflyframework_agentic.pipeline.context import PipelineContext
ctx = PipelineContext(inputs=data, memory=mgr)
result = await engine.run(context=ctx)
# AgentStep and ReasoningStep automatically propagate memoryMemory is injected into the reasoning state dict as state["memory"]:
result = await pattern.execute(agent, "Analyze this", memory=mgr)
# Inside pattern hooks: state["memory"].working.set("key", "value")The framework is a pure in-process library: it serves no HTTP port. When a host
service exposes agents over its own transport, it owns the wiring of
conversation_id from the inbound request to agent.run(..., conversation_id=...).
The memory backend simply provides the stateful continuity behind that call.
Memory settings are configured via environment variables (prefix
FIREFLY_AGENTIC_) and consumed by MemoryManager.from_config():
# Core
export FIREFLY_AGENTIC_MEMORY_BACKEND=in_memory # in_memory | file | postgres | mongodb
export FIREFLY_AGENTIC_MEMORY_MAX_CONVERSATION_TOKENS=128000
export FIREFLY_AGENTIC_MEMORY_SUMMARIZE_THRESHOLD=10 # token count (small default; raise for prod)
export FIREFLY_AGENTIC_MEMORY_FILE_DIR=.firefly_memory # used by the "file" backend
# PostgreSQL backend
export FIREFLY_AGENTIC_MEMORY_POSTGRES_URL="$DATABASE_URL"
export FIREFLY_AGENTIC_MEMORY_POSTGRES_POOL_SIZE=10
export FIREFLY_AGENTIC_MEMORY_POSTGRES_POOL_MIN_SIZE=2
export FIREFLY_AGENTIC_MEMORY_POSTGRES_SCHEMA=firefly_memory
# MongoDB backend
export FIREFLY_AGENTIC_MEMORY_MONGODB_URL=mongodb://localhost:27017/
export FIREFLY_AGENTIC_MEMORY_MONGODB_DATABASE=firefly_memory
export FIREFLY_AGENTIC_MEMORY_MONGODB_COLLECTION=entries
export FIREFLY_AGENTIC_MEMORY_MONGODB_POOL_SIZE=10Note: the from_config() env path supports in_memory, file, postgres,
and mongodb. SQLiteStore is constructed explicitly (SQLiteStore(path))
and passed via MemoryManager(store=...).