修复寻找agent超时问题

.env.example

@@ -9,6 +9,8 @@ OPENROUTER_TIMEOUT_SECONDS=120
 # Agent defaults
 AGENT_MAX_ITERATIONS=20
 AGENT_TEMPERATURE=0.7
+# 单个 find_agent 最长运行 30 分钟，超时后标记失败并继续下一条
+FIND_AGENT_TIMEOUT_SECONDS=1800
 
 # Skills directory (relative to project root or absolute path)
 SKILLS_DIR=skills
@@ -73,6 +75,7 @@ ALIYUN_OSS_ROOT_PREFIX=supply_agent
 # 手动上传日志专用 OSS 目录（与 Agent 自动上传 / MySQL oss_logs 路径分离）
 ALIYUN_OSS_MANUAL_LOG_PREFIX=supply_agent/manual_logs
 ALIYUN_OSS_PUBLIC_BASE_URL=http://rescdn.yishihui.com
+ALIYUN_OSS_CONNECT_TIMEOUT_SECONDS=30
 LOG_OSS_UPLOAD_ENABLED=true
 
 # AIGC platform (视频爬取/发布计划)

Dockerfile

@@ -30,6 +30,7 @@ ENV PYTHONUNBUFFERED=1 \
     SCHEDULER_CRON_MINUTE=0 \
     PIPELINE_WORKER_PROCESSES=4 \
     PIPELINE_MAX_ACTIVE_STEPS=4 \
+    FIND_AGENT_TIMEOUT_SECONDS=1800 \
     MYSQL_CONNECTION_BUDGET=40
 
 RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g; s|security.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \

README.md

@@ -150,8 +150,10 @@ async for event in agent.astream("Your question"):
 |------|------|--------|
 | `OPENROUTER_API_KEY` | OpenRouter API 密钥 | (必填) |
 | `OPENROUTER_MODEL` | 默认模型 | `google/gemini-2.5-flash` |
+| `OPENROUTER_TIMEOUT_SECONDS` | 单次模型请求超时秒数 | `120` |
 | `AGENT_MAX_ITERATIONS` | 最大循环次数 | `20` |
 | `AGENT_TEMPERATURE` | 生成温度 | `0.7` |
+| `FIND_AGENT_TIMEOUT_SECONDS` | 单个 find_agent 总运行超时秒数 | `1800` |
 | `SKILLS_DIR` | Skills 目录 | `skills` |
 | `LOGS_DIR` | Agent 运行日志目录 | `logs` |
 | `LOG_ENABLED` | 是否写入运行日志 | `true` |

agents/find_agent/__init__.py

