খরচ ও লেটেন্সির মূল কথা

import json
from pathlib import Path

from evidencedesk.chunking import chunk_corpus
from evidencedesk.context import estimate_tokens
from evidencedesk.rag import RAG_RULES, SAMPLE_REPLIES, Retriever, build_request

chunks = chunk_corpus(Path("kit/corpus"))
retriever = Retriever(chunks, clock=lambda: "2026-10-12T09:00:00Z")
Q = "How long do I have to return a bike?"
found = retriever.retrieve(Q)
request = build_request(Q, found)
print("part".ljust(26), "tokens")
print("system rules".ljust(26), estimate_tokens(RAG_RULES))
for r in found:
    print(f"evidence {r.chunk.id}".ljust(26), estimate_tokens(r.chunk.text))
print("whole user message".ljust(26), estimate_tokens(request.messages[1].content), "(with tags)")
reply = estimate_tokens(json.dumps(SAMPLE_REPLIES[Q]))
print("reply (JSON)".ljust(26), reply, f"(the cap is max_tokens={request.max_tokens})")

print()
print("scenario".ljust(30), "calls  input  output  billed")
for label, k, calls in (("k=3, one call", 3, 1), ("k=1, one call", 1, 1), ("k=3, two retries", 3, 3)):
    size = sum(estimate_tokens(m.content) for m in build_request(Q, retriever.retrieve(Q, k=k)).messages)
    print(f"{label:<30}{calls:>5}{size * calls:>7}{reply * calls:>8}{(size + reply) * calls:>8}")

part                       tokens
system rules               116
evidence kc-returns-v3:c0  58
evidence kc-warranty-v2:c0 41
evidence kc-returns-v3:c1  31
whole user message         210 (with tags)
reply (JSON)               39 (the cap is max_tokens=200)

scenario                       calls  input  output  billed
k=3, one call                     1    326      39     365
k=1, one call                     1    215      39     254
k=3, two retries                  3    978     117    1095

# file: kit/evidencedesk/usage.py
"""Cost and latency: a hypothetical price table, a usage ledger, and clocks you can inject."""
from __future__ import annotations

import time
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass
from decimal import Decimal

from evidencedesk.context import estimate_tokens
from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse
from evidencedesk.streaming import Assembled, StreamEvent, assemble

# --- 1. A hypothetical price table ---
PER = Decimal(1_000_000)  # rates are "credits per 1M tokens"; a credit is a made-up unit


@dataclass(frozen=True)
class Rate:
    input: Decimal
    output: Decimal
    cached_input: Decimal


NOT_A_REAL_PRICE_TABLE: dict[str, Rate] = {  # invented numbers for tests; never a real provider price
    "fake-model": Rate(Decimal("3"), Decimal("12"), Decimal("0.75")),
    "fake-small": Rate(Decimal("0.6"), Decimal("2.4"), Decimal("0.15")),
}


class UnknownModel(KeyError):
    """No rate for this model: stop, do not count it as free."""

# file: kit/evidencedesk/usage.py (append)
# --- 2. Usage records ---
def request_tokens(request: LLMRequest) -> int:
    """Rough input size: ceil(chars / 4) per message."""
    return sum(estimate_tokens(m.content) for m in request.messages)


@dataclass(frozen=True)
class UsageRecord:
    request_id: str
    model: str
    input_tokens: int  # ALL input tokens, cached ones included
    output_tokens: int
    cached_tokens: int = 0  # the part of input_tokens a provider cache served
    estimated: bool = False  # True: no usage from the provider, we guessed
    attempt: int = 1  # 1 = first try, 2 and more = a retry

    def __post_init__(self) -> None:
        if min(self.input_tokens, self.output_tokens, self.cached_tokens) < 0 or self.cached_tokens > self.input_tokens:
            raise ValueError("token counts must be >= 0 and cached_tokens <= input_tokens")
        if self.attempt < 1:
            raise ValueError("attempt starts at 1")


