CrewAI のガイドから「最初の Flow の構築」を翻訳しました。構造化された、イベント駆動のワークフローを作成する方法を学習します。

CrewAI : ガイド : 最初の Flow の構築

作成 : クラスキャット・セールスインフォメーション
作成日時 : 04/01/2025

* 本記事は github.com/crewAIInc/crewAI/docs の以下のページを独自に翻訳した上でまとめ直し、補足説明を加えています :

• Build Your First Flow

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

 

CrewAI : ガイド : 最初の Flow の構築

実行に対する正確な制御を持つ、構造化された、イベント駆動のワークフローを作成する方法を学習します。

 

Flow で AI ワークフローを制御する

CrewAI Flow は AI オーケストレーションの次のレベルを表現しています – AI エージェントクルーのコラボレーション能力を手続型プログラミングの正確性や柔軟性と組み合わせます。crew がエージェントのコラボに優れている一方で、flow は AI システムの様々なコンポーネントがいつどのように相互作用するかを正確にきめ細かく制御できます。

このガイドでは、任意のトピックについて包括的な学習ガイドを生成する強力な CrewAI Flow の作成を順を追って説明します。このチュートリアルは、通常のコード、直接的な LLM 呼び出しと crew ベースの処理を組み合わせることで、Flow が AI ワークフローに対する構造化された、イベント駆動の制御を提供する方法を実演します。

 

何が Flow を強力にするか

Flow は以下を可能にします :

1. 様々な AI 相互作用 (interaction) パターンを組み合わせる – 複雑な協調タスクには crew を、単純な操作には直接的な LLM 呼び出しを、そして手続的なロジックには通常のコードを使用します。
   

2. イベント駆動型システムの構築 – コンポーネントが特定のイベントやデータ変更に応答する方法を定義します。
   

3. コンポーネントに渡る状態の維持 – アプリケーションの異なるパート間でデータを共有して変換します。
   

4. 外部システムとの統合 – AI ワークフローをデータベース、API やユーザインターフェイスとシームレスに接続します。
   

5. 複雑な実行パスの作成 – 条件分岐、並列処理、そして動的ワークフローを設計します。

 

何を構築して学習するか

このガイドの最後までに、以下が成されます :

1. 洗練されたコンテンツ生成システムの作成、これはユーザ入力、AI プランニング、そしてマルチエージェントのコンテンツ作成を組み合わせます。
   

2. システムの様々なコンポーネント間の 情報のフローのオーケストレーション
   

3. イベント駆動アーキテクチャの実装、そこでは各ステップは前のステップの完了に応答します。
   

4. より複雑な AI アプリケーションの基盤の構築、これは拡張してカスタマイズできます。

このガイド・クリエイター・フローは、以下のようなより高度なアプリケーションを作成するために適用できる、基本パターンを実演します :

• 複数の特殊なサブシステムを組み合わせる、インタラクティブな AI アシスタント
  

• AI 強化された変換による複雑なデータ処理パイプライン
  

• 外部サービスや API を統合する自律エージェント
  

• human-in-the-loop プロセスを含む、多段階意思決定システム

Let’s dive in and build your first flow!

 

前提条件

始める前に、以下を確認してください :

• インストールガイド に従って CrewAI がインストールされている
• 環境変数に OpenAI API キーをセットアップ
• Python の基本的な理解

 

Step 1: 新しい CrewAI Flow プロジェクトを作成する

最初に、CLI を使用して新しい CrewAI Flow プロジェクトを作成しましょう。このコマンドは、flow のためにすべての必要なディレクトリとテンプレートファイルを含む雛形 (scaffolded) プロジェクトをセットアップします。

crewai create flow guide_creator_flow
cd guide_creator_flow

これは flow に必要な基本構造を持つプロジェクトを生成します。

CrewAI フレームワーク概要

 

Step 2: プロジェクト構造の理解

生成されたプロジェクトは以下の構造を持ちます。それに馴染むように少し時間を取ってください、この構造の理解は将来的により複雑なフローを作成するのに役立ちます。

guide_creator_flow/
├── .gitignore
├── pyproject.toml
├── README.md
├── .env
├── main.py
├── crews/
│   └── poem_crew/
│       ├── config/
│       │   ├── agents.yaml
│       │   └── tasks.yaml
│       └── poem_crew.py
└── tools/
    └── custom_tool.py

この構造は flow の様々なコンポーネント間の明確な分離を与えます :

• main.py ファイル内のメイン・フローロジック
• crews ディレクトリの専門クルー
• tools ディレクトリのカスタムツール

