deepstream-sop

DeepStream SOP 推論マイクロサービススキル

このスキルでは、産業用映像フィードの時間的なアクション検出と VLM ベースの SOP 準拠モニタリングを行う、GPU アクセラレーション対応パイプラインである NVIDIA DeepStream SOP（Standard Operating Procedure）推論マイクロサービスの構築、拡張、デバッグを AI コーディングアシスタントに案内します。

リファレンスリポジトリ：https://github.com/NVIDIA/sop-monitoring-blueprints/tree/main/microservices/sop-inference-bp ローカルのリファレンスコード：sop-inference-bp/ ディレクトリ（リポジトリをローカルにクローンしたもの）

---

モデル

推論の両ステージともモデルに依存せず、環境変数（GEBD の場合は Triton ディレクトリも）で切り替えられます。

ステージ	役割	モデルクラス	デフォルト	切り替え方法
ステージ 1（CV）	フレームごとの境界スコアリング → チャンク分割	汎用イベント境界検出（GEBD）	Triton Python バックエンド経由の DDM（MCG-NJU/DDM）	triton_model_repo/<model>/ を置き換え、DDM_MODEL_PATH を設定します（§ 5）
ステージ 3（VLM）	チャンクごとのアクション分類	vLLM 経由の視覚言語モデル	Cosmos Reason 1 7B（Reason 2 もサポート）	VLLM_MODEL_PATH に別の HF ID またはローカルパスを設定します

「GEBD」は切り替え可能なステージ 1 のスロット、「DDM」はデフォルトのアーキテクチャを指します（これらの用語は同じ意味で使用されます）。

チャンク分割方式はリクエストごとに選択できます（§ 2）。デフォルトの ddm-net は GEBD を使用します。uniform は固定長のチャンクを生成し、ステージ 1 の GEBD をバイパスします（§ 3、§ 6）。DDM の時間ウィンドウは FRAMES_PER_SIDE / SEQUENCE_BATCH で設定でき（§ 4、§ 5）、オプションで TensorRT も使用できます（§ 5）。

---

アーキテクチャ概要

Kafka コンテナと並行して、Docker コンテナ（nvds-action-sop）内で実行されます。全体図は references/sop_architecture.svg を参照してください。

4 段階の SOPVideoProcessor パイプラインを通るデータフロー（リクエストごと）：

Input Sources                    Docker Container: nvds-action-sop
─────────────                    ──────────────────────────────────────────────────
Video Files ──┐                  FastAPI Server (port 8300)
RTSP Streams ─┤── base64/       ├─ /v1/chat/completions → SOPProcessManager
Basler Camera ┘   file/rtsp/       │
                  camera           │ ModelInitializer: VLM first, then DDM dummy pipeline
                                   │ 4 Thread Pools: cv(32), clip(32), vlm(64), vlm_req(64)
                                   │
                                   ▼ SOPVideoProcessor (per-request)
                                   ┌────────────────────────────────────────────────┐
                                   │ Stage 1: DeepStream Pipeline (GPU)             │
                                   │   Source → nvstreammux → tee1                  │
                                   │    ├─[inference] queue1 → nvdspreprocess       │
                                   │    │  → nvinferserver (Triton CAPI + DDM)      │
                                   │    │  → InferOutputTensorParser → score_queue  │
                                   │    ├─[frames]  queue3 → nvvideoconvert         │
                                   │    │  → capsfilter → appsink                   │
                                   │    │  → DecodedFrameRetriever → frame_queue    │
                                   │    └─[RTSP out] queue → convert → H.264 enc    │  (optional, § 18)
                                   │       → rtppay → udpsink → RTSPServer (§ 18)   │  opt-in only
                                   │              │ boundary scores                 │
                                   │              ▼                                 │
                                   │ Stage 2: Clip Post-Process                     │
                                   │   Boundary detection → chunk segmentation      │
                                   │              │ video frames + timestamps        │
                                   │              ▼                                 │
                                   │ Stage 3: VLM Inference                         │
                                   │   Embedded vLLM (Cosmos Reason 1/2)            │
                                   │   Frame sampling at VLM_FPS → classification   │
                                   │              │ action labels                    │
                                   │              ▼                                 │
                                   │ Stage 4: SOP Checker                           │
                                   │   Sequence validation → missing/misordered     │
                                   │              │ chunk results                    │
                                   │              ▼                                 │
                                   │         final_queue                            │
                                   └────────────────────────────────────────────────┘
                                          │
