File size: 5,434 Bytes
a112aa8 |
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 131 132 133 134 135 136 137 138 139 |
import os
import logging
from concurrent.futures import ThreadPoolExecutor
from pypdf import PdfReader
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
import ollama
import subprocess
import time
from dotenv import load_dotenv
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.DEBUG
)
logger = logging.getLogger(__name__)
load_dotenv()
logger.debug("Environment variables loaded.")
def load_single_document(filepath):
if filepath.endswith('.pdf'):
with open(filepath, 'rb') as file:
pdf_reader = PdfReader(file)
text = " ".join([page.extract_text() for page in pdf_reader.pages])
elif filepath.endswith('.txt'):
with open(filepath, 'r', encoding='utf-8') as file:
text = file.read()
else:
logger.warning("Unsupported file type: %s", filepath)
return {"content": "", "source": filepath}
return {"content": text, "source": filepath}
def load_documents(directory):
logger.debug("Loading documents from directory: %s", directory)
filepaths = [os.path.join(directory, filename) for filename in os.listdir(directory) if filename.endswith('.pdf') or filename.endswith('.txt')]
documents = []
with ThreadPoolExecutor() as executor:
documents = list(executor.map(load_single_document, filepaths))
logger.debug("Loaded %d documents", len(documents))
return documents
documents = []
with ThreadPoolExecutor() as executor:
documents = list(executor.map(load_single_document, filepaths))
logger.debug("Loaded %d documents", len(documents))
return documents
def prepare_documents(documents):
logger.debug("Preparing documents for embedding.")
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.create_documents([doc["content"] for doc in documents], metadatas=[{"source": os.path.basename(doc["source"])} for doc in documents])
embeddings = HuggingFaceEmbeddings()
db = FAISS.from_documents(texts, embeddings)
logger.debug("Documents prepared and indexed.")
return db
def clarify_ollama(question):
max_retries = 3
for attempt in range(max_retries):
try:
response = ollama.chat(model='gemma2:2b', messages=[
{
'role': 'system',
'content': 'Sei un assistente che deve essere sicuro del topic della domanda. Chiedi se la domanda si riferisce agli osservatori "Blockchain", "Payment" oppure "Metaverse"'
},
{
'role': 'user',
'content': f"Domanda: {question}"
}
])
return response['message']['content']
except Exception as e:
logger.error("Errore nella comunicazione con Ollama: %s", e)
if attempt < max_retries - 1:
logger.debug("Tentativo di riavvio di Ollama...")
subprocess.run(["ollama", "serve"], shell=True)
time.sleep(5)
else:
return "Mi dispiace, c'è un problema di comunicazione con il modello. Per favore, verifica che Ollama sia in esecuzione."
def query_ollama(question, context, sources):
max_retries = 3
for attempt in range(max_retries):
try:
response = ollama.chat(model='gemma2:2b', messages=[
{
'role': 'system',
'content': 'Sei un assistente che risponde in italiano alle domande basandosi solo sulle informazioni fornite. Cita le fonti quando possibile. Se non trovi informazioni, rispondi "Su questo al momento non posso risponderti. Puoi chiedere maggiori informazioni all\'ufficio di riferimento." e rammenta il topic delle fonti spiegando perché le informazioni richieste non sono disponibili.'
},
{
'role': 'user',
'content': f"Contesto: {context}\n\nFonti: {sources}\n\nDomanda: {question}"
}
])
return response['message']['content']
except Exception as e:
logger.error("Errore nella comunicazione con Ollama: %s", e)
if attempt < max_retries - 1:
logger.debug("Tentativo di riavvio di Ollama...")
subprocess.run(["ollama", "serve"], shell=True)
time.sleep(5)
else:
return "Mi dispiace, c'è un problema di comunicazione con il modello. Per favore, verifica che Ollama sia in esecuzione."
def get_answer(question, db):
start_time = time.time()
docs = db.similarity_search(question, k=3)
context = " ".join([doc.page_content for doc in docs])
sources = ", ".join(set([doc.metadata['source'] for doc in docs]))
answer = query_ollama(question, context, sources)
end_time = time.time()
logger.debug("Similarity search and response received in %.2f seconds.", end_time - start_time)
return answer, sources
def get_clarification_answer(question):
start_time = time.time()
clarify_answer = clarify_ollama(question)
end_time = time.time()
logger.debug("Clarification response received in %.2f seconds.", end_time - start_time)
return clarify_answer
|