この構造を変更してガイド・クリエイター flow を作成します、これは包括的な学習ガイドを生成するプロセスのオーケストレーションを行います。

 

Step 3: コンテンツ・ライター Crew の追加

flow はコンテンツ作成プロセスを処理するために専門の crew を必要とします。CrewAI CLI を使用してコンテンツ・ライター crew を追加しましょう :

crewai flow add-crew content-crew

このコマンドは crew のために必要なディレクトリとテンプレートファイルを自動的に作成します。コンテンツ・ライター crew はガイドのセクションの執筆とレビューを担当し、メインのアプリケーションにより編成される flow 全体の中で作業します。

 

Step 4: コンテンツ・ライター crew の設定

次に、コンテンツ・ライター crew 用に生成されたファイルを変更しましょう。2 つの特化型エージェント – ライターとレビュアー – をセットアップします、これらは協力してガイドの高品質なコンテンツを作成します。

1. 最初に、agents 設定ファイルを更新してコンテンツ・クリエイター・チームを定義します :

   # src/guide_creator_flow/crews/content_crew/config/agents.yaml
   content_writer:
     role: >
       Educational Content Writer
     goal: >
       Create engaging, informative content that thoroughly explains the assigned topic
       and provides valuable insights to the reader
     backstory: >
       You are a talented educational writer with expertise in creating clear, engaging
       content. You have a gift for explaining complex concepts in accessible language
       and organizing information in a way that helps readers build their understanding.
     llm: openai/gpt-4o-mini
   
   content_reviewer:
     role: >
       Educational Content Reviewer and Editor
     goal: >
       Ensure content is accurate, comprehensive, well-structured, and maintains
       consistency with previously written sections
     backstory: >
       You are a meticulous editor with years of experience reviewing educational
       content. You have an eye for detail, clarity, and coherence. You excel at
       improving content while maintaining the original author's voice and ensuring
       consistent quality across multiple sections.
     llm: openai/gpt-4o-mini
   

   これらのエージェント定義は、AI エージェントがコンテンツ作成にアプローチする方法を形作る、専門の役割と視点を確立します。各エージェントが明確な目的と専門性をどのように持つかに注目してください。
    

2. 次に、タスク設定ファイルを更新して特定の執筆とレビュータスクを定義します :

   # src/guide_creator_flow/crews/content_crew/config/tasks.yaml
   write_section_task:
     description: >
       Write a comprehensive section on the topic: "{section_title}"
   
       Section description: {section_description}
       Target audience: {audience_level} level learners
   
       Your content should:
       1. Begin with a brief introduction to the section topic
       2. Explain all key concepts clearly with examples
       3. Include practical applications or exercises where appropriate
       4. End with a summary of key points
       5. Be approximately 500-800 words in length
   
       Format your content in Markdown with appropriate headings, lists, and emphasis.
   
       Previously written sections:
       {previous_sections}
   
       Make sure your content maintains consistency with previously written sections
       and builds upon concepts that have already been explained.
     expected_output: >
       A well-structured, comprehensive section in Markdown format that thoroughly
       explains the topic and is appropriate for the target audience.
     agent: content_writer
   
   review_section_task:
     description: >
       Review and improve the following section on "{section_title}":
   
       {draft_content}
   
       Target audience: {audience_level} level learners
   
       Previously written sections:
       {previous_sections}
   
       Your review should:
       1. Fix any grammatical or spelling errors
       2. Improve clarity and readability
       3. Ensure content is comprehensive and accurate
       4. Verify consistency with previously written sections
       5. Enhance the structure and flow
       6. Add any missing key information
   
       Provide the improved version of the section in Markdown format.
     expected_output: >
       An improved, polished version of the section that maintains the original
       structure but enhances clarity, accuracy, and consistency.
     agent: content_reviewer
     context:
       - write_section_task
   

   これらのタスク定義はエージェントへの詳細な指示を提供し、品質基準を満たすコンテンツを生成することを確実にします。review タスクの context パラメータが、reviwer が writer の出力へのアクセスできるようなワークフローを作成する方法に注意してください。
    

3. 次に、crew 実装ファイルを更新してエージェントとタスクがどのように連携するかを定義します :

   # src/guide_creator_flow/crews/content_crew/content_crew.py
   from crewai import Agent, Crew, Process, Task
   from crewai.project import CrewBase, agent, crew, task
   
   @CrewBase
   class ContentCrew():
       """Content writing crew"""
   
       @agent
       def content_writer(self) -> Agent:
           return Agent(
               config=self.agents_config['content_writer'],
               verbose=True
           )
   
       @agent
       def content_reviewer(self) -> Agent:
           return Agent(
               config=self.agents_config['content_reviewer'],
               verbose=True
           )
   
       @task
       def write_section_task(self) -> Task:
           return Task(
               config=self.tasks_config['write_section_task']
           )
   
       @task
       def review_section_task(self) -> Task:
           return Task(
               config=self.tasks_config['review_section_task'],
               context=[self.write_section_task()]
           )
   
       @crew
       def crew(self) -> Crew:
           """Creates the content writing crew"""
           return Crew(
               agents=self.agents,
               tasks=self.tasks,
               process=Process.sequential,
               verbose=True,
           )
   

   crew 定義はエージェントとタスク間の関係を確立し、コンテンツ・ライターが下書きを作成してから reviewer がそれを改良するような順次プロセスをセットアップします。crew が独立的に機能する一方、flow ではより大きなシステムの一部としてオーケストレーションされます。

 

Step 5: Flow の作成

さて、ここからがエキサイティングな部分です – ガイド作成プロセス全体をオーケストレーションする flow を作成します。ここでは、通常の Python コード、直接的な LLM 呼び出し、そしてコンテンツ作成 crew を包括的なシステムに組み合わせていきます。

flow は :

1. トピックと対象 (audience) のレベルについてのユーザ入力を取得します。
   

2. 直接 LLM 呼び出しを行なって構造化されたガイドのアウトラインを作成します。
   

3. コンテンツ・ライター crew を使用して各セクションを順次処理します。
   

4. すべてを最終的な包括的なドキュメントにまとめます。

main.py ファイルに flow を作成しましょう :

#!/usr/bin/env python
import json
from typing import List, Dict
from pydantic import BaseModel, Field
from crewai import LLM
from crewai.flow.flow import Flow, listen, start
from guide_creator_flow.crews.content_crew.content_crew import ContentCrew

# Define our models for structured data
class Section(BaseModel):
    title: str = Field(description="Title of the section")
    description: str = Field(description="Brief description of what the section should cover")

class GuideOutline(BaseModel):
    title: str = Field(description="Title of the guide")
    introduction: str = Field(description="Introduction to the topic")
    target_audience: str = Field(description="Description of the target audience")
    sections: List[Section] = Field(description="List of sections in the guide")
    conclusion: str = Field(description="Conclusion or summary of the guide")

# Define our flow state
class GuideCreatorState(BaseModel):
    topic: str = ""
    audience_level: str = ""
    guide_outline: GuideOutline = None
    sections_content: Dict[str, str] = {}

class GuideCreatorFlow(Flow[GuideCreatorState]):
    """Flow for creating a comprehensive guide on any topic"""

    @start()
    def get_user_input(self):
        """Get input from the user about the guide topic and audience"""
        print("\n=== Create Your Comprehensive Guide ===\n")

        # Get user input
        self.state.topic = input("What topic would you like to create a guide for? ")

        # Get audience level with validation
        while True:
            audience = input("Who is your target audience? (beginner/intermediate/advanced) ").lower()
            if audience in ["beginner", "intermediate", "advanced"]:
                self.state.audience_level = audience
                break
            print("Please enter 'beginner', 'intermediate', or 'advanced'")

        print(f"\nCreating a guide on {self.state.topic} for {self.state.audience_level} audience...\n")
        return self.state

    @listen(get_user_input)
    def create_guide_outline(self, state):
        """Create a structured outline for the guide using a direct LLM call"""
        print("Creating guide outline...")

        # Initialize the LLM
        llm = LLM(model="openai/gpt-4o-mini", response_format=GuideOutline)

        # Create the messages for the outline
        messages = [
            {"role": "system", "content": "You are a helpful assistant designed to output JSON."},
            {"role": "user", "content": f"""
            Create a detailed outline for a comprehensive guide on "{state.topic}" for {state.audience_level} level learners.

            The outline should include:
            1. A compelling title for the guide
            2. An introduction to the topic
            3. 4-6 main sections that cover the most important aspects of the topic
            4. A conclusion or summary

            For each section, provide a clear title and a brief description of what it should cover.
            """}
        ]

        # Make the LLM call with JSON response format
        response = llm.call(messages=messages)

        # Parse the JSON response
        outline_dict = json.loads(response)
        self.state.guide_outline = GuideOutline(**outline_dict)

        # Save the outline to a file
        with open("output/guide_outline.json", "w") as f:
            json.dump(outline_dict, f, indent=2)

        print(f"Guide outline created with {len(self.state.guide_outline.sections)} sections")
        return self.state.guide_outline

    @listen(create_guide_outline)
    def write_and_compile_guide(self, outline):
        """Write all sections and compile the guide"""
        print("Writing guide sections and compiling...")
        completed_sections = []

        # Process sections one by one to maintain context flow
        for section in outline.sections:
            print(f"Processing section: {section.title}")

            # Build context from previous sections
            previous_sections_text = ""
            if completed_sections:
                previous_sections_text = "# Previously Written Sections\n\n"
                for title in completed_sections:
                    previous_sections_text += f"## {title}\n\n"
                    previous_sections_text += self.state.sections_content.get(title, "") + "\n\n"
            else:
                previous_sections_text = "No previous sections written yet."

            # Run the content crew for this section
            result = ContentCrew().crew().kickoff(inputs={
                "section_title": section.title,
                "section_description": section.description,
                "audience_level": self.state.audience_level,
                "previous_sections": previous_sections_text,
                "draft_content": ""
            })

            # Store the content
            self.state.sections_content[section.title] = result.raw
            completed_sections.append(section.title)
            print(f"Section completed: {section.title}")

        # Compile the final guide
        guide_content = f"# {outline.title}\n\n"
        guide_content += f"## Introduction\n\n{outline.introduction}\n\n"

        # Add each section in order
        for section in outline.sections:
            section_content = self.state.sections_content.get(section.title, "")
            guide_content += f"\n\n{section_content}\n\n"

        # Add conclusion
        guide_content += f"## Conclusion\n\n{outline.conclusion}\n\n"

        # Save the guide
        with open("output/complete_guide.md", "w") as f:
            f.write(guide_content)

        print("\nComplete guide compiled and saved to output/complete_guide.md")
        return "Guide creation completed successfully"

