Compaction: the model's notes replace the middle of a long run (#158) #163
No files matched your search
@@ -332,9 +332,15 @@ logs the outage; interrupted outbound effects are covered by intents (L8).
|
|||||||
### L6. Context and memory
|
### L6. Context and memory
|
||||||
|
|
||||||
**1. The working context of one model call** is rendered from the log at call time
|
**1. The working context of one model call** is rendered from the log at call time
|
||||||
for one role, fitted to that model's window, and never stored. Fitting never drops
|
for one role, fitted to that model's window, and never stored.
|
||||||
the system prompt, the current plan or the most recent turns; a call that still
|
- **A long run is compacted:** well before the window fills, the model's own
|
||||||
doesn't fit fails (L9).
|
notes replace the middle of its run in one step (a `compaction` event, L7),
|
||||||
|
keeping word for word what started the work and its last few turns. Rare,
|
||||||
|
large steps keep the start of the prompt unchanged between them, so the
|
||||||
|
model server reuses its cache.
|
||||||
|
- **Fitting is the last guard:** it never drops the system prompt, the current
|
||||||
|
plan or the most recent turns, and a call that still doesn't fit fails (L9).
|
||||||
|
Each cut it makes is logged; after compaction one is a sign of trouble.
|
||||||
|
|
||||||
**2. The work-item transcript** (the log, L7) is durable and scoped to one item:
|
**2. The work-item transcript** (the log, L7) is durable and scoped to one item:
|
||||||
nothing crosses items, and agent-level calls (switching, duty runs) see other items
|
nothing crosses items, and agent-level calls (switching, duty runs) see other items
|
||||||
|
|||||||
@@ -713,6 +713,10 @@ of earlier points, replies on threads, the round cap and its reset, `stop` and
|
|||||||
environment only. A stop ends a git command, and a cut-short first clone is
|
environment only. A stop ends a git command, and a cut-short first clone is
|
||||||
removed. `review.clone` / `review.fetch` log the time and size. No size
|
removed. `review.clone` / `review.fetch` log the time and size. No size
|
||||||
limits yet. Tests: `test_clones.py`, against local bare repositories.
|
limits yet. Tests: `test_clones.py`, against local bare repositories.
|
||||||
|
- **Compaction** (#158; reviews of PRs touching about 20 files hit the turn cap,
|
||||||
|
re-reading files fitting had cut): shared by both agents' driver runs, as in
|
||||||
|
L6 above. Each cut fitting still makes is logged (`context.trimmed`).
|
||||||
|
Tests: `test_compact.py` in nisse-core.
|
||||||
|
|
||||||
**M7.6 Deployment.** *Waits for the agent host (Phase 6).* Its image, container,
|
**M7.6 Deployment.** *Waits for the agent host (Phase 6).* Its image, container,
|
||||||
Gitea account and secrets (R5). The image needs `git`, and `scratch/` a
|
Gitea account and secrets (R5). The image needs `git`, and `scratch/` a
|
||||||
@@ -771,6 +775,15 @@ if the implementation differs.
|
|||||||
proposal comment older than the reply.
|
proposal comment older than the reply.
|
||||||
- **L5 transport:** the `Transport` interface is send, ask and receive. Intake
|
- **L5 transport:** the `Transport` interface is send, ask and receive. Intake
|
||||||
backoff is exponential, capped at a few minutes.
|
backoff is exponential, capped at a few minutes.
|
||||||
|
- **L6 compaction (#158):** before each driver call, over `compact_at` (0.6) of
|
||||||
|
the prompt budget, the same model is asked, over the same request plus one
|
||||||
|
message, for notes on everything between what started the work and the last
|
||||||
|
turns (at most `compact_keep_turns`, 3, in `compact_to`, 0.25). The notes,
|
||||||
|
with code's list of the calls they replace, become a `compaction` event
|
||||||
|
that renders in place of its range; a later one folds in the earlier.
|
||||||
|
Failed notes are logged (`context.compaction_failed`) and fitting takes
|
||||||
|
over; `context.compacted` logs each one. Not yet in nisse-dev's helper
|
||||||
|
loop.
|
||||||
- **L6 fitting order:** first shorten old tool results to their first and last
|
- **L6 fitting order:** first shorten old tool results to their first and last
|
||||||
lines, then drop the oldest events with a note saying so; never the system
|
lines, then drop the oldest events with a note saying so; never the system
|
||||||
prompt, the current plan or the most recent turns. View labels:
|
prompt, the current plan or the most recent turns. View labels:
|
||||||
|
|||||||
+35
-2
@@ -283,9 +283,39 @@ what failed last; the start keeps the exit code.
|
|||||||
- **Too high:** the transcript outgrows the context window sooner. The
|
- **Too high:** the transcript outgrows the context window sooner. The
|
||||||
context budget copes (see below), but by cutting older output harder.
|
context budget copes (see below), but by cutting older output harder.
|
||||||
|
|
||||||
### `tuning.context` — fitting each request into the window
|
### `tuning.context` — compaction, and fitting each request into the window
|
||||||
|
|
||||||
Before any request is sent, the agent checks it against the model's context
|
**Compaction comes first.** A driver run's request grows with every tool
|
||||||
|
result. Once it passes `compact_at` of the prompt budget, the model is asked
|
||||||
|
for notes on its run so far: findings with the code they rest on, what each
|
||||||
|
file told it, what is left to do. The notes replace everything between what
|
||||||
|
started the work (the brief, or the conversation) and the last few turns,
|
||||||
|
which stay word for word. The model server has the request cached already,
|
||||||
|
so the notes cost only their own output. Because it happens rarely and drops
|
||||||
|
the request well below the limit, the start of the prompt stays the same for
|
||||||
|
many calls, and the server keeps reusing its cache.
|
||||||
|
|
||||||
|
Logged as `context.compacted` (tokens before and after, seconds). Notes that
|
||||||
|
fail, come back empty or leave the request still too big are logged as
|
||||||
|
`context.compaction_failed`; the run goes on and fitting (below) cuts
|
||||||
|
instead, and compaction isn't tried again for `compact_keep_turns` turns.
|
||||||
|
|
||||||
|
| Knob | Default | |
|
||||||
|
|---|---|---|
|
||||||
|
| `compact_at` | 0.6 | fraction of the prompt budget that sets compaction off; 0 turns it off |
|
||||||
|
| `compact_to` | 0.25 | what the kept recent turns may take, at most |
|
||||||
|
| `compact_keep_turns` | 3 | recent turns kept word for word, at most |
|
||||||
|
|
||||||
|
- **`compact_at` too high:** little room left for the notes call and a big
|
||||||
|
result that lands in one turn, so fitting cuts more often. **Too low:**
|
||||||
|
compactions come often, and each one loses detail.
|
||||||
|
- **`compact_to` / `compact_keep_turns` too low:** the model loses sight of
|
||||||
|
what it just did. **Too high:** the notes don't bring the request far
|
||||||
|
enough down, and compactions come sooner.
|
||||||
|
- **Deciding:** look for `context.trimmed` in the log. After compaction it
|
||||||
|
should be rare; each one means the model lost something it had read.
|
||||||
|
|
||||||
|
**Fitting is the last guard.** Before any request is sent, the agent checks it against the model's context
|
||||||
window and makes it fit. A server doesn't always refuse a prompt that's too
|
window and makes it fit. A server doesn't always refuse a prompt that's too
|
||||||
long. Ollama silently drops its start, which is where the system prompt and
|
long. Ollama silently drops its start, which is where the system prompt and
|
||||||
the plan are, so an unfitted request can fail in a way nobody sees.
|
the plan are, so an unfitted request can fail in a way nobody sees.
|
||||||
@@ -304,6 +334,9 @@ order, stopping at the first that fits:
|
|||||||
3. **If the system prompt and the recent turns still don't fit,** the
|
3. **If the system prompt and the recent turns still don't fit,** the
|
||||||
request is not sent. The milestone stops and asks you.
|
request is not sent. The milestone stops and asks you.
|
||||||
|
|
||||||
|
Each cut is logged as `context.trimmed`, with how many results were
|
||||||
|
shortened and events dropped.
|
||||||
|
|
||||||
Tokens are estimated from characters (pessimistically, at 3 per token) and
|
Tokens are estimated from characters (pessimistically, at 3 per token) and
|
||||||
the estimate is corrected from the prompt sizes the server reports. The
|
the estimate is corrected from the prompt sizes the server reports. The
|
||||||
correction only ever makes the estimate larger.
|
correction only ever makes the estimate larger.
|
||||||
|
|||||||
@@ -2,7 +2,8 @@
|
|||||||
tool, runs out of turns, or is stopped.
|
tool, runs out of turns, or is stopped.
|
||||||
|
|
||||||
Everything the run does is written to the item's transcript as the driver
|
Everything the run does is written to the item's transcript as the driver
|
||||||
(L7), and every model call sees the driver's view of it (L6). Execution tools
|
(L7), and every model call sees the driver's view of it (L6), compacted
|
||||||
|
when it grows long (``compact``). Execution tools
|
||||||
go to the item's executor, on its tool VM (L13); control tools are answered
|
go to the item's executor, on its tool VM (L13); control tools are answered
|
||||||
here, and a waiting one ends the run, so the scheduler can put it to the
|
here, and a waiting one ends the run, so the scheduler can put it to the
|
||||||
operator. Which tools wait is the agent's: each run is told.
|
operator. Which tools wait is the agent's: each run is told.
|
||||||
@@ -14,6 +15,7 @@ from typing import Any, Literal, Protocol
|
|||||||
|
|
||||||
from nisse_core.log.canonical import AssistantText, Fact, ToolCall, ToolResult
|
from nisse_core.log.canonical import AssistantText, Fact, ToolCall, ToolResult
|
||||||
from nisse_core.loop.tools import names
|
from nisse_core.loop.tools import names
|
||||||
|
from nisse_core.models.compact import Compactor
|
||||||
from nisse_core.models.llm import LLMClient, LLMError
|
from nisse_core.models.llm import LLMClient, LLMError
|
||||||
from nisse_core.state.cancel import Cancel, skip_calls
|
from nisse_core.state.cancel import Cancel, skip_calls
|
||||||
from nisse_core.state.store import Transcript
|
from nisse_core.state.store import Transcript
|
||||||
@@ -82,8 +84,10 @@ class Driver:
|
|||||||
max_turns: int = DEFAULT_MAX_TURNS,
|
max_turns: int = DEFAULT_MAX_TURNS,
|
||||||
cancel: Cancel | None = None,
|
cancel: Cancel | None = None,
|
||||||
failure_limit: int = 0, # 0: no limit
|
failure_limit: int = 0, # 0: no limit
|
||||||
|
compactor: Compactor | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self._client = client
|
self._client = client
|
||||||
|
self._compactor = compactor
|
||||||
self._model = model
|
self._model = model
|
||||||
self._max_turns = max_turns
|
self._max_turns = max_turns
|
||||||
self._failure_limit = failure_limit
|
self._failure_limit = failure_limit
|
||||||
@@ -123,6 +127,12 @@ class Driver:
|
|||||||
# L2: stop takes effect at the next tool boundary.
|
# L2: stop takes effect at the next tool boundary.
|
||||||
if self._cancel.requested:
|
if self._cancel.requested:
|
||||||
return RunOutcome("stopped", note=f"before turn {turn + 1}")
|
return RunOutcome("stopped", note=f"before turn {turn + 1}")
|
||||||
|
if self._compactor is not None:
|
||||||
|
# Not a turn: the model's notes replace the middle of a long
|
||||||
|
# run before it outgrows the window (L6).
|
||||||
|
self._compactor.compact(
|
||||||
|
transcript, system=system, tools=tools, model=model
|
||||||
|
)
|
||||||
try:
|
try:
|
||||||
produced = self._client.complete(
|
produced = self._client.complete(
|
||||||
transcript.view(), tools=tools, model=model, system=system
|
transcript.view(), tools=tools, model=model, system=system
|
||||||
|
|||||||
@@ -0,0 +1,262 @@
|
|||||||
|
"""Compaction (architecture L6, L7): the model's own notes replace the middle
|
||||||
|
of a long run, so it stays inside the window without losing what it found.
|
||||||
|
|
||||||
|
Before each model call of a driver run, ``Compactor.compact`` looks at the
|
||||||
|
size of the request. Below ``compact_at`` (a fraction of the prompt budget)
|
||||||
|
it does nothing. Above it, in one step:
|
||||||
|
|
||||||
|
1. It keeps, word for word, everything before the first turn (the brief, or
|
||||||
|
the conversation that started the work) and the last few turns, as many
|
||||||
|
as fit in ``compact_to`` of the budget, at most ``keep_turns``.
|
||||||
|
2. It asks the same model, over the very same request plus one more
|
||||||
|
message, for notes on everything in between. The model server has that
|
||||||
|
request cached from the last call, so the notes cost only their own
|
||||||
|
output. The model offered the run's tools is not made to call one.
|
||||||
|
3. It records a ``Compaction`` event replacing that range, the notes with a
|
||||||
|
list of the calls made in it appended by code, so the model knows what it
|
||||||
|
can look at again. The log keeps everything it replaces.
|
||||||
|
|
||||||
|
Doing this rarely, and well below the limit, keeps the start of the prompt
|
||||||
|
the same for many calls, which is what lets the server reuse its cache.
|
||||||
|
Cutting a little every call, as fitting does, changes it every time.
|
||||||
|
|
||||||
|
A compaction that fails (the call fails, the notes are empty, or they leave
|
||||||
|
the request still over ``compact_at``) is logged and the run goes on; the
|
||||||
|
request is then fitted as before (``context``). It isn't tried again until
|
||||||
|
``keep_turns`` more turns have passed.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import time
|
||||||
|
from collections.abc import Sequence
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from nisse_core.log.canonical import (
|
||||||
|
AssistantText,
|
||||||
|
Compaction,
|
||||||
|
StoredEvent,
|
||||||
|
ToolCall,
|
||||||
|
UserMessage,
|
||||||
|
)
|
||||||
|
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
|
||||||
|
from nisse_core.oplog import OpLog
|
||||||
|
from nisse_core.state.store import Transcript
|
||||||
|
|
||||||
|
# The list of calls appended to the notes: how many, and how long each line.
|
||||||
|
MAX_LISTED_CALLS = 80
|
||||||
|
LISTED_CALL_CHARS = 200
|
||||||
|
|
||||||
|
NOTES_REQUEST = """\
|
||||||
|
[orchestrator] Your context is nearly full. Write notes to replace the \
|
||||||
|
conversation from your task up to your last few turns (the task itself and \
|
||||||
|
those turns stay as they are). Reply with the notes as plain text, not a \
|
||||||
|
tool call. Include:
|
||||||
|
|
||||||
|
- What the task is, and what you have done so far.
|
||||||
|
- Your findings and conclusions so far, each with its file and line and \
|
||||||
|
the exact code or text it rests on, quoted.
|
||||||
|
- Decisions made and anything the operator or a peer asked for, word for \
|
||||||
|
word where the wording matters.
|
||||||
|
- What each file you read or command you ran told you that you still need.
|
||||||
|
- What is left to do, in order.
|
||||||
|
|
||||||
|
Leave out what you no longer need. These notes are all you will have of that \
|
||||||
|
part of the conversation; anything else you would have to look at again."""
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class CompactionTuning:
|
||||||
|
compact_at: float = 0.6 # of the prompt budget; 0: never compact
|
||||||
|
compact_to: float = 0.25 # what the kept recent turns may take
|
||||||
|
keep_turns: int = 3 # recent turns kept word for word, at most
|
||||||
|
|
||||||
|
|
||||||
|
def check_tuning(tuning: CompactionTuning, section: str, errors: list[str]) -> None:
|
||||||
|
"""The agents' config check: notes must leave the request smaller than
|
||||||
|
what set them off, or every call would compact again."""
|
||||||
|
if tuning.compact_at and not tuning.compact_to < tuning.compact_at <= 1:
|
||||||
|
errors.append(
|
||||||
|
f"{section}: compact_to must be less than compact_at, which is at "
|
||||||
|
"most 1 (or 0: never compact)"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class Compactor:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
client: Any,
|
||||||
|
tuning: CompactionTuning | None = None,
|
||||||
|
*,
|
||||||
|
oplog: OpLog | None = None,
|
||||||
|
) -> None:
|
||||||
|
"""``client``: the run's client. Compaction needs its budget
|
||||||
|
(``budget_for``, ``estimate``); with a client that has none, as in
|
||||||
|
tests with a scripted model, it never compacts."""
|
||||||
|
self._client = client
|
||||||
|
self._tuning = tuning or CompactionTuning()
|
||||||
|
self._oplog = oplog
|
||||||
|
# Per run (work item, first event shown): the number of turns when
|
||||||
|
# compaction last failed.
|
||||||
|
self._failed_at: dict[tuple[str | None, int], int] = {}
|
||||||
|
|
||||||
|
def compact(
|
||||||
|
self,
|
||||||
|
transcript: Transcript,
|
||||||
|
*,
|
||||||
|
system: str | None,
|
||||||
|
tools: list[dict[str, Any]] | None,
|
||||||
|
model: str,
|
||||||
|
) -> bool:
|
||||||
|
"""Compact the transcript if its request is over ``compact_at``.
|
||||||
|
True if a compaction was recorded."""
|
||||||
|
tuning = self._tuning
|
||||||
|
budget_for = getattr(self._client, "budget_for", None)
|
||||||
|
estimate = getattr(self._client, "estimate", None)
|
||||||
|
if not tuning.compact_at or budget_for is None or estimate is None:
|
||||||
|
return False
|
||||||
|
limit = budget_for(model).for_prompt
|
||||||
|
events = effective_events(transcript.view())
|
||||||
|
before = estimate(events, system, tools, model)
|
||||||
|
if before <= tuning.compact_at * limit:
|
||||||
|
return False
|
||||||
|
|
||||||
|
starts = _turn_starts(events)
|
||||||
|
run = (transcript.work_item, events[0].id if events else 0)
|
||||||
|
failed_at = self._failed_at.get(run)
|
||||||
|
if failed_at is not None and len(starts) < failed_at + tuning.keep_turns:
|
||||||
|
return False
|
||||||
|
head = _head_end(events)
|
||||||
|
# The recent turns kept: at most keep_turns, as many as fit in
|
||||||
|
# compact_to, and at least one, so the model sees where it is.
|
||||||
|
later = [s for s in starts if s > head]
|
||||||
|
recent_from = len(events)
|
||||||
|
for start in reversed(later[-tuning.keep_turns :]):
|
||||||
|
if recent_from < len(events) and (
|
||||||
|
estimate(events[start:], None, None, model) > tuning.compact_to * limit
|
||||||
|
):
|
||||||
|
break
|
||||||
|
recent_from = start
|
||||||
|
middle = events[head:recent_from]
|
||||||
|
turns = sum(1 for s in later if s < recent_from)
|
||||||
|
if turns < 2:
|
||||||
|
# Too little to be worth notes (or nothing at all): fitting
|
||||||
|
# covers it, and logs what it cuts.
|
||||||
|
return False
|
||||||
|
|
||||||
|
started = time.monotonic()
|
||||||
|
try:
|
||||||
|
notes = self._notes(events, system, tools, model)
|
||||||
|
except LLMError as exc: # a failed compaction never fails the run
|
||||||
|
return self._failed(run, transcript, len(starts), f"the call failed: {exc}")
|
||||||
|
if not notes:
|
||||||
|
return self._failed(run, transcript, len(starts), "the notes were empty")
|
||||||
|
|
||||||
|
first = min(
|
||||||
|
e.event.replaces[0] if isinstance(e.event, Compaction) else e.id
|
||||||
|
for e in middle
|
||||||
|
)
|
||||||
|
last = max(e.id for e in middle if not isinstance(e.event, Compaction))
|
||||||
|
listed = _calls(middle)
|
||||||
|
summary = notes + (
|
||||||
|
"\n\nCalls made in the part these notes replace (make one again "
|
||||||
|
"to see its result):\n" + "\n".join(listed)
|
||||||
|
if listed
|
||||||
|
else ""
|
||||||
|
)
|
||||||
|
compaction = Compaction(summary=summary, model=model, replaces=[first, last])
|
||||||
|
placed = StoredEvent(id=first, ts="", event=compaction)
|
||||||
|
after = estimate(
|
||||||
|
[*events[:head], placed, *events[recent_from:]], system, tools, model
|
||||||
|
)
|
||||||
|
if after > tuning.compact_at * limit:
|
||||||
|
return self._failed(
|
||||||
|
run,
|
||||||
|
transcript,
|
||||||
|
len(starts),
|
||||||
|
f"the request would still be {after} tokens, over "
|
||||||
|
f"{int(tuning.compact_at * limit)}",
|
||||||
|
)
|
||||||
|
transcript.append(compaction)
|
||||||
|
self._failed_at.pop(run, None)
|
||||||
|
if self._oplog is not None:
|
||||||
|
self._oplog.emit(
|
||||||
|
"context.compacted",
|
||||||
|
cause=f"the request was {before} tokens of {limit}",
|
||||||
|
work_item=transcript.work_item,
|
||||||
|
model=model,
|
||||||
|
tokens_before=before,
|
||||||
|
tokens_after=after,
|
||||||
|
replaced=[first, last],
|
||||||
|
turns_replaced=turns,
|
||||||
|
seconds=round(time.monotonic() - started, 1),
|
||||||
|
)
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _notes(
|
||||||
|
self,
|
||||||
|
events: list[StoredEvent],
|
||||||
|
system: str | None,
|
||||||
|
tools: list[dict[str, Any]] | None,
|
||||||
|
model: str,
|
||||||
|
) -> str:
|
||||||
|
# The same system prompt and tools as the run's calls, so the server
|
||||||
|
# reuses the cached request: only the last message is new.
|
||||||
|
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
|
||||||
|
)
|
||||||
|
return "\n".join(
|
||||||
|
e.text.strip() for e in produced if isinstance(e, AssistantText)
|
||||||
|
).strip()
|
||||||
|
|
||||||
|
def _failed(
|
||||||
|
self,
|
||||||
|
run: tuple[str | None, int],
|
||||||
|
transcript: Transcript,
|
||||||
|
turns: int,
|
||||||
|
why: str,
|
||||||
|
) -> bool:
|
||||||
|
self._failed_at[run] = turns
|
||||||
|
if self._oplog is not None:
|
||||||
|
self._oplog.emit(
|
||||||
|
"context.compaction_failed",
|
||||||
|
cause=why[:300],
|
||||||
|
work_item=transcript.work_item,
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _head_end(events: Sequence[StoredEvent]) -> int:
|
||||||
|
"""Where the part before the first turn ends: the brief, or what started
|
||||||
|
the work. An earlier compaction is the start of the middle."""
|
||||||
|
for index, stored in enumerate(events):
|
||||||
|
if isinstance(stored.event, (*_MODEL_SIDE, Compaction)):
|
||||||
|
return index
|
||||||
|
return len(events)
|
||||||
|
|
||||||
|
|
||||||
|
def _calls(events: Sequence[StoredEvent]) -> list[str]:
|
||||||
|
"""The calls made in ``events``, once each, in order, as lines; earlier
|
||||||
|
compactions' lists carried over."""
|
||||||
|
lines: list[str] = []
|
||||||
|
for stored in events:
|
||||||
|
event = stored.event
|
||||||
|
if isinstance(event, Compaction):
|
||||||
|
_, marker, listed = event.summary.partition(
|
||||||
|
"(make one again to see its result):\n"
|
||||||
|
)
|
||||||
|
if marker:
|
||||||
|
lines.extend(listed.splitlines())
|
||||||
|
elif isinstance(event, ToolCall):
|
||||||
|
arguments = json.dumps(event.arguments, ensure_ascii=False)
|
||||||
|
line = f"- {event.tool} {arguments}"
|
||||||
|
if len(line) > LISTED_CALL_CHARS:
|
||||||
|
line = line[: LISTED_CALL_CHARS - 3] + "..."
|
||||||
|
lines.append(line)
|
||||||
|
unique = list(dict.fromkeys(lines))
|
||||||
|
return unique[-MAX_LISTED_CALLS:]
|
||||||
@@ -6,8 +6,15 @@ start of the prompt, which is where the system prompt and the plan are. So a
|
|||||||
request is made to fit the model's declared window *before* it is sent, and
|
request is made to fit the model's declared window *before* it is sent, and
|
||||||
one that cannot be made to fit is never sent.
|
one that cannot be made to fit is never sent.
|
||||||
|
|
||||||
The log itself is never changed. Fitting works on the events about to be
|
The log itself is never changed. Compaction (``compact``) is what keeps a
|
||||||
rendered, in three steps, stopping at the first that fits:
|
long run inside the window: the model's own notes replace the middle of the
|
||||||
|
conversation, now and then, in one large step. Fitting is the last guard
|
||||||
|
under it, and normally sends everything as it is. When it has to cut, the
|
||||||
|
cut is recorded (``Trim``) and logged as ``context.trimmed``: after
|
||||||
|
compaction that should be rare, and each one is worth a look.
|
||||||
|
|
||||||
|
Fitting works on the events about to be rendered, compactions applied, in
|
||||||
|
three steps, stopping at the first that fits:
|
||||||
|
|
||||||
1. Tool results older than the most recent turns are cut to their first and
|
1. Tool results older than the most recent turns are cut to their first and
|
||||||
last lines, with a note saying how much was left out.
|
last lines, with a note saying how much was left out.
|
||||||
@@ -39,6 +46,8 @@ from nisse_core.log.canonical import (
|
|||||||
UserMessage,
|
UserMessage,
|
||||||
)
|
)
|
||||||
from nisse_core.models.llm import DEFAULT_MAX_OUTPUT_TOKENS, LLMError
|
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
|
||||||
|
|
||||||
# English and code run at about four characters a token. Three is deliberately
|
# English and code run at about four characters a token. Three is deliberately
|
||||||
# pessimistic, before any correction from what the server reports.
|
# pessimistic, before any correction from what the server reports.
|
||||||
@@ -129,6 +138,18 @@ class Budget:
|
|||||||
return int(self.window * (1 - self.safety_margin)) - self.max_output
|
return int(self.window * (1 - self.safety_margin)) - self.max_output
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Trim:
|
||||||
|
"""What fitting cut from one request."""
|
||||||
|
|
||||||
|
step: int # 1: older results shortened; 2: the oldest events dropped too
|
||||||
|
tokens_before: int
|
||||||
|
tokens_after: int
|
||||||
|
limit: int
|
||||||
|
results_cut: int
|
||||||
|
events_dropped: int
|
||||||
|
|
||||||
|
|
||||||
def fit(
|
def fit(
|
||||||
events: Sequence[StoredEvent],
|
events: Sequence[StoredEvent],
|
||||||
system: str | None,
|
system: str | None,
|
||||||
@@ -138,14 +159,29 @@ def fit(
|
|||||||
) -> list[StoredEvent]:
|
) -> list[StoredEvent]:
|
||||||
"""The events to render: all of them if they fit, otherwise the result of
|
"""The events to render: all of them if they fit, otherwise the result of
|
||||||
the three steps in the module docstring."""
|
the three steps in the module docstring."""
|
||||||
|
return fit_report(events, system, tools, budget, factor)[0]
|
||||||
|
|
||||||
|
|
||||||
|
def fit_report(
|
||||||
|
events: Sequence[StoredEvent],
|
||||||
|
system: str | None,
|
||||||
|
tools: list[dict[str, Any]] | None,
|
||||||
|
budget: Budget,
|
||||||
|
factor: float = 1.0,
|
||||||
|
) -> tuple[list[StoredEvent], Trim | None]:
|
||||||
|
"""``fit``, and what it cut: None when nothing was."""
|
||||||
limit = budget.for_prompt
|
limit = budget.for_prompt
|
||||||
|
|
||||||
|
def tokens(candidate: Sequence[StoredEvent]) -> int:
|
||||||
|
return estimate_tokens(candidate, system, tools, factor)
|
||||||
|
|
||||||
def fits(candidate: Sequence[StoredEvent]) -> bool:
|
def fits(candidate: Sequence[StoredEvent]) -> bool:
|
||||||
return estimate_tokens(candidate, system, tools, factor) <= limit
|
return tokens(candidate) <= limit
|
||||||
|
|
||||||
kept = list(events)
|
kept = list(events)
|
||||||
if fits(kept):
|
before = tokens(kept)
|
||||||
return kept
|
if before <= limit:
|
||||||
|
return kept, None
|
||||||
|
|
||||||
starts = _turn_starts(kept)
|
starts = _turn_starts(kept)
|
||||||
recent_from = (
|
recent_from = (
|
||||||
@@ -155,6 +191,13 @@ def fit(
|
|||||||
)
|
)
|
||||||
|
|
||||||
# Step 1: cut older tool results down.
|
# Step 1: cut older tool results down.
|
||||||
|
results_cut = sum(
|
||||||
|
1
|
||||||
|
for index, stored in enumerate(kept)
|
||||||
|
if index < recent_from
|
||||||
|
and isinstance(stored.event, ToolResult)
|
||||||
|
and len(stored.event.result) > ELIDED_RESULT_CHARS
|
||||||
|
)
|
||||||
kept = [
|
kept = [
|
||||||
replace(
|
replace(
|
||||||
stored,
|
stored,
|
||||||
@@ -168,7 +211,7 @@ def fit(
|
|||||||
for index, stored in enumerate(kept)
|
for index, stored in enumerate(kept)
|
||||||
]
|
]
|
||||||
if fits(kept):
|
if fits(kept):
|
||||||
return kept
|
return kept, Trim(1, before, tokens(kept), limit, results_cut, 0)
|
||||||
|
|
||||||
# Step 2: drop the oldest events, one turn at a time, never the recent
|
# Step 2: drop the oldest events, one turn at a time, never the recent
|
||||||
# ones. The note keeps the transcript starting with a user message, which
|
# ones. The note keeps the transcript starting with a user message, which
|
||||||
@@ -189,7 +232,10 @@ def fit(
|
|||||||
)
|
)
|
||||||
candidate = [note, *remaining]
|
candidate = [note, *remaining]
|
||||||
if fits(candidate):
|
if fits(candidate):
|
||||||
return candidate
|
dropped = len(kept) - len(remaining)
|
||||||
|
return candidate, Trim(
|
||||||
|
2, before, tokens(candidate), limit, results_cut, dropped
|
||||||
|
)
|
||||||
|
|
||||||
# Step 3: nothing more can go.
|
# Step 3: nothing more can go.
|
||||||
needed = estimate_tokens(kept[recent_from:], system, tools, factor)
|
needed = estimate_tokens(kept[recent_from:], system, tools, factor)
|
||||||
@@ -216,8 +262,10 @@ class BudgetedClient:
|
|||||||
keep_recent_turns: int = 6,
|
keep_recent_turns: int = 6,
|
||||||
safety_margin: float = 0.1,
|
safety_margin: float = 0.1,
|
||||||
enforces_tool_choice: Callable[[str], bool] = lambda model: False,
|
enforces_tool_choice: Callable[[str], bool] = lambda model: False,
|
||||||
|
oplog: OpLog | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self._inner = inner
|
self._inner = inner
|
||||||
|
self._oplog = oplog
|
||||||
self._limits_for = limits_for
|
self._limits_for = limits_for
|
||||||
# Whether a model's server holds it to the offered tools when asked
|
# Whether a model's server holds it to the offered tools when asked
|
||||||
# (LiteLLM's supports_tool_choice). Asking one that doesn't can be
|
# (LiteLLM's supports_tool_choice). Asking one that doesn't can be
|
||||||
@@ -231,6 +279,7 @@ class BudgetedClient:
|
|||||||
# from reported usage. Starts at 1 and only grows.
|
# from reported usage. Starts at 1 and only grows.
|
||||||
self._factors: dict[str, float] = {}
|
self._factors: dict[str, float] = {}
|
||||||
self.trimmed_requests = 0
|
self.trimmed_requests = 0
|
||||||
|
self.last_trim: Trim | None = None # the last request's, if it was cut
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def last_usage(self) -> dict[str, int]:
|
def last_usage(self) -> dict[str, int]:
|
||||||
@@ -260,20 +309,57 @@ class BudgetedClient:
|
|||||||
safety_margin=self._safety_margin,
|
safety_margin=self._safety_margin,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def estimate(
|
||||||
|
self,
|
||||||
|
events: Sequence[StoredEvent],
|
||||||
|
system: str | None,
|
||||||
|
tools: list[dict[str, Any]] | None,
|
||||||
|
model: str,
|
||||||
|
) -> int:
|
||||||
|
"""A request's size in tokens, compactions applied, as fitting
|
||||||
|
would count it."""
|
||||||
|
factor = self._factors.get(model, 1.0)
|
||||||
|
return estimate_tokens(effective_events(list(events)), system, tools, factor)
|
||||||
|
|
||||||
def complete(
|
def complete(
|
||||||
self,
|
self,
|
||||||
events: list[StoredEvent],
|
events: list[StoredEvent],
|
||||||
tools: list[dict[str, Any]] | None,
|
tools: list[dict[str, Any]] | None,
|
||||||
model: str,
|
model: str,
|
||||||
system: str | None = None,
|
system: str | None = None,
|
||||||
|
*,
|
||||||
|
require_tool_call: bool | None = None,
|
||||||
) -> list[CanonicalEvent]:
|
) -> list[CanonicalEvent]:
|
||||||
|
"""``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."""
|
||||||
budget = self.budget_for(model)
|
budget = self.budget_for(model)
|
||||||
factor = self._factors.get(model, 1.0)
|
factor = self._factors.get(model, 1.0)
|
||||||
fitted = fit(events, system, tools, budget, factor)
|
events = effective_events(list(events))
|
||||||
# Counted so the benchmark can report how often a model's window was
|
fitted, trim = fit_report(events, system, tools, budget, factor)
|
||||||
# the constraint: any request that had to be cut down or shortened.
|
self.last_trim = trim
|
||||||
if fitted != list(events):
|
if trim is not None:
|
||||||
|
# Counted so the benchmark can report how often a model's window
|
||||||
|
# was the constraint, and logged: each cut lost the model
|
||||||
|
# something it had read.
|
||||||
self.trimmed_requests += 1
|
self.trimmed_requests += 1
|
||||||
|
if self._oplog is not None:
|
||||||
|
self._oplog.emit(
|
||||||
|
"context.trimmed",
|
||||||
|
cause=(
|
||||||
|
"older tool results shortened"
|
||||||
|
if trim.step == 1
|
||||||
|
else "the oldest events dropped"
|
||||||
|
),
|
||||||
|
model=model,
|
||||||
|
tokens_before=trim.tokens_before,
|
||||||
|
tokens_after=trim.tokens_after,
|
||||||
|
limit=trim.limit,
|
||||||
|
results_cut=trim.results_cut,
|
||||||
|
events_dropped=trim.events_dropped,
|
||||||
|
)
|
||||||
|
if require_tool_call is None:
|
||||||
|
require_tool_call = bool(tools) and self._enforces_tool_choice(model)
|
||||||
produced = self._inner.complete(
|
produced = self._inner.complete(
|
||||||
fitted,
|
fitted,
|
||||||
tools,
|
tools,
|
||||||
@@ -283,7 +369,7 @@ class BudgetedClient:
|
|||||||
# Every caller that offers tools needs one called: the executor
|
# Every caller that offers tools needs one called: the executor
|
||||||
# finishes with finish, the planner with submit_plan or
|
# finishes with finish, the planner with submit_plan or
|
||||||
# ask_human, a delegation with hand_back.
|
# ask_human, a delegation with hand_back.
|
||||||
require_tool_call=bool(tools) and self._enforces_tool_choice(model),
|
require_tool_call=require_tool_call,
|
||||||
)
|
)
|
||||||
reported = int(self.last_usage.get("input", 0))
|
reported = int(self.last_usage.get("input", 0))
|
||||||
if reported > 0:
|
if reported > 0:
|
||||||
|
|||||||
@@ -35,10 +35,11 @@ class Interruptible:
|
|||||||
tools: list[dict[str, Any]] | None,
|
tools: list[dict[str, Any]] | None,
|
||||||
model: str,
|
model: str,
|
||||||
system: str | None = None,
|
system: str | None = None,
|
||||||
|
**options: Any,
|
||||||
) -> list[CanonicalEvent]:
|
) -> list[CanonicalEvent]:
|
||||||
self._calling = True
|
self._calling = True
|
||||||
try:
|
try:
|
||||||
return self._client.complete(events, tools, model, system)
|
return self._client.complete(events, tools, model, system, **options)
|
||||||
finally:
|
finally:
|
||||||
self._calling = False
|
self._calling = False
|
||||||
|
|
||||||
|
|||||||
@@ -4,8 +4,7 @@ call time, never stored provider-shaped.
|
|||||||
First adapter: the OpenAI-compatible chat format LiteLLM's normalized route
|
First adapter: the OpenAI-compatible chat format LiteLLM's normalized route
|
||||||
speaks — the executor role's path (L9). The Anthropic passthrough adapter
|
speaks — the executor role's path (L9). The Anthropic passthrough adapter
|
||||||
arrives with delegations (L9). Token budgeting and prompt-cache breakpoints
|
arrives with delegations (L9). Token budgeting and prompt-cache breakpoints
|
||||||
are render-time concerns and are never stored (L7); neither exists yet in
|
are render-time concerns and are never stored (L7).
|
||||||
this skeleton.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import json
|
import json
|
||||||
@@ -27,20 +26,54 @@ from nisse_core.log.canonical import (
|
|||||||
|
|
||||||
|
|
||||||
def effective_events(stored: list[StoredEvent]) -> list[StoredEvent]:
|
def effective_events(stored: list[StoredEvent]) -> list[StoredEvent]:
|
||||||
"""Apply compactions: events whose ids fall in a compaction's replaced
|
"""Apply compactions: the events in a compaction's replaced range give way
|
||||||
range are dropped; the compaction event itself stays as their summary."""
|
to the compaction, which takes the place of the first of them, so the
|
||||||
replaced: list[tuple[int, int]] = [
|
summary sits where the conversation it summarises was. A compaction
|
||||||
(e.event.replaces[0], e.event.replaces[1])
|
whose range a later one covers is folded into it and not shown. One
|
||||||
for e in stored
|
whose range is no longer in the list (it was applied already) stays
|
||||||
if isinstance(e.event, Compaction)
|
where it is, so applying this twice changes nothing."""
|
||||||
|
compactions = [e for e in stored if isinstance(e.event, Compaction)]
|
||||||
|
active = [
|
||||||
|
c
|
||||||
|
for c in compactions
|
||||||
|
if not any(
|
||||||
|
later.id > c.id
|
||||||
|
and later.event.replaces[0] <= c.event.replaces[0]
|
||||||
|
and later.event.replaces[1] >= c.event.replaces[1]
|
||||||
|
for later in compactions
|
||||||
|
)
|
||||||
]
|
]
|
||||||
|
|
||||||
def is_replaced(event_id: int) -> bool:
|
active_ids = {c.id for c in active}
|
||||||
return any(first <= event_id <= last for first, last in replaced)
|
|
||||||
|
|
||||||
return [
|
def covering(event_id: int) -> StoredEvent | None:
|
||||||
e for e in stored if isinstance(e.event, Compaction) or not is_replaced(e.id)
|
for c in active:
|
||||||
]
|
first, last = c.event.replaces
|
||||||
|
if first <= event_id <= last:
|
||||||
|
return c
|
||||||
|
return None
|
||||||
|
|
||||||
|
present = {e.id for e in stored if not isinstance(e.event, Compaction)}
|
||||||
|
placed = {
|
||||||
|
c.id
|
||||||
|
for c in active
|
||||||
|
if any(c.event.replaces[0] <= i <= c.event.replaces[1] for i in present)
|
||||||
|
}
|
||||||
|
out: list[StoredEvent] = []
|
||||||
|
shown: set[int] = set()
|
||||||
|
for e in stored:
|
||||||
|
if isinstance(e.event, Compaction):
|
||||||
|
if e.id in active_ids and e.id not in placed and e.id not in shown:
|
||||||
|
out.append(e)
|
||||||
|
shown.add(e.id)
|
||||||
|
continue
|
||||||
|
c = covering(e.id)
|
||||||
|
if c is None:
|
||||||
|
out.append(e)
|
||||||
|
elif c.id not in shown:
|
||||||
|
out.append(c)
|
||||||
|
shown.add(c.id)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
class OpenAIChatRenderer:
|
class OpenAIChatRenderer:
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ from typing import Any, Self
|
|||||||
|
|
||||||
from nisse_core.log.canonical import (
|
from nisse_core.log.canonical import (
|
||||||
CanonicalEvent,
|
CanonicalEvent,
|
||||||
|
Compaction,
|
||||||
StoredEvent,
|
StoredEvent,
|
||||||
UserMessage,
|
UserMessage,
|
||||||
event_type,
|
event_type,
|
||||||
@@ -573,9 +574,15 @@ class Transcript:
|
|||||||
def view(self, after: int = 0) -> list[StoredEvent]:
|
def view(self, after: int = 0) -> list[StoredEvent]:
|
||||||
"""The log as this actor is shown it: what a model's context is
|
"""The log as this actor is shown it: what a model's context is
|
||||||
built from (L6). See ``views``."""
|
built from (L6). See ``views``."""
|
||||||
return project(
|
start = max(after, self._start)
|
||||||
self.read(max(after, self._start)), self.actor, self._store.SHOWN
|
# A compaction of events before the start belongs to an earlier
|
||||||
)
|
# run's view, not this one's.
|
||||||
|
events = [
|
||||||
|
e
|
||||||
|
for e in self.read(start)
|
||||||
|
if not (isinstance(e.event, Compaction) and e.event.replaces[0] <= start)
|
||||||
|
]
|
||||||
|
return project(events, self.actor, self._store.SHOWN)
|
||||||
|
|
||||||
def since(self, event_id: int) -> "Transcript":
|
def since(self, event_id: int) -> "Transcript":
|
||||||
"""The same transcript, its view starting after ``event_id``: a run
|
"""The same transcript, its view starting after ``event_id``: a run
|
||||||
|
|||||||
@@ -0,0 +1,310 @@
|
|||||||
|
"""Compaction (L6, L7): a long driver run's middle gives way to the model's
|
||||||
|
own notes, in one step now and then, so the run stays inside the window and
|
||||||
|
the start of its prompt stays the same between compactions."""
|
||||||
|
|
||||||
|
import io
|
||||||
|
import json
|
||||||
|
from itertools import pairwise
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from nisse_core.log.canonical import (
|
||||||
|
AssistantText,
|
||||||
|
CanonicalEvent,
|
||||||
|
Compaction,
|
||||||
|
Fact,
|
||||||
|
StoredEvent,
|
||||||
|
ToolCall,
|
||||||
|
ToolResult,
|
||||||
|
UserMessage,
|
||||||
|
)
|
||||||
|
from nisse_core.loop.driver import Driver, ToolOutcome
|
||||||
|
from nisse_core.loop.tools import tool
|
||||||
|
from nisse_core.models.compact import CompactionTuning, Compactor
|
||||||
|
from nisse_core.models.context import BudgetedClient
|
||||||
|
from nisse_core.models.interruptible import Interruptible, Stopped
|
||||||
|
from nisse_core.oplog import OpLog
|
||||||
|
from nisse_core.state.store import StateStore
|
||||||
|
|
||||||
|
READ = tool("read_file", "Read a file.", path="Its path.")
|
||||||
|
POST = tool("post_review", "Post the review. Ends your turn.", body="The review.")
|
||||||
|
TOOLS = [READ, POST]
|
||||||
|
BRIEF = "BRIEF: review PR #1"
|
||||||
|
FILE_CHARS = 6000 # about 2000 tokens a read
|
||||||
|
|
||||||
|
|
||||||
|
class Model:
|
||||||
|
"""Reads one file a turn, then posts; writes notes when asked for them."""
|
||||||
|
|
||||||
|
def __init__(self, reads: int, notes: Any = "NOTES: all fine so far") -> None:
|
||||||
|
self.reads = reads
|
||||||
|
self.notes = notes
|
||||||
|
self.requests: list[dict[str, Any]] = []
|
||||||
|
self.turn = 0
|
||||||
|
self.on_notes: Any = None
|
||||||
|
|
||||||
|
def complete(
|
||||||
|
self,
|
||||||
|
events: list[StoredEvent],
|
||||||
|
tools: Any,
|
||||||
|
model: str,
|
||||||
|
system: str | None = None,
|
||||||
|
*,
|
||||||
|
max_tokens: int | None = None,
|
||||||
|
require_tool_call: bool = False,
|
||||||
|
) -> list[CanonicalEvent]:
|
||||||
|
asked = _asks_for_notes(events)
|
||||||
|
self.requests.append(
|
||||||
|
{
|
||||||
|
"events": list(events),
|
||||||
|
"tools": tools,
|
||||||
|
"system": system,
|
||||||
|
"require_tool_call": require_tool_call,
|
||||||
|
"notes": asked,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
if asked:
|
||||||
|
if self.on_notes is not None:
|
||||||
|
self.on_notes()
|
||||||
|
notes = self.notes(events) if callable(self.notes) else self.notes
|
||||||
|
return [AssistantText(text=notes, model=model)]
|
||||||
|
self.turn += 1
|
||||||
|
if self.turn > self.reads:
|
||||||
|
return [ToolCall(call_id="c-post", tool="post_review", arguments={})]
|
||||||
|
return [
|
||||||
|
AssistantText(text=f"reading f{self.turn}", model=model),
|
||||||
|
ToolCall(
|
||||||
|
call_id=f"c-{self.turn}",
|
||||||
|
tool="read_file",
|
||||||
|
arguments={"path": f"f{self.turn}.py"},
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
def turns_sent(self) -> list[list[StoredEvent]]:
|
||||||
|
return [r["events"] for r in self.requests if not r["notes"]]
|
||||||
|
|
||||||
|
|
||||||
|
def _asks_for_notes(events: list[StoredEvent]) -> bool:
|
||||||
|
last = events[-1].event if events else None
|
||||||
|
return isinstance(last, UserMessage) and last.source == "compaction"
|
||||||
|
|
||||||
|
|
||||||
|
class Files:
|
||||||
|
def dispatch(self, name: str, arguments: dict[str, Any]) -> ToolOutcome:
|
||||||
|
path = arguments["path"]
|
||||||
|
return ToolOutcome(f"{path}\n" + "x" * FILE_CHARS, True)
|
||||||
|
|
||||||
|
|
||||||
|
class World:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
tmp_path: Path,
|
||||||
|
model: Model,
|
||||||
|
*,
|
||||||
|
tuning: CompactionTuning | None = None,
|
||||||
|
interruptible: bool = False,
|
||||||
|
) -> None:
|
||||||
|
self.stream = io.StringIO()
|
||||||
|
oplog = OpLog("t", self.stream)
|
||||||
|
store = StateStore(tmp_path / "s.db", oplog=OpLog("t", io.StringIO()))
|
||||||
|
self.log = store.transcript("x", "driver")
|
||||||
|
self.log.as_actor("orchestrator").append(Fact(text=BRIEF))
|
||||||
|
self.model = model
|
||||||
|
# A 20k window less the margin and output ceiling: 17,000 tokens for
|
||||||
|
# the prompt, so compaction starts at about 10,200.
|
||||||
|
budgeted = BudgetedClient(
|
||||||
|
model,
|
||||||
|
lambda m: (20_000, None),
|
||||||
|
max_output_tokens=1000,
|
||||||
|
oplog=oplog,
|
||||||
|
enforces_tool_choice=lambda m: True,
|
||||||
|
)
|
||||||
|
self.client: Any = Interruptible(budgeted) if interruptible else budgeted
|
||||||
|
self.driver = Driver(
|
||||||
|
self.client,
|
||||||
|
"m",
|
||||||
|
max_turns=60,
|
||||||
|
compactor=Compactor(self.client, tuning, oplog=oplog),
|
||||||
|
)
|
||||||
|
|
||||||
|
def run(self) -> Any:
|
||||||
|
return self.driver.run(
|
||||||
|
self.log,
|
||||||
|
system="you review",
|
||||||
|
tools=TOOLS,
|
||||||
|
waiting={"post_review"},
|
||||||
|
workspace=Files(),
|
||||||
|
)
|
||||||
|
|
||||||
|
def oplog(self, event: str) -> list[dict[str, Any]]:
|
||||||
|
records = [json.loads(line) for line in self.stream.getvalue().splitlines()]
|
||||||
|
return [r for r in records if r["event"] == event]
|
||||||
|
|
||||||
|
def compactions(self) -> list[Compaction]:
|
||||||
|
return [e.event for e in self.log.read() if isinstance(e.event, Compaction)]
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_long_run_is_compacted_keeping_its_brief_and_recent_turns(
|
||||||
|
tmp_path: Path,
|
||||||
|
) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=7))
|
||||||
|
assert world.run().kind == "wait"
|
||||||
|
|
||||||
|
[compaction] = world.compactions()
|
||||||
|
[logged] = world.oplog("context.compacted")
|
||||||
|
assert logged["tokens_after"] < logged["tokens_before"]
|
||||||
|
assert logged["work_item"] == "x"
|
||||||
|
assert world.oplog("context.trimmed") == [] # fitting never had to cut
|
||||||
|
|
||||||
|
last = world.model.turns_sent()[-1]
|
||||||
|
# The brief first, word for word; then the notes; then the recent turns.
|
||||||
|
assert last[0].event.text == BRIEF
|
||||||
|
assert isinstance(last[1].event, Compaction)
|
||||||
|
assert last[1].event.summary.startswith("NOTES: all fine so far")
|
||||||
|
assert any(
|
||||||
|
isinstance(e.event, ToolResult) and e.event.result.startswith("f7.py")
|
||||||
|
for e in last
|
||||||
|
)
|
||||||
|
assert not any(
|
||||||
|
isinstance(e.event, ToolResult) and e.event.result.startswith("f1.py")
|
||||||
|
for e in last
|
||||||
|
)
|
||||||
|
# The turn just before the compaction is still there, word for word.
|
||||||
|
sent = world.model.turns_sent()
|
||||||
|
after = next(
|
||||||
|
i for i, r in enumerate(sent) if any(isinstance(e.event, Compaction) for e in r)
|
||||||
|
)
|
||||||
|
latest = [e for e in sent[after - 1] if isinstance(e.event, ToolResult)][-1]
|
||||||
|
assert latest in sent[after]
|
||||||
|
# The calls it replaced are listed by code, so the model can make them again.
|
||||||
|
assert '- read_file {"path": "f1.py"}' in compaction.summary
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_notes_are_asked_for_over_the_runs_own_request(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=8))
|
||||||
|
world.run()
|
||||||
|
requests = world.model.requests
|
||||||
|
asked = next(i for i, r in enumerate(requests) if r["notes"])
|
||||||
|
before, notes = requests[asked - 1], requests[asked]
|
||||||
|
# Same system prompt and tools, and the last request plus the model's
|
||||||
|
# reply and its result before it: only the request for notes is new,
|
||||||
|
# so the server reuses its cache.
|
||||||
|
assert notes["system"] == before["system"]
|
||||||
|
assert notes["tools"] == before["tools"]
|
||||||
|
assert before["require_tool_call"] is True
|
||||||
|
assert notes["require_tool_call"] is False
|
||||||
|
ids = [e.id for e in notes["events"][:-1]]
|
||||||
|
assert ids[: len(before["events"])] == [e.id for e in before["events"]]
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_prompt_start_stays_the_same_between_compactions(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=20))
|
||||||
|
world.run()
|
||||||
|
assert len(world.compactions()) >= 2
|
||||||
|
sent = world.model.turns_sent()
|
||||||
|
unchanged = 0
|
||||||
|
for earlier, later in pairwise(sent):
|
||||||
|
compacted = any(isinstance(e.event, Compaction) for e in later) and not any(
|
||||||
|
isinstance(e.event, Compaction) and e.event == c.event
|
||||||
|
for e in later
|
||||||
|
for c in earlier
|
||||||
|
if isinstance(c.event, Compaction)
|
||||||
|
)
|
||||||
|
if compacted:
|
||||||
|
continue
|
||||||
|
# Between compactions, each request extends the one before it.
|
||||||
|
assert [e.id for e in later[: len(earlier)]] == [e.id for e in earlier]
|
||||||
|
unchanged += 1
|
||||||
|
assert unchanged > len(sent) // 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_second_compaction_folds_in_the_first(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=20))
|
||||||
|
world.run()
|
||||||
|
first, second, *_ = world.compactions()
|
||||||
|
assert second.replaces[0] == first.replaces[0]
|
||||||
|
assert second.replaces[1] > first.replaces[1]
|
||||||
|
# The first one's list of calls is carried into the second.
|
||||||
|
assert '- read_file {"path": "f1.py"}' in second.summary
|
||||||
|
last = world.model.turns_sent()[-1]
|
||||||
|
assert sum(isinstance(e.event, Compaction) for e in last) == 1
|
||||||
|
assert last[0].event.text == BRIEF
|
||||||
|
|
||||||
|
|
||||||
|
def test_tool_calls_are_never_parted_from_their_results(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=20))
|
||||||
|
world.run()
|
||||||
|
for sent in world.model.turns_sent():
|
||||||
|
calls = {e.event.call_id for e in sent if isinstance(e.event, ToolCall)}
|
||||||
|
answered = {e.event.call_id for e in sent if isinstance(e.event, ToolResult)}
|
||||||
|
assert answered <= calls
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_short_run_is_never_compacted(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=2))
|
||||||
|
world.run()
|
||||||
|
assert world.compactions() == []
|
||||||
|
assert not any(r["notes"] for r in world.model.requests)
|
||||||
|
|
||||||
|
|
||||||
|
def test_empty_notes_fail_without_failing_the_run(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=8, notes=""))
|
||||||
|
assert world.run().kind == "wait"
|
||||||
|
assert world.compactions() == []
|
||||||
|
failed = world.oplog("context.compaction_failed")
|
||||||
|
assert failed and failed[0]["cause"] == "the notes were empty"
|
||||||
|
# Not tried again every turn: only after keep_turns more.
|
||||||
|
asked = [i for i, r in enumerate(world.model.requests) if r["notes"]]
|
||||||
|
assert all(b - a > 3 for a, b in pairwise(asked))
|
||||||
|
|
||||||
|
|
||||||
|
def test_with_no_compaction_fitting_cuts_and_says_so(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=12, notes=""))
|
||||||
|
world.run()
|
||||||
|
[first, *_] = world.oplog("context.trimmed")
|
||||||
|
assert first["results_cut"] > 0
|
||||||
|
assert first["tokens_after"] <= first["limit"] < first["tokens_before"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_notes_that_leave_it_too_long_are_not_kept(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=8, notes="n" * 40_000))
|
||||||
|
world.run()
|
||||||
|
assert world.compactions() == []
|
||||||
|
[failed, *_] = world.oplog("context.compaction_failed")
|
||||||
|
assert "would still be" in failed["cause"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_stop_while_writing_notes_breaks_off_cleanly(tmp_path: Path) -> None:
|
||||||
|
model = Model(reads=8)
|
||||||
|
world = World(tmp_path, model, interruptible=True)
|
||||||
|
model.on_notes = world.client.interrupt
|
||||||
|
with pytest.raises(Stopped):
|
||||||
|
world.run()
|
||||||
|
assert world.compactions() == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_compact_at_zero_turns_it_off(tmp_path: Path) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=10), tuning=CompactionTuning(compact_at=0))
|
||||||
|
world.run()
|
||||||
|
assert world.compactions() == []
|
||||||
|
assert world.oplog("context.trimmed")
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_view_from_a_start_hides_compactions_of_what_came_before(
|
||||||
|
tmp_path: Path,
|
||||||
|
) -> None:
|
||||||
|
world = World(tmp_path, Model(reads=0))
|
||||||
|
log = world.log
|
||||||
|
log.as_actor("orchestrator").append(Fact(text="old work"))
|
||||||
|
start = log.read()[-1].id
|
||||||
|
log.as_actor("orchestrator").append(Fact(text="BRIEF again"))
|
||||||
|
log.append(Compaction(summary="stale", model="m", replaces=[1, start]))
|
||||||
|
log.append(
|
||||||
|
Compaction(summary="this run's", model="m", replaces=[start + 1, start + 1])
|
||||||
|
)
|
||||||
|
shown = log.since(start).view()
|
||||||
|
assert [e.event.summary for e in shown if isinstance(e.event, Compaction)] == [
|
||||||
|
"this run's"
|
||||||
|
]
|
||||||
@@ -9,6 +9,7 @@ import pytest
|
|||||||
from nisse_core.log.canonical import (
|
from nisse_core.log.canonical import (
|
||||||
AssistantText,
|
AssistantText,
|
||||||
CanonicalEvent,
|
CanonicalEvent,
|
||||||
|
Compaction,
|
||||||
StoredEvent,
|
StoredEvent,
|
||||||
ToolCall,
|
ToolCall,
|
||||||
ToolResult,
|
ToolResult,
|
||||||
@@ -189,6 +190,46 @@ def test_reported_usage_only_ever_raises_the_estimate() -> None:
|
|||||||
assert under_then_over._factors["m"] == 2.0 # never loosened
|
assert under_then_over._factors["m"] == 2.0 # never loosened
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_request_that_fits_is_sent_whole_and_nothing_is_logged() -> None:
|
||||||
|
stream = io.StringIO()
|
||||||
|
client = BudgetedClient(
|
||||||
|
RecordingClient(), lambda m: (100_000, None), oplog=OpLog("t", stream)
|
||||||
|
)
|
||||||
|
events = transcript(3, 100)
|
||||||
|
client.complete(events, None, "m")
|
||||||
|
assert client.last_trim is None
|
||||||
|
assert stream.getvalue() == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_cut_is_recorded_and_logged() -> None:
|
||||||
|
stream = io.StringIO()
|
||||||
|
client = BudgetedClient(
|
||||||
|
RecordingClient(),
|
||||||
|
lambda m: (4000, 500),
|
||||||
|
keep_recent_turns=2,
|
||||||
|
oplog=OpLog("t", stream),
|
||||||
|
)
|
||||||
|
client.complete(transcript(20, 3000), None, "m", system="system")
|
||||||
|
assert client.last_trim is not None and client.last_trim.step == 2
|
||||||
|
assert '"event": "context.trimmed"' in stream.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_compaction_counts_for_what_it_replaced() -> None:
|
||||||
|
events = transcript(20, 3000)
|
||||||
|
last = events[-1].id
|
||||||
|
compaction = StoredEvent(
|
||||||
|
id=last + 1,
|
||||||
|
ts="t",
|
||||||
|
event=Compaction(summary="notes", model="m", replaces=[2, last - 4]),
|
||||||
|
)
|
||||||
|
inner = RecordingClient()
|
||||||
|
client = BudgetedClient(inner, lambda m: (8000, 500), keep_recent_turns=2)
|
||||||
|
client.complete([*events, compaction], None, "m")
|
||||||
|
assert client.last_trim is None # fits once the summary stands in
|
||||||
|
sent = inner.calls[0]["events"]
|
||||||
|
assert isinstance(sent[1].event, Compaction)
|
||||||
|
|
||||||
|
|
||||||
def test_last_usage_passes_through() -> None:
|
def test_last_usage_passes_through() -> None:
|
||||||
client = BudgetedClient(RecordingClient(reported_input=7), lambda m: (100, None))
|
client = BudgetedClient(RecordingClient(reported_input=7), lambda m: (100, None))
|
||||||
assert client.last_usage["input"] == 7
|
assert client.last_usage["input"] == 7
|
||||||
|
|||||||
@@ -66,11 +66,25 @@ def test_compaction_replaces_its_range_only() -> None:
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
remaining = effective_events(events)
|
remaining = effective_events(events)
|
||||||
assert [e.id for e in remaining] == [3, 4]
|
# The summary takes the place of what it replaces.
|
||||||
|
assert [e.id for e in remaining] == [4, 3]
|
||||||
|
assert effective_events(remaining) == remaining # applying again: no change
|
||||||
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
||||||
contents = [m["content"] for m in body["messages"]]
|
contents = [m["content"] for m in body["messages"]]
|
||||||
# Both render as user messages, so they share one: roles must alternate.
|
# Both render as user messages, so they share one: roles must alternate.
|
||||||
assert contents == ["kept\n\n[summary of earlier conversation]\nthe old stuff"]
|
assert contents == ["[summary of earlier conversation]\nthe old stuff\n\nkept"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_later_compaction_folds_in_an_earlier_one() -> None:
|
||||||
|
events = stored([UserMessage(text=f"e{i}", source="m") for i in range(1, 7)])
|
||||||
|
events.append(
|
||||||
|
StoredEvent(id=7, ts="t", event=Compaction("first", "q", replaces=[2, 3]))
|
||||||
|
)
|
||||||
|
events.append(
|
||||||
|
StoredEvent(id=8, ts="t", event=Compaction("second", "q", replaces=[2, 5]))
|
||||||
|
)
|
||||||
|
remaining = effective_events(events)
|
||||||
|
assert [e.id for e in remaining] == [1, 8, 6]
|
||||||
|
|
||||||
|
|
||||||
def test_tool_call_without_a_result_gets_a_placeholder() -> None:
|
def test_tool_call_without_a_result_gets_a_placeholder() -> None:
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ from nisse_core.log.canonical import (
|
|||||||
)
|
)
|
||||||
from nisse_core.loop.driver import Driver, RunOutcome
|
from nisse_core.loop.driver import Driver, RunOutcome
|
||||||
from nisse_core.models.catalog import ModelFacts
|
from nisse_core.models.catalog import ModelFacts
|
||||||
|
from nisse_core.models.compact import Compactor
|
||||||
from nisse_core.oplog import OpLog
|
from nisse_core.oplog import OpLog
|
||||||
from nisse_dev.bench.tasks import Task
|
from nisse_dev.bench.tasks import Task
|
||||||
from nisse_dev.config import Tuning
|
from nisse_dev.config import Tuning
|
||||||
@@ -78,10 +79,13 @@ class MeteringClient:
|
|||||||
tools: list[dict[str, Any]] | None,
|
tools: list[dict[str, Any]] | None,
|
||||||
model: str,
|
model: str,
|
||||||
system: str | None = None,
|
system: str | None = None,
|
||||||
|
**options: Any,
|
||||||
) -> list[CanonicalEvent]:
|
) -> list[CanonicalEvent]:
|
||||||
self.calls += 1
|
self.calls += 1
|
||||||
try:
|
try:
|
||||||
produced = self._inner.complete(events, tools, model, system=system)
|
produced = self._inner.complete(
|
||||||
|
events, tools, model, system=system, **options
|
||||||
|
)
|
||||||
finally:
|
finally:
|
||||||
# Counted even when the call fails: a rate limit that never
|
# Counted even when the call fails: a rate limit that never
|
||||||
# cleared still cost the run its wait.
|
# cleared still cost the run its wait.
|
||||||
@@ -95,10 +99,16 @@ class MeteringClient:
|
|||||||
answered = getattr(self._inner, "last_model", None)
|
answered = getattr(self._inner, "last_model", None)
|
||||||
if not self._facts.answered(answered):
|
if not self._facts.answered(answered):
|
||||||
self.fallbacks.append(str(answered))
|
self.fallbacks.append(str(answered))
|
||||||
if not any(isinstance(e, ToolCall) for e in produced):
|
# Compaction's notes are prose by request, not a broken turn.
|
||||||
|
asked_for_prose = options.get("require_tool_call") is False
|
||||||
|
if not asked_for_prose and not any(isinstance(e, ToolCall) for e in produced):
|
||||||
self.prose_turns += 1
|
self.prose_turns += 1
|
||||||
return produced
|
return produced
|
||||||
|
|
||||||
|
def __getattr__(self, name: str) -> Any:
|
||||||
|
# The budget (budget_for, estimate) compaction reads.
|
||||||
|
return getattr(self._inner, name)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class RunRecord:
|
class RunRecord:
|
||||||
@@ -222,6 +232,8 @@ def run_task(
|
|||||||
model,
|
model,
|
||||||
max_turns=tuning.driver.max_turns,
|
max_turns=tuning.driver.max_turns,
|
||||||
failure_limit=tuning.driver.failure_limit,
|
failure_limit=tuning.driver.failure_limit,
|
||||||
|
# As the agent runs it, so a long task measures the same thing.
|
||||||
|
compactor=Compactor(meter, tuning.context.compaction()),
|
||||||
)
|
)
|
||||||
system = SYSTEM.format(
|
system = SYSTEM.format(
|
||||||
repo=f"bench/{task.name}",
|
repo=f"bench/{task.name}",
|
||||||
|
|||||||
@@ -34,6 +34,7 @@ from nisse_core.config import (
|
|||||||
resolve,
|
resolve,
|
||||||
)
|
)
|
||||||
from nisse_core.conversation.matrix import MatrixConfig
|
from nisse_core.conversation.matrix import MatrixConfig
|
||||||
|
from nisse_core.models.compact import CompactionTuning, check_tuning
|
||||||
from nisse_core.models.llm import (
|
from nisse_core.models.llm import (
|
||||||
DEFAULT_MAX_OUTPUT_TOKENS,
|
DEFAULT_MAX_OUTPUT_TOKENS,
|
||||||
DEFAULT_REQUEST_TIMEOUT_S,
|
DEFAULT_REQUEST_TIMEOUT_S,
|
||||||
@@ -134,6 +135,19 @@ class ToolsTuning:
|
|||||||
class ContextTuning:
|
class ContextTuning:
|
||||||
keep_recent_turns: int = 6
|
keep_recent_turns: int = 6
|
||||||
safety_margin: float = 0.1
|
safety_margin: float = 0.1
|
||||||
|
# Compaction (L6): over compact_at of the prompt budget, the model's
|
||||||
|
# notes replace the middle of the run, keeping at most compact_keep_turns
|
||||||
|
# recent turns, in compact_to of it. compact_at 0: never.
|
||||||
|
compact_at: float = field(default=0.6, metadata={"zero": True})
|
||||||
|
compact_to: float = 0.25
|
||||||
|
compact_keep_turns: int = 3
|
||||||
|
|
||||||
|
def compaction(self) -> CompactionTuning:
|
||||||
|
return CompactionTuning(
|
||||||
|
compact_at=self.compact_at,
|
||||||
|
compact_to=self.compact_to,
|
||||||
|
keep_turns=self.compact_keep_turns,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
@@ -182,6 +196,7 @@ def _parse_tuning(raw: Any, errors: list[str]) -> Tuning:
|
|||||||
)
|
)
|
||||||
if tuning.context.safety_margin >= 1:
|
if tuning.context.safety_margin >= 1:
|
||||||
errors.append("tuning.context.safety_margin: must be less than 1")
|
errors.append("tuning.context.safety_margin: must be less than 1")
|
||||||
|
check_tuning(tuning.context.compaction(), "tuning.context", errors)
|
||||||
return tuning
|
return tuning
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ from nisse_core.knowledge.memory import Memory
|
|||||||
from nisse_core.knowledge.vault import Vault
|
from nisse_core.knowledge.vault import Vault
|
||||||
from nisse_core.loop.driver import Driver
|
from nisse_core.loop.driver import Driver
|
||||||
from nisse_core.models.catalog import ModelCatalog, select_client
|
from nisse_core.models.catalog import ModelCatalog, select_client
|
||||||
|
from nisse_core.models.compact import Compactor
|
||||||
from nisse_core.models.context import BudgetedClient
|
from nisse_core.models.context import BudgetedClient
|
||||||
from nisse_core.models.llm import AnthropicPassthroughClient, LiteLLMClient
|
from nisse_core.models.llm import AnthropicPassthroughClient, LiteLLMClient
|
||||||
from nisse_core.oplog import OpLog
|
from nisse_core.oplog import OpLog
|
||||||
@@ -110,7 +111,7 @@ class RunningLoop:
|
|||||||
|
|
||||||
|
|
||||||
def make_llm_clients(
|
def make_llm_clients(
|
||||||
config: AgentConfig, catalog: ModelCatalog
|
config: AgentConfig, catalog: ModelCatalog, oplog: OpLog | None = None
|
||||||
) -> tuple[BudgetedClient, BudgetedClient]:
|
) -> tuple[BudgetedClient, BudgetedClient]:
|
||||||
"""The two LLM routes, as every model call in the agent uses them.
|
"""The two LLM routes, as every model call in the agent uses them.
|
||||||
|
|
||||||
@@ -136,6 +137,7 @@ def make_llm_clients(
|
|||||||
enforces_tool_choice=lambda m: (
|
enforces_tool_choice=lambda m: (
|
||||||
catalog.facts(m).supports_tool_choice is True
|
catalog.facts(m).supports_tool_choice is True
|
||||||
),
|
),
|
||||||
|
oplog=oplog, # a cut that fitting has to make is logged
|
||||||
)
|
)
|
||||||
|
|
||||||
normalized = budgeted(
|
normalized = budgeted(
|
||||||
@@ -276,7 +278,7 @@ def build_loop_agent(
|
|||||||
project_runtime(config, entry, store, gitea, account)
|
project_runtime(config, entry, store, gitea, account)
|
||||||
for entry in config.projects
|
for entry in config.projects
|
||||||
]
|
]
|
||||||
normalized, passthrough = make_llm_clients(config, catalog)
|
normalized, passthrough = make_llm_clients(config, catalog, oplog)
|
||||||
cancel = Cancel()
|
cancel = Cancel()
|
||||||
# The configured driver; each work item runs on the one it pinned (L9).
|
# The configured driver; each work item runs on the one it pinned (L9).
|
||||||
driver = Driver(
|
driver = Driver(
|
||||||
@@ -285,6 +287,9 @@ def build_loop_agent(
|
|||||||
max_turns=config.tuning.driver.max_turns,
|
max_turns=config.tuning.driver.max_turns,
|
||||||
cancel=cancel,
|
cancel=cancel,
|
||||||
failure_limit=config.tuning.driver.failure_limit,
|
failure_limit=config.tuning.driver.failure_limit,
|
||||||
|
compactor=Compactor(
|
||||||
|
normalized, config.tuning.context.compaction(), oplog=oplog
|
||||||
|
),
|
||||||
)
|
)
|
||||||
tuning = config.tuning
|
tuning = config.tuning
|
||||||
helpers = Helpers(
|
helpers = Helpers(
|
||||||
|
|||||||
@@ -354,6 +354,18 @@ def test_context_tuning_is_read_and_checked() -> None:
|
|||||||
build(tuning={"context": {"safety_margin": 1}})
|
build(tuning={"context": {"safety_margin": 1}})
|
||||||
|
|
||||||
|
|
||||||
|
def test_compaction_tuning_is_read_and_checked() -> None:
|
||||||
|
compaction = build().tuning.context.compaction()
|
||||||
|
assert (compaction.compact_at, compaction.compact_to) == (0.6, 0.25)
|
||||||
|
assert compaction.keep_turns == 3
|
||||||
|
off = build(tuning={"context": {"compact_at": 0}}).tuning.context
|
||||||
|
assert off.compaction().compact_at == 0
|
||||||
|
with pytest.raises(ConfigError, match="compact_to must be less than compact_at"):
|
||||||
|
build(tuning={"context": {"compact_at": 0.3, "compact_to": 0.5}})
|
||||||
|
with pytest.raises(ConfigError, match="compact_to must be less than compact_at"):
|
||||||
|
build(tuning={"context": {"compact_at": 1.5}})
|
||||||
|
|
||||||
|
|
||||||
def test_the_build_host_is_required_and_defaults_to_root_on_22() -> None:
|
def test_the_build_host_is_required_and_defaults_to_root_on_22() -> None:
|
||||||
host = only(build()).build_host
|
host = only(build()).build_host
|
||||||
assert (host.address, host.user) == ("build.test", "root")
|
assert (host.address, host.user) == ("build.test", "root")
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ from nisse_core.knowledge.vault import Vault
|
|||||||
from nisse_core.loop.driver import Driver
|
from nisse_core.loop.driver import Driver
|
||||||
from nisse_core.loop.expert import ExpertDesk
|
from nisse_core.loop.expert import ExpertDesk
|
||||||
from nisse_core.models.catalog import ModelCatalog
|
from nisse_core.models.catalog import ModelCatalog
|
||||||
|
from nisse_core.models.compact import Compactor
|
||||||
from nisse_core.models.context import BudgetedClient
|
from nisse_core.models.context import BudgetedClient
|
||||||
from nisse_core.models.interruptible import Interruptible
|
from nisse_core.models.interruptible import Interruptible
|
||||||
from nisse_core.models.llm import LiteLLMClient
|
from nisse_core.models.llm import LiteLLMClient
|
||||||
@@ -65,7 +66,7 @@ def check_access(
|
|||||||
return account, repos
|
return account, repos
|
||||||
|
|
||||||
|
|
||||||
def llm_client(config: ReviewConfig) -> BudgetedClient:
|
def llm_client(config: ReviewConfig, oplog: OpLog | None = None) -> BudgetedClient:
|
||||||
"""The one route to the models, every request fitted to its window
|
"""The one route to the models, every request fitted to its window
|
||||||
(L6), against what LiteLLM says the configured models are (L9)."""
|
(L6), against what LiteLLM says the configured models are (L9)."""
|
||||||
catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token)
|
catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token)
|
||||||
@@ -92,6 +93,7 @@ def llm_client(config: ReviewConfig) -> BudgetedClient:
|
|||||||
keep_recent_turns=tuning.keep_recent_turns,
|
keep_recent_turns=tuning.keep_recent_turns,
|
||||||
safety_margin=tuning.safety_margin,
|
safety_margin=tuning.safety_margin,
|
||||||
enforces_tool_choice=lambda m: catalog.facts(m).supports_tool_choice is True,
|
enforces_tool_choice=lambda m: catalog.facts(m).supports_tool_choice is True,
|
||||||
|
oplog=oplog, # a cut that fitting has to make is logged
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -106,7 +108,7 @@ def build(
|
|||||||
gitea = gitea or GiteaClient(config.gitea_url, config.gitea_token)
|
gitea = gitea or GiteaClient(config.gitea_url, config.gitea_token)
|
||||||
account, repos = check_access(gitea, config.repositories, oplog)
|
account, repos = check_access(gitea, config.repositories, oplog)
|
||||||
# Every model call, the reviewer's and ask_expert's, can be broken off.
|
# Every model call, the reviewer's and ask_expert's, can be broken off.
|
||||||
model = Interruptible(client if client is not None else llm_client(config)) # type: ignore[arg-type]
|
model = Interruptible(client if client is not None else llm_client(config, oplog)) # type: ignore[arg-type]
|
||||||
config.state_db.parent.mkdir(parents=True, exist_ok=True)
|
config.state_db.parent.mkdir(parents=True, exist_ok=True)
|
||||||
store = StateStore(config.state_db, oplog=oplog)
|
store = StateStore(config.state_db, oplog=oplog)
|
||||||
vault = (
|
vault = (
|
||||||
@@ -129,7 +131,12 @@ def build(
|
|||||||
reviewer = Reviewer(
|
reviewer = Reviewer(
|
||||||
store,
|
store,
|
||||||
gitea,
|
gitea,
|
||||||
Driver(model, config.roles.resolve("driver"), cancel=cancel), # type: ignore[arg-type]
|
Driver(
|
||||||
|
model, # type: ignore[arg-type]
|
||||||
|
config.roles.resolve("driver"),
|
||||||
|
cancel=cancel,
|
||||||
|
compactor=Compactor(model, tuning.context.compaction(), oplog=oplog),
|
||||||
|
),
|
||||||
account=account,
|
account=account,
|
||||||
gitea_url=config.gitea_url,
|
gitea_url=config.gitea_url,
|
||||||
oplog=oplog,
|
oplog=oplog,
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ from nisse_core.config import (
|
|||||||
read_file,
|
read_file,
|
||||||
resolve,
|
resolve,
|
||||||
)
|
)
|
||||||
|
from nisse_core.models.compact import CompactionTuning, check_tuning
|
||||||
from nisse_core.models.llm import (
|
from nisse_core.models.llm import (
|
||||||
DEFAULT_MAX_OUTPUT_TOKENS,
|
DEFAULT_MAX_OUTPUT_TOKENS,
|
||||||
DEFAULT_REQUEST_TIMEOUT_S,
|
DEFAULT_REQUEST_TIMEOUT_S,
|
||||||
@@ -72,6 +73,19 @@ class ExpertTuning:
|
|||||||
class ContextTuning:
|
class ContextTuning:
|
||||||
keep_recent_turns: int = 6
|
keep_recent_turns: int = 6
|
||||||
safety_margin: float = 0.1
|
safety_margin: float = 0.1
|
||||||
|
# Compaction (L6): over compact_at of the prompt budget, the model's
|
||||||
|
# notes replace the middle of the run, keeping at most compact_keep_turns
|
||||||
|
# recent turns, in compact_to of it. compact_at 0: never.
|
||||||
|
compact_at: float = field(default=0.6, metadata={"zero": True})
|
||||||
|
compact_to: float = 0.25
|
||||||
|
compact_keep_turns: int = 3
|
||||||
|
|
||||||
|
def compaction(self) -> CompactionTuning:
|
||||||
|
return CompactionTuning(
|
||||||
|
compact_at=self.compact_at,
|
||||||
|
compact_to=self.compact_to,
|
||||||
|
keep_turns=self.compact_keep_turns,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
@@ -118,6 +132,7 @@ class ReviewConfig:
|
|||||||
tuning = parse_tuning(tree.get("tuning"), Tuning, errors)
|
tuning = parse_tuning(tree.get("tuning"), Tuning, errors)
|
||||||
if tuning.context.safety_margin >= 1:
|
if tuning.context.safety_margin >= 1:
|
||||||
errors.append("tuning.context.safety_margin: must be less than 1")
|
errors.append("tuning.context.safety_margin: must be less than 1")
|
||||||
|
check_tuning(tuning.context.compaction(), "tuning.context", errors)
|
||||||
vault = parse_vault(tree.get("vault"), errors)
|
vault = parse_vault(tree.get("vault"), errors)
|
||||||
flat = flatten({k: v for k, v in tree.items() if k != "vault"})
|
flat = flatten({k: v for k, v in tree.items() if k != "vault"})
|
||||||
flat.pop("tuning", None)
|
flat.pop("tuning", None)
|
||||||
|
|||||||
@@ -494,6 +494,30 @@ def test_the_ci_waits_may_be_zero(tmp_path: Path) -> None:
|
|||||||
assert (review.ci_wait_s, review.ci_grace_s) == (0, 0)
|
assert (review.ci_wait_s, review.ci_grace_s) == (0, 0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_compaction_is_tuned_in_the_config(tmp_path: Path) -> None:
|
||||||
|
from nisse_core.config import ConfigError
|
||||||
|
from nisse_review.config import ReviewConfig
|
||||||
|
|
||||||
|
tree: dict[str, Any] = {
|
||||||
|
"state_db": str(tmp_path / "s.db"),
|
||||||
|
"llm": {"base_url": "http://llm", "token": "k"},
|
||||||
|
"roles": {"reviewer": "local", "expert": "frontier"},
|
||||||
|
"gitea": {"url": "https://g", "token": "t", "operators": ["don"]},
|
||||||
|
}
|
||||||
|
assert (
|
||||||
|
ReviewConfig.from_tree(tree, {}).tuning.context.compaction().compact_at == 0.6
|
||||||
|
)
|
||||||
|
tuned = {
|
||||||
|
**tree,
|
||||||
|
"tuning": {"context": {"compact_at": 0.7, "compact_keep_turns": 2}},
|
||||||
|
}
|
||||||
|
compaction = ReviewConfig.from_tree(tuned, {}).tuning.context.compaction()
|
||||||
|
assert (compaction.compact_at, compaction.keep_turns) == (0.7, 2)
|
||||||
|
wrong = {**tree, "tuning": {"context": {"compact_at": 0.2, "compact_to": 0.3}}}
|
||||||
|
with pytest.raises(ConfigError, match="compact_to must be less than compact_at"):
|
||||||
|
ReviewConfig.from_tree(wrong, {})
|
||||||
|
|
||||||
|
|
||||||
def test_diff_lines_are_the_heads_lines_a_diff_shows() -> None:
|
def test_diff_lines_are_the_heads_lines_a_diff_shows() -> None:
|
||||||
diff = (
|
diff = (
|
||||||
"--- a/x.py\n+++ b/x.py\n@@ -1,3 +1,4 @@\n a\n-b\n+c\n+d\n e\n"
|
"--- a/x.py\n+++ b/x.py\n@@ -1,3 +1,4 @@\n a\n-b\n+c\n+d\n e\n"
|
||||||
|
|||||||
@@ -163,6 +163,8 @@ def test_build_wires_a_stop_to_every_model_call(tmp_path: Path) -> None:
|
|||||||
try:
|
try:
|
||||||
driver = built.scheduler._reviewer._driver
|
driver = built.scheduler._reviewer._driver
|
||||||
assert driver._client is built.model
|
assert driver._client is built.model
|
||||||
|
# Compaction's notes are a model call a stop breaks off too.
|
||||||
|
assert driver._compactor._client is built.model
|
||||||
assert built.scheduler._reviewer._expert._client is built.model
|
assert built.scheduler._reviewer._expert._client is built.model
|
||||||
built.stop() # no call in flight: nothing raised
|
built.stop() # no call in flight: nothing raised
|
||||||
assert driver._cancel.requested and built.scheduler._stop.is_set()
|
assert driver._cancel.requested and built.scheduler._stop.is_set()
|
||||||
|
|||||||
Reference in new issue
Block a user