Output                                    ▼
──────                             ┌─────────────────┐
SSE Stream (chat.completion.chunk) │ Kafka Messages   │
Non-streaming (chat.completion)    │ (JSON/Protobuf)  │
Prometheus metrics (/v1/metrics)   └────────┬────────┘
                                            ▼
                                   Docker Container: kafka
                                   (apache/kafka:3.7.0)

---

セクション一覧

各セクションは references/ 内の独立したファイルです。タスクに必要なものだけを読み込んでください。

§	ファイル	担当範囲
1	skill_01_fastapi_endpoints.md	FastAPI エンドポイント、サーバーの初期化、Prometheus メトリクス
2	skill_02_pydantic_schemas.md	リクエスト／レスポンスの Pydantic モデル（api_types.py）
3	skill_03_deepstream_pipeline.md	DeepStream pyservicemaker パイプライン、テンソルパーサー、ダミーパイプライン
4	skill_04_config_templates.md	nvdspreprocess / nvinferserver の設定テンプレートとレンダリング
5	skill_05_triton_ddm_model.md	Triton モデルリポジトリ、config.pbtxt、model.py、ddm_net.py
5b	skill_05b_custom_postprocess.md	C++ 後処理プラグイン、Makefile、IOptions API
6	skill_06_sop_process_manager.md	SOPProcessManager、SOPVideoProcessor、VLLMInference、Kafka
6b	skill_06b_sop_checker.md	SOP シーケンスとチェッカーの準拠：MissingNumberDetector、SopCheckerCache、SopCheckerRequest/Response
7	skill_07_sse_streaming.md	SSE ジェネレーター、ストリームレスポンスの形式、ダミーテストモード
8	skill_08_basler_camera.md	Basler カメラのサポート、Pylon SDK、エミュレーション、形式
9	skill_09_docker_build_deploy.md	Docker のビルド、デプロイ、.env の設定
10	skill_10_test_suite.md	テストスイートのカバレッジ、アサーション、テストの実行
11	skill_11_env_variables.md	すべての環境変数のリファレンス
12	skill_12_evaluation_workflow.md	エンドツーエンド評価のワークフロー：静的チェック、ビルド、起動、テスト、API／カメラ／Kafka のチェック、レポート
13	skill_13_verification_curl.md	検証手順と curl の例
14	skill_14_implementation_checklist.md	実装チェックリスト：ファイルのコピー一覧、生成ファイル、Docker の前提条件、検証
15	skill_15_latency_measurement.md	SSE ストリーミングによるファイル入力の TTFC と C2C レイテンシ測定
16	skill_16_message_schema.md	Kafka メッセージスキーマの選択（デフォルトの JSON と NvProtoSchema）と、カスタムデータによるメッセージの拡張
17	skill_17_camera_latency_measurement.md	内部パイプラインのタイムスタンプを使用した、カメラ／ライブストリームの chunk_e2e レイテンシ測定
18	skill_18_rtsp_streaming_output.md	オプトインの RTSP ストリーミング出力：tee1 タップによる再ストリーミング、RTSPStreamingServer、SW_ENCODER の切り替え。RTSP を明示的にリクエストされた場合にのみ生成してください

エンドツーエンド評価では、まず § 12 を読み、必要に応じてビルド／テスト／curl／レイテンシ／カメラ／Kafka の各セクションを読み込んでください。

