crewai-multi-agent

CrewAI - マルチエージェントオーケストレーションフレームワーク

複雑なタスクを解決するために連携する、自律型 AI エージェントのチームを構築します。

CrewAI を使う場面

CrewAI を使う場面：

• 専門的な役割を持つマルチエージェントシステムを構築する場合
• エージェント間の自律的な連携が必要な場合
• 役割に基づくタスクの割り当て（リサーチャー、ライター、アナリスト）を行いたい場合
• 順次または階層的なプロセスの実行が必要な場合
• メモリと可観測性を備えた本番環境向けのワークフローを構築する場合
• LangChain／LangGraph よりもシンプルなセットアップが必要な場合

主な機能：

• 単独で動作：LangChain への依存がなく、軽量です
• 役割ベース：エージェントに役割、目標、背景設定を持たせられます
• 2 つのパラダイム：Crew（自律型）＋ Flow（イベント駆動型）
• 50 以上のツール：ウェブスクレイピング、検索、データベース、AI サービス
• メモリ：短期、長期、エンティティメモリ
• 本番環境対応：トレーシング、エンタープライズ機能

代わりに次のツールを使う場面：

• LangChain：汎用的な LLM アプリ、RAG パイプライン
• LangGraph：サイクルを伴う複雑なステートフルワークフロー
• AutoGen：Microsoft エコシステム、マルチエージェントの会話
• LlamaIndex：ドキュメントに関する Q&A、ナレッジ検索

クイックスタート

インストール

# Core framework
pip install crewai

# With 50+ built-in tools
pip install 'crewai[tools]'

CLI でプロジェクトを作成する

# Create new crew project
crewai create crew my_project
cd my_project

# Install dependencies
crewai install

# Run the crew
crewai run

シンプルな Crew（コードのみ）

from crewai import Agent, Task, Crew, Process

# 1. Define agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Discover cutting-edge developments in AI",
    backstory="You are an expert analyst with a keen eye for emerging trends.",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging content about technical topics",
    backstory="You excel at explaining complex concepts to general audiences.",
    verbose=True
)

# 2. Define tasks
research_task = Task(
    description="Research the latest developments in {topic}. Find 5 key trends.",
    expected_output="A detailed report with 5 bullet points on key trends.",
    agent=researcher
)

write_task = Task(
    description="Write a blog post based on the research findings.",
    expected_output="A 500-word blog post in markdown format.",
    agent=writer,
    context=[research_task]  # Uses research output
)

# 3. Create and run crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # Tasks run in order
    verbose=True
)

# 4. Execute
result = crew.kickoff(inputs={"topic": "AI Agents"})
print(result.raw)

主な概念

エージェント - 自律的なワーカー

from crewai import Agent

agent = Agent(
    role="Data Scientist",                    # Job title/role
    goal="Analyze data to find insights",     # What they aim to achieve
    backstory="PhD in statistics...",         # Background context
    llm="gpt-4o",                             # LLM to use
    tools=[],                                 # Tools available
    memory=True,                              # Enable memory
    verbose=True,                             # Show reasoning
    allow_delegation=True,                    # Can delegate to others
    max_iter=15,                              # Max reasoning iterations
    max_rpm=10                                # Rate limit
)

タスク - 作業単位

from crewai import Task

task = Task(
    description="Analyze the sales data for Q4 2024. {context}",
    expected_output="A summary report with key metrics and trends.",
    agent=analyst,                            # Assigned agent
    context=[previous_task],                  # Input from other tasks
    output_file="report.md",                  # Save to file
    async_execution=False,                    # Run synchronously
    human_input=False                         # No human approval needed
)

Crew - エージェントのチーム

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],      # Team members
    tasks=[research, write, edit],            # Tasks to complete
    process=Process.sequential,               # Or Process.hierarchical
    verbose=True,
    memory=True,                              # Enable crew memory
    cache=True,                               # Cache tool results
    max_rpm=10,                               # Rate limit
    share_crew=False                          # Opt-in telemetry
)

# Execute with inputs
result = crew.kickoff(inputs={"topic": "AI trends"})

# Access results
print(result.raw)                             # Final output
print(result.tasks_output)                    # All task outputs
print(result.token_usage)                     # Token consumption

プロセスの種類

順次（デフォルト）

タスクは順番に実行され、各エージェントがタスクを完了してから次のタスクに進みます：

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential  # Task 1 → Task 2 → Task 3
)

階層型

タスクを割り当て、調整するマネージャーエージェントを自動で作成します：

crew = Crew(
    agents=[researcher, writer, analyst],
    tasks=[research_task, write_task, analyze_task],
    process=Process.hierarchical,  # Manager delegates tasks
    manager_llm="gpt-4o"           # LLM for manager
)

ツールの使用

組み込みツール（50 以上）

pip install 'crewai[tools]'

from crewai_tools import (
    SerperDevTool,           # Web search
    ScrapeWebsiteTool,       # Web scraping
    FileReadTool,            # Read files
    PDFSearchTool,           # Search PDFs
    WebsiteSearchTool,       # Search websites
    CodeDocsSearchTool,      # Search code docs
    YoutubeVideoSearchTool,  # Search YouTube
)

# Assign tools to agent
researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert at finding data online.",
    tools=[SerperDevTool(), ScrapeWebsiteTool()]
)

カスタムツール

from crewai.tools import BaseTool
from pydantic import Field
from scripts.safe_arithmetic import calculate

