Posted inAI

CrewAI完全マスター：調査・レポーティングを自動化するマルチエージェントワークフローの構築

AI tutorial - IT technology blog
AI tutorial - IT technology blog

アーキテクチャのアプローチ：連鎖プロンプト vs グラフエンジン vs 自律型エージェント

自動化されたリサーチパイプラインを設計する際、通常は「連鎖プロンプト（Chained Prompts）」「グラフベースの状態エンジン」「自律型マルチエージェントCrew」という3つのパラダイムのいずれかを選択することになります。

長年、開発チームはPythonスクリプトや標準的なLangChainのRunnableを介してOpenAI API呼び出しをパイプライン化してきました。1回目の呼び出しで生テキストをスクレイピングし、2回目で主要指標を抽出し、3回目でMarkdownの下書きを作成する、といった具合です。しかし、タスクにバックトラック（後戻り）や動的な情報源の検証、専門ドメインごとの分担が必要になると、この脆弱な組み立てラインは破綻してしまいます。

  • 連鎖プロンプトパイプライン：実装がシンプルで、各ステップが次のステップへ直接データを渡します。しかし、ステップ1でレート制限エラーページや切り詰められたJSONを取り込んでしまうと、後続のモデルすべてがハルシネーションを起こし、リカバリーの余地がありません。
  • グラフベースのワークフロー（LangGraph）：極めて決定論的です。状態遷移と循環グラフにより、エッジのルーティングを厳密に制御できます。ただし、さほど複雑でないタスクであっても、定型的な状態スキーマ、ノードハンドラー、手動のリデューサーの作成に何時間も費やすことになります。
  • ロールプレイング型マルチエージェントシステム（CrewAI）：実際のエジニアリングチームを模倣します。各エージェントにペルソナ、固有の目標、ローカルメモリ、特化ツールが割り当てられます。エージェント同士がサブタスクを委任し、下書きをレビューし、人手を介さずにフィードバックをループ処理します。

本番環境におけるCrewAIの真のトレードオフ

社内プラットフォームでCrewAIを標準採用する前に、日々の運用の現実とシステムの信頼性SLAを慎重に比較検討する必要があります。

CrewAIの強み

  • 明確な役割の境界：データ収集担当とテクニカルライターを分けることで、プロンプトをシンプルに維持できます。システムコンテキストが散らからず、プロンプトターンあたりのトークン消費量を30%〜50%削減できます。
  • すぐに使えるツールガバナンス：リトライ処理、入力バリデーション、実行キャッシュ、エージェント間のハンドオフが標準で組み込まれています。
  • 最小限のセットアップ：カスタムツールを備えた完全に機能する2エージェントチームを、50行未満の簡潔なPythonコードで立ち上げられます。

運用上の落とし穴

  • 急増するトークンオーバーヘッドとレイテンシ：自律性には大きな代償が伴います。マルチエージェント間の議論や検索の検証により、単一の1,500トークンのプロンプトが8〜15回のAPIラウンドトリップに膨れ上がります。その結果、1回の実行あたりの応答時間が4秒から45秒近くまで増大します。
  • 制御不能な実行ループ：曖昧なプロンプトを与えると、エージェントが高コストなクエリーループに陥る危険があります。厳格なガードがなければ、エージェントは何十回も検索に失敗し続け、タイムアウトに達するまでに5〜10ドルのAPIクレジットを消費してしまいます。

アーキテクチャと前提条件

エージェントのプロトタイプを信頼性の高い本番ジョブへと昇華させるには、厳格なトークン境界、明示的なモデルルーティング、ハードな反復回数制限（iteration cap）が不可欠です。

環境構築

専用の仮想環境を作成し、必要な依存パッケージをインストールします：

python3 -m venv venv
source venv/bin/activate
pip install crewai==0.80.0 crewai-tools langchain-openai

APIキーを設定します。大量のWebクエリにはgpt-4o-miniを使用し、編集・要約にはフルスペックのgpt-4oを確保します。この階層型ルーティングにより、推論コストを約70%削減できます：

export OPENAI_API_KEY="sk-proj-your-openai-key"
export SERPER_API_KEY="your-serper-dev-api-key"

ステップ別実装：DevOpsリサーチCrewの構築

以下は実際に動作する2エージェントパイプラインの例です。CVE開示情報やクラスターベンチマークを収集するインフラアナリストと、それらの調査結果をエグゼクティブ向けブリーフィングにまとめるテクニカルエディターを組み合わせています。

1. ツールとモデルルーティングの設定

まず、リアルタイムのWeb検索用にSerperDevToolを初期化し、コストと速度のバランスを考慮してモデルをマッピングします：

import os
from crewai import Agent, Crew, Process, Task
from crewai_tools import SerperDevTool
from langchain_openai import ChatOpenAI

# Web検索ツール
search_tool = SerperDevTool()

