103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
"""Shrink conversation snapshots that the browser hydrates.
|
|
|
|
The LangGraph checkpoint stays complete so the next model turn still sees
|
|
every tool argument and file body. ``GET /state`` and ``POST /history``
|
|
only need enough structure for the chat UI: file cards already reopen
|
|
artifacts from disk, so shipping a 200 KB ``write_file`` argument on every
|
|
page load is what makes a content-heavy thread feel frozen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
# File bodies / search dumps. Keep enough that a missing-path inline preview
|
|
# still shows a useful prefix; the rest lives in the artifact store.
|
|
DISPLAY_TOOL_TEXT_MAX_CHARS = 4_096
|
|
# Provider reasoning is shown collapsed; a long chain must not dominate the
|
|
# hydrate payload the way file bodies do.
|
|
DISPLAY_REASONING_MAX_CHARS = 16_384
|
|
|
|
_LARGE_TOOL_ARG_KEYS = frozenset(
|
|
{
|
|
"content",
|
|
"contents",
|
|
"file_text",
|
|
"new_str",
|
|
"old_str",
|
|
"text",
|
|
"body",
|
|
"patch",
|
|
}
|
|
)
|
|
|
|
|
|
def _truncate_text(value: str, limit: int) -> str:
|
|
if len(value) <= limit:
|
|
return value
|
|
omitted = len(value) - limit
|
|
return f"{value[:limit]}\n\n…(display truncated, {omitted} chars omitted)"
|
|
|
|
|
|
def _slim_mapping(payload: Mapping[str, Any]) -> dict[str, Any]:
|
|
out: dict[str, Any] = {}
|
|
for key, value in payload.items():
|
|
if isinstance(value, str) and key in _LARGE_TOOL_ARG_KEYS:
|
|
out[key] = _truncate_text(value, DISPLAY_TOOL_TEXT_MAX_CHARS)
|
|
elif isinstance(value, dict):
|
|
out[key] = _slim_mapping(value)
|
|
else:
|
|
out[key] = value
|
|
return out
|
|
|
|
|
|
def _slim_tool_call(tool_call: Any) -> Any:
|
|
if not isinstance(tool_call, Mapping):
|
|
return tool_call
|
|
slimmed = dict(tool_call)
|
|
args = slimmed.get("args")
|
|
if isinstance(args, dict):
|
|
slimmed["args"] = _slim_mapping(args)
|
|
elif isinstance(args, str) and len(args) > DISPLAY_TOOL_TEXT_MAX_CHARS:
|
|
slimmed["args"] = _truncate_text(args, DISPLAY_TOOL_TEXT_MAX_CHARS)
|
|
function = slimmed.get("function")
|
|
if isinstance(function, dict):
|
|
function = dict(function)
|
|
arguments = function.get("arguments")
|
|
if isinstance(arguments, str) and len(arguments) > DISPLAY_TOOL_TEXT_MAX_CHARS:
|
|
function["arguments"] = _truncate_text(arguments, DISPLAY_TOOL_TEXT_MAX_CHARS)
|
|
slimmed["function"] = function
|
|
return slimmed
|
|
|
|
|
|
def _slim_message(message: Any) -> Any:
|
|
if not isinstance(message, Mapping):
|
|
return message
|
|
slimmed = dict(message)
|
|
tool_calls = slimmed.get("tool_calls")
|
|
if isinstance(tool_calls, list):
|
|
slimmed["tool_calls"] = [_slim_tool_call(call) for call in tool_calls]
|
|
kwargs = slimmed.get("additional_kwargs")
|
|
if isinstance(kwargs, dict):
|
|
kwargs = dict(kwargs)
|
|
raw_calls = kwargs.get("tool_calls")
|
|
if isinstance(raw_calls, list):
|
|
kwargs["tool_calls"] = [_slim_tool_call(call) for call in raw_calls]
|
|
reasoning = kwargs.get("reasoning_content")
|
|
if isinstance(reasoning, str) and len(reasoning) > DISPLAY_REASONING_MAX_CHARS:
|
|
kwargs["reasoning_content"] = _truncate_text(reasoning, DISPLAY_REASONING_MAX_CHARS)
|
|
slimmed["additional_kwargs"] = kwargs
|
|
message_type = str(slimmed.get("type") or slimmed.get("role") or "")
|
|
if message_type in {"tool", "ToolMessage", "function"}:
|
|
content = slimmed.get("content")
|
|
if isinstance(content, str) and len(content) > DISPLAY_TOOL_TEXT_MAX_CHARS:
|
|
slimmed["content"] = _truncate_text(content, DISPLAY_TOOL_TEXT_MAX_CHARS)
|
|
return slimmed
|
|
|
|
|
|
def slim_display_messages(messages: list[Any]) -> list[Any]:
|
|
"""Return a shallow-copied message list safe to send to the browser."""
|
|
|
|
return [_slim_message(message) for message in messages]
|