deepstream-sop

DeepStream SOP 推理微服务技能

此技能指导 AI 编码助手构建、扩展和调试 NVIDIA DeepStream SOP（标准操作程序）推理微服务 —— 这是一个用于工业视频流中时序动作检测和基于 VLM 的 SOP 合规性监控的 GPU 加速流水线。

参考仓库：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）。

---

架构概览

在 Docker 容器（nvds-action-sop）中运行，与 Kafka 容器并行。完整图示：references/sop_architecture.svg。

数据流经 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 | 在 main() 中、uvicorn.run() 之前初始化 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 checker | 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_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() 会将 None 放入 _score_queue | 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 | Eval/延迟请求默认省略请求文本，以便 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 | 分块 schema：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 | 将 pipeline_chunk_end_timestamp（最后一帧的 wall_clock）和 pipeline_vlm_ready_timestamp（tm_e2e.now()）写入 chunk_info，用于摄像头延迟测量（§ 17） | 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) 会跳过 DDM，但保留 tee1 扇出；Stage 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 的序列维度为 -1 | skill_04_config_templates.md、skill_05_triton_ddm_model.md | | DDM_TRT_OPTIONAL_PATH | DDM_TRT_OPTIMIZATION=true 会通过 TensorRT 运行 DDM（每线程独立上下文，固定 batch = 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 时，RTSP 输出分支才会从现有的 tee1（在主推理链接之后添加）连接。 | skill_18_rtsp_streaming_output.md | | RTSP_LEAKY_QUEUE_TINY | RTSP 分支队列必须设置为 leaky=2，并使用极小的容量（max-size-buffers=2），以防止背压和 NVMM 池耗尽。 | 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 |

---