はじめに
本記事では、AIエージェントを構築する際によく使われる 5 つのワークフロー
- Prompt-Chaining
- Parallelization
- Routing
- Evaluator-Optimizer Workflow
- Orchestrator-Workers Workflow
について、それぞれコード例や簡単な図を用いながら解説します。
さらに、「どのようなタスクに向いているか」も加えて解説しますので、設計の際の参考にしてください。
Anthropicのcookbookのワークフローについて勉強したので、そのまとめ記事となります。
1. Prompt-Chaining
1.1 概要
Prompt-Chaining とは、1 つの複雑なタスクを複数のサブタスクに分解し、前のステップの結果を次のステップに渡しながら段階的に処理するワークフローです。
たとえば「テキスト抽出 → 情報整形 → 最終フォーマット」という一連のステップを、LLM に順々に実行させます。
1.2 コード例
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 は、依存しない複数のサブタスクを同時並行で実行するワークフローです。
サブタスク同士が独立している場合、処理を同時に走らせることで大幅に時間を短縮できます。
2.2 コード例
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 用テンプレートかなどを動的に切り替えます。
3.2 コード例
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 は、
- まず Generator LLM で回答を生成
- Evaluator LLM がその回答を評価し、フィードバックを出す
- フィードバックを受けて再度 Generator LLM が回答を修正
- 条件を満たすまで 1〜3 を繰り返す
というループを回すワークフローです。
4.2 コード例
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 が自律的に判断できる点が特徴です。
5.2 コード例
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 つの手法を紹介し、それぞれにコード例と図、また「どのようなタスクに向いているか」をまとめました。
-
Prompt-Chaining
- 段階的に処理することで分かりやすく、ステップごとの検証が容易。
- 順序依存のあるタスクや分割ステップが明確なタスクに有効。
-
Parallelization
- 依存のないサブタスクを並列実行して高速化。
- 同型タスクの大量処理や、サブタスクに依存関係がない場合に適切。
-
Routing
- 入力を自動分類して適切なLLMやテンプレートに振り分ける。
- 多岐にわたる問い合わせや、専門モデルの使い分けに向いている。
-
Evaluator-Optimizer Workflow
- 生成と評価を何度も繰り返すことで品質を高める。
- コード生成や契約書など、高い精度が求められるタスクに活用可能。
-
Orchestrator-Workers Workflow
- 中央のOrchestratorが動的にタスク分解し、複数のWorkerで処理。
- 専門性が多岐にわたる複雑なタスクや、多段プロセスを自動化するケースに向く。
実際のAIエージェント開発では、これらのワークフローを組み合わせたり、部分的にカスタマイズしたりして適用するケースが少なくありません。
タスクの性質や要求精度、システムのコスト要件などに合わせて最適な方法を選定していただければ幸いです。