from tools.AIAgent import AIAgent
from tools.command_list import load_ai_agent_name_list
from tools.command_list import get_main_agent_name_list
from tools.operational_command_list import load_operation_program_list
from tools.operational_command_list import append_operation_program_function_direct
from flow.flow_controller import get_lengest_program_code_from_response
from flow.AIOperation import AIOperation
import tools.command_list
import streamlit as st
class OperationcodeCreatorAgent():
def __init__(self, max_check=3):
name = "OperationcodeCreator"
self.agent_dict = load_ai_agent_name_list("OperationcodeCreator")
print(self.agent_dict)
self.agent = AIAgent(name,
"",
[],
False)
self.max_check = max_check
self.temperature = 0.5
self.latest_respons = ""
load_operation_program_list()
self.ai_operation = AIOperation()
def __get_agent_prompt(self, name):
return self.agent_dict[name][1]
# AIAgentと関数的互換性確保のために必要
def get_respons(self, command):
return self.FunctionCreate(command)
def clear_memory(self):
self.agent.clear_memory()
def update_temperature(self, temperature):
self.temperature = temperature
self.agent.update_temperature(temperature)
def update_tools(self, tools):
self.agent.update_tools(tools)
################################################
def FunctionCreate(self, command):
load_ai_agent_name_list()
agent_list_str = get_main_agent_name_list()
pre_prompt = "\r\n現在利用できるAIAgentは\r\n" + agent_list_str + "\r\n-------\r\n"
# アイデアだし
self.agent.update_system_prompt(self.__get_agent_prompt("OperationcodeCreator"))
# コールバック関数の設定 (エージェントの動作の可視化用)
tools.command_list.g_time_keeper.wait()
respons = self.agent.get_respons(pre_prompt + command+"\r\n-----\r\nを実現するための作業手順を指示プログラムで作成してください。")
#respons = self.agent.get_respons("なんでprcess:Agent名\"::\"Agentに渡すプロンプト で::のように:を2つ続けているの？")
# 批評
self.agent.update_system_prompt(self.__get_agent_prompt("OperationcodeCriticism"))
# コールバック関数の設定 (エージェントの動作の可視化用)
tools.command_list.g_time_keeper.wait()
respons = self.agent.get_respons(command+"\r\nの実現に対して指示プログラムを批評してください。")
# 修正
self.agent.update_system_prompt(self.__get_agent_prompt("OperationcodeCreator"))
# コールバック関数の設定 (エージェントの動作の可視化用)
tools.command_list.g_time_keeper.wait()
respons = self.agent.get_respons(pre_prompt + "批評に基づいて\r\n"+command+"\r\n----\r\nを実現するための作業手順の指示プログラムを修正してください。")
operatoin = get_lengest_program_code_from_response(respons)
self.ai_operation.analyze_operation_code(operatoin)
result,respons = self.ai_operation.run(command)
return respons
def save_operation_functions(self, respons):
"""レスポンスに"""
code_list = tools.flow_controller.get_program_code_list_from_response(respons)
if "operation" in code_list.keys():
# 複数のコードがある場合書くコードについてエラーチェックをして保存する。
if list is type(code_list["operation"]):
for code in code_list["operation"]:
function_name, explanation, program = self.get_append_operation_program_function_parameter(code)
if None is not function_name:
append_operation_program_function_direct(function_name, explanation, program)
def get_append_operation_program_function_parameter(self, program):
count = 0
code_list = program.split("\n")
name = None
explanation = None
for line in code_list:
l_line = line.rstrip()
if 0 < len(l_line):
if "#" == l_line[0]:
l_line = l_line[1:]
l_line = l_line.strip()
if 0 == count:
name = l_line
if 1 == count:
explanation = l_line
break
count += 1
return name, explanation, program
#########################################
def __programe_check_sub(self, respons_text, error_respons, count):
if False is tools.flow_controller.get_code_error():
error_respons = ""
return False, error_respons, respons_text, count
if self.max_check <= count:
error_respons = "3回以上やり直してもエラーを修正しきれませんでした。"
return False, error_respons, respons_text, count
print(error_respons)
tools.command_list.g_time_keeper.wait()
if 0 <= error_respons.find("ソースコードが前から変わっていません。"):
self.update_temperature(1.0)
respons_text = self.agent.get_respons(error_respons+"\r\n直してください。必要ならrun_pip_commandを使用してください。")
count += 1
return True, error_respons, respons_text, count
def respons_programe_check(self, check_prompt, timeout=5):
# operationで エラーが出るようなら自動で作りなおしを依頼。
count = 0
respons_text = check_prompt
loop_flag = True
error_respons = ""
while loop_flag:
error_respons = tools.flow_controller.operation_error_check(respons_text, error_respons, timeout)
loop_flag, error_respons, respons_text, count = self.__programe_check_sub(respons_text, error_respons, count)
count += 1
return error_respons, respons_text
#################################
def think_agent(self, prompt):
tools.command_list.g_time_keeper.wait()
# AIエージェントの思考
with st.chat_message("operationProgramer"):
# エージェントを実行
response = self.agent.get_respons(prompt)
st.write(response)
return response