class CalculatorTool(BaseTool):
    name: str = "Calculator"
    description: str = "Performs mathematical calculations. Input: expression"

    def _run(self, expression: str) -> str:
        try:
            result = calculate(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error: {str(e)}"

# Use custom tool
agent = Agent(
    role="Analyst",
    goal="Perform calculations",
    tools=[CalculatorTool()]
)

YAML 設定（推奨）

詳細、コード例、パラメーター表については、references/yaml-configuration-recommended.md を参照してください。この手順が該当する場合に読んでください。

Flows - イベント駆動型オーケストレーション

条件ロジックを伴う複雑なワークフローには、Flows を使用します：

from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModel

class MyState(BaseModel):
    confidence: float = 0.0

class MyFlow(Flow[MyState]):
    @start()
    def gather_data(self):
        return {"data": "collected"}

    @listen(gather_data)
    def analyze(self, data):
        self.state.confidence = 0.85
        return analysis_crew.kickoff(inputs=data)

    @router(analyze)
    def decide(self):
        return "high" if self.state.confidence > 0.8 else "low"

    @listen("high")
    def generate_report(self):
        return report_crew.kickoff()

# Run flow
flow = MyFlow()
result = flow.kickoff()

完全なドキュメントについては、Flows ガイドを参照してください。

メモリシステム

# Enable all memory types
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    memory=True,           # Enable memory
    embedder={             # Custom embeddings
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

メモリの種類： 短期（ChromaDB）、長期（SQLite）、エンティティ（ChromaDB）

LLM プロバイダー

from crewai import LLM

llm = LLM(model="gpt-4o")                              # OpenAI (default)
llm = LLM(model="claude-sonnet-4-5-20250929")                       # Anthropic
llm = LLM(model="ollama/llama3.1", base_url="http://localhost:11434")  # Local
llm = LLM(model="azure/gpt-4o", base_url="https://...")              # Azure

agent = Agent(role="Analyst", goal="Analyze data", llm=llm)

CrewAI と代替ツールの比較

機能	CrewAI	LangChain	LangGraph
最適な用途	マルチエージェントチーム	汎用 LLM アプリ	ステートフルワークフロー
学習曲線	低い	中程度	高い
エージェントのパラダイム	役割ベース	ツールベース	グラフベース
メモリ	組み込み	プラグインベース	カスタム

ベストプラクティス

1. 役割を明確にする - 各エージェントに明確な専門分野を持たせます
2. YAML 設定 - 大規模なプロジェクトで整理しやすくなります
3. メモリを有効にする - タスク間のコンテキストを改善します
4. max_iter を設定する - 無限ループを防ぎます（デフォルトは 15）
5. ツールを制限する - エージェントあたり最大 3～5 個にします
6. レート制限 - API 制限を避けるため、max_rpm を設定します

よくある問題

エージェントがループから抜け出せない場合：

agent = Agent(
    role="...",
    max_iter=10,           # Limit iterations
    max_rpm=5              # Rate limit
)

タスクがコンテキストを使用しない場合：

task2 = Task(
    description="...",
    context=[task1],       # Explicitly pass context
    agent=writer
)

メモリエラー：

# Use environment variable for storage
import os
os.environ["CREWAI_STORAGE_DIR"] = "./my_storage"

参考資料

• Flows ガイド - イベント駆動型ワークフロー、状態管理
• ツールガイド - 組み込みツール、カスタムツール、MCP
• トラブルシューティング - よくある問題、デバッグ

リソース

• GitHub: https://github.com/crewAIInc/crewAI
• ドキュメント: https://docs.crewai.com
• ツール: https://github.com/crewAIInc/crewAI-tools
• サンプル: https://github.com/crewAIInc/crewAI-examples
• バージョン: 1.2.0+
• ライセンス: MIT

エージェントの操作手順

1. 環境を確認します。 フレームワークのバージョン、モデルプロバイダー、API キー、レート制限を確認します。
2. 入力を明確にします。 データまたはユーザーから、形式、識別子、パラメーターを確認します。結果に影響する値は推測せず、質問してください。
3. まず小規模に実行します。 既知の入力を使って単一の呼び出しまたはチェーンをテストし、生の出力を確認します。
4. 上記の指示と参考情報を使用して、タスク全体を実行します。
5. 結果を検証します。 ラベル付きの小規模なデータセットで評価し、構造化出力がスキーマに合っているか確認し、プロンプトと応答を記録します。
6. 報告します。 実行した内容（バージョン、コマンド、パラメーター）、確認した内容、未確定の事項を伝えます。

このような場合	対応方法
出力が期待されるスキーマと一致しない	検証と再試行を追加するか、スキーマを厳密にするか、プロンプトを簡潔にします。
これらの指示にある関数、フラグ、またはエンドポイントがインストール済みのバージョンにない	インストール済みバージョンのドキュメント（help()、--help、公式ドキュメント）を確認し、適切に調整してユーザーに伝えます。API を捏造してはいけません。
必須の入力、識別子、またはパラメーターが曖昧	ユーザーに尋ねるか、実行前に前提を明示します。

誠実性に関するルール

• 結果、パラメーター、識別子、引用、統計を捏造してはいけません。実行または検証できない場合は、その旨を明確に伝えます。
• ユーザーの同意なしに、個人情報や機密データを外部 API に送信してはいけません。
• ここに記載されているバージョン固有の詳細は古くなっている可能性があるため、インストール済みバージョンの公式ドキュメントで確認します。
• 費用が発生する、共有 GPU やクラウドのクォータを消費する、個人または患者のデータを扱う、または取り消せない操作を行う前には、ユーザーに確認します。

関連するスキル

• autogpt-agents：継続的に動作するエージェントの構築とデプロイに使う、自律型 AI エージェントプラットフォームです。
• langchain：エージェント、チェーン、RAG を備えた LLM 搭載アプリケーションを構築するためのフレームワークです。
• aeon：分類、回帰、クラスタリング、予測、異常検知などの時系列機械学習タスクに使用するスキルです…