Spaces:
Sleeping
Sleeping
File size: 2,989 Bytes
500516e |
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 |
# src/utils.py
import os
import json
from src.memory import MemoryManager # Corrected import path
from src.llm_interface import LLMInterface # Import LLMInterface
import logging
import spacy
from sklearn.cluster import KMeans
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def chunk_text(text, chunk_size=1000, overlap=100):
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
def extract_and_summarize(query: str, memory_manager: MemoryManager, llm_interface: LLMInterface, system_prompt: str = "", max_tokens: int = 512, temperature: float = 0.7, top_p: float = 0.95) -> str:
# Retrieve relevant memories from the database
relevant_memories = memory_manager.retrieve_relevant_memories(query, limit=30)
logging.info(f"Retrieved {len(relevant_memories)} relevant memories for query: {query}")
# Combine relevant memories into a single context
context = " ".join([memory['description'] for memory in relevant_memories])
logging.info(f"Built context: {context}")
# Truncate the context if it exceeds the token limit
max_context_length = 30000 # Adjust this based on your LLM's token limit
if len(context) > max_context_length:
context = context[:max_context_length]
logging.info(f"Truncated context to {max_context_length} characters.")
# Use spaCy to generate sentence embeddings
nlp = spacy.load('en_core_web_lg')
sentences = context.split('.')
sentence_embeddings = [nlp(sent).vector for sent in sentences]
# Cluster sentences
num_clusters = min(len(sentences), 10) # Adjust the number of clusters
kmeans = KMeans(n_clusters=num_clusters)
kmeans.fit(sentence_embeddings)
labels = kmeans.labels_
# Select representative sentences from each cluster
representative_sentences = []
for i in range(num_clusters):
cluster_sentences = [sentences[j] for j in range(len(sentences)) if labels[j] == i]
if cluster_sentences:
representative_sentences.append(max(cluster_sentences, key=len)) # Select the longest sentence as representative
# Combine representative sentences to form a summary
summary = " ".join(representative_sentences)
logging.info(f"Generated summary: {summary}")
# Use LLM to refine the summary
try:
refined_summary = llm_interface.send_message(f"Context: {summary}\nQuestion: {query}", system_prompt=system_prompt, max_tokens=max_tokens, temperature=temperature, top_p=top_p)
logging.info(f"Refined summary: {refined_summary}")
except Exception as e:
refined_summary = f"Error refining summary: {e}"
logging.error(f"Error refining summary: {e}")
return refined_summary
|