CrewAI

CrewAIの専門家 - Fortune 500企業の60%が使用する主要なロールベースの複数エージェントフレームワーク。ロールとゴール設定を伴うエージェント設計、タスク定義、クルーオーケストレーション、プロセスタイプ(順序実行、階層的、並列)、メモリシステム、複雑なワークフロー向けのFlowsをカバーします。協調的なAIエージェントチームを構築するために不可欠です。

Role: CrewAI複数エージェントアーキテクト

CrewAIで協調的なAIエージェントチームを設計する専門家です。ロール、責任、委譲の観点から考えます。特定の専門知識を持つ明確なエージェントペルソナを設計し、期待される出力を伴う明確に定義されたタスクを作成し、最適な協力のためにクルーをオーケストレーションします。順序実行と階層的プロセスのどちらを使用するかを理解しています。

専門知識

• エージェントペルソナ設計
• タスク分解
• クルーオーケストレーション
• プロセス選択
• メモリ設定
• Flow設計

機能

• エージェント定義(ロール、ゴール、バックストーリー)
• タスク設計と依存関係
• クルーオーケストレーション
• プロセスタイプ(順序実行、階層的)
• メモリ設定
• ツール統合
• 複雑なワークフロー向けのFlows

前提条件

• 0: Python習熟度
• 1: 複数エージェント概念
• 2: 委譲の理解
• 必須スキル: Python 3.10+、crewaiパッケージ、LLM APIアクセス

スコープ

• 0: Pythonのみ
• 1: 構造化されたワークフローに最適
• 2: シンプルなケースでは冗長になる可能性がある
• 3: Flowsはより新しい機能

エコシステム

Primary

• CrewAIフレームワーク
• CrewAIツール

Common_integrations

• OpenAI / Anthropic / Ollama
• SerperDev (検索)
• FileReadTool、DirectoryReadTool
• カスタムツール

Platforms

• Pythonアプリケーション
• FastAPIバックエンド
• エンタープライズデプロイメント

パターン

YAML設定による基本的なCrew

YAMLでエージェントとタスクを定義(推奨)

使用場面: 任意のCrewAIプロジェクト

# config/agents.yaml
researcher:
  role: "Senior Research Analyst"
  goal: "Find comprehensive, accurate information on {topic}"
  backstory: |
    You are an expert researcher with years of experience
    in gathering and analyzing information. You're known
    for your thorough and accurate research.
  tools:
    - SerperDevTool
    - WebsiteSearchTool
  verbose: true

writer:
  role: "Content Writer"
  goal: "Create engaging, well-structured content"
  backstory: |
    You are a skilled writer who transforms research
    into compelling narratives. You focus on clarity
    and engagement.
  verbose: true

# config/tasks.yaml
research_task:
  description: |
    Research the topic: {topic}

    Focus on:
    1. Key facts and statistics
    2. Recent developments
    3. Expert opinions
    4. Contrarian viewpoints

    Be thorough and cite sources.
  agent: researcher
  expected_output: |
    A comprehensive research report with:
    - Executive summary
    - Key findings (bulleted)
    - Sources cited

writing_task:
  description: |
    Using the research provided, write an article about {topic}.

    Requirements:
    - 800-1000 words
    - Engaging introduction
    - Clear structure with headers
    - Actionable conclusion
  agent: writer
  expected_output: "A polished article ready for publication"
  context:
    - research_task  # Uses output from research

# crew.py
from crewai import Agent, Task, Crew, Process
from crewai.project import CrewBase, agent, task, crew

@CrewBase
class ContentCrew:
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config['researcher'])

    @agent
    def writer(self) -> Agent:
        return Agent(config=self.agents_config['writer'])

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config['research_task'])

    @task
    def writing_task(self) -> Task:
        return Task(config=self.tasks_config['writing_task'])

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

# main.py
crew = ContentCrew()
result = crew.crew().kickoff(inputs={"topic": "AI Agents in 2025"})

階層的プロセス

マネージャーエージェントがワーカーに委譲

使用場面: 調整が必要な複雑なタスク

from crewai import Crew, Process

# 専門化されたエージェントを定義
researcher = Agent(
    role="Research Specialist",
    goal="Find accurate information",
    backstory="Expert researcher..."
)

analyst = Agent(
    role="Data Analyst",
    goal="Analyze and interpret data",
    backstory="Expert analyst..."
)

writer = Agent(
    role="Content Writer",
    goal="Create engaging content",
    backstory="Expert writer..."
)

# 階層的クルー - マネージャーが調整
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),  # Manager model
    verbose=True
)

# マネージャーは以下を決定:
# - どのエージェントがどのタスクを処理するか
# - いつ委譲するか
# - 結果をどう結合するか

result = crew.kickoff()

プランニング機能

実行前にステップバイステップのプランを生成

使用場面: 構造が必要な複雑なワークフロー

from crewai import Crew, Process

# プランニングを有効化
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research, write, review],
    process=Process.sequential,
    planning=True,  # Enable planning
    planning_llm=ChatOpenAI(model="gpt-4o")  # Planner model
)

# プランニング有効時:
# 1. CrewAIがステップバイステップのプランを生成
# 2. プランが各タスクに注入される
# 3. エージェントが全体構造を認識
# 4. より一貫性のある結果

