Memory

Two memory primitives extend what an agent can recall. Working memory keeps conversation history alive across multiple calls within a single workflow execution. Long-term memory stores discrete facts in a database so agents can retrieve them in future executions.

Both are optional and independent. Use either, both, or neither.

Working memory

Pass working_memory() to agent() to maintain conversation history across turns. Each message the agent exchanges is written to the Flux event log as a task event, so the history is durable and replay-safe — no separate database required.

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent
from flux.tasks.ai.memory import working_memory

@workflow
async def chatbot(ctx: ExecutionContext):
    bot = await agent(
        system_prompt="You are a helpful assistant.",
        model="ollama/llama3.2",
        working_memory=working_memory(),
    )
    r1 = await bot("My name is Alice. What is Python?")
    r2 = await bot("What did I just tell you my name is?")  # bot knows "Alice"
    return r2

Because messages live in the event log, the agent retains context across workflow pause and resume. On resume, every completed task replays from the log without re-contacting the LLM. Only the first new bot() call after the resume point sends a request, and it includes the full prior conversation.

Controlling context size

Two parameters trim the message window sent to the LLM. Earlier messages stay in the event log; they are excluded from the next LLM call.

Limit by message count:

working_memory(window=20)  # pass only the 20 most recent messages

Limit by approximate token count:

working_memory(max_tokens=4000)  # trim oldest messages until estimated tokens fit

Token estimation uses len(content) // 4; treat it as an approximation, not an exact count from the model’s tokenizer.

Automatic compaction

When both max_tokens and compact_model are set, the agent summarizes old messages into a single replacement before they are dropped. This preserves key facts across long conversations while keeping the context window under budget.

working_memory(
    max_tokens=4000,
    compact_model="ollama/llama3.2",  # model used to summarize
    compact_threshold=0.70,           # compact when at 70% of max_tokens
    compact_preserve=4,               # keep the N most recent messages verbatim
)

compact_threshold and compact_preserve are optional; the defaults shown above apply when omitted.

Full working_memory() signature

working_memory(
    window: int | None = None,
    max_tokens: int | None = None,
    compact_model: str | None = None,
    compact_threshold: float = 0.70,
    compact_preserve: int = 4,
) -> WorkingMemory

Long-term memory

Long-term memory stores named facts (key/value pairs) in a persistent backend. The LLM decides what to store and retrieve through four tools that agent() adds automatically when long_term_memory is provided.

The agent calls these tools autonomously. No explicit calls are needed in workflow code.

Scoping facts

Every long_term_memory instance carries an agent name and a scope string. Together they namespace the stored facts so different agents or different users never share the same key space.

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent
from flux.tasks.ai.memory import working_memory, long_term_memory, sqlite

@workflow
async def personal_assistant(ctx: ExecutionContext):
    assistant = await agent(
        system_prompt=(
            "You are a personal assistant. Remember important facts about the user "
            "with store_memory. Always check memory first with recall_memory."
        ),
        model="ollama/llama3.2",
        working_memory=working_memory(),
        long_term_memory=long_term_memory(
            provider=sqlite("assistant.db"),
            agent="personal_assistant",
            scope=f"user:{ctx.input['user_id']}",
        ),
    )
    return await assistant(ctx.input["message"])

On the first execution, the agent stores facts with store_memory. On subsequent executions, recall_memory retrieves them.

Use any string as the scope to separate memory by user, session, or resource:

# Per-user
long_term_memory(provider=sqlite("users.db"), agent="assistant", scope="user:123")

# Per pull-request
long_term_memory(provider=sqlite("reviews.db"), agent="reviewer", scope="pr:456")

# Per session
long_term_memory(provider=sqlite("sessions.db"), agent="bot", scope="session:abc")

Full long_term_memory() signature

long_term_memory(
    provider,      # MemoryProvider instance: sqlite(...), postgresql(...), in_memory(), or custom
    agent: str,    # name that namespaces facts (e.g., the workflow or agent name)
    scope: str,    # secondary namespace (e.g., "user:123", "session:abc")
) -> LongTermMemory

Backends

SQLite

Stores facts in a local file. Suitable for single-process deployments and development:

from flux.tasks.ai.memory import sqlite

provider = sqlite("memory.db")  # creates the file on first use

The sqlite() helper prepends sqlite:/// and passes the full URL to SQLAlchemy. The memory table is created automatically on first write with (agent, scope, key) as the primary key.

PostgreSQL

Stores facts in a PostgreSQL database. Use this in multi-process or distributed deployments where multiple workers need to read and write the same fact store:

from flux.tasks.ai.memory import postgresql

provider = postgresql("postgresql://user:password@localhost/mydb")

Requires the psycopg (v3) driver, installed by the postgresql extra:

pip install 'flux-core[postgresql]'

The provider connects through SQLAlchemy and pins the psycopg dialect automatically — pass a plain postgresql:// URL.

Both backends use a dialect-aware upsert (INSERT ... ON CONFLICT DO UPDATE), so writing the same key twice overwrites the previous value.

In-memory

Stores facts in process memory with no persistence. Intended for testing and examples:

from flux.tasks.ai.memory import in_memory

provider = in_memory()  # all facts lost when the process exits

Shared memory across agents

Pass the same long_term_memory instance to multiple agents to give them a shared fact store. Facts written by one agent are immediately visible to the other.

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent
from flux.tasks.ai.memory import long_term_memory, in_memory

shared = long_term_memory(
    provider=in_memory(),
    agent="shared_agent",
    scope="review:pr-42",
)

@workflow
async def code_review(ctx: ExecutionContext):
    reviewer = await agent(
        system_prompt=(
            "You are a code reviewer. Store your findings with store_memory, "
            "organized by category (bugs, style, security)."
        ),
        model="ollama/llama3.2",
        long_term_memory=shared,
    )
    summarizer = await agent(
        system_prompt=(
            "Use recall_memory and list_memory_keys to read the reviewer's findings, "
            "then write a concise summary."
        ),
        model="ollama/llama3.2",
        long_term_memory=shared,
    )

    await reviewer(f"Review this code:\n\n{ctx.input['code']}")
    return await summarizer("Summarize the code review findings.")

The reviewer writes findings; the summarizer reads them within the same scope. Switch to sqlite(...) or postgresql(...) when the agents run on different workers.

Custom backends

Any class that implements the MemoryProvider protocol works as a provider:

from typing import Any, Protocol

class MemoryProvider(Protocol):
    async def memorize(self, agent: str, scope: str, key: str, value: Any) -> None: ...
    async def recall(self, agent: str, scope: str, key: str | None = None) -> Any: ...
    async def forget(self, agent: str, scope: str, key: str | None = None) -> None: ...
    async def keys(self, agent: str, scope: str) -> list[str]: ...
    async def scopes(self, agent: str) -> list[str]: ...

Pass an instance directly to long_term_memory():

provider = MyRedisProvider("redis://localhost:6379")
memory = long_term_memory(provider=provider, agent="my_agent", scope="user:123")

recall(key=None) returns a dict of all key-value pairs in the scope. forget(key=None) clears all keys. Initialize the backend connection lazily on first use, as the built-in providers do.