|
import re
|
|
from typing import Any, Dict, List
|
|
|
|
|
|
class AdvancedContextManager:
|
|
|
|
def __init__(self, knowledge_store=None, conversation_memory=None):
|
|
self.knowledge_store = knowledge_store
|
|
self.conversation_memory = conversation_memory
|
|
|
|
def adaptive_context_window(self, messages: List[Dict[str, Any]], complexity: str) -> int:
|
|
"""Calculate adaptive context window size based on message complexity."""
|
|
base_window = 10
|
|
complexity_multipliers = {"simple": 1.0, "medium": 2.0, "complex": 3.5, "very_complex": 5.0}
|
|
multiplier = complexity_multipliers.get(complexity, 2.0)
|
|
return int(base_window * multiplier)
|
|
|
|
def _analyze_message_complexity(self, messages: List[Dict[str, Any]]) -> float:
|
|
"""Analyze the complexity of messages and return a score between 0.0 and 1.0."""
|
|
if not messages:
|
|
return 0.0
|
|
total_complexity = 0.0
|
|
for message in messages:
|
|
content = message.get("content", "")
|
|
if not content:
|
|
continue
|
|
word_count = len(content.split())
|
|
sentence_count = len(re.split("[.!?]+", content))
|
|
avg_word_length = sum((len(word) for word in content.split())) / max(word_count, 1)
|
|
length_score = min(1.0, word_count / 100)
|
|
structure_score = min(1.0, sentence_count / 10)
|
|
vocabulary_score = min(1.0, avg_word_length / 8)
|
|
message_complexity = (length_score + structure_score + vocabulary_score) / 3
|
|
total_complexity += message_complexity
|
|
return min(1.0, total_complexity / len(messages))
|
|
|
|
def extract_key_sentences(self, text: str, top_k: int = 5) -> List[str]:
|
|
if not text.strip():
|
|
return []
|
|
sentences = re.split("(?<=[.!?])\\s+", text)
|
|
if not sentences:
|
|
return []
|
|
scored_sentences = []
|
|
for i, sentence in enumerate(sentences):
|
|
length_score = min(1.0, len(sentence) / 50)
|
|
position_score = 1.0 if i == 0 else 0.8 if i < len(sentences) / 2 else 0.6
|
|
score = (length_score + position_score) / 2
|
|
scored_sentences.append((sentence, score))
|
|
scored_sentences.sort(key=lambda x: x[1], reverse=True)
|
|
return [s[0] for s in scored_sentences[:top_k]]
|
|
|
|
def advanced_summarize_messages(self, messages: List[Dict[str, Any]]) -> str:
|
|
all_content = " ".join([msg.get("content", "") for msg in messages])
|
|
key_sentences = self.extract_key_sentences(all_content, top_k=3)
|
|
summary = " ".join(key_sentences)
|
|
return summary if summary else "No content to summarize."
|
|
|
|
def score_message_relevance(self, message: Dict[str, Any], context: str) -> float:
|
|
content = message.get("content", "")
|
|
content_words = set(re.findall("\\b\\w+\\b", content.lower()))
|
|
context_words = set(re.findall("\\b\\w+\\b", context.lower()))
|
|
intersection = content_words & context_words
|
|
union = content_words | context_words
|
|
if not union:
|
|
return 0.0
|
|
return len(intersection) / len(union)
|
|
|
|
def create_enhanced_context(
|
|
self, messages: List[Dict[str, Any]], user_message: str, include_knowledge: bool = True
|
|
) -> tuple:
|
|
"""Create enhanced context with knowledge base and conversation memory integration."""
|
|
working_messages = messages.copy()
|
|
all_results = []
|
|
|
|
# Search knowledge base
|
|
if include_knowledge and self.knowledge_store:
|
|
knowledge_results = self.knowledge_store.search_entries(user_message, top_k=3)
|
|
for entry in knowledge_results:
|
|
score = entry.metadata.get("search_score", 0.5)
|
|
all_results.append(
|
|
{
|
|
"content": entry.content,
|
|
"score": score,
|
|
"source": f"Knowledge Base ({entry.category})",
|
|
"type": "knowledge",
|
|
}
|
|
)
|
|
|
|
# Search conversation memory
|
|
if self.conversation_memory:
|
|
from rp.core.knowledge_context import calculate_text_similarity
|
|
|
|
history_results = self.conversation_memory.search_conversations(user_message, limit=3)
|
|
for conv in history_results:
|
|
conv_messages = self.conversation_memory.get_conversation_messages(
|
|
conv["conversation_id"]
|
|
)
|
|
for msg in conv_messages[-5:]: # Last 5 messages from each conversation
|
|
if msg["role"] == "user" and msg["content"] != user_message:
|
|
relevance = calculate_text_similarity(user_message, msg["content"])
|
|
if relevance > 0.3:
|
|
all_results.append(
|
|
{
|
|
"content": msg["content"],
|
|
"score": relevance,
|
|
"source": f"Previous conversation: {conv['conversation_id'][:8]}",
|
|
"type": "conversation",
|
|
}
|
|
)
|
|
|
|
# Sort and limit results
|
|
all_results.sort(key=lambda x: x["score"], reverse=True)
|
|
top_results = all_results[:5]
|
|
|
|
if top_results:
|
|
knowledge_parts = []
|
|
for idx, result in enumerate(top_results, 1):
|
|
content = result["content"]
|
|
if len(content) > 1500:
|
|
content = content[:1500] + "..."
|
|
score_indicator = f"({result['score']:.2f})" if result["score"] < 1.0 else "(exact)"
|
|
knowledge_parts.append(
|
|
f"Match {idx} {score_indicator} - {result['source']}:\n{content}"
|
|
)
|
|
knowledge_message_content = (
|
|
"[KNOWLEDGE_BASE_CONTEXT]\nRelevant information from knowledge base and conversation history:\n\n"
|
|
+ "\n\n".join(knowledge_parts)
|
|
)
|
|
knowledge_message = {"role": "user", "content": knowledge_message_content}
|
|
working_messages.append(knowledge_message)
|
|
context_info = (
|
|
f"Added {len(top_results)} matches from knowledge and conversation history"
|
|
)
|
|
else:
|
|
context_info = "No relevant knowledge or conversation matches found"
|
|
|
|
return (working_messages, context_info)
|