Spaces:
Running
Running
File size: 4,120 Bytes
851f2ed 695b33f 851f2ed 695b33f 851f2ed 695b33f |
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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
#!/usr/bin/env python3
"""
Conversation management for the agent
"""
from typing import Dict, List, Any
from datetime import datetime
class ConversationManager:
"""
Manages conversation history and context
"""
def __init__(self, max_history: int = 10):
self.max_history = max_history
def add_exchange(self, history: List[Dict[str, str]], user_query: str, agent_response: str) -> List[Dict[str, str]]:
"""
Add a new user-agent exchange to the conversation history
"""
updated_history = history.copy()
# Add user message
updated_history.append({
"role": "user",
"content": user_query,
"timestamp": datetime.now().isoformat()
})
# Add agent response
updated_history.append({
"role": "assistant",
"content": agent_response,
"timestamp": datetime.now().isoformat()
})
# Keep only the last max_history exchanges (pairs)
if len(updated_history) > self.max_history * 2:
updated_history = updated_history[-self.max_history * 2:]
return updated_history
def format_for_lightrag(self, history: List[Dict[str, str]]) -> List[Dict[str, str]]:
"""
Format conversation history for LightRAG API
"""
formatted = []
for exchange in history:
formatted.append({
"role": exchange["role"],
"content": exchange["content"]
})
return formatted
def get_context_summary(self, history: List[Dict[str, str]]) -> str:
"""
Generate a summary of recent conversation context
"""
if not history:
return "No previous conversation context."
recent_exchanges = history[-6:] # Last 3 exchanges
context_parts = []
for i, exchange in enumerate(recent_exchanges):
role = "User" if exchange["role"] == "user" else "Assistant"
context_parts.append(f"{role}: {exchange['content']}")
return "\n".join(context_parts)
class ConversationFormatter:
"""
Format conversation data for different purposes
"""
@staticmethod
def build_conversation_history(history: List[Dict[str, str]], max_turns: int = 10) -> List[Dict[str, str]]:
"""
Build conversation history for LightRAG API
"""
if not history:
return []
# Take last max_turns pairs (user + assistant)
recent_history = history[-max_turns*2:]
formatted = []
for exchange in recent_history:
# Handle both Message objects and dictionary formats
if hasattr(exchange, 'role'):
role = exchange.role
content = exchange.content
else:
role = exchange["role"]
content = exchange["content"]
formatted.append({
"role": role,
"content": content
})
return formatted
@staticmethod
def create_context_summary(history: List[Dict[str, str]]) -> str:
"""
Create a summary of conversation context
"""
if not history:
return "No previous conversation."
recent_exchanges = history[-4:] # Last 2 exchanges
context_parts = []
for exchange in recent_exchanges:
# Handle both Message objects and dictionary formats
if hasattr(exchange, 'role'):
role = "User" if exchange.role == "user" else "Assistant"
content = exchange.content
else:
role = "User" if exchange["role"] == "user" else "Assistant"
content = exchange["content"]
content = content[:100] + "..." if len(content) > 100 else content
context_parts.append(f"{role}: {content}")
return "\n".join(context_parts)
|