def record_usage(request: LLMRequest, response: LLMResponse, request_id: str, *, cached_tokens: int = 0,
                 attempt: int = 1) -> UsageRecord:
    """Keep reported usage; if there is none (0 + 0 with text), estimate it and flag it."""
    usage = response.usage
    if usage.total_tokens == 0 and response.text:
        out = min(estimate_tokens(response.text), request.max_tokens)
        return UsageRecord(request_id, response.model, request_tokens(request), out, 0, True, attempt)
    return UsageRecord(request_id, response.model, usage.input_tokens, usage.output_tokens, cached_tokens, False, attempt)


def cost_of(record: UsageRecord, table: dict[str, Rate] = NOT_A_REAL_PRICE_TABLE) -> Decimal:
    """Exact cost in made-up credits: fresh input + cached input + output."""
    try:
        rate = table[record.model]
    except KeyError:
        raise UnknownModel(record.model) from None
    fresh = record.input_tokens - record.cached_tokens
    return (fresh * rate.input + record.cached_tokens * rate.cached_input + record.output_tokens * rate.output) / PER


@dataclass(frozen=True)
class Totals:
    calls: int
    retries: int
    input_tokens: int
    output_tokens: int
    cached_tokens: int
    estimated: int
    cost: Decimal


class Ledger:
    """One record per model call. A cache hit makes no call, so it adds none."""

    def __init__(self, table: dict[str, Rate] = NOT_A_REAL_PRICE_TABLE) -> None:
        self.table, self.records = table, []

    def add(self, record: UsageRecord) -> Decimal:
        cost = cost_of(record, self.table)  # raises UnknownModel first
        self.records.append(record)
        return cost

    def summary(self) -> Totals:
        r = self.records
        return Totals(len(r), sum(x.attempt > 1 for x in r), sum(x.input_tokens for x in r), sum(x.output_tokens for x in r),
                      sum(x.cached_tokens for x in r), sum(x.estimated for x in r), sum((cost_of(x, self.table) for x in r), Decimal(0)))

from dataclasses import replace

from evidencedesk.llm import FakeClient, LLMRequest, Message
from evidencedesk.rag import SAMPLE_REPLIES, RecordedReplies
from evidencedesk.usage import NOT_A_REAL_PRICE_TABLE, Ledger, UnknownModel, UsageRecord, cost_of, record_usage

for name, rate in NOT_A_REAL_PRICE_TABLE.items():
    print(name, "per 1M tokens: input", rate.input, "| output", rate.output, "| cached input", rate.cached_input)

ask = LLMRequest((Message("user", "How long do I have to return a new bike?"),))
reported = record_usage(ask, FakeClient().complete(ask), "req-001")
rag_request = build_request(Q, found)
silent = record_usage(rag_request, RecordedReplies(SAMPLE_REPLIES).complete(rag_request), "req-002")
cached = UsageRecord("req-003", "fake-model", input_tokens=reported.input_tokens, output_tokens=19, cached_tokens=30)
try:
    cost_of(silent)
except UnknownModel as error:
    print("no rate for", error)
silent = replace(silent, model="fake-model")  # price it as the model it stands in for
for rec in (reported, silent, cached):
    print(rec.request_id, rec.input_tokens, rec.output_tokens, rec.cached_tokens, "estimated" if rec.estimated else "reported", cost_of(rec))

ledger = Ledger()
for attempt in (1, 2, 3):  # the same call, tried three times
    ledger.add(UsageRecord("req-004", "fake-model", 400, 60, attempt=attempt))
print(ledger.summary())

fake-model per 1M tokens: input 3 | output 12 | cached input 0.75
fake-small per 1M tokens: input 0.6 | output 2.4 | cached input 0.15
no rate for 'recorded'
req-001 41 19 0 reported 0.000351
req-002 326 39 0 estimated 0.001446
req-003 41 19 30 reported 0.0002835
Totals(calls=3, retries=2, input_tokens=1200, output_tokens=180, cached_tokens=0, estimated=0, cost=Decimal('0.00576'))

# file: kit/evidencedesk/usage.py (append)
# --- 3. Latency with an injected clock ---
Clock = Callable[[], float]


class ManualClock:
    """Moves only when told. No wall clock."""

    def __init__(self, start: float = 0.0) -> None:
        self.now = start

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


