CrewAI でマルチエージェントシステムを構築する際、各エージェントが異なるモデルを使えると、より興味深い設計が可能になります。
研究者には高速かつ経済的なモデルが有益で、アナリストには強力な推論モデルが必要となり、ライターには高品質な長文生成に最適化されたモデルが求められる場合があります。従来、これらのエージェントを異なるプロバイダに接続するには、個別の API 資格情報、エンドポイント、SDK、課金システム、プロバイダ固有の設定を管理する必要がありました。
よりクリーンなアーキテクチャは、エージェントとワークフローの管理を CrewAI に任せ、モデルアクセスを CometAPI が担うことです。
CometAPI は OpenAI 互換エンドポイント https://api.cometapi.com/v1 を提供しており、アプリケーションは共通の API インターフェースを通じて複数プロバイダのモデルへリクエストをルーティングできます。現行のクイックスタートドキュメントでは、API キーと base URL を変更することで標準の OpenAI Python SDK の利用もサポートしています。
このチュートリアルでは、以下を備えた 3 エージェントの CrewAI ワークフローを構築します。
- 研究: Gemini 3.7 Flash
- 分析: Claude Opus 5
- ライティング: GPT-5.6
- CometAPI の API キー 1 つ
- API base URL 1 つ
- エージェントごとのモデル設定
- 一時的な障害に対する制限付きフォールバック
- 本番環境向けの CrewAI チェックポイント
- トークンおよび実行の利用状況トラッキング
- サーバー側のモデル検証
重要なアーキテクチャ上の境界はシンプルです。
CrewAI はエージェントのオーケストレーションを担当。CometAPI はモデルアクセスを担当。ルーティングは Model ID で定義。
CrewAI のマルチエージェント・モデルルーティングとは？
CrewAI は、エージェント、タスク、クルー、マルチエージェントワークフローを作成するための Python フレームワークです。各エージェントは独自の LLM 設定を持て、Crew がそれらのタスク実行とコンテキストの受け渡しを調整します。
CrewAI の現行の LLM 設定は、model、api_key、base_url を明示的に指定でき、OpenAI 互換エンドポイントも設定可能です。
これにより、マルチモデルのアーキテクチャが素直に設計できます。
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
エージェントは論理的には分離されたままですが、モデルアクセスが一元化されます。
これは「すべてのモデルが置き換え可能」という意味ではありません。OpenAI 互換 API は共通のリクエストインターフェースを提供しますが、コンテキスト上限、ツールサポート、推論制御、出力挙動、レイテンシ、価格が同一であることを保証するものではありません。
この違いは、プロダクション向けのルーティング設計において重要です。
なぜ CrewAI と CometAPI を組み合わせるのか？
最大の利点は、CrewAI が突然マルチプロバイダ対応になることではありません。CrewAI 自体が複数の LLM プロバイダをすでにサポートしています。
利点は、モデルアクセスを 1 つの API レイヤーの背後に統合できることです。
統合 API レイヤーがない場合、3 エージェントのワークフローは次のようになります。
| エージェント | プロバイダ | 資格情報 | 統合方式 |
|---|---|---|---|
| Researcher | Google API key | プロバイダ固有 | |
| Analyst | Anthropic | Anthropic API key | プロバイダ固有 |
| Writer | OpenAI | OpenAI API key | プロバイダ固有 |
CometAPI を使うと:
| エージェント | モデル | 資格情報 | エンドポイント |
|---|---|---|---|
| Researcher | Gemini 3.7 Flash | CometAPI key | CometAPI |
| Analyst | Claude Opus 5 | CometAPI key | CometAPI |
| Writer | GPT-5.6 | CometAPI key | CometAPI |
CometAPI の現行クイックスタートでは、OpenAI API の base URL を置き換えるドロップインエンドポイントとして説明され、同一サービス内で複数プロバイダのモデル一覧を提供します。
これにより、アプリケーションで有用な責務分離が得られます。
CrewAI
- エージェントの役割を定義
- タスクを定義
- コンテキストを受け渡し
- 実行を制御
- エージェントの反復を管理
- クルー全体のオーケストレーションを担当
CometAPI
- 共通のモデルアクセスレイヤーを提供
- API 認証を一元化
- Model ID によるモデルルーティングを提供
- アプリケーションに 1 つの API エンドポイントを提供
- 利用状況と課金の可視化を集約
この CrewAI ワークフローは何を構築するのか？
この例では、3 つのエージェントを逐次的に実行します。
| CrewAI エージェント | メインのモデル | フォールバック | 役割 |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | 事実収集とリサーチ |
| Product Analyst | claude-opus-5 | gpt-5.6 | 証拠とトレードオフの統合 |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | 最終的な意思決定メモの作成 |
これはあくまで例示的なルーティング方針であり、ベンチマーク順位ではありません。
エージェントに適したモデルは次に依存します。
- タスクの複雑さ
- 必要なコンテキスト長
- ツール使用の有無
- 構造化出力の要件
- レイテンシ
- 信頼性
- トークンコスト
- 出力品質
- アプリケーション固有の評価結果
有用な原則は次のとおりです。
プロバイダではなく、エージェントの仕事に適したモデルを選ぶこと。
各 CrewAI エージェントはどのモデルを使うべき？
この例では、シンプルなコスト対能力の戦略でモデルを割り当てます。
Researcher: Gemini 3.7 Flash
リサーチでは、比較的大量の情報を処理し、コンパクトな中間成果物を作ることが多いです。
高速モデルは、高ボリュームのリサーチタスクに有用です。
"researcher": "gemini-3.7-flash"
Analyst: Claude Opus 5
アナリストの役割はより狭いものの、推論の負荷が高いです。リサーチの出力を受け取り、推奨に変換します。
"analyst": "claude-opus-5"
Writer: GPT-5.6
最終エージェントは、リサーチと分析を開発者向けの意思決定メモに変換します。
"writer": "gpt-5.6"
重要なのはこの 3 つの割り当てそのものではありません。ルーティングポリシーを固定する前に、代表的なタスクに対して候補モデルを評価してください。
開始前に必要なものは？
必要なもの:
- Python 3.10+
* CrewAI
* OpenAI Python SDK 互換性
* `python-dotenv`
* CometAPI の API キー
* 使用予定の Model ID
CometAPI の現行 Python 連携は OpenAI 互換 API をサポートしており、公式の CometAPI Python パッケージでは COMETAPI_KEY と COMETAPI_BASE_URL を環境変数で設定する方法が記載されています。
標準エンドポイントは次のとおりです。
https://api.cometapi.com/v1
デプロイ前に、選択した Model ID が現在利用可能であり、CrewAI ワークロードで必要なエンドポイントやパラメータをサポートしていることを確認してください。モデルカタログや価格は変わり得ます。
CrewAI と依存関係はどうやってインストールする？
新しい Python 環境を作成します。
python -m venv .venv
有効化します。
source .venv/bin/activate
Windows の場合:
.venv\Scripts\Activate.ps1
依存関係をインストールします。
pip install "crewai[openai]" openai python-dotenv
以下のフォールバック実装では OpenAI SDK の例外クラスを直接 import しているため、openai を明示的に指定しています。
本番運用では、最新の浮動バージョンに依存し続けるのではなく、テストしたバージョンを固定してください。
例:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
CrewAI の LLM レイヤーは活発に進化しているため、正確なコンストラクタやプロバイダ設定は、アプリケーションで使用する CrewAI のバージョンに合わせて確認してください。現行の CrewAI ドキュメントでは、カスタム base_url と API キーを指定した LLM の構成をサポートしています。
CometAPI の API キーはどう設定する？
.env ファイルを作成します。
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Python で読み込みます。
import osfrom dotenv import load_dotenvload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv( "COMETAPI_BASE_URL", "https://api.cometapi.com/v1",)
.env を Git にコミットしないでください。
.gitignore に追加します。
.env.venv/__pycache__/
API キーはサーバー側の資格情報として保持すべきです。CometAPI の現行クイックスタートでも、キーはソースコードではなく環境変数に保管することを推奨しています。
CrewAI を CometAPI にどう接続する？
CrewAI の LLM オブジェクトには、モデル名、API キー、カスタム base URL を渡せます。
ヘルパーを作成します。
from crewai import LLMdef cometapi_llm(model_id: str) -> LLM: return LLM( model=model_id, base_url=COMETAPI_BASE_URL, api_key=COMETAPI_KEY, timeout=60.0, max_retries=0, )
同じ設定を各エージェントに個別埋め込みするよりも望ましいです。
各エージェントには Model ID だけが必要になります。
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
なぜ max_retries=0 にするのか？
理由はフォールバック制御のためです。
基盤の LLM クライアントが自動リトライし、アプリケーションもフォールバックを実装していると、1 回の失敗でフォールバックに入る前に複数の隠れたリクエストが発生し得ます。
ルーティングを明示するチュートリアルでは、再試行やモデル切り替えの判断をアプリケーション側に任せる方がクリーンです。
モデルルーティング方針はどう定義する？
ルーティングはプロンプトの外に出しましょう。
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6",}FALLBACK_MODELS = { "researcher": "gpt-5.6", "analyst": "gpt-5.6", "writer": "gemini-3.7-flash",}
これにより明確な設定境界が生まれます。
後からこのマッピングを以下に移すことも可能です。
- 環境設定
- YAML
- JSON
- データベース
- フィーチャーフラグ
- 社内のモデルルーティングサービス
エージェントのプロンプトを書き換えることなく実施できます。
3 つの CrewAI エージェントはどう構築する？
エージェントごとに 1 つの LLM オブジェクトを作成します。
from crewai import Agentdef build_agents(model_map: dict[str, str]): researcher = Agent( role="Market Researcher", goal="Collect the facts needed to answer the topic", backstory=( "You create concise, source-aware research briefs " "and clearly separate facts from assumptions." ), llm=cometapi_llm(model_map["researcher"]), max_iter=3, allow_delegation=False, ) analyst = Agent( role="Product Analyst", goal="Turn research into a defensible recommendation", backstory=( "You identify evidence, assumptions, risks, " "and trade-offs before making recommendations." ), llm=cometapi_llm(model_map["analyst"]), max_iter=3, allow_delegation=False, ) writer = Agent( role="Technical Writer", goal="Produce a concise technical decision memo", backstory=( "You write clear technical explanations " "without unnecessary marketing language." ), llm=cometapi_llm(model_map["writer"]), max_iter=3, allow_delegation=False, ) return researcher, analyst, writer
モデルの割り当ては、エージェントの役割定義と完全に独立しました。
これこそが、モデルルーティングを実用的にする理由です。
エージェントを逐次タスクでどうつなぐ？
3 つのタスクを作成します。
from crewai import Taskdef build_tasks(researcher, analyst, writer): research_task = Task( description=( "Research this topic: {topic}. " "Return the key facts, uncertainties, " "and relevant sources that the analyst should consider." ), expected_output=( "A compact research brief containing facts, " "uncertainties, and source references." ), agent=researcher, ) analysis_task = Task( description=( "Using the research brief, analyze {topic}. " "Identify the strongest conclusion and explain " "the major trade-offs." ), expected_output=( "A decision outline with evidence, " "assumptions, risks, and trade-offs." ), agent=analyst, context=[research_task], ) writing_task = Task( description=( "Write a concise technical decision memo about {topic}. " "State the recommendation early and preserve " "important caveats." ), expected_output="A polished technical decision memo in Markdown.", agent=writer, context=[research_task, analysis_task], ) return research_task, analysis_task, writing_task
依存チェーンは次のとおりです。
Topic ↓Research ↓Analysis ↓Final memo
アナリストはリサーチタスクの出力を受け取り、ライターはリサーチと分析の両方のコンテキストを受け取ります。
クルーはどう組み立てる？
エージェントとタスクを結合します。
from crewai import Crew, Processdef build_crew(model_map: dict[str, str]) -> Crew: researcher, analyst, writer = build_agents(model_map) research_task, analysis_task, writing_task = build_tasks( researcher, analyst, writer, ) return Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, verbose=True, )
これでモデルルーティングは完全に設定駆動になりました。
次の変更:
"researcher": "gemini-3.7-flash"
を別のサポートモデルに置き換えても、リサーチのプロンプトやタスク定義を変える必要はありません。
CrewAI のモデルフォールバックはどうあるべき？
本番志向の実装では、ここにより注意が必要です。
ありがちな誤りは次のとおりです。
Any error ↓Switch model
これは過剰です。
例えば、次のエラーでは通常フォールバックすべきではありません。
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
無効な API キーや不正なリクエストは、モデルを切り替えても修正できません。
フォールバックが適切なのは、次のような一時的な障害です。
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
したがってフォールバック方針は次のとおりです。
フォールバックや再試行は、有界な一時的障害に限定し、フォールバックモデルが同一のリクエスト契約を満たす場合に限る。
再試行可能なエラーはどう検出する？
OpenAI SDK のエラークラスを使えます。
from collections.abc import Iteratorfrom openai import ( APIConnectionError, APIStatusError, APITimeoutError,)def exception_chain(error: BaseException) -> Iterator[BaseException]: current: BaseException | None = error seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) yield current current = ( current.__cause__ or current.__context__ )def should_fallback(error: BaseException) -> bool: for current in exception_chain(error): if isinstance( current, (APIConnectionError, APITimeoutError), ): return True if isinstance(current, APIStatusError): return ( current.status_code in {408, 429} or current.status_code >= 500 ) return False
ここでは、408 と 429 を除く 400 台の構成エラーを意図的に除外しています。
クルー全体をリトライすべきか、失敗したエージェントだけを再実行すべきか？
フォールバック戦略には 2 つあります。
クルーレベルのフォールバック
最も単純な実装は次のとおりです。
Start crew ↓failure ↓change routing ↓run crew again
理解は容易ですが、完了済みタスクを再実行することがあります。
例:
Research → completedAnalysis → completedWriter → failed
kickoff() の全体リトライでは、次の実行が行われ得ます。
Research → againAnalysis → againWriter → fallback
この結果、以下が増加します。
- トークン使用量
- レイテンシ
- API コスト
- 副作用のリスク
タスクレベルのリカバリ
本番ワークフローでは、完了済みの作業にチェックポイントを残すべきです。
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
CrewAI は現在、チェックポイントによって実行状態を保存し、失敗後に再開できる機能を提供しています。ドキュメントのチェックポイント挙動では、完了済みタスクをスキップして保存状態から下流の作業を継続できます。
これは、コストが高い、あるいは副作用を伴うワークフローにとって優れたアーキテクチャです。
CrewAI のチェックポイント機能はどう追加する？
本番ワークフローでは、クルーでチェックポイントを有効化します。
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
CrewAI のチェックポイントシステムは、タスク完了後に実行状態を永続化し、チェックポイントからクルーを復元できます。
例えば、復元した実行では次のように使えます。
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
正確なチェックポイント設定は、プロジェクトで使用する CrewAI のバージョンに従ってください。
重要なアーキテクチャ上の要点は次のとおりです。
フォールバックよりも先にチェックポイント。
これにより、一時的なモデル障害によって高コストの完了済み作業が再実行される事態を防げます。
シンプルな制限付きフォールバックをどう実装する？
チュートリアルとして、クルーレベルの簡易フォールバックを示せます。
def run_with_fallback(topic: str): routes = [ PRIMARY_MODELS, { **PRIMARY_MODELS, "writer": FALLBACK_MODELS["writer"], }, { **PRIMARY_MODELS, "analyst": FALLBACK_MODELS["analyst"], "writer": FALLBACK_MODELS["writer"], }, ] last_error = None for attempt, model_map in enumerate(routes, start=1): try: crew = build_crew(model_map) result = crew.kickoff( inputs={"topic": topic} ) return result, model_map except Exception as error: last_error = error if not should_fallback(error): raise if attempt == len(routes): raise print( f"Transient failure on attempt {attempt}. " f"Trying bounded fallback route.", flush=True, ) raise RuntimeError( "Crew execution failed after all fallback routes." ) from last_error
ここで重要な区別があります。
これは、失敗したエージェントを特定したと主張しているわけではありません。
これはあくまで制限付きクルーレベルのフォールバック戦略です。
小規模でステートレスなワークフローでは許容できるかもしれません。本番ワークフローで高価なリサーチやツール、または副作用がある場合は、チェックポイントに基づく回復を使用してください。
CrewAI のトークン使用量はどう追跡する？
利用状況の追跡は、ルーティングレイヤーの一部として設計すべきです。
実行の最後に、CrewAI の結果を検査します。
result, selected_models = run_with_fallback(topic)print("Selected models:")print(selected_models)print("Final result:")print(result.raw)print("Usage:")print(result.token_usage)
利用可能な使用量フィールドは、CrewAI のバージョンや実行パスに依存する場合があるため、デプロイするバージョンにおける返却オブジェクトを信頼できる唯一の情報源としてください。
本番の利用記録には理想的には以下を含めます。
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
これにより、次のような問いに答えられます。
どのエージェントが予算を最も消費しているか？
アナリストはどれくらいの頻度でフォールバックしているか？
どのモデルが最もレイテンシが高いか？
各ワークフローはいくらかかっているか？
エージェント単位でコストをどう制御する？
マルチモデルルーティングは、実際のワークロード差を反映するほど有用です。
例:
Researcher→ 高ボリューム→ 低コストモデルAnalyst→ 低ボリューム→ 強力な推論モデルWriter→ 中ボリューム→ 汎用の本番モデル
また、エージェント設定でもコストを制約できます。
例:
max_iter=3
はエージェントの反復ループに上限を設けます。これは API コール数やトークン予算が厳密に 3 に限定されるという意味ではありません。
その他の制御には以下が含まれます。
- タスクコンテキストの制限
- 中間出力の要約
- 再利用可能なリサーチのキャッシュ
- 最大入力サイズの制限
- サポートされる範囲で最大出力トークンの制限
- ツール呼び出しの制限
- ユーザー単位の予算設定
- ワークフロー単位の予算設定
- フォールバック頻度の追跡
デプロイ前にモデルをどう検証する？
Model ID を固定値で書き続けないでください。
モデルは次のようになり得ます。
- 利用不可
- 改名
- 非推奨
- 制限付き
- 能力の変更
- 価格の変更
- アプリが使用するパラメータと非互換
CometAPI はプログラムから問い合わせ可能なモデルカタログエンドポイントを提供し、公開のモデルディレクトリは人間がモデルを探索するのに利用できます。
デプロイ前チェックは次のように実行できます。
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
その後、構成済みの Model ID がデプロイ前に存在することを検証します。
例えば、CI プロセスで次を検証できます。
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
可用性チェックはアプリのテストの代替ではありません。モデルがカタログに存在しても、CrewAI エージェントが使用するすべてのパラメータ、ツール、出力形式がサポートされるとは限りません。
完全な CrewAI の例はどうなる？
統合実装は以下のとおりです。
import jsonimport osimport sysfrom collections.abc import Iteratorfrom dotenv import load_dotenvfrom openai import ( APIConnectionError, APIStatusError, APITimeoutError,)from crewai import Agent, Crew, LLM, Process, Taskload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv( "COMETAPI_BASE_URL", "https://api.cometapi.com/v1",)PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6",}FALLBACK_MODELS = { "researcher": "gpt-5.6", "analyst": "gpt-5.6", "writer": "gemini-3.7-flash",}def cometapi_llm(model_id: str) -> LLM: return LLM( model=model_id, base_url=COMETAPI_BASE_URL, api_key=COMETAPI_KEY, timeout=60.0, max_retries=0, )def build_crew(model_map: dict[str, str]) -> Crew: researcher = Agent( role="Market Researcher", goal="Collect the facts needed to answer the topic", backstory=( "You create concise, source-aware research briefs " "and distinguish facts from assumptions." ), llm=cometapi_llm(model_map["researcher"]), max_iter=3, allow_delegation=False, ) analyst = Agent( role="Product Analyst", goal="Turn research into a defensible recommendation", backstory=( "You evaluate evidence, assumptions, risks, " "and trade-offs." ), llm=cometapi_llm(model_map["analyst"]), max_iter=3, allow_delegation=False, ) writer = Agent( role="Technical Writer", goal="Produce a concise technical decision memo", backstory=( "You write clear technical explanations " "without unnecessary hype." ), llm=cometapi_llm(model_map["writer"]), max_iter=3, allow_delegation=False, ) research_task = Task( description=( "Research this topic: {topic}. " "Return the key facts, uncertainties, " "and relevant sources." ), expected_output=( "A concise research brief with facts " "and open questions." ), agent=researcher, ) analysis_task = Task( description=( "Using the research brief, analyze {topic}. " "Identify the strongest conclusion and " "explain the major trade-offs." ), expected_output=( "A decision outline with evidence, " "assumptions, risks, and trade-offs." ), agent=analyst, context=[research_task], ) writing_task = Task( description=( "Write a concise technical decision memo " "about {topic}. State the recommendation early " "and preserve important caveats." ), expected_output=( "A polished technical decision memo in Markdown." ), agent=writer, context=[ research_task, analysis_task, ], ) return Crew( agents=[ researcher, analyst, writer, ], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, verbose=True, )def exception_chain( error: BaseException,) -> Iterator[BaseException]: current = error seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) yield current current = ( current.__cause__ or current.__context__ )def should_fallback(error: BaseException) -> bool: for current in exception_chain(error): if isinstance( current, ( APIConnectionError, APITimeoutError, ), ): return True if isinstance(current, APIStatusError): return ( current.status_code in {408, 429} or current.status_code >= 500 ) return Falsedef run_with_fallback(topic: str): routes = [ PRIMARY_MODELS, { **PRIMARY_MODELS, "writer": FALLBACK_MODELS["writer"], }, { **PRIMARY_MODELS, "analyst": FALLBACK_MODELS["analyst"], "writer": FALLBACK_MODELS["writer"], }, ] last_error = None for attempt, model_map in enumerate( routes, start=1, ): try: crew = build_crew(model_map) result = crew.kickoff( inputs={ "topic": topic, } ) return result, model_map except Exception as error: last_error = error if not should_fallback(error): raise if attempt == len(routes): raise print( f"Transient failure on attempt " f"{attempt}; trying fallback.", file=sys.stderr, ) raise RuntimeError( "No model route completed the crew." ) from last_errordef main(): topic = ( sys.argv[1] if len(sys.argv) > 1 else ( "Should a small SaaS add " "AI-generated meeting summaries?" ) ) result, selected_models = ( run_with_fallback(topic) ) output = { "selected_models": selected_models, "raw": result.raw, "tasks_output": [ task.raw for task in result.tasks_output ], "token_usage": str( result.token_usage ), } print( json.dumps( output, indent=2, default=str, ) )if __name__ == "__main__": main()
元の版に対する重要な改善点は、例外が「どのエージェントが失敗したか」を特定すると誤って示唆しないことです。
これは明示的に制限付きのクルーレベル・フォールバック実装です。
本番では、同じルーティング方針を CrewAI のチェックポイントと組み合わせてください。
CrewAI ワークフローはどう実行する？
ファイル名を次のように保存します。
crewai_multi_model.py
次のように実行します。
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
成功時のレスポンスは次のような情報を含みます。
{ "selected_models": { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6" }, "raw": "<final decision memo>", "tasks_output": [ "<research output>", "<analysis output>", "<writing output>" ], "token_usage": "<usage information>"}
正確なレスポンスや使用量は、入力、モデルの挙動、CrewAI のバージョン、実行パスに依存します。
再試行可能エラーでフォールバック経路が有効になると、selected_models にそのクルー実行で使用されたルートが反映されます。
本番のモデルルーティングはどう設計すべき？
本番のルーティング方針は、モデル品質だけでなく多面的に検討すべきです。
有用な評価関数は次のとおりです。
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
これを複数のレベルで実装できます。
コストベースのルーティング
単純タスク → 経済的モデル複雑タスク → プレミアムモデル
レイテンシベースのルーティング
対話的リクエスト → 高速モデルバッチ/バックグラウンド → 高品質モデル
信頼性ベースのルーティング
プライマリモデル ↓一時的障害 ↓フォールバックモデル
タスクベースのルーティング
リサーチ → Model A分析 → Model Bライティング → Model Cコード → Model D
CrewAI では、もともと各エージェントが異なる役割を持つため、この最後のアプローチが特に自然です。
フォールバックを安全にするには？
堅牢なフォールバックシステムは 4 つのルールを徹底すべきです。
認証エラーではフォールバックしない
API キーが無効な場合:
401
モデルを変えても解決しません。
不正リクエストではフォールバックしない
リクエストが無効な場合:
400422
モデルを変えるのではなくリクエストを修正してください。
無制限にフォールバックしない
ハードな上限を設定します。
MAX_FALLBACK_ATTEMPTS = 2
上限のないフォールバックは高コストなリトライループになります。
フォールバックモデルのリクエスト互換性を確保する
フォールバックモデルは、エージェントが必要とする機能をサポートしている必要があります。
例えば、プライマリエージェントが特定のツールや構造化出力を必要とする場合、フォールバックモデルも同じ契約を満たさなければなりません。
OpenAI 互換であっても、機能互換を意味しません。
CrewAI + CometAPI でよくあるエラーは？
| 症状 | 想定原因 | 対処 |
|---|---|---|
| 401 Unauthorized | API キーが無効/不足 | COMETAPI_KEY を確認。フォールバックしない |
| 400 Bad Request | リクエストパラメータが無効 | リクエストを修正 |
| 404 Model Not Found | 古い Model ID | 現行のモデルカタログを確認 |
| 408 Timeout | 一時的なタイムアウト | 有界なポリシーで再試行 |
| 429 Rate Limited | リクエスト過多 | バックオフして再試行 |
| 500–504 | 一時的なサーバ/ゲートウェイ障害 | 制限付きフォールバック |
| エージェントが繰り返しリトライする | SDK の隠れた自動リトライ | max_retries を制御 |
| 完了済みタスクが再実行される | クルー全体のリトライ | チェックポイントに基づく回復 |
| モデルごとの挙動差 | モデル能力の差異 | 各モデルを独立にテスト |
| CrewAI のコンストラクタで予期せぬエラー | バージョン不一致 | CrewAI のバージョンを固定・検証 |
CrewAI のエラーとモデルのエラーをどう切り分ける？
デバッグ時に重要な区別です。
構成エラー
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
これらは速やかに失敗すべきです。
プロバイダ/API エラー
401403404429500503
ステータスに応じて異なるハンドリングが必要です。
アプリケーションエラー
Agent output invalidTool returned malformed dataTask context missingSide effect failed
モデルを変えても解決しない場合があります。
成熟したエージェントシステムでは、次を個別に扱うべきです。
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
これは汎用的な
except Exception: use_fallback()
よりはるかに安全です。
外部の副作用をどう保護する？
エージェントがテキスト生成以外のことを行うと、フォールバックは格段に複雑になります。
例えば、あるエージェントが:
- データベースレコードを作成
- メール送信
- 外部 API を呼び出し
- CRM を更新
し、外部アクション成功後にモデルがタイムアウトした場合、クルー全体を再実行すると操作が重複する恐れがあります。
以下を使ってください。
- 冪等性キー
- タスクのチェックポイント
- トランザクション境界
- 実行 ID
- 耐久的なタスク状態
- 明示的な副作用の確認
例:
job_id = crew_run_123task_id = writer_456
これらの識別子を外部操作に保存し、リトライ時にすでに操作が行われたかどうかを判定できるようにします。
マルチモデルの CrewAI ワークフローをどう監視する？
最低限、次をログに記録します。
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
次はログに記録しないでください。
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
モデルごとに監視すべき指標は以下です。
信頼性
success ratetimeout rate5xx ratefallback rate
パフォーマンス
p50 latencyp95 latencyp99 latency
コスト
input tokensoutput tokenscost per taskcost per completed workflow
品質
task success ratehuman evaluationstructured-output validitytool-call success
これにより、モデルルーティングをハードコードの選好から、可観測なエンジニアリングシステムへと変えられます。
直接プロバイダ API と CometAPI、どちらを選ぶべき？
アーキテクチャに依存します。
| アーキテクチャ | 資格情報 | モデル切替 | プロバイダ統合負荷 | ルーティングの一元化 |
|---|---|---|---|---|
| 直接プロバイダ API | 複数 | カスタム | 高 | なし |
| 単一プロバイダ | 1 つ | 限定的 | 低 | 限定的 |
| CrewAI + CometAPI | CometAPI 資格情報 1 つ | Model ID ベース | 低 | あり |
アプリが 1 つのプロバイダとそのネイティブ機能だけで十分な場合、直接統合は合理的です。
CrewAI アプリで複数プロバイダのモデルが必要で、アクセスレイヤーを 1 つにしたい場合は、CometAPI が有力候補になります。
重要なのは、CometAPI が CrewAI の代替ではないという点です。
代わりに次のような関係です。
CrewAIAgent orchestration ↓CometAPIModel access ↓Multiple models
各レイヤーは異なる責務を持ちます。
このアーキテクチャはどうスケールする？
ルーティング方針をエージェント定義から切り離すと、モデルを追加してもアプリ全体を作り直す必要がありません。
例:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
同じアーキテクチャで次をサポートできます。
リサーチエージェント分析エージェントコーディングエージェントレビューエージェントライティングエージェントファクトチェックエージェント
各エージェントは異なるモデルを持ちながら、同じ CometAPI アクセスレイヤーを共有できます。
次のステップはルーティングの動的化です。
固定の
"analyst": "claude-opus-5"
ではなく、最終的には次のようにできます。
select_model( task="analysis", budget=budget, latency_target=latency_target,)
すると、ルーティングシステムはアプリ要件に基づき、承認済みモデルから選択できます。
CrewAI + CometAPI の本番向けベストアーキテクチャは？
小規模ワークフローでは:
User Input ↓CrewAI ↓CometAPI ↓Models
本番では:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
本番の重要コンポーネントは次の通りです。
- モデル許可リスト
- エージェント別ルーティング
- 制限付きリトライ
- タスクのチェックポイント
- 利用状況トラッキング
- コスト制御
- モデル互換性テスト
- 可観測性
- 副作用の冪等化
これは、crew.kickoff() を単に try/except で包むだけの実装より、はるかに堅牢です。
1 つの CometAPI キー、異なるモデル、明確なエージェント役割
CrewAI と CometAPI を組み合わせる最も有用な考え方は、補完的な 2 つのレイヤーとして捉えることです。
CrewAI はエージェントが何をするかを定義する。
CometAPI はエージェントがどうモデルにアクセスするかを定義する。
この分離により、高ボリュームのリサーチエージェントには高速モデル、分析エージェントには強力な推論モデル、最終ライターには汎用のモデルを割り当てることができ、ワークフロー内部で個別のプロバイダ統合を維持する必要がなくなります。
最も簡単な実装は、1 つの CometAPI キーと 1 つの OpenAI 互換 base URL の利用です。
https://api.cometapi.com/v1
本番では、アーキテクチャをもう一歩進めてください。モデルルーティングは設定に置き、デプロイ前にモデル可用性を検証し、一時的な障害に対してのみ制限付きフォールバックを使い、完了済みタスクにはチェックポイントを取り、各実行でモデルと使用メタデータを記録します。
これは、CrewAI を 1 つの LLM に接続するだけの単純なパターンより、はるかに堅牢です。
CrewAI はエージェントをオーケストレーション。CometAPI はモデルアクセスを集約。Model ID がルーティングを制御。チェックポイントが完了済み作業を保護。利用トラッキングがコストを管理。
よくある質問
CrewAI は同じクルー内で複数の AI モデルを使えますか？
はい。各 CrewAI エージェントに異なる LLM 設定を割り当てられます。各設定は同じ CometAPI の API キーと base URL を使いつつ、独自のモデルを指定できます。
CrewAI は OpenAI 互換 API に接続できますか？
はい。CrewAI の LLM 設定は、OpenAI 互換エンドポイントに対するカスタム base_url と API キーをサポートします。
CometAPI では base URL は次のとおりです。
https://api.cometapi.com/v1
GPT、Claude、Gemini に別々の API キーは必要ですか？
CometAPI を経由してこれらのモデルへアクセスする場合、アプリケーションは CometAPI の資格情報とエンドポイントを使い、各 CrewAI エージェントで個別のプロバイダ資格情報を実装する必要はありません。
1 つの API キーでモデルの能力は同一になりますか？
いいえ。API インターフェースは統一されても、モデルの能力は異なります。コンテキストウィンドウ、ツールサポート、パラメータ、出力挙動、レイテンシ、価格はモデルごとに異なります。
1 つのモデルが失敗したら CrewAI ワークフロー全体をリトライすべき？
単純でステートレスなワークフローに限ります。クルー全体の再実行は、完了済みタスクの再実行やコスト増につながります。本番では、完了済みタスクをチェックポイントし、可能であれば失敗箇所から再開してください。
CrewAI の現行チェックポイント機能は、実行状態を保持し、失敗後に再開するために設計されています。
CrewAI の例外すべてでモデルフォールバックすべき？
いいえ。認証、リクエスト不正、無効な Model ID、未サポートパラメータは、モデル変更ではなく構成の修正が必要です。
フォールバックは、タイムアウトやレート制限、一時的な 5xx 応答などの一時的障害に限定するのが適切です。
各 CrewAI エージェントのコストはどう追跡する？
エージェント名、モデル ID、トークン使用量、レイテンシ、実行ステータス、フォールバック情報を各タスクごとに記録してください。これらのデータでエージェント別およびワークフロー別のコストを算出できます。
エージェントに割り当てるモデルを動的に変更できますか？
はい。Model ID はルーティング設定に保持し、エージェント定義に直接埋め込まないでください。アプリケーションは、コスト、レイテンシ、タスク種別、可用性に基づいてモデルを選択できます。
CometAPI は CrewAI の代替ですか？
いいえ。異なるレイヤーで動作します。CrewAI はエージェントとタスクのオーケストレーションを行い、CometAPI は統一されたモデルアクセスレイヤーを提供します。
CometAPI の現行モデルはどこで見つかる？
CometAPI のモデルディレクトリで人間がモデルを探索でき、モデル API でプログラム的な検証が可能です。CometAPI の現行クイックスタートでは、テキスト、画像、動画、音声を含む 500+ のモデルが掲載されています。