The call record: every model call, for the operator's study (M6.6a)
nisse-core records each model call of every agent into a file beside its state store (<state>.calls.db), for the status page (L7, X6): - the request body exactly as sent (never headers), the raw response with its reasoning, the model asked and the one that answered, time, tokens and llama-server's own prompt reuse (timings.cache_n); - its work item, role, run and purpose: Driver.run and nisse-dev's delegation loop open runs, nested; compaction, expert consults, the classifier and the go-ahead reader label their calls; - provenance, worked out in BudgetedClient by comparing the events at each stage: each event sent with its log id and content as sent, which a compaction replaced, which fitting shortened (by how much) and dropped. Content is stored once by hash, compressed. It is diagnostic only: the agent never reads it, a failed write is logged once and recording stops while the call goes on, and the file is deleted when the agent starts, so its layout never migrates; CallReader refuses a layout it doesn't know. While recording, the system prompt ends with one sentence saying so. `call_record: false` in either agent's YAML turns it off. The benchmark records each run beside that run's own state store, and `--keep-runs DIR` keeps both for the page. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
This commit is contained in:
1 parent
cba35eb7d4
commit
dd060566b2
22 files changed
+1346
-48
No files matched your search
@@ -15,6 +15,7 @@ from typing import Any, Literal, Protocol
|
||||
|
||||
from nisse_core.log.canonical import AssistantText, Fact, ToolCall, ToolResult
|
||||
from nisse_core.loop.tools import names
|
||||
from nisse_core.models import calls
|
||||
from nisse_core.models.compact import Compactor
|
||||
from nisse_core.models.llm import LLMClient, LLMError
|
||||
from nisse_core.state.cancel import Cancel, skip_calls
|
||||
@@ -93,7 +94,19 @@ class Driver:
|
||||
self._failure_limit = failure_limit
|
||||
self._cancel = cancel or Cancel()
|
||||
|
||||
def run(
|
||||
def run(self, transcript: Transcript, **options: Any) -> RunOutcome:
|
||||
"""One run, recorded as one (L7): its calls are its turns. See
|
||||
``_run`` for the options."""
|
||||
with calls.run(
|
||||
calls.recorder_of(self._client),
|
||||
work_item=transcript.work_item,
|
||||
role=transcript.actor,
|
||||
) as recorded:
|
||||
outcome = self._run(transcript, **options)
|
||||
recorded.end(outcome.kind, outcome.note)
|
||||
return outcome
|
||||
|
||||
def _run(
|
||||
self,
|
||||
transcript: Transcript,
|
||||
*,
|
||||
|
||||
@@ -32,6 +32,7 @@ from nisse_core.log.canonical import (
|
||||
StoredEvent,
|
||||
UserMessage,
|
||||
)
|
||||
from nisse_core.models import calls
|
||||
from nisse_core.models.llm import LLMClient, LLMError, ModelRefused, RoleMap
|
||||
from nisse_core.state.store import Transcript
|
||||
|
||||
@@ -156,18 +157,19 @@ class ExpertDesk:
|
||||
# consult neither depends on nor leaks repo context. It cannot
|
||||
# ride in the system prompt alone — the Messages API requires
|
||||
# at least one non-system message.
|
||||
produced = self._client.complete(
|
||||
[
|
||||
StoredEvent(
|
||||
id=0,
|
||||
ts="",
|
||||
event=UserMessage(text=question, source="ask_expert"),
|
||||
)
|
||||
],
|
||||
tools=None,
|
||||
model=model,
|
||||
system=system,
|
||||
)
|
||||
with calls.purpose("consult", role="expert"):
|
||||
produced = self._client.complete(
|
||||
[
|
||||
StoredEvent(
|
||||
id=0,
|
||||
ts="",
|
||||
event=UserMessage(text=question, source="ask_expert"),
|
||||
)
|
||||
],
|
||||
tools=None,
|
||||
model=model,
|
||||
system=system,
|
||||
)
|
||||
except ModelRefused as exc:
|
||||
# Not retried: the same question is declined again.
|
||||
return None, f"{exc}. Ask it differently, or work from the repository"
|
||||
|
||||
@@ -0,0 +1,493 @@
|
||||
"""The call record (architecture L7, X6): every model call an agent makes,
|
||||
kept for the operator to study, never for the agent to read.
|
||||
|
||||
Each call is recorded as it was sent and answered: the request body exactly
|
||||
(never its headers, so no credential), the raw response with its reasoning,
|
||||
the model asked and the one that answered, time, tokens and the server's own
|
||||
count of prompt reuse. Beside that, where the request came from: its work
|
||||
item, role and run, and for each event sent, its log id and its content as
|
||||
sent, with what compaction replaced, what fitting shortened or dropped. The
|
||||
status page reads a request against the state store it was made from.
|
||||
|
||||
It is diagnostic only. Nothing in the agent reads it, a write that fails is
|
||||
logged once and recording stops while the call goes ahead, and the file is
|
||||
deleted when the agent starts, so its layout never migrates: it carries a
|
||||
version, and a reader that doesn't know it says so.
|
||||
|
||||
Content is stored once by hash and compressed, so a conversation that grows
|
||||
by a turn a call costs about one copy of itself.
|
||||
|
||||
Runs and purposes reach the recorder through a context variable: the driver
|
||||
opens a run around its turns (``run``), and a call made for something else
|
||||
(a compaction, a consult) says so (``purpose``).
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import hashlib
|
||||
import json
|
||||
import sqlite3
|
||||
import threading
|
||||
import uuid
|
||||
import zlib
|
||||
from collections.abc import Iterator, Mapping, Sequence
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from nisse_core.log.canonical import StoredEvent, event_type, to_payload
|
||||
from nisse_core.oplog import OpLog
|
||||
|
||||
LAYOUT_VERSION = 1
|
||||
|
||||
# Said to the model, at the end of its system prompt, while its calls are
|
||||
# recorded (L7): it is the operator's record, and the agent should know.
|
||||
NOTICE = "Your model calls are recorded for the operator's study."
|
||||
|
||||
_SCHEMA = (
|
||||
"CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
|
||||
"CREATE TABLE blobs (hash TEXT PRIMARY KEY, data BLOB NOT NULL)",
|
||||
"""
|
||||
CREATE TABLE runs (
|
||||
id TEXT PRIMARY KEY,
|
||||
parent TEXT,
|
||||
work_item TEXT,
|
||||
role TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
started_at TEXT NOT NULL,
|
||||
ended_at TEXT,
|
||||
outcome TEXT,
|
||||
note TEXT
|
||||
)
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE calls (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at TEXT NOT NULL,
|
||||
seconds REAL NOT NULL,
|
||||
run TEXT,
|
||||
work_item TEXT,
|
||||
role TEXT,
|
||||
purpose TEXT,
|
||||
model TEXT NOT NULL,
|
||||
answered_by TEXT,
|
||||
outcome TEXT NOT NULL,
|
||||
error TEXT,
|
||||
tokens_in INTEGER,
|
||||
tokens_out INTEGER,
|
||||
tokens_cached INTEGER,
|
||||
cache_n INTEGER,
|
||||
prompt_n INTEGER,
|
||||
estimate INTEGER,
|
||||
factor REAL,
|
||||
route TEXT,
|
||||
status INTEGER,
|
||||
request TEXT,
|
||||
response TEXT,
|
||||
sent TEXT NOT NULL,
|
||||
compacted TEXT NOT NULL,
|
||||
shortened TEXT NOT NULL,
|
||||
dropped TEXT NOT NULL,
|
||||
trim TEXT
|
||||
)
|
||||
""",
|
||||
"CREATE INDEX calls_by_run ON calls (run, id)",
|
||||
"CREATE INDEX calls_by_item ON calls (work_item, id)",
|
||||
)
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
def path_for(state_db: Path) -> Path:
|
||||
"""Where an agent's call record lives: beside its state store, named
|
||||
after it, so two agents sharing a directory keep two records."""
|
||||
return state_db.with_name(f"{state_db.stem}.calls.db")
|
||||
|
||||
|
||||
def recorder_for(
|
||||
state_db: Path, enabled: bool, oplog: OpLog | None = None
|
||||
) -> "CallRecorder | None":
|
||||
"""An agent's recorder, as its configuration has it (L7): made, and so
|
||||
the old record deleted, when the agent starts."""
|
||||
return CallRecorder(path_for(state_db), oplog=oplog) if enabled else None
|
||||
|
||||
|
||||
def with_notice(system: str | None) -> str:
|
||||
return f"{system}\n\n{NOTICE}" if system else NOTICE
|
||||
|
||||
|
||||
# Where a call is made from, for its record.
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Scope:
|
||||
run: str | None = None
|
||||
parent: str | None = None
|
||||
work_item: str | None = None
|
||||
role: str | None = None
|
||||
purpose: str | None = None
|
||||
|
||||
|
||||
_OUTSIDE = Scope() # no run, no purpose: frozen, so safe to share
|
||||
_scope: ContextVar[Scope] = ContextVar("nisse_call_scope", default=_OUTSIDE)
|
||||
|
||||
|
||||
def current() -> Scope:
|
||||
return _scope.get()
|
||||
|
||||
|
||||
class RunHandle:
|
||||
"""Says how a run ended, for its record."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.outcome = ""
|
||||
self.note = ""
|
||||
|
||||
def end(self, outcome: str, note: str = "") -> None:
|
||||
self.outcome, self.note = outcome, note
|
||||
|
||||
|
||||
def recorder_of(client: Any) -> "CallRecorder | None":
|
||||
"""The recorder a client records into, through any wrappers (they pass
|
||||
attributes on); None if it records nothing."""
|
||||
found = getattr(client, "call_recorder", None)
|
||||
return found if isinstance(found, CallRecorder) else None
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def run(
|
||||
recorder: "CallRecorder | None",
|
||||
*,
|
||||
work_item: str | None,
|
||||
role: str,
|
||||
kind: str = "run",
|
||||
) -> Iterator[RunHandle]:
|
||||
"""A run: the calls made inside it are its turns. Nested in any run
|
||||
already open (a delegation inside a driver run)."""
|
||||
outer = _scope.get()
|
||||
run_id = uuid.uuid4().hex[:16]
|
||||
handle = RunHandle()
|
||||
if recorder is not None:
|
||||
recorder.run_started(run_id, outer.run, work_item, role, kind)
|
||||
token = _scope.set(
|
||||
Scope(run=run_id, parent=outer.run, work_item=work_item, role=role)
|
||||
)
|
||||
try:
|
||||
yield handle
|
||||
except BaseException as exc:
|
||||
handle.end("error", f"{type(exc).__name__}: {exc}"[:500])
|
||||
raise
|
||||
finally:
|
||||
_scope.reset(token)
|
||||
if recorder is not None:
|
||||
recorder.run_ended(run_id, handle.outcome, handle.note)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def purpose(name: str, *, role: str | None = None) -> Iterator[None]:
|
||||
"""Calls made inside are for ``name`` (a compaction, a consult), by
|
||||
``role`` when it isn't the run's."""
|
||||
outer = _scope.get()
|
||||
token = _scope.set(replace(outer, purpose=name, role=role or outer.role))
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_scope.reset(token)
|
||||
|
||||
|
||||
# What a call did to the events it was given (L6), for its record.
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Provenance:
|
||||
sent: Sequence[StoredEvent] # what went out, in order, as sent
|
||||
compacted: Mapping[int, int | None] # event id -> the compaction that replaced it
|
||||
shortened: Mapping[int, int] # event id -> characters fitting cut
|
||||
dropped: Sequence[int] # event ids fitting left out
|
||||
|
||||
|
||||
def _encode(value: Any) -> tuple[str, bytes]:
|
||||
text = json.dumps(value, ensure_ascii=False, sort_keys=False)
|
||||
digest = hashlib.sha256(text.encode()).hexdigest()[:32]
|
||||
return digest, zlib.compress(text.encode(), 6)
|
||||
|
||||
|
||||
class CallRecorder:
|
||||
"""Writes the call record. Never raises into the call it records: the
|
||||
first failure is logged and recording stops."""
|
||||
|
||||
def __init__(self, path: str | Path, *, oplog: OpLog | None = None) -> None:
|
||||
self.path = Path(path)
|
||||
self._oplog = oplog
|
||||
self._lock = threading.Lock()
|
||||
self._conn: sqlite3.Connection | None = None
|
||||
self.active = False
|
||||
try:
|
||||
# Deleted at start (L7): records live as long as the process, so
|
||||
# the layout never needs migrating.
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
Path(f"{self.path}{suffix}").unlink(missing_ok=True)
|
||||
conn = sqlite3.connect(
|
||||
self.path, isolation_level=None, check_same_thread=False
|
||||
)
|
||||
# WAL so the status page can read while the agent writes; no
|
||||
# fsync per write, since a record lost in a power cut costs nothing.
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=NORMAL")
|
||||
conn.execute("BEGIN")
|
||||
for statement in _SCHEMA:
|
||||
conn.execute(statement)
|
||||
conn.execute(
|
||||
"INSERT INTO meta (key, value) VALUES ('layout_version', ?)",
|
||||
(str(LAYOUT_VERSION),),
|
||||
)
|
||||
conn.execute("COMMIT")
|
||||
self._conn = conn
|
||||
self.active = True
|
||||
except (OSError, sqlite3.Error) as exc:
|
||||
self._failed(exc)
|
||||
|
||||
def _failed(self, exc: BaseException) -> None:
|
||||
self.active = False
|
||||
if self._oplog is not None:
|
||||
self._oplog.emit(
|
||||
"calls.record_failed",
|
||||
cause="recording stopped; the calls go on",
|
||||
path=str(self.path),
|
||||
error=f"{type(exc).__name__}: {exc}"[:300],
|
||||
)
|
||||
|
||||
def _write(self, write: Any) -> None:
|
||||
if not self.active or self._conn is None:
|
||||
return
|
||||
with self._lock:
|
||||
try:
|
||||
self._conn.execute("BEGIN")
|
||||
write(self._conn)
|
||||
self._conn.execute("COMMIT")
|
||||
except Exception as exc: # noqa: BLE001 -- never into the call
|
||||
with contextlib.suppress(sqlite3.Error):
|
||||
self._conn.execute("ROLLBACK")
|
||||
self._failed(exc)
|
||||
|
||||
def _blob(self, conn: sqlite3.Connection, value: Any) -> str:
|
||||
digest, data = _encode(value)
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO blobs (hash, data) VALUES (?, ?)", (digest, data)
|
||||
)
|
||||
return digest
|
||||
|
||||
def run_started(
|
||||
self,
|
||||
run_id: str,
|
||||
parent: str | None,
|
||||
work_item: str | None,
|
||||
role: str,
|
||||
kind: str,
|
||||
) -> None:
|
||||
self._write(
|
||||
lambda conn: conn.execute(
|
||||
"INSERT INTO runs (id, parent, work_item, role, kind, started_at) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?)",
|
||||
(run_id, parent, work_item, role, kind, utc_now()),
|
||||
)
|
||||
)
|
||||
|
||||
def run_ended(self, run_id: str, outcome: str, note: str) -> None:
|
||||
self._write(
|
||||
lambda conn: conn.execute(
|
||||
"UPDATE runs SET ended_at = ?, outcome = ?, note = ? WHERE id = ?",
|
||||
(utc_now(), outcome, note, run_id),
|
||||
)
|
||||
)
|
||||
|
||||
def record(
|
||||
self,
|
||||
*,
|
||||
at: str,
|
||||
seconds: float,
|
||||
model: str,
|
||||
answered_by: str | None,
|
||||
error: str | None,
|
||||
usage: Mapping[str, int],
|
||||
exchange: Mapping[str, Any] | None,
|
||||
provenance: Provenance,
|
||||
estimate: int | None,
|
||||
factor: float,
|
||||
trim: Mapping[str, Any] | None,
|
||||
) -> None:
|
||||
scope = _scope.get()
|
||||
|
||||
def write(conn: sqlite3.Connection) -> None:
|
||||
request = response = route = status = None
|
||||
timings: Mapping[str, Any] = {}
|
||||
if exchange is not None:
|
||||
route = exchange.get("route")
|
||||
status = exchange.get("status")
|
||||
body = exchange.get("request")
|
||||
if isinstance(body, Mapping):
|
||||
request = json.dumps(_split(conn, self, body))
|
||||
if "response" in exchange:
|
||||
raw = exchange["response"]
|
||||
response = self._blob(conn, raw)
|
||||
if isinstance(raw, Mapping) and isinstance(
|
||||
raw.get("timings"), Mapping
|
||||
):
|
||||
timings = raw["timings"]
|
||||
sent = [
|
||||
[
|
||||
s.id,
|
||||
s.actor,
|
||||
event_type(s.event),
|
||||
self._blob(conn, to_payload(s.event)),
|
||||
]
|
||||
for s in provenance.sent
|
||||
]
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO calls (
|
||||
at, seconds, run, work_item, role, purpose, model, answered_by,
|
||||
outcome, error, tokens_in, tokens_out, tokens_cached, cache_n,
|
||||
prompt_n, estimate, factor, route, status, request,
|
||||
response, sent, compacted, shortened, dropped, trim
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
|
||||
?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
at,
|
||||
round(seconds, 3),
|
||||
scope.run,
|
||||
scope.work_item,
|
||||
scope.role,
|
||||
scope.purpose,
|
||||
model,
|
||||
answered_by,
|
||||
"failed" if error is not None else "answered",
|
||||
error,
|
||||
usage.get("input"),
|
||||
usage.get("output"),
|
||||
usage.get("cached"),
|
||||
_int(timings.get("cache_n")),
|
||||
_int(timings.get("prompt_n")),
|
||||
estimate,
|
||||
factor,
|
||||
route,
|
||||
status,
|
||||
request,
|
||||
response,
|
||||
json.dumps(sent),
|
||||
json.dumps({str(k): v for k, v in provenance.compacted.items()}),
|
||||
json.dumps({str(k): v for k, v in provenance.shortened.items()}),
|
||||
json.dumps(list(provenance.dropped)),
|
||||
json.dumps(dict(trim)) if trim is not None else None,
|
||||
),
|
||||
)
|
||||
|
||||
self._write(write)
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
self.active = False
|
||||
if self._conn is not None:
|
||||
with contextlib.suppress(sqlite3.Error):
|
||||
self._conn.close()
|
||||
self._conn = None
|
||||
|
||||
|
||||
def _int(value: Any) -> int | None:
|
||||
return (
|
||||
int(value)
|
||||
if isinstance(value, int | float) and not isinstance(value, bool)
|
||||
else None
|
||||
)
|
||||
|
||||
|
||||
def _split(
|
||||
conn: sqlite3.Connection, recorder: CallRecorder, body: Mapping[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""A request body as hashes: each top-level field once, and the messages
|
||||
one by one, since each call repeats the conversation before it."""
|
||||
fields: list[list[Any]] = []
|
||||
for key, value in body.items():
|
||||
if key == "messages" and isinstance(value, list):
|
||||
fields.append([key, [recorder._blob(conn, m) for m in value]])
|
||||
else:
|
||||
fields.append([key, recorder._blob(conn, value)])
|
||||
return {"fields": fields}
|
||||
|
||||
|
||||
# Reading it back: the status page's side, and the tests'.
|
||||
|
||||
|
||||
class UnknownLayout(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class CallReader:
|
||||
"""Opens a call record read-only (``mode=ro``) and never writes to it."""
|
||||
|
||||
def __init__(self, path: str | Path) -> None:
|
||||
self._conn = sqlite3.connect(f"file:{Path(path)}?mode=ro", uri=True)
|
||||
self._conn.row_factory = sqlite3.Row
|
||||
row = self._conn.execute(
|
||||
"SELECT value FROM meta WHERE key = 'layout_version'"
|
||||
).fetchone()
|
||||
self.layout_version = int(row["value"]) if row else 0
|
||||
if self.layout_version != LAYOUT_VERSION:
|
||||
self._conn.close()
|
||||
raise UnknownLayout(
|
||||
f"call record layout {self.layout_version}; this reader knows "
|
||||
f"{LAYOUT_VERSION}"
|
||||
)
|
||||
|
||||
def blob(self, digest: str) -> Any:
|
||||
row = self._conn.execute(
|
||||
"SELECT data FROM blobs WHERE hash = ?", (digest,)
|
||||
).fetchone()
|
||||
if row is None:
|
||||
raise KeyError(digest)
|
||||
return json.loads(zlib.decompress(row["data"]).decode())
|
||||
|
||||
def runs(self) -> list[dict[str, Any]]:
|
||||
return [
|
||||
dict(r)
|
||||
for r in self._conn.execute("SELECT * FROM runs ORDER BY started_at")
|
||||
]
|
||||
|
||||
def calls(self, *, run: str | None = None) -> list[dict[str, Any]]:
|
||||
if run is None:
|
||||
rows = self._conn.execute("SELECT * FROM calls ORDER BY id")
|
||||
else:
|
||||
rows = self._conn.execute(
|
||||
"SELECT * FROM calls WHERE run = ? ORDER BY id", (run,)
|
||||
)
|
||||
out = []
|
||||
for row in rows:
|
||||
call = dict(row)
|
||||
for key in ("sent", "compacted", "shortened", "dropped"):
|
||||
call[key] = json.loads(call[key])
|
||||
call["trim"] = json.loads(call["trim"]) if call["trim"] else None
|
||||
out.append(call)
|
||||
return out
|
||||
|
||||
def request(self, call: Mapping[str, Any]) -> dict[str, Any] | None:
|
||||
"""The request body exactly as sent, rebuilt."""
|
||||
if not call.get("request"):
|
||||
return None
|
||||
body: dict[str, Any] = {}
|
||||
for key, value in json.loads(call["request"])["fields"]:
|
||||
body[key] = (
|
||||
[self.blob(h) for h in value]
|
||||
if isinstance(value, list)
|
||||
else self.blob(value)
|
||||
)
|
||||
return body
|
||||
|
||||
def response(self, call: Mapping[str, Any]) -> Any:
|
||||
return self.blob(call["response"]) if call.get("response") else None
|
||||
|
||||
def close(self) -> None:
|
||||
self._conn.close()
|
||||
@@ -39,6 +39,7 @@ from nisse_core.log.canonical import (
|
||||
ToolCall,
|
||||
UserMessage,
|
||||
)
|
||||
from nisse_core.models import calls
|
||||
from nisse_core.models.context import _MODEL_SIDE, _turn_starts
|
||||
from nisse_core.models.llm import LLMError
|
||||
from nisse_core.models.render import effective_events
|
||||
@@ -207,9 +208,10 @@ class Compactor:
|
||||
ask = StoredEvent(
|
||||
id=0, ts="", event=UserMessage(text=NOTES_REQUEST, source="compaction")
|
||||
)
|
||||
produced = self._client.complete(
|
||||
[*events, ask], tools, model, system=system, require_tool_call=False
|
||||
)
|
||||
with calls.purpose("compaction"):
|
||||
produced = self._client.complete(
|
||||
[*events, ask], tools, model, system=system, require_tool_call=False
|
||||
)
|
||||
return "\n".join(
|
||||
e.text.strip() for e in produced if isinstance(e, AssistantText)
|
||||
).strip()
|
||||
|
||||
@@ -39,6 +39,7 @@ from typing import Any
|
||||
from nisse_core.log.canonical import (
|
||||
AssistantText,
|
||||
CanonicalEvent,
|
||||
Compaction,
|
||||
DelegationStarted,
|
||||
PlanSubmitted,
|
||||
StoredEvent,
|
||||
@@ -46,6 +47,7 @@ from nisse_core.log.canonical import (
|
||||
ToolResult,
|
||||
UserMessage,
|
||||
)
|
||||
from nisse_core.models import calls
|
||||
from nisse_core.models.llm import DEFAULT_MAX_OUTPUT_TOKENS, LLMError
|
||||
from nisse_core.models.render import effective_events
|
||||
from nisse_core.oplog import OpLog
|
||||
@@ -264,8 +266,12 @@ class BudgetedClient:
|
||||
safety_margin: float = 0.1,
|
||||
enforces_tool_choice: Callable[[str], bool] = lambda model: False,
|
||||
oplog: OpLog | None = None,
|
||||
call_recorder: calls.CallRecorder | None = None,
|
||||
) -> None:
|
||||
self._inner = inner
|
||||
# Every call is recorded here for the operator's study (L7), while it
|
||||
# is active; the benchmark swaps it per run.
|
||||
self.call_recorder = call_recorder
|
||||
self._oplog = oplog
|
||||
self._limits_for = limits_for
|
||||
# Whether a model's server holds it to the offered tools when asked
|
||||
@@ -359,10 +365,36 @@ class BudgetedClient:
|
||||
"""``require_tool_call``: None asks for a tool call whenever tools
|
||||
are offered (and the model's server enforces it); False lets the
|
||||
model answer in prose, as compaction's notes do."""
|
||||
recorder = self.call_recorder
|
||||
if recorder is not None and not recorder.active:
|
||||
recorder = None
|
||||
if recorder is not None:
|
||||
# The agent is told its calls are recorded, while they are (L7).
|
||||
system = calls.with_notice(system)
|
||||
budget = self.budget_for(model)
|
||||
factor = self._factors.get(model, 1.0)
|
||||
events = effective_events(list(events))
|
||||
fitted, trim = fit_report(events, system, tools, budget, factor)
|
||||
given = list(events)
|
||||
events = effective_events(given)
|
||||
at = calls.utc_now()
|
||||
started = time.monotonic()
|
||||
try:
|
||||
fitted, trim = fit_report(events, system, tools, budget, factor)
|
||||
except ContextOverflow as exc:
|
||||
if recorder is not None:
|
||||
self._record(
|
||||
recorder,
|
||||
at,
|
||||
started,
|
||||
model,
|
||||
given,
|
||||
events,
|
||||
events,
|
||||
None,
|
||||
factor,
|
||||
exchange=None,
|
||||
error=f"ContextOverflow: {exc}",
|
||||
)
|
||||
raise
|
||||
self.last_trim = trim
|
||||
if trim is not None:
|
||||
# Counted so the benchmark can report how often a model's window
|
||||
@@ -386,6 +418,7 @@ class BudgetedClient:
|
||||
)
|
||||
if require_tool_call is None:
|
||||
require_tool_call = bool(tools) and self._enforces_tool_choice(model)
|
||||
before = getattr(self._inner, "last_exchange", None)
|
||||
started = time.monotonic()
|
||||
try:
|
||||
produced = self._inner.complete(
|
||||
@@ -399,13 +432,139 @@ class BudgetedClient:
|
||||
# ask_human, a delegation with hand_back.
|
||||
require_tool_call=require_tool_call,
|
||||
)
|
||||
except Exception as e:
|
||||
self._log_call(model, started, error=type(e).__name__)
|
||||
except BaseException as e:
|
||||
if isinstance(e, Exception):
|
||||
self._log_call(model, started, error=type(e).__name__)
|
||||
if recorder is not None:
|
||||
self._record(
|
||||
recorder,
|
||||
at,
|
||||
started,
|
||||
model,
|
||||
given,
|
||||
events,
|
||||
fitted,
|
||||
trim,
|
||||
factor,
|
||||
exchange=self._exchange(before),
|
||||
error=f"{type(e).__name__}: {e}"[:1000],
|
||||
tools=tools,
|
||||
system=system,
|
||||
)
|
||||
raise
|
||||
self._log_call(model, started)
|
||||
if recorder is not None:
|
||||
self._record(
|
||||
recorder,
|
||||
at,
|
||||
started,
|
||||
model,
|
||||
given,
|
||||
events,
|
||||
fitted,
|
||||
trim,
|
||||
factor,
|
||||
exchange=self._exchange(before),
|
||||
error=None,
|
||||
tools=tools,
|
||||
system=system,
|
||||
)
|
||||
reported = int(self.last_usage.get("input", 0))
|
||||
if reported > 0:
|
||||
estimated = estimate_tokens(fitted, system, tools)
|
||||
observed = reported / max(estimated, 1)
|
||||
self._factors[model] = max(factor, observed)
|
||||
return produced
|
||||
|
||||
def _exchange(self, before: Any) -> dict[str, Any] | None:
|
||||
"""This call's request and response, if the client set them anew."""
|
||||
exchange = getattr(self._inner, "last_exchange", None)
|
||||
return (
|
||||
exchange if isinstance(exchange, dict) and exchange is not before else None
|
||||
)
|
||||
|
||||
def _record(
|
||||
self,
|
||||
recorder: calls.CallRecorder,
|
||||
at: str,
|
||||
started: float,
|
||||
model: str,
|
||||
given: list[StoredEvent],
|
||||
effective: list[StoredEvent],
|
||||
fitted: list[StoredEvent],
|
||||
trim: Trim | None,
|
||||
factor: float,
|
||||
*,
|
||||
exchange: dict[str, Any] | None,
|
||||
error: str | None,
|
||||
tools: list[dict[str, Any]] | None = None,
|
||||
system: str | None = None,
|
||||
) -> None:
|
||||
answered = None if error is not None else self.last_model
|
||||
recorder.record(
|
||||
at=at,
|
||||
seconds=time.monotonic() - started,
|
||||
model=model,
|
||||
answered_by=answered if answered and answered != model else None,
|
||||
error=error,
|
||||
usage={} if error is not None else dict(self.last_usage),
|
||||
exchange=exchange,
|
||||
provenance=provenance(given, effective, fitted),
|
||||
estimate=(
|
||||
estimate_tokens(fitted, system, tools, factor)
|
||||
if error is None or exchange is not None
|
||||
else None
|
||||
),
|
||||
factor=factor,
|
||||
trim=None
|
||||
if trim is None
|
||||
else {
|
||||
"step": trim.step,
|
||||
"tokens_before": trim.tokens_before,
|
||||
"tokens_after": trim.tokens_after,
|
||||
"limit": trim.limit,
|
||||
"results_cut": trim.results_cut,
|
||||
"events_dropped": trim.events_dropped,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def provenance(
|
||||
given: Sequence[StoredEvent],
|
||||
effective: Sequence[StoredEvent],
|
||||
fitted: Sequence[StoredEvent],
|
||||
) -> calls.Provenance:
|
||||
"""What compaction and fitting did to the events a call was given (L6),
|
||||
for its record: which a compaction replaced, which fitting shortened and
|
||||
by how much, and which it dropped. Events the harness made up for one
|
||||
call (id 0) are in what was sent, and in nothing else."""
|
||||
compactions = [e for e in effective if isinstance(e.event, Compaction)]
|
||||
kept = {e.id for e in effective}
|
||||
compacted: dict[int, int | None] = {}
|
||||
for stored in given:
|
||||
if stored.id == 0 or stored.id in kept or isinstance(stored.event, Compaction):
|
||||
continue
|
||||
compacted[stored.id] = next(
|
||||
(
|
||||
c.id
|
||||
for c in compactions
|
||||
if c.event.replaces[0] <= stored.id <= c.event.replaces[1]
|
||||
),
|
||||
None,
|
||||
)
|
||||
by_id = {e.id: e for e in effective if e.id != 0}
|
||||
sent_ids = {e.id for e in fitted}
|
||||
shortened: dict[int, int] = {}
|
||||
for stored in fitted:
|
||||
original = by_id.get(stored.id)
|
||||
if (
|
||||
original is not None
|
||||
and isinstance(stored.event, ToolResult)
|
||||
and isinstance(original.event, ToolResult)
|
||||
and stored.event.result != original.event.result
|
||||
):
|
||||
shortened[stored.id] = len(original.event.result) - len(stored.event.result)
|
||||
dropped = [i for i in by_id if i not in sent_ids]
|
||||
return calls.Provenance(
|
||||
sent=list(fitted), compacted=compacted, shortened=shortened, dropped=dropped
|
||||
)
|
||||
@@ -188,6 +188,10 @@ class LiteLLMClient:
|
||||
# Set even when the call then fails: time spent waiting out a rate
|
||||
# limit is the proxy's, not the model's (the benchmark subtracts it).
|
||||
self.last_rate_limit_wait_s: float = 0.0
|
||||
# The last call's request body and what came back, for the call
|
||||
# record (L7): a new dict each call, set before it is sent. Never the
|
||||
# headers, which carry the token.
|
||||
self.last_exchange: dict[str, Any] | None = None
|
||||
|
||||
def complete(
|
||||
self,
|
||||
@@ -208,10 +212,12 @@ class LiteLLMClient:
|
||||
# nothing to check against. A caller that knows the model's limits
|
||||
# (the context budget) passes a tighter one.
|
||||
body["max_tokens"] = max_tokens or self._max_tokens
|
||||
self.last_exchange = exchange = {"route": "chat", "request": body}
|
||||
self.last_rate_limit_wait_s = 0.0 # not the last call's, if this one raises
|
||||
response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry(
|
||||
self._http, "/chat/completions", body, self._sleep
|
||||
)
|
||||
_keep_response(exchange, response)
|
||||
if response.status_code != 200:
|
||||
_raise_for_upstream(response)
|
||||
raise LLMError(
|
||||
@@ -321,6 +327,10 @@ class AnthropicPassthroughClient:
|
||||
# Set even when the call then fails: time spent waiting out a rate
|
||||
# limit is the proxy's, not the model's (the benchmark subtracts it).
|
||||
self.last_rate_limit_wait_s: float = 0.0
|
||||
# The last call's request body and what came back, for the call
|
||||
# record (L7): a new dict each call, set before it is sent. Never the
|
||||
# headers, which carry the token.
|
||||
self.last_exchange: dict[str, Any] | None = None
|
||||
|
||||
def complete(
|
||||
self,
|
||||
@@ -341,10 +351,12 @@ class AnthropicPassthroughClient:
|
||||
)
|
||||
if require_tool_call and tools:
|
||||
body["tool_choice"] = {"type": "any"} # Anthropic's name for it
|
||||
self.last_exchange = exchange = {"route": "messages", "request": body}
|
||||
self.last_rate_limit_wait_s = 0.0 # not the last call's, if this one raises
|
||||
response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry(
|
||||
self._http, "/v1/messages", body, self._sleep
|
||||
)
|
||||
_keep_response(exchange, response)
|
||||
if response.status_code != 200:
|
||||
_raise_for_upstream(response)
|
||||
raise LLMError(
|
||||
@@ -444,6 +456,16 @@ def _post_with_rate_limit_retry(
|
||||
# Model families we have a native renderer adapter for. L9: delegations take
|
||||
# the vendor's passthrough route because a normalising middlebox is lossy
|
||||
# exactly where a long-running delegation lives.
|
||||
def _keep_response(exchange: dict[str, Any], response: httpx.Response) -> None:
|
||||
"""What came back, for the call record: the JSON as the server sent it
|
||||
(reasoning included), or its text when it isn't JSON."""
|
||||
exchange["status"] = response.status_code
|
||||
try:
|
||||
exchange["response"] = response.json()
|
||||
except ValueError:
|
||||
exchange["response"] = {"text": response.text[:20000]}
|
||||
|
||||
|
||||
def _response_cost(response: httpx.Response) -> float:
|
||||
try:
|
||||
return float(response.headers.get("x-litellm-response-cost") or 0.0)
|
||||
|
||||
Reference in new issue
Block a user