概要
こんにちは、KDDIアジャイル開発センター新入社員のひさふるです。今回はAmazon BedrockとLangChainの勉強を兼ねて、簡易的なスクラムを行うLLM Agentを作ってみました。
結果としてはまたまだ精度は良くないですが、自律的にスクラムを進められるAgentが作成出来たかなと思っています。LLMを使ったタスクの具体化等の技術に興味のある方は、ぜひご覧ください。(詳しい方からのアドバイスなどあれば、ぜひいただきたいです...！)
はじめに
まず、今回LLM Agentを作成するに至った経緯について説明します。
Agentの課題
今回、LLMの勉強のために簡単なプログラムを組んでみようと思い立ったのですが、単純なAPIの呼び出し等は既にやり尽くされているので、(新入社員の私としては)少しチャレンジングなAgentの作成をやってみようと思いつきました。
LLMにおけるAgentとはユーザーからの指示を受けて自律的にタスクを実行するプログラムのことですが、LLM用フレームワークであるLangChainに実装されているAgentはシンプルながら非常に良く出来ており、Action、Observation、Thoughtを繰り返すことで自律的に目標を分析・行動し、タスクを解決することが出来ます。
しかし、Action、Observation、Thoughtを繰り返すような単純なAgentには巨大/ハイレベルなタスクに対応出来ないという問題点があります。
この問題点を解決するために、タスク細分化という方法が提案されています。
例えばSELFGOALというAgentの研究では、ゴールを達成するために必要なサブゴールを定義し、更にそのサブゴールを達成するためのタスクを定義するという方法で、目標の具体化・細分化を行っています。
ゴール
├─ サブゴール1
│ ├─ タスク1-1
│ └─ タスク1-2
├─ サブゴール2
│ ├─ タスク2-1
│ └─ タスク2-2
︙
また、他の解決策としてはLLM同士にディベートをさせるという方法もあります。1つのLLMには小さなタスクのみをさせ、複数のLLMから成るチームとして最終的な意思決定をさせよう、というのがこの方法です。
しかし別の研究では、LLMは同調圧力に弱かったり、逆に一度自信を持つと自分の意見に固執する傾向があるようで、研究によって段々とLLMの持つ人格が判明しつつあります。この傾向が影響し、LLMに集団行動をさせると同調圧力により、上手く最適解を導き出せないという問題も存在するようです。
この問題に対処しようとしたのが以下の研究で、ディベート(集団での議論)とリフレクション(１人で熟考)を組み合わせることで、タスクの正答率を上げようというものです。具体的には、ディベートを先に行ってからリフレクションを行ったほうが、正答率が良くなるようです。
ここまでの話をまとめると、現状のLLM Agentの課題は大きく以下の2つと言えるでしょう。
- 巨大なタスクに対応するためには、タスクの細分化・具体化が必要不可欠である
- LLMには性格(解答の傾向)が存在するため、集団で協議させる場合はその性格を考慮する必要がある
これは私達が普段チームで協働するときの課題と似ていますね。ということは、私達が普段チームで開発するときのフレームワークをLLM Agentにも適用したら、精度が良くなるのでは...？
スクラム開発をLLM Agentに応用する
私の勤めているKDDIアジャイル開発センターは、その名前の通りアジャイル開発を主軸に開発を行う企業です。そんなアジャイル開発の最も有名なフレームワークの1つがスクラムです。
スクラムでは最初に、プロダクトゴールとして達成したい目標を掲げ、それを達成するためのタスク一覧であるプロダクトバックログを作成します。次に、スプリントという短い期間の中で、プロダクトバックログの中から今回のスプリントで達成したいものをスプリントゴールとして定義します。スプリント中では、更にスプリントゴールを達成するための具体的なタスクの一覧であるスプリントバックログを計画し、それに沿って実装を進めます。スプリントの最後には出来上がった成果物(インクリメント)に対するレビューを行い、必要であればプロダクトバックログリファインメントとして計画の修正を行います。
私は、このスクラム開発の特徴は大きく以下の2つであると考えています。
- プロダクトバックログとスプリントバックログをもとにした目標の具体化と柔軟な開発
- スクラムマスターの存在による強いチーム
1.については上記で説明した通りです。2.のスクラムマスターとは、スクラム開発を行う上で全体のサポートを行う存在であり、スクラムガイドでは以下のように説明されています。
スクラムマスターは、スクラムガイドで定義されたスクラムを確立させることの結果に責任を持つ。
すなわちスクラムマスターとは、開発がスクラムというフレームワークを逸脱せず、様々な関係者間でスムーズな開発が出来るようチームのサポートを行う立場、と言えると私は考えています。
ここまでの話をもとに、Agentの持つ課題とスクラム開発の特徴を照らし合わせると...
| Agentの課題 | スクラム開発の特徴(強み) |
|---|---|
| タスクの具体化・細分化 | バックログを用いた目標の具体化と柔軟な開発 |
| LLMの性格に応じたチームマネジメント | スクラムマスターによるチームマネジメント |
このように、Agentの課題に対してスクラム開発の強みが上手く合致すると思いませんか？
ということで、(前置きが長くなりましたが)、スクラム開発というフレームワークがLLM Agentに対してどれだけ有効なのかを検証するために、今回は簡単なAgentを作成していこうと思います。
Agentの作成
今回の目標
今回は、まず簡単なAgent(もどき)を作成することを目標とし、スクラム開発の流れに則りLLMが自分でタスクを具体化出来るところまでを行います。
Agentとスクラム開発の相性について長々と語りましたが、他の部分(スクラムマスターの挙動、AgentのFunction呼び出し等)は別の機会に実装します。(すなわち実際の開発は行わせず、Agentはタスクの具体化と修正のみを行うものとします。)
今回実装するエージェントの概要を図に示します。
今回のプログラムは大きく分けてプロダクトバックログエージェントとスプリントスプリントバックログエージェント、開発者エージェントに分けられ、それぞれ以下のような役割を持ちます。
- プロダクトバックログエージェント：ユーザーからプロダクトゴール(達成したい目標)の入力を受け付け、プロダクトバックログ(目標を達成するための大まかなタスク一覧)を出力する。また、ステークホルダーからのレビューを受け付け、プロダクトバックログリファインメント(プロダクトバックログの修正)を実施する。
- スプリントバックログエージェント：プロダクトバックログ中のアイテムをもとにスプリントバックログ(プロダクトバックログアイテムを達成するための具体的なタスク一覧)を出力する。前回のスプリントにおいて未達成のタスクがある場合、それも考慮した上でスプリントバックログを出力する。
- 開発者エージェント：スプリントバックログ中のタスクを実行する。本来は関数の実行やプログラムの生成を行う部分であるが、今回は簡単のため一定確率でタスクが成功または失敗するものとする。
実装したプログラム
次に、上記のエージェントを実装したプログラムをご紹介します。
メインプログラム(main.py)
まずはそれぞれのAgentを統括しスクラムを進めるメインプログラムから。main.pyでは最初にプロダクトバックログエージェントを呼び出し、プロダクトバックログを定義させます。
次にそのプロダクトバックログからプロダクトバックログアイテムを抽出し、それをもとにスプリントバックログを作成、スプリントバックログアイテムごとに開発を実施し、未達成のタスクがある場合は記録しておきます。
最後に、開発の実施結果に対して、プログラムの実行者がレビューを行い、必要があればプロダクトバックログリファインメントでプロダクトバックログを修正します。
from product_backlog_agent import ProductBacklogAgent
from sprint_backlog_agent import SprintBacklogAgent
import environment
def main():
model_id = "anthropic.claude-instant-v1"
product_goal = input("プロダクトゴール:")
# プロダクトバックログエージェント
# プロダクトゴールを入力し、プロダクトバックログを定義
pba = ProductBacklogAgent(model_id,product_goal)
pba.create_product_backlog()
pba.print_product_backlog()
count = 1 # スプリント数
unresolved_task = []
#バックログが空になるまでスプリントを繰り返す
while len(pba.product_backlog.backlogs) > 0:
print("===== スプリント {0} =====\n".format(count))
# スプリントゴールの定義
product_backlog_item = pba.product_backlog.backlogs.pop(0)
sprint_goal = product_backlog_item.title + ":" + product_backlog_item.story
print("===== スプリントゴール ======")
print(sprint_goal)
# スプリントゴールからスプリントバックログを作成
sba = SprintBacklogAgent(model_id,product_goal,sprint_goal,unresolved_task)
sba.sprint_planning()
sba.print_sprint_backlog()
# スプリントバックログアイテムをもとに開発を実行
# 未解決タスクがある場合は配列に格納
unresolved_task = []
for item in sba.sprint_backlog.backlogs:
result = environment.action()
if not result:
unresolved_task.append(item)
print("=====未解決タスク=====")
print(unresolved_task)
# スプリントレビューとプロダクトバックログリファインメント
review = input("レビュー(無い場合はそのままEnterを押してください):")
if review != "":
print("===プロダクトバックログリファインメントの実施===")
pba.product_backlog_refinement(review)
pba.print_product_backlog()
count+=1
if __name__ == "__main__":
main()
プロダクトバックログエージェント(product_backlog_agent.py)
プロダクトバックログの管理・生成・リファインメントを行うエージェントとして、product_backlog_agent.pyを実装しました。LLMにプロダクトゴールやレビューの内容を与えることで、バックログの生成やリファインメントを行います。
from typing import List
from base_agent import BaseAgent
from langchain.prompts import PromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
# プロダクトバックログアイテムの定義
class ProductBacklogItem(BaseModel):
number: int = Field(description="プロダクトバックログアイテムの番号")
title: str = Field(description="プロダクトバックログアイテムのタイトル")
story: str = Field(description="プロダクトバックログアイテムの内容")
class ProductBacklog(BaseModel):
backlogs: List[ProductBacklogItem]
class ProductBacklogAgent(BaseAgent):
product_backlog = []
product_goal = ""
def __init__(self,model_id,product_goal) -> None:
super().__init__(model_id)
self.product_goal = product_goal
def print_product_backlog(self):
print("=====プロダクトバックログ======")
for item in self.product_backlog.backlogs:
item_str = "ID:{0} | Title:{1} | Content:{2}".format(item.number,item.title,item.story)
print(item_str)
print("\n")
# 格納されているプロダクトゴールからプロダクトバックログを作成
def create_product_backlog(self):
template = self.load_text("./prompts/product_backlog.txt")
output_parser = PydanticOutputParser(pydantic_object=ProductBacklog)
prompt_template = PromptTemplate(
template = template,
input_variables = ["product_goal"],
partial_variables={"format_instructions": output_parser.get_format_instructions()}
)
prompt = prompt_template.format_prompt(product_goal=self.product_goal)
output = self.llm(prompt.to_string())
self.product_backlog = output_parser.parse(output)
return self.product_backlog
# レビューをもとにプロダクトバックログを修正
def product_backlog_refinement(self,review):
template = self.load_text("./prompts/product_backlog_refinement.txt")
output_parser = PydanticOutputParser(pydantic_object=ProductBacklog)
prompt_template = PromptTemplate(
template = template,
input_variables = ["product_goal", "current_product_backlog", "review"],
partial_variables={"format_instructions": output_parser.get_format_instructions()}
)
prompt = prompt_template.format_prompt(
product_goal=self.product_goal,
current_product_backlog=self.product_backlog,
review=review
)
output = self.llm(prompt.to_string())
self.product_backlog = output_parser.parse(output)
return self.product_backlog
スプリントバックログエージェント(sprint_backlog_agent.py)
最後に、スプリントバックログの生成と管理を行うプログラム(sprint_backlog_agent.py)です。基本構成はプロダクトバックログと同じで、スプリント毎に、プロダクトバックログアイテム等をもとにスプリントバックログを生成します。
from typing import List
from base_agent import BaseAgent
from langchain.prompts import PromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
class SprintBacklogItem(BaseModel):
number: int = Field(description="スプリントバックログアイテムの番号")
title: str = Field(description="スプリントバックログアイテムのタイトル")
story: str = Field(description="スプリントバックログアイテムの内容")
class SprintBacklog(BaseModel):
backlogs: List[SprintBacklogItem]
class SprintBacklogAgent(BaseAgent):
sprint_backlog = []
sprint_goal = ""
def __init__(self,model_id,product_goal,sprint_goal,unresolved_task) -> None:
super().__init__(model_id)
self.product_goal = product_goal
self.sprint_goal = sprint_goal
self.unresolved_task = unresolved_task
def print_sprint_backlog(self):
print("\n=====スプリントバックログ======")
for item in self.sprint_backlog.backlogs:
item_str = "ID:{0} | Title:{1} | Content:{2}".format(item.number,item.title,item.story)
print(item_str)
print("\n")
# スプリントバックログの生成
def sprint_planning(self):
template = self.load_text("./prompts/sprint_backlog.txt")
output_parser = PydanticOutputParser(pydantic_object=SprintBacklog)
prompt_template = PromptTemplate(
template = template,
input_variables = ["product_goal","sprint_goal","unresolved_task"],
partial_variables={"format_instructions": output_parser.get_format_instructions()}
)
prompt = prompt_template.format_prompt(
product_goal=self.product_goal,
sprint_goal=self.sprint_goal,
unresolved_task=self.unresolved_task
)
output = self.llm(prompt.to_string())
self.sprint_backlog = output_parser.parse(output)
return self.sprint_backlog
開発者エージェント(environment.py)
今回は簡単のため実際にはタスク(スプリントバックログアイテム)を実行しないものとし、一定確率で成功または失敗になる処理を実装しました。
import random
def action():
return random.choices([True, False], weights = [0.7,0.3])[0]
if __name__ == "__main__":
for _ in range(10):
print(action())
ベースエージェント(base_agent.py)
他のエージェントプログラムの共通部分となるクラスです。
from langchain_aws import BedrockLLM
from langchain_community.document_loaders import TextLoader
class BaseAgent():
llm = None
def __init__(self,model_id) -> None:
self.llm = BedrockLLM(model_id=model_id)
def load_text(self,path):
loader = TextLoader(path)
document = loader.load()
template = document[0].page_content
return template
与えるプロンプト
LLMに与えるプロンプトは以下のように設定しました。
プロダクトバックログ生成時(product_backlog.txt)
product_goalにはユーザーが入力したプロダクトゴール、format_instructionsにはLangChainが生成する出力形式の指示が与えられます。詳しい使われ方はproduct_backlog_agent.pyをご覧ください。
あなたには、この後ユーザーが実現したいこと(プロダクトゴール)が与えられます。
あなたは、そのプロダクトゴールを実現するために必要なプロダクトバックログを作成してください。
プロダクトゴールは以下のとおりです。
プロダクトゴール：{product_goal}
{format_instructions}
出力は日本語でお願いします。
スプリントバックログ生成時(sprint_backlog.txt)
重要な情報であるproduct_goalを与えつつ、前回のスプリントで達成出来なかったタスク一覧(unresolved_task)を前情報として与えます。また、今回のスプリントの達成目標であるsprint_goalを与えることで、スプリントバックログを生成させています。
現在、次のようなプロダクトゴールの実現に向けてスクラム開発を進めています。
プロダクトゴール:{product_goal}
また、前回のスプリントの未解決タスクとして以下のタスクが残っています。
未解決タスク：{unresolved_task}
あなたには、プロダクトゴールを実現するために、次に実行されるスプリントでが実現したいこと(スプリントゴール)が与えられます。
あなたは、そのスプリントゴールを実現するために必要なスプリントバックログを作成してください。
スプリントゴールは以下のとおりです。
スプリントゴール：{sprint_goal}
{format_instructions}
出力は日本語でお願いします。
プロダクトバックログリファインメント(product_backlog_refinement.txt)
最後に、プロダクトバックログリファインメントを行う際に与えるプロンプトです。現在のプロダクトバックログ(current_product_backlog)に加え、ユーザーからのレビュー(review)を与えることで、プロダクトバックログの修正を促します。
現在、プロダクトバックログは次のようになっています。
====プロダクトバックログ====
{current_product_backlog}
====
今、ステークホルダーから次のようなレビューがありました。
レビュー:{review}
このレビューをもとに、現在のプロダクトバックログを修正し、修正後のプロダクトバックログを出力してください。
ただし、プロダクトゴールを達成するという目標からは逸脱しないでください。
プロダクトゴール：{product_goal}
{format_instructions}
出力は日本語でお願いします。
プログラムの実行結果
では、実際に上記プログラムを実行してみた結果をご紹介します。
プロダクトバックログ生成時
まず、最初にプロダクトバックログエージェントにプロダクトゴールを入力した結果を示します。今回は、プロダクトゴールとして以下のような文言を入力しました。
Twitterを模したSNSのAPIをPythonで実装してください。フロントエンドは既に存在し、今回はバックエンドAPIを作成するものとします。今回は簡易的な実装を心がけ、必要最低限の機能のみを実装してください。テストや検証は必要ありません。
結果として、図中に示すように、エージェントは4つのプロダクトバックログアイテムを出力しました。どれもユーザーの登録やツイートの取得といった、Twitterを作成する上で必要な機能が羅列されており、プロダクトバックログの作成は成功したと言えるでしょう。
スプリント1
次に、生成されたプロダクトバックログをもとにスプリント1を行った結果です。最初のプロダクトバックログアイテムである、
1.ユーザーID、パスワード、名前などの基本情報を受け取り、DBに登録するAPIエンドポイントを作成します。
というタスクをもとに、スプリントバックログエージェントがスプリントバックログを生成しています。スプリントバックログも、大まかな部分は妥当なアイテムが生成されています。
しかし、入力画面の作成など今回指定した範囲外のタスクを含めてしまっているため、ここで与えるプロンプトは改良が必要かもしれません。(今回、フロントエンドは既に存在しているとの仮定があるため、画面作成は必要ありませんでした。)
今回は簡単のため、それぞれのスプリントバックログアイテムが一定確率で成功、または失敗するものとしており、今回のスプリントでは1つのアイテムが未達成タスクとして出力されました。
ユーザー登録処理後、登録成功時と失敗時で異なるレスポンスを返す処理を実装します。
スプリント2
スプリント2では、2.テキストデータと画像データを受け取り、DBにツイート情報を登録し、一覧画面に返すAPIエンドポイントを作成します。というプロダクトバックログアイテムをもとにスプリントバックログが出力されました。スプリントバックログの内容も、目標を達成するために妥当なものとなっています。
また、前回の未達成タスクである登録処理の成功失敗レスポンスを返すがスプリントバックログ含まれており、前回のスプリントの反省が出来ていることもわかります。
加えて、今回はスプリント2のレビューとして、APIエンドポイントの検証をバックログに含めてくださいという指示を入力しました。
スプリント3
スプリント3では、スプリント2で入力したレビューをもとにプロダクトバックログリファインメントが行われ、プロダクトバックログに5つ目のアイテム(APIエンドポイントの検証)が追加されています。エージェントが、ステークホルダーからレビューをもとに、正しくリファインメントを行うことが出来ました。
また、前回の未達成タスクがスプリントバックログに含まれていることも確認出来ました。
スプリント4・スプリント5
残りのスプリント4・5も今までと同様に、プロダクトバックログアイテムをもとにスプリントバックログを生成し、スプリントバックログ内のタスクの消化を繰り返すことで、当初予定していたプロダクトゴールを達成することが出来ました。
(スプリント5のみ、未達成タスクが漏れてしまっていました。)
おわりに
今回はLLM Agentにスクラム開発を行わせる意義からはじめ、簡単なスクラム開発(っぽいこと)が出来るエージェントの実装までを行いました。
今回の結果だけでも、スクラム開発のフレームワークを踏襲することでエージェントがタスクの具体化を行うことができ、さらにいくつかのスプリントに分けてタスクを実行することで、ユーザーのレビューをもとにしたタスクの柔軟な変更が可能となりました。
今回は実際にはタスクの実行は行っておりませんが、最終的にはAgentがタスクの実行(今回の場合は、プログラムの生成)まで行えることを目標としています。ゆくゆくは生成したコードのレビューやコンフリクトの解決等も自動で行えるようにすれば、私達人間は指示を出すだけでプロジェクトを完遂させることが出来るかもしれません。
一方で課題も多数あることがわかりました。今回だけでもLLMが指示通りの動作をしてくれないことが頻繁にあり、指定に無い出力があった場合にどのようにカバーするかが課題となりそうです。
それから、次回以降は機会があれば複数のLLMに対して同一のプロンプトを投げ、異なる意見を出力したときに上手く取りまとめるスクラムマスターエージェントを実装してみたいと思っています。