Skip to content

MemoryKnowledgeStore

autogen.beta.knowledge.memory.MemoryKnowledgeStore #

MemoryKnowledgeStore()

In-memory KnowledgeStore. Development default.

Backed by a flat dict. Paths are keys. Directories are inferred from stored paths via prefix matching.

Source code in autogen/beta/knowledge/memory.py 
36
37
38
39
def __init__(self) -> None:
    self._data: dict[str, str] = {}
    self._append_lock = asyncio.Lock()
    self._subscribers: dict[str, list[ChangeCallback]] = {}

read async #

read(path)
Source code in autogen/beta/knowledge/memory.py 
41
42
async def read(self, path: str) -> str | None:
    return self._data.get(_normalize(path))

write async #

write(path, content)
Source code in autogen/beta/knowledge/memory.py 
44
45
46
47
async def write(self, path: str, content: str) -> None:
    normalized = _normalize(path)
    self._data[normalized] = content
    await self._notify(normalized)

list async #

list(path='/')
Source code in autogen/beta/knowledge/memory.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
async def list(self, path: str = "/") -> list[str]:
    prefix = _normalize(path).rstrip("/") + "/"

    children: set[str] = set()
    for key in self._data:
        if not key.startswith(prefix):
            continue
        remainder = key[len(prefix) :]
        if "/" in remainder:
            children.add(remainder.split("/")[0] + "/")
        else:
            children.add(remainder)
    return sorted(children)

delete async #

delete(path)
Source code in autogen/beta/knowledge/memory.py 
63
64
65
66
67
68
69
70
71
72
73
74
async def delete(self, path: str) -> None:
    normalized = _normalize(path)
    affected: list[str] = []
    if normalized in self._data:
        del self._data[normalized]
        affected.append(normalized)
    prefix = normalized.rstrip("/") + "/"
    for key in [k for k in self._data if k.startswith(prefix)]:
        del self._data[key]
        affected.append(key)
    for key in affected:
        await self._notify(key)

exists async #

exists(path)
Source code in autogen/beta/knowledge/memory.py 
76
77
78
79
80
81
async def exists(self, path: str) -> bool:
    normalized = _normalize(path)
    if normalized in self._data:
        return True
    prefix = normalized.rstrip("/") + "/"
    return any(k.startswith(prefix) for k in self._data)

append async #

append(path, content)
Source code in autogen/beta/knowledge/memory.py 
83
84
85
86
87
88
89
90
async def append(self, path: str, content: str) -> int:
    normalized = _normalize(path)
    async with self._append_lock:
        existing = self._data.get(normalized, "")
        offset = len(existing.encode("utf-8"))
        self._data[normalized] = existing + content
    await self._notify(normalized)
    return offset

read_range async #

read_range(path, start, end=None)
Source code in autogen/beta/knowledge/memory.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def read_range(self, path: str, start: int, end: int | None = None) -> str:
    normalized = _normalize(path)
    existing = self._data.get(normalized)
    if existing is None:
        return ""
    data = existing.encode("utf-8")
    stop = len(data) if end is None else min(end, len(data))
    if start >= stop:
        return ""
    return data[start:stop].decode("utf-8", errors="strict")

on_change async #

on_change(path, callback)
Source code in autogen/beta/knowledge/memory.py 
103
104
105
106
async def on_change(self, path: str, callback: ChangeCallback) -> ChangeSubscription:
    normalized = _normalize(path)
    self._subscribers.setdefault(normalized, []).append(callback)
    return _MemoryChangeSubscription(self._subscribers, normalized, callback)