AutoGen で現実世界の AI ワークフローを構築する方法: ステップバイステップ ガイド

大規模言語モデルを試してみて、Generative AI をアプリに統合し始めたとしたら、それは素晴らしいことです。しかし、現実的に考えてみましょう。LLM は必ずしも私たちが望むように動作するわけではありません。LLM は、自分の意思を持った邪悪な幼児のようなものです。

すぐに、単純なプロンプト チェーンだけでは不十分であることに気付くでしょう。 場合によっては、それ以上のものが必要になります。 場合によっては、マルチエージェント ワークフローが必要になります。 ここでAutoGenの出番です。

例を見てみましょう。メモを取るアプリを作っていると想像してください (明らかに、世の中にはメモを取るアプリが足りません。😝)。でも、何か特別なことをしたいのです。ユーザーから提供されたシンプルで生のメモを、要約、オプションのタイトル、自動化されたタスクの ToDo リストを備えた、完全に再構成されたドキュメントに変換したいのです。そして、このすべてを苦労せずに実現したいのです。少なくとも、AI エージェントにとっては。

さて、皆さんが何を考えているかはわかっています。「これらは初心者向けプログラムのようなものじゃないの？」それに対して、私は、その通りだと言います。意地悪ですが、その通りです。しかし、ワークフローのシンプルさに騙されないでください。ここで学ぶスキル、たとえば AI エージェントの処理、ワークフロー制御の実装、会話履歴の管理などは、AI ゲームを次のレベルに引き上げるのに役立ちます。

さあ、準備はいいですか。AutoGen を使用して AI ワークフローを作成する方法を学びましょう。

始める前に、すべてのソース コードへのリンクがGitHubにあることに注意してください。

カスタムワークフローの構築

最初のユースケース「メモの要約とそれに続く条件付きタイトルの生成」から始めましょう。公平を期すために言うと、ここではエージェントを使用する必要はありません。でも、どこかから始めなければなりませんよね?

ステップ1: ベースLLM構成の作成

AutoGen のようなエージェント フレームワークでは、常にモデル パラメータを構成する必要があります。使用するモデルとフォールバック モデル、温度、さらにはタイムアウトやキャッシュなどの設定についてです。AutoGen の場合、その設定は次のようになります。

 # build the gpt_configuration object base_llm_config = { "config_list": [ { "model": "Llama-3-8B-Instruct", "api_key": os.getenv("OPENAI_API_KEY"), "base_url": os.getenv("OPENAI_API_URL"), } ], "temperature": 0.0, "cache_seed": None, "timeout": 600, }

ご覧のとおり、私はオープンソース AI の大ファンで、Llama 3 を信頼しています。api_key とbase_url値を設定するだけで、AutoGen を任意の OpenAI 互換推論サーバーに向けることができます。したがって、Groq、Together.ai、または vLLM を使用してモデルをローカルにホストapi_keyます。私はInferixを使用しています。

本当に簡単です！

興味があります！オープンソース AI ホスティングに関する同様のガイドに興味がありますか？コメントでお知らせください。

ステップ2: エージェントの作成

