""" Module: api_interface.py -------------------------- Obsahuje funkce pro komunikaci s OpenAI API a správu konverzačního vlákna. """ import openai import json import time import threading import logging from configparser import ConfigParser logger = logging.getLogger("robovojtik.api_interface") config = ConfigParser() config.read("config.ini") try: openai.api_key = config["OpenAI"]["api_key"] except KeyError: raise ValueError("API key not found in config.ini. Please set the 'api_key' under [OpenAI] section or set OPENAI_API_KEY environment variable.") ASSISTANT_ID = config["OpenAI"].get("assistant_id", None) if not ASSISTANT_ID: raise ValueError("assistant_id not found in config.ini. Please set the 'assistant_id' under [OpenAI] section.") # Globální proměnná pro uchování ID konverzačního vlákna thread_id = None def vytvor_nove_vlakno(): """ Vytvoří nové vlákno konverzace s asistentem pomocí OpenAI API. """ global thread_id thread = openai.beta.threads.create() thread_id = thread.id logger.debug(f"Vytvořeno nové vlákno: {thread_id}") return thread_id def posli_dotaz_do_assistenta(prompt): """ Odesílá dotaz v přirozeném jazyce do asistenta a vrací jeho odpověď. """ global thread_id if thread_id is None: vytvor_nove_vlakno() logger.debug(f"Odesílám dotaz: {prompt}") openai.beta.threads.messages.create( thread_id=thread_id, role="user", content=prompt ) run = openai.beta.threads.runs.create( thread_id=thread_id, assistant_id=ASSISTANT_ID ) while True: run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) if run_status.status == "completed": break time.sleep(1) messages = openai.beta.threads.messages.list(thread_id=thread_id) answer = messages.data[0].content[0].text.value logger.debug(f"Asistent odpověděl: {answer}") return answer def posli_prikaz_do_assistenta(command): """ Odesílá schválený příkaz k vykonání do asistenta a vrací jeho odpověď. Před odesláním se příkaz vyčistí. """ from shell_functions import spust_prikaz, vytvor_skript # Importujeme zde, abychom předešli cyklickým závislostem. global thread_id if thread_id is None: vytvor_nove_vlakno() logger.debug(f"Odesílám příkaz k vykonání: {command}") run = openai.beta.threads.runs.create( thread_id=thread_id, assistant_id=ASSISTANT_ID, instructions=f"Prosím, spusť tento příkaz: {command}" ) while True: run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) if run_status.status == "requires_action": tool_calls = run_status.required_action.submit_tool_outputs.tool_calls tool_outputs = [] for tool_call in tool_calls: tool_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) if tool_name == "execute_shell_command": vysledek = spust_prikaz(arguments["command"]) elif tool_name == "create_script": vysledek = vytvor_skript(arguments["file_name"], arguments["content"]) else: vysledek = "Neznámá funkce." tool_outputs.append({ "tool_call_id": tool_call.id, "output": vysledek }) openai.beta.threads.runs.submit_tool_outputs( thread_id=thread_id, run_id=run.id, tool_outputs=tool_outputs ) elif run_status.status == "completed": break time.sleep(1) messages = openai.beta.threads.messages.list(thread_id=thread_id) answer = messages.data[0].content[0].text.value logger.debug(f"Výsledek spuštěného příkazu: {answer}") return answer def volani_asistenta(prompt, spinner_func=None): """ Spustí dotaz do asistenta v samostatném vlákně a během čekání volá spinner_func. Vrací odpověď asistenta. Pokud dojde k chybě, vrací prázdný řetězec. """ result_container = {} def worker(): try: odpoved = posli_dotaz_do_assistenta(prompt) result_container['answer'] = odpoved except Exception as ex: result_container['answer'] = f"Chyba při volání asistenta: {str(ex)}" logger.exception("Chyba v worker funkci volani_asistenta.") thread = threading.Thread(target=worker) thread.start() idx = 0 spinner = ["|", "/", "-", "\\"] while thread.is_alive(): if spinner_func: spinner_func(spinner[idx % len(spinner)]) time.sleep(0.2) idx += 1 thread.join() return result_container.get('answer', "")