|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("devii.agentic.compaction")
|
|
|
|
SUMMARY_INPUT_CAP = 600_000
|
|
SUMMARY_PROMPT = (
|
|
"Summarize the following assistant conversation segment as a concise factual log of "
|
|
"actions taken, tools called, entities created or changed, conclusions reached, and "
|
|
"outstanding work. Keep identifiers, slugs, uids, and decisions verbatim. Maximum 800 words.\n\n"
|
|
"---\n\n"
|
|
)
|
|
|
|
|
|
def context_size(messages: list[dict[str, Any]]) -> int:
|
|
return len(json.dumps(messages, default=str))
|
|
|
|
|
|
def find_compaction_split(messages: list[dict[str, Any]], keep_tail: int) -> int:
|
|
if len(messages) <= keep_tail:
|
|
return 1
|
|
candidate = len(messages) - keep_tail
|
|
while candidate > 1:
|
|
if messages[candidate].get("role") == "user":
|
|
return candidate
|
|
candidate -= 1
|
|
return 1
|
|
|
|
|
|
async def compact_messages(llm: Any, messages: list[dict[str, Any]], keep_tail: int) -> list[dict[str, Any]]:
|
|
if len(messages) < keep_tail + 3:
|
|
return messages
|
|
split = find_compaction_split(messages, keep_tail)
|
|
if split <= 1:
|
|
return messages
|
|
system_message = messages[0]
|
|
middle = messages[1:split]
|
|
tail = messages[split:]
|
|
if not middle:
|
|
return messages
|
|
|
|
segment = json.dumps(middle, default=str)[:SUMMARY_INPUT_CAP]
|
|
try:
|
|
summary = await llm.summarize(SUMMARY_PROMPT + segment)
|
|
except Exception: # noqa: BLE001 - compaction must never break the loop
|
|
logger.exception("Compaction summary failed; keeping full context")
|
|
return messages
|
|
|
|
logger.info("Compacted %d messages into a summary", len(middle))
|
|
return [
|
|
system_message,
|
|
{"role": "assistant", "content": f"[compacted earlier turns]\n\n{summary}"},
|
|
*tail,
|
|
]
|