Knowledge Store

A KnowledgeStore is a virtual, path-based key-value store with filesystem-like semantics. It gives agents a durable place to persist conversation logs, artifacts, working memory, and any other content that should outlive a single agent.ask() call.

All implementations share the same protocol, so you can swap an in-memory store for a SQLite file or Redis instance without changing callers.

The protocol#

The KnowledgeStore protocol (importable from autogen.beta) defines the full API every implementation satisfies:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from typing import Protocol

class KnowledgeStore(Protocol):
    async def read(self, path: str) -> str | None: ...
    async def write(self, path: str, content: str) -> None: ...
    async def list(self, path: str = "/") -> list[str]: ...
    async def delete(self, path: str) -> None: ...
    async def exists(self, path: str) -> bool: ...

    async def append(self, path: str, content: str) -> int: ...
    async def read_range(self, path: str, start: int, end: int | None = None) -> str: ...

    async def on_change(self, path: str, callback) -> ChangeSubscription: ...

Paths follow Unix conventions: absolute (/dir/subdir/file.txt), directories are implicit, and list() returns immediate children with / suffixes for directories.

The append / read_range pair supports WAL-style workloads: append returns the byte offset, which you can later hand to read_range to retrieve just the newly written slice.

Choosing an implementation#

Basic usage#

Memory store — fastest, non-persistent#

1
2
3
4
5
6
7

from autogen.beta import MemoryKnowledgeStore

store = MemoryKnowledgeStore()

await store.write("/artifacts/report.md", "# Q3 Summary\n...")
print(await store.read("/artifacts/report.md"))
print(await store.list("/"))  # ['artifacts/']

Sqlite store — persistent across process restarts#

1
2
3
4
5
6
7
8

from autogen.beta import SqliteKnowledgeStore

store = SqliteKnowledgeStore("/var/agents/alice/knowledge.db")
await store.write("/config/model.txt", "claude-opus-4-7")

# In a later run, the same DB:
store2 = SqliteKnowledgeStore("/var/agents/alice/knowledge.db")
print(await store2.read("/config/model.txt"))  # "claude-opus-4-7"

Disk store — files on the filesystem#

1
2
3
4
5

from autogen.beta import DiskKnowledgeStore

store = DiskKnowledgeStore("/var/agents/alice/knowledge/")
await store.write("/artifacts/data.json", '{"ok": true}')
# produces /var/agents/alice/knowledge/artifacts/data.json

Append and read_range#

These two methods support WAL-style event logs, turn-by-turn transcripts, and any append-only workload where you want to retrieve only new content.

1
2
3
4
5
6

off1 = await store.append("/log/events.jsonl", '{"t": 1}\n')
off2 = await store.append("/log/events.jsonl", '{"t": 2}\n')

# Read everything appended in the second write
new_slice = await store.read_range("/log/events.jsonl", off1)
print(new_slice)  # '{"t": 2}\n'

Change subscriptions#

Callers can react to writes via on_change. Backends that observe changes efficiently (DiskKnowledgeStore using watchdog) call the callback directly. Backends that cannot (MemoryKnowledgeStore, SqliteKnowledgeStore) return a NoopChangeSubscription — the caller is expected to poll.

1
2
3
4
5
6

async def on_log_change(path: str) -> None:
    print(f"{path} changed")

sub = await store.on_change("/log/", on_log_change)
# ... later:
await sub.cancel()

DefaultBootstrap#

DefaultBootstrap populates a store with a standard layout and SKILL.md files that explain each directory to an LLM reader. It's designed to be called once per agent:

1
2
3
4
5
6
7

from autogen.beta import DefaultBootstrap, MemoryKnowledgeStore

store = MemoryKnowledgeStore()
await DefaultBootstrap().bootstrap(store, actor_name="alice")

print(await store.list("/"))
# ['SKILL.md', 'artifacts/', 'log/', 'memory/']

Resulting layout:

Implement your own StoreBootstrap if you need a different layout.

EventLogWriter — persist stream history#

EventLogWriter serializes a Stream's events to a KnowledgeStore as JSONL, and can reconstruct them later. Useful for replay, audit, or multi-run aggregation.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from autogen.beta import Agent, EventLogWriter, MemoryKnowledgeStore, MemoryStream
from autogen.beta.config import OpenAIConfig

stream = MemoryStream()
agent = Agent("assistant", config=OpenAIConfig(model="gpt-5"))
await agent.ask("Hello!", stream=stream)

# Persist all events from this stream
store = MemoryKnowledgeStore()
writer = EventLogWriter(store)
events = list(await stream.history.get_events())
await writer.persist(stream.id, events)

# Reload later
loaded = await writer.load(stream.id)  # -> list[BaseEvent]

Persisted events land at /log/{stream_id}.jsonl. Events of types that cannot be deserialized (e.g. a removed class) come back as UnknownEvent — no data is lost.

When an Agent is given a KnowledgeConfig, it runs this persist(...) step for you at the end of every ask() — so you only call EventLogWriter directly in a custom harness. Set KnowledgeConfig.write_event_log=False to turn the automatic write off (e.g. when the store is purely user-facing memory rather than a transcript archive); if the writer raises, the turn still returns its reply and an EventLogFailed event is emitted on the stream.

LockedKnowledgeStore — serialize writers#

LockedKnowledgeStore wraps any KnowledgeStore to serialize concurrent writes. It delegates locking to a user-provided object implementing acquire(name, ttl) / release(name) — typically a distributed lock (Redis, database advisory locks, etc.) so multiple processes sharing the same store can coordinate.

1
2
3
4
5

from autogen.beta import LockedKnowledgeStore, SqliteKnowledgeStore

inner = SqliteKnowledgeStore("/var/agents/shared.db")
store = LockedKnowledgeStore(inner, lock=your_distributed_lock)
# ... hand `store` to every agent that shares the DB