Add Venkai to an agent¶

A working integration in one file, then the decisions behind each part.

The complete integration¶

"""venkai_memory.py — the whole integration. Two calls."""
import os
import requests

BASE = os.environ.get("VENKAI_BASE_URL", "https://api.venkai.fr")
KEY = os.environ["VENKAI_API_KEY"]          # fail at import, not mid-run
PROJECT = "billing-service"                  # constant: a typo makes a new project

_H = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}
_TIMEOUT = 10


def load_context(task: str, limit: int = 5, min_score: float = 0.0,
                 max_chars: int = 4000) -> str:
    """Ranked memories for `task`, as a prompt block.

    Venkai ranks and bounds by count; the score floor and the character
    budget are the caller's job — neither exists server-side.
    """
    r = requests.get(f"{BASE}/api/context/{PROJECT}/relevant",
                     headers=_H, params={"query": task, "limit": limit},
                     timeout=_TIMEOUT)
    r.raise_for_status()

    selected, used = [], 0
    for m in r.json()["memories"]:
        if m["score"] < min_score:
            continue
        cost = len(m["content"])
        if used + cost > max_chars:
            continue           # skip, not break: a shorter high-scorer may still fit
        selected.append(f"- [{m['type']}] {m['content']}")
        used += cost

    if not selected:
        return ""
    return "Established context for this project:\n" + "\n".join(selected)


def record(content: str, type: str = "decision", importance: float = 0.8,
           agent: str = "planner", metadata: dict | None = None) -> None:
    """Store a conclusion. Failing to remember must not fail the task."""
    try:
        requests.post(f"{BASE}/api/memory", headers=_H, timeout=_TIMEOUT, json={
            "project_id": PROJECT, "agent_id": agent, "content": content,
            "type": type, "importance": importance, "metadata": metadata or {},
        }).raise_for_status()
    except requests.RequestException as exc:
        # ponytail: log and continue. Memory is an enhancement; an agent that
        # crashes because it could not write a note is worse than a forgetful one.
        print(f"[venkai] could not record memory: {exc}")

Use it:

def run(task: str):
    prompt = f"{load_context(task)}\n\nTask: {task}"
    result = call_your_llm(prompt)
    if result.decision:
        record(result.decision, type="decision", importance=0.85)
    return result

That is the integration. The rest of this page is why.

Decisions in that code¶

Read fails loudly, write fails quietly¶

load_context raises: an agent running without its context will produce confidently wrong output, and you want to know. record swallows: losing one note is a small loss, crashing a task is a big one.

The project key is a module constant¶

Writing to a key that does not exist creates it and returns 200. There is no error anywhere in the typo path — the write succeeds into a new empty project, and later reads on the correct key come back empty. A constant removes the class of bug.

The budget is enforced client-side¶

limit bounds the number of memories, not their size. min_score and max_chars are the caller's, because neither exists server-side. See Context selection.

`continue`, not `break`¶

Skipping an oversized memory lets a shorter, still-highly-ranked one through. Breaking on the first overflow throws away the tail of a good list.

Where the calls belong¶

flowchart TB
  T["task arrives"] --> L["load_context(task)"]
  L --> P["build prompt"]
  P --> M["LLM"]
  M --> A["act / iterate"]
  A --> D{"did we conclude<br/>something durable?"}
  D -->|yes| R["record(...)"]
  D -->|no| E["done"]
  R --> E

Read once per task, at the top — not per LLM call. A ten-step agent that recalls each step pays ten times and repeats the same memories in every prompt.

Write on conclusions, not observations:

Write it	Skip it
"We chose Postgres because the ranking query needs joins"	"I am reading config.py"
"Refunds over 500 EUR need a second approver"	Tool output you can re-read
"The earlier assumption about tz handling was wrong"	The step-by-step trace
"This user prefers terse answers"	Anything already in the repo

Every low-value row competes for retrieval budget forever. Restraint on the write path is what keeps the read path useful.

Placing the block in the prompt¶

messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "system", "content": load_context(task)},   # before the task
    {"role": "user", "content": task},
]

Keep it in a system message, distinct from the user's words, and labelled as established context. Merged into the user turn, models start treating stored constraints as part of the current request and try to act on them.

Letting the model write¶

Give the model record as a tool — it knows when it concluded something and you do not:

TOOLS = [{
    "name": "remember",
    "description": (
        "Record a durable conclusion about this project: a decision and its "
        "reason, a hard constraint, or a correction of an earlier belief. "
        "Do NOT record intermediate steps, tool output, or anything already "
        "in the repository."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "content": {"type": "string"},
            "type": {"enum": ["fact", "decision", "preference",
                              "event", "constraint", "relationship"]},
            "importance": {"type": "number", "minimum": 0, "maximum": 1},
        },
        "required": ["content", "type"],
    },
}]

Do not do the same for recall. Models call it with vague queries or forget entirely; deterministic loading at the top of the task is more reliable.

Verify it works¶

# 1. the memory landed
curl -s "https://api.venkai.fr/api/context/billing-service?limit=5" \
  -H "Authorization: Bearer $VENKAI_API_KEY" | python -m json.tool

# 2. it comes back for a plausible query
curl -sG https://api.venkai.fr/api/context/billing-service/relevant \
  -H "Authorization: Bearer $VENKAI_API_KEY" \
  --data-urlencode 'query=database choice' --data-urlencode 'limit=3'

If (1) shows rows and (2) does not return them, the ranking is the issue, not the integration — raise importance, or switch to the semantic embedding provider (Embeddings).