poolsplit
poolsplit is a Python library for pool-split retrieval in agent memory systems. It solves the problem of memory starvation where high-priority entries monopolize the context window, by giving each type-group its own reserved token budget and allowing always-scope entries (like behavioral corrections) to load unconditionally. It is for developers building AI agents with memory stores that need fair and reliable retrieval.
✨ Key features
- Reserved token budgets per type-group (pools) prevent starvation.
- Always-scope entries load unconditionally, bypassing budget limits.
- Deterministic scoring and sorting with tie-breakers for stable results.
- Pluggable scorer, token counter, and activity checker.
- No runtime dependencies; single module, Python 3.9+.
- Entries are never mutated; deduplication by object identity.
🎯 Use cases
- Retrieve mixed-type memory entries with guaranteed representation for low-priority types.
- Ensure behavioral corrections always load even under heavy new content.
- Implement fair retrieval for agent memory with multiple entry categories.
- Build a custom retrieval system with pluggable scoring and token counting.
📦 Installation
🧰 Requirements: Python 3.9 or newer. No runtime dependencies or API keys required.
pip install .
For development (tests):
pip install .[dev]
pytest -q
Vendoring is supported: copy poolsplit.py into your project and import it.
🚀 Usage
import datetime
from poolsplit import Pool, retrieve_pools
entries = [
{"id": "c1", "type": "correction", "scope": "always",
"content": "Never claim a write that did not happen.", "created": "2026-07-06T00:00:00Z"},
{"id": "t1", "type": "task", "base_priority": 7,
"content": "Ship the refactor.", "created": "2026-07-06T00:00:00Z"},
{"id": "s1", "type": "skill", "base_priority": 2,
"content": "grep to the symbol before reading the file.", "created": "2026-07-06T00:00:00Z"},
]
pools = [
Pool("corrections", max_tokens=1, max_entries=8, types=("correction",)),
Pool("content", max_tokens=200, max_entries=15, types=("task",)),
Pool("skill", max_tokens=80, max_entries=2, types=("skill",)),
]
by_pool = retrieve_pools(entries, pools, now=datetime.datetime(2026, 7, 7))
# {"corrections": [c1], "content": [t1], "skill": [s1]}
⚠️ Good to know
poolsplit does not handle persistence, semantic scoring, access-frequency tracking, or async operations; it is a synchronous, in-process library that reads and returns dicts.
❓ FAQ
How does poolsplit prevent memory starvation?
It gives each type-group its own reserved token budget (pool) retrieved independently, so high-priority types cannot consume the budget of low-priority types.
What are always-scope entries?
Entries with scope="always" load unconditionally, bypassing both token and entry caps, ensuring they are never crowded out by fresh content.
Can I use my own scoring function?
Yes, the scorer is pluggable; you can pass your own scorer to retrieve or retrieve_pools for custom ranking.
Does poolsplit track access frequency?
No, it deliberately avoids access-frequency tracking to prevent feedback loops that can make entries unreachable.
📊 Repository
🤖 Overview, features, install steps and FAQ were generated from the project's README on Sep 4, 2026. Always check the original source before running commands.