fix(ai): NDJSON 流式接口空闲心跳保活 + LLM 首包前重试

根因: 分析/复盘端点 yield meta 后等 LLM 首包期间流上零字节可达数分钟,
反向代理按空闲超时切断连接, 前端报裸 network error; 且 max_retries=0
使上游(DeepSeek)高峰过载时一次失败即终止。

- 新增 services/ndjson_heartbeat.with_heartbeat: 空闲超 15s 插入
  {"type":"ping"} 行保活(间隔与 mining.py SSE 心跳一致)
- 5 个流式端点接入: 复盘/个股/财务/轮动/策略生成; 策略生成端点补齐
  Cache-Control 与 X-Accel-Buffering 头
- AsyncOpenAI max_retries 0→2: SDK 重试仅覆盖首包前, 未产出内容, 安全
- 前端事件类型联合加入 'ping'(下游对未知类型静默忽略); 新增
  friendlyStreamError 把浏览器裸网络错误翻译为可操作提示, 6 处落点应用
- 新增 test_ndjson_heartbeat.py 覆盖心跳插入/异常透传/断连清理/重试配置

兼容性: ping 为新增事件类型, 旧前端与 daily_pipeline 消费方静默跳过,
前后端可独立发布。

backend/app/api/financials.py

@@ -11,6 +11,7 @@ from pydantic import BaseModel
 from app.services.financial_sync import FINANCIAL_TABLES, get_financial_df
 from app.services.financial_analyzer import analyze_financials_stream
 from app.services import ai_reports
+from app.services.ndjson_heartbeat import with_heartbeat
 from app.tickflow.capabilities import Cap
 
 logger = logging.getLogger(__name__)
@@ -188,7 +189,7 @@ async def analyze_financials(request: Request, req: AnalyzeRequest):
     data_dir = request.app.state.repo.store.data_dir
 
     async def stream_gen():
