Zenn
🏭

OpenAI Agents SDKを活用したマルチAIエージェントの協調によるセンサーデータ分析システム

に公開
OpenAI
Claude
AIエージェント
agentsdk
tech

はじめに

OpenAI Agents SDKを活用した、複数の専門エージェントが連携するシステムの実装方法について解説します。各エージェントは、センサーデータの監視、AIによる詳細分析、アラート通知、運用改善の助言、質問応答、Embedding生成などの役割を持ち、OrchestrationAgentを介して全体の処理を統括しています。さらに、AgentSDKの柔軟な定義や簡単な統合、同期／非同期実行、効率的なマルチエージェントオーケストレーションなど、基本的な機能についても紹介します。

実際に作成したコンセプトモックの動画は以下から確認できます。

https://youtu.be/YUCq6R0qwk4

1. AgentSDKの機能概要と使い方

AgentSDKは、単一エージェントおよびマルチエージェントのワークフローをオーケストレーションするためのオープンソースライブラリです。従来の複雑な非同期処理やプロンプトエンジニアリングの実装負荷を大幅に軽減し、シンプルなコードで高度なタスク処理を実現します。
https://openai.github.io/openai-agents-python/

1-1. エージェントの作成と設定

機能:
各エージェントは、Agentオブジェクトとして生成され、名前、役割、指示（instructions）および利用可能な機能（functions）を設定します。さらに、Python の関数をツールとしてデコレータで登録することで、既存の機能や外部APIとの連携がスムーズに行える点が大きなメリットです。

使い方の例:

from agents import Agent, Runner

# AnalysisAgentの初期化例
self.agent = Agent(
    name="SensorDataAnalyzer",
    instructions="""
    あなたはセンサーデータの解析の専門家です。グレーゾーンのデータに対して、なぜその値が正常または異常なのかを詳細に分析してください。
    回答は「normal」または「abnormal」から始め、理由を付記してください。
    """
)

ポイント:

nameでエージェントを識別
instructionsでエージェントの専門知識や回答ルールを明示

1-2. エージェントの実行と同期／非同期処理

AgentSDK は、Runner クラスを提供しており、同期的にエージェントを実行するための Runner.run_sync() や、非同期処理を利用するための Runner.run()、さらにはストリーミングで応答を受け取る Runner.run_streamed() といったメソッドが用意されており、ユースケースに応じた実行方法を選択できます。

import asyncio
from agents import Agent, Runner

async def main():
    # センサーデータ分析エージェントの作成
    agent = Agent(
        name="SensorAnalyzer",
        instructions="センサーの測定値が正常範囲内か異常かを判断してください。"
    )
    
    # 分析するセンサーデータ
    sensor_data = "センサーID: S001, 測定値: 78.3, 正常範囲: 20-50"
    
    # 非同期で実行
    result = await Runner.run(agent, sensor_data)
    
    # 結果を表示
    print(result.final_output)

# 非同期関数の実行
if __name__ == "__main__":
    asyncio.run(main())

ポイント:

同期実行によりシンプルなコードで AI との対話が可能
非同期実行にも対応しているため、バックグラウンド処理や高速レスポンスが必要なシステムにも適用可能

1-3. エージェント間の連携とタスク転送（ハンドオフ）

AgentSDK では、複数のエージェントを組み合わせたワークフローの構築が容易です。
特定の条件下では、あるエージェントがタスクを他のエージェントへ ハンドオフ することで、より専門的な回答や処理を委譲できます。
さらに、LLMがコンテキストを判断しながら適切なエージェントに自律的にタスクを委譲できるため、柔軟性が高まり効果的に対応できるようになります

from agents import Agent, Runner, handoff, RunContextWrappe
# 分析エージェントの定義
analysis_agent = Agent(
    name="AnalysisAgent",
    instructions="センサーデータを詳細に分析し、異常の有無と原因を特定してください。"
)

# ハンドオフコールバック関数
def on_analysis_handoff(ctx: RunContextWrapper[None]):
    # ハンドオフ時に渡されたパラメータを取得
    sensor_data = ctx.parameters.get("sensor_data", "")
    print(f"分析エージェントにハンドオフします: {sensor_data}")
    return f"センサーデータ「{sensor_data}」の分析を開始します..."

# ハンドオフを定義
analysis_handoff = handoff(
    agent=analysis_agent,
    on_handoff=on_analysis_handoff,
    tool_name_override="handoff_to_analysis",  # ツール名のカスタマイズ
    tool_description_override="センサーデータの分析が必要な場合に分析エージェントにタスクを転送します。"
)

# メインエージェントの定義
dispatcher = Agent(
    name="DataDispatcher",
    instructions="""
    あなたはデータ転送を行うエージェントです。入力されたタスクが分析を要する場合は、
    analysis_handoff ツールを使用して AnalysisAgent に転送してください。
    """,
    tools=[analysis_handoff]  # ハンドオフをツールとして追加
)

ポイント:

