The artifact store

Chapter 06 ends with an unsolved problem, and it is the most serious one in the whole system.

The tiny agent read 38,000 tokens and returned 118. So it decided what matters. Except it does not know the overall task - it sees a brief, not the conversation. Perhaps the caller wanted to know afterwards which countries were involved. The agent threw that paragraph away, because the brief only asked about start and end.

Compression without a way back is data loss with better PR.

The goal

Compression is only defensible once it is reversible. The caller gets the compact answer and keeps access to the original. That is the difference between a summary and a loss.

The idea in one sentence

Throw nothing away, just do not send it along.

tool output (38k tokens)
   ├─> full text to the file system   -> artifact_id
   ├─> cut into chunks (~512 tokens, 64 overlap)
   ├─> embed the chunks               -> in-memory matrix, as in chapter 03
   └─> handle to the caller           -> "art_7f3a1c"

And then expand(handle, question): a vector search inside this one artifact that returns the three most relevant chunks. Typically 400 to 1,500 tokens instead of 38,000.

Step 1: the store

# tiny/artifacts.py
from __future__ import annotations

import hashlib
import json
import time
from dataclasses import dataclass
from pathlib import Path

import numpy as np

from tiny.llm import embed

STORE = Path(".artifacts")
CHUNK = 2000       # characters, roughly 500 tokens
OVERLAP = 250      # so a sentence on the seam is not lost
TTL_HOURS = 24


@dataclass
class Artifact:
    handle: str
    source: str
    tokens_estimated: int
    preview: str


def _chunks(text: str) -> list[str]:
    step = CHUNK - OVERLAP
    return [text[i : i + CHUNK] for i in range(0, len(text), step)] or [""]


