fix(ai-gateway): 修复 ASR 降级链断裂与 Redis 配置未生效

FallbackProvider.transcribe 缺少 **kwargs，call_with_fallback 透传的
_feature 撞到签名抛 TypeError，导致降级链末端失败、ASR 请求返回 503
而非友好降级响应（已补两条回归测试锁定）。

get_cache 以无参构造 RedisCache，忽略 REDIS_URL 环境变量，容器部署下
恒连 localhost 失败，限流与响应缓存自部署起从未生效；main.py 原先无论
成败都打印'连接已建立'掩盖了该故障。

其余：transcribe 限流路径由 /api/v1/ai/ 更正为 /api/v1/asr/，配额
30->600 并豁免全局每日总量（段级高频调用）；开启 Qwen ASR 的 ITN；
上游 429 不再误报为本地配额耗尽。

Author: Aparencia

server/ai-gateway/cache/redis_cache.py

@@ -156,8 +156,15 @@ async def set_ai_cache(
 
 
 def get_cache() -> RedisCache:
-    """获取全局缓存实例"""
+    """
+    获取全局缓存实例
+
+    @ai-context: 必须用 APP_CONFIG["redis_url"]（来自环境变量 REDIS_URL）构造，
+    容器部署时 Redis 位于独立容器（redis:6379，带密码），若用无参默认值
+    redis://localhost:6379/0 会永远连接失败，导致限流与响应缓存静默失效。
+    """
     global _cache_instance
     if _cache_instance is None:
-        _cache_instance = RedisCache()
+        from config import APP_CONFIG
+        _cache_instance = RedisCache(APP_CONFIG["redis_url"])
     return _cache_instance

server/ai-gateway/config/limits.py

@@ -49,7 +49,9 @@
     "evaluate": 10,
     "recommend": 15,
     "vision_extract": 20,
-    "transcribe": 30,
+    # 课堂实时转录为段级高频调用（VAD 每 5-30s 产生一段，一节课数百段），
+    # 对齐主流 ASR 按时长计费模式放宽次数限制；中间件对其豁免全局每日总量
+    "transcribe": 600,
     "tag_content": 30,
     "optimize_card": 15,
     "feynman_question": 15,

server/ai-gateway/errors.py

@@ -29,11 +29,21 @@ def __init__(self, provider: str, reason: str = ""):
 
 
 class RateLimitExceededError(AIError):
-    """频率限制超限"""
+    """
+    频率限制超限
+
+    @ai-context: limit=0 表示上游服务商返回 429（其配额/并发受限），
+    而非本项目的每日配额耗尽——两者文案必须区分，否则日志会出现
+    "已达上限（0 次）"这类自相矛盾的误导信息。
+    """
 
     def __init__(self, feature: str, limit: int):
+        if limit > 0:
+            message = f"今日 {feature} 功能使用次数已达上限（{limit} 次），请明天再试"
+        else:
+            message = f"{feature} 服务商当前访问量过大或配额受限（上游 429），请稍后重试"
         super().__init__(
-            message=f"今日 {feature} 功能使用次数已达上限（{limit} 次），请明天再试",
+            message=message,
             status_code=429,
             detail={"feature": feature, "limit": limit},
         )

server/ai-gateway/main.py

@@ -69,7 +69,15 @@ async def lifespan(app: FastAPI):
     # 初始化 Redis 连接
     cache = get_cache()
     await cache.connect()
-    logger.info("Redis 连接已建立")
+    # connect 内部失败时会降级（_client=None），此处按实际结果记录，
+    # 避免"连接已建立"的误导日志掩盖限流/缓存静默失效
+    if cache._client is not None:
+        logger.info("Redis 连接已建立")
+    else:
+        logger.error(
+            "Redis 连接失败，限流与响应缓存将全部降级失效（限流放行）。"
+            "请检查 REDIS_URL 环境变量与 redis 容器状态"
+        )
 
     # 初始化各 Provider 并检查 API Key 配置
     init_providers(app)

server/ai-gateway/middleware/rate_limit.py

@@ -35,13 +35,17 @@
     "/api/v1/ai/evaluate-explanation": "evaluate",
     "/api/v1/ai/recommend-duration": "recommend",
     "/api/v1/ai/vision": "vision_extract",
-    "/api/v1/ai/transcribe": "transcribe",
+    "/api/v1/asr/transcribe": "transcribe",
     "/api/v1/ai/tag-content": "tag_content",
     "/api/v1/ai/optimize-card": "optimize_card",
     "/api/v1/ai/feynman-question": "feynman_question",
     "/api/v1/ai/feynman-evaluate-answers": "feynman_evaluate",
 }
 
+# 豁免全局每日总量的功能：段级高频调用（如课堂实时转录一节课数百段），
+# 若计入 daily_total 会在几分钟内耗尽全部 AI 配额，仅受各自功能级上限约束
+GLOBAL_EXEMPT_FEATURES: frozenset[str] = frozenset({"transcribe"})
+
 
 class RateLimitMiddleware(BaseHTTPMiddleware):
     """频率限制中间件 — 基于 Redis 滑动窗口"""
@@ -183,6 +187,10 @@ async def _check_rate_limit(
                 "请明天再试，或升级套餐获取更多配额。"
             )
 
