jpmschweitzerandClaude Opus 4.5
2523db4da7
- SQLAlchemy async database layer (SQLite dev, PostgreSQL prod) - Conversation and Message models with UUID primary keys - Token counting utilities using litellm - Context summarization at 80% token threshold - REST API endpoints for multi-turn conversations - 19 conversation tests, 6 token tests (176 total passing) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
"""
|
|
Context summarization for conversations.
|
|
|
|
Compresses old messages when approaching token limits.
|
|
"""
|
|
from src.domains.conversations.models import Message
|
|
from src.shared.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
SUMMARIZE_PROMPT = """Summarize this conversation history concisely for context preservation.
|
|
|
|
Focus on:
|
|
- Key decisions made and their rationale
|
|
- Important files, functions, or code discussed
|
|
- Current task state and progress
|
|
- Any unresolved questions or blockers
|
|
- Technical details that would be needed to continue the work
|
|
|
|
Keep the summary under 500 words. Be factual and technical, not conversational.
|
|
Preserve specific file paths, function names, and code references.
|
|
|
|
CONVERSATION HISTORY:
|
|
{history}
|
|
|
|
CONCISE SUMMARY:"""
|
|
|
|
|
|
def format_messages_for_summary(messages: list[Message]) -> str:
|
|
"""
|
|
Format messages into a string for summarization.
|
|
|
|
Args:
|
|
messages: List of Message objects to format
|
|
|
|
Returns:
|
|
Formatted conversation string
|
|
"""
|
|
parts = []
|
|
for msg in messages:
|
|
if msg.is_summary:
|
|
parts.append(f"[Previous Summary]: {msg.content}")
|
|
else:
|
|
role = msg.role.upper()
|
|
parts.append(f"{role}: {msg.content}")
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
async def generate_summary(
|
|
messages: list[Message],
|
|
working_dir: str = "."
|
|
) -> str:
|
|
"""
|
|
Generate a summary of conversation messages using the Explore agent.
|
|
|
|
Args:
|
|
messages: Messages to summarize
|
|
working_dir: Working directory for agent context
|
|
|
|
Returns:
|
|
Summary text
|
|
"""
|
|
from src.domains.agents.explore import explore
|
|
|
|
history = format_messages_for_summary(messages)
|
|
prompt = SUMMARIZE_PROMPT.format(history=history)
|
|
|
|
logger.info(f"Generating summary for {len(messages)} messages")
|
|
|
|
try:
|
|
summary = await explore(prompt, working_dir=working_dir)
|
|
return summary.strip()
|
|
except Exception as e:
|
|
logger.error(f"Summary generation failed: {e}")
|
|
# Fallback: create a simple truncated summary
|
|
return _fallback_summary(messages)
|
|
|
|
|
|
def _fallback_summary(messages: list[Message]) -> str:
|
|
"""
|
|
Create a simple fallback summary if agent summarization fails.
|
|
|
|
Args:
|
|
messages: Messages to summarize
|
|
|
|
Returns:
|
|
Basic summary string
|
|
"""
|
|
# Take first and last few messages
|
|
if len(messages) <= 4:
|
|
return format_messages_for_summary(messages)
|
|
|
|
first_two = messages[:2]
|
|
last_two = messages[-2:]
|
|
|
|
parts = [
|
|
"Conversation started with:",
|
|
format_messages_for_summary(first_two),
|
|
f"\n[... {len(messages) - 4} messages omitted ...]\n",
|
|
"Most recent exchange:",
|
|
format_messages_for_summary(last_two),
|
|
]
|
|
return "\n".join(parts)
|