From b850a11bb4986ceaa683c4c4e2735adb481fb08e Mon Sep 17 00:00:00 2001 From: sinuhet Date: Wed, 19 Mar 2025 23:31:00 +0100 Subject: [PATCH] =?UTF-8?q?P=C5=99idat=20robovojtik.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- robovojtik.py | 141 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100644 robovojtik.py diff --git a/robovojtik.py b/robovojtik.py new file mode 100644 index 0000000..712bdc9 --- /dev/null +++ b/robovojtik.py @@ -0,0 +1,141 @@ +import openai +import subprocess +import time +import configparser +import json + +# === 1. Načtení konfigurace === +config = configparser.ConfigParser() +config.read("config.ini") + +openai.api_key = config["OpenAI"]["api_key"] +ASSISTANT_ID = config["OpenAI"]["assistant_id"] +thread_id = None + +# === 2. Definice funkcí pro asistenta === +FUNCTIONS = [ + { + "name": "execute_shell_command", + "description": "Spustí shellový příkaz a vrátí jeho výstup.", + "parameters": { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "Shellový příkaz k vykonání." + } + }, + "additionalProperties": False, + "required": ["command"] + } + } +] + +def vytvor_nove_vlakno(): + """Vytvoří nové vlákno pro zachování historie.""" + thread = openai.beta.threads.create() + return thread.id + +def spust_prikaz(prikaz): + """Spustí shellový příkaz a vrátí jeho výstup.""" + try: + output = subprocess.check_output(prikaz, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) + return output.strip() + except subprocess.CalledProcessError as e: + return f"Chyba při vykonávání příkazu:\n{e.output}" + +def posli_dotaz_do_assistenta(prompt): + """Pošle dotaz do asistenta a získá návrh příkazu.""" + global thread_id + if thread_id is None: + thread_id = vytvor_nove_vlakno() + + 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) + return messages.data[0].content[0].text.value + +def posli_prikaz_do_assistenta(command): + """Pošle asistentovi schválený příkaz k provedení.""" + global thread_id + if thread_id is None: + thread_id = vytvor_nove_vlakno() + + 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"]) + 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) + return messages.data[0].content[0].text.value + +def main(): + print("Linuxový shell s OpenAI Assistant v2 API a funkcemi. Napiš 'exit' pro ukončení.") + + while True: + user_input = input("\nZadej příkaz v přirozeném jazyce: ") + if user_input.strip().lower() == "exit": + break + + assistant_odpoved = posli_dotaz_do_assistenta(user_input) + print("\nRobovojtík navrhuje příkaz:") + print(assistant_odpoved) + + potvrzeni = input("\nSpustit tento příkaz? (y/n): ").strip().lower() + if potvrzeni == "y": + vysledek = posli_prikaz_do_assistenta(assistant_odpoved) + print("\nVýstup příkazu:") + print(vysledek) + else: + print("❌ Příkaz nebyl spuštěn.") + +if __name__ == "__main__": + main()