CrewAI Workflow Telemetry: 境界から検証された行まで
制御されたアプリケーション境界で CrewAI Workflow Telemetry を使用し、イベント コントラクトを小さく保ち、集約ビューを構築する前に既知の結果を検証します。
- 1
結果を選択してください
マルチエージェントワークフローの可観測性
- 2
契約を定義する
run_id、crew_name、process_type、task_count、agent_count、およびリリース
- 3
境界を計測する
すべての内部メッセージを個別のビジネス イベントとして記録するのではなく、Crew.kickoff または所有する Flow メソッドをラップし、1 つの論理ワークフロー結果を出力します。
- 4
証拠を検証する
Exercise a known fixture, then inspect agent_workflow_completed for one correctly typed terminal row.
始める前に
前提条件と境界
- CrewAI と telemetry-sh は信頼できる Python ランタイムで初期化されました
- 安定したスタッフ、プロセス、タスクカテゴリ、および最終結果の名前
- タスクの入力、エージェントの出力、メモリ、知識、ツールのペイロードをカバーするコンテンツ ポリシー
配信設定
サーバー側のインストールと初期化
サーバー側キーを使用して、プロセスごとに 1 回、同期コードの場合は Telemetry を、非同期コードの場合は TelemetryAsync を初期化します。 ブラウザ バンドル、クライアントに表示される環境変数、ソース管理、ログ、および例外メッセージに取り込み資格情報が含まれないようにします。
pipのインストール
python -m pip install telemetry-sh- 1制限されたネットワーク動作を持つ再利用可能なサーバー側配信クライアントを 1 つ準備します。
- 2成功、失敗、再試行、またはタイムアウトの境界に結果イベントを追加します。
- 3アラートを有効にする前に、制御されたフィクスチャを送信し、保存されている行を検査します。
スニペット
1 つの構造化されたイベントから始める
ワークフローが完了、失敗、または再試行される場所にこの図形を追加します。次に、実際のフィールドからダッシュボードを構築します。
CrewAI Workflow Telemetryイベント
from time import perf_counter
from uuid import uuid4
def run_research_crew(inputs):
run_id = str(uuid4())
started_at = perf_counter()
status = "success"
error_type = None
try:
return research_crew.kickoff(inputs=inputs)
except Exception as error:
status = "failed"
error_type = classify_crew_error(error)
raise
finally:
telemetry.log("agent_workflow_completed", {
"run_id": run_id,
"crew_name": "research_crew",
"process_type": "sequential",
"status": status,
"error_type": error_type,
"duration_ms": round((perf_counter() - started_at) * 1000),
"release": APP_RELEASE,
})イベント契約
run_id、crew_name、process_type、task_count、agent_count、およびリリース
ステータス、duration_ms、retry_count、handoff_count、および error_type
承認済み、reviewer_outcome、承認された場合は estimated_cost_usd、および環境
実装のチェックポイント
チェックポイント 1
すべての内部メッセージを個別のビジネス イベントとして記録するのではなく、Crew.kickoff または所有する Flow メソッドをラップし、1 つの論理ワークフロー結果を出力します。
チェックポイント 2
ステップレベルの診断が必要な場合は、CrewAI 独自のトレースまたはコールバックを使用し、承認された実行 ID と関連付けます。
チェックポイント 3
アラートを定義する前に、部分的なタスクの失敗、委任ループ、人間による入力、キャンセル、および非同期キックオフ動作を実行します。
検証
イベントが到着したことを証明する
既知の成功例と失敗例を実行した後、これを実行します。最終的なイベント コントラクトがスニペットと異なる場合は、フォールバック テーブル名を置き換えます。
CrewAI Workflow Telemetry 検証クエリ
SELECT *
FROM agent_workflow_completed
ORDER BY timestamp_utc DESC
LIMIT 20;実装の参考資料
新しい運用パスを有効にする前に、イベント契約、データ安全性に関するガイダンス、上流の主要ドキュメントを確認してください。
生産境界
結果イベントを小さく、回復可能なものに保つ
このパターンが提供するのは、
- 上流のワークフローの横にある、制限付きの SQL 対応の結果。
- ダッシュボード、アラート、およびイベント間の相関関係の安定したフィールド。
- 成功、失敗、再試行、タイムアウトの動作を検証するためのフィクスチャ駆動のパス。
このパターンでは提供されません
- OTLP エクスポーター、自動収集パイプライン、または詳細なトレースと診断ログの代替。
- ペイロードにイベント ID が含まれているという理由だけで、1 回だけ配信されます。
- 生のプロバイダー ペイロード、ユーザー コンテンツ、資格情報、または規制されたデータを収集する許可。
イベントスキーマの開始点
このワークフローのイベント コントラクト
クエリまたはスニペットを運用環境に適用する前に、行粒度、出力境界、必要なタイプ、プライバシー クラス、サンプル ペイロード、および検証チェックリストを確認してください。
関連製品の機能
このワークフローを続行します AIエージェントの監視
レビュー可能な SQL を使用して、エージェントの実行、ツールの使用、モデルのコスト、品質、製品の結果を結び付けます。
関連する SQL レシピ
SQL で次の質問に答えてください
このワークフローの構造化フィールドに対してクエリを実行し、結果の例を検査して、有用な回答をダッシュボードまたはアラートに変換します。
AI エージェントのタスクの成功と人間による引き継ぎを測定する
どのエージェントのワークフローが正常に終了し、受け入れられた結果をもたらしますか?
レシピを開くネストされた AI ツール呼び出しイベントのクエリ
最も失敗した呼び出しに関連する AI ツールと引数はどれですか?
レシピを開く繰り返しを検出する AI エージェント ツール Loops
どのエージェントの実行が繰り返しのツール ループに陥っているように見えますか?
レシピを開く相関のあるワークフローのタイムラインを再構築する
最近失敗したワークフロー中に、順番に何が起こったのでしょうか?
レシピを開く1 ドルあたりの受け入れられた AI 出力を測定する
どのモデルと機能の組み合わせが、1 ドルあたり最も受け入れられる出力を生成しますか?
レシピを開く実装ファミリーごとに参照する
関連する統合パターンを比較する
この統合と組み合わせるテンプレート
さらなる統合
クロード エージェント SDK Telemetry
プロンプトやツールのペイロードを保存せずに、クロード エージェント SDK の実行結果、期間、ツール アクティビティ、ターン、コスト、および承認された製品シグナルを追跡します。
オープンガイドOpenAI エージェント Telemetry
OpenAI エージェントの実行、ツール呼び出し、モデルの使用状況、レイテンシー、コスト、および構造化イベントによる最終結果をログに記録します。
オープンガイドLangChain エージェントの結果 Telemetry
生のトレースを複製することなく、LangChain エージェントの実行とツールの結果を製品に関わるコスト、信頼性、ハンドオフ、および受け入れメトリクスに結び付けます。
オープンガイド