def store(text: str, source: str) -> Artifact:
    STORE.mkdir(exist_ok=True)
    handle = "art_" + hashlib.sha256(f"{source}{text[:400]}".encode()).hexdigest()[:8]
    folder = STORE / handle
    folder.mkdir(exist_ok=True)

    (folder / "fulltext.txt").write_text(text, encoding="utf-8")
    chunks = _chunks(text)
    (folder / "chunks.json").write_text(json.dumps(chunks), encoding="utf-8")

    vectors = np.array(embed(chunks), dtype=np.float32)
    vectors /= np.maximum(np.linalg.norm(vectors, axis=1, keepdims=True), 1e-12)
    np.save(folder / "vectors.npy", vectors)

    (folder / "meta.json").write_text(
        json.dumps({"source": source, "created": time.time()}), encoding="utf-8"
    )
    return Artifact(handle, source, len(text) // 4, text[:200].replace("\n", " "))


def expand(handle: str, question: str, k: int = 3) -> str:
    """Vector search INSIDE one artifact. The rest stays where it is."""
    folder = STORE / handle
    if not folder.exists():
        return f"Artifact {handle} does not exist (any more)."

    chunks = json.loads((folder / "chunks.json").read_text(encoding="utf-8"))
    vectors = np.load(folder / "vectors.npy")

    q = np.array(embed([question])[0], dtype=np.float32)
    q /= max(float(np.linalg.norm(q)), 1e-12)
    best = np.argsort(-(vectors @ q))[:k]
    return "\n\n---\n\n".join(chunks[i] for i in sorted(best))


def cleanup() -> int:
    """Artifacts are a session cache, not an archive."""
    limit = time.time() - TTL_HOURS * 3600
    removed = 0
    for folder in STORE.glob("art_*"):
        meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
        if meta["created"] < limit:
            for file in folder.iterdir():
                file.unlink()
            folder.rmdir()
            removed += 1
    return removed

Two decisions in there are deliberate, not convenient.

The overlap. Without it a sentence sitting exactly on the cut line ends up half in each chunk - and is found in neither. 250 characters are roughly one or two sentences and cost 12 percent more storage. A good trade.

The expiry. Artifacts are a session cache. Keep them and after two weeks you have an unintended archive of scraped third-party content on disk - with everything that means legally and practically. cleanup() belongs in a cron job, at the latest the day after the prototype.

Step 2: wire it into the runner

Remember the ugly 12,000-character emergency brake from chapter 05? It gets replaced now:

# tiny/runner.py, inside the tool loop
from tiny.artifacts import store

ARTIFACT_THRESHOLD = 16000   # characters, roughly 4000 tokens

...
            else:
                function = REGISTRY[name][0]
                content = str(function(**arguments))

            if len(content) > ARTIFACT_THRESHOLD:
                artifact = store(content, source=f"{name}:{arguments}")
                collected_artifacts.append(artifact)
                content = (
                    content[:ARTIFACT_THRESHOLD]
                    + f"\n\n[truncated. Full text under handle={artifact.handle}]"
                )

This first of all protects the tiny agent itself - otherwise a single scrape eats its own context before it gets round to condensing anything. And it produces the handle that travels on.

In the digest the handle comes up with it:

    return Result(
        ...,
        artifacts=[
            {
                "handle": a.handle,
                "source": a.source,
                "tokens": a.tokens_estimated,
                "preview": a.preview,
            }
            for a in collected_artifacts
        ],
    )

Those four fields cost around 40 tokens together. In exchange the caller knows that 38,000 more are lying around, where they came from and roughly how they start. That is enough to decide whether following up is worth it.

Step 3: the caller's third tool

The calling agent now has exactly three tools, no matter how many capabilities are registered:

# tiny/host.py
from tiny.artifacts import expand
from tiny.concierge import load_concierge
from tiny.runner import delegate

_concierge = load_concierge()


def find_capability(goal: str) -> dict:
    """Can anybody here do this? No LLM call, calibrated answer."""
    d = _concierge.route(goal)
    if d.kind == "abstain":
        return {"can_do": False, "confidence": round(d.score, 3)}
    card = _concierge.cards[d.agent_id]
    return {
        "can_do": True,
        "agent": card.id,
        "summary": card.summary,
        "limitations": card.does_not,
        "confidence": round(d.score, 3),
        "ambiguous": d.kind == "ambiguous",
        "alternatives": [a for a, _ in d.alternatives],
    }


def host_delegate(task: str) -> dict:
    return dict(delegate(_concierge, task))


def host_expand(handle: str, question: str) -> str:
    return expand(handle, question)

Three functions, three tool definitions in the main context. By the arithmetic from chapter 01 that is around 600 tokens - constant, whether three capabilities are registered or five hundred. That is the target figure of the whole architecture, and now it is there.

find_capability is the declarative answer to the question you first want to solve dialogically: just ask the agent whether it can do this. That would be an LLM call, a round of latency and an uncalibrated answer. Here it is a vector search with a number behind it.

The milestone

The test consists of two calls, and the second one is the point.

# scripts/07_expand_test.py
from tiny.host import host_delegate, host_expand

result = host_delegate(
    "read https://en.wikipedia.org/wiki/World_War_II and tell me "
    "when the war began and when it ended"
)
print("Digest  :", result["summary"][:120])
print("Fields  :", result["fields"])
for a in result["artifacts"]:
    print(f"Artifact: {a['handle']}  ~{a['tokens']} tokens  {a['source'][:40]}")

# This information was NOT in the digest - nobody asked for it.
handle = result["artifacts"][0]["handle"]
follow_up = host_expand(handle, "which countries took part in the wannsee conference")
print(f"\nFollow-up: ~{len(follow_up) // 4} tokens")
print(follow_up[:400])

Digest  : The Second World War began on 1 September 1939 with the invasion of Poland ...
Fields  : {'start': '1 September 1939', 'end': '2 September 1945'}
Artifact: art_7f3a1c  ~38412 tokens  http_get:{'url': 'https://en.wikipe

Follow-up: ~610 tokens
... the Wannsee Conference on 20 January 1942 ...

Now the bookkeeping, because only that proves anything:

The state is reached when you can retrieve a piece of information that did not appear in the digest - while the 38,000 tokens never show up in your main context. They are on disk. Reachable, but not present.

That is the difference this was all about.