はじめに
マルチエージェントAIシステムは、AI自動化の次なる進化を代表するものです。単一のAIがすべてを処理するのではなく、それぞれ異なる役割、目標、ツールを持つ専門エージェントが協力して、複雑なタスクをより効果的に解決します。
CrewAIは、このようなシステムの構築を簡単にするPythonフレームワークです。このチュートリアルでは、技術トピックを自動的に調査し、市場トレンドを分析し、プロフェッショナルなレポートを作成するリサーチクルーを構築します。
学習内容
- CrewAIプロジェクトをゼロから構築
- 専門的な役割とツールを持つエージェントの定義
- 依存関係と期待される出力を持つタスクの作成
- マルチエージェントワークフローのオーケストレーション
- Web検索とカスタムツールの統合
- Python 3.10+
- OpenAI APIキー
- Pythonの基本知識
前提条件
---
ステップ1: 環境セットアップ
新しいプロジェクトを作成し、依存関係をインストールします：
mkdir tech-research-crew && cd tech-research-crew
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
pip install crewai crewai-tools langchain-openai python-dotenv
APIキー用の.envファイルを作成します：
# .env
OPENAI_API_KEY=sk-your-api-key-here
SERPER_API_KEY=your-serper-key-here # Web検索用
---
ステップ2: エージェントの定義
agents.pyを作成します。各エージェントには役割、目標、バックストーリーがあります：
# agents.py
from crewai import Agent
from crewai_tools import SerperDevTool, WebsiteSearchTool
# ツール
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()
# エージェント1: リサーチスペシャリスト
researcher = Agent(
role="シニアテクノロジーリサーチャー",
goal="新興技術に関する包括的で正確な情報を見つける",
backstory="""あなたはトップコンサルティングファームで15年の経験を持つ
ベテラン技術リサーチャーです。信頼性の高い情報源を見つけ、
主要トレンドを特定し、誇大広告と実態を区別することに優れています。""",
tools=[search_tool, web_tool],
verbose=True,
allow_delegation=False,
max_iter=5,
)
# エージェント2: データアナリスト
analyst = Agent(
role="テクノロジーマーケットアナリスト",
goal="技術トレンドを分析し、データ駆動型のインサイトを提供する",
backstory="""あなたはテクノロジー市場分析を専門とするデータ駆動型の
アナリストです。生のリサーチを実用的なインサイトに変換し、
パターンを特定し、エビデンスに基づいた予測を行います。""",
verbose=True,
allow_delegation=False,
)
# エージェント3: テクニカルライター
writer = Agent(
role="シニアテクニカルライター",
goal="明確で魅力的、構造化された技術レポートを作成する",
backstory="""あなたは複雑なトピックをわかりやすくすることで知られる
受賞歴のあるテクニカルライターです。エグゼクティブとエンジニアの
両方が価値を見出すレポートを作成します。""",
verbose=True,
allow_delegation=False,
)
---
ステップ3: タスクの作成
tasks.pyを作成します。タスクは各エージェントが達成すべきことを定義します：
# tasks.py
from crewai import Task
from agents import researcher, analyst, writer
def create_research_tasks(topic: str):
"""指定トピックのリサーチタスクシーケンスを作成"""
# タスク1: リサーチ
research_task = Task(
description=f"""以下のトピックについて徹底的に調査: {topic}
調査内容:
1. 技術の現状
2. 主要プレイヤーと企業
3. 最近の動向（過去6ヶ月）
4. ユースケースと実世界での応用
5. 技術的課題と制限事項
すべてのソースのURLを提供してください。""",
expected_output="""詳細なリサーチドキュメント:
- エグゼクティブサマリー（2-3文）
- 主要な発見（5-7箇条書き）
- 各調査分野の詳細セクション
- 検証用ソースURL""",
agent=researcher,
)
# タスク2: 分析（リサーチに依存）
analysis_task = Task(
description=f"""リサーチ結果を分析: {topic}
分析内容:
1. SWOT分析
2. 市場導入タイムライン予測
3. 競合技術との比較
4. エンタープライズ導入のリスク評価
5. アーリーアダプターのROI可能性""",
expected_output="""構造化された分析ドキュメント:
- SWOTマトリックス
- 導入タイムライン（フェーズ付き）
- 競合比較表
- リスクマトリックス
- ROI予測""",
agent=analyst,
context=[research_task],
)
# タスク3: レポート作成（両方に依存）
writing_task = Task(
description=f"""プロフェッショナルな技術レポートを作成: {topic}
レポート要件:
1. エグゼクティブサマリーで開始
2. 論理的な流れで結果を提示
3. 分析フレームワークを含める
4. 実行可能な推奨事項を提供
5. 次のステップで締めくくる
形式: Markdown、明確な見出し、箇条書き、表を使用""",
expected_output="""完成されたMarkdownレポート（2000-3000語）:
- タイトルと日付
- エグゼクティブサマリー
- 調査結果
- 分析とインサイト
- 推奨事項（優先順位付き）
- ソース付録""",
agent=writer,
context=[research_task, analysis_task],
output_file="output/report.md",
)
return [research_task, analysis_task, writing_task]
---
ステップ4: クルーの組み立て
main.pyを作成します：
# main.py
import os
from dotenv import load_dotenv
from crewai import Crew, Process
from agents import researcher, analyst, writer
from tasks import create_research_tasks
load_dotenv()
os.makedirs("output", exist_ok=True)
def run_research_crew(topic: str):
tasks = create_research_tasks(topic)
crew = Crew(
agents=[researcher, analyst, writer],
tasks=tasks,
process=Process.sequential,
verbose=True,
memory=True,
max_rpm=10,
)
result = crew.kickoff()
print(f"\nレポート保存先: output/report.md")
print(f"使用トークン数: {crew.usage_metrics}")
return result
if __name__ == "__main__":
topic = input("リサーチトピックを入力: ") or "エンタープライズにおけるAIエージェント"
result = run_research_crew(topic)
---
ステップ5: カスタムツールの追加
tools.pyでカスタムツールを作成します：
# tools.py
from crewai_tools import tool
import json
from datetime import datetime
@tool("JSONレポート保存")
def save_json_report(content: str, filename: str) -> str:
"""構造化データをJSONレポートファイルとして保存します。
Args:
content: JSON文字列としてのレポート内容
filename: 出力ファイル名（拡張子なし）
"""
try:
data = json.loads(content)
except json.JSONDecodeError:
data = {"raw_content": content}
report = {
"generated_at": datetime.now().isoformat(),
"data": data,
}
filepath = f"output/{filename}.json"
with open(filepath, "w") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
return f"レポートを{filepath}に保存しました"
---
ステップ6: 階層型プロセス
複雑なワークフローでは、マネージャーエージェントが委任する階層型を使用します：
# hierarchical_crew.py
from crewai import Crew, Process
from langchain_openai import ChatOpenAI
manager_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
crew = Crew(
agents=[researcher, analyst, writer],
tasks=tasks,
process=Process.hierarchical,
manager_llm=manager_llm,
verbose=True,
)
result = crew.kickoff()
階層モードでは、CrewAIがマネージャーエージェントを自動作成し、タスクの割り当てと品質管理を行います。
---
ステップ7: 実行とテスト
python main.py
# 出力例:
# > Agent: シニアテクノロジーリサーチャー
# > Task: AIエージェントについて調査中...
# > ツール使用: インターネット検索
# > ...
# > Agent: テクノロジーマーケットアナリスト
# > Task: リサーチ結果を分析中...
# > ...
# > CREW EXECUTION COMPLETE
# > レポート保存先: output/report.md
---
まとめ
| 概念 | 説明 |
|------|------|
| Agent | 役割、目標、ツールを持つ自律AIエンティティ |
| Task | 期待される出力とコンテキストを持つ具体的な課題 |
| Crew | タスクに協力して取り組むエージェントのグループ |
| Process | 実行戦略（順次または階層型） |
| Tools | エージェントが外部と対話するための関数 |
| Memory | タスク実行間で維持される共有コンテキスト |
---
ベストプラクティス
1. 具体的な役割 — 広範ではなく、狭く明確に定義された役割をエージェントに与える
2. 明確な期待値 — expected_outputを正確に定義する
3. タスク依存関係 — contextを使用してタスクをチェーンする
4. レート制限 — max_rpmでAPIレート制限を回避
5. エラーハンドリング — max_iterで無限ループを防止
6. モデル選択 — 複雑な推論にはGPT-4o、シンプルなタスクにはGPT-3.5
---
参考リソース
Techsfree発行 — テクノロジーイノベーションのパートナー