Pipeline error log ================== 各パイプラインにはエラーログテーブルがあります。レコードが任意のステップで失敗すると、ログは、ソースレコードID、失敗したステップ、操作タイプ、およびエラーメッセージをキャプチャします。バッチの残りの部分は処理を継続します。 その仕組み ---------- - レコードに失敗すると、ソースID、失敗したステップ番号、操作タイプ、エラーメッセージとともに\ ``{source_schema}.pipeline_{pipeline_name}_errors`` ログに記録されます。 - 単一の不良レコードはパイプラインをブロックしません。バッチ内の他のレコードは、残りのすべてのステップを続行します。 - エラーログテーブルはパイプラインで作成され、パイプラインが削除されるときに削除されます。 7.2.0より前に作成されたパイプラインは、アップグレード中にテーブルを取得します。 - 書き込みは失敗したステップのトランザクション内で発生するため、成功したステップが孤立したログ行を残すことはなく、障害がレコードを失うことはありません。 - ログに記録された各エラーは、 Postgres ``WARNING`` も生成します。 :ref:`promoteb.pipeline_error_warnings ` で切り替えます ;エラーは関係なくログに残ります。 - queryおよびsummaryファンクションは、エラーログテーブルが欠落しているパイプラインに対して空の結果を返します。 - エラーログテーブルを手動で削除しないでください。 7.4.0の時点で、ソース\ ``DELETE`` ハンドラーとステップ後クリーンアップは、テーブルが見つからない場合にエラーを発生させます。 - エラーは、4つの\ ``error_category`` 値のいずれかを取得します :ref:`helpb.ErrorBlocking ` を参照してください。 .. csv-table:: :header: Category,Scope,Blocks pipeline?,Description :widths: 15,12,12,20 :align: left :class: longtable `RecordTemporary`,Record,No,単一レコードの一時的な障害。再試行で成功する場合があります。 `RecordPermanent`,Record,No,単一のレコードの永続的な障害たとえば、破損したデータ。 `PipelineTemporary`,Pipeline,Yes,ステップ全体を停止した一時的な障害たとえば、サービスの停止。再試行で解決される場合があります。 `PipelinePermanent`,Pipeline,Yes,ステップ全体を停止した永続的な障害たとえば、無効なモデル構成。 レコードレベルのエラーはパイプラインをブロックしません。他のレコードは処理を続けます。パイプラインレベルのエラーは、ステップ全体をブロックします。一時的なエラーは、再試行すると解決される場合があります。永続的なエラーは、不正なデータまたは構成ミスを示しています。具体的には ``RecordTemporary`` は現在モデルレート制限エラーの場合にのみ起動し、\ ``RecordPermanent`` は不正な入力データたとえば、破損したPDF、不正なHTML、無効なUTF-8をカバーし、\ ``PipelineTemporary`` は一時的な外部障害たとえば、オブジェクトストアの停止、モデルサービスタイムアウトをカバーします)、\ ``PipelinePermanent`` は、一般的な構成の誤りたとえば、削除されたモデル、無効なステップ構成などです。 エラーのクエリー ---------------- ``aidb.get_error_logs()`` を使用して、パイプラインのエラーログを検査します。 .. code:: sql SELECT * FROM aidb.get_error_logs(my_pipeline); __OUTPUT__ id | source_id | part_ids | pipeline_step | step_operation | error_message | error_category | failed_at | retry_count | last_retry_at - ---+-----------+----------+---------------+----------------+----------------------------------------+-------------------+----------------------------+-------------+--------------- 3 | doc_99 | {0,2} | 2 | ChunkText | text segment exceeded maximum length | RecordPermanent | 2026-04-10 14:32:01.123+00 | 0 | 2 | doc_42 | {0} | 1 | ParsePdf | failed to extract text from page 0 | RecordPermanent | 2026-04-10 14:31:58.456+00 | 0 | 1 | doc_7 | | 1 | ParsePdf | corrupt PDF header | RecordPermanent | 2026-04-10 14:31:55.789+00 | 0 | (3 rows) ソースレコード、ステップ番号、またはエラーカテゴリでフィルタリングします。 .. code:: sql SELECT * FROM aidb.get_error_logs( my_pipeline, p_source_id => doc_42 ); .. code:: sql SELECT * FROM aidb.get_error_logs( my_pipeline, p_pipeline_step => 1::SMALLINT, p_error_category => RecordPermanent ); ``p_limit`` および\ ``p_offset`` を使用したページ付け .. code:: sql SELECT * FROM aidb.get_error_logs(my_pipeline, p_limit => 20, p_offset => 40); 完全なパラメーターと戻り型の詳細については、 :ref:`Operations reference ` を参照してください。 エラーの概要 ------------ 1つのパイプラインのステップ、オペレーション、およびカテゴリ別にグループ化されたエラー数を取得します。 .. code:: sql SELECT * FROM aidb.get_error_log_summary(my_pipeline); __OUTPUT__ pipeline_step | step_operation | error_category | error_count | latest_failed_at - --------------+----------------+-------------------+-------------+---------------------------- 1 | ParsePdf | RecordPermanent | 2 | 2026-04-10 14:31:58.456+00 2 | ChunkText | RecordPermanent | 1 | 2026-04-10 14:32:01.123+00 (2 rows) または、すべてのパイプラインで同じロールアップを一度に取得します。 .. code:: sql SELECT * FROM aidb.get_all_error_summaries(); __OUTPUT__ pipeline_name | pipeline_step | step_operation | error_category | error_count | latest_failed_at - --------------+---------------+----------------+-------------------+-------------+---------------------------- my_pipeline | 1 | ParsePdf | RecordPermanent | 2 | 2026-04-10 14:31:58.456+00 my_pipeline | 2 | ChunkText | RecordPermanent | 1 | 2026-04-10 14:32:01.123+00 pdf_ingester | 1 | ParsePdf | PipelinePermanent | 1 | 2026-04-10 15:00:12.000+00 (3 rows) 完全なパラメーターと戻り型の詳細については、 :ref:`Operations reference ` および :ref:`promoteb.get_all_error_summaries ` を参照してください。 失敗したレコードの再キューイング -------------------------------- 7. 5.0の新機能。 ``aidb.requeue_pipeline_errors()`` は、ターゲットのレコードレベルのエラー行を削除し、パイプラインのステートサーフェスでソースレコードをダーティとしてマークします。次の通常の実行で再処理されます。呼び出し自分自身は処理を行いません。 .. code:: sql SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[1, 2, 99]); __OUTPUT__ error_id | action - ---------+------------------------ 1 | requeued 2 | requeued 99 | not_found (3 rows) 結果は\ ``error_id`` の昇順で並べられます。各行は、3つのアクションのいずれかをレポートします。 - ``requeued`` レコードレベルの行が削除され、ソースがダーティとしてマークされました。テーブルソースは状態イベントをエンキューします。ボリュームソースは、状態行にワンショットの\ ``retry_requested_at`` マーカーを設定します。 - ``skipped_pipeline_level`` エントリーにはNULL ``source_id`` 以下を参照してください、そして適切な場所に残されます。 - ``not_found`` IDがエラーログになかったか、パイプラインにエラーログテーブルがありません古いパイプラインまたはSemantic KBのような\ ``Empty`` ソース。 バックグラウンドパイプラインは、次のワーカーティックに再キューイングされた行を自動的にドレインします。無効またはライブパイプラインを実際に再処理するには、フォローアップの\ ``aidb.run_pipeline('my_pipeline')`` が必要です。 リキューイングは遅延マークダーティ操作であり、同期再試行ではありません。呼び出し自分自身は処理を行わないため、レコードごとの成功または失敗は、フォローアップの実行後にのみ表面化します。再処理後に何が起こったのかを確認するには、 ``aidb.get_error_logs`` を再照会します。 重複するIDのコンカレント・リキューは、行ごとの\ ``FOR UPDATE`` ロックでシリアル化されます。重複しないIDセットは並列的に処理されます。 Requeueは、 ``aidb.run_pipeline`` またはバックグラウンドワーカーをブロックしません。 パイプラインレベルのエラー ^^^^^^^^^^^^^^^^^^^^^^^^^^ パイプラインレベルのエラーNULL ``source_id`` は、特定のレコードに関連付けられていないバッチ障害を記録します。レコードごとに再度キューイングすることはできないため、 ``aidb.requeue_pipeline_errors`` は\ ``action='skipped_pipeline_level'`` を返し、ログにエントリを残します。 .. code:: sql SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[7]); __OUTPUT__ error_id | action - ---------+------------------------ 7 | skipped_pipeline_level (1 row) 根本的な問題たとえば、モデルの可用性、構成、削除されたモデルなどを修正し、パイプラインを再実行します。次のステップがエラーなしで実行されると、NULL行は自動的にクリアされます。再実行せずに終了するには、 ``aidb.clear_error_logs`` を使用します。 無効モードのボリュームに関する注意事項 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ボリュームソースを使用する無効モードのパイプラインでは、 ``aidb.requeue_pipeline_errors`` はマークされたファイルに\ ``retry_requested_at`` マーカーを設定しますが、ドレインをスケジュールできないため、 ``aidb.run_pipeline`` を指す\ ``NOTICE`` を発行します。手動\ ``aidb.run_pipeline`` は、マークされたファイルのみではなくボリューム全体を再読み取ります。これにより、大規模なボリュームのフルボリュームスキャンがコストがかかります。 大規模な無効モードボリュームの場合、パイプラインをバックグラウンドモードに切り替え、\ ``aidb.requeue_pipeline_errors`` を呼び出し、ワーカーにマークされたファイルをドレインさせて、パイプラインを無効に戻します。 完全なパラメーターと戻り型の詳細については、 :ref:`Operations reference ` を参照してください。 エラーのクリア -------------- IDで特定のエラーログエントリを削除します。 .. code:: sql SELECT aidb.clear_error_logs(my_pipeline, ARRAY[1, 3]); __OUTPUT__ clear_error_logs - ----------------- 2 (1 row) 削除された行数を返します。決済はIDベースです。 ``get_error_logs()`` を使用してエントリを削除する前に確認してください。 完全なパラメーターと戻り型の詳細については、 :ref:`Operations reference ` を参照してください。 自動クリーンアップ ------------------ エラーログエントリは、3つの場合、\ ``aidb.clear_error_logs`` または\ ``aidb.requeue_pipeline_errors`` なしでクリアされます。 - ソースレコードが削除されました。ライブ\ ``DELETE`` トリガー、バックグラウンドワーカー\ ``DELETE`` 処理、およびボリュームの古い状態のクリーンアップはすべて、宛先行とともに一致するエラーログエントリを削除します。 - ソースエンベロープは、後の実行で宛先に到達します。パイプラインエグゼキューターは、\ ``(source_id, part_ids)`` によってスコープされるエントリをクリアするため、部分的粒度の成功によって兄弟エラーがワイプされません。 - そのステップが次に正常に実行された後、ステップのパイプラインレベルの行NULL ``source_id`` 。その実行中に同時に失敗したエグゼキューターによって書き込まれたエラーは保存されます。 パイプラインを削除すると、エラーログテーブル全体が削除されます。 パイプラインステータスの統合 ---------------------------- エラー数は :ref:`promoteb.pipeline_metrics ` ビューにも表示されます。パイプラインステータスはエラーの重大度を反映します。 - ``PartialErrors`` 一部のレコードは失敗しました。残りは通常に処理されました。 - ``BlockingErrors`` パイプラインレベルのエラーにより、ステップが完全に停止しました。 パイプラインの状態のモニタリングの詳細については、 :ref:`モニタリングとオブザーバビリティ <モニタリングとオブザーバビリティ>` を参照してください。 エンドツーエンドの例 -------------------- 3つのドキュメントソースを使用した短いウォークスルー。1つの行には破損したPDFがあり、1つは解析しますが、特大のチャンクを生成し、3番目は埋め込みステップでモデルレート制限に達します。 パイプラインを実行します。 3つの障害はいずれもバッチを中止しません。次に、結果のエラーを検査します。 .. code:: sql SELECT aidb.run_pipeline(my_pipeline); SELECT id, source_id, error_category FROM aidb.get_error_logs(my_pipeline); __OUTPUT__ id | source_id | error_category - ---+-----------+----------------- 3 | doc_200 | RecordTemporary 2 | doc_42 | RecordPermanent 1 | doc_7 | RecordPermanent (3 rows) パイプラインは\ ``PartialErrors`` を報告し、数を表示します。 .. code:: sql SELECT pipeline, "Status", "count(record errors)", "count(blocking errors)" FROM aidb.pipeline_metrics WHERE pipeline = my_pipeline; __OUTPUT__ pipeline | Status | count(record errors) | count(blocking errors) - -------------+---------------+----------------------+------------------------ my_pipeline | PartialErrors | 3 | 0 (1 row) 上流のレート制限の問題を修正した後、レート制限された行を再度キューに入れます。これは、ソースがダーティであることを示します。自分自身は再処理しません。 .. code:: sql SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[3]); __OUTPUT__ error_id | action - ---------+---------- 3 | requeued (1 row) 再度キューに入れられた行をドレインします。バックグラウンドパイプラインがこれを自動的に取得します。無効またはライブパイプラインには明示的な呼び出しが必要です。トリアージ後に2つの永続的な障害を無視します。 .. code:: sql SELECT aidb.run_pipeline(my_pipeline); SELECT aidb.clear_error_logs(my_pipeline, ARRAY[1, 2]); __OUTPUT__ clear_error_logs - ----------------- 2 (1 row) ステータスは\ ``UpToDate`` に戻ります。 .. code:: sql SELECT pipeline, "Status" FROM aidb.pipeline_metrics WHERE pipeline = my_pipeline; __OUTPUT__ pipeline | Status - -------------+---------- my_pipeline | UpToDate (1 row)