修改视频理解

find_agent_v2/qwen_video_understanding_30s.py

@@ -13,6 +13,7 @@ import signal
 import shutil
 import subprocess
 import tempfile
+import threading
 import time
 from pathlib import Path
 from typing import Any
@@ -40,11 +41,14 @@ CLIP_SECONDS = 30.0
 MIN_USABLE_CLIP_SECONDS = 15.0
 REMOTE_CLIP_TIMEOUT_SECONDS = 120.0
 REMOTE_CLIP_ATTEMPTS = 2
-UPLOAD_TIMEOUT_SECONDS = 60.0
+UPLOAD_TIMEOUT_SECONDS = 120.0
+OSS_UPLOAD_ATTEMPTS = 2
+OSS_UPLOAD_CONCURRENCY = 2
 MODEL_TIMEOUT_SECONDS = 300.0
 TOOL_TIMEOUT_SECONDS = 480.0
 PROBE_TIMEOUT_SECONDS = 15.0
 _DURATION_RE = re.compile(r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)")
+_OSS_UPLOAD_SLOTS = threading.BoundedSemaphore(OSS_UPLOAD_CONCURRENCY)
 
 
 class RemoteClipError(RuntimeError):
@@ -52,6 +56,7 @@ class RemoteClipError(RuntimeError):
 
 
 def _result(**payload: Any) -> str:
+    payload.setdefault("success", not bool(payload.get("error_code")))
     return json.dumps(payload, ensure_ascii=False)
 
 
@@ -166,6 +171,57 @@ def _upload_to_oss(clipped: Path, video_url: str) -> str:
     return client.upload_file(clipped, object_key)
 
 
+def _consume_background_upload(future: asyncio.Future) -> None:
+    """Consume a timed-out thread result so a later exception is not left unobserved."""
+    try:
+        future.exception()
+    except (asyncio.CancelledError, Exception):
+        pass
+
+
+async def _upload_attempt(clipped: Path, video_url: str) -> str:
+    """Run one upload with a timeout that starts only after a concurrency slot is acquired."""
+    loop = asyncio.get_running_loop()
+    started = asyncio.Event()
+
+    def upload_with_slot() -> str:
+        with _OSS_UPLOAD_SLOTS:
+            try:
+                loop.call_soon_threadsafe(started.set)
+            except RuntimeError:
+                pass
+            return _upload_to_oss(clipped, video_url)
+
+    future = loop.run_in_executor(None, upload_with_slot)
+    await started.wait()
+    try:
+        return await asyncio.wait_for(
+            asyncio.shield(future),
+            timeout=UPLOAD_TIMEOUT_SECONDS,
+        )
+    except TimeoutError:
+        # asyncio cannot stop a running upload thread. Keep the thread-owned
+        # semaphore until the SDK call really exits, then consume its result.
+        future.add_done_callback(_consume_background_upload)
+        raise TimeoutError(f"OSS 上传视频超时（{int(UPLOAD_TIMEOUT_SECONDS)} 秒）") from None
+
+
+async def _upload_to_oss_with_one_retry(clipped: Path, video_url: str) -> str:
+    errors: list[str] = []
+    for attempt in range(1, OSS_UPLOAD_ATTEMPTS + 1):
+        try:
+            return await _upload_attempt(clipped, video_url)
+        except Exception as exc:
+            errors.append(f"attempt {attempt}/{OSS_UPLOAD_ATTEMPTS}: {type(exc).__name__}: {exc}")
+            logger.warning("find_agent_v2 OSS upload %s", errors[-1])
+    last = errors[-1] if errors else "unknown error"
+    if "TimeoutError" in last:
+        raise TimeoutError(
+            f"OSS 上传视频连续 {OSS_UPLOAD_ATTEMPTS} 次超时（单次 {int(UPLOAD_TIMEOUT_SECONDS)} 秒）"
+        ) from None
+    raise RuntimeError("OSS 上传视频失败：" + " | ".join(errors))
+
+
 def _qwen_client() -> OpenAI:
     load_dotenv(find_project_root() / ".env")
     api_key = os.getenv("DASHSCOPE_API_KEY")
@@ -193,13 +249,7 @@ async def _prepare_oss_video(video_url: str) -> tuple[str, float, bool]:
     temp_dir = Path(tempfile.mkdtemp(prefix="find_agent_v2_video_"))
     try:
         clipped, duration, complete = await _clip_remote_with_one_retry(video_url, temp_dir)
