视频识别看起来只是把一个文件交给模型，再等待一段文字结果。但在实际系统中，真正困难的部分通常不在 HTTP 请求本身，而在于如何控制输入规模、处理耗时任务、约束输出格式，并且在失败后能够定位原因。
例如，一个客服质检系统可能需要从视频中提取时间轴、人物动作、画面文字和风险事件。直接把原始视频发送给模型会遇到几个问题：文件大小可能超过接口限制；连续帧会产生大量重复信息；网络重试可能导致重复计费或重复任务；自然语言结果也很难直接进入数据库和告警系统。
因此，更稳妥的做法是把视频识别拆成一条可观测流水线：
- 校验视频来源、格式和大小。
- 按固定间隔或场景变化抽取代表帧。
- 将帧、时间戳和任务约束发送给视觉模型 API。
- 对返回结果进行 JSON 解析、字段校验和时间范围校验。
- 保存原始响应、规范化结果、模型标识和请求追踪号。
本文以 Python 为例，展示这条流水线的最小实现。示例中的模型名称、上传接口和请求路径仅表示一种常见的 OpenAI 兼容调用方式，实际使用时必须以服务商当前文档为准。
核心原理
为什么要抽帧
视频可以看作带有时间顺序的图像序列。对大多数“识别某个事件”类任务而言，不需要把每一帧都发送给模型。可以先按时间间隔抽帧，例如每 2 秒取一帧，再把时间戳作为上下文传给模型。
固定间隔抽帧实现简单，但可能错过很短的动作。更复杂的方案可以先用视频处理工具检测场景变化，再对变化明显的位置补帧。抽帧策略应由业务目标决定：概览生成适合稀疏采样，安全事件检测则需要更密集采样，并且最好保留原视频以便人工复核。
为什么要异步化
模型调用的耗时与视频时长、帧数、服务负载和网络状况有关。让 Web 请求一直等待会占用连接和线程，也不利于重试。因此，接口层只负责创建任务，后台 worker 再完成抽帧和模型调用，客户端通过任务 ID 查询状态。
任务至少应有 PENDING、RUNNING、SUCCEEDED、FAILED 四种状态。状态更新需要带版本号或使用条件更新，避免两个 worker 同时处理同一任务。重试也必须设置上限，并区分可重试错误与不可重试错误，例如超时通常可以重试，而输入格式错误不应无限重试。
为什么要约束输出
模型返回的文本不等于可靠的数据接口。生产代码应在提示词中要求 JSON，同时在客户端再次解析并校验。校验失败时，可以保留原始响应，进入人工复核或有限次数的修复请求，不能直接把未经验证的文本写入告警和自动化流程。
可执行实现
准备环境
安装 OpenCV 和 HTTP 客户端：
python -m venv .venv
. .venv/bin/activate
pip install opencv-python requests jsonschema
密钥和接口地址从环境变量读取：
export VISION_API_BASE="https://api.example.com/v1"
export VISION_API_KEY="replace-with-your-key"
export VISION_MODEL="your-vision-model"
如果使用 HaerAPI 或其他中转接口，应先确认其当前文档是否支持目标视觉模型、图片输入格式、鉴权方式、数据保留规则和速率限制，再据此调整请求封装。
抽取带时间戳的帧
下面的函数只负责视频读取和缩放，不负责调用模型。将媒体处理与网络请求分开，便于单独测试和替换采样策略。
from pathlib import Path
import base64
import cv2
def sample_frames(video_path: str, interval_seconds: float = 2.0,
max_width: int = 1280) -> list[dict]:
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
raise ValueError(f"cannot open video: {video_path}")
fps = cap.get(cv2.CAP_PROP_FPS)
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if fps <= 0 or frame_count <= 0:
cap.release()
raise ValueError("video metadata is unavailable")
step = max(1, round(fps * interval_seconds))
result = []
index = 0
while True:
ok, frame = cap.read()
if not ok:
break
if index % step == 0:
height, width = frame.shape[:2]
if width > max_width:
scale = max_width / width
frame = cv2.resize(frame, (max_width, round(height * scale)))
ok, encoded = cv2.imencode(".jpg", frame)
if ok:
result.append({
"timestamp": round(index / fps, 3),
"image_base64": base64.b64encode(encoded).decode("ascii")
})
index += 1
cap.release()
return result
这里使用 Base64 只是为了演示把帧放入 JSON。若服务商提供对象存储 URL 或 multipart 上传，生产环境可以改用对应方式，避免请求体过大。上传 URL 还应设置有效期，并限制访问权限。
组织模型请求
先定义一个尽可能明确的任务契约：
import json
import os
import requests
SCHEMA = {
"type": "object",
"required": ["events"],
"properties": {
"events": {
"type": "array",
"items": {
"type": "object",
"required": ["start", "end", "label", "confidence"],
"properties": {
"start": {
"type": "number", "minimum": 0},
"end": {
"type": "number", "minimum": 0},
"label": {
"type": "string", "minLength": 1},
"confidence": {
"type": "number", "minimum": 0, "maximum": 1}
}
}
}
}
}
def analyze_frames(frames: list[dict], instruction: str) -> dict:
base = os.environ["VISION_API_BASE"].rstrip("/")
key = os.environ["VISION_API_KEY"]
model = os.environ["VISION_MODEL"]
content = [{
"type": "text", "text": (
"请只返回 JSON，不要使用 Markdown 代码围栏。"
"每个事件必须包含 start、end、label、confidence。"
"start 和 end 使用秒数，不能超出输入帧的时间范围。\n" + instruction
)}]
for item in frames:
content.append({
"type": "text", "text": f"frame_timestamp={item['timestamp']}"})
content.append({
"type": "image_url",
"image_url": {
"url": "data:image/jpeg;base64," + item["image_base64"]
}
})
response = requests.post(
base + "/chat/completions",
headers={
"Authorization": f"Bearer {key}"},
json={
"model": model,
"messages": [{
"role": "user", "content": content}],
"temperature": 0
},
timeout=(10, 180)
)
response.raise_for_status()
payload = response.json()
text = payload["choices"][0]["message"]["content"]
return json.loads(text)
这里没有假设所有服务商都支持同一请求结构。某些 API 可能要求先上传文件，某些 API 可能使用专门的视频输入字段，也可能只接受公网可访问的图片地址。接入时应把差异收敛在一个适配器中，而不是散落在业务代码里。
校验与归一化
使用 jsonschema 校验结构，再检查业务约束：
from jsonschema import validate
def validate_result(result: dict, frames: list[dict]) -> dict:
validate(instance=result, schema=SCHEMA)
if not frames:
raise ValueError("no frames were sampled")
max_time = frames[-1]["timestamp"]
normalized = []
for event in result["events"]:
start = float(event["start"])
end = float(event["end"])
if end < start:
raise ValueError("event end is before start")
if end > max_time + 2:
raise ValueError("event exceeds sampled time range")
normalized.append({
"start": round(max(0, start), 3),
"end": round(end, 3),
"label": event["label"].strip(),
"confidence": round(float(event["confidence"]), 4)
})
return {
"events": normalized}
置信度只能作为模型输出的一个信号，不能自动等同于业务概率。涉及处罚、封禁、医疗或安全决策时，应增加人工复核、规则校验和原始视频回看环节。
任务与可靠性设计
数据库可以保存以下字段：task_id、status、source_uri、model、prompt_version、attempt、request_id、raw_response_uri、normalized_result、error_code、created_at 和 updated_at。其中 prompt_version 很重要，否则同一视频在不同提示词下产生差异时难以追溯。
任务提交时生成幂等键，例如由业务单号、文件哈希和规则版本组成。worker 获取任务后使用条件更新：只有状态仍为 PENDING 的记录才能变更为 RUNNING。网络超时重试时使用指数退避，并给每次请求设置连接超时和读取超时。不要把完整视频、Base64 帧和模型原始响应写入普通应用日志，以免造成隐私泄露和日志膨胀。
对于数据保护，至少需要明确：视频是否包含个人信息，上传前是否需要脱敏，传输和存储是否加密，第三方处理方保存多久，谁有权限读取结果，以及删除请求如何传递到缓存、对象存储和备份。HaerAPI 这类接口服务是否适合具体数据，仍应以其现行条款和组织自身的合规评估为准。
常见问题
抽帧越密，结果一定越好吗？
不一定。帧数增加会扩大请求体和处理成本，也可能让模型面对更多重复信息。应根据事件持续时间、视频分辨率和允许的漏检风险选择间隔，并用业务样本验证，而不是直接采用固定值。
为什么模型已经被要求返回 JSON，仍然会解析失败？
提示词约束不是协议保证。模型可能添加解释文字、输出非法转义，或返回字段类型错误。因此必须保留原文、执行解析和 Schema 校验；必要时使用服务商提供的结构化输出能力，但仍要处理接口错误和不完整响应。
网络超时后能否直接重试？
需要先判断请求是否已经在服务端执行。若服务端支持幂等键，应在重试时复用同一键；若不支持，则应结合任务状态查询或业务侧去重，避免重复创建任务。超时并不等于服务端没有收到请求。
是否应该让模型直接判断高风险事件？
对于高风险场景，不建议把一次模型输出作为唯一依据。可以让模型负责候选事件提取，再由确定性规则、第二阶段模型或人工审核进行确认，并记录最终决策链路。
总结
可靠的视频识别接入，本质上是一个包含媒体预处理、模型调用、结构化校验、任务调度和审计存储的工程系统。抽帧降低输入压力，异步任务隔离长耗时操作，幂等和重试处理不确定的网络行为，Schema 校验则把自然语言结果变成可检查的数据。