-        async for chunk in analyze_financials_stream(data_dir, req.symbol, req.focus):
+        async for chunk in with_heartbeat(analyze_financials_stream(data_dir, req.symbol, req.focus)):
             yield chunk + "\n"
 
     return StreamingResponse(

backend/app/api/market_recap.py

@@ -21,6 +21,7 @@ from pydantic import BaseModel
 
 from app.services import auction_benchmark, dragon_tiger, market_recap_reports, preferences
 from app.services.market_recap import recap_market_stream
+from app.services.ndjson_heartbeat import with_heartbeat
 
 logger = logging.getLogger(__name__)
 
@@ -100,7 +101,9 @@ async def analyze_market(request: Request, req: AnalyzeRequest):
             raise HTTPException(400, f"as_of 格式应为 YYYY-MM-DD,收到: {req.as_of}")
 
     async def stream_gen():
-        async for chunk in recap_market_stream(repo, quote_service, depth_service, as_of, req.focus):
+        async for chunk in with_heartbeat(
+            recap_market_stream(repo, quote_service, depth_service, as_of, req.focus),
+        ):
             yield chunk + "\n"
 
     return StreamingResponse(

backend/app/api/rps.py

@@ -11,6 +11,7 @@ from pydantic import BaseModel
 
 from app.services import rps_rotation
 from app.services.concept_rotation_analyzer import analyze_rotation_stream
+from app.services.ndjson_heartbeat import with_heartbeat
 
 router = APIRouter(prefix="/api/rps", tags=["rps"])
 
@@ -61,8 +62,10 @@ async def analyze_rotation(request: Request, req: AnalyzeRequest):
     level = req.level if (kind == "industry" and req.level in (1, 2, 3)) else None
 
     async def stream_gen():
-        async for chunk in analyze_rotation_stream(
-            repo, days, req.focus, quote_service, depth_service, kind, level,
+        async for chunk in with_heartbeat(
+            analyze_rotation_stream(
+                repo, days, req.focus, quote_service, depth_service, kind, level,
+            ),
         ):
             yield chunk + "\n"
 

backend/app/api/stock_analysis.py

@@ -22,6 +22,7 @@ from pydantic import BaseModel
 
 from app.indicators.levels import compute_levels, summarize_levels
 from app.services import stock_reports
+from app.services.ndjson_heartbeat import with_heartbeat
 from app.services.stock_analyzer import analyze_stock_stream
 
 logger = logging.getLogger(__name__)
@@ -165,7 +166,7 @@ async def analyze_stock(request: Request, req: AnalyzeRequest):
     data_dir = repo.store.data_dir
 
     async def stream_gen():
-        async for chunk in analyze_stock_stream(repo, data_dir, req.symbol, req.focus):
+        async for chunk in with_heartbeat(analyze_stock_stream(repo, data_dir, req.symbol, req.focus)):
             yield chunk + "\n"
 
     return StreamingResponse(

backend/app/api/strategy.py

@@ -29,6 +29,7 @@ from app.strategy.scoring import (
     effective_scoring,
     effective_scoring_directions,
 )
+from app.services.ndjson_heartbeat import with_heartbeat
 
 router = APIRouter(prefix="/api/strategies", tags=["strategies"])
 logger = logging.getLogger(__name__)
@@ -936,7 +937,11 @@ async def build_strategy_stream(req: BuildRequest, request: Request):
         except Exception as e:
             yield json.dumps({"type": "error", "message": f"AI生成失败: {e}"}, ensure_ascii=False) + "\n"
 
-    return StreamingResponse(event_generator(), media_type="application/x-ndjson")
+    return StreamingResponse(
+        with_heartbeat(event_generator()),
+        media_type="application/x-ndjson",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
 
 
 

backend/app/services/ai_provider.py

@@ -611,7 +611,10 @@ def _openai_client(api_key: str, timeout: float):
         api_key=api_key,
         base_url=normalize_openai_base_url(secrets_store.get_ai_config("ai_base_url", settings.ai_base_url)),
         timeout=timeout,
-        max_retries=0,
+        # SDK 层重试仅覆盖首包前的连接错误/超时/429/5xx, 此时尚未产出任何内容,
+        # 重试安全; 首 chunk 之后的断流不在此列, 由上层协议报错处理。
+        # 根因: DeepSeek 等上游高峰过载时首包失败率高, max_retries=0 导致一次抖动即终止。
+        max_retries=2,
         default_headers={"User-Agent": user_agent},
     )
 

backend/app/services/ndjson_heartbeat.py

@@ -0,0 +1,62 @@
+"""NDJSON 流式响应的心跳保活包装。
+
+根因: AI 分析/复盘类端点先 yield meta 再等 LLM 首包, 推理模型思考期间
+流上零字节可达数分钟, 反向代理与浏览器代理会按空闲超时切断连接, 前端
+fetch 抛出裸 network error。空闲期插入 {"type":"ping"} 心跳行保活;
+前端各消费方与 daily_pipeline 的 recap 消费循环对未知事件类型均静默
+忽略, 新增类型无兼容影响。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import logging
+from collections.abc import AsyncIterator
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# 与 api/mining.py 的 SSE 心跳间隔保持一致
+HEARTBEAT_INTERVAL_SECONDS = 15.0
+
+_HEARTBEAT_LINE = json.dumps({"type": "ping"})
+
+
+async def with_heartbeat(
+    events: AsyncIterator[str],
+    interval: float = HEARTBEAT_INTERVAL_SECONDS,
+) -> AsyncIterator[str]:
+    """包装 NDJSON 事件生成器: 源空闲超过 interval 秒时插入一行心跳。
+
+    源生成器的 yield 与异常均原样透传; 消费端停止(客户端断连)时取消
+    后台泵任务, 不遗留挂起协程。
+    """
+    queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
+
+    async def pump() -> None:
+        try:
+            async for item in events:
+                queue.put_nowait(("item", item))
+            queue.put_nowait(("eof", None))
+        except Exception as e:  # 由消费端原样重抛
+            queue.put_nowait(("error", e))
+
+    pump_task = asyncio.create_task(pump())
+    try:
+        while True:
+            try:
+                kind, payload = await asyncio.wait_for(queue.get(), timeout=interval)
+            except TimeoutError:
+                yield _HEARTBEAT_LINE
+                continue
+            if kind == "item":
+                yield payload
+            elif kind == "error":
+                raise payload
+            else:
+                return
+    finally:
+        pump_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await pump_task

backend/tests/test_ndjson_heartbeat.py

@@ -0,0 +1,89 @@
+"""NDJSON 心跳包装与 LLM 客户端重试配置的测试。
+
+背景: AI 分析/复盘端点等 LLM 首包期间流上零字节, 会被代理按空闲超时
+切断连接(前端报裸 network error)。with_heartbeat 在空闲期插入
+{"type":"ping"} 行保活; _openai_client 的 max_retries 让首包前失败可重试。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from app import secrets_store
+from app.services import ai_provider
+from app.services.ndjson_heartbeat import with_heartbeat
+
+
+async def _collect(gen) -> list[dict]:
+    out = []
+    async for line in gen:
+        out.append(json.loads(line))
+    return out
+
+
+@pytest.mark.asyncio
+async def test_with_heartbeat_inserts_ping_only_when_idle():
+    async def slow_source():
+        yield json.dumps({"type": "meta"})
+        await asyncio.sleep(0.15)  # 空闲窗口, 应插 ping
+        yield json.dumps({"type": "delta", "content": "x"})
+        yield json.dumps({"type": "done"})
+
+    events = await _collect(with_heartbeat(slow_source(), interval=0.05))
+    types = [e["type"] for e in events]
+
+    assert types[0] == "meta"
+    assert types[-2] == "delta"
+    assert types[-1] == "done"
+    # 中间全部是心跳, 数据完整且相对顺序不变
+    assert types[1:-2] and all(t == "ping" for t in types[1:-2])
+    assert events[-2]["content"] == "x"
+
+
+@pytest.mark.asyncio
+async def test_with_heartbeat_no_ping_when_source_active():
+    async def fast_source():
+        for i in range(5):
+            yield json.dumps({"type": "delta", "content": str(i)})
+
+    events = await _collect(with_heartbeat(fast_source(), interval=30.0))
+    assert [e["type"] for e in events] == ["delta"] * 5
+    assert [e["content"] for e in events] == [str(i) for i in range(5)]
+
+
+@pytest.mark.asyncio
+async def test_with_heartbeat_propagates_source_error():
+    async def bad_source():
+        yield json.dumps({"type": "meta"})
+        raise RuntimeError("boom")
+
+    lines: list[str] = []
+    with pytest.raises(RuntimeError, match="boom"):
+        async for line in with_heartbeat(bad_source(), interval=0.05):
+            lines.append(line)
+    assert len(lines) == 1  # 异常前已收到的数据不丢
+
+
+@pytest.mark.asyncio
+async def test_with_heartbeat_cancels_pump_on_client_disconnect():
+    async def endless_source():
+        while True:
+            await asyncio.sleep(0.01)
+            yield json.dumps({"type": "delta", "content": "tick"})
+
+    gen = with_heartbeat(endless_source(), interval=0.05)
+    first = await gen.__anext__()
+    assert json.loads(first)["type"] == "delta"
+    await gen.aclose()  # 模拟客户端断连; pump 应被取消且不挂起
+
+
+def test_openai_client_enables_bounded_retries(monkeypatch):
+    monkeypatch.setattr(
+        secrets_store,
+        "get_ai_config",
+        lambda key, default="": "https://api.deepseek.com/v1" if key == "ai_base_url" else "",
+    )
+    client = ai_provider._openai_client("test-key", 30.0)
+    assert client.max_retries == 2

frontend/src/components/RpsRotationDialog.tsx

@@ -2,7 +2,7 @@ import { useState, useMemo, useRef, useEffect, useCallback } from 'react'
 import { motion, AnimatePresence } from 'framer-motion'
 import { X, Repeat, Sparkles, ArrowDownUp, RefreshCw, AlertCircle } from 'lucide-react'
 import { useQuery } from '@tanstack/react-query'
-import { api } from '@/lib/api'
+import { api, friendlyStreamError } from '@/lib/api'
 import { QK } from '@/lib/queryKeys'
 import { cn } from '@/lib/cn'
 import { fmtPct } from '@/lib/format'
@@ -77,7 +77,7 @@ export function RpsRotationDialog({ onClose, kind = 'concept' }: Props) {
         // done: 无操作
       }
     } catch (e) {
-      setAnalysisError(e instanceof Error ? e.message : String(e))
+      setAnalysisError(friendlyStreamError(e instanceof Error ? e.message : String(e)))
     } finally {
       setAnalyzing(false)
     }

frontend/src/components/screener/StrategyBuilderDialog.tsx

@@ -1,7 +1,7 @@
 import { useState, useEffect, useCallback, useRef } from 'react'
 import { Modal } from '@/components/Modal'
 import { X, Sparkles, Save, Loader2, ChevronLeft, ChevronRight, AlertTriangle, Settings2, FileText, Copy, Check, Terminal } from 'lucide-react'
-import { api } from '@/lib/api'
+import { api, friendlyStreamError } from '@/lib/api'
 import type { AiIterateRound } from '@/lib/api'
 import { storage } from '@/lib/storage'
 import { cn } from '@/lib/cn'
@@ -349,7 +349,7 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
         if (genRules) setRules(genRules)
       }
     } catch (e: any) {
-      const msg = String(e?.message ?? '')
+      const msg = friendlyStreamError(String(e?.message ?? ''))
       setError(msg.includes('API Key') || msg.includes('api_key') ? 'AI API Key 未配置或无效' : (msg || '生成失败'))
     } finally { setLoading(false) }
   }
