import gradio as gr import json import os from typing import List from concurrent.futures import ThreadPoolExecutor, as_completed from openai import OpenAI import itertools import ast # API Keys OPENAI_API_KEY = os.getenv("YOUR_OPENAI_API_KEY") openai_client = OpenAI(api_key=OPENAI_API_KEY) openai_tools = [{"type": "web_search_preview", "search_context_size": "high", "user_location": {"type": "approximate","country": "UA","city": "Kiev"}}] # JSON parser def parse_json_response(response: str) -> list: cleaned = response.replace("json", "").replace("```", "").replace("\n", "") print(cleaned) return json.loads(cleaned) def query_openai(prompt: str) -> str: try: response = openai_client.responses.create( model="gpt-4.1", temperature=0.5, max_output_tokens=10000, tools=openai_tools, input=prompt, ) return response.output_text except Exception as e: return f"❌ Error: {e}" # Parallel LLM execution def run_parallel(prompts: List[str], query_fn, max_workers: int = 5) -> List[str]: results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {executor.submit(query_fn, prompt): prompt for prompt in prompts} for future in as_completed(futures): try: res = parse_json_response(future.result()) results.append(res) except Exception as e: print(e) pass return results #Parallel LLM execution def run_parallel_wo_validation(prompts: List[str], query_fn, max_workers: int = 5) -> List[str]: results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {executor.submit(query_fn, prompt): prompt for prompt in prompts} for future in as_completed(futures): results.append(future.result()) return results # Prompt builders JSON_INSTRUCTION = 'Поверни list з структурою ["name", "name", ..., "name", "name"]. Поверни тільки цей list. Без будь-якого іншого тексту' def build_company_prompt(query: str) -> str: return f"""Твоя задача надати мені список топ 10 компаній, які пов`язані з (входять в) {query}. Надай список із 10 успішних компаній які відносяться до {query}. Шукай все що напряму пов'язано з {query}""" + JSON_INSTRUCTION def build_people_prompt(area: str) -> str: return f"""Твоя задача надати мені список топ 5 успішних людей, які пов`язані з {area}. Надай список 5 успішних людей (ім`я та прізвище). Шукай все що напряму пов'язано з {area} (компанії, дочірні компанії, співробітники, і тд)""" + JSON_INSTRUCTION def build_books_prompt(name: str, query: str) -> str: return f"""Ти найкращий пошуковий помічник по книгам у світі. Надай мені список книг, які рекомендував {name}. Мене цікавлять тільки назви книг та автори, не пиши нічого іншого. В відповіді повинні бути лише назва книги. Проаналізуй всі статті, публікації, згадки у соціальних мережах про {name} і знайди саме книги які ця людина рекомендувала. Ці книги потрібні щоб віповісти на запит користувача рекомендації книг від {name} зі сфери {query}. Якщо ти нічого не зміг знайти поверни "". БІЛЬШЕ НІЧОГО НЕ ПИШИ.""" # Generator function for progressive output def on_click(query, context): query = f"{query} {context}".strip() if context else query.strip() log = "" log += "🔍 Шукаю пов`язані бізнеси...\n" yield log.strip(), gr.update(visible=False) company_responses = run_parallel([build_company_prompt(query)] * 3, query_openai) company_responses = list(set([item for sublist in company_responses for item in sublist])) log += "👤 Шукаю найуспішніших людей у цій сфері...\n" yield log.strip(), gr.update(visible=False) company_responses.append(query) print(company_responses) people_prompts = [build_people_prompt(f"{k} from {query}") for k in list(company_responses)] people_responses = run_parallel(people_prompts, query_openai) people_responses = list(set([item for sublist in people_responses for item in sublist])) print(people_responses) people = people_responses log += f"✅ Знайдено людей: {', '.join(people)}\n" log += "📚 Збираю рекомендації книжок...\n" yield log.strip(), gr.update(visible=False) book_prompts = [build_books_prompt(name, query) for name in people] book_results = run_parallel_wo_validation(book_prompts, query_openai, max_workers=10) print(book_results) all_raw_text = " ".join(filter(lambda x: x not in (None, '', '""', "''"), book_results)) #log += all_raw_text extract_prompt = f"Твоя задача дістати із всього тексту, який я тобі надішлю, всі книги та написати їх автора. Поверни лише назву книги та авторів. Сформуй список із всіх цих книг разом з автором та поверни його. Видали дублікати, якщо вони є. Весь текст: {all_raw_text}. Поверни лише новий список з авторами і нічого більше." final_response = query_openai(extract_prompt) books = final_response.strip().split("\n") def format_books_for_textbox(book_list): # Filter out empty entries and clean the text cleaned_books = [book.strip() for book in book_list if book.strip()] # Remove duplicates while preserving order seen = set() unique_books = [] for book in cleaned_books: if book.lower() not in seen: seen.add(book.lower()) unique_books.append(book) # Format with proper line breaks and bullet points if not unique_books: return "На жаль, не вдалося знайти рекомендовані книги для цієї сфери." return "\n\n".join(f"📖 {book}" for book in unique_books) log += "✅ Готово! Книги, рекомендовані експертами у цій сфері:" yield log.strip(), gr.update(value=format_books_for_textbox(books), visible=True) # Gradio interface with gr.Blocks(title="📚 BookRecommender") as demo: gr.Markdown(""" # 📚 BookRecommender Введи сферу або компанію, і ми зберемо книжки, які рекомендують найуспішніші люди в цій галузі. """) query_input = gr.Textbox(label="Назва сфери або компанії", placeholder="Наприклад: Universe Group, Genesis, IT-компанії України") context_input = gr.Textbox(label="Контекст", placeholder="Наприклад: в Україні, на світовому ринку") search_button = gr.Button("🔍 Знайти книжки") status_box = gr.Textbox(label="Прогрес виконання", interactive=False) book_output = gr.Textbox(label="Список рекомендованих книжок", lines=20, interactive=True, visible=False) search_button.click( fn=on_click, inputs=[query_input, context_input], outputs=[status_box, book_output] ) if __name__ == "__main__": demo.launch()