§ 18 はオプトインです。RTSP 出力を明示的にリクエストされた場合にのみ生成してください。それ以外の場合は § 18 と以下の RTSP_* ルールをスキップしてください。

---

主要ファイルの対応表

ソースからターゲットへのファイル対応全体は、 skill_14_implementation_checklist.md に記載されています。

• そのままコピーするファイル：references/ にある、重要なアルゴリズム（サイクル検出、qwen_vl_utils の前処理、DeepStream の IOptions API、protobuf のソースなど）を含むファイルです。ファイルごとの理由も記載されています。
• 調整可能なテンプレートとしてコピーするファイル（Dockerfile、compose.yaml、Triton の設定と model.py、ddm_net.py、Pylon エミュレーション設定など）。
• スキルの各セクションから生成するファイル：生成時に厳密に従う必要がある、以下の重要なルールがそれぞれ注記されています。
• Docker ビルドの前提条件と、ビルド後の検証チェックリスト。

設定ファイル（nvds_preprocess_template.txt、nvds_inference_template.txt、vlm_prompts.txt）は、configs/ のものをそのまま使用します。

skill_06b を読み込んだら、プロジェクトのルートにある configs/actions.json を読み込み、§ 6b-G の生成ワークフローを実行して nvds_action_detector/missing_number_detector.py を生成します。 configs/actions.json が存在しない、または無効な場合は、参照ファイルをコピーしてください。

---

重要なルール

