streamlitはmain関数を何度も呼び出してくれる。ただ、そのため何度も初期化コードが呼ばれやすく、AiAgetntの初期化がむつかしい。
RoleAssignmentOfficerを単純な分配人として、メモリを1回ごとに初期化しているが、現在あまりうまくいっていない。調整が必要。もともとはマネージャーとしてもっと複雑なことをやらせていたが、それはそれでうまくいかなかった。
respons_programe_checkでpythonプログラムがエラーを出す間は、直させる処理をしている。
import os
import streamlit as st
from langchain_community.callbacks import StreamlitCallbackHandler
# custom tools
from tools.command_list import set_work_space
from tools.command_list import load_ai_agent_name_list
from tools.command_list import get_ai_agent
from tools.command_list import get_tool_list
from tools.command_list import get_next_agent
import tools.command_list
import tools.fllow_controller
from tools.AIAgent import AIAgent
import msvcrt
################################################
##########################################################
def init_page():
st.set_page_config(
page_title="In front of AGI start gate.",
page_icon=""
)
st.header("In front of AGI start gate.")
st.sidebar.title("Options")
def init_messages(ai_agent):
clear_button = st.sidebar.button("Clear Conversation", key="clear")
if clear_button or "messages" not in st.session_state:
st.session_state.messages = [
{"role": "assistant", "content": "入力してください。"}
]
ai_agent.clear_memory()
def create_ai_agent(name, private_memory=False):
return AIAgent(name, get_ai_agent(name), get_tool_list(), private_memory)
def cange_agent(agent, name):
agent.update_system_prompt(get_ai_agent(name))
def think_agent(ai_agent, name, prompt):
tools.command_list.g_time_keeper.wait()
# AIエージェントの思考
print("think_agent name", name, prompt)
with st.chat_message(name):
# コールバック関数の設定 (エージェントの動作の可視化用)
st_cb = StreamlitCallbackHandler(
st.container(), expand_new_thoughts=True)
# エージェントを実行
response = ai_agent.get_respons(prompt, st_cb)
st.write(response["output"])
print("response", response["output"])
return response["output"]
def respons_programe_check(ai_agent, agent_name, check_prompt):
# 次の作業者を決定。
count = 0
respons_text = check_prompt
# pythonで エラーが出るようなら自動で作りなおしを依頼。
while True:
error_respons = tools.fllow_controller.python_error_check(respons_text)
if False is tools.fllow_controller.get_code_error():
error_respons = ""
return error_respons, respons_text
if 3 <= count:
error_respons = "3回以上やり直してもエラーを修正しきれませんでした。"
return error_respons, respons_text
print(error_respons)
respons_text = think_agent(ai_agent, agent_name, error_respons+"\r\n直してください。")
count += 1
def assign_roles(ra_ai_agent, work_prompt):
# 次の作業者を決定。
ra_ai_agent.clear_memory()
think_agent(ra_ai_agent, "RoleAssignmentOfficer",
work_prompt + "\r\n この作業の続きをだれに作業させるのがいいですか？")
next_agent = get_next_agent()
if None is next_agent:
# 見つからなければもう一度だけ確認
think_agent(ra_ai_agent, "RoleAssignmentOfficer", "該当するエージェントが見つかりませんでした。get_agent_name_listの機能で確認してください。")
next_agent = get_next_agent()
else:
pass
return next_agent, work_prompt
def get_next_agent_program_check(ai_agent, ra_ai_agent, agent_name, respons):
# 次の作業者を決定。
r_prompt, respons = respons_programe_check(ai_agent, agent_name, respons)
next_agent, respons = assign_roles(ra_ai_agent, respons)
return next_agent, respons
def main():
loop_count_max = 6
loop_count = 0
now_agent_name = "programer"
pre_agent_name = now_agent_name
load_ai_agent_name_list()
ra_ai_agent = create_ai_agent("RoleAssignmentOfficer", True)
ai_agent = create_ai_agent(now_agent_name)
init_page()
init_messages(ai_agent)
set_work_space(os.getcwd())
print("current folder", os.getcwd())
for msg in st.session_state['memory'].chat_memory.messages:
st.chat_message(msg.type).write(msg.content)
# セッション状態に値がある場合のみテキスト入力欄を表示
if user_prompt := st.chat_input(placeholder="作業内容を入れてください"):
st.chat_message("user").write(user_prompt)
ra_ai_agent.clear_memory()
think_agent(ra_ai_agent, "RoleAssignmentOfficer",
user_prompt + "\r\n この作業をだれに作業させるのがいいですか？")
now_agent_name = get_next_agent()
prompt = user_prompt
if None is not now_agent_name and "User" != now_agent_name:
while True:
cange_agent(ai_agent, now_agent_name)
respons = think_agent(ai_agent, now_agent_name, prompt)
pre_agent_name = now_agent_name
# 分配者でないとき
now_agent_name, respons =\
get_next_agent_program_check(ai_agent,
ra_ai_agent,
pre_agent_name,
respons)
tools.command_list.g_time_keeper.wait()
if None is now_agent_name:
break
if "User" == now_agent_name:
break
# キー入力があれば、確認。ESCでいったん作業終了
input_data = msvcrt.getch()
if input_data == b'\x1b': # escキー
print("escキーが押されました")
break
prompt = now_agent_name + "さん続きをお願い。"
if loop_count_max <= loop_count:
now_agent_name = "User"
print("count over")
break
loop_count += 1
print("loop_count", loop_count)
print("end_of_main")
if __name__ == '__main__':
main()
AIによる解説
AGI スタートゲート前：リファレンスマニュアル
このプログラムは、複数のAIエージェントを連携させて、ユーザーの指示に基づいたタスクを実行するシステムです。
システムの概要
- AIエージェント: それぞれ異なる役割を持つAIエージェントが複数存在します。
- 役割割り当てオフィサー: ユーザーの指示を受け取り、適切なAIエージェントにタスクを割り当てます。
- タスク実行: 割り当てられたAIエージェントは、タスクを実行し、結果をユーザーに返します。
- エラー処理: エージェントがエラーを発生させた場合、自動的に修正を試みます。
使用方法
- 作業内容の入力: ユーザーは、作業内容をテキスト入力欄に入力します。
- 役割割り当て: 役割割り当てオフィサーが、入力された作業内容に基づいて、適切なAIエージェントを決定します。
- タスク実行: 決定されたAIエージェントが、タスクを実行します。
- 結果の表示: AIエージェントは、実行結果をユーザーに表示します。
- エラー処理: エラーが発生した場合、AIエージェントは自動的に修正を試みます。
- 次の作業者の決定: 役割割り当てオフィサーは、次の作業を行うAIエージェントを決定します。
- ループ: ステップ3から6を繰り返します。
AIエージェント
- programer: プログラムを作成します。
- RoleAssignmentOfficer: ユーザーの指示に基づいて、適切なAIエージェントを決定します。
- その他: システムに登録されている他のAIエージェント。
コマンド
- get_ai_agent_name_list: 登録されているAIエージェントの名前一覧を取得します。
- get_ai_agent: 指定されたAIエージェントの情報を取得します。
- get_tool_list: AIエージェントが使用できるツールのリストを取得します。
- get_next_agent: 次の作業を行うAIエージェントを決定します。
- set_work_space: 作業ディレクトリを設定します。
- python_error_check: Pythonコードのエラーをチェックします。
注意点
- エージェントがエラーを発生させた場合、自動的に修正を試みますが、必ずしも成功するとは限りません。
- エラーが修正できない場合は、ユーザーが手動で修正する必要があります。
- システムは、ユーザーの指示に基づいて動作します。ユーザーの指示が曖昧な場合、システムは適切な動作を行えない可能性があります。
参考情報
コード例
# AIエージェントの思考
print("think_agent name", name, prompt)
with st.chat_message(name):
# コールバック関数の設定 (エージェントの動作の可視化用)
st_cb = StreamlitCallbackHandler(
st.container(), expand_new_thoughts=True)
# エージェントを実行
response = ai_agent.get_respons(prompt, st_cb)
st.write(response["output"])
print("response", response["output"])
return response["output"]まとめ
このプログラムは、複数のAIエージェントを連携させて、ユーザーの指示に基づいたタスクを実行するシステムです。ユーザーは、作業内容を入力するだけで、AIエージェントが自動的にタスクを実行します。エラーが発生した場合、システムは自動的に修正を試みます。