class Timer:
    """`with Timer(clock) as t: ...`, then `t.elapsed`. Tests pass a ManualClock."""

    def __init__(self, clock: Clock = time.perf_counter) -> None:
        self.clock, self.start, self.elapsed = clock, 0.0, 0.0

    def __enter__(self) -> Timer:
        self.start = self.clock()
        return self

    def __exit__(self, *exc: object) -> None:
        self.elapsed = self.clock() - self.start


class SlowClient:
    """Moves a ManualClock: a base delay plus a delay per output token. A stand-in for the network."""

    def __init__(self, inner: LLMClient, clock: ManualClock, base_s: float = 0.4, per_output_token_s: float = 0.02) -> None:
        self.inner, self.clock, self.base_s, self.per_token_s = inner, clock, base_s, per_output_token_s

    def complete(self, request: LLMRequest) -> LLMResponse:
        response = self.inner.complete(request)
        self.clock.advance(self.base_s + self.per_token_s * response.usage.output_tokens)
        return response

from evidencedesk.usage import ManualClock, SlowClient, Timer

clock = ManualClock()
slow = SlowClient(FakeClient(), clock)  # 0.4 s + 0.02 s per output token
with Timer(clock) as t:
    slow.complete(ask)
print("fake clock:", round(t.elapsed, 3), "s for 19 output tokens")

with Timer() as real:  # the default clock is time.perf_counter()
    FakeClient().complete(ask)
print("real clock used:", real.elapsed >= 0, "(value not printed)")

fake clock: 0.78 s for 19 output tokens
real clock used: True (value not printed)

# file: kit/evidencedesk/usage.py (append)
# --- 4. First-chunk and full-response latency ---
@dataclass(frozen=True)
class StreamTiming:
    first_chunk_s: float | None  # None: no chunk arrived
    total_s: float
    chunks: int


def slow_stream(events: Iterable[StreamEvent], clock: ManualClock, first_s: float = 0.4, per_chunk_s: float = 0.05) -> Iterator[StreamEvent]:
    """Delay a fake stream: first chunk after first_s, each later one per_chunk_s after the last."""
    first = True
    for event in events:
        if event.kind == "chunk":
            clock.advance(first_s if first else per_chunk_s)
            first = False
        yield event


def timed_stream(events: Iterable[StreamEvent], clock: Clock = time.perf_counter) -> tuple[Assembled, StreamTiming]:
    """Read a stream to its end; record time to the first chunk and to the last event."""
    start, first, chunks, seen = clock(), None, 0, []
    for event in events:
        if event.kind == "chunk":
            chunks += 1
            if first is None:
                first = clock() - start
        seen.append(event)
    return assemble(seen), StreamTiming(first, clock() - start, chunks)

from evidencedesk.streaming import FakeStreamClient
from evidencedesk.usage import slow_stream, timed_stream

clock = ManualClock()
events = slow_stream(FakeStreamClient(chunk_size=20).stream(ask), clock, first_s=0.4, per_chunk_s=0.1)
done, timing = timed_stream(events, clock)
print(done.status, "|", done.text)
print(f"chunks={timing.chunks}  first chunk after {timing.first_chunk_s:.2f} s  whole answer after {timing.total_s:.2f} s")

complete | You can return a new, unridden bike within 30 days of delivery for a full refund.
chunks=5  first chunk after 0.40 s  whole answer after 0.80 s

# file: kit/evidencedesk/cache.py
"""Exact-match response cache: a deterministic key, a policy per request class, a TTL on an injected clock."""
from __future__ import annotations

import hashlib
import json
import re
import time
from collections.abc import Callable, Iterable
from dataclasses import dataclass

from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse
from evidencedesk.rag import Retrieved


# --- 1. The key ---
def normalise(text: str) -> str:
    """Collapse whitespace; keep case (RD2-24 is not rd2-24)."""
    return re.sub(r"\s+", " ", text).strip()


def evidence_refs(retrieved: Iterable[Retrieved]) -> tuple[str, ...]:
    """The evidence set as chunk-id@version."""
    return tuple(f"{r.chunk.id}@{r.chunk.version}" for r in retrieved)


