Add media pipeline timing metrics

content_agent/business_modules/content_discovery/pattern_recall/recall_decision.py

@@ -109,40 +109,52 @@ def _update_content_media_records(
     judgments: list[dict[str, Any]],
 ) -> list[dict[str, Any]]:
     updates_by_key: dict[tuple[Any, Any], dict[str, Any]] = {}
+    timings_by_key: dict[tuple[Any, Any], dict[str, Any]] = {}
     for item, judgment in zip(discovered_content_items, judgments):
+        key = (item.get("platform"), item.get("platform_content_id"))
         update = judgment.get("media_storage_update") if isinstance(judgment, dict) else None
         if isinstance(update, dict):
-            updates_by_key[(item.get("platform"), item.get("platform_content_id"))] = update
-    if not updates_by_key:
+            updates_by_key[key] = update
+        timing = judgment.get("timing_metrics") if isinstance(judgment, dict) else None
+        if isinstance(timing, dict):
+            timings_by_key[key] = timing
+    if not updates_by_key and not timings_by_key:
         return content_media_records
     updated_records: list[dict[str, Any]] = []
     for record in content_media_records:
         key = (record.get("platform"), record.get("platform_content_id"))
         update = updates_by_key.get(key)
-        if not update:
+        timing = timings_by_key.get(key)
+        if not update and not timing:
             updated_records.append(record)
             continue
         raw_payload = dict(record.get("raw_payload") or {})
-        raw_payload.update(
-            {
-                "content_media_status": update.get("content_media_status", record.get("content_media_status")),
-                "oss_url": update.get("oss_url"),
-                "local_path": update.get("local_path"),
-            }
-        )
-        if update.get("failure_reason"):
-            raw_payload["failure_reason"] = update["failure_reason"]
-        update_payload = update.get("raw_payload")
-        if isinstance(update_payload, dict):
-            raw_payload.update(update_payload)
+        if update:
+            raw_payload.update(
+                {
+                    "content_media_status": update.get("content_media_status", record.get("content_media_status")),
+                    "oss_url": update.get("oss_url"),
+                    "local_path": update.get("local_path"),
+                }
+            )
+            if update.get("failure_reason"):
+                raw_payload["failure_reason"] = update["failure_reason"]
+            update_payload = update.get("raw_payload")
+            if isinstance(update_payload, dict):
+                raw_payload.update(update_payload)
+        if timing:
+            raw_payload["gemini_timing_metrics"] = timing
         merged = {
             **record,
-            "content_media_status": update.get("content_media_status", record.get("content_media_status")),
-            "oss_url": update.get("oss_url"),
-            "local_path": update.get("local_path"),
+            "content_media_status": (
+                update.get("content_media_status", record.get("content_media_status"))
+                if update else record.get("content_media_status")
+            ),
+            "oss_url": update.get("oss_url") if update else record.get("oss_url"),
+            "local_path": update.get("local_path") if update else record.get("local_path"),
             "raw_payload": {key: value for key, value in raw_payload.items() if value is not None},
         }
-        if update.get("failure_reason"):
+        if update and update.get("failure_reason"):
             merged["failure_reason"] = update["failure_reason"]
         updated_records.append(merged)
     return updated_records