各ルールの詳細は、リンク先の skill_NN_*.md 参照ファイルに記載されています。 | タグ | ルールの概要 | 詳細 | |-----|---|---| | MANAGER_INIT_IN_MAIN | uvicorn.run() の前に main() 内で SOPProcessManager を初期化します。lifespan() 内では初期化しません | skill_01_fastapi_endpoints.md | | NAMED_KWARGS | create_video_processor() では名前付き kwargs を使用します。カメラ引数は個別の kwargs として渡します | skill_06_sop_process_manager.md | | LIVE_REQUIRES_STREAM_TRUE | ライブ入力（RTSP／カメラ）には stream: true が必要です | skill_08_basler_camera.md | | VLM_DISABLED_DISABLES_SOP_CHECKER | DISABLE_VLM_INFERENCE=true にすると、インポート時に SOP チェッカーが自動的に無効になります | skill_06_sop_process_manager.md | | CHUNK_PARAMS_MAX_LENGTH | ChunkParams.max_length_sec は内部で 10s、API のデフォルト値は 60s です | skill_06_sop_process_manager.md | | VLM_WARMUP_BEFORE_DDM | ModelInitializer では、最初に VLM をウォームアップしてから CV のダミーパイプラインを実行します | skill_06_sop_process_manager.md | | VLM_WARMUP_3_FRAMES | VLM のウォームアップには 3 フレーム（torch.zeros）が必要です。Qwen3VL は 3 フレーム未満だとハングします | skill_06_sop_process_manager.md | | THREAD_POOL_SIZES | スレッドプールは 4 つです：cv(32)、clip(32)、vlm_inference(64)、vlm_request(64) | skill_06_sop_process_manager.md | | MEDIA_INFO_PYMEDIAINFO | メディア情報は pymediainfo で取得します。ライブソースでは fps=30／duration=inf を直接設定します | skill_06_sop_process_manager.md | | CAMERA_EMULATION_PYLON_CAMEMU | カメラエミュレーションには PYLON_CAMEMU=1 を使用します（シリアル番号 0815-0000） | skill_08_basler_camera.md | | DEEPSTREAM_LIB_HIDE | DeepStream ライブラリを隠す方法：gst-plugin-pylon のビルド中にライブラリ名を lib.tmp に変更します | skill_08_basler_camera.md | | VLM_REAL_GPU_FRAMES | VLM は DecodedFrameRetriever 経由で実際の GPU フレームを使用します。推論に torch.zeros は使用しないでください | skill_06_sop_process_manager.md | | BUFFER_RETRIEVER_STATIC_BASE | DecodedFrameRetriever は super().__init__() 経由で BufferRetriever を静的に継承する必要があります。実行時に __class__.__bases__ を変更すると pipeline.attach() がハングします | skill_06_sop_process_manager.md | | FRAME_RETRIEVER_PRIORITY | create_inference_pipeline では、frame_retriever= kwarg が frame_queue より優先されます | skill_03_deepstream_pipeline.md | | MUX_ORIGINAL_RESOLUTION | nvstreammux は元の解像度（224 ではありません）を使用します。get_media_info() から mux_width/mux_height を渡します（カメラ以外の入力ではライブ RTSP をプローブします。カメラの経路には影響しません） | skill_03_deepstream_pipeline.md、skill_06_sop_process_manager.md | | FILE_URI_NO_DOUBLE_PREFIX | create_inference_pipeline のファイルソースでは、file:// を付加する前に file_path.startswith("file://") を確認します。API は file:// URL をそのまま渡します | skill_03_deepstream_pipeline.md | | CLEANUP_ON_DISCONNECT | クライアント切断時のパイプラインのクリーンアップは、try/finally 内の trigger_stop_processors で行います | skill_07_sse_streaming.md | | UNIFIED_CLIP_POST_PROCESS | ファイルとライブの両方で統一された clip_post_process() を使用します。stop() は _score_queue に None を入れます | skill_06_sop_process_manager.md | | ABORT_INFLIGHT_VLM | stop() の際に llm.abort(req_id) で処理中の VLM リクエストを中断します | skill_06_sop_process_manager.md | | LOGGER_EXPORT_GET_LOGGER | ds_logger.py は get_logger をエクスポートする必要があります | skill_06_sop_process_manager.md | | KAFKA_USE_CREATE_PRODUCER | Kafka では messager.py の create_producer() を使用します。Messager クラスはありません | skill_06_sop_process_manager.md | | USER_PROMPT_PRIORITY | ユーザーのリクエストテキストは VLM_PROMPT_PATH ファイルより優先されます。リクエスト内の {"type":"text"} は設定ファイルのプロンプトを上書きします | skill_06_sop_process_manager.md | | EVAL_USE_CONFIG_PROMPT | 評価／レイテンシのリクエストでは、デフォルトでリクエストテキストを省略し、VLM が VLM_PROMPT_PATH を使用するようにします | skill_12_evaluation_workflow.md、skill_13_verification_curl.md、skill_15_latency_measurement.md、skill_17_camera_latency_measurement.md | | CHUNK_SCHEMA_FIELD_NAMES | チャンクスキーマ：chunk_idx、cv_boundary_score、checker_result。サマリーでは chunk_idx=-1 | skill_06_sop_process_manager.md | | SEQUENTIAL_FRAME_DRAIN | decoded_frame_queue（チャンク間で共有される FIFO）からの取り出しは単一のスレッドで行い、チャンクごとに VLM を順次送信します。並列で取り出すとフレームが奪われ、フレーム数が 0 のチャンクや誤った VLM 入力が発生します | skill_06_sop_process_manager.md | | WALL_CLOCK_BEFORE_GPU | DecodedFrameRetriever.consume() では、GPU dlpack の前に wall_clock_entry = time.time() を取得し、キューには 3 要素のタプル (timestamp, wall_clock_entry, tensor) を入れます | skill_06_sop_process_manager.md、skill_17_camera_latency_measurement.md | | CHUNK_E2E_PIPELINE_TIMESTAMPS | カメラのレイテンシ（§ 17）用に、pipeline_chunk_end_timestamp（最後のフレームの wall_clock）と pipeline_vlm_ready_timestamp（tm_e2e.now()）を chunk_info に書き込みます | skill_06_sop_process_manager.md、skill_17_camera_latency_measurement.md | | VLM_INFERENCE_REQUIRED_KWARGS | すべての VLLMInference.inference() 呼び出しで video_fps、system_prompt、max_completion_tokens を渡す必要があります | skill_06_sop_process_manager.md | | UNIFORM_CHUNKING_BYPASSES_DDM | chunking_options.algorithm="uniform" → 固定長チャンクを使用します。create_inference_pipeline(uniform_chunk=True) は tee1 の分岐を維持したまま DDM をスキップします。ステージ 2 では uniform_clip_post_process を使用します | skill_02_pydantic_schemas.md、skill_03_deepstream_pipeline.md、skill_06_sop_process_manager.md | | DDM_TEMPORAL_CONFIGURABLE | SLIDING_WINDOWS_SIZE = 2*FRAMES_PER_SIDE + SEQUENCE_BATCH を preprocess／nvinferserver に反映します（18 をハードコードしません）。Triton config.pbtxt の sequence 次元は -1 です | skill_04_config_templates.md、skill_05_triton_ddm_model.md | | DDM_TRT_OPTIONAL_PATH | DDM_TRT_OPTIMIZATION=true では TensorRT を介して DDM を実行します（スレッドごとのコンテキスト、固定バッチサイズ = SEQUENCE_BATCH）。PyTorch にフォールバックします。両方を同時に使用しないでください。デフォルトは PyTorch です | skill_05_triton_ddm_model.md | | DDM_TRT_STREAM_ORDERING | DDMTensorRTEngine.infer()：wait_stream(current) → execute_async_v3 → torch.cuda.synchronize(device)（ストリーム単位ではありません）。ストリーム単位で同期すると TRT の補助ストリーム処理が未完了のままになり、gst-CV SIGSEGV が発生します（NVBug 6289256） | skill_05_triton_ddm_model.md | | METADATA_LICENSE_FROM_FILE | /v1/metadata は DS_SOP_LICENSE_PATH（デフォルト /opt/nvidia/nvds_sop/license.txt）から licenseInfo を読み取ります。ライセンス文をハードコードしないでください | skill_01_fastapi_endpoints.md | | CAMERA_EMULATION_FRAMES_RGB | Pylon エミュレーション用 PNG は、明示的に 3 チャネル RGB にする必要があります（Emulation_0815-0000.pfs PixelFormat=RGB8Packed に一致）。nvvideoconvert ! videoconvert ! "video/x-raw,format=RGB" ! pngenc を使って生成します | skill_08_basler_camera.md | | COMPOSE_ENV_PASSTHROUGH | docker compose は ${VAR} 参照のみを置換します。実行時環境変数をコンテナーに渡すには、すべて environment: の下に明示的に列挙する必要があります。 | skill_09_docker_build_deploy.md | 以下の 4 つの RTSP_* ルールは、オプションの RTSP ストリーミング出力機能 （§ 18）が要求された場合にのみ適用されます。デフォルトのビルドには適用されないため、ユーザーが RTSP 出力を求めていない場合はスキップしてください。

| RTSP_OUTPUT_TAPS_TEE1 | rtsp_port が指定されている場合に限り、既存の tee1（メインの推論リンクの後に追加）から RTSP 出力ブランチをリンクします。 | skill_18_rtsp_streaming_output.md | | RTSP_LEAKY_QUEUE_TINY | バックプレッシャーと NVMM プールの枯渇を防ぐため、RTSP ブランチのキューは leaky=2 と小さい上限（max-size-buffers=2）にする必要があります。 | skill_18_rtsp_streaming_output.md | | RTSP_KEYINT_MAX_30 | 下流でのシークを可能にするため、RTSP H.264 エンコーダーでは key-int-max=30 を設定し（B フレームは無効化）、使用する必要があります。 | skill_18_rtsp_streaming_output.md | | RTSP_ENCODER_FALLBACK | SW_ENCODER に基づいてソフトウェア／ハードウェアの H.264 エンコーダーを選択し、MJPEG にフォールバックします。 | skill_18_rtsp_streaming_output.md |

---