@@ -30,7 +30,13 @@ def run_find_agent(
     此函数内部会走 agent.arun()，并在结束前关闭异步 HTTP 客户端。
     """
     agent = create_find_agent(settings=settings, model=model)
-    result = _run_coroutine(arun_find_agent(agent, user_input))
+    result = _run_coroutine(
+        arun_find_agent(
+            agent,
+            user_input,
+            timeout_seconds=agent.settings.find_agent_timeout_seconds,
+        )
+    )
     try:
         persist_model_recommendations(result)
     except Exception:

agents/find_agent/async_runner.py

@@ -11,26 +11,39 @@ from supply_agent.types import AgentResult
 logger = logging.getLogger(__name__)
 
 _thread_loop = threading.local()
+_ASYNC_CLEANUP_TIMEOUT_SECONDS = 5.0
 
 
 async def _drain_event_loop() -> None:
-    """等待 httpx/OpenAI 等异步客户端的收尾任务完成。"""
+    """Bound cleanup of httpx/OpenAI background tasks."""
     loop = asyncio.get_running_loop()
-    for _ in range(5):
-        pending = [
-            task
-            for task in asyncio.all_tasks(loop)
-            if task is not asyncio.current_task() and not task.done()
-        ]
-        if not pending:
-            break
-        results = await asyncio.gather(*pending, return_exceptions=True)
-        for result in results:
-            if isinstance(result, Exception) and not isinstance(
-                result, asyncio.CancelledError
-            ):
-                logger.debug("pending async task error during drain: %s", result)
-        await asyncio.sleep(0)
+    pending = {
+        task
+        for task in asyncio.all_tasks(loop)
+        if task is not asyncio.current_task() and not task.done()
+    }
+    if not pending:
+        return
+
+    done, still_pending = await asyncio.wait(
+        pending,
+        timeout=_ASYNC_CLEANUP_TIMEOUT_SECONDS,
+    )
+    for task in done:
+        if task.cancelled():
+            continue
+        error = task.exception()
+        if error is not None:
+            logger.debug("pending async task error during drain: %s", error)
+    if still_pending:
+        logger.warning(
+            "Cancelling %d async cleanup task(s) after %.0fs",
+            len(still_pending),
+            _ASYNC_CLEANUP_TIMEOUT_SECONDS,
+        )
+        for task in still_pending:
+            task.cancel()
+        await asyncio.wait(still_pending, timeout=1.0)
 
 
 async def _close_agent_async_resources(agent: Agent) -> None:
@@ -38,16 +51,37 @@ async def _close_agent_async_resources(agent: Agent) -> None:
     async_client = getattr(agent.llm, "_async_client", None)
     if async_client is not None:
         try:
-            await async_client.close()
+            await asyncio.wait_for(
+                async_client.close(),
+                timeout=_ASYNC_CLEANUP_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            logger.warning(
+                "close async llm client timed out after %.0fs",
+                _ASYNC_CLEANUP_TIMEOUT_SECONDS,
+            )
         except Exception:
             logger.debug("close async llm client failed", exc_info=True)
     await _drain_event_loop()
 
 
-async def arun_find_agent(agent: Agent, user_input: str) -> AgentResult:
+async def arun_find_agent(
+    agent: Agent,
+    user_input: str,
+    *,
+    timeout_seconds: float,
+) -> AgentResult:
     """在单个事件循环内运行 find_agent 并确保资源释放。"""
     try:
-        return await agent.arun(user_input)
+        return await asyncio.wait_for(
+            agent.arun(user_input),
+            timeout=timeout_seconds,
+        )
+    except TimeoutError:
+        logger.error("find_agent timed out after %.0fs", timeout_seconds)
+        raise TimeoutError(
+            f"find_agent timed out after {timeout_seconds:.0f}s"
+        ) from None
     finally:
         await _close_agent_async_resources(agent)
 

agents/find_agent/tools/qwen_video_analyze.py

@@ -22,7 +22,7 @@ from typing import Any, Optional
 
 import httpx
 from dotenv import load_dotenv
-from openai import APIStatusError, OpenAI
+from openai import APIStatusError, APITimeoutError, OpenAI
 
 from supply_agent.paths import find_project_root
 from supply_agent.tools import tool
@@ -39,6 +39,8 @@ DEFAULT_FPS = 2.0
 DEFAULT_TIMEOUT = 300.0
 DEFAULT_DOWNLOAD_TIMEOUT = 120.0
 DEFAULT_MAX_DURATION_SECONDS = 180.0
+FFPROBE_TIMEOUT_SECONDS = 30.0
+FFMPEG_TIMEOUT_SECONDS = 300.0
 
 _DURATION_RE = re.compile(
     r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)",
@@ -48,6 +50,10 @@ _DURATION_RE = re.compile(
 _env_loaded = False
 
 
+class ToolTimeoutError(RuntimeError):
+    """A bounded local tool operation exceeded its deadline."""
+
+
 def _ensure_env_loaded() -> None:
     """从项目根目录加载 .env，使 os.getenv 能读到其中的变量。"""
     global _env_loaded
@@ -109,35 +115,53 @@ def _parse_duration_text(text: str) -> float | None:
 def _probe_duration(ffmpeg: str, target: str) -> float | None:
     ffprobe = _resolve_ffprobe()
     if ffprobe:
-        result = subprocess.run(
-            [
-                ffprobe,
-                "-v",
-                "error",
-                "-show_entries",
-                "format=duration",
-                "-of",
-                "default=noprint_wrappers=1:nokey=1",
+        try:
+            result = subprocess.run(
+                [
+                    ffprobe,
+                    "-v",
+                    "error",
+                    "-show_entries",
+                    "format=duration",
+                    "-of",
+                    "default=noprint_wrappers=1:nokey=1",
+                    target,
+                ],
+                capture_output=True,
+                text=True,
+                check=False,
+                timeout=FFPROBE_TIMEOUT_SECONDS,
+            )
+        except subprocess.TimeoutExpired:
+            logger.warning(
+                "ffprobe timed out after %.0fs: %s",
+                FFPROBE_TIMEOUT_SECONDS,
                 target,
-            ],
+            )
+        else:
+            if result.returncode == 0:
+                raw = result.stdout.strip()
+                if raw:
+                    try:
+                        return float(raw)
+                    except ValueError:
+                        pass
+
+    try:
+        result = subprocess.run(
+            [ffmpeg, "-i", target],
             capture_output=True,
             text=True,
             check=False,
+            timeout=FFPROBE_TIMEOUT_SECONDS,
         )
-        if result.returncode == 0:
-            raw = result.stdout.strip()
-            if raw:
-                try:
-                    return float(raw)
-                except ValueError:
-                    pass
-
-    result = subprocess.run(
-        [ffmpeg, "-i", target],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
+    except subprocess.TimeoutExpired:
+        logger.warning(
+            "ffmpeg duration probe timed out after %.0fs: %s",
+            FFPROBE_TIMEOUT_SECONDS,
+            target,
+        )
+        return None
     return _parse_duration_text(result.stderr)
 
 
@@ -161,13 +185,31 @@ def _truncate_video(
         "+faststart",
         str(output_path),
     ]
-    result = subprocess.run(copy_cmd, capture_output=True, text=True, check=False)
-    if result.returncode == 0 and output_path.is_file() and output_path.stat().st_size > 0:
+    try:
+        result = subprocess.run(
+            copy_cmd,
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=FFMPEG_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired:
+        logger.warning(
+            "ffmpeg stream copy timed out after %.0fs",
+            FFMPEG_TIMEOUT_SECONDS,
+        )
+        result = None
+    if (
+        result is not None
+        and result.returncode == 0
+        and output_path.is_file()
+        and output_path.stat().st_size > 0
+    ):
         return
 
     logger.warning(
         "ffmpeg stream copy failed, fallback to re-encode: %s",
-        (result.stderr or result.stdout)[-500:],
+        ((result.stderr or result.stdout)[-500:] if result is not None else "timeout"),
     )
     encode_cmd = [
         ffmpeg,
@@ -186,12 +228,18 @@ def _truncate_video(
         "+faststart",
         str(output_path),
     ]
-    encode_result = subprocess.run(
-        encode_cmd,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
+    try:
+        encode_result = subprocess.run(
+            encode_cmd,
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=FFMPEG_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired:
+        raise ToolTimeoutError(
+            f"ffmpeg 截断视频超时（{FFMPEG_TIMEOUT_SECONDS:.0f}秒）"
+        ) from None
     if encode_result.returncode != 0 or not output_path.is_file():
         detail = (encode_result.stderr or encode_result.stdout)[-500:]
         raise RuntimeError(f"ffmpeg 截断视频失败: {detail}")
@@ -419,6 +467,7 @@ def verify_truncation_runtime(*, check_oss: bool = False) -> dict[str, Any]:
             check=True,
             capture_output=True,
             text=True,
+            timeout=FFMPEG_TIMEOUT_SECONDS,
         )
         _truncate_video(ffmpeg, source_path, clipped_path, 1.0)
         if not clipped_path.is_file() or clipped_path.stat().st_size <= 0:
@@ -505,9 +554,31 @@ async def qwen_video_analyze(
             extra=truncate_meta,
         )
 
+    except ToolTimeoutError as e:
+        logger.warning(
+            "qwen_video_analyze local operation timed out: video_url=%s error=%s",
+            video_url,
+            e,
+        )
+        return _error_result(
+            str(e),
+            error_code="tool_timeout",
+            retryable=True,
+        )
     except ValueError as e:
         logger.error("qwen_video_analyze config error: %s", e)
         return _error_result(str(e))
+    except httpx.TimeoutException as e:
+        logger.warning(
+            "qwen_video_analyze download timed out: video_url=%s error=%s",
+            video_url,
+            e,
+        )
+        return _error_result(
+            f"下载视频超时: {e}",
+            error_code="tool_timeout",
+            retryable=True,
+        )
     except httpx.HTTPError as e:
         logger.error(
             "qwen_video_analyze download error: video_url=%s error=%s",
@@ -516,6 +587,17 @@ async def qwen_video_analyze(
             exc_info=True,
         )
         return _error_result(f"下载视频失败: {e}")
+    except APITimeoutError as e:
+        logger.warning(
+            "qwen_video_analyze model request timed out: video_url=%s error=%s",
+            video_url,
+            e,
+        )
+        return _error_result(
+            f"视频模型请求超时: {e}",
+            error_code="tool_timeout",
+            retryable=True,
+        )
     except APIStatusError as e:
         error_code, title, message, retryable = _classify_api_error(e)
         logger.warning(

deploy/README.md

@@ -13,6 +13,8 @@ Scheduler、多个 Pipeline Worker 和 Reconciler。
 5. Docker 停止窗口至少 300 秒。
 6. 容器和 Alembic 连接 MySQL 后会将会话时区固定为 `+08:00`（中国标准时间）；定时日批有次日
    调度 deadline，手工/API/CLI 补数不设置 deadline。
+7. 单个 `find_agent` 默认最多运行 1800 秒；运行中的找片批次每 60 秒输出一次
+   进度日志，超时记录会标记失败，批次继续处理下一条需求。
 
 示例：
 

supply_agent/config.py

@@ -39,6 +39,12 @@ class Settings(BaseSettings):
     # Agent
     agent_max_iterations: int = Field(default=20, alias="AGENT_MAX_ITERATIONS")
     agent_temperature: float = Field(default=0.7, alias="AGENT_TEMPERATURE")
+    find_agent_timeout_seconds: float = Field(
+        default=1800.0,
+        ge=60.0,
+        le=7200.0,
+        alias="FIND_AGENT_TIMEOUT_SECONDS",
+    )
 
     # Skills
     skills_dir: Path = Field(default=Path("skills"), alias="SKILLS_DIR")

supply_infra/config.py

@@ -158,6 +158,12 @@ class InfraSettings(BaseSettings):
         default="http://rescdn.yishihui.com",
         alias="ALIYUN_OSS_PUBLIC_BASE_URL",
     )
+    aliyun_oss_connect_timeout_seconds: int = Field(
+        default=30,
+        ge=5,
+        le=300,
+        alias="ALIYUN_OSS_CONNECT_TIMEOUT_SECONDS",
+    )
     log_oss_upload_enabled: bool = Field(default=True, alias="LOG_OSS_UPLOAD_ENABLED")
 
     # AIGC platform

supply_infra/oss/client.py

@@ -29,7 +29,12 @@ class OssClient:
             self.settings.aliyun_oss_access_key_id,
             self.settings.aliyun_oss_access_key_secret,
         )
-        self._bucket = oss2.Bucket(auth, endpoint, self.settings.aliyun_oss_bucket)
+        self._bucket = oss2.Bucket(
+            auth,
+            endpoint,
+            self.settings.aliyun_oss_bucket,
+            connect_timeout=self.settings.aliyun_oss_connect_timeout_seconds,
+        )
         prefix = root_prefix if root_prefix is not None else self.settings.aliyun_oss_root_prefix
         self._root = prefix.strip("/")
         self._public_base = self.settings.aliyun_oss_public_base_url.rstrip("/")

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -7,8 +7,9 @@ from __future__ import annotations
 
 import logging
 import uuid
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
 from datetime import datetime
+from time import monotonic
 from typing import Any
 from zoneinfo import ZoneInfo
 
@@ -53,6 +54,13 @@ def process_single_discover(
 ) -> dict[str, Any]:
     """并发 worker：对单条需求记录执行 find_agent。"""
     summary = serialize_find_demand_context(ctx)
+    logger.info(
+        "discover videos started: grade_id=%s demand=%s videos=%d points=%d",
+        ctx.demand_grade_id,
+        ctx.demand_name,
+        ctx.video_count,
+        ctx.point_count,
+    )
     try:
         execution = discover_videos_for_demand(ctx, force=force)
         if execution.skipped:
@@ -216,37 +224,56 @@ def discover_videos_from_demands(
                 executor.submit(process_single_discover, ctx, force=force)
                 for ctx in batch
             ]
-            for future in as_completed(futures):
-                try:
-                    item_result = future.result()
-                except Exception as exc:
-                    logger.exception(
-                        "discover videos worker 出现未捕获错误: biz_dt=%s",
+            pending = set(futures)
+            batch_started = monotonic()
+            while pending:
+                completed, pending = wait(
+                    pending,
+                    timeout=60,
+                    return_when=FIRST_COMPLETED,
+                )
+                if not completed:
+                    logger.info(
+                        "discover videos still running: biz_dt=%s "
+                        "batch_pending=%d processed=%d elapsed_seconds=%d",
                         resolved_biz_dt,
+                        len(pending),
+                        result["processed"],
+                        int(monotonic() - batch_started),
                     )
-                    result["failed"] += 1
-                    result["processed"] += 1
-                    result["errors"].append({"error": str(exc)})
                     continue
 
-                result["processed"] += 1
-                if item_result.get("skipped"):
-                    result["skipped"] += 1
-                    continue
-                if item_result.get("success"):
-                    result["succeeded"] += 1
-                    continue
+                for future in completed:
+                    try:
+                        item_result = future.result()
+                    except Exception as exc:
+                        logger.exception(
+                            "discover videos worker 出现未捕获错误: biz_dt=%s",
+                            resolved_biz_dt,
+                        )
+                        result["failed"] += 1
+                        result["processed"] += 1
+                        result["errors"].append({"error": str(exc)})
+                        continue
 
-                result["failed"] += 1
-                result["errors"].append(
-                    {
-                        "demand_grade_id": item_result.get("demand_grade_id"),
-                        "demand_name": item_result.get("demand_name"),
-                        "video_count": item_result.get("video_count"),
-                        "run_id": item_result.get("run_id"),
-                        "error": item_result.get("error"),
-                    }
-                )
+                    result["processed"] += 1
+                    if item_result.get("skipped"):
+                        result["skipped"] += 1
+                        continue
+                    if item_result.get("success"):
+                        result["succeeded"] += 1
+                        continue
+
+                    result["failed"] += 1
+                    result["errors"].append(
+                        {
+                            "demand_grade_id": item_result.get("demand_grade_id"),
+                            "demand_name": item_result.get("demand_name"),
+                            "video_count": item_result.get("video_count"),
+                            "run_id": item_result.get("run_id"),
+                            "error": item_result.get("error"),
+                        }
+                    )
 
             passed_videos = _count_passed_videos(resolved_biz_dt)
             result["passed_videos"] = passed_videos

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -1,7 +1,22 @@
 from __future__ import annotations
 
-from unittest.mock import patch
+import asyncio
+import json
+import subprocess
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
 
+import pytest
+
+from agents.find_agent.async_runner import arun_find_agent
+from agents.find_agent.tools.qwen_video_analyze import (
+    FFMPEG_TIMEOUT_SECONDS,
+    FFPROBE_TIMEOUT_SECONDS,
+    ToolTimeoutError,
+    _probe_duration,
+    _truncate_video,
+    qwen_video_analyze,
+)
 from supply_infra.scheduler.jobs.discover_videos_from_demands import (
     discover_videos_from_demands,
 )
@@ -38,3 +53,95 @@ def test_stops_discovery_after_200_passed_videos(
     assert result["processed"] == 1
     assert result["passed_videos"] == 200
     assert result["stopped_by_passed_video_limit"] is True
+
+
+class _AsyncClient:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+class _SlowAgent:
+    def __init__(self) -> None:
+        self.llm = type("LLM", (), {"_async_client": _AsyncClient()})()
+
+    async def arun(self, _user_input: str) -> None:
+        await asyncio.sleep(60)
+
+
+@pytest.mark.asyncio
+async def test_find_agent_timeout_closes_async_client() -> None:
+    agent = _SlowAgent()
+
+    with pytest.raises(TimeoutError, match="find_agent timed out"):
+        await arun_find_agent(
+            agent,  # type: ignore[arg-type]
+            "test",
+            timeout_seconds=0.01,
+        )
+
+    assert agent.llm._async_client.closed is True
+
+
+@patch(
+    "agents.find_agent.tools.qwen_video_analyze._resolve_ffprobe",
+    return_value="/usr/bin/ffprobe",
+)
+@patch("agents.find_agent.tools.qwen_video_analyze.subprocess.run")
+def test_probe_duration_stops_after_subprocess_timeouts(
+    mock_run,
+    _mock_resolve_ffprobe,
+) -> None:
+    mock_run.side_effect = [
+        subprocess.TimeoutExpired("ffprobe", FFPROBE_TIMEOUT_SECONDS),
+        subprocess.TimeoutExpired("ffmpeg", FFPROBE_TIMEOUT_SECONDS),
+    ]
+
+    assert _probe_duration("/usr/bin/ffmpeg", "https://example.com/video.mp4") is None
+    assert [call.kwargs["timeout"] for call in mock_run.call_args_list] == [
+        FFPROBE_TIMEOUT_SECONDS,
+        FFPROBE_TIMEOUT_SECONDS,
+    ]
+
+
+@patch("agents.find_agent.tools.qwen_video_analyze.subprocess.run")
+def test_truncate_video_raises_after_bounded_ffmpeg_timeouts(
+    mock_run,
+    tmp_path: Path,
+) -> None:
+    mock_run.side_effect = [
+        subprocess.TimeoutExpired("ffmpeg-copy", FFMPEG_TIMEOUT_SECONDS),
+        subprocess.TimeoutExpired("ffmpeg-encode", FFMPEG_TIMEOUT_SECONDS),
+    ]
+
+    with pytest.raises(RuntimeError, match="ffmpeg 截断视频超时"):
+        _truncate_video(
+            "/usr/bin/ffmpeg",
+            tmp_path / "input.mp4",
+            tmp_path / "output.mp4",
+            180,
+        )
+
+    assert [call.kwargs["timeout"] for call in mock_run.call_args_list] == [
+        FFMPEG_TIMEOUT_SECONDS,
+        FFMPEG_TIMEOUT_SECONDS,
+    ]
+
+
+@pytest.mark.asyncio
+@patch(
+    "agents.find_agent.tools.qwen_video_analyze._prepare_analysis_url",
+    new_callable=AsyncMock,
+)
+async def test_qwen_tool_timeout_is_returned_to_agent(
+    mock_prepare,
+) -> None:
+    mock_prepare.side_effect = ToolTimeoutError("ffmpeg timeout")
+
+    result = json.loads(await qwen_video_analyze("https://example.com/video.mp4"))
+
+    assert result["error"] == "ffmpeg timeout"
+    assert result["error_code"] == "tool_timeout"
+    assert result["retryable"] is True