def cache_key(request: LLMRequest, *, prompt_version: str, evidence: Iterable[str] = (), corpus_version: str = "",
              scope: str = "") -> str:
    """Hash of everything that can change the answer. Nothing random in it: no ids, no times."""
    material = {
        "model": request.model, "max_tokens": request.max_tokens,
        "messages": [[m.role, normalise(m.content)] for m in request.messages],
        "prompt_version": prompt_version, "evidence": sorted(evidence), "corpus_version": corpus_version, "scope": scope,
    }
    text = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

from evidencedesk.cache import cache_key, evidence_refs

base = dict(prompt_version="v1", evidence=evidence_refs(found), corpus_version="2026-10-12")
rag_a = build_request(Q, found)
variants = {
    "same request again": (rag_a, {}),
    "extra spaces in the question": (build_request("  How long do I   have to return a bike? ", found), {}),
    "different model": (LLMRequest(rag_a.messages, model="fake-small"), {}),
    "different max_tokens": (LLMRequest(rag_a.messages, max_tokens=50), {}),
    "different letter case": (build_request(Q.lower(), found), {}),
    "prompt_version v2": (rag_a, {"prompt_version": "v2"}),
    "one evidence chunk fewer": (rag_a, {"evidence": evidence_refs(found[:-1])}),
    "corpus_version changed": (rag_a, {"corpus_version": "2026-10-13"}),
    "scope user:nadia": (rag_a, {"scope": "user:nadia"}),
}
key0 = cache_key(rag_a, **base)
print("base key:", key0[:8], "...")
for label, (req, change) in variants.items():
    print(f"{label:<30}", "same key" if cache_key(req, **{**base, **change}) == key0 else "NEW key")

base key: 81f3f88d ...
same request again             same key
extra spaces in the question   same key
different model                NEW key
different max_tokens           NEW key
different letter case          NEW key
prompt_version v2              NEW key
one evidence chunk fewer       NEW key
corpus_version changed         NEW key
scope user:nadia               NEW key

# file: kit/evidencedesk/cache.py (append)
# --- 2. The store ---
@dataclass
class _Entry:
    response: LLMResponse
    stored_at: float
    tags: frozenset[str]


class ResponseCache:
    """In memory. Entries expire after ttl_s on the injected clock; the oldest goes first when full."""

    def __init__(self, ttl_s: float = 300.0, clock: Callable[[], float] = time.monotonic, max_entries: int = 256) -> None:
        self.ttl_s, self.clock, self.max_entries = ttl_s, clock, max_entries
        self.entries: dict[str, _Entry] = {}
        self.hits = self.misses = self.expired = 0

    def __len__(self) -> int:
        return len(self.entries)

    def get(self, key: str) -> LLMResponse | None:
        entry = self.entries.get(key)
        if entry is not None and self.clock() - entry.stored_at >= self.ttl_s:
            del self.entries[key]
            self.expired += 1
            entry = None
        if entry is None:
            self.misses += 1
            return None
        self.hits += 1
        return entry.response

    def put(self, key: str, response: LLMResponse, tags: Iterable[str] = ()) -> None:
        if key not in self.entries and len(self.entries) >= self.max_entries:
            del self.entries[next(iter(self.entries))]
        self.entries[key] = _Entry(response, self.clock(), frozenset(tags))

    def invalidate(self, tag: str) -> int:
        """Remove entries with this tag; return how many."""
        doomed = [k for k, e in self.entries.items() if tag in e.tags]
        for key in doomed:
            del self.entries[key]
        return len(doomed)

    def clear(self) -> None:
        self.entries.clear()

# file: kit/evidencedesk/cache.py (append)
# --- 3. The policy and the client ---
@dataclass(frozen=True)
class CachePolicy:
    """Which request classes may be cached. The default: none."""
    shared: frozenset[str] = frozenset()  # any user may get the cached answer
    per_user: frozenset[str] = frozenset()  # cached inside one user's scope only


NO_CACHE = CachePolicy()
FAQ_ONLY = CachePolicy(shared=frozenset({"faq"}))


