[LLM]アプリ全自動開発"ChatDev"のソースコード解説
ChatDevは作ってほしい製品の要望を出すだけで、あとはAIがうまいこと自動開発してくれるツールです。
例えば、「スネークゲーム(餌を食べてどんどん体が伸びていくやつ)を作って」と言うと、数分待つだけで実際に遊べるスネークゲームが出来上がります。
今回は、その論文と共にアップされたソースコードについて解説していきます。
前回の記事はこちら
ChatDevの概要
AIに専門職になりきってもらうと、その分野について良い返答をするようになるので、例えば、「あなたはプロのプログラマーです」というと、返ってくるプログラムの精度が良くなったりします。
で、ChatDevは専門職を複数用意して、実際のソフトウェア開発会社を真似をすることで、質の高いソフトを作れるようにしよう、というアイデアです。
また、そのための工夫として、チャットチェーンを特徴としてあげています。
・チャットチェーン
まずはソフトウェア開発プロセスを参考にフェーズがいくつか用意されてます。
「設計」「コーディング」「テスト」「文書化」の4フェーズです。
通常、これらのフェーズは質の高いコミュニケーションを必要とします。
それを可能にするため、各専門家でペアを組んでもらい、指示役・アシスタント(というより作業員？)の役割分担のもと、相互に協力して問題解決に向かってもらう、というのが概要みたいです。
例えばコーディングのフェーズでは「最高技術責任者(指示役)」と「プログラマー(アシスタント)」がペアを組んでたりします。
全体の処理の流れ
まず、専門職を用意します。
・最高経営責任者
・最高製品責任者
・最高技術責任者
・プログラマー
・コードレビューアー
・ソフトウェアテストエンジニア
・最高クリエイティブ責任者
そして以下の流れをループします。
1、フェーズ進行
2、指示役がプロンプトを作成
3、アシスタントがそのプロンプト等をOpenAIにAPIを送って、返答をもらう
4、返答内容が要件を満たすものならそれを保存して次のフェーズへ移行、満たさなければ2に戻る
これの繰り返しです。
指示役とアシスタント、というか、各専門職には専用のプロンプトが組み込まれているので、自分の職と指示役のメッセージを一緒に送ることで、チャットチェーンと呼んでいる相互協力を作っているようです。
例えば、「最高技術責任者(指示役)」と「プログラマー(アシスタント)」がペアだったとして、
システムプロンプト「ここはソフトウェア開発会社だ。具体的にはうんぬん…」
プログラマープロンプト「僕はプログラマーだ。僕の役割はうんぬん…」
最高技術責任者プロンプト「こういうのを作ってくれ！」
こういうのを一式まとめてOpenAIにAPI飛ばすと、プログラマー視点で要件に沿ったものがレスポンスされるわけです。
APIを理解してる人向けに話すと、それぞれが
systemプロンプト
assistantプロンプト
userプロンプト
に該当します。
assistantは要はAIで、userは僕ら人間みたいなものなので、
「AIはプログラマーになりきってね。で、こういうの作ってね」と送っているのと同等になります。
・プロンプト
ソースコードにあるプロンプト、ものすごく分量が多いのでここではほとんど割愛しますが、一例だけ紹介します。あとは実際にここ見て日本語に訳すと良いと思います。
(urlがttps://になっているので先頭にhをつけてhttps://にしてください)
ttps://github.com/OpenBMB/ChatDev/tree/main/CompanyConfig/Default
・PhaseConfig.json
{"Coding": {
"assistant_role_name": "Programmer",
"user_role_name": "Chief Technology Officer",
"phase_prompt": [
"According to the new user's task and our software designs listed below: ",
"Task: \"{task}\".",
"Modality: \"{modality}\".",
"Programming Language: \"{language}\"",
"Ideas:\"{ideas}\"",
"We have decided to complete the task through a executable software with multiple files implemented via {language}. As the {assistant_role}, to satisfy the new user's demands, you should write one or multiple files and make sure that every detail of the architecture is, in the end, implemented as code. {gui}",
"Think step by step and reason yourself to the right decisions to make sure we get it right.",
"You will first lay out the names of the core classes, functions, methods that will be necessary, as well as a quick comment on their purpose.",
"Then you will output the content of each file including complete code. Each file must strictly follow a markdown code block format, where the following tokens must be replaced such that \"FILENAME\" is the lowercase file name including the file extension, \"LANGUAGE\" in the programming language, \"DOCSTRING\" is a string literal specified in source code that is used to document a specific segment of code, and \"CODE\" is the original code:",
"FILENAME",
"```LANGUAGE",
"'''",
"DOCSTRING",
"'''",
"CODE",
"```",
"You will start with the \"main\" file, then go to the ones that are imported by that file, and so on.",
"Please note that the code should be fully functional. Ensure to implement all functions. No placeholders (such as 'pass' in Python)."
]
}
}"Coding"が、フェーズ名
"assistant_role_name"がアシスタントの職の名前
"user_role_name"が指示役の職の名前 です。
"phase_prompt"が、そのフェーズで指示役がアシスタントに出すプロンプトです。
上に書いた例でいうと、
最高技術責任者プロンプト「こういうのを作ってくれ！」
に該当します。
"Task: \"{task}\".",みたいに、{ } で囲まれているところがいくつかありますが、それは処理の流れの中で保持された文字が入ります。
例えば、僕がツールに対して「スネークゲーム作って！」と冒頭に指示を出すと、Taskの欄に「スネークゲーム作って！」という言葉がそのまま入っていた気がします。
・RoleConfig.json
"Programmer": [
"{chatdev_prompt}",
"You are Programmer. we are both working at ChatDev. We share a common interest in collaborating to successfully complete a task assigned by a new customer.",
"You can write/create computer software or applications by providing a specific programming language to the computer. You have extensive computing and coding experience in many varieties of programming languages and platforms, such as Python, Java, C, C++, HTML, CSS, JavaScript, XML, SQL, PHP, etc,.",
"Here is a new customer's task: {task}.",
"To complete the task, you must write a response that appropriately solves the requested instruction based on your expertise and customer's needs."
],"Programmer"が役職名です。
で、その下に複数行書いてあることは全て、その役職になりきってもらうためのプロンプトです。
上に書いた例でいうと、
プログラマープロンプト「僕はプログラマーだ。僕の役割はうんぬん…」
に該当します。
ソースコード解説
処理の流れ自体は上で書いたのが全てなので、ここからはソースコードを読む人向けに、ヒントになりそうな部分をちょこっと書きます。
・run.py
全てはここから始まります。下を見ていくと
# ----------------------------------------
# Pre Processing
# ----------------------------------------
chat_chain.pre_processing()
# ----------------------------------------
# Personnel Recruitment
# ----------------------------------------
chat_chain.make_recruitment()
# ----------------------------------------
# Chat Chain
# ----------------------------------------
chat_chain.execute_chain()
# ----------------------------------------
# Post Processing
# ----------------------------------------
chat_chain.post_processing()こんなのがありますが、chat_chain.execute_chain()以外は初期化とか事後処理みたいな細々としたやつなので、メインを追いたい場合はchat_chain.execute_chain()だけ見ればいいと思います。
・chat_chain.py
run.pyからすぐにdef execute_chain に最初にたどり着くと思います。
def execute_chain(self):
"""
execute the whole chain based on ChatChainConfig.json
Returns: None
"""
for phase_item in self.chain:
self.execute_step(phase_item)この self.chainは、CompanyConfig/Default/ChatChainConfig.jsonに書いてあるフェーズ情報がそのまま入ってます。なので、ChatChainConfig.jsonに書いてある順番で処理が進みます。
DemandAnalysis -> LanguageChoose -> Coding -> …
def execute_step(self, phase_item: dict):
"""
execute single phase in the chain
Args:
phase_item: single phase configuration in the ChatChainConfig.json
Returns:
"""
phase = phase_item['phase']
phase_type = phase_item['phaseType']
# For SimplePhase, just look it up from self.phases and conduct the "Phase.execute" method
if phase_type == "SimplePhase":
max_turn_step = phase_item['max_turn_step']
need_reflect = check_bool(phase_item['need_reflect'])
if phase in self.phases:
self.chat_env = self.phases[phase].execute(self.chat_env,
self.chat_turn_limit_default if max_turn_step <= 0 else max_turn_step,
need_reflect)
else:
raise RuntimeError(f"Phase '{phase}' is not yet implemented in chatdev.phase")
# For ComposedPhase, we create instance here then conduct the "ComposedPhase.execute" method
elif phase_type == "ComposedPhase":
cycle_num = phase_item['cycleNum']
composition = phase_item['Composition']
compose_phase_class = getattr(self.compose_phase_module, phase)
if not compose_phase_class:
raise RuntimeError(f"Phase '{phase}' is not yet implemented in chatdev.compose_phase")
compose_phase_instance = compose_phase_class(phase_name=phase,
cycle_num=cycle_num,
composition=composition,
config_phase=self.config_phase,
config_role=self.config_role,
model_type=self.model_type,
log_filepath=self.log_filepath)
self.chat_env = compose_phase_instance.execute(self.chat_env)
else:
raise RuntimeError(f"PhaseType '{phase_type}' is not yet implemented.")次にくる関数です。
基本的にフェーズはSimplePhaseとComposedPhaseに分けられます。フェーズ単体で済むものと、フェーズ中にさらに細かいフェーズがあるかどうかです。
self.chat_env = self.phases[phase].executeで、フェーズが実行されます。
・phase.py
次はここにきます。
def execute(self, chat_env, chat_turn_limit, need_reflect) -> ChatEnv:
self.update_phase_env(chat_env)
if "ModuleNotFoundError" in self.phase_env['test_reports']:
chat_env.fix_module_not_found_error(self.phase_env['test_reports'])
log_and_print_online(
f"Software Test Engineer found ModuleNotFoundError:\n{self.phase_env['test_reports']}\n")
pip_install_content = ""
for match in re.finditer(r"No module named '(\S+)'", self.phase_env['test_reports'], re.DOTALL):
module = match.group(1)
pip_install_content += "{}\n```{}\n{}\n```\n".format("cmd", "bash", f"pip install {module}")
log_and_print_online(f"Programmer resolve ModuleNotFoundError by:\n{pip_install_content}\n")
self.seminar_conclusion = "nothing need to do"
else:
self.seminar_conclusion = \
self.chatting(chat_env=chat_env,
task_prompt=chat_env.env_dict['task_prompt'],
need_reflect=need_reflect,
assistant_role_name=self.assistant_role_name,
user_role_name=self.user_role_name,
phase_prompt=self.phase_prompt,
phase_name=self.phase_name,
assistant_role_prompt=self.assistant_role_prompt,
user_role_prompt=self.user_role_prompt,
chat_turn_limit=chat_turn_limit,
placeholders=self.phase_env)
chat_env = self.update_chat_env(chat_env)
return chat_env上は細々としたエラー処理なので、
とりあえず下段のself.seminar_conclusion = self.chattingだけ見ればいいと思います。
self.seminar_conclusionとはそのフェーズにおける結論の文章が格納されていて、self.chatting関数でチャットというか専門職同士の話し合いを開始します。
def chatting(
self,
chat_env,
task_prompt: str,
assistant_role_name: str,
user_role_name: str,
phase_prompt: str,
phase_name: str,
assistant_role_prompt: str,
user_role_prompt: str,
task_type=TaskType.CHATDEV,
need_reflect=False,
with_task_specify=False,
model_type=ModelType.GPT_3_5_TURBO,
placeholders=None,
chat_turn_limit=10
) -> str:
"""
Args:
chat_env: global chatchain environment TODO: only for employee detection, can be deleted
task_prompt: user query prompt for building the software
assistant_role_name: who receives the chat
user_role_name: who starts the chat
phase_prompt: prompt of the phase
phase_name: name of the phase
assistant_role_prompt: prompt of assistant role
user_role_prompt: prompt of user role
task_type: task type
need_reflect: flag for checking reflection
with_task_specify: with task specify
model_type: model type
placeholders: placeholders for phase environment to generate phase prompt
chat_turn_limit: turn limits in each chat
Returns:
"""
if placeholders is None:
placeholders = {}
assert 1 <= chat_turn_limit <= 100
if not chat_env.exist_employee(assistant_role_name):
raise ValueError(f"{assistant_role_name} not recruited in ChatEnv.")
if not chat_env.exist_employee(user_role_name):
raise ValueError(f"{user_role_name} not recruited in ChatEnv.")
# init role play
role_play_session = RolePlaying(
assistant_role_name=assistant_role_name,
user_role_name=user_role_name,
assistant_role_prompt=assistant_role_prompt,
user_role_prompt=user_role_prompt,
task_prompt=task_prompt,
task_type=task_type,
with_task_specify=with_task_specify,
model_type=model_type,
)
# log_and_print_online("System", role_play_session.assistant_sys_msg)
# log_and_print_online("System", role_play_session.user_sys_msg)
# start the chat
_, input_user_msg = role_play_session.init_chat(None, placeholders, phase_prompt)
seminar_conclusion = None
# handle chats
# the purpose of the chatting in one phase is to get a seminar conclusion
# there are two types of conclusion
# 1. with "<INFO>" mark
# 1.1 get seminar conclusion flag (ChatAgent.info) from assistant or user role, which means there exist special "<INFO>" mark in the conversation
# 1.2 add "<INFO>" to the reflected content of the chat (which may be terminated chat without "<INFO>" mark)
# 2. without "<INFO>" mark, which means the chat is terminated or normally ended without generating a marked conclusion, and there is no need to reflect
for i in range(chat_turn_limit):
# start the chat, we represent the user and send msg to assistant
# 1. so the input_user_msg should be assistant_role_prompt + phase_prompt
# 2. then input_user_msg send to LLM and get assistant_response
# 3. now we represent the assistant and send msg to user, so the input_assistant_msg is user_role_prompt + assistant_response
# 4. then input_assistant_msg send to LLM and get user_response
# all above are done in role_play_session.step, which contains two interactions with LLM
# the first interaction is logged in role_play_session.init_chat
assistant_response, user_response = role_play_session.step(input_user_msg, chat_turn_limit == 1)
conversation_meta = "**" + assistant_role_name + "<->" + user_role_name + " on : " + str(
phase_name) + ", turn " + str(i) + "**\n\n"
# TODO: max_tokens_exceeded errors here
if isinstance(assistant_response.msg, ChatMessage):
# we log the second interaction here
log_and_print_online(role_play_session.assistant_agent.role_name,
conversation_meta + "[" + role_play_session.user_agent.system_message.content + "]\n\n" + assistant_response.msg.content)
if role_play_session.assistant_agent.info:
seminar_conclusion = assistant_response.msg.content
break
if assistant_response.terminated:
break
if isinstance(user_response.msg, ChatMessage):
# here is the result of the second interaction, which may be used to start the next chat turn
log_and_print_online(role_play_session.user_agent.role_name,
conversation_meta + "[" + role_play_session.assistant_agent.system_message.content + "]\n\n" + user_response.msg.content)
if role_play_session.user_agent.info:
seminar_conclusion = user_response.msg.content
break
if user_response.terminated:
break
# continue the chat
if chat_turn_limit > 1 and isinstance(user_response.msg, ChatMessage):
input_user_msg = user_response.msg
else:
break
# conduct self reflection
if need_reflect:
if seminar_conclusion in [None, ""]:
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session, phase_name,
chat_env)
if "recruiting" in phase_name:
if "Yes".lower() not in seminar_conclusion.lower() and "No".lower() not in seminar_conclusion.lower():
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session,
phase_name,
chat_env)
elif seminar_conclusion in [None, ""]:
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session, phase_name,
chat_env)
else:
seminar_conclusion = assistant_response.msg.content
log_and_print_online("**[Seminar Conclusion]**:\n\n {}".format(seminar_conclusion))
seminar_conclusion = seminar_conclusion.split("<INFO>")[-1]
return seminar_conclusion次にここに決ます。くっそ長いですが、見る点は二つだけです。
# init role play
role_play_session = RolePlaying(
assistant_role_name=assistant_role_name,
user_role_name=user_role_name,
assistant_role_prompt=assistant_role_prompt,
user_role_prompt=user_role_prompt,
task_prompt=task_prompt,
task_type=task_type,
with_task_specify=with_task_specify,
model_type=model_type,
)openaiにapi送る前に、準備として指示役とアシスタントを詰め込んだクラスを作ってます。それだけです。
for i in range(chat_turn_limit):
# start the chat, we represent the user and send msg to assistant
# 1. so the input_user_msg should be assistant_role_prompt + phase_prompt
# 2. then input_user_msg send to LLM and get assistant_response
# 3. now we represent the assistant and send msg to user, so the input_assistant_msg is user_role_prompt + assistant_response
# 4. then input_assistant_msg send to LLM and get user_response
# all above are done in role_play_session.step, which contains two interactions with LLM
# the first interaction is logged in role_play_session.init_chat
assistant_response, user_response = role_play_session.step(input_user_msg, chat_turn_limit == 1)for i in range(chat_turn_limit):とありますが、納得する回答が出なかった時にループしたりしなかったりします。
コーディングフェーズみたいにフェーズ内フェーズがあったりすると別のとこでループ処理するので、ここではループしないです。というか他フェーズでもこの箇所でループしてた記憶があまりないです。
で、
assistant_response, user_response = role_play_session.step(i
で、OpenAIにapi送るための準備段階に入ります。
・role_playing.py
次にここにきます。
def step(
self,
user_msg: ChatMessage,
assistant_only: bool,
) -> Tuple[ChatAgentResponse, ChatAgentResponse]:
assert isinstance(user_msg, ChatMessage), print("broken user_msg: " + str(user_msg))
# print("assistant...")
user_msg_rst = user_msg.set_user_role_at_backend()
assistant_response = self.assistant_agent.step(user_msg_rst)
if assistant_response.terminated or assistant_response.msgs is None:
return (
ChatAgentResponse([assistant_response.msgs], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([], False, {}))
assistant_msg = self.process_messages(assistant_response.msgs)
if self.assistant_agent.info:
return (ChatAgentResponse([assistant_msg], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([], False, {}))
self.assistant_agent.update_messages(assistant_msg)
if assistant_only:
return (
ChatAgentResponse([assistant_msg], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([], False, {})
)
# print("user...")
assistant_msg_rst = assistant_msg.set_user_role_at_backend()
user_response = self.user_agent.step(assistant_msg_rst)
if user_response.terminated or user_response.msgs is None:
return (ChatAgentResponse([assistant_msg], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([user_response], user_response.terminated, user_response.info))
user_msg = self.process_messages(user_response.msgs)
if self.user_agent.info:
return (ChatAgentResponse([assistant_msg], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([user_msg], user_response.terminated, user_response.info))
self.user_agent.update_messages(user_msg)
return (
ChatAgentResponse([assistant_msg], assistant_response.terminated, assistant_response.info),
ChatAgentResponse([user_msg], user_response.terminated, user_response.info),
)assistant_response = self.assistant_agent.step(user_msg_rst)
というところで、アシスタントに指示役のプロンプトを合成して、OpenAIのAPI使ってます。
self.assistant_agentクラス自体に専門職のプロンプトを持っていて、user_msg_rstに指示役のプロンプトが入っているので、
上の例で書いた、
システムプロンプト「ここはソフトウェア開発会社だ。具体的にはうんぬん…」
プログラマープロンプト「僕はプログラマーだ。僕の役割はうんぬん…」
最高技術責任者プロンプト「こういうのを作ってくれ！」
というのをデータとしてまとめたあとに、それをAPIで送ってサーバーから返答もらってる感じです。
self.assistant_agent.update_messages(assistant_msg)
ここで履歴に追加してます。履歴は使ったり使わなかったりします。
だいたい使わないのですが、それは別の箇所で履歴代わりの変数があるからです。(chat_envとphase_envです)
その下の
if assistant_only:
ここでだいたい処理終了します。もしここで終了しなかったら、
さっきやった指示役とアシスタントを逆にして実行するだけです。
・phase.py
@log_arguments
def chatting(
self,
chat_env,
task_prompt: str,
assistant_role_name: str,
user_role_name: str,
phase_prompt: str,
phase_name: str,
assistant_role_prompt: str,
user_role_prompt: str,
task_type=TaskType.CHATDEV,
need_reflect=False,
with_task_specify=False,
model_type=ModelType.GPT_3_5_TURBO,
placeholders=None,
chat_turn_limit=10
) -> str:
"""
Args:
chat_env: global chatchain environment TODO: only for employee detection, can be deleted
task_prompt: user query prompt for building the software
assistant_role_name: who receives the chat
user_role_name: who starts the chat
phase_prompt: prompt of the phase
phase_name: name of the phase
assistant_role_prompt: prompt of assistant role
user_role_prompt: prompt of user role
task_type: task type
need_reflect: flag for checking reflection
with_task_specify: with task specify
model_type: model type
placeholders: placeholders for phase environment to generate phase prompt
chat_turn_limit: turn limits in each chat
Returns:
"""
if placeholders is None:
placeholders = {}
assert 1 <= chat_turn_limit <= 100
if not chat_env.exist_employee(assistant_role_name):
raise ValueError(f"{assistant_role_name} not recruited in ChatEnv.")
if not chat_env.exist_employee(user_role_name):
raise ValueError(f"{user_role_name} not recruited in ChatEnv.")
# init role play
role_play_session = RolePlaying(
assistant_role_name=assistant_role_name,
user_role_name=user_role_name,
assistant_role_prompt=assistant_role_prompt,
user_role_prompt=user_role_prompt,
task_prompt=task_prompt,
task_type=task_type,
with_task_specify=with_task_specify,
model_type=model_type,
)
# log_and_print_online("System", role_play_session.assistant_sys_msg)
# log_and_print_online("System", role_play_session.user_sys_msg)
# start the chat
_, input_user_msg = role_play_session.init_chat(None, placeholders, phase_prompt)
seminar_conclusion = None
# handle chats
# the purpose of the chatting in one phase is to get a seminar conclusion
# there are two types of conclusion
# 1. with "<INFO>" mark
# 1.1 get seminar conclusion flag (ChatAgent.info) from assistant or user role, which means there exist special "<INFO>" mark in the conversation
# 1.2 add "<INFO>" to the reflected content of the chat (which may be terminated chat without "<INFO>" mark)
# 2. without "<INFO>" mark, which means the chat is terminated or normally ended without generating a marked conclusion, and there is no need to reflect
for i in range(chat_turn_limit):
# start the chat, we represent the user and send msg to assistant
# 1. so the input_user_msg should be assistant_role_prompt + phase_prompt
# 2. then input_user_msg send to LLM and get assistant_response
# 3. now we represent the assistant and send msg to user, so the input_assistant_msg is user_role_prompt + assistant_response
# 4. then input_assistant_msg send to LLM and get user_response
# all above are done in role_play_session.step, which contains two interactions with LLM
# the first interaction is logged in role_play_session.init_chat
assistant_response, user_response = role_play_session.step(input_user_msg, chat_turn_limit == 1)
conversation_meta = "**" + assistant_role_name + "<->" + user_role_name + " on : " + str(
phase_name) + ", turn " + str(i) + "**\n\n"
# TODO: max_tokens_exceeded errors here
if isinstance(assistant_response.msg, ChatMessage):
# we log the second interaction here
log_and_print_online(role_play_session.assistant_agent.role_name,
conversation_meta + "[" + role_play_session.user_agent.system_message.content + "]\n\n" + assistant_response.msg.content)
if role_play_session.assistant_agent.info:
seminar_conclusion = assistant_response.msg.content
break
if assistant_response.terminated:
break
if isinstance(user_response.msg, ChatMessage):
# here is the result of the second interaction, which may be used to start the next chat turn
log_and_print_online(role_play_session.user_agent.role_name,
conversation_meta + "[" + role_play_session.assistant_agent.system_message.content + "]\n\n" + user_response.msg.content)
if role_play_session.user_agent.info:
seminar_conclusion = user_response.msg.content
break
if user_response.terminated:
break
# continue the chat
if chat_turn_limit > 1 and isinstance(user_response.msg, ChatMessage):
input_user_msg = user_response.msg
else:
break
# conduct self reflection
if need_reflect:
if seminar_conclusion in [None, ""]:
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session, phase_name,
chat_env)
if "recruiting" in phase_name:
if "Yes".lower() not in seminar_conclusion.lower() and "No".lower() not in seminar_conclusion.lower():
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session,
phase_name,
chat_env)
elif seminar_conclusion in [None, ""]:
seminar_conclusion = "<INFO> " + self.self_reflection(task_prompt, role_play_session, phase_name,
chat_env)
else:
seminar_conclusion = assistant_response.msg.content
log_and_print_online("**[Seminar Conclusion]**:\n\n {}".format(seminar_conclusion))
seminar_conclusion = seminar_conclusion.split("<INFO>")[-1]
return seminar_conclusionクソ長い関数に戻ってきました。
現在はここです。
assistant_response, user_response = role_play_session.step(input_user_msg, chat_turn_limit == 1)
といっても、もうほとんど見なくていいです。
細々とした処理が色々ありますけど、最後の行
return seminar_conclusion
で、さっきのAPIのメッセージ内容を綺麗な加工したものを返してるだけです。
def execute(self, chat_env, chat_turn_limit, need_reflect) -> ChatEnv:
self.update_phase_env(chat_env)
if "ModuleNotFoundError" in self.phase_env['test_reports']:
chat_env.fix_module_not_found_error(self.phase_env['test_reports'])
log_and_print_online(
f"Software Test Engineer found ModuleNotFoundError:\n{self.phase_env['test_reports']}\n")
pip_install_content = ""
for match in re.finditer(r"No module named '(\S+)'", self.phase_env['test_reports'], re.DOTALL):
module = match.group(1)
pip_install_content += "{}\n```{}\n{}\n```\n".format("cmd", "bash", f"pip install {module}")
log_and_print_online(f"Programmer resolve ModuleNotFoundError by:\n{pip_install_content}\n")
self.seminar_conclusion = "nothing need to do"
else:
self.seminar_conclusion = \
self.chatting(chat_env=chat_env,
task_prompt=chat_env.env_dict['task_prompt'],
need_reflect=need_reflect,
assistant_role_name=self.assistant_role_name,
user_role_name=self.user_role_name,
phase_prompt=self.phase_prompt,
phase_name=self.phase_name,
assistant_role_prompt=self.assistant_role_prompt,
user_role_prompt=self.user_role_prompt,
chat_turn_limit=chat_turn_limit,
placeholders=self.phase_env)
chat_env = self.update_chat_env(chat_env)
return chat_env処理を終えたので戻ってきました。
chat_env = self.update_chat_env(chat_env)
で、結果をうまいこと分別して、履歴として保存してます。それを次のフェーズに渡します。
あとは関数をどんどん戻っていって、
def execute_chain(self):
"""
execute the whole chain based on ChatChainConfig.json
Returns: None
"""
for phase_item in self.chain:
self.execute_step(phase_item)で、次のフェーズに移るだけです。
基本はこうですが、実は重要な別ルートがあって見落としてたなんてことがあるかもしれません。
ガチでプロジェクトに組み込みたい人は上記を参考にソースコードをブレークポイントしかけながら一行一行追っていくと良いと思います。
この記事が気に入ったらサポートをしてみませんか？