# コスト効率化のための階層型LLMルーティング
fast_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
reasoning_llm = ChatOpenAI(model="gpt-4o", temperature=0.4)

2. 特化型エージェントの構築

backstoryに明確な指示を与えて各エージェントの基盤を固めます。無限ループを防ぐために必ずmax_iterを設定し、開発中はverbose=Trueにして生（raw）のツール呼び出しを確認できるようにします。

# データ収集エージェント
researcher = Agent(
    role="シニアインフラ＆クラウドセキュリティアナリスト",
    goal="{topic} に関する最新のセキュリティCVE、リソースオーバーヘッドの傾向、およびデプロイメントのベンチマークを特定する",
    backstory="""あなたは経験豊富なSite Reliability Engineerです。
    マーケティング的な誇張を排し、確かなテレメトリ、公式CVEデータベース、
    査読済みの本番環境ベンチマークに専念します。""",
    tools=[search_tool],
    llm=fast_llm,
    verbose=True,
    memory=True,
    max_iter=4,
    allow_delegation=False
)

# 編集・要約エージェント
writer = Agent(
    role="プリンシパルテクニカルライター",
    goal="インフラのテレメトリとCVEの調査結果を統合し、{topic} に関するエグゼクティブ向けブリーフィングを作成する",
    backstory="""あなたは複雑な分散システムのデータを簡潔明瞭なエグゼクティブサマリーに変換します。
    具体的なトレードオフ、影響指標、実践的なエンジニアリング修正策を重視します。""",
    tools=[],
    llm=reasoning_llm,
    verbose=True,
    memory=True,
    max_iter=3,
    allow_delegation=False
)

3. 連携タスクの定義

descriptionに具体的なステップを記述し、フォーマットの乱れ（ドリフト）を防ぐためにexpected_outputで厳格な出力契約を定義します：

# タスク1: 調査と検証
research_task = Task(
    description="""
    1. {topic} に関連する実際のデプロイにおける課題と最新のCVE開示情報を調査する。
    2. レイテンシおよびメモリのオーバーヘッドに関する具体的なベンチマークを少なくとも3つ特定する。
    3. すべての主張を公式プロジェクトリポジトリおよびベンダーのアドバイザリと照合してクロスリファレンスする。
    """,
    expected_output="検証済みメトリクス、CVE識別子、アーキテクチャパターンを含む箇条書きの概要。",
    agent=researcher
)

# タスク2: エグゼクティブ向けMarkdownレポート作成
reporting_task = Task(
    description="""
    リサーチアナリストのメモをもとに、明瞭なMarkdownブリーフィングを作成してください。
    ブリーフは以下のセクションで正確に構成してください：
    - エグゼクティブサマリー
    - 検証済みレイテンシ＆メモリベンチマーク
    - 重要なセキュリティベクター
    - DevOps修正チェックリスト
    トーンは簡潔かつ事実に基づき、マーケティング的な誇張は排除してください。
    """,
    expected_output="明確な見出し、メトリクステーブル、実用的なアクション手順でフォーマットされた、公開可能なMarkdownファイル。",
    agent=writer,
    output_file="devops_intelligence_report.md"
)

4. Crewの統合と実行

エージェントとタスクをシーケンシャル（順次実行）なCrewにまとめ、調査対象のトピックを指定してパイプラインを実行します：

# エージェントとタスクを統合
devops_intel_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, reporting_task],
    process=Process.sequential,
    verbose=True
)

# 動的入力で実行
if __name__ == "__main__":
    topic_input = "2026年におけるKubernetesマルチクラスターサービスメッシュのオーバーヘッド"
    print(f"[*] {topic_input} のCrew実行を開始中...")
    
    result = devops_intel_crew.kickoff(inputs={"topic": topic_input})
    
    print("\n[+] ワークフローの実行が完了しました！")
    print("[+] レポートが保存されました: devops_intelligence_report.md")

本番環境向けのガードレール

自動化されたCI/CDランナーやヘッドレスなcronジョブ内でマルチエージェントシステムを実行する場合、プロアクティブなガードレールが必須です：

  1. max_iterの厳格な上限設定：反復回数の制限は絶対に省略しないでください。max_iter=3またはmax_iter=4を設定することで、エージェントがスタックしてもクレジット残高を消費し尽くすことなく安全に失敗（Fail Gracefully）させることができます。
  2. output_fileの直接指定：CIのログコレクターではターミナル出力が途中で切り捨てられることがあります。最終成果物をディスクに保存することで、後続のジョブが検証可能なイミュータブル（不変）なファイルを得られます。
  3. モデルタイムアウトの設定：ネットワーク呼び出しはハングすることがあります。ChatOpenAIクライアントインスタンスに明示的にrequest_timeout=60を渡し、応答のないリクエストを迅速に失敗（Fail Fast）させてランナーキュー全体の停止を防ぎます。
Share:
Tags: