AIエージェント入門① - よく使われる 5 つのワークフロー

# はじめに

本記事では、AIエージェントを構築する際によく使われる 5 つのワークフロー

1. **Prompt-Chaining**  
2. **Parallelization**  
3. **Routing**  
4. **Evaluator-Optimizer Workflow**  
5. **Orchestrator-Workers Workflow**  

について、それぞれコード例や簡単な図を用いながら解説します。  
さらに、「どのようなタスクに向いているか」も加えて解説しますので、設計の際の参考にしてください。

Anthropicのcookbookのワークフローについて勉強したので、そのまとめ記事となります。

https://github.com/anthropics/anthropic-cookbook/tree/main/patterns/agents

https://www.anthropic.com/research/building-effective-agents

---

# 1. Prompt-Chaining

## 1.1 概要

**Prompt-Chaining** とは、1 つの複雑なタスクを複数のサブタスクに分解し、**前のステップの結果を次のステップに渡しながら**段階的に処理するワークフローです。
たとえば「テキスト抽出 → 情報整形 → 最終フォーマット」という一連のステップを、LLM に順々に実行させます。

```mermaid
flowchart LR
    subgraph Prompt-Chaining
    A[Raw Input] --> B[Step 1]
    B --> C[Step 2]
    C --> D[Step 3]
    D --> E[Final Output]
    end
```

## 1.2 コード例

```python
from typing import List
from util import llm_call  # LLMコールのラッパー関数（仮）

def chain(input_text: str, prompts: List[str]) -> str:
    """
    シンプルなPrompt-Chainingの例:
    前のステップ結果を次のステップに引き継ぐ
    """
    result = input_text
    for i, prompt in enumerate(prompts, 1):
        print(f"\n[Prompt Step {i}]")
        # 前のステップ結果を新しいプロンプトと合体
        chained_prompt = f"{prompt}\nInput: {result}"
        # LLM呼び出し
        result = llm_call(chained_prompt)
        print(result)
    return result

if __name__ == "__main__":
    # 例: テキストを抽出→数値変換→ソート→テーブル化など
    text = """Q3 Performance Report: 
    - Revenue grew by 45%.
    - Customer satisfaction reached 92 points.
    ..."""
    steps = [
        "Extract the metrics and numbers from the text:",
        "Convert the extracted values into a uniform format (percentage or decimal):",
        "Sort the lines by numerical value (descending):",
        "Format as a Markdown table:"
    ]
    final_output = chain(text, steps)
    print("\n[Final Chained Output]")
    print(final_output)
```

## 1.3 メリット・デメリット

- **メリット**  
  - 複雑なタスクを**段階的**に処理するため、エラーが起きてもどのステップに原因があるかを特定しやすい。  
  - 途中結果を人間が確認・修正することも簡単。

- **デメリット**  
  - ステップ数が増えると**APIコスト**や**レスポンス時間**がかさむ。  
  - 中間結果のクオリティが低いと後続ステップへ影響が波及する。

## 1.4 向いているタスク

- **順序依存のある処理**  
  - （例）文書から段階的にデータを抽出・変換し、最終的にレポートを作成する  
- **複数の小ステップをかみ砕いて実行**する必要があるケース  
  - （例）プロンプト → 中間出力 → 次の入力 … と段階を踏んで精度を上げたい場合  
- **ドキュメント分析やロジカルなステップ**を必要とするタスク  
  - （例）構造化 → 整形 → 再構成 など

---

# 2. Parallelization

## 2.1 概要

**Parallelization** は、**依存しない複数のサブタスク**を同時並行で実行するワークフローです。
サブタスク同士が独立している場合、処理を同時に走らせることで大幅に時間を短縮できます。

```mermaid
flowchart TB
    A[Main Task] --> B[Subtask 1]
    A[Main Task] --> C[Subtask 2]
    A[Main Task] --> D[Subtask 3]
    B --> E[Result 1]
    C --> F[Result 2]
    D --> G[Result 3]
    E & F & G --> H[Combine or Collate Results]
```

## 2.2 コード例

```python
import concurrent.futures
from util import llm_call

def parallel(prompt: str, inputs: list[str], max_workers: int = 3) -> list[str]:
    """
    同じプロンプトで複数の入力を並列処理
    """
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(llm_call, f"{prompt}\nInput: {inp}") for inp in inputs]
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results

if __name__ == "__main__":
    inputs = [
        "Stakeholder: Customers",
        "Stakeholder: Employees",
        "Stakeholder: Investors",
        "Stakeholder: Suppliers",
    ]
    prompt = """Analyze market impact for the following stakeholder. Provide recommended actions."""
    
    # 並列で一気に処理
    analysis_results = parallel(prompt, inputs)
    for res in analysis_results:
        print("==Parallel Result==")
        print(res)
```

## 2.3 メリット・デメリット

- **メリット**  
  - **処理時間の短縮**が大きい (サブタスク数が多いほど恩恵が増える)。  
  - 同じプロンプトを使い回す場合、実装がシンプル。

- **デメリット**  
  - サブタスクに**依存関係があると並列化が困難**。  
  - 一度に大量のリクエストを送ると、API 制限・コストが増大。

## 2.4 向いているタスク

- **大量の同型タスク**をまとめて処理したい場合  
  - （例）1000件のレビューを要約・感情分析する  
- **サブタスク同士に依存が無い**ケース  
  - （例）同じ指示で複数の異なる入力を処理  
- **マルチユーザーの問い合わせを一斉に捌く**ようなリアルタイム業務

---

# 3. Routing

## 3.1 概要

**Routing** は、入力テキストの内容に応じて「どの LLM やプロンプトを使うべきか」を自動で判定するワークフローです。
たとえば、問い合わせ内容を解析して **Billing** 用テンプレートか **Technical** 用テンプレートかなどを動的に切り替えます。

```mermaid
flowchart LR
    A[Incoming Request] --> B{Routing Logic}
    B -- billing --> C[Billing Prompt]
    B -- account --> D[Account Prompt]
    B -- technical --> E[Technical Prompt]
    B -- product --> F[Product Prompt]
    C --> G[Billing LLM]
    D --> H[Account LLM]
    E --> I[Technical LLM]
    F --> J[Product LLM]
```

## 3.2 コード例

```python
from typing import Dict
from util import llm_call, extract_xml

def route(input_text: str, routes: Dict[str, str]) -> str:
    """
    LLMでまずルーティング先を選び、その後にルート別の専用プロンプトを用いる例
    """
    # ルーティング先候補を列挙
    available_routes = list(routes.keys())

    # まずはLLMに「どのルートが適切か」判定させる
    selector_prompt = f"""
    You have these options: {available_routes}
    Decide the best route for the following input:
    <input>{input_text}</input>
    Output in XML:
    <selection>The chosen route</selection>
    """

    selector_response = llm_call(selector_prompt)
    chosen_route = extract_xml(selector_response, "selection").strip().lower()

    # 選ばれたルート用のプロンプトを実行
    route_prompt = routes[chosen_route]
    response = llm_call(f"{route_prompt}\nInput: {input_text}")
    return response

if __name__ == "__main__":
    # 各ルート用のプロンプト
    support_routes = {
        "billing": "You are a billing specialist. ...",
        "technical": "You are a technical support engineer. ...",
        "account": "You are an account security specialist. ...",
        "product": "You are a product specialist. ...",
    }
    # 例の問い合わせ
    ticket = "Subject: My account was locked after I forgot my password twice."
    answer = route(ticket, support_routes)
    print("[Routed Answer]")
    print(answer)
```

## 3.3 メリット・デメリット

- **メリット**  
  - 各分野に特化した応答が可能 → **精度向上**。  
  - 分類の仕組みを追加するだけで簡単に拡張できる。

- **デメリット**  
  - 分類（ルーティング）を誤ると、全く無関係の回答が返るリスク。  
  - ルーティング用のLLMが正しく分類できるようにするための設計が必要。

## 3.4 向いているタスク

- **問い合わせ内容が多岐にわたるサポート業務**  
  - （例）「請求」「アカウント」「技術」「製品」などに分かれるカスタマーサポート  
- **特定分野向けに最適化されたモデルを使い分けたい**場合  
  - （例）法律文書、医療文章、ビジネス文書など  
- **マルチエージェント構成**で役割分担を明確化したいケース  
  - （例）1つのチャットボットで複数の役割を統括するが、問い合わせに応じて担当LLMを変える

---

# 4. Evaluator-Optimizer Workflow

## 4.1 概要

**Evaluator-Optimizer Workflow** は、

1. まず **Generator LLM** で回答を生成  
2. **Evaluator LLM** がその回答を評価し、フィードバックを出す  
3. フィードバックを受けて再度 **Generator LLM** が回答を修正  
4. 条件を満たすまで **1〜3** を繰り返す

というループを回すワークフローです。

```mermaid
flowchart LR
    subgraph Evaluator-Optimizer
    G[Generate] --> E[Evaluate]
    E -- Needs Improvement --> G
    E -- PASS --> F[Final Output]
    end
```

## 4.2 コード例

```python
from util import llm_call, extract_xml

def generate(generator_prompt: str, user_task: str) -> str:
    """
    Generator LLMが回答を1回作る
    """
    full_prompt = f"{generator_prompt}\nTask: {user_task}"
    response = llm_call(full_prompt)
    result = extract_xml(response, "response")
    return result

def evaluate(evaluator_prompt: str, content: str, user_task: str) -> (str, str):
    """
    Evaluator LLMが回答の合否とフィードバックを出す
    """
    full_prompt = f"{evaluator_prompt}\nOriginal Task: {user_task}\nContent: {content}"
    response = llm_call(full_prompt)
    evaluation = extract_xml(response, "evaluation")  # PASS / NEEDS_IMPROVEMENT / FAIL
    feedback = extract_xml(response, "feedback")
    return evaluation, feedback

def evaluator_optimizer_loop(task: str, evaluator_prompt: str, generator_prompt: str):
    while True:
        # 1. 生成フェーズ
        result = generate(generator_prompt, task)
        # 2. 評価フェーズ
        status, feedback = evaluate(evaluator_prompt, result, task)
        print(f"[Evaluation] {status}: {feedback}")
        if status == "PASS":
            return result
        else:
            # フィードバックをプロンプトに追加し再生成
            generator_prompt += f"\nFeedback: {feedback}"

if __name__ == "__main__":
    # 例: コード生成タスク
    task_description = "Implement a MinStack with O(1) operations for push, pop, and getMin."
    evaluator_prompt = """
    Evaluate the code. Output:
    <evaluation>PASS or NEEDS_IMPROVEMENT</evaluation>
    <feedback>Improvement points</feedback>
    """
    generator_prompt = """
    <response>
    # code implementation here
    </response>
    """
    final_code = evaluator_optimizer_loop(task_description, evaluator_prompt, generator_prompt)
    print("[Final Optimized Code]")
    print(final_code)
```

## 4.3 メリット・デメリット

- **メリット**  
  - **自動 QA** のような働きにより、最終的に高品質な結果を得やすい。  
  - 修正指示が具体的なので、**改善サイクル**が回しやすい。

- **デメリット**  
  - **評価と再生成の繰り返し**でコスト・時間が増す。  
  - フィードバックの正しさが保証されなければ、誤った修正へ誘導されるおそれ。

## 4.4 向いているタスク

- **出力の品質が非常に重要なタスク**  
  - （例）コード生成、論文執筆、契約書など、ミスが許されない文書  
- **何度か試行錯誤しながら完成形に近づけたい**場面  
  - （例）クリエイティブな文章作成、デザイン案のブラッシュアップなど  
- **QAやレビュー工程を自動化したい**場合  
  - （例）生成されたコンテンツを自動採点・フィードバックして最適化する

---

# 5. Orchestrator-Workers Workflow

## 5.1 概要

**Orchestrator-Workers Workflow** では、
- **Orchestrator LLM** が、与えられたタスクを**動的に**いくつかのサブタスクに分解  
- 分解されたサブタスクを、複数の **Worker LLM** が並行または段階的に処理  
- **Orchestrator** が結果をまとめて最終出力

という流れを取ります。
どのようなサブタスクに分解するかを Orchestrator LLM が**自律的に判断**できる点が特徴です。

```mermaid
flowchart LR
    A[Input Task] --> O[Orchestrator LLM]
    O --> W1[Worker LLM1]
    O --> W2[Worker LLM2]
    O --> W3[Worker LLM3]
    W1 --> O
    W2 --> O
    W3 --> O
    O --> B[Aggregated Result]
```

## 5.2 コード例