ハンドオフ機能を活用することで、エージェント間のタスク振り分けが自律的に行われます。
LLM がコンテキストを判断しながら適切なエージェントにタスクを委譲するため、柔軟性が高まり効果的に対応可能になります。

1-4. FunctionTools

AgentSDK の function_toolは、AI エージェントに外部機能やデータアクセスの能力を付与するための強力な仕組みです。これにより、AI は単なる会話だけでなく、実際のシステムと対話できるようになります。

from agents import Agent, function_tool

@function_tool
def get_sensor_data(sensor_id: str):
    """センサーの最新データを取得します。

    Args:
        sensor_id: データを取得するセンサーのID
        
    Returns:
        センサーデータの文字列表現
    """
    # 実際のデータ取得ロジック
    return f"センサー{sensor_id}の測定値: 42.5°C, 状態: 正常"

# エージェントにツールを登録
agent = Agent(
    name="MonitoringAssistant",
    instructions="ユーザーの質問に応じてセンサーデータを提供してください",
    tools=[get_sensor_data]
)

1-5. ガードレールとトレーシング

ガードレール:

入力・出力の検証機能が内蔵されており、不正な入力や予期しない出力を防止する仕組みが実装されています。これにより、プロンプトインジェクションなどの攻撃リスクが大幅に軽減されます。

トレーシング:

各エージェントの動作が詳細にトレースされ、ログとして記録されます。これにより、問題の早期発見やデバッグが容易になり、システムの信頼性と安全性が向上します。

2. OpenAI Agents SDKを活用したマルチAIエージェントの協調によるセンサーデータ分析システムの概要

2-1. システムの動作フロー

1.データ読み込み:

設定された時間にCSVファイルからセンサーデータを読み込み、各レコードに対して初期評価（正常、異常、グレーゾーン）をルールベースで実施

2.Embedding生成:

グレーゾーンや異常値の場合、EmbeddingAgentが、センサーデータからベクトルデータを生成。グレーゾーンの場合は、結果にカテゴリーを付与せずにデータをAnalysisAgentに受け渡す。

3.AI分析:

グレーゾーンや異常値の場合、AnalysisAgentがAI分析を実施。EmbeddingAgentからベクトルデータを受け取り、類似度を計算する。プロンプトに基づいて「normal」または「abnormal」の判定と分析理由を出力する。また、ベクトルデータをカテゴライズする

4.アクション:

abnormalと判定された場合、ActionAgentが状況を整理した内容を含む異常検知のメール通知を行う

全体の統括:

ルールベースで動くOrchestrationAgentが全体の処理結果とログを履歴として管理

ユーザとの対話:

QuestionAnswerAgentがOrchestrationAgentからシステム全体の状態や履歴情報を取得して、ユーザーからの問い合わせに対して適切な回答を生成する
質問内容に応じて、AnalysisAgent、ActionAgent、AdviceAgentに自動でハンドオフして、専門的な回答を取得することが可能

2-2. 各エージェントの役割

AnalysisAgent:

グレーゾーンデータに対して、agentSDK経由でAI分析を実行し、正常／異常を判断

ActionAgent:

異常検知時にメール通知を送信する

AdviceAgent:

センサーデータに基づいた運用改善の助言を生成

QuestionAnswerAgent:

システムの状態や履歴情報を元に、ユーザーからの質問に回答

EmbeddingAgent:

センサーデータからEmbedding（数値ベクトル）を生成し、類似度計算でカテゴリ判定を支援

OrchestrationAgent:

各エージェントの処理結果を統括し、全体の処理フロー（データの読み込み→評価→AI分析→通知→履歴管理）を実現
ルールベースのAgentであり、生成AIの機能はなし

2-3. 連携のメリット

責務の分離:

各エージェントが専門分野に特化しているため、コードの可読性と保守性が向上

スケーラビリティの向上:

新たなエージェントや機能の追加が容易で、既存の処理フローにシームレスに統合可能
将来的なシステムの拡大に柔軟に対応出来る

柔軟なワークフローの構築:

複数のエージェントを連携させることで、システム全体の処理フローを柔軟に設計できます。各エージェントは自律的にタスクを受け取り、LLM がコンテキストを判断しながら適切なエージェントに自律的にタスクを委譲できるため、より効果的に処理が進みます
また、ルールベースのフローを組み合わせることで、ハイブリッドな意思決定により、決定の一貫性と柔軟性の両方が実現されます

堅牢なエラーハンドリング:

各エージェントが個別にエラーを検出・ログ出力し、OrchestrationAgentが統合的に管理するため、システム全体の信頼性が向上

3. まとめと今後の展望

本システムは、AgentSDKの活用により、複雑なセンサーデータの解析や通知処理をシンプルなコードで実現できる点が大きな特徴です。エージェント間の柔軟な連携と効率的なワークフローの構築により、各専門分野の知識を最大限に活かした高度なAI分析が可能となります。さらに、今回は実装していませんがガードレールやトレーシングを実装することでシステム全体の信頼性をさらに高めることが可能になります。

Discussion