Spaces:
Sleeping
Sleeping
File size: 3,902 Bytes
6e5b27a f88de80 6e5b27a f88de80 6e5b27a f88de80 6e5b27a f88de80 6e5b27a f88de80 |
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 |
from config import initialize
from utilities.vectorstore.SummaryManager import SummaryManager
def manage_collection(file_name, collection_name):
if not qdrant_manager.get_collection(collection_name):
if not qdrant_manager.create_collection(collection_name):
print("❌ Error: Failed to create collection in Qdrant. Exiting application.", flush=True)
return
success, total_tokens, text = qdrant_manager.insert_document(file_name)
if success:
print(f"✅ Documento inserito correttamente. Token totali: {total_tokens}")
if text:
print(f"✅ Testo completo disponibile (entro il limite token): {text[:100]}...")
else:
print("❌ Errore durante l'inserimento del documento.")
#qdrant_manager.delete_collection(collection_name)
def get_initial_summary(collection_name):
"""Retrieve initial summary from a Qdrant collection."""
# Carica la collection se esiste
if not qdrant_manager.get_collection(collection_name):
print(f"❌ Collection '{collection_name}' non trovata.")
return None
# Inizializza il SummaryManager
summary_manager = SummaryManager(language="en", qdrant_manager=qdrant_manager)
# Genera il riassunto iniziale
return summary_manager.do_initial_summary()
def get_summary(collection_name, type="map_reduce"):
"""Retrieve initial summary from a Qdrant collection."""
# Carica la collection se esiste
if not qdrant_manager.get_collection(collection_name):
print(f"❌ Collection '{collection_name}' non trovata.")
return None
# Inizializza il SummaryManager
summary_manager = SummaryManager(language="en", qdrant_manager=qdrant_manager)
if type == "map_reduce":
return summary_manager.do_summary_map_reduce()
elif type == "stuff":
print("Using stuff method")
return summary_manager.do_summary_stuff()
else:
return None
def chat_with_bot(llm_manager, contextualize=True):
print("🤖 Chatbot! Write 'exit' or 'quit' to close the conversation.\n")
try:
if contextualize:
llm_manager.initialize_conversation()
print(f"🤖 ELI: {llm_manager.messages[-1].content}\n")
except Exception as e:
print(f"⚠️ Could not load initial summary: {e}\n")
while True:
try:
user_input = input("👤 You: ")
if user_input.lower() in ["exit", "quit"]:
print("👋 End of conversation.")
break
response = llm_manager.send_message(user_input, contextualize=contextualize)
print(f"🤖 ELI: {response}\n")
except KeyboardInterrupt:
print("\n👋 Conversation stopped.")
break
except Exception as e:
print(f"⚠️ Error: {e}\n")
def get_chunk(collection_name, chunk_id):
"""Retrieve a specific chunk from a Qdrant collection."""
# Carica la collection se esiste
if not qdrant_manager.get_collection(collection_name):
print(f"❌ Collection '{collection_name}' non trovata.")
return None
return qdrant_manager.get_chunk_by_index(chunk_id)
llm_manager, qdrant_manager = initialize()
# file_name="data/txt/Key statisitcs startups.txt"
collection_name="key_statistics"
llm_manager, qdrant_manager = initialize()
# if qdrant_manager.get_collection(collection_name):
# llm_manager.set_qdrant_manager(qdrant_manager)
#chat_with_bot(llm_manager)
#manage_collection(file_name, collection_name)
#summary=get_initial_summary(collection_name)
#summary=get_summary(collection_name,"map_reduce")
#summary=get_summary(collection_name,type="stuff")
# if summary:
# print(f"✅ Summary:\n{summary}")
# else:
# print("⚠️ Nessun riassunto generato.")
text=get_chunk(collection_name, 1)
print(text) |