```python
from typing import Dict, List
from util import llm_call, extract_xml

class FlexibleOrchestrator:
    def __init__(self, orchestrator_prompt: str, worker_prompt: str):
        self.orchestrator_prompt = orchestrator_prompt
        self.worker_prompt = worker_prompt

    def process(self, task: str) -> Dict:
        # 1. Orchestratorにタスク分解を依頼
        orchestrator_input = self.orchestrator_prompt.format(task=task)
        orchestrator_response = llm_call(orchestrator_input)

        analysis = extract_xml(orchestrator_response, "analysis")
        tasks_xml = extract_xml(orchestrator_response, "tasks")

        # 2. サブタスクをパース (詳細実装は省略)
        subtask_list = self._parse_tasks(tasks_xml)

        # 3. Workerにそれぞれのサブタスクを処理させる
        results = []
        for subtask in subtask_list:
            worker_input = self.worker_prompt.format(
                original_task=task,
                task_type=subtask["type"],
                description=subtask["description"]
            )
            worker_response = llm_call(worker_input)
            results.append({
                "type": subtask["type"],
                "result": extract_xml(worker_response, "response")
            })
        
        return {"analysis": analysis, "worker_results": results}
    
    def _parse_tasks(self, tasks_xml: str) -> List[Dict]:
        # XMLをパースして {type, description} のリストを作る想定
        return []  # 実際はXML解析処理が必要

if __name__ == "__main__":
    # Orchestrator用のプロンプトとWorker用のプロンプトを用意
    orchestrator_prompt = """
    Analyze the task: {task}
    Output in XML:
    <analysis>...</analysis>
    <tasks>
       <task>
         <type>formal</type>
         <description>...</description>
       </task>
       <task>
         <type>conversational</type>
         <description>...</description>
       </task>
    </tasks>
    """
    worker_prompt = """
    <response>
    Here is the {task_type} version of {original_task}:
    (content)
    </response>
    """

    orchestrator = FlexibleOrchestrator(orchestrator_prompt, worker_prompt)
    final_result = orchestrator.process("Write a product description for an eco-friendly water bottle")
    print(final_result)
```

## 5.3 メリット・デメリット

- **メリット**  
  - **動的にタスクを分割**するため、複雑・包括的なタスクに柔軟対応できる。  
  - 複数の Worker に割り振ることで、同時並行実行も可能。

- **デメリット**  
  - Orchestrator と複数 Worker の連携が必要で**実装がやや複雑**。  
  - Orchestrator が誤ったタスク分解を行うと、成果物全体に影響が出る。

## 5.4 向いているタスク

- **タスク分解が予測困難な複雑タスク**  
  - （例）長文要約、様々な切り口での分析、別々のジャンルのテキスト生成  
- **複数の専門性が必要なケース**  
  - （例）法律＋テクニカル＋経営戦略など、分野ごとに別々のモデルやプロンプトを使いたい  
- **複雑なワークフローを全自動化したい**場合  
  - （例）マーケティング用キャッチコピーの生成・分析・評価をまとめて行う

---

# まとめ

本記事では、AIエージェントをより高精度・高効率に構築するための 5 つの手法を紹介し、それぞれにコード例と図、また「どのようなタスクに向いているか」をまとめました。

1. **Prompt-Chaining**  
   - **段階的に処理**することで分かりやすく、ステップごとの検証が容易。  
   - 順序依存のあるタスクや分割ステップが明確なタスクに有効。

2. **Parallelization**  
   - 依存のないサブタスクを**並列実行**して高速化。  
   - 同型タスクの大量処理や、サブタスクに依存関係がない場合に適切。

3. **Routing**  
   - 入力を**自動分類**して適切なLLMやテンプレートに振り分ける。  
   - 多岐にわたる問い合わせや、専門モデルの使い分けに向いている。

4. **Evaluator-Optimizer Workflow**  
   - 生成と評価を何度も繰り返すことで**品質を高める**。  
   - コード生成や契約書など、**高い精度**が求められるタスクに活用可能。

5. **Orchestrator-Workers Workflow**  
   - 中央のOrchestratorが**動的にタスク分解**し、複数のWorkerで処理。  
   - 専門性が多岐にわたる複雑なタスクや、多段プロセスを自動化するケースに向く。

実際のAIエージェント開発では、**これらのワークフローを組み合わせ**たり、部分的にカスタマイズしたりして適用するケースが少なくありません。
タスクの性質や要求精度、システムのコスト要件などに合わせて最適な方法を選定していただければ幸いです。

# 次のアクション

本記事の続編・関連記事があるので、ご参考にしてみてk

https://qiita.com/syukan3/items/ddb51cd2e6b853f0f948

https://qiita.com/syukan3/items/153409d7f387ea8c3065

https://qiita.com/syukan3/items/3f9e857dc0203430ea76

https://qiita.com/syukan3/items/174e43235bde8a1a0694

---
### 会社紹介

**株式会社 Mosaica**  
最先端テクノロジーで社会課題を解決し、持続可能な未来を創造する IT カンパニー。
AI ソリューション、クラウド統合、DX 推進、経営コンサルティングなど包括的なサービスでビジネス変革を支援しています。  

詳しくは [公式サイト](https://mosaica.co.jp/) までお気軽にご相談ください。
公式サイト： https://mosaica.co.jp/