AutoGen で会話エージェントを初期化するのは非常に簡単です。システム メッセージとともに基本 LLM 構成を提供するだけで準備完了です。

 import autogen def get_note_summarizer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note. Generate a summary describing what the note is about. The summary must follow the provided "RULES". "RULES": - The summary should be not more than 3 short sentences. - Don't use bullet points. - The summary should be short and concise. - Identify and retain any "catchy" or memorable phrases from the original text - Identify and correct all grammatical errors. - Output the summary and nothing else.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Note_Summarizer", # Lets give our agent a nice name llm_config=base_llm_config, # This is where we pass the llm configuration system_message=system_message, ) def get_title_generator(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note along with a summary. Generate a title based on the user's input. The title must be witty and easy to read. The title should accurate present what the note is about. The title must strictly be less than 10 words. Make sure you keep the title short. Make sure you print the title and nothing else. """ # Create and return our assistant agent return autogen.AssistantAgent( name="Title_Generator", llm_config=base_llm_config, system_message=system_message, )

エージェントを作成する上で最も重要な部分は、 system_messageです。エージェントを構成するために使用したsystem_messageを見てみましょう。

AutoGen の AI エージェントは会話に参加することで機能することを覚えておくことが重要です。会話を解釈して進める方法は、設定されているsystem_messageによって完全に異なります。これは、正しく動作させるために時間を費やすことになる場所の 1 つです。

必要なのは、あと 1 つのエージェントだけです。私たち人間の代理として機能するエージェントです。「メモ」を最初のプロンプトとして会話を開始できるエージェントです。

 def get_user(): # A system message to define the role and job of our agent system_message = "A human admin. Supplies the initial prompt and nothing else." # Create and return our user agent return autogen.UserProxyAgent( name="Admin", system_message=system_message, human_input_mode="NEVER", # We don't want interrupts for human-in-loop scenarios code_execution_config=False, # We definitely don't want AI executing code. default_auto_reply=None, )

ここでは特別なことは何も行われていません。default_auto_reply パラメータdefault_auto_reply Noneに設定していることに留意してください。これは重要です。これを none に設定すると、ユーザー エージェントにメッセージが送信されるたびに会話が終了するようになります。

おっと、エージェントを作成するのを完全に忘れていました。すぐに作成しましょう。

 # Create our agents user = get_user() note_summarizer = get_note_summarizer(base_llm_config) title_generator = get_title_generator(base_llm_config)

ステップ3: GroupChatを使用してエージェントの調整を設定する

パズルの最後のピースは、エージェントを調整させることです。参加の順序を決定し、どのエージェントが参加すべきかを決定する必要があります。

はい、それは 1 個以上です。でも、要点はわかりますよね! 🙈

考えられる解決策の 1 つは、エージェントが参加する順序を AI に判断させることです。これは悪い考えではありません。実際、ワークフローの性質が動的である複雑な問題を扱う場合、これが私の頼みの綱のオプションです。

ただし、このアプローチには欠点があります。現実は再び襲い掛かります。これらの決定を行うエージェントは、多くの場合、大規模なモデルを必要とするため、レイテンシとコストが高くなります。さらに、誤った決定を下すリスクもあります。

事前に手順の順序がわかっている決定論的なワークフローの場合、私は自分で手綱を握って船を操縦することを好みます。幸いなことに、AutoGen はGroupChatと呼ばれる便利な機能でこのユースケースをサポートしています。

 from autogen import GroupChatManager from autogen.agentchat.groupchat import GroupChat from autogen.agentchat.agent import Agent def get_group_chat(agents, generate_title: bool = False): # Define the function which decides the agent selection order def speaker_selection_method(last_speaker: Agent, group_chat: GroupChat): # The admin will always forward the note to the summarizer if last_speaker.name == "Admin": return group_chat.agent_by_name("Note_Summarizer") # Forward the note to the title generator if the user wants a title if last_speaker.name == "Note_Summarizer" and generate_title: return group_chat.agent_by_name("Title_Generator") # Handle the default case - exit return None return GroupChat( agents=agents, messages=[], max_round=3, # There will only be 3 turns in this group chat. The group chat will exit automatically post that. speaker_selection_method=speaker_selection_method, )

GroupChat 、すべてのエージェントがチャットしたりGroupChat作業したりできる WhatsApp グループとして想像してください。この設定により、エージェントは互いの作業に基づいて作業を進めることができます。GroupChat クラスは、 GroupChatManagerと呼ばれるコンパニオン クラスとともにグループ管理者のように機能し、各エージェントが送信するすべてのメッセージを追跡して、全員が会話履歴を把握できるようにします。

上記のコード スニペットでは、カスタムspeaker_selection_methodを使用してGroupChatを作成しました。 speaker_selection_method使用すると、カスタム ワークフローを指定できます。以下は、その視覚的な表現です。

speaker_selection_method基本的に Python 関数なので、これを使って何でもできます。これにより、非常に強力なワークフローを作成できます。たとえば、次のことができます。

• エージェントをペアにして、お互いの作業を確認します。
• 以前のエージェントのいずれかがエラーを出力した場合は、 「トラブルシューター」エージェントを関与させます。
• 外部システムへの Webhook をトリガーして、進捗状況を通知します。

可能性を想像してみてください！😜

ステップ4: すべてを準備し、会話を始める

最後のステップは、 GroupChatのインスタンスを作成し、それをGroupChatManager内にラップして会話を開始することです。

 # Create our group chat groupchat = get_group_chat([user, note_summarizer, title_generator], generate_title=True) manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=base_llm_config) # Start the chat user.initiate_chat( manager, clear_history=True, message=note, )

注: ユーザーは個々のエージェントではなくGroupChatManagerとチャットしています。どのエージェントが会話に参加して最終的な返信をするかはわかりません。ずるいですよね?

出力は次のようになります。

 Admin (to chat_manager): Note: Convo with editor: - discuss titles and thumbnails - discuss video editing tips tracker - Zeeshan presents the tracker - what trick helps with what - he decidedls if we can experiment with something new - make sure all all videos since how config management works in k8s are backed u - make list of YouTube thumbnail templates - make list of YouTube idea generation limits -------------------------------------------------------------------------------- Next speaker: Note_Summarizer Note_Summarizer (to chat_manager): The note is about a conversation with an editor regarding video production. They discussed titles and thumbnails, as well as a video editing tips tracker presented by Zeeshan, which highlights tricks for specific tasks. Additionally, they ensured that all videos on Kubernetes configuration management are backed up and created lists of YouTube thumbnail templates and idea generation limits. -------------------------------------------------------------------------------- Next speaker: Title_Generator Title_Generator (to chat_manager): "Video Production Chat: Titles, Thumbnails, and Editing Tips" --------------------------------------------------------------------------------

会話をコントロールする

次に、最後のユースケースである、特定の「メモ」を取得し、それをよりわかりやすく再構成して、ユーザーのタスク リストを作成します。

具体的な手順は以下のとおりです。

まず、メモで取り上げるトピックのリストを特定します。このリストは、プロセス全体の原動力となります。このリストによって、再フォーマットされたメモのセクションが確立され、生成されるタスクの詳細レベルが決まります。

ただ 1 つ小さな問題があります。Paraphrazer ParaphrazerとTask_Creatorエージェントは、お互いの出力をあまり気にしません。気にするのはTopic_Analyzerの出力だけです。

したがって、エージェントの応答が会話履歴を乱雑にしないようにする方法が必要です。そうしないと、完全な混乱に陥ります。ワークフローの制御はすでに完了しています。今度は、会話履歴のボスになる番です! 😎

ステップ1: エージェントの作成

まず最初に、エージェントを設定する必要があります。詳細を説明して皆さんを退屈させるつもりはないので、コードを以下に示します。

 def get_topic_analyzer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note. Generate a list of topics discussed in that note. The output must obey the following "RULES": "RULES": - Output should only contain the important topics from the note. - There must be atleast one topic in output. - Don't reuse the same text from user's note. - Don't have more than 10 topics in output.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Topic_Analyzer", llm_config=base_llm_config, system_message=system_message, ) def get_paraphrazer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI content editor. The user will provide you a note along with a summary. Rewrite that note and make sure you cover everything in the note. Do not include the title. The output must obey the following "RULES": "RULES": - Output must be in markdown. - Make sure you use each points provided in summary as headers. - Each header must start with `##`. - Headers are not bullet points. - Each header can optionally have a list of bullet points. Don't put bullet points if the header has no content. - Strictly use "-" to start bullet points. - Optionally make an additional header named "Addional Info" to cover points not included in the summary. Use "Addional Info" header for unclassified points. - Identify and correct spelling & grammatical mistakes.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Paraphrazer", llm_config=base_llm_config, system_message=system_message, ) def get_tasks_creator(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI personal assistant. The user will provide you a note along with a summary. Identify each task the user has to do as next steps. Make sure to cover all the action items mentioned in the note. The output must obey the following "RULES": "RULES": - Output must be an YAML object with a field named tasks. - Make sure each task object contains fields title and description. - Extract the title based on the tasks the user has to do as next steps. - Description will be in markdown format. Feel free to include additional formatting and numbered lists. - Strictly use "-" or "dashes" to start bullet points in the description field. - Output empty tasks array if no tasks were found. - Identify and correct spelling & grammatical mistakes. - Identify and fix any errors in the YAML object. - Output should strictly be in YAML with no ``` or any additional text.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Task_Creator", llm_config=base_llm_config, system_message=system_message, )

ステップ2: カスタムGroupChatを作成する

残念ながら、AutoGen では会話履歴を直接制御することはできません。そのため、先に進んで、カスタム実装でGroupChatクラスを拡張する必要があります。

 class CustomGroupChat(GroupChat): def __init__(self, agents): super().__init__(agents, messages=[], max_round=4) # This function get's invoked whenever we want to append a message to the conversation history. def append(self, message: Dict, speaker: Agent): # We want to skip messages from the Paraphrazer and the Task_Creator if speaker.name != "Paraphrazer" and speaker.name != "Task_Creator": super().append(message, speaker) # The `speaker_selection_method` now becomes a function we will override from the base class def select_speaker(self, last_speaker: Agent, selector: AssistantAgent): if last_speaker.name == "Admin": return self.agent_by_name("Topic_Analyzer") if last_speaker.name == "Topic_Analyzer": return self.agent_by_name("Paraphrazer") if last_speaker.name == "Paraphrazer": return self.agent_by_name("Task_Creator") # Return the user agent by default return self.agent_by_name("Admin")

基本GroupChatクラスから 2 つの関数をオーバーライドします。

• append - 会話履歴に追加されるメッセージを制御します。
• select_speaker - これはspeaker_selection_methodを指定する別の方法です。

しかし、ちょっと待ってください。AutoGen のコードを詳しく調べてみると、 GroupChatManagerによって各エージェントが会話履歴も維持されることがわかりました。理由は聞かないでください。本当にわかりません!

そこで、これを修正するためにGroupChatManagerも拡張してみましょう。

 class CustomGroupChatManager(GroupChatManager): def __init__(self, groupchat, llm_config): super().__init__(groupchat=groupchat, llm_config=llm_config) # Don't forget to register your reply functions self.register_reply(Agent, CustomGroupChatManager.run_chat, config=groupchat, reset_config=GroupChat.reset) def run_chat( self, messages: Optional[List[Dict]] = None, sender: Optional[Agent] = None, config: Optional[GroupChat] = None, ) -> Union[str, Dict, None]: """Run a group chat.""" if messages is None: messages = self._oai_messages[sender] message = messages[-1] speaker = sender groupchat = config for i in range(groupchat.max_round): # set the name to speaker's name if the role is not function if message["role"] != "function": message["name"] = speaker.name groupchat.append(message, speaker) if self._is_termination_msg(message): # The conversation is over break # We do not want each agent to maintain their own conversation history history # broadcast the message to all agents except the speaker # for agent in groupchat.agents: # if agent != speaker: # self.send(message, agent, request_reply=False, silent=True) # Pro Tip: Feel free to "send" messages to the user agent if you want to access the messages outside of autogen for agent in groupchat.agents: if agent.name == "Admin": self.send(message, agent, request_reply=False, silent=True) if i == groupchat.max_round - 1: # the last round break try: # select the next speaker speaker = groupchat.select_speaker(speaker, self) # let the speaker speak # We'll now have to pass their entire conversation of messages on generate_reply # Commented OG code: reply = speaker.generate_reply(sender=self) reply = speaker.generate_reply(sender=self, messages=groupchat.messages) except KeyboardInterrupt: # let the admin agent speak if interrupted if groupchat.admin_name in groupchat.agent_names: # admin agent is one of the participants speaker = groupchat.agent_by_name(groupchat.admin_name) # We'll now have to pass their entire conversation of messages on generate_reply # Commented OG code: reply = speaker.generate_reply(sender=self) reply = speaker.generate_reply(sender=self, messages=groupchat.messages) else: # admin agent is not found in the participants raise if reply is None: break # The speaker sends the message without requesting a reply speaker.send(reply, self, request_reply=False) message = self.last_message(speaker) return True, None

元の実装にいくつか小さな編集を加えました。詳細についてはコメントを参照してください。

しかし、強調したいことが 1 つあります。GroupChatManagerの「run_chat」メソッドをオーバーライドして、Apache Airflow や Temporal などの独自のワークフロー エンジンをプラグインすることができます。分散システムの実践者なら、この機能がいかに強力であるかをよくご存知でしょう。

ステップ3: すべてを準備し、会話を始める

これを前例のように設定して、この赤ちゃんがゴロゴロ鳴くのを見てみましょう！🐱

 # Create our agents user = get_user() topic_analyzer = get_topic_analyzer(base_llm_config) paraphrazer = get_paraphrazer(base_llm_config) task_creator = get_tasks_creator(base_llm_config) # Create our group chat groupchat = CustomGroupChat(agents=[user, topic_analyzer, paraphrazer, task_creator]) manager = CustomGroupChatManager(groupchat=groupchat, llm_config=base_llm_config) # Start the chat user.initiate_chat( manager, clear_history=True, message=note, ) # Lets print the count of tasks just for fun chat_messages = user.chat_messages.get(manager) if chat_messages is not None: for message in chat_messages: if message.get("name") == "Task_Creator": taskList = yaml.safe_load(message.get("content")) # type: ignore l = len(taskList.get("tasks")) print(f"Got {l} tasks from Task_Creator.")

出力は次のようになります。

 Admin (to chat_manager): Note: Convo with editor: - discuss titles and thumbnails - discuss video editing tips tracker - Zeeshan presents the tracker - what trick helps with what - he decidedls if we can experiment with something new - make sure all all videos since how config management works in k8s are backed u - make list of YouTube thumbnail templates - make list of YouTube idea generation limits -------------------------------------------------------------------------------- Topic_Analyzer (to chat_manager): Here is the list of topics discussed in the note: 1. Titles 2. Thumbnails 3. Video editing tips 4. Config management in Kubernetes (k8s) 5. YouTube thumbnail templates 6. YouTube idea generation limits -------------------------------------------------------------------------------- Paraphrazer (to chat_manager): Here is the rewritten note in markdown format: ## Titles - Discuss titles and thumbnails with the editor ## Video Editing Tips Tracker ### Zeeshan presents the tracker - What trick helps with what - He decides if we can experiment with something new ## Config Management in Kubernetes (k8s) - Make sure all videos since how config management works in k8s are backed up ## YouTube Thumbnail Templates - Make a list of YouTube thumbnail templates ## YouTube Idea Generation Limits - Make a list of YouTube idea generation limits ## Additional Info - Discuss video editing tips tracker with Zeeshan - Present the tracker and decide if we can experiment with something new -------------------------------------------------------------------------------- Task_Creator (to chat_manager): tasks: - title: Discuss Titles and Thumbnails description: >- - Discuss titles and thumbnails with the editor This task involves having a conversation with the editor to discuss the titles and thumbnails for the videos. - title: Discuss Video Editing Tips Tracker description: >- - Zeeshan presents the tracker - Discuss what trick helps with what - Decide if we can experiment with something new This task involves discussing the video editing tips tracker presented by Zeeshan, understanding what tricks help with what, and deciding if it's possible to experiment with something new. - title: Back up All Videos Since How Config Management Works in k8s description: >- - Make sure all videos since how config management works in k8s are backed up This task involves ensuring that all videos related to config management in Kubernetes (k8s) are backed up. - title: Create List of YouTube Thumbnail Templates description: >- - Make list of YouTube thumbnail templates This task involves creating a list of YouTube thumbnail templates. - title: Create List of YouTube Idea Generation Limits description: >- - Make list of YouTube idea generation limits This task involves creating a list of YouTube idea generation limits. -------------------------------------------------------------------------------- Got 5 tasks from Task_Creator.

そうです。AI が人間に仕事を与えてくれる時代へようこそ！（一体どこで間違えたのでしょうか？🤷‍♂️）

結論

生成型 AI 駆動型アプリケーションの構築は困難です。しかし、適切なツールを使用すれば可能です。要約すると、次のようになります。

• AI エージェントは、複雑な問題を会話としてモデル化するという驚くべき機能を実現します。
• 正しく実行すれば、AI アプリケーションをより決定論的かつ信頼性の高いものにすることができます。
• AutoGenのようなツールは、AI エージェントを構築するためのシンプルな抽象化を備えたフレームワークを提供します。

次のステップとして、次のリソースを確認して、AI エージェントの世界をさらに深く理解することができます。

• 自分で試してみるための GitHub リポジトリ (コード)
• AutoGen を使用した会話パターンの実装 (ドキュメント)