总结层篇幅预算治理：prompt 收敛 + max_tokens + 压缩重试 + 分带监控

config/config.example.jsonc

@@ -215,6 +215,22 @@
         "min_summary_threshold": 500,    // 文本长度低于此值时跳过总结
         "min_calibrate_ratio": 0.8,      // 校对后文本长度/原文长度的最小比例，低于则视为异常
 
+        // ------------------------------------------------------
+        // 总结篇幅预算（prompt / max_tokens / 后验硬顶共用同一曲线）
+        // ------------------------------------------------------
+        "summary_budget": {
+            "s_target_min": 500,
+            "s_target_max": 3000,
+            "s_hard_cap_max": 4500,
+            "m_target_min": 2000,
+            "m_target_max": 4000,
+            "m_hard_cap": 5000,
+            "l_target_min": 4000,
+            "l_target_max": 6000,
+            "l_hard_cap": 8000,
+            "max_tokens_multiplier": 1.5
+        },
+
         // ------------------------------------------------------
         // 内容审查降级（llm-compat content_fallbacks）
         // 当主模型因内容审查拒绝时，自动切换到 fallback 模型

src/video_transcript_api/api/routes/audit.py

@@ -6,19 +6,23 @@
 
 import asyncio
 import sqlite3
+from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 
 from ..context import (
     get_audit_logger,
     get_cache_manager,
+    get_config,
     get_logger,
     get_usage_recorder,
     get_user_manager,
     lazy_resource,
 )
+from ..services.summary_ratio_stats import compute_summary_ratio_stats
 from ..services.transcription import TranscribeResponse, verify_token
+from ...llm.core.summary_budget import SummaryBudgetConfig
 from ..services.view_token_resolver import ViewTokenResolver
 from ...utils.llm_status import SummaryStatus
 