@@ -378,7 +378,7 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
       const updatedRules = parseRules(finalResult.code)
       if (genDesc) setDescription(genDesc)
       if (updatedRules) setRules(updatedRules)
-    } catch (e: any) { setError(String(e?.message ?? '修改失败')) }
+    } catch (e: any) { setError(friendlyStreamError(String(e?.message ?? '')) || '修改失败') }
     finally { setLoading(false) }
   }
 

frontend/src/lib/aiReportStore.ts

@@ -1,5 +1,5 @@
 import { useSyncExternalStore } from 'react'
-import { api } from './api'
+import { api, friendlyStreamError } from './api'
 
 /**
  * AI 财务分析 —— 全局任务/报告 store(与 UI 解耦)。
@@ -264,7 +264,7 @@ async function runStream(id: string, symbol: string, focus: string) {
     const msg = String(e?.message ?? '分析失败')
     patchTask(id, {
       phase: 'error',
-      error: normalizeAiError(msg),
+      error: normalizeAiError(friendlyStreamError(msg)),
     })
   }
 }

frontend/src/lib/api.ts

@@ -25,6 +25,19 @@ export class ApiError extends Error {
   }
 }
 
+/**
+ * 把浏览器 fetch 抛出的裸网络错误文案翻译成可操作的提示。
+ * 各浏览器文案不同: Chrome "Failed to fetch" / Safari "network error" / "Load failed"。
+ * 典型根因: 后端等 AI 首包期间流式连接被代理/网关按空闲超时切断, 或 AI 服务繁忙。
+ */
+export function friendlyStreamError(message: string | undefined | null): string {
+  if (!message) return ''
+  if (/failed to fetch|network error|load failed|networkerror/i.test(message)) {
+    return '网络连接中断: 通常是 AI 服务繁忙, 或代理/网关超时切断了长连接, 请重试'
+  }
+  return message
+}
+
 const DEFAULT_REQUEST_TIMEOUT_MS = 30_000
 /** 同步计算型接口 (回测/筛选等) 的放宽超时: 合法耗时可能远超轮询类接口。 */
 const COMPUTE_REQUEST_TIMEOUT_MS = 300_000