@dataclass(frozen=True)
class Cached:
    response: LLMResponse
    status: str  # hit | miss | bypass
    key: str | None = None  # None for bypass


class CachedClient:
    """`ask` says what happened (hit, miss, bypass); `complete` keeps the LLMClient shape."""

    def __init__(self, inner: LLMClient, cache: ResponseCache, policy: CachePolicy = NO_CACHE, prompt_version: str = "v1") -> None:
        self.inner, self.cache, self.policy, self.prompt_version = inner, cache, policy, prompt_version
        self.bypassed = 0

    def ask(self, request: LLMRequest, *, request_class: str = "unclassified", user_id: str | None = None,
            evidence: Iterable[str] = (), corpus_version: str = "") -> Cached:
        if request_class in self.policy.shared:
            scope = ""
        elif request_class in self.policy.per_user and user_id:
            scope = f"user:{user_id}"
        else:  # not allowed, or no user to isolate by: fail closed
            self.bypassed += 1
            return Cached(self.inner.complete(request), "bypass")
        refs = tuple(evidence)
        key = cache_key(request, prompt_version=self.prompt_version, evidence=refs, corpus_version=corpus_version, scope=scope)
        found = self.cache.get(key)
        if found is not None:
            return Cached(found, "hit", key)
        response = self.inner.complete(request)
        if response.stop_reason == "end":  # a cut-off answer is never cached
            tags = {f"prompt:{self.prompt_version}", f"model:{request.model}", f"corpus:{corpus_version}"}
            tags |= {"doc:" + ref.split(":")[0] for ref in refs}
            self.cache.put(key, response, tags)
        return Cached(response, "miss", key)

    def complete(self, request: LLMRequest, **kwargs: object) -> LLMResponse:
        return self.ask(request, **kwargs).response  # type: ignore[arg-type]

from evidencedesk.cache import FAQ_ONLY, NO_CACHE, CachedClient, ResponseCache
from evidencedesk.usage import Ledger

def run(policy, n=5):
    clock, inner = ManualClock(), FakeClient()
    client, ledger = CachedClient(inner, ResponseCache(300, clock), policy), Ledger()
    for i in range(n):
        got = client.ask(ask, request_class="faq")
        if got.status != "hit":  # a hit makes no model call, so there is nothing to record
            ledger.add(record_usage(ask, got.response, f"req-{i}"))
    return client, inner, ledger.summary()

for label, policy in (("no cache", NO_CACHE), ("faq cached", FAQ_ONLY)):
    client, inner, total = run(policy)
    print(f"{label:<11} model calls={len(inner.calls)} tokens={total.input_tokens + total.output_tokens} cost={total.cost}"
          f" hits={client.cache.hits} misses={client.cache.misses} bypassed={client.bypassed}")

no cache    model calls=5 tokens=300 cost=0.001755 hits=0 misses=0 bypassed=5
faq cached  model calls=1 tokens=60 cost=0.000351 hits=4 misses=1 bypassed=0

from dataclasses import replace

clock, inner = ManualClock(), FakeClient()
client = CachedClient(inner, ResponseCache(3600, clock), FAQ_ONLY, prompt_version="v1")
refs = evidence_refs(found)
print("chunk versions:", [ref.split("@")[1] for ref in refs], "->", [client.ask(ask, request_class="faq", evidence=refs).status for _ in range(2)])

newer = [replace(r.chunk, version=r.chunk.version + 1) for r in found]  # the returns page was updated
refs_v2 = tuple(f"{c.id}@{c.version}" for c in newer)
print("evidence changed        ->", client.ask(ask, request_class="faq", evidence=refs_v2).status)
client.prompt_version = "v2"
print("prompt version changed  ->", client.ask(ask, request_class="faq", evidence=refs).status)
print("entries before purge:", len(client.cache), "| removed by tag doc:kc-returns-v3:", client.cache.invalidate("doc:kc-returns-v3"),
      "| left:", len(client.cache))
client.prompt_version = "v1"
print("old request after purge ->", client.ask(ask, request_class="faq", evidence=refs).status)