@@ -222,6 +234,8 @@ def _build_evidence_row(
     for field in ["failure_type", "exception_type", "http_status_code"]:
         if field in judgment:
             summary[field] = judgment.get(field)
+    if isinstance(judgment.get("timing_metrics"), dict):
+        summary["timing_metrics"] = judgment["timing_metrics"]
     return {
         "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
         "run_id": run_id,

content_agent/integrations/gemini_video.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import json
 import os
+import time
 from typing import Any, Callable, Mapping
 
 import httpx
@@ -57,6 +58,10 @@ def _with_media_update(result: dict[str, Any], update: dict[str, Any] | None) ->
     return result
 
 
+def _with_timing(result: dict[str, Any], timing: dict[str, Any]) -> dict[str, Any]:
+    return {**result, "timing_metrics": timing}
+
+
 def _clamp_score(value: Any) -> float:
     try:
         number = float(value)
@@ -157,16 +162,24 @@ class GeminiVideoClient:
         play_url = media.get("play_url")
         if not play_url:
             return _with_media_update(
-                _fail("no_play_url", query_text=query_text),
+                _with_timing(_fail("no_play_url", query_text=query_text), {"video_fetch": {"skipped": "no_play_url"}}),
                 _media_unavailable_update("no_play_url"),
             )
+        timing_metrics: dict[str, Any] = {}
         try:
-            data_url = self.fetch_fn(play_url, content.get("platform", "douyin"))
+            data_url, fetch_metrics = self._fetch_video(play_url, content.get("platform", "douyin"))
+            timing_metrics["video_fetch"] = fetch_metrics
         except Exception as exc:
-            return _fail(
-                "video_fetch_failed",
-                query_text=query_text,
-                exception_type=type(exc).__name__,
+            fetch_metrics = getattr(exc, "metrics", {}) or {}
+            if fetch_metrics:
+                timing_metrics["video_fetch"] = fetch_metrics
+            return _with_timing(
+                _fail(
+                    "video_fetch_failed",
+                    query_text=query_text,
+                    exception_type=type(exc).__name__,
+                ),
+                timing_metrics,
             )
 
         messages = [
@@ -180,8 +193,11 @@ class GeminiVideoClient:
             },
         ]
         last_failure: dict[str, Any] | None = None
+        request_attempts: list[dict[str, Any]] = []
+        request_started_at = time.monotonic()
         for attempt in range(2):
             retry_count = attempt + 1
+            attempt_started_at = time.monotonic()
             try:
                 response = self.http_post(
                     f"{self.base_url}/chat/completions",
@@ -190,9 +206,35 @@ class GeminiVideoClient:
                     timeout=self.timeout_seconds,
                 )
                 response.raise_for_status()
-                return _parse(response.json(), query_text)
+                request_attempts.append(
+                    {
+                        "attempt": retry_count,
+                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
+                        "http_status_code": getattr(response, "status_code", None),
+                        "status": "ok",
+                    }
+                )
+                timing_metrics["gemini_request"] = {
+                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
+                    "attempts": request_attempts,
+                }
+                return _with_timing(_parse(response.json(), query_text), timing_metrics)
             except httpx.HTTPError as exc:
                 failure_type = "gemini_client_timeout" if isinstance(exc, httpx.TimeoutException) else "gemini_http_error"
+                request_attempts.append(
+                    {
+                        "attempt": retry_count,
+                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
+                        "http_status_code": _http_status(exc),
+                        "status": "failed",
+                        "failure_type": failure_type,
+                        "exception_type": type(exc).__name__,
+                    }
+                )
+                timing_metrics["gemini_request"] = {
+                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
+                    "attempts": request_attempts,
+                }
                 last_failure = _fail(
                     failure_type,
                     query_text=query_text,
@@ -202,8 +244,21 @@ class GeminiVideoClient:
                 )
                 if attempt == 0 and _retryable_http(exc):
                     continue
-                return last_failure
+                return _with_timing(last_failure, timing_metrics)
             except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                request_attempts.append(
+                    {
+                        "attempt": retry_count,
+                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
+                        "status": "failed",
+                        "failure_type": "gemini_response_invalid",
+                        "exception_type": type(exc).__name__,
+                    }
+                )
+                timing_metrics["gemini_request"] = {
+                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
+                    "attempts": request_attempts,
+                }
                 last_failure = _fail(
                     "gemini_response_invalid",
                     query_text=query_text,
@@ -212,8 +267,17 @@ class GeminiVideoClient:
                 )
                 if attempt == 0:
                     continue
-                return last_failure
-        return last_failure or _fail("gemini_unknown_error", query_text=query_text)
+                return _with_timing(last_failure, timing_metrics)
+        return _with_timing(last_failure or _fail("gemini_unknown_error", query_text=query_text), timing_metrics)
+
+    def _fetch_video(self, play_url: str, platform: str) -> tuple[str, dict[str, Any]]:
+        if self.fetch_fn is video_fetch.fetch_and_compress:
+            return video_fetch.fetch_and_compress_with_metrics(play_url, platform)
+        started_at = time.monotonic()
+        result = self.fetch_fn(play_url, platform)
+        if isinstance(result, tuple) and len(result) == 2 and isinstance(result[1], dict):
+            return result[0], result[1]
+        return result, {"total_duration_ms": _elapsed_ms(started_at, time.monotonic())}
 
 
 class MissingGeminiVideoClient:
@@ -237,3 +301,7 @@ def _media_unavailable_update(failure_reason: str) -> dict[str, Any]:
         "failure_reason": failure_reason,
         "raw_payload": {"failure_reason": failure_reason},
     }
+
+
+def _elapsed_ms(start: float, end: float) -> int:
+    return max(0, int((end - start) * 1000))

content_agent/integrations/oss_archive.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, timezone
 from threading import Lock
@@ -223,11 +224,17 @@ def _archive_one(
     deadline = _parse_time(raw_payload.get("oss_archive_deadline_at"))
     if deadline and now >= deadline:
         return _with_failed_archive(record, raw_payload, now, attempt_count, "oss_upload_failed")
+    attempt_started_at = time.monotonic()
     upload_result = upload_fn(
         record.get("play_url") or "",
         referer=video_fetch._download_headers(str(record.get("platform") or ""), None),
         timeout_seconds=attempt_timeout_seconds,
     )
+    timing_metrics = {
+        "oss_upload_duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
+        "oss_upload_timeout_seconds": attempt_timeout_seconds,
+        "oss_archive_attempt_started_at": _iso(now),
+    }
     if upload_result.get("status") == "ok":
         updated_raw = {
             **raw_payload,
@@ -240,6 +247,7 @@ def _archive_one(
             "oss_archive_updated_at": _iso(now),
             "oss_object_key": upload_result.get("oss_object_key"),
             "save_oss_timestamp": upload_result.get("save_oss_timestamp"),
+            "oss_timing_metrics": timing_metrics,
         }
         return {
             **record,
@@ -265,6 +273,7 @@ def _archive_one(
         "oss_archive_last_http_status_code": upload_result.get("http_status_code"),
         "oss_archive_updated_at": _iso(now),
         "upload_failure_reason": failure,
+        "oss_timing_metrics": timing_metrics,
     }
     return {
         **record,
@@ -352,6 +361,10 @@ def _compact(payload: dict[str, Any]) -> dict[str, Any]:
     return dict(payload)
 
 
+def _elapsed_ms(start: float, end: float) -> int:
+    return max(0, int((end - start) * 1000))
+
+
 def _resolve_max_workers(value: int | None = None) -> int:
     if value is not None and value > 0:
         return value

content_agent/integrations/video_fetch.py

@@ -39,6 +39,10 @@ COMPRESS_TIMEOUT_SECONDS = 20 * 60.0
 class VideoFetchError(RuntimeError):
     """下载/压缩/超限失败,由 GeminiVideoClient 捕获转 fail。"""
 
+    def __init__(self, message: str, *, metrics: dict[str, Any] | None = None) -> None:
+        super().__init__(message)
+        self.metrics = metrics or {}
+
 
 def _download_headers(platform: str, override: dict[str, str] | None) -> dict[str, str]:
     if override is not None:
@@ -82,39 +86,86 @@ def fetch_and_compress(
     compress_timeout_seconds: float = COMPRESS_TIMEOUT_SECONDS,
     clock: Any = time.monotonic,
 ) -> str:
+    data_url, _ = fetch_and_compress_with_metrics(
+        play_url,
+        platform,
+        headers=headers,
+        http_client=http_client,
+        ffmpeg_exe=ffmpeg_exe,
+        timeout_seconds=timeout_seconds,
+        compress_timeout_seconds=compress_timeout_seconds,
+        clock=clock,
+    )
+    return data_url
+
+
+def fetch_and_compress_with_metrics(
+    play_url: str,
+    platform: str,
+    *,
+    headers: dict[str, str] | None = None,
+    http_client: Any | None = None,
+    ffmpeg_exe: str | None = None,
+    timeout_seconds: float = DOWNLOAD_TIMEOUT_SECONDS,
+    compress_timeout_seconds: float = COMPRESS_TIMEOUT_SECONDS,
+    clock: Any = time.monotonic,
+) -> tuple[str, dict[str, Any]]:
+    metrics: dict[str, Any] = {
+        "download_timeout_seconds": timeout_seconds,
+        "ffmpeg_timeout_seconds": compress_timeout_seconds,
+    }
+    total_started_at = clock()
     if not play_url:
-        raise VideoFetchError("missing play_url")
+        raise VideoFetchError("missing play_url", metrics=metrics)
     client = http_client or httpx
-    start = clock()
     try:
+        download_started_at = clock()
         video_path = _download_to_tempfile(
             client,
             play_url,
             platform,
             headers=headers,
             timeout_seconds=timeout_seconds,
-            started_at=start,
+            started_at=total_started_at,
             clock=clock,
         )
+        metrics["download_duration_ms"] = _elapsed_ms(download_started_at, clock())
     except httpx.HTTPError as exc:
-        raise VideoFetchError(f"download failed: {type(exc).__name__}") from exc
+        metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+        raise VideoFetchError(f"download failed: {type(exc).__name__}", metrics=metrics) from exc
+    except VideoFetchError as exc:
+        metrics.update(exc.metrics)
+        metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+        raise VideoFetchError(str(exc), metrics=metrics) from exc
     try:
-        if os.path.getsize(video_path) <= 0:
-            raise VideoFetchError("empty download")
+        downloaded_bytes = os.path.getsize(video_path)
+        metrics["downloaded_bytes"] = downloaded_bytes
+        if downloaded_bytes <= 0:
+            metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+            raise VideoFetchError("empty download", metrics=metrics)
+        ffmpeg_started_at = clock()
         compressed = _compress(
             video_path,
             ffmpeg_exe or imageio_ffmpeg.get_ffmpeg_exe(),
             timeout_seconds=compress_timeout_seconds,
             input_path=True,
         )
+        metrics["ffmpeg_duration_ms"] = _elapsed_ms(ffmpeg_started_at, clock())
+        metrics["compressed_bytes"] = len(compressed)
+    except VideoFetchError as exc:
+        metrics.update(exc.metrics)
+        metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+        raise VideoFetchError(str(exc), metrics=metrics) from exc
     finally:
         try:
             os.unlink(video_path)
         except OSError:
             pass
     if len(compressed) > MAX_INLINE_BYTES:
-        raise VideoFetchError(f"compressed video oversize: {len(compressed)} bytes")
-    return f"data:video/mp4;base64,{base64.b64encode(compressed).decode('ascii')}"
+        metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+        raise VideoFetchError(f"compressed video oversize: {len(compressed)} bytes", metrics=metrics)
+    metrics["total_duration_ms"] = _elapsed_ms(total_started_at, clock())
+    return f"data:video/mp4;base64,{base64.b64encode(compressed).decode('ascii')}", metrics
 
 
 def _download_to_tempfile(
@@ -176,3 +227,7 @@ def _unlink_quietly(path: str) -> None:
         os.unlink(path)
     except OSError:
         pass
+
+
+def _elapsed_ms(start: float, end: float) -> int:
+    return max(0, int((end - start) * 1000))

tests/test_gemini_video.py

@@ -47,14 +47,14 @@ def test_analyze_returns_v4_relevance_fields():
 
     result = _client(body).analyze(_ITEM, _MEDIA, _CTX)
 
-    assert result == {
-        "schema_version": "v4_gemini_query_relevance.v1",
-        "query_text": "中医养生",
-        "query_relevance_score": 83.0,
-        "query_relevance_reason": "贴切",
-        "final_status": "ok",
-        "retry_count": 0,
-    }
+    assert result["schema_version"] == "v4_gemini_query_relevance.v1"
+    assert result["query_text"] == "中医养生"
+    assert result["query_relevance_score"] == 83.0
+    assert result["query_relevance_reason"] == "贴切"
+    assert result["final_status"] == "ok"
+    assert result["retry_count"] == 0
+    assert result["timing_metrics"]["video_fetch"]["total_duration_ms"] >= 0
+    assert result["timing_metrics"]["gemini_request"]["attempts"][0]["status"] == "ok"
 
 
 def test_analyze_prompt_does_not_contain_legacy_50_plus_fields():

tests/test_oss_archive.py

@@ -63,6 +63,7 @@ def test_archive_due_records_updates_success_to_oss_uploaded():
     assert row["raw_payload"]["oss_archive_status"] == "uploaded"
     assert row["raw_payload"]["oss_archive_attempt_count"] == 1
     assert row["raw_payload"]["oss_object_key"] == "crawler/video/content_001.mp4"
+    assert row["raw_payload"]["oss_timing_metrics"]["oss_upload_duration_ms"] >= 0
 
 
 def test_archive_due_records_uploads_due_records_with_bounded_workers():
@@ -220,6 +221,7 @@ def test_archive_due_records_keeps_failed_attempt_pending_before_deadline():
     assert row["raw_payload"]["oss_archive_attempt_count"] == 1
     assert row["raw_payload"]["oss_archive_last_error"] == "oss_upload_http_error"
     assert row["raw_payload"]["oss_archive_next_retry_at"] == (NOW + timedelta(minutes=15)).isoformat()
+    assert row["raw_payload"]["oss_timing_metrics"]["oss_upload_duration_ms"] >= 0
 
 
 def test_archive_due_records_marks_failed_after_deadline():

tests/test_pattern_recall_archive.py

@@ -17,9 +17,14 @@ def test_pattern_recall_marks_all_judged_videos_for_oss_archive(tmp_path):
     ]
     media = [_media(run_id, policy_run_id, item["platform_content_id"]) for item in items]
     bundles = [_bundle(item) for item in items]
+    pool_result = fake_gemini_pool()
+    pool_result["timing_metrics"] = {
+        "video_fetch": {"download_duration_ms": 100, "ffmpeg_duration_ms": 200},
+        "gemini_request": {"total_duration_ms": 300},
+    }
     gemini = FakeGeminiVideoClient(
         result_by_content_id={
-            "content_pool": fake_gemini_pool(),
+            "content_pool": pool_result,
             "content_review": fake_gemini_review(),
             "content_failed": fake_gemini_fail("gemini_client_timeout"),
         }
@@ -46,6 +51,15 @@ def test_pattern_recall_marks_all_judged_videos_for_oss_archive(tmp_path):
     }
     for row in result["content_media_records"]:
         assert row["raw_payload"]["oss_archive_status"] == "pending"
+    media_by_id = {row["platform_content_id"]: row for row in result["content_media_records"]}
+    assert media_by_id["content_pool"]["raw_payload"]["gemini_timing_metrics"]["video_fetch"] == {
+        "download_duration_ms": 100,
+        "ffmpeg_duration_ms": 200,
+    }
+    evidence_by_id = {row["platform_content_id"]: row for row in result["pattern_recall_evidence"]}
+    assert evidence_by_id["content_pool"]["evidence_summary"]["timing_metrics"]["gemini_request"] == {
+        "total_duration_ms": 300,
+    }
 
 
 def test_pattern_recall_submits_oss_archive_before_gemini_judgment(tmp_path):

tests/test_video_fetch.py

@@ -65,9 +65,9 @@ def test_fetch_download_failure_raises():
 
 def test_fetch_enforces_total_download_timeout(monkeypatch):
     _patch_compress(monkeypatch)
-    ticks = iter([0.0, 31.0])
+    ticks = iter([0.0, 0.0, 31.0, 31.0])
 
-    with pytest.raises(VideoFetchError, match="download timeout"):
+    with pytest.raises(VideoFetchError, match="download timeout") as exc_info:
         fetch_and_compress(
             "http://v/x",
             "douyin",
@@ -76,6 +76,7 @@ def test_fetch_enforces_total_download_timeout(monkeypatch):
             timeout_seconds=30.0,
             clock=lambda: next(ticks),
         )
+    assert exc_info.value.metrics["total_duration_ms"] == 31000
 
 
 def test_fetch_oversize_raises(monkeypatch):