nová verze

This commit is contained in:
sinuhet 2025-03-19 23:45:30 +01:00
parent 7bf4a40a0c
commit 39b6842b98

View File

@ -3,8 +3,9 @@ import subprocess
import time import time
import configparser import configparser
import json import json
import sys
# === 1. Načtení konfigurace === # === 1. Načtení konfigurace z config.ini ===
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read("config.ini") config.read("config.ini")
@ -12,7 +13,7 @@ openai.api_key = config["OpenAI"]["api_key"]
ASSISTANT_ID = config["OpenAI"]["assistant_id"] ASSISTANT_ID = config["OpenAI"]["assistant_id"]
thread_id = None thread_id = None
# === 2. Definice funkcí pro asistenta === # === 2. Definice funkcí, které asistent může volat ===
FUNCTIONS = [ FUNCTIONS = [
{ {
"name": "execute_shell_command", "name": "execute_shell_command",
@ -31,6 +32,7 @@ FUNCTIONS = [
} }
] ]
# === 3. Pomocné funkce ===
def vytvor_nove_vlakno(): def vytvor_nove_vlakno():
"""Vytvoří nové vlákno pro zachování historie.""" """Vytvoří nové vlákno pro zachování historie."""
thread = openai.beta.threads.create() thread = openai.beta.threads.create()
@ -45,65 +47,59 @@ def spust_prikaz(prikaz):
return f"Chyba při vykonávání příkazu:\n{e.output}" return f"Chyba při vykonávání příkazu:\n{e.output}"
def posli_dotaz_do_assistenta(prompt): def posli_dotaz_do_assistenta(prompt):
"""Pošle dotaz do asistenta a získá návrh příkazu.""" """Odesílá dotaz (v přirozeném jazyce) do asistenta a vrací jeho návrh příkazu."""
global thread_id global thread_id
if thread_id is None: if thread_id is None:
thread_id = vytvor_nove_vlakno() thread_id = vytvor_nove_vlakno()
openai.beta.threads.messages.create( openai.beta.threads.messages.create(
thread_id=thread_id, thread_id=thread_id,
role="user", role="user",
content=prompt content=prompt
) )
run = openai.beta.threads.runs.create( run = openai.beta.threads.runs.create(
thread_id=thread_id, thread_id=thread_id,
assistant_id=ASSISTANT_ID assistant_id=ASSISTANT_ID
) )
while True: while True:
run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
if run_status.status == "completed": if run_status.status == "completed":
break break
time.sleep(1) time.sleep(1)
messages = openai.beta.threads.messages.list(thread_id=thread_id) messages = openai.beta.threads.messages.list(thread_id=thread_id)
return messages.data[0].content[0].text.value return messages.data[0].content[0].text.value
def posli_prikaz_do_assistenta(command): def posli_prikaz_do_assistenta(command):
"""Pošle asistentovi schválený příkaz k provedení.""" """Odesílá schválený příkaz k vykonání asistentovi pomocí funkce execute_shell_command."""
global thread_id global thread_id
if thread_id is None: if thread_id is None:
thread_id = vytvor_nove_vlakno() thread_id = vytvor_nove_vlakno()
run = openai.beta.threads.runs.create( run = openai.beta.threads.runs.create(
thread_id=thread_id, thread_id=thread_id,
assistant_id=ASSISTANT_ID, assistant_id=ASSISTANT_ID,
instructions=f"Prosím, spusť tento příkaz: {command}" instructions=f"Prosím, spusť tento příkaz: {command}"
) )
while True: while True:
run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
if run_status.status == "requires_action": if run_status.status == "requires_action":
tool_calls = run_status.required_action.submit_tool_outputs.tool_calls tool_calls = run_status.required_action.submit_tool_outputs.tool_calls
tool_outputs = [] tool_outputs = []
for tool_call in tool_calls: for tool_call in tool_calls:
tool_name = tool_call.function.name tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments) arguments = json.loads(tool_call.function.arguments)
if tool_name == "execute_shell_command": if tool_name == "execute_shell_command":
vysledek = spust_prikaz(arguments["command"]) vysledek = spust_prikaz(arguments["command"])
else: else:
vysledek = "Neznámá funkce." vysledek = "Neznámá funkce."
tool_outputs.append({ tool_outputs.append({
"tool_call_id": tool_call.id, "tool_call_id": tool_call.id,
"output": vysledek "output": vysledek
}) })
openai.beta.threads.runs.submit_tool_outputs( openai.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id, thread_id=thread_id,
run_id=run.id, run_id=run.id,
@ -111,31 +107,55 @@ def posli_prikaz_do_assistenta(command):
) )
elif run_status.status == "completed": elif run_status.status == "completed":
break break
time.sleep(1) time.sleep(1)
messages = openai.beta.threads.messages.list(thread_id=thread_id) messages = openai.beta.threads.messages.list(thread_id=thread_id)
return messages.data[0].content[0].text.value return messages.data[0].content[0].text.value
# === 4. Hlavní interaktivní smyčka ===
def main(): def main():
print("Linuxový shell s OpenAI Assistant v2 API a funkcemi. Napiš 'exit' pro ukončení.") print("Linuxový shell s OpenAI Assistant v2 API a funkcemi.")
print("Pro přímý příkaz použij prefix 'cmd:'")
print("Pro ukončení Robovojtíka zadej 'vypni' nebo 'exit'.")
while True: while True:
user_input = input("\nZadej příkaz v přirozeném jazyce: ") user_input = input("\nZadej příkaz v přirozeném jazyce: ").strip()
if user_input.strip().lower() == "exit":
break # Ukončení shellu Robovojtíka
if user_input.lower() in ["vypni", "exit"]:
assistant_odpoved = posli_dotaz_do_assistenta(user_input) potvrzeni = input("Skutečně chceš ukončit Robovojtíka? (y/n): ").strip().lower()
if potvrzeni == "y":
print("Ukončuji Robovojtíka...")
sys.exit(0)
else:
print("Ukončení zrušeno. Pokračuji.")
continue
# Přímý příkaz bez asistenta
if user_input.startswith("cmd:"):
command = user_input[4:].strip()
print(f"Rozpoznán přímý příkaz: {command}")
potvrzeni = input("Spustit tento příkaz? (y/n): ").strip().lower()
if potvrzeni == "y":
vysledek = spust_prikaz(command)
print("\nVýstup příkazu:")
print(vysledek)
else:
print("Příkaz nebyl spuštěn.")
continue
# Normální dotaz pro asistenta, který navrhne příkaz
assistant_response = posli_dotaz_do_assistenta(user_input)
print("\nRobovojtík navrhuje příkaz:") print("\nRobovojtík navrhuje příkaz:")
print(assistant_odpoved) print(assistant_response)
potvrzeni = input("\nSpustit tento příkaz? (y/n): ").strip().lower() potvrzeni = input("Spustit tento příkaz? (y/n): ").strip().lower()
if potvrzeni == "y": if potvrzeni == "y":
vysledek = posli_prikaz_do_assistenta(assistant_odpoved) execution_result = posli_prikaz_do_assistenta(assistant_response)
print("\nVýstup příkazu:") print("\nVýstup příkazu:")
print(vysledek) print(execution_result)
else: else:
print("❌ Příkaz nebyl spuštěn.") print("Příkaz nebyl spuštěn. Čekám na další vstup.")
if __name__ == "__main__": if __name__ == "__main__":
main() main()