chunk versions: ['3', '2', '3'] -> ['miss', 'hit']
evidence changed        -> miss
prompt version changed  -> miss
entries before purge: 3 | removed by tag doc:kc-returns-v3: 3 | left: 0
old request after purge -> miss

from evidencedesk.cache import CachePolicy

policy = CachePolicy(shared=frozenset({"faq"}), per_user=frozenset({"orders"}))
clock, inner = ManualClock(), FakeClient()
client = CachedClient(inner, ResponseCache(60, clock), policy)
steps = [
    ("faq, nadia", dict(request_class="faq", user_id="nadia")),
    ("faq, omar (shared)", dict(request_class="faq", user_id="omar")),
    ("orders, nadia", dict(request_class="orders", user_id="nadia")),
    ("orders, nadia again", dict(request_class="orders", user_id="nadia")),
    ("orders, omar", dict(request_class="orders", user_id="omar")),
    ("orders, no user", dict(request_class="orders")),
    ("account (not allowed)", dict(request_class="account", user_id="nadia")),
]
for label, kw in steps:
    print(f"{label:<24}{client.ask(ask, **kw).status}")
clock.advance(61)
print(f"{'faq after 61 s (ttl 60)':<24}{client.ask(ask, request_class='faq').status}")
print("model calls:", len(inner.calls), "| entries:", len(client.cache))

faq, nadia              miss
faq, omar (shared)      hit
orders, nadia           miss
orders, nadia again     hit
orders, omar            miss
orders, no user         bypass
account (not allowed)   bypass
faq after 61 s (ttl 60) miss
model calls: 6 | entries: 3

python -m pytest -q kit/tests/test_usage.py kit/tests/test_cache.py | sed 's/ in [0-9.]*s//'

........................                                                 [100%]
24 passed

from evidencedesk.usage import Ledger, record_usage

small, ledger = FakeClient(), Ledger()
for n, question in enumerate(small.fixtures, 1):
    req = LLMRequest((Message("user", question),), model="fake-small")
    resp = small.complete(req)
    ledger.add(record_usage(req, resp, f"e-{n}"))
    print(question[:32].ljust(34), resp.stop_reason.ljust(7), resp.usage.total_tokens)
print(ledger.summary())

How long do I have to return a n   end     60
How long does standard delivery    end     58
How should I store the e-bike ba   length  65
Totals(calls=3, retries=0, input_tokens=124, output_tokens=59, cached_tokens=0, estimated=0, cost=Decimal('0.0002160'))

from evidencedesk.context import estimate_tokens
from evidencedesk.streaming import FakeStreamClient
from evidencedesk.usage import request_tokens, slow_stream, timed_stream

clock = ManualClock()
events = slow_stream(FakeStreamClient(chunk_size=20, stop_after=2).stream(ask), clock)
done, timing = timed_stream(events, clock)
print(done.status, "| usage:", done.usage, "|", timing)
rec = UsageRecord("e-stream", "fake-model", request_tokens(ask), estimate_tokens(done.text), estimated=True)
print("estimated:", rec.estimated, "| cost:", cost_of(rec))

interrupted | usage: None | StreamTiming(first_chunk_s=0.4, total_s=0.45, chunks=2)
estimated: True | cost: 0.00015

from dataclasses import replace

from evidencedesk.cache import FAQ_ONLY, CachedClient, ResponseCache, evidence_refs
from evidencedesk.rag import SAMPLE_REPLIES, RecordedReplies, Retriever, build_request


def cached_rag(client, retriever, question):
    found = retriever.retrieve(question)
    return client.ask(build_request(question, found), request_class="faq", evidence=evidence_refs(found))


replies = RecordedReplies(SAMPLE_REPLIES)
client = CachedClient(replies, ResponseCache(3600, ManualClock()), FAQ_ONLY)
print([cached_rag(client, retriever, Q).status for _ in range(2)], "model calls:", len(replies.calls))
bumped = [replace(c, version=c.version + 1) if c.doc_id == "kc-returns-v3" else c for c in chunks]
print(cached_rag(client, Retriever(bumped, clock=lambda: "2026-10-12T09:00:00Z"), Q).status, "model calls:", len(replies.calls))

['miss', 'hit'] model calls: 1
miss model calls: 2