Skip to content

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:

memories = [m for m in resp["memories"] if m["score"] >= 0.35]

Choosing the threshold:

  1. Run 20 representative queries.
  2. Read the results and mark each useful / not.
  3. Pick the score that separates them best.
  4. Re-check it when Venkai announces a ranking change.

Absolute scores are not a contract

Scores are meaningful relative to each other within a response. Ranking improvements can shift absolute values, so keep your floor in configuration, not hard-coded, and re-validate it periodically.

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.
    Go further only if you measure that paraphrase duplicates cost 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 matched 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.