forked from retoor/devplacepy
105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from ..text import normalize_newlines
|
|
|
|
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. "
|
|
"Write plain markdown with real line breaks (never the two-character sequence \\n). "
|
|
"Use headings and bullet lists where helpful. 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
|
|
|
|
|
|
def _segment_plain(messages: list[dict[str, Any]]) -> str:
|
|
parts: list[str] = []
|
|
for message in messages:
|
|
role = str(message.get("role") or "unknown")
|
|
content = message.get("content")
|
|
if isinstance(content, str) and content.strip():
|
|
parts.append(f"{role}:\n{normalize_newlines(content)}")
|
|
continue
|
|
if isinstance(content, list):
|
|
chunks: list[str] = []
|
|
for part in content:
|
|
if isinstance(part, dict) and part.get("type") == "text":
|
|
chunks.append(str(part.get("text") or ""))
|
|
elif isinstance(part, str):
|
|
chunks.append(part)
|
|
text = normalize_newlines("\n".join(c for c in chunks if c))
|
|
if text.strip():
|
|
parts.append(f"{role}:\n{text}")
|
|
continue
|
|
tool_calls = message.get("tool_calls")
|
|
if tool_calls:
|
|
names = []
|
|
for call in tool_calls:
|
|
fn = (call or {}).get("function") or {}
|
|
name = fn.get("name") or "tool"
|
|
names.append(str(name))
|
|
if names:
|
|
parts.append(f"{role}: called {', '.join(names)}")
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
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 = _segment_plain(middle)[:SUMMARY_INPUT_CAP]
|
|
if not segment.strip():
|
|
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
|
|
|
|
summary = normalize_newlines(summary or "").strip()
|
|
if not summary:
|
|
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,
|
|
]
|