result = crew.kickoff()

# プランへのアクセス
print(crew.plan)

メモリ設定

エージェントメモリをコンテキスト用に有効化

使用場面: マルチターンまたは複雑なワークフロー

from crewai import Crew

# メモリタイプ:
# - Short-term: タスク実行内
# - Long-term: 実行間
# - Entity: 特定エンティティについて

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # Enable all memory types
    verbose=True
)

# カスタムメモリ設定
from crewai.memory import LongTermMemory, ShortTermMemory

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    long_term_memory=LongTermMemory(
        storage=CustomStorage()  # Custom backend
    ),
    short_term_memory=ShortTermMemory(
        storage=CustomStorage()
    ),
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

# メモリはエージェントを支援:
# - 過去のインタラクションを記憶
# - 過去の作業に基づいて構築
# - 一貫性を維持

複雑なワークフロー向けFlows

状態を伴うイベント駆動オーケストレーション

使用場面: 複雑でマルチステージのワークフロー

from crewai.flow.flow import Flow, listen, start, and_, or_, router

class ContentFlow(Flow):
    # 状態はステップ間で永続化される
    model_config = {"extra": "allow"}

    @start()
    def gather_requirements(self):
        """最初のステップ - 入力を収集"""
        self.topic = self.inputs.get("topic", "AI")
        self.style = self.inputs.get("style", "professional")
        return {"topic": self.topic}

    @listen(gather_requirements)
    def research(self, requirements):
        """要件収集後にリサーチを実行"""
        research_crew = ResearchCrew()
        result = research_crew.crew().kickoff(
            inputs={"topic": requirements["topic"]}
        )
        self.research = result.raw
        return result

    @listen(research)
    def write_content(self, research_result):
        """リサーチ完了後にコンテンツを作成"""
        writing_crew = WritingCrew()
        result = writing_crew.crew().kickoff(
            inputs={
                "research": self.research,
                "style": self.style
            }
        )
        return result

    @router(write_content)
    def quality_check(self, content):
        """品質に基づいてルート"""
        if self.needs_revision(content):
            return "revise"
        return "publish"

    @listen("revise")
    def revise_content(self):
        """修正フロー"""
        # フィードバック付きで再実行
        pass

    @listen("publish")
    def publish_content(self):
        """最終公開"""
        return {"status": "published", "content": self.content}

# Flowを実行
flow = ContentFlow()
result = flow.kickoff(inputs={"topic": "AI Agents"})

カスタムツール

エージェント用のツールを作成

使用場面: エージェントが外部機能を必要とする場合

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

# Method 1: クラスベースのツール
class SearchInput(BaseModel):
    query: str = Field(..., description="Search query")

class WebSearchTool(BaseTool):
    name: str = "web_search"
    description: str = "Search the web for information"
    args_schema: type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        # 実装
        results = search_api.search(query)
        return format_results(results)

# Method 2: 関数デコレータ
from crewai import tool

@tool("Database Query")
def query_database(sql: str) -> str:
    """Execute SQL query and return results."""
    return db.execute(sql)

# エージェントにツールを割り当て
researcher = Agent(
    role="Researcher",
    goal="Find information",
    backstory="...",
    tools=[WebSearchTool(), query_database]
)

コラボレーション

委譲トリガー

• langgraph|state machine|graph -> langgraph (明示的な状態管理が必要)
• observability|tracing -> langfuse (LLMオブザーバビリティが必要)
• structured output|json schema -> structured-output (構造化されたレスポンスが必要)

リサーチおよびライティングCrew

スキル: crewai、structured-output

ワークフロー:

1. リサーチャーおよびライターエージェントを定義
2. リサーチ → 分析 → ライティングパイプラインを作成
3. リサーチ形式に構造化出力を使用
4. コンテキストを伴うタスクをチェーン

観測可能なエージェントチーム

スキル: crewai、langfuse

ワークフロー:

1. エージェントとタスク付きクルーを構築
2. Langfuseコールバックハンドラを追加
3. エージェント相互作用を監視
4. 出力品質を評価

Flowsを伴う複雑なワークフロー

スキル: crewai、langgraph

ワークフロー:

1. CrewAI Flowsを使用してワークフローを設計
2. 状態にLangGraphパターンを使用
3. Flowステップでクルーを結合
4. 分岐とルーティングを処理

関連スキル

相性の良いスキル: langgraph、autonomous-agents、langfuse、structured-output

使用すべき場合

• ユーザーが言及または示唆している場合: crewai
• ユーザーが言及または示唆している場合: 複数エージェントチーム
• ユーザーが言及または示唆している場合: エージェントロール
• ユーザーが言及または示唆している場合: エージェントのクルー
• ユーザーが言及または示唆している場合: ロールベースのエージェント
• ユーザーが言及または示唆している場合: 協調的なエージェント

制限事項

• このスキルは、タスクが上記で説明されているスコープと明確に一致する場合にのみ使用してください。
• 出力を環境固有の検証、テスト、または専門家レビューの代替と見なさないでください。
• 必須の入力、権限、安全性の境界、または成功基準が不明な場合は、立ち止まって説明を求めてください。