Context optimization¶
Getting the most useful prompt out of the smallest number of characters.
Venkai gives you a ranked, bounded list. Turning that into a good prompt block is client-side work, and this page is that work.
What Venkai does and does not bound¶
| Bounded by Venkai | The number of memories (limit) |
| Not bounded by Venkai | Characters, tokens, redundancy, relevance floor |
limit=5 can return 200 characters or 40 000. Everything below exists because
of that gap.
Budget by characters¶
def pack(memories, max_chars=4000):
"""Take ranked memories until the budget is spent.
`continue`, not `break`: an oversized memory should not discard the
shorter, still-highly-ranked ones behind it.
"""
out, used = [], 0
for m in memories:
cost = len(m["content"]) + 20 # ~formatting overhead per line
if used + cost > max_chars:
continue
out.append(m)
used += cost
return out
Characters are a proxy for tokens — roughly 4 characters per token for English prose, worse for code and non-Latin scripts. If you need a real token budget, count with your model's tokenizer; Venkai does not do it for you.
Apply a relevance floor¶
There is none server-side. Every result has a score, and nothing is dropped
for scoring badly:
Choosing the threshold:
- Run 20 representative queries.
- Read the results and mark each useful / not.
- Pick the score that separates them best.
- Re-derive it after any provider change.
Thresholds do not transfer between providers
hashing similarities cluster near zero, so scores sit low across the
board. A floor tuned on hashing will reject almost everything under
semantic, and one tuned on semantic will accept everything under hashing.
Ask for more, then filter¶
raw = recall(task, limit=15) # over-fetch
scored = [m for m in raw if m["score"] >= 0.35] # floor
final = pack(scored, max_chars=3000) # budget
Retrieval is cheap — no model call, deterministic. Over-fetching and filtering client-side gives you a real selection stage without waiting for one to exist server-side.
Collapse near-duplicates¶
def dedupe(memories, overlap=0.8):
"""Drop a memory whose word set is mostly covered by a higher-ranked one.
Crude, and enough: exact and near-exact repeats are the common case.
ponytail: token-set overlap, not MMR. Upgrade to embedding-based MMR only
if you measure that paraphrase duplicates are actually costing you budget.
"""
kept = []
for m in memories:
words = set(m["content"].lower().split())
if any(len(words & set(k["content"].lower().split())) / max(len(words), 1) > overlap
for k in kept):
continue
kept.append(m)
return kept
Format compactly¶
def render(memories):
"""Group by type. Constraints first — they are what must not be violated."""
order = ["constraint", "decision", "preference", "fact", "event", "relationship"]
groups = {}
for m in memories:
groups.setdefault(m["type"], []).append(m["content"])
lines = []
for t in order:
if t in groups:
lines.append(f"{t.upper()}S:" if t != "preference" else "PREFERENCES:")
lines.extend(f" - {c}" for c in groups[t])
return "\n".join(lines)
CONSTRAINTS:
- Refunds over 500 EUR require a second approver.
- The retrieval endpoint must answer in under 200 ms at p95.
DECISIONS:
- We use Postgres, not MongoDB, because the ranking query needs joins.
Do not ship score, id, created_at, metadata or justification into
the prompt. They are for your filtering logic, not for the model — and they
cost tokens on every call.
Choose limit by call site¶
| Call site | limit |
Why |
|---|---|---|
| Tool call, focused sub-task | 3 | The task is narrow; more is noise |
| Top-of-task planning | 5–8 | The model needs the shape of the project |
| Human-facing summary | 10–20 | A person filters better than a floor does |
| Analysis / export | 50+ | Not a prompt at all |
Query with the task, not with keywords¶
The query is embedded and compared against memory content. Give it the real sentence:
recall("implement the refund endpoint with the approval rule") # ✓
recall("refund") # ✗ under-specified
recall(entire_conversation_history) # ✗ diluted
A one-word query embeds to something generic and matches everything weakly. The whole transcript embeds to an average of every subject in it and matches nothing sharply.
Cache within a task¶
class TaskContext:
"""One retrieval per task. Sub-steps reuse it rather than re-querying."""
def __init__(self, task, limit=8):
self._block = render(pack(dedupe(recall(task, limit=limit))))
def block(self) -> str:
return self._block
Recall again only when the subject changes — a sub-agent working on a different component should query for its own component.
Measure¶
import time
t0 = time.perf_counter()
memories = recall(task, limit=10)
latency = time.perf_counter() - t0
selected = pack([m for m in memories if m["score"] >= 0.35])
print(f"retrieved={len(memories)} kept={len(selected)} "
f"chars={sum(len(m['content']) for m in selected)} latency={latency:.3f}s "
f"top_score={memories[0]['score'] if memories else 0:.3f}")
Log those five numbers per call. kept/retrieved falling and top_score
falling are the two early signals that your store needs consolidation — see
Long-running agents.
Related¶
- Context selection — what is and is not implemented
- Ranking — the score you are filtering on
- Evaluation — measuring whether any of this helps