Skip to content

KnowledgeStore

autogen.beta.knowledge.base.KnowledgeStore #

Bases: Protocol

Virtual path-based store for actor knowledge.

Provides filesystem semantics over any storage backend. Paths use Unix conventions: /dir/subdir/file.txt Directories are implicit -- writing /a/b/c.txt implies /a/ and /a/b/ exist. Listing returns immediate children. Directory entries end with '/'.

The network layer uses the same protocol to back the hub's virtual file system; see :mod:autogen.beta.network. Three methods beyond the basic CRUD are required for WAL-backed sessions: append and read_range are mandatory, while on_change is optional (backends that cannot observe changes efficiently return a :class:NoopChangeSubscription and callers fall back to polling).

read async #

read(path)

Read content at path. Returns None if not found.

Source code in autogen/beta/knowledge/base.py 
53
54
55
async def read(self, path: str) -> str | None:
    """Read content at path. Returns None if not found."""
    ...

write async #

write(path, content)

Write content to path. Creates parent directories implicitly.

Source code in autogen/beta/knowledge/base.py 
57
58
59
async def write(self, path: str, content: str) -> None:
    """Write content to path. Creates parent directories implicitly."""
    ...

list async #

list(path='/')

List immediate children of a directory path.

Returns relative names. Directories end with '/'. Example: list("/log/") might return ["stream-abc.jsonl", "stream-def.jsonl"]

Source code in autogen/beta/knowledge/base.py 
61
62
63
64
65
66
67
async def list(self, path: str = "/") -> list[str]:
    """List immediate children of a directory path.

    Returns relative names. Directories end with '/'.
    Example: list("/log/") might return ["stream-abc.jsonl", "stream-def.jsonl"]
    """
    ...

delete async #

delete(path)

Delete entry at path. No-op if not found.

Source code in autogen/beta/knowledge/base.py 
69
70
71
async def delete(self, path: str) -> None:
    """Delete entry at path. No-op if not found."""
    ...

exists async #

exists(path)

Check if path exists.

Source code in autogen/beta/knowledge/base.py 
73
74
75
async def exists(self, path: str) -> bool:
    """Check if path exists."""
    ...

append async #

append(path, content)

Atomically append content to the file at path.

Creates the file (and its parents) if it does not exist. Returns the byte offset at which content was written, so callers can record a cursor for later read_range calls.

Source code in autogen/beta/knowledge/base.py 
77
78
79
80
81
82
83
84
async def append(self, path: str, content: str) -> int:
    """Atomically append ``content`` to the file at ``path``.

    Creates the file (and its parents) if it does not exist. Returns the
    byte offset at which ``content`` was written, so callers can record a
    cursor for later ``read_range`` calls.
    """
    ...

read_range async #

read_range(path, start, end=None)

Read the byte slice [start, end) of the file at path.

end of None means "up to the current end of file". Returns an empty string if the file does not exist. Slices are returned as UTF-8 text; callers that append multi-byte content must align offsets to character boundaries themselves.

Source code in autogen/beta/knowledge/base.py 
86
87
88
89
90
91
92
93
94
async def read_range(self, path: str, start: int, end: int | None = None) -> str:
    """Read the byte slice ``[start, end)`` of the file at ``path``.

    ``end`` of ``None`` means "up to the current end of file". Returns an
    empty string if the file does not exist. Slices are returned as UTF-8
    text; callers that append multi-byte content must align offsets to
    character boundaries themselves.
    """
    ...

on_change async #

on_change(path, callback)

Subscribe to change notifications at path.

Backends that can observe changes efficiently invoke callback(path) whenever a file under path changes. Backends that cannot must return :class:NoopChangeSubscription; the hub then polls on a short interval instead.

Source code in autogen/beta/knowledge/base.py 
 96
 97
 98
 99
100
101
102
103
104
async def on_change(self, path: str, callback: ChangeCallback) -> ChangeSubscription:
    """Subscribe to change notifications at ``path``.

    Backends that can observe changes efficiently invoke
    ``callback(path)`` whenever a file under ``path`` changes. Backends
    that cannot must return :class:`NoopChangeSubscription`; the hub
    then polls on a short interval instead.
    """
    ...