import openai from langchain_community.document_loaders import PyPDFDirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma import chromadb import uuid from docx import Document from datetime import datetime from docx.shared import Pt, RGBColor from docx import Document from docx.shared import Pt, RGBColor from docx.oxml import OxmlElement, ns from datetime import datetime import os import re import uuid import tempfile import shutil import time import gradio as gr # Panel Kontrolny # HFS_vs_GoogleColab = 1 # Access if HFS_vs_GoogleColab == 0: from google.colab import drive drive.mount('/content/drive') class CFG: BASE_PATH = r'/content/drive/MyDrive/Colab Notebooks/MyProjects/Asystent_Analityka/' if HFS_vs_GoogleColab == 0 else "./" nazwa_projektu_HF = "etfy" # zrobić automatyczny przełącznik GC vs HFS - asystent vs chatbot rola = "Jesteś asystentem doradcy finansowego" kolekcja_bd = "etfy" # jeszcze można dodać nazwy i opisy Interface i ChatBota model_llm = "gpt-4o-mini" # gpt-4o-mini, gpt-4o, o1-mini, gpt-4o, claude-3-opus-20240229, speakleash/Bielik-11B-v2.3-Instruct temperature = 0.6 # od 0.1 do 0.6 model_embeddings = "text-embedding-3-small" # "text-embedding-ada-002", text-embedding-3-small, text-embedding-3-large, ipipan/silver-retriever-base-v1.1 (razem z Bielkiem) dimensions_embeddings = 1536 chunk_size = 3200 # / 500 / 1500 chunk_overlap = 500 # / 200 / 100 / 35 / 200 # No_ReRanking SMALL retriever_num_base_results = 5 reranked_num_results = 3 if HFS_vs_GoogleColab == 1: # Hugging Face Secrets openai_api_key = os.getenv("OPENAI_API_KEY") openai.api_key = openai_api_key # na HFS dodać do Secrets w Settings else: # Google Colab Secrets from google.colab import userdata os.environ["OPENAI_API_KEY"] = userdata.get('Elephant-key') # client = OpenAI(api_key = userdata.get('Elephant-key')) # client = Anthropic(api_key = userdata.get('anthropic-key')) # Ścieżki do konkretnych katalogów dla konkretnych spółek DATA_PATH = os.path.join(CFG.BASE_PATH, f"data_{CFG.kolekcja_bd}") CHROMA_PATH = os.path.join(CFG.BASE_PATH, f"chroma_db_{CFG.kolekcja_bd}") TEMP_PATH = os.path.join(CFG.BASE_PATH, f"answers_{CFG.kolekcja_bd}") # Create the DATA directory if it doesn't exist os.makedirs(DATA_PATH, exist_ok=True) # Create the CHROMA directory if it doesn't exist os.makedirs(CHROMA_PATH, exist_ok=True) # Create the TEMP directory if it doesn't exist os.makedirs(TEMP_PATH, exist_ok=True) def initiate_embeding_model(model_embeddings=CFG.model_embeddings, model_dimensions=CFG.dimensions_embeddings): # initiate the embeddings model embeddings_model = OpenAIEmbeddings( model = model_embeddings, dimensions =model_dimensions ) return embeddings_model embeddings_model = initiate_embeding_model(CFG.model_embeddings, CFG.dimensions_embeddings) # !!! # To pewnie należy zmienić, żeby najpierw sprawdzał, czy istnieje taka baza - czyli żeby tylko inicjował ją a nie tworzył # !!! def create_vector_store(embeddings_model, CHROMA_PATH): # Tworzenie pustej bazy ChromaDB - (!) natywnie, nie przez LangChain client = chromadb.PersistentClient(path=CHROMA_PATH) # initiate the vector store vector_store = Chroma( embedding_function = embeddings_model, persist_directory = CHROMA_PATH, ) # Sprawdzenie, czy baza jest pusta (powinna być) print("Dostępne kolekcje:", client.list_collections()) return vector_store, client vector_store, client = create_vector_store(embeddings_model, CHROMA_PATH) def read_data_from_chroma(collection_name): # Set up the vectorstore to be the retriever # 📌 1. Inicjalizacja kolekcji vector_store = Chroma( collection_name=collection_name, embedding_function=embeddings_model, persist_directory=CHROMA_PATH, ) # Określenie liczby chunków zwracanych przez retriever base_num_results = CFG.retriever_num_base_results # Chroma nie obsługuje rerankingu, więc nie działa "fetch_k" i trzeba albo samemu zrobić reranking # przy pomocy LLM (poniżej) albo użyć jakieś inne biblioteki robiącej to automatycznie (MultiQueryRetriever) #rr_num_results = CFG.reranked_num_results retriever = vector_store.as_retriever(search_kwargs={"k": base_num_results}) return retriever # usatw kolekcję do odczytu collection_name = CFG.kolekcja_bd retriever = read_data_from_chroma(collection_name) # initiate the model llm = ChatOpenAI(temperature=CFG.temperature, model=CFG.model_llm) def response(query, historia=None): # NO_ReRanking chunków relevant_chunks = retriever.invoke(query) # add all the chunks to 'knowledge' knowledge = "" sources_markdown = "" zrodla = "" # ✅ Źródła dla pliku Word cytaty = "" nazwa_projektu = CFG.nazwa_projektu_HF if HFS_vs_GoogleColab == 1: # Dynamiczny link do GFS # Pobranie nazwy aktywnego projektu Spaces (jeśli uruchomiony na HF) HF_SPACE_ID = os.getenv("SPACE_ID", "smolinski/test") # Domyślnie "test", jeśli brak zmiennej # Tworzenie dynamicznego BASE_URL dla aktualnego projektu BASE_URL = f"https://huggingface.co/spaces/{HF_SPACE_ID}/resolve/main/data_{CFG.kolekcja_bd}/" else: BASE_URL = DATA_PATH + "/" for relevant_chunk in relevant_chunks: # Tworzenie linku do źródła full_source = relevant_chunk.metadata.get("source", "Nieznane źródło") # Pobiera źródło file_name = os.path.basename(full_source) # Usuwa ścieżkę, zostawia nazwę pliku file_link = f"{BASE_URL}{file_name}" file_link = file_link.replace(' ', '%20') # na wypadek spacji w nazwie – kodujemy je do URL #page_number = relevant_chunk.metadata.get("page_number", "nieznana strona") # Pobranie numeru strony page_number_raw = relevant_chunk.metadata.get("page_number", None) try: page_number = int(page_number_raw) + 1 page_number = str(page_number) except (ValueError, TypeError): page_number = "nieznana strona" # ✅ Linki zapisane w Markdownie dla Gradio sources_markdown += f"\n- {file_name}, str. {page_number}: [otwórz]({file_link})" # ✅ Chunki w osobnej zmiennej cytaty += f"Cytat z {file_name}, strona {page_number}:\n\n{relevant_chunk.page_content}\n\n---\n\n" knowledge += relevant_chunk.page_content + "\n\n---\n\n" # print(cytaty) # dodajemy historię do prompta (jeśli istnieje) historia_text = "" if historia: for i, (q, a) in enumerate(historia[-5:], 1): historia_text += f"\nPoprzednia rozmowa {i}:\nPytanie: {q}\nOdpowiedź: {a}\n" # make the call to the LLM (including prompt) if query is not None: rag_prompt = f""" {CFG.rola}, który szczegółowo i dokładnie odpowiada na pytania w oparciu o przekazaną wiedzę. Dziel się wszystkimi posiadanymi informacjami na dany temat. Na pytanie o ETF-y na GPW wymieniaj wszystkie dostępne, chyba że to pytanie szczegółowe o etf-y long, short, lewarowane itp. Na pytania o aktualne notowania odpowiadaj: Aktualne notowania dostępne są na stronie GPW. Do podkreślenia lub wypunktowania najważniejszych rzeczy używaj pogrubionej czcionki, a do ciekawostek i dodatkowych rzeczy kursywy. Podczas udzielania odpowiedzi korzystaj wyłącznie z poniższych informacji zawartych w sekcji „Wiedza”. Bądź miły i uprzejmy, ale rzeczowy. Przykładaj większą wagę do nowszych informacji. Jeśli pytanie jest zbyt ogólne nie odpowiadaj na nie, lecz poproś o doprecyzowanie. Jeśli nie znasz odpowiedzi, napisz: Niestety nie posiadam informacji na ten temat. NIE WYMYŚLAJ NICZEGO.\n\n {historia_text} Pytanie: {query} Wiedza:\n {knowledge} """ # the response to the Gradio App response = llm(rag_prompt) return response.content if response and response.content else "Brak odpowiedzi.", sources_markdown, cytaty def zaktualizuj_historie(pytanie, odpowiedz, historia): historia.append((pytanie, odpowiedz)) return historia[-5:] def wyczysc_formularz(): return "", "", "", [] # ✅ Funkcja do dodania cytatów do odpowiedzi def dodaj_cytaty(odpowiedz, cytaty): return f"{odpowiedz}\n\n---\n**Cytaty ze śródeł**\n---\n\n{cytaty}" if cytaty else odpowiedz # 1. Inicjalizacja unikalnego folderu sesji def init_user_session(): cleanup_old_sessions(base_path=tempfile.gettempdir(), max_age_days=1) session_id = str(uuid.uuid4()) user_temp_dir = os.path.join(tempfile.gettempdir(), f"asystent_{session_id}") os.makedirs(user_temp_dir, exist_ok=True) return user_temp_dir # 2. Czyszczenie katalogów starszych niż 1 dzień def cleanup_old_sessions(base_path, max_age_days=1): now = time.time() for folder in os.listdir(base_path): if folder.startswith("asystent_"): folder_path = os.path.join(base_path, folder) if os.path.isdir(folder_path): folder_age = now - os.path.getctime(folder_path) if folder_age > max_age_days * 86400: shutil.rmtree(folder_path) print(f"Usunięto stary folder sesji: {folder_path}") # 3. Zapis odpowiedzi do pliku .docx def zapisz_odpowiedz(odpowiedz, pytanie, sources, user_path): if not odpowiedz or odpowiedz.strip() == "" or not pytanie.strip(): print("Błąd: Odpowiedź lub pytanie są puste!") return None date_str = datetime.now().strftime("%Y-%m-%d") file_name = "".join(c if c.isalnum() or c in (" ", "_", "-") else "_" for c in pytanie)[:50] file_path = os.path.join(user_path, f"{file_name}_{date_str}.docx") try: doc = Document() def formatuj_naglowek(paragraph, text, font_size=14, color=(0, 0, 0), bold=True): run = paragraph.add_run(text) run.bold = bold run.font.size = Pt(font_size) run.font.color.rgb = RGBColor(*color) run.font.name = "Calibri" paragraph.paragraph_format.line_spacing = 1.25 paragraph.paragraph_format.space_before = Pt(5) paragraph.paragraph_format.space_after = Pt(0) def formatuj_paragraf(paragraph): for run in paragraph.runs: run.font.name = "Calibri" run.font.size = Pt(12) paragraph.paragraph_format.line_spacing = 1.25 paragraph.paragraph_format.space_before = Pt(0) paragraph.paragraph_format.space_after = Pt(5) def dodaj_markdown_tekst(doc, text): lines = text.splitlines() for line in lines: # Lista wypunktowana if line.strip().startswith(("- ", "* ")): para = doc.add_paragraph(style='List Bullet') content = line.strip()[2:] # Cytat elif line.strip().startswith("> "): para = doc.add_paragraph() para.paragraph_format.left_indent = Pt(20) content = line.strip()[2:] else: para = doc.add_paragraph() content = line # Markdown inline: **bold**, *italic*, ~~strikethrough~~ pattern = r"(\*\*.*?\*\*|\*.*?\*|~~.*?~~|[^*~]+)" parts = re.findall(pattern, content) for part in parts: clean = part.replace("**", "").replace("*", "").replace("~~", "") run = para.add_run(clean) if part.startswith("**") and part.endswith("**"): run.bold = True elif part.startswith("*") and part.endswith("*"): run.italic = True elif part.startswith("~~") and part.endswith("~~"): run.font.strike = True run.font.name = "Calibri" run.font.size = Pt(12) p1 = doc.add_paragraph() formatuj_naglowek(p1, "Pytanie:") p1 = doc.add_paragraph(pytanie) formatuj_paragraf(p1) doc.add_paragraph(" ") p2 = doc.add_paragraph() formatuj_naglowek(p2, "Odpowiedź:") dodaj_markdown_tekst(doc, odpowiedz) doc.add_paragraph(" ") if sources and sources.strip(): p3 = doc.add_paragraph() formatuj_naglowek(p3, "Źródła:") p3 = doc.add_paragraph(re.sub(r":.*", "", sources)) formatuj_paragraf(p3) doc.save(file_path) print(f"Plik zapisany: {file_path}") return file_path if os.path.exists(file_path) else None except Exception as e: print(f"Błąd podczas zapisu pliku: {e}") return None # 4. Lista plików użytkownika def lista_plikow(user_path): pliki = [os.path.join(user_path, f) for f in os.listdir(user_path) if f.endswith(".docx")] pliki.sort(key=os.path.getctime, reverse=True) return pliki if pliki else None # 5. Czyszczenie folderu użytkownika def wyczysc_folder(user_path): if os.path.exists(user_path): shutil.rmtree(user_path) # call this function for every message added to the chatbot def stream_response(query, history): """Obsługuje strumieniowanie i poprawnie czyści pole tekstowe.""" history = history or [] # Inicjalizacja pustej historii, jeśli brak danych # Pobranie pasujących fragmentów wiedzy relevant_chunks = retriever.invoke(query) knowledge = "\n\n---\n\n".join([relevant_chunk.page_content for relevant_chunk in relevant_chunks]) # Tworzenie promptu dla modelu LLM rag_prompt = f""" {CFG.rola}, który szczegółowo i dokładnie odpowiada na pytania w oparciu o przekazaną wiedzę. Dziel się wszystkimi posiadanymi informacjami na dany temat, tak by Twoje odpowiedzi były wyczerpujące. Na pytanie o ETF-y na GPW wymieniaj wszystkie dostępne, chyba że to pytanie szczegółowe o etf-y long, short, lewarowane itp. Na pytania o aktualne notowania odpowiadaj: Aktualne notowania dostępne są na stronie GPW. Do podkreślenia lub wypunktowania najważniejszych rzeczy używaj pogrubionej czcionki, a do ciekawostek i dodatkowych rzeczy kursywy. Podczas udzielania odpowiedzi korzystaj wyłącznie z poniższych informacji zawartych w sekcji „Wiedza”. Bądź miły i uprzejmy, ale rzeczowy. Przykładaj większą wagę do nowszych informacji. Jeśli pytanie jest zbyt ogólne nie odpowiadaj na nie, lecz poproś o doprecyzowanie. Jeśli nie znasz odpowiedzi, napisz: Niestety nie posiadam informacji na ten temat. NIE WYMYŚLAJ NICZEGO.\n\n Pytanie: {query}\n\n Historia rozmowy:\n {history} Wiedza:\n {knowledge} """ print("Prompt:\n", rag_prompt) print("Odpowiedź:") # Strumieniowanie odpowiedzi do Gradio partial_message = "" for response in llm.stream(rag_prompt): partial_message += response.content yield history + [(query, partial_message)], query # **Tymczasowo zwracamy query, by pole nie było puste** # Po zakończeniu strumieniowania dodajemy pełną wiadomość do historii i czyścimy input_text history.append((query, partial_message)) yield history, "" # **Finalnie zwracamy pusty string, by wyczyścić pole tekstowe** with gr.Blocks(css=""" .button_wyczysc-color { background-color: #A9A9A9 !important; color: white !important; } #markdown_odpowiedz { border: 1px solid #ccc; border-radius: 6px; padding: 12px; background-color: #f9f9f9; margin-top: 6px; min-height: 8em; } """) as gui: session_dir = gr.State(value=init_user_session) historia_formularza = gr.State([]) gr.Markdown("# Asystent Finansowy") gr.Markdown("### Odpowiadam na pytania z zakresu ETF-ów notowanych na GPW.") gr.Markdown("###### Pamiętaj: jestem tylko chatbotem i czasami się mylę, a moje odpowiedzi nie mogą być traktowane jako rekomendacje inwestycyjne!") with gr.Tabs(): # ChatBot with gr.TabItem("💬 Chat"): chatbot = gr.Chatbot() input_text_chat = gr.Textbox(placeholder="Napisz tutaj pytanie...", container=False, autoscroll=True, scale=7) input_text_chat.submit(fn=stream_response, inputs=[input_text_chat, chatbot], outputs=[chatbot, input_text_chat]) # Formularz with gr.TabItem("📝 Formularz"): input_text_form = gr.Textbox(label="Zadaj pytanie:", placeholder="Napisz tutaj pytanie...", lines=2, interactive=True) with gr.Row(): with gr.Column(scale=1): submit_button = gr.Button("Wyślij pytanie") with gr.Column(scale=1): clear_answer_button = gr.Button("Wyczyść formularz", elem_classes="button_wyczysc-color") with gr.Column(scale=7): gr.Markdown("") gr.Markdown("### Odpowiedź:") output_answer = gr.Markdown( value="", elem_id="markdown_odpowiedz" ) output_cytaty = gr.State("") with gr.Row(): with gr.Column(scale=1): zacytuj_button = gr.Button("Przytocz źródła") with gr.Column(scale=8): gr.Markdown("") gr.Markdown("### Źródła:") output_sources = gr.Markdown() gr.Markdown("### Pobierz odpowiedzi:") download_files = gr.File(label="Pliki do pobrania", interactive=False, file_types=[".docx"]) # Logika przycisków submit_button.click( response, inputs=[input_text_form, historia_formularza], outputs=[output_answer, output_sources, output_cytaty] ).then( zaktualizuj_historie, inputs=[input_text_form, output_answer, historia_formularza], outputs=historia_formularza ).then( zapisz_odpowiedz, inputs=[output_answer, input_text_form, output_sources, session_dir], outputs=None ).then( lista_plikow, inputs=session_dir, outputs=download_files ) clear_answer_button.click( wyczysc_formularz, inputs=[], outputs=[output_answer, input_text_form, output_sources, historia_formularza] ) zacytuj_button.click( dodaj_cytaty, inputs=[output_answer, output_cytaty], outputs=output_answer ).then( zapisz_odpowiedz, inputs=[output_answer, input_text_form, output_sources, session_dir], outputs=None ).then( lista_plikow, inputs=session_dir, outputs=download_files ) gui.launch()