-        try:
-            oss_url = await asyncio.wait_for(
-                asyncio.to_thread(_upload_to_oss, clipped, video_url),
-                timeout=UPLOAD_TIMEOUT_SECONDS,
-            )
-        except TimeoutError:
-            raise TimeoutError("OSS 上传视频超时（60 秒）") from None
+        oss_url = await _upload_to_oss_with_one_retry(clipped, video_url)
         return oss_url, duration, complete
     finally:
         for path in temp_dir.iterdir():
@@ -215,7 +265,7 @@ async def _prepare_oss_video(video_url: str) -> tuple[str, float, bool]:
     name="understand_candidate_video_30s_v2",
     description=(
         "理解当前评估分片中的候选视频。按 candidate_id 读取播放地址，由 ffmpeg 远程截取前 30 秒"
-        "（单次 60 秒超时，失败重试 1 次）；两次均不足 30 秒时只接受超过 15 秒的最长片段。上传 OSS"
+        "（单次 120 秒超时，失败重试 1 次）；两次均不足 30 秒时只接受超过 15 秒的最长片段。上传 OSS"
         "并清理全部本地文件后调用千问返回内容理解。"
         "仅用于核验实际主题、需求相关性、表达与场景、传播理由及视频中的时间线索；不能用画面人物"
         "年龄推断受众年龄。prompt 应写明当前需求和需要核验的疑点。"
@@ -265,17 +315,20 @@ async def understand_candidate_video_30s_v2(
             output=content,
             duration_ms=int((time.monotonic() - started) * 1000),
         )
-    except (RemoteClipError, TimeoutError) as exc:
+    except RemoteClipError as exc:
         return _result(
             error=str(exc),
-            error_code=(
-                "video_understanding_timeout"
-                if isinstance(exc, TimeoutError)
-                else "remote_clip_failed"
-            ),
+            error_code="remote_clip_failed",
             candidate_id=candidate_id,
             retryable=False,
         )
+    except TimeoutError as exc:
+        return _result(
+            error=str(exc),
+            error_code="video_understanding_timeout",
+            candidate_id=candidate_id,
+            retryable=True,
+        )
     except APIStatusError as exc:
         return _result(
             error=str(exc),

find_agent_v2/runtime.py

@@ -54,9 +54,23 @@ def _as_langchain_tool(fn: ToolFn) -> StructuredTool:
 def _message_dict(message: BaseMessage) -> dict[str, Any]:
     data = message.model_dump(mode="json")
     data["type"] = message.type
+    if isinstance(message, ToolMessage) and _tool_result_error_code(message.content):
+        data["status"] = "error"
     return data
 
 
+def _tool_result_error_code(content: Any) -> str:
+    if not isinstance(content, str):
+        return ""
+    try:
+        payload = json.loads(content)
+    except (TypeError, ValueError):
+        return ""
+    if not isinstance(payload, dict):
+        return ""
+    return str(payload.get("error_code") or "")
+
+
 def _usage(messages: list[BaseMessage]) -> dict[str, int | float]:
     totals: dict[str, int | float] = {
         "input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost": 0.0,
@@ -85,12 +99,14 @@ def _events(messages: list[BaseMessage]) -> list[dict[str, Any]]:
                 "usage": message.usage_metadata,
             })
         elif isinstance(message, ToolMessage):
+            error_code = _tool_result_error_code(message.content)
             events.append({
                 "type": "tool_call",
                 "name": message.name,
                 "tool_call_id": message.tool_call_id,
                 "result": message.content,
-                "status": message.status,
+                "status": "error" if error_code else message.status,
+                "error_code": error_code or None,
             })
     return events
 

tests/supply_agent/test_find_agent_v2.py

@@ -7,11 +7,16 @@ from types import SimpleNamespace
 
 import pytest
 from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
-from langchain_core.messages import AIMessage
+from langchain_core.messages import AIMessage, ToolMessage
 
 from find_agent_v2.graph import FindAgentRoundGraph
 from find_agent_v2.agent import decide_continued_exploration
-from find_agent_v2.runtime import DelegateArgs, FindAgentNodeHost
+from find_agent_v2.runtime import (
+    DelegateArgs,
+    FindAgentNodeHost,
+    _events,
+    _message_dict,
+)
 from find_agent_v2.demand_context import (
     V2DemandContext,
     V2ReferencePoint,
@@ -66,6 +71,22 @@ def _names(functions) -> set[str]:
     return {getattr(fn, "_tool_name", fn.__name__) for fn in functions}
 
 
+def test_video_tool_error_code_is_observed_as_failed() -> None:
+    message = ToolMessage(
+        content=json.dumps({
+            "success": False,
+            "error": "OSS timeout",
+            "error_code": "video_understanding_timeout",
+        }),
+        name="understand_candidate_video_30s_v2",
+        tool_call_id="call-1",
+    )
+
+    assert _message_dict(message)["status"] == "error"
+    assert _events([message])[0]["status"] == "error"
+    assert _events([message])[0]["error_code"] == "video_understanding_timeout"
+
+
 def _evaluation(**overrides):
     value = {
         "candidate_id": 11,

tests/supply_agent/test_find_agent_v2_video_understanding.py

@@ -2,6 +2,10 @@
 
 from __future__ import annotations
 
+import asyncio
+import json
+import threading
+import time
 from pathlib import Path
 
 import pytest
@@ -28,6 +32,9 @@ async def test_remote_clip_retries_once_and_prefers_complete_clip(monkeypatch, t
     assert duration == 30.0
     assert complete is True
     assert video_tool.REMOTE_CLIP_TIMEOUT_SECONDS == 120.0
+    assert video_tool.UPLOAD_TIMEOUT_SECONDS == 120.0
+    assert video_tool.OSS_UPLOAD_ATTEMPTS == 2
+    assert video_tool.OSS_UPLOAD_CONCURRENCY == 2
 
 
 @pytest.mark.asyncio
@@ -114,3 +121,77 @@ async def test_prepare_cleans_all_files_when_upload_fails(monkeypatch) -> None:
 
     assert not observed["selected"].exists()
     assert not observed["temp_dir"].exists()
+
+
+@pytest.mark.asyncio
+async def test_oss_upload_retries_once_after_timeout(monkeypatch, tmp_path) -> None:
+    attempts = 0
+    clip = tmp_path / "clip.mp4"
+    clip.write_bytes(b"clip")
+
+    async def upload_attempt(_clip: Path, _url: str) -> str:
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise TimeoutError("first timeout")
+        return "https://oss.example/clip.mp4"
+
+    monkeypatch.setattr(video_tool, "_upload_attempt", upload_attempt)
+
+    result = await video_tool._upload_to_oss_with_one_retry(clip, "video-url")
+
+    assert result == "https://oss.example/clip.mp4"
+    assert attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_oss_upload_concurrency_is_bounded(monkeypatch, tmp_path) -> None:
+    active = 0
+    maximum = 0
+    lock = threading.Lock()
+    clip = tmp_path / "clip.mp4"
+    clip.write_bytes(b"clip")
+
+    def upload(_clip: Path, video_url: str) -> str:
+        nonlocal active, maximum
+        with lock:
+            active += 1
+            maximum = max(maximum, active)
+        try:
+            time.sleep(0.05)
+            return f"https://oss.example/{video_url}.mp4"
+        finally:
+            with lock:
+                active -= 1
+
+    monkeypatch.setattr(video_tool, "_upload_to_oss", upload)
+
+    results = await asyncio.gather(*(
+        video_tool._upload_attempt(clip, f"video-{index}")
+        for index in range(4)
+    ))
+
+    assert len(results) == 4
+    assert maximum <= video_tool.OSS_UPLOAD_CONCURRENCY
+
+
+@pytest.mark.asyncio
+async def test_video_understanding_timeout_is_retryable(monkeypatch) -> None:
+    class Service:
+        @staticmethod
+        def candidate_input(_run_id: str, _candidate_id: int):
+            return {"video_url": "https://example.test/video.mp4"}
+
+    async def timeout(_url: str):
+        raise TimeoutError("OSS upload timeout")
+
+    monkeypatch.setattr(video_tool, "get_find_agent_v2_service", lambda: Service())
+    monkeypatch.setattr(video_tool, "_prepare_oss_video", timeout)
+
+    result = json.loads(await video_tool.understand_candidate_video_30s_v2(
+        run_id="run", candidate_id=1, prompt="check",
+    ))
+
+    assert result["success"] is False
+    assert result["error_code"] == "video_understanding_timeout"
+    assert result["retryable"] is True