@@ -806,6 +819,7 @@ export type StrategyBuildStreamEvent =
   | { type: 'delta'; content: string }
   | ({ type: 'result' } & StrategyBuildResult)
   | { type: 'error'; message: string }
+  | { type: 'ping' }
 
 export interface StrategyCodeSaveResult {
   ok: boolean
@@ -3212,7 +3226,7 @@ export const api = {
    * 用 ReadableStream 解析(而非 SSE EventSource),支持 POST body 且更简单。
    */
   async *financialAnalyzeStream(symbol: string, focus?: string): AsyncGenerator<{
-    type: 'meta' | 'delta' | 'error' | 'done'
+    type: 'meta' | 'delta' | 'error' | 'done' | 'ping'
     symbol?: string
     summary?: string
     periods?: number
@@ -3283,7 +3297,7 @@ export const api = {
    * meta 里额外带 levels(关键价位)供图表回放。
    */
   async *stockAnalyzeStream(symbol: string, focus?: string): AsyncGenerator<{
-    type: 'meta' | 'delta' | 'error' | 'done'
+    type: 'meta' | 'delta' | 'error' | 'done' | 'ping'
     symbol?: string
     summary?: string
     levels?: Record<LevelType, PriceLevel[]>
@@ -3358,7 +3372,7 @@ export const api = {
    * meta 里带 as_of / emotion_score / emotion_label / summary,供前端先渲染信号灯。
    */
   async *reviewStream(asOf?: string, focus?: string): AsyncGenerator<{
-    type: 'meta' | 'delta' | 'error' | 'done'
+    type: 'meta' | 'delta' | 'error' | 'done' | 'ping'
     as_of?: string
     emotion_score?: number
     emotion_label?: string
@@ -3402,7 +3416,7 @@ export const api = {
 
   /** AI 概念轮动分析 — 流式 NDJSON。 */
   async *rotationAnalyzeStream(days: number, focus?: string, kind?: 'concept' | 'industry', level?: number): AsyncGenerator<{
-    type: 'meta' | 'delta' | 'error' | 'done'
+    type: 'meta' | 'delta' | 'error' | 'done' | 'ping'
     days?: number
     summary?: string
     content?: string

frontend/src/lib/reviewStore.ts

@@ -10,7 +10,7 @@
  *  - AbortController 存模块级 ref,组件卸载不中断流
  *  - 订阅者列表(notify 机制),Review mount 时订阅、unmount 时退订
  */
-import { api } from '@/lib/api'
+import { api, friendlyStreamError } from '@/lib/api'
 
 export type ReviewPhase = 'idle' | 'loading' | 'streaming' | 'done' | 'error'
 
@@ -121,7 +121,7 @@ export async function startReviewGeneration(
     }
   } catch (e: any) {
     if (!abortCtrl.signal.aborted) {
-      state = { ...state, error: e?.message ?? '复盘失败', phase: 'error' }
+      state = { ...state, error: friendlyStreamError(e?.message) || '复盘失败', phase: 'error' }
       notify()
     }
   } finally {

frontend/src/lib/stockAnalysisStore.ts

@@ -1,5 +1,5 @@
 import { useSyncExternalStore } from 'react'
-import { api, type PriceLevel, type LevelType } from './api'
+import { api, friendlyStreamError, type PriceLevel, type LevelType } from './api'
 
 /**
  * AI 个股分析 —— 全局任务/报告 store(与 aiReportStore 解耦、并行存在)。
@@ -235,7 +235,7 @@ async function runStream(id: string, symbol: string, _name: string, focus: strin
     const msg = String(e?.message ?? '分析失败')
     patchTask(id, {
       phase: 'error',
-      error: normalizeAiError(msg),
+      error: normalizeAiError(friendlyStreamError(msg)),
     })
   }
 }