@@ -166,6 +170,47 @@ async def get_audit_stats(days: int = 30, user_info: dict = Depends(verify_token
         raise HTTPException(status_code=500, detail=f"获取统计信息失败: {exc}")
 
 
+@router.get("/summary-ratio")
+async def get_summary_ratio_stats(
+    days: int = Query(30, ge=1, le=365, description="统计时间窗口（天）"),
+    user_info: dict = Depends(verify_token),
+):
+    """按 S/M/L 分带返回总结相对原文的长度比例监控指标。"""
+    can_view_global = (
+        not user_manager.is_multi_user_mode() or user_info.get("is_legacy", False)
+    )
+    if not can_view_global:
+        raise HTTPException(status_code=403, detail="无权访问全局总结比例统计")
+
+    try:
+        app_config = get_config()
+        storage = app_config.get("storage", {})
+        audit_db_path = storage.get("audit_db")
+        cache_manager = get_cache_manager()
+        cache_db_path = str(cache_manager.db_path)
+        cache_root = Path(cache_manager.cache_dir)
+        budget_config = SummaryBudgetConfig.from_dict(
+            (app_config.get("llm") or {}).get("summary_budget")
+        )
+
+        data = await asyncio.to_thread(
+            compute_summary_ratio_stats,
+            audit_db_path=audit_db_path,
+            cache_db_path=cache_db_path,
+            cache_root=cache_root,
+            days=days,
+            budget_config=budget_config,
+        )
+        return TranscribeResponse(
+            code=200,
+            message="获取总结比例统计成功",
+            data=data,
+        )
+    except Exception as exc:
+        logger.exception("获取总结比例统计异常: %s", exc)
+        raise HTTPException(status_code=500, detail=f"获取总结比例统计失败: {exc}")
+
+
 @router.get("/calls")
 async def get_audit_calls(
     limit: int = Query(100, ge=1, le=10000, description="返回记录数量限制"),

src/video_transcript_api/api/services/summary_ratio_stats.py

@@ -0,0 +1,229 @@
+"""Compute summary-to-original length ratio stats for audit monitoring."""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from ...transcriber import FunASRSpeakerClient
+from ...utils.logging import setup_logger
+from ...llm.core.summary_budget import (
+    classify_original_length_band,
+    compute_summary_budget,
+    SummaryBudgetConfig,
+)
+
+logger = setup_logger(__name__)
+
+_BAND_ORDER = ("S", "M", "L")
+
+
+def _median(values: List[float]) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    mid = len(ordered) // 2
+    if len(ordered) % 2:
+        return ordered[mid]
+    return (ordered[mid - 1] + ordered[mid]) / 2.0
+
+
+def _p90(values: List[float]) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = int(0.9 * (len(ordered) - 1))
+    return ordered[index]
+
+
+def _read_original_length(cache_dir: Path) -> Optional[int]:
+    funasr_file = cache_dir / "transcript_funasr.json"
+    capswriter_file = cache_dir / "transcript_capswriter.txt"
+    try:
+        if funasr_file.exists():
+            with funasr_file.open("r", encoding="utf-8") as handle:
+                funasr_data = json.load(handle)
+            client = FunASRSpeakerClient()
+            transcript_text = client.format_transcript_with_speakers(funasr_data)
+            return len(transcript_text)
+        if capswriter_file.exists():
+            with capswriter_file.open("r", encoding="utf-8") as handle:
+                return len(handle.read())
+    except Exception as exc:
+        logger.warning(f"Failed to read original transcript length from {cache_dir}: {exc}")
+    return None
+
+
+def _read_summary_length(cache_dir: Path) -> Optional[int]:
+    summary_file = cache_dir / "llm_summary.txt"
+    if not summary_file.exists():
+        return None
+    try:
+        with summary_file.open("r", encoding="utf-8") as handle:
+            return len(handle.read())
+    except OSError as exc:
+        logger.warning(f"Failed to read summary length from {cache_dir}: {exc}")
+        return None
+
+
+def _resolve_cache_dir(
+    cache_conn: sqlite3.Connection,
+    cache_root: Path,
+    *,
+    cache_id: Optional[int],
+    platform: str,
+    media_id: str,
+    use_speaker_recognition: bool,
+) -> Optional[Path]:
+    """Resolve artifact directory for the task's actual cache variant."""
+    cache_root_resolved = cache_root.resolve()
+    cache_row = None
+
+    if cache_id is not None:
+        cache_row = cache_conn.execute(
+            "SELECT files_loc FROM video_cache WHERE id = ?",
+            (cache_id,),
+        ).fetchone()
+
+    if not cache_row:
+        cache_row = cache_conn.execute(
+            """
+            SELECT files_loc FROM video_cache
+            WHERE platform = ? AND media_id = ? AND use_speaker_recognition = ?
+            ORDER BY updated_at DESC
+            LIMIT 1
+            """,
+            (platform, media_id, 1 if use_speaker_recognition else 0),
+        ).fetchone()
+
+    if not cache_row:
+        return None
+
+    cache_dir = (cache_root / Path(cache_row["files_loc"])).resolve()
+    if not cache_dir.is_relative_to(cache_root_resolved):
+        logger.warning(
+            f"files_loc escapes cache_root, skipping: {cache_row['files_loc']}"
+        )
+        return None
+    return cache_dir
+
+
+def _aggregate_band(ratios: List[float], over_100: int, over_hardcap: int) -> Dict[str, Any]:
+    return {
+        "n": len(ratios),
+        "median_ratio": round(_median(ratios), 4) if ratios else 0.0,
+        "p90_ratio": round(_p90(ratios), 4) if ratios else 0.0,
+        "over_100pct": over_100,
+        "over_hardcap": over_hardcap,
+    }
+
+
+def compute_summary_ratio_stats(
+    *,
+    audit_db_path: str,
+    cache_db_path: str,
+    cache_root: Path,
+    days: int,
+    budget_config: Optional[SummaryBudgetConfig] = None,
+) -> Dict[str, Any]:
+    """Join audit snapshots to cache artifacts and compute S/M/L band ratios."""
+    budget_config = budget_config or SummaryBudgetConfig()
+    cutoff = (datetime.now() - timedelta(days=max(1, days))).strftime("%Y-%m-%d %H:%M:%S")
+
+    audit_conn = sqlite3.connect(f"file:{audit_db_path}?mode=ro", uri=True)
+    audit_conn.row_factory = sqlite3.Row
+    cache_conn = sqlite3.connect(f"file:{cache_db_path}?mode=ro", uri=True)
+    cache_conn.row_factory = sqlite3.Row
+
+    try:
+        snapshot_rows = audit_conn.execute(
+            """
+            SELECT task_id, platform, summary_status, archived_at
+            FROM task_audit_snapshots
+            WHERE summary_status = 'generated'
+              AND status = 'success'
+              AND COALESCE(content_expired, 0) = 0
+              AND archived_at >= ?
+            """,
+            (cutoff,),
+        ).fetchall()
+
+        band_ratios: Dict[str, List[float]] = {band: [] for band in _BAND_ORDER}
+        band_over_100: Dict[str, int] = {band: 0 for band in _BAND_ORDER}
+        band_over_hardcap: Dict[str, int] = {band: 0 for band in _BAND_ORDER}
+        skipped = 0
+
+        for row in snapshot_rows:
+            task_id = row["task_id"]
+            platform = row["platform"]
+            task_row = cache_conn.execute(
+                """
+                SELECT platform, media_id, use_speaker_recognition, cache_id
+                FROM task_status
+                WHERE task_id = ?
+                """,
+                (task_id,),
+            ).fetchone()
+            if not task_row:
+                skipped += 1
+                continue
+
+            resolved_platform = task_row["platform"] or platform
+            media_id = task_row["media_id"]
+            if not resolved_platform or not media_id:
+                skipped += 1
+                continue
+
+            use_speaker = bool(task_row["use_speaker_recognition"])
+            cache_dir = _resolve_cache_dir(
+                cache_conn,
+                cache_root,
+                cache_id=task_row["cache_id"],
+                platform=resolved_platform,
+                media_id=media_id,
+                use_speaker_recognition=use_speaker,
+            )
+            if not cache_dir:
+                skipped += 1
+                continue
+
+            original_length = _read_original_length(cache_dir)
+            summary_length = _read_summary_length(cache_dir)
+            if original_length is None or summary_length is None or original_length <= 0:
+                skipped += 1
+                continue
+
+            band = classify_original_length_band(original_length)
+            if band not in band_ratios:
+                skipped += 1
+                continue
+
+            ratio = summary_length / original_length
+            band_ratios[band].append(ratio)
+            if ratio > 1.0:
+                band_over_100[band] += 1
+            hard_cap = compute_summary_budget(original_length, budget_config).hard_cap
+            if summary_length > hard_cap:
+                band_over_hardcap[band] += 1
+
+        bands = {
+            band: _aggregate_band(
+                band_ratios[band],
+                band_over_100[band],
+                band_over_hardcap[band],
+            )
+            for band in _BAND_ORDER
+        }
+        return {
+            "days": days,
+            "cutoff": cutoff,
+            "bands": bands,
+            "skipped_tasks": skipped,
+            "sampled_tasks": sum(bands[b]["n"] for b in _BAND_ORDER),
+        }
+    finally:
+        audit_conn.close()
+        cache_conn.close()

src/video_transcript_api/llm/core/config.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, Union
 
+from .summary_budget import SummaryBudgetConfig
+
 
 @dataclass
 class LLMConfig:
@@ -129,6 +131,9 @@ class LLMConfig:
     notes_reasoning_effort: Optional[str] = None
     notes_concurrency: int = 10
 
+    # Summary output budget (prompt + max_tokens + post-hoc cap share one curve).
+    summary_budget: SummaryBudgetConfig = field(default_factory=SummaryBudgetConfig)
+
     @classmethod
     def from_dict(cls, config_dict: dict) -> "LLMConfig":
         """从配置字典创建 LLMConfig 实例
@@ -321,6 +326,9 @@ def from_dict(cls, config_dict: dict) -> "LLMConfig":
                 )
             ),
             notes_concurrency=llm_config.get("notes_concurrency", 10),
+            summary_budget=SummaryBudgetConfig.from_dict(
+                llm_config.get("summary_budget")
+            ),
         )
 
     def get_models(self) -> dict:

src/video_transcript_api/llm/core/llm_client.py

@@ -52,6 +52,7 @@ def call(
         reasoning_effort: Optional[str] = None,
         task_type: str = "unknown",
         force_json_mode: Optional[str] = None,
+        max_tokens: Optional[int] = None,
     ) -> LLMResponse:
         """调用 LLM API
 
@@ -93,6 +94,7 @@ def call(
                 system_prompt=system_prompt,
                 config=self.config,
                 force_json_mode=force_json_mode,
+                max_tokens=max_tokens,
             )
 
             if isinstance(result, StructuredResult):