OperationcodeCreatorAgent:ソース:tools:FunctionCreatorAgent.py

EmotionalBehaviorAgent.pyから抜き出しただけで、まだ動作確認していない。


from tools.AIAgent import AIAgent
from tools.command_list import load_ai_agent_name_list
from tools.program_called_command_list import load_python_program_list
from tools.program_called_command_list import get_python_program_list
from tools.program_called_command_list import append_python_program_function_direct
from tools.command_list import run_pip_command
import tools.command_list 
import streamlit as st



class FunctionCreatorAgent():
    def __init__(self, max_check=3):
        name = "FunctionCreator"
        self.agent_dict = load_ai_agent_name_list("FunctionCreator")
        print(self.agent_dict)
        self.agent = AIAgent(name,
                             "",
                             [],
                             False)
        self.max_check = max_check
        self.temperature = 0.5
        self.pre_emotional_respons = ""
        self.latest_respons = ""
        load_python_program_list()

    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):
        program_list_prompt = get_python_program_list()
        if "" == program_list_prompt:
            pre_promput = "現在扱える機能はありません。\r\n"
        else:
            pre_promput = "現在扱える機能は\r\n"+program_list_prompt

        # アイデアだし
        self.agent.update_system_prompt(pre_promput+self.__get_agent_prompt("ProgramIdeaGeneration"))
        # コールバック関数の設定 (エージェントの動作の可視化用)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons("まだ機能として作られていない機能のアイデアを書いてください。")
        # 批評
        self.agent.update_system_prompt(pre_promput+self.__get_agent_prompt("ProgramCriticism"))
        # コールバック関数の設定 (エージェントの動作の可視化用)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons("")
        # 修正
        self.agent.update_system_prompt(pre_promput+self.__get_agent_prompt("Fixes"))
        # コールバック関数の設定 (エージェントの動作の可視化用)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons("")
        # まとめ
        self.agent.update_system_prompt(pre_promput+self.__get_agent_prompt("ProgramSummary"))
        # コールバック関数の設定 (エージェントの動作の可視化用)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons("")
        # 作成
        self.agent.update_system_prompt(pre_promput+self.__get_agent_prompt("PythonProgramer"))
        self.update_tools([run_pip_command])
        # コールバック関数の設定 (エージェントの動作の可視化用)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons("まだ機能として作られていない機能のプログラムを書いてください。")
        error, respons = self.respons_programe_check(respons)
        self.latest_respons = respons
        self.save_python_functions(respons)

        return respons

    def save_python_functions(self, respons):
        """レスポンスに"""
        code_list = tools.fllow_controller.get_program_code_list_from_response(respons)
        if "python" in code_list:
            # 複数のコードがある場合書くコードについてエラーチェックをして保存する。
            if list is type(code_list["python"]):
                for code in code_list["python"]:
                    function_name, explanation, program = self.get_append_python_program_function_parameter(code)
                    if None is not function_name:
                        append_python_program_function_direct(function_name, explanation, program)

    def get_append_python_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.fllow_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):
        # pythonで エラーが出るようなら自動で作りなおしを依頼。

        count = 0
        respons_text = check_prompt
        loop_flag = True
        error_respons = ""
        while loop_flag:
            error_respons = tools.fllow_controller.python_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("PythonProgramer"):
            # エージェントを実行
            response = self.agent.get_respons(prompt)
            st.write(response)
        return response