def kickoff():
    """Run the guide creator flow"""
    GuideCreatorFlow().kickoff()
    print("\n=== Flow Complete ===")
    print("Your comprehensive guide is ready in the output directory.")
    print("Open output/complete_guide.md to view it.")

def plot():
    """Generate a visualization of the flow"""
    flow = GuideCreatorFlow()
    flow.plot("guide_creator_flow")
    print("Flow visualization saved to guide_creator_flow.html")

if __name__ == "__main__":
    kickoff()

 
この flow で何が起こっているか分析しましょう :

1. 構造化データ用に Pydantic モデルを定義して、型の安全性と明確なデータ表現を確実にします
   

2. 状態 (state) クラスを作成して、flow の様々なステップに渡りデータを保持します
   

3. 3 つの主要な flow ステップを実装します :
   • @start() デコレータを使用してユーザ入力を取得する
   • 直接 LLM 呼び出しを使用してガイドのアウトラインを作成
   • コンテンツ crew でセクションを処理

4. @listen() デコレータを使用してステップ間のイベント駆動関係を確立します

これが flow のパワーです – 様々な種類の処理 (ユーザ・インタラクション、直接 LLM 呼び出し、crew ベースのタスク) を一貫性のある、イベント駆動型システムに組み合わせます。

 

Step 6: 環境変数のセットアップ

プロジェクト・ルートに API キーを含む .env ファイルを作成します :

OPENAI_API_KEY=your_openai_api_key

 

Step 7: 依存関係のインストール

必要な依存関係をインストールします :

crewai install

 

Step 8: Flow の実行

flow が実際に動作するのを確認するときです！CrewAI CLI を使用して実行します :

crewai flow kickoff

このコマンドを実行すると、flow が活発に動き出すのを見るでしょう :

1. トピックと対象者レベルの入力が求められます。
   

2. ガイド用に構造化されたアウトラインを作成します。
   

3. コンテンツ・ライターとレビュアーが互いに協力して、各セクションを処理します。
   

4. 最後に、すべてを包括的なガイドに編集します。

これは、AI と非 AI の両方の複数のコンポーネントを含む複雑なプロセスのオーケストレーションを行なう flow のパワーを実演しています。

 

Step 9: Flow の視覚化

flow の協力な機能の一つは、構造を視覚化する機能です :

crewai flow plot

これは、様々なステップ間の関係やそれらの間を流れるデータを含む、flow の構造を表示する HTML ファイルを作成します。この視覚化は複雑なフローの理解やデバッグに大きな価値があります。

 

Step 10: 出力のレビュー

flow が完了すると、output ディレクトリに 2 つのファイルが生成されます :

1. guide_outline.json: ガイドの構造化されたアウトラインを含みます。
   

2. complete_guide.md: すべてのセクションを含む包括的なガイド

少し時間を取ってこれらのファイルをレビューし、構築したもの – ユーザ入力、直接的な AI インタラクション、協調的なエージェント作業をまとめて複雑で高品質な出力を生成するシステムを評価してください。

以上