Spaces:
Paused
Paused
File size: 2,372 Bytes
7d4338a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """API handler for chat compaction."""
from helpers.api import ApiHandler, Input, Output, Request, Response
from agent import AgentContext
from plugins._chat_compaction.helpers.compactor import (
MIN_COMPACTION_TOKENS,
get_compaction_stats,
run_compaction,
)
class CompactChat(ApiHandler):
"""Compact the current chat history into a summarized message."""
async def process(self, input: Input, request: Request) -> Output:
ctxid = input.get("context", "")
action = input.get("action", "compact")
if not ctxid:
return Response("Missing context id", 400)
context = AgentContext.get(ctxid)
if not context:
return Response("Context not found", 404)
if context.is_running():
return Response("Cannot compact while agent is running", 409)
visible_count = len(context.log.logs)
if visible_count <= 1:
return Response("Not enough messages to compact", 400)
# Gate both stats and compact — no point opening the modal for tiny chats
stats = await get_compaction_stats(context)
if stats["token_count"] < MIN_COMPACTION_TOKENS:
return {
"ok": False,
"message": f"Not enough content to compact (minimum {MIN_COMPACTION_TOKENS:,} tokens)",
}
if action == "stats":
return {"ok": True, "stats": stats}
elif action == "compact":
use_chat_model = input.get("use_chat_model", True)
preset_name = input.get("preset_name") or None
context.run_task(
_run_compaction_task, context, use_chat_model, preset_name
)
return {"ok": True, "message": "Compaction started"}
else:
return Response(f"Unknown action: {action}", 400)
async def _run_compaction_task(context, use_chat_model: bool, preset_name: str | None):
"""Wrapper to run compaction and handle errors."""
try:
await run_compaction(context, use_chat_model, preset_name)
except Exception as e:
context.log.log(
type="error",
heading="Compaction Failed",
content=str(e),
)
from helpers.state_monitor_integration import mark_dirty_all
mark_dirty_all(reason="plugins._chat_compaction.compact_chat_error")
|