diff --git a/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py b/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py index c0b96c6..18a8296 100644 --- a/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py +++ b/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py @@ -48,55 +48,60 @@ def parse_vlm_json(raw: str) -> dict: def validate_schema(data: dict) -> dict: """Schema 校验 + 脏数据清洗 + 新架构（整视频分析）schema: global_summary / events / people_mentioned + 兼容旧结构: 模型若输出 frame_details / entities_json 也接受并转换。 注意: compute_provider 是内部记账字段（orchestrator 在解析成功后填充）， 提示词从不要求模型输出它，故不做必填校验。 """ - required = ["global_summary", "entities_json", "frame_details"] + required = ["global_summary", "events"] for k in required: if k not in data: + if k == "events" and "frame_details" in data: + continue raise VLMOutputInvalidError(f"缺失字段: {k}") - # entities_json 结构校验 - if not isinstance(data["entities_json"], list): - raise VLMOutputInvalidError("entities_json 必须为数组") + # events 结构校验（兼容 frame_details 旧结构转换） + events = data.get("events") + if events is None and "frame_details" in data: + events = [] + for f in data["frame_details"]: + events.append({ + "timestamp": f.get("frame_timestamp", ""), + "description": f.get("action", ""), + "people": [f.get("person", "")] if f.get("person") else [], + "is_attention_event": bool(f.get("is_attention_event", False)), + }) + if not isinstance(events, list): + raise VLMOutputInvalidError("events 必须为数组") - cleaned_entities = [] - for ent in data["entities_json"]: - if not isinstance(ent, dict): + cleaned = [] + for ev in events: + if not isinstance(ev, dict): continue - if "person" not in ent or "action" not in ent: - raise VLMOutputInvalidError("entity 缺少 person 或 action 字段") - cleaned_entities.append({ - "person": str(ent["person"]), - "action": str(ent["action"]), - "clothing": str(ent.get("clothing", "")) + if "timestamp" not in ev or "description" not in ev: + raise VLMOutputInvalidError("event 缺少 timestamp 或 description 字段") + people = ev.get("people", []) + if isinstance(people, str): + people = [people] + if not isinstance(people, list): + people = [] + cleaned.append({ + "timestamp": str(ev["timestamp"]), + "description": str(ev["description"]), + "people": [str(p) for p in people if p], + "is_attention_event": bool(ev.get("is_attention_event", False)), }) - data["entities_json"] = cleaned_entities + data["events"] = cleaned - # frame_details 结构校验 - if not isinstance(data["frame_details"], list): - raise VLMOutputInvalidError("frame_details 必须为数组") - - cleaned_frames = [] - for frame in data["frame_details"]: - if not isinstance(frame, dict): - continue - for k in ["frame_index", "frame_timestamp", "person", "action", "source_providers"]: - if k not in frame: - raise VLMOutputInvalidError(f"frame_details 缺少字段: {k}") - sp = frame["source_providers"] - if not isinstance(sp, list) or len(sp) == 0: - raise VLMOutputInvalidError("frame_details.source_providers 必须为非空数组") - cleaned_frames.append({ - "frame_index": int(frame["frame_index"]), - "frame_timestamp": str(frame["frame_timestamp"]), - "person": str(frame["person"]), - "action": str(frame["action"]), - "clothing": str(frame.get("clothing", "")), - "is_attention_event": bool(frame.get("is_attention_event", False)), - "source_providers": [str(p) for p in sp] - }) - data["frame_details"] = cleaned_frames + # people_mentioned（兼容 entities_json 旧结构） + people = data.get("people_mentioned") + if people is None and "entities_json" in data: + people = data["entities_json"] + if isinstance(people, list) and people and isinstance(people[0], dict): + people = [p.get("person", "") for p in people] + if not isinstance(people, list): + people = [] + data["people_mentioned"] = [str(p) for p in people if p] # compute_provider: 模型通常不输出（提示词不要求），缺失/非法时归一为空数组， # 由 orchestrator.format_cloud_result 用实际成功的 provider 覆盖填充 diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py index 5a4209b..3a70746 100644 --- a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py @@ -74,6 +74,7 @@ class NvidiaVisionAdapter(BaseModelAdapter): def _upload_asset(self, video_path: str) -> Optional[str]: """用 NVIDIA Assets API 上传大视频文件，返回 asset_id 供 video_url 引用。""" + content_type = "video/mp4" upload_timeout = max(self.timeout, 900) try: r = requests.post( @@ -82,7 +83,7 @@ class NvidiaVisionAdapter(BaseModelAdapter): "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, - json={"contentType": "video/mp4", "description": "fam-edge video asset"}, + json={"contentType": content_type, "description": "fam-edge video asset"}, timeout=60, ) except Exception as e: @@ -106,8 +107,9 @@ class NvidiaVisionAdapter(BaseModelAdapter): up = requests.put( upload_url, data=f, - # 注意：必须用小写 header 名，预签名 S3 URL 的签名覆盖小写 content-type - headers={"content-type": "application/octet-stream", + # 必须全小写 header 名且 content-type 值与 POST 的 contentType 一致： + # 预签名 S3 URL 签名覆盖这两个值，不一致会 SignatureDoesNotMatch + headers={"content-type": content_type, "x-amz-meta-nvcf-asset-description": "fam-edge video asset"}, timeout=upload_timeout, )