+        # 段级高频功能豁免全局每日总量（仅受功能级上限约束）
+        if feature in GLOBAL_EXEMPT_FEATURES:
+            return True, ""
+
         # ---- 第二层：全局每日总量限制（预检查，仅读取） ----
         global_key = f"rate_limit:{user_id}:global:{today}"
         try:
@@ -236,6 +244,10 @@ async def _increment_rate_limit(self, user_id: str, feature: str) -> None:
         except Exception as exc:
             logger.warning("频率限制计数失败(功能级): %s", exc)
 
+        # 段级高频功能不计入全局每日总量
+        if feature in GLOBAL_EXEMPT_FEATURES:
+            return
+
         # ---- 第二层：全局每日总量计数 ----
         global_key = f"rate_limit:{user_id}:global:{today}"
         try:

server/ai-gateway/providers/fallback_provider.py

@@ -178,8 +178,16 @@ async def transcribe(
         sample_rate: int = 16000,
         channels: int = 1,
         model: str = "",
+        **kwargs: Any,
     ) -> dict[str, Any]:
-        """Fallback 不支持 ASR，返回友好降级结果"""
+        """
+        Fallback 不支持 ASR，返回友好降级结果
+
+        @ai-context: 必须接受 **kwargs —— 云端 Provider 的 transcribe 由
+        with_retry_and_timeout 装饰并在内部吞掉 _feature，而本方法无装饰器，
+        call_with_fallback 传入的 _feature 会直接透传到签名上。缺少 **kwargs
+        会抛 TypeError 使 fallback 链最后一环也失败，请求最终返回 503 而非降级响应。
+        """
         import time as _time
         start = _time.monotonic()
         logger.warning("使用 FallbackProvider 降级 ASR 响应")

server/ai-gateway/providers/qwen_provider.py

@@ -120,7 +120,9 @@ async def transcribe(
         try:
             # 音频以 Data URL 内嵌（客户端上送 WAV/PCM base64）
             data_uri = f"data:audio/wav;base64,{audio_base64}"
-            asr_options: dict[str, Any] = {"enable_itn": False}
+            # ITN 开启：数字/单位规范化（"三点一四"→"3.14"），对齐主流 ASR 默认行为，
+            # 课堂场景公式/数据密集，规范化文本对笔记质量至关重要
+            asr_options: dict[str, Any] = {"enable_itn": True}
             if language != "auto":
                 asr_options["language"] = language
 

server/ai-gateway/tests/test_providers.py

@@ -183,6 +183,66 @@ def test_recommend_duration_fallback_no_history(self):
         assert result["recommended_minutes"] == 25
         assert result["source"] == "local_rule"
 
+    @pytest.mark.asyncio
+    async def test_transcribe_accepts_feature_kwarg(self):
+        """
+        回归：transcribe 必须容忍 call_with_fallback 透传的 _feature 关键字。
+
+        云端 Provider 的 transcribe 由 with_retry_and_timeout 装饰并吞掉 _feature，
+        但 FallbackProvider 无装饰器，若签名不接受 **kwargs 会抛 TypeError，
+        导致 fallback 链最后一环失败、ASR 请求返回 503 而非降级响应。
+        """
+        # Arrange
+        provider = FallbackProvider()
+
+        # Act
+        result = await provider.transcribe(
+            audio_base64="ZmFrZQ==", language="zh", _feature="transcribe",
+        )
+
+        # Assert
+        assert result["model"] == "fallback"
+        assert result["text"] == ""
+        assert result["warning"]  # 必须透传降级提示供客户端识别失败
+
+    @pytest.mark.asyncio
+    async def test_transcribe_fallback_reachable_when_cloud_asr_fails(self):
+        """
+        回归：云端 ASR 全失败时，fallback 链末端必须能返回降级响应。
+
+        真实故障场景——Qwen 404 + GLM 400 后，若 FallbackProvider.transcribe
+        因签名不接受 _feature 抛 TypeError，整个请求会变成 503。
+        """
+        # Arrange：模拟 qwen/glm 的 transcribe 均失败，链末端为真实 FallbackProvider
+        class FailingASRProvider:
+            def __init__(self, name):
+                self.provider_name = name
+
+            async def transcribe(self, *args, **kwargs):
+                raise RuntimeError(f"{self.provider_name} ASR 上游故障")
+
+        app = MagicMock()
+        app.state.providers = {
+            "qwen": FailingASRProvider("qwen"),
+            "glm": FailingASRProvider("glm"),
+            "fallback": FallbackProvider(),
+        }
+
+        async def _run_chain(provider, model_name):
+            # 与 transcribe 路由一致：显式透传 _feature
+            return await provider.transcribe(
+                audio_base64="ZmFrZQ==", language="zh", model=model_name,
+                _feature="transcribe",
+            )
+
+        # Act
+        result, provider_key = await call_with_fallback(app, "transcribe", _run_chain)
+
+        # Assert
+        assert provider_key == "fallback"
+        assert result["model"] == "fallback"
+        assert result["warning"]
+
     def test_recommend_duration_fallback_short_history(self):
         history = [
             {"duration_minutes": 10, "completed": True},