Spaces:
Running
Running
File size: 2,618 Bytes
b7f63db | 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 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """
Conversation windowing — keeps context size bounded.
Strategy:
1. Always preserve the last *max_recent* messages in full.
2. Older messages are summarised into a single system block using a
lightweight (FAST tier) LLM call.
3. If no summariser is provided, older messages are simply dropped.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import HumanMessage, SystemMessage
logger = logging.getLogger(__name__)
DEFAULT_MAX_RECENT = 8
def prepare_context(
messages: List[Dict[str, Any]],
*,
max_recent: int = DEFAULT_MAX_RECENT,
summarizer_llm: Optional[BaseChatModel] = None,
) -> List[Dict[str, Any]]:
"""
Prepare a bounded context window from *messages*.
Returns a list of message dicts ready for conversion to LangChain objects.
"""
if len(messages) <= max_recent:
return messages
recent = messages[-max_recent:]
older = messages[:-max_recent]
if summarizer_llm and older:
summary = _summarize(older, summarizer_llm)
if summary:
summary_msg: Dict[str, Any] = {
"role": "system",
"content": f"[Conversation summary so far]\n{summary}",
}
return [summary_msg] + recent
# No summariser → just keep the recent window
return recent
def _summarize(
messages: List[Dict[str, Any]],
llm: BaseChatModel,
) -> Optional[str]:
"""Summarise *messages* into a compact paragraph."""
try:
transcript_lines: list[str] = []
for msg in messages:
role = msg.get("role", "user")
content = (msg.get("content") or "")[:500] # cap per message
transcript_lines.append(f"{role}: {content}")
transcript = "\n".join(transcript_lines[-30:]) # last 30 msgs max
prompt = (
"Summarise the following conversation excerpt in 2-4 concise "
"sentences. Preserve any DeFi operation context (tokens, "
"networks, amounts, actions) if present. Reply in English.\n\n"
f"{transcript}"
)
response = llm.invoke([
SystemMessage(content="You are a concise conversation summariser."),
HumanMessage(content=prompt),
])
text = getattr(response, "content", None)
if isinstance(text, str) and text.strip():
return text.strip()
except Exception:
logger.exception("Conversation summarisation failed; skipping.")
return None
|