From 0fc73d58fa6512de4a31fe2e263c95964f0c8c34 Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 07:09:51 +0800 Subject: [PATCH 01/10] =?UTF-8?q?fix(trending):=20=E4=BB=8A=E6=97=A5?= =?UTF-8?q?=E9=80=89=E9=A2=98=E4=B8=8E=E6=8A=A5=E5=91=8A=E7=AA=97=E5=8F=A3?= =?UTF-8?q?=E6=8C=89=E5=8E=9F=E6=96=87=E5=8F=91=E5=B8=83=E6=97=B6=E9=97=B4?= =?UTF-8?q?=E5=BD=92=E6=A1=A3=E6=97=A7=E6=96=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit today_picks / 日报 / 周报的候选时间窗从 crawled_at 改为 COALESCE(published_at, crawled_at)：3 天前发布、今天才抓到的 旧文不再进「今天」，按原文发布时间归档；published_at 为空的 信源回退按抓取时间计。覆盖 DuckDB 主路径、OLTP fallback 与 repo 两侧共 6 处窗口，主备口径逐条对齐。 --- backend/app/repositories/content_repo.py | 10 +- backend/app/services/_duckdb_picks_mixin.py | 2 +- backend/app/services/_duckdb_reports_mixin.py | 8 +- backend/app/services/today_picks.py | 6 +- backend/tests/test_duckdb_service.py | 29 ++--- backend/tests/test_scoring_flow_repo.py | 107 ++++++++++++++++++ 6 files changed, 140 insertions(+), 22 deletions(-) diff --git a/backend/app/repositories/content_repo.py b/backend/app/repositories/content_repo.py index 3b7abbea..995ae2af 100644 --- a/backend/app/repositories/content_repo.py +++ b/backend/app/repositories/content_repo.py @@ -808,7 +808,9 @@ async def list_for_today_picks( selectinload(self.model.source), ) .join(AiAnalysis, AiAnalysis.id == latest_analysis_id) - .where(self.model.crawled_at >= cutoff) + # 窗口按原文发布时间优先（published_at 为空回退抓取时间），旧文晚抓 + # 不进「今天」；与 DuckDB query_today_picks 的 COALESCE 口径对齐。 + .where(func.coalesce(self.model.published_at, self.model.crawled_at) >= cutoff) .where(AiAnalysis.risk_score <= risk_threshold) ) if category: @@ -847,8 +849,10 @@ async def list_for_report_window( selectinload(self.model.source), ) .join(AiAnalysis, AiAnalysis.id == latest_analysis_id) - .where(self.model.crawled_at >= window_start) - .where(self.model.crawled_at <= window_end) + # 报告窗口同样按原文发布时间归档：晚抓到的旧文归入其发布期， + # 不出现在之后的日报里（口径与 list_for_today_picks 一致）。 + .where(func.coalesce(self.model.published_at, self.model.crawled_at) >= window_start) + .where(func.coalesce(self.model.published_at, self.model.crawled_at) <= window_end) .where(AiAnalysis.risk_score <= risk_threshold) .where(AiAnalysis.curation_score.isnot(None)) .where(~self._accepted_event_member_exists()) diff --git a/backend/app/services/_duckdb_picks_mixin.py b/backend/app/services/_duckdb_picks_mixin.py index 23e0e070..4a5744f7 100644 --- a/backend/app/services/_duckdb_picks_mixin.py +++ b/backend/app/services/_duckdb_picks_mixin.py @@ -105,7 +105,7 @@ def query_today_picks( LEFT JOIN oltp_db.sources s ON s.id = c.source_id LEFT JOIN feedback_scores f ON f.content_id = c.id LEFT JOIN ignored_content ignored ON ignored.content_id = c.id - WHERE c.crawled_at >= ? + WHERE COALESCE(c.published_at, c.crawled_at) >= ? AND ignored.content_id IS NULL AND a.risk_score <= {risk_threshold} AND a.curation_score IS NOT NULL diff --git a/backend/app/services/_duckdb_reports_mixin.py b/backend/app/services/_duckdb_reports_mixin.py index 7af242fb..f5e01262 100644 --- a/backend/app/services/_duckdb_reports_mixin.py +++ b/backend/app/services/_duckdb_reports_mixin.py @@ -29,7 +29,8 @@ def query_content_for_report(self, hours: int = 48) -> list[dict[str, Any]]: FROM oltp_db.content_items c LEFT JOIN latest_analysis a ON a.content_id = c.id LEFT JOIN ignored_content ignored ON ignored.content_id = c.id - WHERE c.crawled_at >= '{cutoff}' + -- 窗口按原文发布时间优先（口径同 list_for_report_window / query_today_picks） + WHERE COALESCE(c.published_at, c.crawled_at) >= '{cutoff}' AND ignored.content_id IS NULL AND a.curation_score IS NOT NULL ORDER BY (COALESCE(a.creator_score, 0) + COALESCE(a.viral_score, 0)) DESC @@ -85,8 +86,9 @@ def query_content_for_weekly(self, start_date: str, end_date: str) -> list[dict[ LEFT JOIN oltp_db.sources s ON s.id = c.source_id LEFT JOIN feedback_scores f ON f.content_id = c.id LEFT JOIN ignored_content ignored ON ignored.content_id = c.id - WHERE CAST(c.crawled_at AS DATE) >= DATE '{start_date}' - AND CAST(c.crawled_at AS DATE) <= DATE '{end_date}' + -- 周报/月报同样按原文发布时间归档，口径与日报窗口一致 + WHERE CAST(COALESCE(c.published_at, c.crawled_at) AS DATE) >= DATE '{start_date}' + AND CAST(COALESCE(c.published_at, c.crawled_at) AS DATE) <= DATE '{end_date}' AND ignored.content_id IS NULL AND a.curation_score IS NOT NULL ORDER BY adjusted_score DESC, COALESCE(a.creator_score, 0) DESC diff --git a/backend/app/services/today_picks.py b/backend/app/services/today_picks.py index 0a9c2172..4c05e98e 100644 --- a/backend/app/services/today_picks.py +++ b/backend/app/services/today_picks.py @@ -6,7 +6,7 @@ import logging from datetime import UTC, datetime, timedelta -from sqlalchemy import or_, select +from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -277,7 +277,9 @@ async def _build_today_picks_via_oltp( selectinload(ContentItem.analyses), selectinload(ContentItem.source), ) - .where(ContentItem.crawled_at >= cutoff) + # 窗口按原文发布时间优先（published_at 为空的信源回退抓取时间）， + # 旧文晚抓不进「今天」；与 DuckDB query_today_picks 的 COALESCE 口径对齐。 + .where(func.coalesce(ContentItem.published_at, ContentItem.crawled_at) >= cutoff) .order_by(ContentItem.crawled_at.desc()) ) if category: diff --git a/backend/tests/test_duckdb_service.py b/backend/tests/test_duckdb_service.py index 8fa4e2cb..ab9904af 100644 --- a/backend/tests/test_duckdb_service.py +++ b/backend/tests/test_duckdb_service.py @@ -826,7 +826,8 @@ def test_digest_content_query_uses_latest_analysis_and_feedback_order(monkeypatc category VARCHAR, source_name VARCHAR, platform VARCHAR, - crawled_at TIMESTAMP + crawled_at TIMESTAMP, + published_at TIMESTAMP ) """) conn.execute(""" @@ -873,12 +874,12 @@ def test_digest_content_query_uses_latest_analysis_and_feedback_order(monkeypatc now = datetime.now(UTC).replace(tzinfo=None).replace(tzinfo=None) conn.execute("INSERT INTO oltp_db.sources VALUES (10, 4)") conn.execute( - "INSERT INTO oltp_db.content_items VALUES (1, 10, '反馈后的最新分析', 'https://example.com/1', 'AI', '测试信源', 'rss', ?)", - [now], + "INSERT INTO oltp_db.content_items VALUES (1, 10, '反馈后的最新分析', 'https://example.com/1', 'AI', '测试信源', 'rss', ?, ?)", + [now, now], ) conn.execute( - "INSERT INTO oltp_db.content_items VALUES (2, NULL, '无反馈样本', 'https://example.com/2', 'AI', '测试信源', 'rss', ?)", - [now], + "INSERT INTO oltp_db.content_items VALUES (2, NULL, '无反馈样本', 'https://example.com/2', 'AI', '测试信源', 'rss', ?, ?)", + [now, now], ) conn.execute( "INSERT INTO oltp_db.ai_analyses VALUES (1, 1, '旧摘要', 95, 95, 95, 95, 95, 10, 99, 95, 95, 95, '[\"旧\"]', '旧推荐', '旧理由', ?)", @@ -930,7 +931,8 @@ def test_daily_report_content_query_uses_latest_analysis_only(monkeypatch): url VARCHAR, category VARCHAR, source_name VARCHAR, - crawled_at TIMESTAMP + crawled_at TIMESTAMP, + published_at TIMESTAMP ) """) conn.execute(""" @@ -951,12 +953,12 @@ def test_daily_report_content_query_uses_latest_analysis_only(monkeypatch): now = datetime.now(UTC).replace(tzinfo=None) conn.execute( - "INSERT INTO oltp_db.content_items VALUES (1, '多次分析样本', 'https://example.com/1', 'AI', '测试信源', ?)", - [now], + "INSERT INTO oltp_db.content_items VALUES (1, '多次分析样本', 'https://example.com/1', 'AI', '测试信源', ?, ?)", + [now, now], ) conn.execute( - "INSERT INTO oltp_db.content_items VALUES (2, '普通样本', 'https://example.com/2', 'AI', '测试信源', ?)", - [now], + "INSERT INTO oltp_db.content_items VALUES (2, '普通样本', 'https://example.com/2', 'AI', '测试信源', ?, ?)", + [now, now], ) conn.execute( "INSERT INTO oltp_db.ai_analyses VALUES (1, 1, '旧日报摘要', 99, 99, 99, 10, 99, '旧理由', ?)", @@ -996,7 +998,8 @@ def test_digest_content_queries_exclude_ignored_content(monkeypatch): category VARCHAR, source_name VARCHAR, platform VARCHAR, - crawled_at TIMESTAMP + crawled_at TIMESTAMP, + published_at TIMESTAMP ) """) conn.execute(""" @@ -1040,8 +1043,8 @@ def test_digest_content_queries_exclude_ignored_content(monkeypatch): now = datetime.now(UTC).replace(tzinfo=None) for content_id, title in ((1, "已忽略素材"), (2, "保留素材")): conn.execute( - "INSERT INTO oltp_db.content_items VALUES (?, NULL, ?, ?, 'AI', '测试信源', 'rss', ?)", - [content_id, title, f"https://example.com/{content_id}", now], + "INSERT INTO oltp_db.content_items VALUES (?, NULL, ?, ?, 'AI', '测试信源', 'rss', ?, ?)", + [content_id, title, f"https://example.com/{content_id}", now, now], ) conn.execute( """ diff --git a/backend/tests/test_scoring_flow_repo.py b/backend/tests/test_scoring_flow_repo.py index b52656c3..c0f14c78 100644 --- a/backend/tests/test_scoring_flow_repo.py +++ b/backend/tests/test_scoring_flow_repo.py @@ -654,3 +654,110 @@ async def test_scoring_flow_counts_respect_visible_user_id(): assert count_all == 3, f"全局应见 3 条，实际 {count_all}" await engine.dispose() + + +def _window_test_source(now: datetime) -> Source: + return Source( + id=1, + name="窗口测试信源", + source_type=SourceType.RSS, + url="https://example.com/window.xml", + category="AI", + status=SourceStatus.ACTIVE, + enabled=True, + weight=3, + ) + + +def _analyzed_item( + item_id: int, + *, + crawled_at: datetime, + published_at: datetime | None, +) -> ContentItem: + return ContentItem( + id=item_id, + title=f"窗口样本 {item_id}", + url=f"https://example.com/window-{item_id}", + source_id=1, + source_name="窗口测试信源", + source_type="RSS", + category="AI", + status=ContentStatus.ANALYZED, + crawled_at=crawled_at, + published_at=published_at, + ) + + +async def _seed_window_items(db, now: datetime) -> None: + """三条内容：旧文晚抓 / 无发布时间 / 正常新文，各带一条低风险分析。""" + items = [ + # 旧文晚抓：3 天前发布、刚刚才发现 —— 不应进「今天」的候选 + _analyzed_item(1, crawled_at=now, published_at=now - timedelta(days=3)), + # 无发布时间的信源：回退按抓取时间计（2 小时前抓到） + _analyzed_item(2, crawled_at=now - timedelta(hours=2), published_at=None), + # 正常新文：半小时前发布、刚刚抓到（发布时间在报告窗口内） + _analyzed_item(3, crawled_at=now, published_at=now - timedelta(minutes=30)), + ] + db.add_all(items) + db.add_all( + [ + AiAnalysis( + id=item_id, + content_id=item_id, + curation_score=80, + risk_score=10, + created_at=now, + ) + for item_id in (1, 2, 3) + ] + ) + await db.commit() + + +@pytest.mark.asyncio +async def test_today_picks_window_prefers_published_at(): + """旧文晚抓不进今日候选；published_at 缺失回退 crawled_at。""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime.now(UTC) + async with session_factory() as db: + db.add(_window_test_source(now)) + await _seed_window_items(db, now) + + repo = ContentRepo(db) + items = await repo.list_for_today_picks(hours=24) + + assert {item.id for item in items} == {2, 3}, "3 天前发布的旧文即使刚被抓到也不应进 24h 候选窗口" + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_report_window_prefers_published_at(): + """报告窗口按原文发布时间归档：晚抓到的旧文不进之后的日报。""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime.now(UTC) + async with session_factory() as db: + db.add(_window_test_source(now)) + await _seed_window_items(db, now) + + repo = ContentRepo(db) + # 报告窗口覆盖「刚刚」（旧文 id=1 的 crawled_at 在窗口内、published_at 在窗口外） + report_items = await repo.list_for_report_window( + window_start=now - timedelta(hours=1), + window_end=now + timedelta(hours=1), + ) + assert 1 not in { + item.id for item in report_items + }, "旧文（published_at 在报告窗口外）不应因晚抓而出现在本期日报" + assert 3 in {item.id for item in report_items} + + await engine.dispose() From 5e8660538548cf4f31c73fd34ad4b3b1c46e8eee Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 07:10:03 +0800 Subject: [PATCH 02/10] =?UTF-8?q?feat(llm):=20=E5=A2=9E=E5=8A=A0=E4=B8=89?= =?UTF-8?q?=E7=AA=97=E9=A2=84=E7=AE=97=E7=86=94=E6=96=AD=E9=97=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 budget_guard：分钟/小时/日三窗预算保险丝，按 llm_call_logs 实际 DONE 调用数计数，超限拒绝新付费调用（LlmBudgetExceededError）。 插入点在熔断检查与响应缓存之后——缓存命中（免费）照常放行。 分析链路遇预算拒绝走本地降级（与 CircuitOpenError 同分支）， 不消耗分析重试次数；预算拒绝记 BUDGET_REJECTED 审计行与 budget_rejected 指标事件；计数查询失败 fail-open。 默认 0/0/0 全关（opt-in），.env.example 附开启建议。 --- backend/.env.example | 6 + backend/app/core/config.py | 7 + backend/app/services/analysis.py | 14 +- backend/app/services/creation.py | 3 + backend/app/services/llm/_call_engine.py | 5 +- backend/app/services/llm/budget_guard.py | 81 +++++++++ backend/app/services/llm/provider.py | 11 +- backend/app/services/llm_usage.py | 29 ++++ backend/tests/test_llm_budget_guard.py | 205 +++++++++++++++++++++++ 9 files changed, 354 insertions(+), 7 deletions(-) create mode 100644 backend/app/services/llm/budget_guard.py create mode 100644 backend/tests/test_llm_budget_guard.py diff --git a/backend/.env.example b/backend/.env.example index 91ce7cfc..ecfddef6 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -70,6 +70,12 @@ AUTH_REFRESH_ATTEMPTS_PER_MINUTE=20 LLM_REQUESTS_PER_MINUTE=30 # 单次模型调用硬截止时间；模型配置中的 extra_params.litellm_params.timeout 只能缩短。 LLM_COMPLETION_TIMEOUT_SECONDS=45 +# 预算熔断：按实际 DONE 调用数的三窗保险丝（分钟/小时/日），超限拒绝新付费调用， +# 分析链路自动走本地降级，窗口过后恢复。0 = 关闭对应窗口（默认全关，opt-in）。 +# 开启建议：分钟窗 ≥ LLM_REQUESTS_PER_MINUTE；日窗按可接受的烧钱上限设置。 +LLM_BUDGET_CALLS_PER_MINUTE=60 +LLM_BUDGET_CALLS_PER_HOUR=2000 +LLM_BUDGET_CALLS_PER_DAY=20000 # OAuth (Google / GitHub 登录) # 留空则该 provider 不启用。前端登录页会根据已启用 provider 渲染对应按钮。 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 9738c6f2..3383d3ee 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -176,6 +176,13 @@ class Settings(BaseSettings): LLM_WORKER_CONCURRENCY: int = 4 # 单次运行时调用的硬上限；模型 extra_params.timeout 只能把它调小，不能放大。 LLM_COMPLETION_TIMEOUT_SECONDS: float = 45.0 + # 预算熔断（budget_guard）：按 llm_call_logs 实际 DONE 调用数的三窗保险丝， + # 超限直接拒绝新调用（分析链路走本地降级，窗口过后自动恢复）。 + # 默认 0 = 全部关闭（opt-in，不改变现有部署行为）；0 同样适用于单窗关闭。 + # 开启建议：分钟窗不低于 LLM_REQUESTS_PER_MINUTE，日窗按可接受的烧钱上限设。 + LLM_BUDGET_CALLS_PER_MINUTE: int = 0 + LLM_BUDGET_CALLS_PER_HOUR: int = 0 + LLM_BUDGET_CALLS_PER_DAY: int = 0 ANALYSIS_WORKER_CONCURRENCY: int = 3 ANALYSIS_MAX_ATTEMPTS: int = 5 ANALYSIS_RETRY_BASE_DELAY_SECONDS: int = 60 diff --git a/backend/app/services/analysis.py b/backend/app/services/analysis.py index 6530e4d6..4cc6a1eb 100644 --- a/backend/app/services/analysis.py +++ b/backend/app/services/analysis.py @@ -506,15 +506,19 @@ async def analyze_content(content: ContentItem, db: AsyncSession) -> AiAnalysis: ) final_model = final_metadata.get("actual_model") or final_model except Exception as llm_exc: - # CircuitOpenError (breaker tripped) and BadRequestError (400, - # e.g. GLM contentFilter code=1301) trigger local fallback. - # Other LLM failures (timeout, network, RuntimeError) still - # propagate up so the caller can record ERROR status + retry. + # CircuitOpenError (breaker tripped), LlmBudgetExceededError (budget + # fuse) and BadRequestError (400, e.g. GLM contentFilter code=1301) + # trigger local fallback. Other LLM failures (timeout, network, + # RuntimeError) still propagate up so the caller can record ERROR + # status + retry. Budget rejection must especially NOT consume + # analysis attempts: a day-window fuse can outlast the retry budget + # and would otherwise burn the whole backlog into permanent ERROR. from litellm.exceptions import BadRequestError + from app.services.llm.budget_guard import LlmBudgetExceededError from app.services.llm.circuit_breaker import CircuitOpenError - if isinstance(llm_exc, CircuitOpenError | BadRequestError): + if isinstance(llm_exc, CircuitOpenError | BadRequestError | LlmBudgetExceededError): logger.warning( "LLM call failed for content id=%d (%s), using local fallback", content.id, diff --git a/backend/app/services/creation.py b/backend/app/services/creation.py index 961eda4a..6457b5bc 100644 --- a/backend/app/services/creation.py +++ b/backend/app/services/creation.py @@ -26,6 +26,7 @@ def _creation_llm_error_message(exc: Exception) -> str: """Return an actionable, non-internal error message for creation callers.""" + from app.services.llm.budget_guard import LlmBudgetExceededError from app.services.llm.circuit_breaker import CircuitOpenError from app.services.llm.provider import LlmCapacityUnavailableError @@ -33,6 +34,8 @@ def _creation_llm_error_message(exc: Exception) -> str: return "创作方案暂时排队等待可用模型渠道，请稍后重试。" if isinstance(exc, CircuitOpenError): return "创作方案服务暂时不可用，系统正在自动恢复，请稍后重试。" + if isinstance(exc, LlmBudgetExceededError): + return "模型调用预算已达今日上限，创作方案稍后自动恢复，请明日再试。" return "创作方案生成失败，请稍后重试。" diff --git a/backend/app/services/llm/_call_engine.py b/backend/app/services/llm/_call_engine.py index 4dcef06f..0420f253 100644 --- a/backend/app/services/llm/_call_engine.py +++ b/backend/app/services/llm/_call_engine.py @@ -251,8 +251,11 @@ def _should_retry(exc: BaseException) -> bool: 覆盖 DeepSeek/GLM/智谱等通过 openai-compat 抛通用 APIError 的场景。 BadRequestError (400) 也不重试：内容过滤等确定性错误重试只会浪费时间。 + 预算熔断拒绝（LlmBudgetExceededError）同理：窗口未恢复前重试必被再拒。 """ - if isinstance(exc, RateLimitError) or _is_deterministic_request_error(exc): + from app.services.llm.budget_guard import LlmBudgetExceededError + + if isinstance(exc, RateLimitError | LlmBudgetExceededError) or _is_deterministic_request_error(exc): return False return not _is_rate_limit_error(exc) diff --git a/backend/app/services/llm/budget_guard.py b/backend/app/services/llm/budget_guard.py new file mode 100644 index 00000000..6bdef19e --- /dev/null +++ b/backend/app/services/llm/budget_guard.py @@ -0,0 +1,81 @@ +"""LLM 预算熔断闸。 + +三窗（分钟/小时/日）预算保险丝：按 ``llm_call_logs`` 里实际 DONE 的调用 +计数，超限直接拒绝新调用（``LlmBudgetExceededError``），调用方走既有的 +失败退避/降级路径。与 ``_rate_limit`` 的语义区别：限流是「等一等再发」 +（保护对端配额），预算是「不再花下去」（保护自己的钱包）——拒绝而非等待。 + +- 各窗口上限为 0 表示关闭该窗口；三窗全 0 时零开销直通。 +- 计数查询失败时 fail-open（保险丝自身故障不阻断主链路），记 warning。 +- 预算拒绝写一行 ``llm_call_logs`` 审计记录（status="BUDGET_REJECTED"， + 不参与 DONE 计数），使拒绝在用量看板 ALL 视图可见。 +""" + +from __future__ import annotations + +import logging + +from app.core.config import settings +from app.services.llm._rate_limit import record_llm_pool_circuit_event + +logger = logging.getLogger(__name__) + + +class LlmBudgetExceededError(Exception): + """Rolling budget window exhausted; deterministic, must not be retried.""" + + +def _budget_limits() -> tuple[int, int, int] | None: + limits = ( + settings.LLM_BUDGET_CALLS_PER_MINUTE, + settings.LLM_BUDGET_CALLS_PER_HOUR, + settings.LLM_BUDGET_CALLS_PER_DAY, + ) + if not any(limits): + return None + return limits + + +async def ensure_llm_budget(scene: str) -> None: + """Raise :class:`LlmBudgetExceededError` when a budget window is exhausted.""" + limits = _budget_limits() + if limits is None: + return + + from app.services.llm_usage import count_recent_llm_calls + + try: + minute_calls, hour_calls, day_calls = await count_recent_llm_calls() + except Exception as exc: # noqa: BLE001 — fuse failure must not block the main path + logger.warning("LLM budget check skipped (count query failed): %s", exc) + return + + for label, calls, limit in ( + ("minute", minute_calls, limits[0]), + ("hour", hour_calls, limits[1]), + ("day", day_calls, limits[2]), + ): + if limit > 0 and calls >= limit: + logger.warning("LLM budget exceeded in %s window: %d/%d calls, scene=%s", label, calls, limit, scene) + record_llm_pool_circuit_event(None, scene, "budget_rejected") + await _record_budget_rejection(scene=scene, detail=f"{label} window: {calls}/{limit}") + raise LlmBudgetExceededError( + f"LLM budget exceeded in {label} window: {calls}/{limit} calls (scene={scene})" + ) + + +async def _record_budget_rejection(*, scene: str, detail: str) -> None: + """审计行：BUDGET_REJECTED 不进 DONE 计数，只在用量看板可见。""" + from app.services.llm_usage import record_llm_call_in_new_session + + try: + await record_llm_call_in_new_session( + model=None, + request_model=None, + scene=scene, + status="BUDGET_REJECTED", + duration_ms=0, + error_message=detail, + ) + except Exception as exc: # noqa: BLE001 — 审计写失败不能影响拒绝路径本身 + logger.warning("LLM budget rejection audit log skipped: %s", exc) diff --git a/backend/app/services/llm/provider.py b/backend/app/services/llm/provider.py index a8a0f672..49812f4b 100644 --- a/backend/app/services/llm/provider.py +++ b/backend/app/services/llm/provider.py @@ -152,6 +152,13 @@ async def call_llm_with_metadata( if cached is not None: return cached, {"cache_hit": True} + # Budget fuse: reject paid calls once a rolling budget window is exhausted. + # Checked after the response cache on purpose — cache hits are free and + # must keep serving while new paid calls are refused. + from app.services.llm.budget_guard import ensure_llm_budget + + await ensure_llm_budget(scene) + try: result = await _call_llm_with_metadata_inner( messages, @@ -172,12 +179,14 @@ async def call_llm_with_metadata( return result except Exception as exc: # 输入或内容策略错误不反映模型可用性，不能污染全局熔断器。 + from app.services.llm.budget_guard import LlmBudgetExceededError from app.services.llm.circuit_breaker import CircuitOpenError # 429 和本地候选冷却代表局部容量耗尽，由 per-model failover 管理； # 把它们累计到路由熔断器会让一个配额不足的渠道阻断全部调用。 + # 预算熔断同理：拒绝花钱不是故障，不得让预算恢复后仍被熔断。 if ( - not isinstance(exc, CircuitOpenError | LlmCapacityUnavailableError) + not isinstance(exc, CircuitOpenError | LlmCapacityUnavailableError | LlmBudgetExceededError) and not _is_deterministic_request_error(exc) and not _is_rate_limit_error(exc) ): diff --git a/backend/app/services/llm_usage.py b/backend/app/services/llm_usage.py index 037ef032..f1ffc9d8 100644 --- a/backend/app/services/llm_usage.py +++ b/backend/app/services/llm_usage.py @@ -323,3 +323,32 @@ async def _write(): except Exception as exc: await db.rollback() logger.warning("LLM usage log skipped: %s", exc) + + +async def count_recent_llm_calls() -> tuple[int, int, int]: + """Count DONE LLM calls in the trailing 1m / 1h / 24h windows. + + Single aggregate query backing the budget fuse (budget_guard); only + billable (status="DONE") calls count, rejected-by-fuse calls never + reach this table. + """ + from datetime import UTC, datetime, timedelta + + from sqlalchemy import case, func + + from app.core.database import async_session + + now = datetime.now(UTC) + async with async_session() as db: + result = await db.execute( + select( + func.count(case((LlmCallLog.created_at >= now - timedelta(minutes=1), 1))), + func.count(case((LlmCallLog.created_at >= now - timedelta(hours=1), 1))), + func.count(), + ).where( + LlmCallLog.status == "DONE", + LlmCallLog.created_at >= now - timedelta(hours=24), + ) + ) + minute_calls, hour_calls, day_calls = result.one() + return int(minute_calls), int(hour_calls), int(day_calls) diff --git a/backend/tests/test_llm_budget_guard.py b/backend/tests/test_llm_budget_guard.py new file mode 100644 index 00000000..e8c0466f --- /dev/null +++ b/backend/tests/test_llm_budget_guard.py @@ -0,0 +1,205 @@ +"""LLM 预算熔断闸（budget_guard）的行为回归测试。 + +覆盖三窗语义： +- 超限拒绝（分钟/日窗各自生效），拒绝事件可观测（budget_rejected 事件 + warning）； +- 三窗全 0 时零开销直通（不触发计数查询）； +- 计数查询失败 fail-open（保险丝自身故障不阻断主链路）； +- provider 集成：预算拒绝不进入 failover/真实调用、不污染路由熔断器、 + 响应缓存命中（免费）在预算耗尽时仍放行。 +""" + +from __future__ import annotations + +import pytest + +from app.core.config import settings +from app.services.llm import provider +from app.services.llm._call_engine import _should_retry +from app.services.llm.budget_guard import ( + LlmBudgetExceededError, + _budget_limits, + ensure_llm_budget, +) +from app.services.llm.circuit_breaker import ( + get_llm_circuit_breaker, + reset_llm_circuit_breakers, +) +from app.services.llm.response_cache import get_llm_cache + + +def _set_budget(monkeypatch, minute: int, hour: int, day: int) -> None: + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_MINUTE", minute) + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_HOUR", hour) + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_DAY", day) + + +@pytest.fixture(autouse=True) +def _isolate_route_state(): + provider._failover.reset() + reset_llm_circuit_breakers() + get_llm_cache().clear() + yield + provider._failover.reset() + reset_llm_circuit_breakers() + get_llm_cache().clear() + + +@pytest.mark.asyncio +async def test_all_zero_budget_skips_count_query(monkeypatch): + """三窗全 0 = 预算闸关闭：不应触碰计数查询。""" + _set_budget(monkeypatch, 0, 0, 0) + assert _budget_limits() is None + + async def _boom(): + raise AssertionError("budget disabled must not query call counts") + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _boom) + await ensure_llm_budget(scene="test") + + +@pytest.mark.asyncio +async def test_minute_window_rejects_and_emits_event(monkeypatch): + _set_budget(monkeypatch, minute=2, hour=0, day=0) + + async def _counts(): + return 2, 5, 10 # 分钟窗已到 2/2 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + events: list[tuple] = [] + monkeypatch.setattr( + "app.services.llm.budget_guard.record_llm_pool_circuit_event", + lambda *args: events.append(args), + ) + audits: list[dict] = [] + + async def _fake_audit(**kwargs): + audits.append(kwargs) + + monkeypatch.setattr("app.services.llm.budget_guard._record_budget_rejection", _fake_audit) + + with pytest.raises(LlmBudgetExceededError, match="minute"): + await ensure_llm_budget(scene="analysis") + assert events and events[0][2] == "budget_rejected" + assert audits and audits[0]["scene"] == "analysis" + assert "minute window: 2/2" in audits[0]["detail"] + + +@pytest.mark.asyncio +async def test_day_window_rejects(monkeypatch): + _set_budget(monkeypatch, minute=0, hour=0, day=100) + + async def _counts(): + return 1, 50, 100 # 日窗已到 100/100 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + + async def _fake_audit(**kwargs): + return None + + monkeypatch.setattr("app.services.llm.budget_guard._record_budget_rejection", _fake_audit) + with pytest.raises(LlmBudgetExceededError, match="day"): + await ensure_llm_budget(scene="analysis") + + +@pytest.mark.asyncio +async def test_hour_window_rejects(monkeypatch): + _set_budget(monkeypatch, minute=0, hour=500, day=0) + + async def _counts(): + return 10, 500, 600 # 小时窗已到 500/500 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + + async def _fake_audit(**kwargs): + return None + + monkeypatch.setattr("app.services.llm.budget_guard._record_budget_rejection", _fake_audit) + with pytest.raises(LlmBudgetExceededError, match="hour"): + await ensure_llm_budget(scene="analysis") + + +@pytest.mark.asyncio +async def test_within_budget_passes(monkeypatch): + _set_budget(monkeypatch, minute=10, hour=100, day=1000) + + async def _counts(): + return 3, 40, 500 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + await ensure_llm_budget(scene="analysis") # 不抛即通过 + + +@pytest.mark.asyncio +async def test_count_query_failure_fails_open(monkeypatch, caplog): + _set_budget(monkeypatch, minute=10, hour=100, day=1000) + + async def _boom(): + raise RuntimeError("db unavailable") + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _boom) + with caplog.at_level("WARNING"): + await ensure_llm_budget(scene="analysis") # fail-open，不抛 + assert any("budget check skipped" in rec.message for rec in caplog.records) + + +def test_should_retry_rejects_budget_error(): + assert _should_retry(LlmBudgetExceededError("minute window")) is False + + +@pytest.mark.asyncio +async def test_provider_rejects_before_failover_and_keeps_breaker_clean(monkeypatch): + """预算拒绝：不进入真实调用/failover，也不计入路由熔断器失败。""" + _set_budget(monkeypatch, minute=1, hour=0, day=0) + + async def _counts(): + return 1, 0, 0 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + + async def _forbidden(*args, **kwargs): + raise AssertionError("budget-exceeded call must not reach the failover layer") + + monkeypatch.setattr(provider, "_call_llm_with_metadata_inner", _forbidden) + + breaker = get_llm_circuit_breaker("budget_test") + failures_before = breaker.status()["failure_count"] + + with pytest.raises(LlmBudgetExceededError): + await provider.call_llm_with_metadata( + [{"role": "user", "content": "预算耗尽时的调用"}], + routing_group="budget_test", + scene="test", + ) + assert breaker.status()["failure_count"] == failures_before, "预算拒绝不是故障，不得污染熔断器" + + +@pytest.mark.asyncio +async def test_cache_hit_still_served_when_budget_exhausted(monkeypatch): + """预算耗尽时缓存命中（免费）必须继续放行。""" + _set_budget(monkeypatch, minute=100, hour=0, day=0) + + async def _inner(messages, temperature, max_tokens, scene, routing_group, response_format=None): + return "cached-answer", {"model": "m1"} + + async def _zero_counts(): + return 0, 0, 0 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _zero_counts) + monkeypatch.setattr(provider, "_call_llm_with_metadata_inner", _inner) + + messages = [{"role": "user", "content": "会被缓存的问题"}] + text, meta = await provider.call_llm_with_metadata(messages, routing_group="budget_cache", scene="test") + assert text == "cached-answer" + assert meta.get("cache_hit") is not True + + # 预算降到已耗尽，同参调用必须仍能从缓存拿到答案 + _set_budget(monkeypatch, minute=1, hour=0, day=0) + + async def _counts(): + return 1, 0, 0 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + + text2, meta2 = await provider.call_llm_with_metadata(messages, routing_group="budget_cache", scene="test") + assert text2 == "cached-answer" + assert meta2.get("cache_hit") is True From 40ec6098023894472ed2a98f5e330f53a33854d1 Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 07:38:12 +0800 Subject: [PATCH 03/10] =?UTF-8?q?docs:=20#88=20=E4=B8=89=E6=9D=A1=E7=BC=BA?= =?UTF-8?q?=E9=99=B7=E5=B7=B2=E4=BF=AE=E5=A4=8D=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E5=9B=9E=E5=BD=92=E7=9F=A9=E9=98=B5=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #89 合并进 main（squash 08071e8），三条缺陷的回归测试已随返工提交补齐并 在 main 上复验通过，状态从「修复完成待合并」改为「已修复关闭」。 --- docs/quality/regression-matrix.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/quality/regression-matrix.md b/docs/quality/regression-matrix.md index aafc1b8a..53b411f4 100644 --- a/docs/quality/regression-matrix.md +++ b/docs/quality/regression-matrix.md @@ -89,7 +89,7 @@ - 症状：`_cache_warmup_task` 非 CancelledError 异常会中断后续全部清理步骤；jieba 预热 await 无超时且 `to_thread` 不可取消，可挂死停机；整体停机无 deadline。 - 回归测试：`tests/test_shutdown_prewarm.py`（异常不外抛且留痕 / jieba 超时不挂死 / 正常与已取消路径）。owner：#73（Parent #6，修复 PR #77：`_shutdown_prewarm_tasks`）。 -### 管理后台批量操作静默失败 — #88 🔧 修复完成待合并（2026-09-29） +### 管理后台批量操作静默失败 — #88 ✅ 已修复关闭（2026-09-29，PR #89 合并） - 症状：`sources/page.tsx` `handleBatchToggle` 逐条 `sourcesApi.update` 的 `catch` 只写 `console.error`，循环结束后**无条件** `setSelectedIds(new Set())` + `fetchSources()`。批量停用 30 个信源、其中 5 个失败时，UI 表现为「全部处理完」，失败的 5 个仍按原状态继续采集并污染内容池，而操作者无从察觉。 - 环境/前提：生产数据，仅管理员账号；`PATCH /api/v1/sources/{id}` 任一请求失败（网络抖动 / 404 / 5xx）即可触发。 - 复现：信源管理页多选 ≥2 个信源 → 批量停用 → 中途阻断网络或使其中之一更新失败 → 观察 UI 与浏览器控制台对照。 @@ -99,7 +99,7 @@ - 回归测试：`npx vitest run src/app/admin/sources/_batch-utils.test.ts`（11 条）。选择态重算与结果文案已抽为纯函数 `_batch-utils.ts`，`.tsx` 只做调用；已用变异测试验证——把实现改回旧行为（失败进 console + 无条件清空）后 3 条转红，其中包含「中途新勾选项被静默丢弃」这条。 - owner：#88。 -### Webhook 日志页读数口径撒谎 — #88 🔧 修复完成待合并（2026-09-29） +### Webhook 日志页读数口径撒谎 — #88 ✅ 已修复关闭（2026-09-29，PR #89 合并） - 症状：`webhook-logs/page.tsx` 的 `successCount` / `failCount` 只统计**当前页** 30 行，却与全局 `total` 并排渲染成同款 Badge。读者会把「本页 2 失败」除以「共 1240 条」读成 0.16% 失败率，真相是第 5 页可能还躺着 40 条。更严重的是失败徽章以 `failCount > 0` 为条件渲染——翻到失败为 0 的分页时「失败」整枚徽章消失，**徽章的缺席本身制造错误信念**。 - 环境/前提：任何有 ≥1 页推送日志的生产数据。 - 复现：`/admin/webhook-logs` 翻页，观察任意分页顶部的 Badge 组。 @@ -110,7 +110,7 @@ - **覆盖边界（独立复核 M3 实证，勿夸大）**：修复前 `summarizeLogPage` 的 6 个公式与修复后逐字相同，只测它**证明不了**本缺陷；真正的修复是把「渲染成什么」下沉成 `buildSummaryBadges`，其断言（失败徽章 0 时仍存在且为中性色、全部徽章带「本页」/「全部」口径词、空页只留 1 枚）才真正对应缺陷本身。JSX 渲染层仍无组件测试（见下方遗留项）。 - owner：#88。 -### 管理后台面包屑缺 4 项映射 — #88 🔧 修复完成待合并（2026-09-29） +### 管理后台面包屑缺 4 项映射 — #88 ✅ 已修复关闭（2026-09-29，PR #89 合并） - 症状：`AdminTopBar.tsx` 的 `ADMIN_PAGE_LABELS` 只有 10 条，侧边栏 `ADMIN_NAV_ITEMS` 有 15 项。`prompts` / `scoring-dashboard` / `evidence` / `webhook-logs` 四页无映射，`findPageLabel` 回退显示兜底文案「管理」，与侧边栏自相矛盾。（另经核实：概览卡片 13 张、面包屑 10 条、侧边栏 15 项三份目录互相矛盾，说明分类从未被写下来过。） - 环境/前提：always。 - 复现：访问上述四页之一，观察顶栏面包屑第二段。 From 7ea83c854c03378e637086804700cf0f2b69cb1d Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 07:38:17 +0800 Subject: [PATCH 04/10] =?UTF-8?q?docs:=20=E8=A1=A5=E8=AE=B0=20LLM=20?= =?UTF-8?q?=E9=99=8D=E7=BA=A7=E5=86=85=E5=AE=B9=E6=97=A0=E6=B3=95=E9=87=8D?= =?UTF-8?q?=E6=96=B0=E5=88=86=E6=9E=90=E7=9A=84=E7=BC=BA=E9=99=B7=20#90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LLM 降级路径（熔断 / 内容过滤 / 预算超限）把内容写成 ANALYZED 终态、 summary_source='local_fallback' 落库后，全代码库没有消费者会把它捡回来重跑： 重分析资格谓词只认 PENDING、僵死 ANALYZING 和 ERROR，手动分析端点遇到已有 记录直接早退。触发原因消失后内容也起不来，永久停在降级时算出的假分上。 --- docs/quality/regression-matrix.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/docs/quality/regression-matrix.md b/docs/quality/regression-matrix.md index 53b411f4..b7b4bbbc 100644 --- a/docs/quality/regression-matrix.md +++ b/docs/quality/regression-matrix.md @@ -120,6 +120,17 @@ - 回归测试：`npx vitest run src/lib/__tests__/admin-nav.test.ts`（13 条）。导航清单与面包屑映射合并到 `src/lib/admin-nav.ts` 单一事实源，映射由清单派生；`nav-checklist` 等价断言（壳内每项都有映射、label 与清单逐项一致）已在其中，漏项即红。顺带修掉一个被测试抓出的旧行为：`/admin/任何未知路径` 曾因壳根前缀匹配被判成「概览」，现已回退兜底。 - owner：#88。 +### LLM 降级内容永久结疤：local_fallback 无恢复路径 — #90 🐛 待修复（2026-09-30） +- 症状：LLM 降级路径（熔断 `CircuitOpenError` / 内容过滤 `BadRequestError` / 预算 `LlmBudgetExceededError` 三种触发源共用）把内容写为 `ANALYZED` 终态、`AiAnalysis.summary_source='local_fallback'` 落库后，**全代码库没有任何 requeue 消费者**——重分析资格谓词只认 PENDING/僵死 ANALYZING/ERROR，手动分析端点遇已有记录早退。降级内容是确定性假分数（curation 61-64 窄带、risk 恒 28、模板推荐语），today-picks 以 40px 大字渲染且前端不读 `summary_source`，用户无法分辨；窄带假分参与 P70 计算会拖低百分位门槛、挤掉真实 65-70 分内容。 +- 环境/前提：任一降级触发源命中即触发；当前部署全走本地模型（`llm_call_logs.total_cost` 全 0），预算闸实际不触发，触发面在切换付费 API 后成为现实。 +- 复现：触发降级（如断言 `LlmBudgetExceededError` 路径或熔断 OPEN）→ 确认内容 `status=ANALYZED`、`summary_source='local_fallback'` → 检索全部重分析入口（`_analysis_candidate_condition`、`api/v1/analyses.py` 手动端点、scripts/）确认无路径捡起该内容。 +- 期望 vs 实际：期望触发原因消除后（预算窗口滚过/熔断恢复）内容可被重新 LLM 分析；实际永久停留假分形态。 +- 边界：`analysis.py` 降级路径 + `content_repo.py` 重分析资格谓词；修复为 requeue job + P70 排除 + 前端「本地速览·待 AI 复核」标记 + 日报豁免的一揽子（详见 #90）。 +- 严重度：P1（预算日窗最坏锁 ~24h × 实测 ~250 条/时摄入 ≈ 数千条永久假分；当前零账单部署不触发故非 P0）；**复现性**：always（降级发生即结疤）。 +- 回归测试：待 requeue 落地补——造 fallback 内容 → 模拟预算恢复 → 断言内容回 PENDING 并被重新分析；P70 排除与前端标记各有对应测试（教训机器化随修复交付）。 +- owner：#90。 +- **存量定性（勿误读）**：非预算闸（`feat(llm): 增加三窗预算熔断闸`）引入——熔断降级早就同样结疤；预算闸放大触发面，圆桌评审三席独立发现。发现方式：预算数值圆桌（2026-09-30，Memory 存 `.vidt/roundtable/budget-fuse-values/`，会话态）。 + ## 三、关键流程基线（9 项） 状态标记：✅ = 2026-09-27 在 main @ 7203847 新鲜复跑通过；📋 = 现有套件覆盖、未逐项复跑（跑全量即覆盖）；🔧 = 修复完成待合并（PR 已开、CI 绿、独立复核非 pass 尚未全部处置）。 From 2a8e67c6379e4ec0ae2d622b65979878b44f4765 Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 21:10:56 +0800 Subject: [PATCH 05/10] =?UTF-8?q?feat(analysis):=20=E9=99=8D=E7=BA=A7?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E8=87=AA=E5=8A=A8=E8=A1=A5=E5=88=86=E6=9E=90?= =?UTF-8?q?=E4=B8=8E=E9=A2=84=E7=AE=97=E8=B1=81=E5=85=8D=E6=97=A5=E6=8A=A5?= =?UTF-8?q?=E5=9C=BA=E6=99=AF=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit requeue job 每 15 分钟在预算余量 ≥30% 时（fail-closed），把最新分析为 local_fallback 的 ANALYZED 内容重置回 PENDING 让分析队列重新拿真实 LLM 分析——把 #90 的永久降级疤变一过性。四重防护：余量门控、60 分钟 冷却、每轮 50 条最旧优先、已有 ≥2 条 fallback 的不再回收（防确定性 降级在 attempts 归零后无限循环）。 预算闸豁免 daily_report/weekly_digest/monthly_digest 低频核心场景， 防日窗耗尽致日报硬失败；新增 budget_headroom_ok 供 requeue 门控。 .env.example 补圆桌四席数值对照表与收紧路线。 --- backend/.env.example | 17 +- backend/app/core/config.py | 3 + backend/app/repositories/content_repo.py | 71 +++++ backend/app/scheduler.py | 24 ++ backend/app/services/analysis_requeue.py | 47 +++ backend/app/services/llm/budget_guard.py | 45 +++ backend/tests/test_analysis_requeue.py | 354 +++++++++++++++++++++++ 7 files changed, 559 insertions(+), 2 deletions(-) create mode 100644 backend/app/services/analysis_requeue.py create mode 100644 backend/tests/test_analysis_requeue.py diff --git a/backend/.env.example b/backend/.env.example index ecfddef6..5909e26c 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -71,11 +71,24 @@ LLM_REQUESTS_PER_MINUTE=30 # 单次模型调用硬截止时间；模型配置中的 extra_params.litellm_params.timeout 只能缩短。 LLM_COMPLETION_TIMEOUT_SECONDS=45 # 预算熔断：按实际 DONE 调用数的三窗保险丝（分钟/小时/日），超限拒绝新付费调用， -# 分析链路自动走本地降级，窗口过后恢复。0 = 关闭对应窗口（默认全关，opt-in）。 -# 开启建议：分钟窗 ≥ LLM_REQUESTS_PER_MINUTE；日窗按可接受的烧钱上限设置。 +# 分析链路自动走本地降级（卡片标「本地速览·待 AI 复核」），预算余量 >30% 时由 +# requeue job 自动补分析；窗口滚过后恢复。0 = 关闭对应窗口（代码默认全关，opt-in）。 LLM_BUDGET_CALLS_PER_MINUTE=60 LLM_BUDGET_CALLS_PER_HOUR=2000 LLM_BUDGET_CALLS_PER_DAY=20000 +# +# 数值校准依据（2026-09-30 圆桌，四席对照；实测负载：日峰 5865 / 时峰 508 / 分峰 24）： +# 现阶段维持 60/2000/20000——圆桌裁决：requeue 补分析刚落地（#90）、 +# 尚未稳定运行观察，误触发代价（本地降级）未完全可逆前天平向宽松； +# 灾难场景仍有 3.4x+ 余量。 +# SRE 席建议收紧 70/1500/10000-12000：限流 30/min 下时窗 2000 数学上不可达（持续上限 1800/h）； +# 成本席建议收紧 60/1200/10000：全旗舰路由最坏 ¥240-3600/日，次数上限锁不住金额上限； +# 产品席偏松 60/3000/20000：误触发=降级，代价不可逆时宁可松； +# 数据席维持 60/2000/20000：现值对正常流量零误伤；relation_discovery 9 天涨 3.6 倍，日窗最怕增长而非异常。 +# 收紧前提：requeue 补分析（已随 #90 落地）稳定运行后，可收紧至 60/1200/10000 并加 10k 告警线； +# 切换付费 API 后建议二期增加金额三窗（当前 total_cost 记账未实现，暂无数据可依）。 +# 预算豁免场景（逗号分隔）：低频核心承诺场景不做预算检查，防日报在日窗耗尽时硬失败。 +LLM_BUDGET_EXEMPT_SCENES=daily_report,weekly_digest,monthly_digest # OAuth (Google / GitHub 登录) # 留空则该 provider 不启用。前端登录页会根据已启用 provider 渲染对应按钮。 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 3383d3ee..84552c2c 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -183,6 +183,9 @@ class Settings(BaseSettings): LLM_BUDGET_CALLS_PER_MINUTE: int = 0 LLM_BUDGET_CALLS_PER_HOUR: int = 0 LLM_BUDGET_CALLS_PER_DAY: int = 0 + # 预算豁免场景（逗号分隔）：这些 scene 的调用不做预算检查。日报/周报/月报 + # 低频且是核心承诺（预算耗尽会让 daily_report 直接走 ERROR），默认豁免。 + LLM_BUDGET_EXEMPT_SCENES: str = "daily_report,weekly_digest,monthly_digest" ANALYSIS_WORKER_CONCURRENCY: int = 3 ANALYSIS_MAX_ATTEMPTS: int = 5 ANALYSIS_RETRY_BASE_DELAY_SECONDS: int = 60 diff --git a/backend/app/repositories/content_repo.py b/backend/app/repositories/content_repo.py index 995ae2af..f93031c9 100644 --- a/backend/app/repositories/content_repo.py +++ b/backend/app/repositories/content_repo.py @@ -783,6 +783,77 @@ async def list_favorites( # ── Today picks candidates (SQLite fallback) ───────────────── + async def reset_local_fallback_for_reanalysis( + self, + *, + limit: int = 50, + cooldown_minutes: int = 60, + max_age_days: int = 7, + ) -> int: + """Requeue local_fallback-analyzed content for real LLM analysis (#90). + + 降级触发原因（预算窗口/熔断）消除后由 requeue job 调用，把永久假分 + 变回真实分析。只回收「最新一条分析是 local_fallback」的 ANALYZED 内容： + + - 最旧优先（``limit`` 限速，避免一次灌满分析队列）； + - ``cooldown_minutes``：降级落库后至少等待，防止预算临界值附近 + requeue→再降级→再 requeue 的反复横跳； + - ``max_age_days``：只回收近 N 天（旧内容重分析价值低）； + - ``skip_analysis`` 为 True 的内容不回收（显式跳过 LLM 的语义优先）； + - 已有 ≥2 条 local_fallback 分析的内容不回收：attempts=0 重置使 + ANALYSIS_MAX_ATTEMPTS 失效，内容过滤类确定性降级（不消耗预算 + 计数、余量门控拦不住）反复 requeue 会浪费调用（#90 复核结论）。 + + 返回实际重置条数。SELECT+UPDATE 两步：UPDATE 复核 status=ANALYZED + 防两步间隙状态漂移。 + """ + from datetime import timedelta + + from sqlalchemy import update + + from app.models.analysis import AiAnalysis + + now = naive_utc_now() + latest_analysis_id = self._latest_analysis_id_subquery(AiAnalysis) + candidate_ids_stmt = ( + select(self.model.id) + .join(AiAnalysis, AiAnalysis.id == latest_analysis_id) + .where(self.model.status == ContentStatus.ANALYZED) + .where(self.model.skip_analysis.is_(False)) + .where(self.model.updated_at <= now - timedelta(minutes=cooldown_minutes)) + .where(self.model.created_at >= now - timedelta(days=max_age_days)) + .where(AiAnalysis.summary_source == "local_fallback") + .where( + self.model.id.notin_( + select(AiAnalysis.content_id) + .where(AiAnalysis.summary_source == "local_fallback") + .group_by(AiAnalysis.content_id) + .having(func.count() >= 2) + ) + ) + .order_by(self.model.created_at.asc()) + .limit(limit) + ) + rows = await self.db.execute(candidate_ids_stmt) + ids = [int(row[0]) for row in rows.all()] + if not ids: + return 0 + + result = await self.db.execute( + update(self.model) + .where(self.model.id.in_(ids)) + .where(self.model.status == ContentStatus.ANALYZED) + .values( + status=ContentStatus.PENDING, + updated_at=now, + analysis_attempts=0, + analysis_next_retry_at=None, + analysis_claim_token=None, + analysis_lease_expires_at=None, + ) + ) + return int(result.rowcount or 0) + async def list_for_today_picks( self, hours: int = 48, diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 8d5c1d3f..2ec70ad3 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -890,6 +890,20 @@ async def _normalize_content_events() -> dict: # ── Lifecycle helpers ───────────────────────────────────────────────── +async def _requeue_local_fallback() -> None: + """#90：预算/熔断恢复后补分析降级内容（余量门控见 analysis_requeue）。""" + from app.services.analysis_requeue import requeue_local_fallback_content + + try: + async with async_session() as db: + count = await requeue_local_fallback_content(db) + await db.commit() + if count: + logger.info("Scheduler: requeued %d local_fallback item(s) for re-analysis", count) + except Exception: + logger.exception("Scheduler: local_fallback requeue failed") + + def start_scheduler() -> None: """Register all scheduled jobs and start the scheduler.""" if scheduler.running: @@ -1113,6 +1127,16 @@ def start_scheduler() -> None: replace_existing=True, ) + # 降级内容补分析 (#90)：每 15 分钟在预算余量充足时，把 local_fallback + # 降级内容重置回 PENDING 让分析队列重新拿真实 LLM 分析 + scheduler.add_job( + _requeue_local_fallback, + trigger=IntervalTrigger(minutes=15), + id="requeue_local_fallback", + name="降级内容补分析（预算余量门控）", + replace_existing=True, + ) + # 趋势快照：每日04:30计算话题热度+关键词频率 # (错开 04:00 知乎抓取 + 04:15 聚类,避免资源竞争) scheduler.add_job( diff --git a/backend/app/services/analysis_requeue.py b/backend/app/services/analysis_requeue.py new file mode 100644 index 00000000..44499f36 --- /dev/null +++ b/backend/app/services/analysis_requeue.py @@ -0,0 +1,47 @@ +"""降级内容补分析（requeue）——#90 的核心修复。 + +LLM 降级路径（熔断 / 内容过滤 / 预算三种触发源）曾把内容写成 +``ANALYZED`` 终态后永久结疤（假分 61-64 窄带、无任何消费者捡起）。 +本模块在触发原因消除后把 ``summary_source='local_fallback'`` 的内容 +重置回 PENDING，让既有分析队列重新拿真实 LLM 分析——把「永久疤」 +变成「一过性」。 + +门控与限速（防预算临界值附近的反复横跳）： + +- **预算余量门控**：最紧启用窗口余量 < 30% 时不发起（fail-closed， + 见 budget_guard.budget_headroom_ok）； +- **冷却**：降级落库后至少等 60 分钟才可回收； +- **限量**：每轮最多 50 条，最旧优先。 +""" + +from __future__ import annotations + +import logging + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.repositories.content_repo import ContentRepo +from app.services.llm.budget_guard import budget_headroom_ok + +logger = logging.getLogger(__name__) + + +async def requeue_local_fallback_content( + db: AsyncSession, + *, + limit: int = 50, + cooldown_minutes: int = 60, +) -> int: + """预算余量充足时，把降级内容重置回待分析。返回重置条数。""" + if not await budget_headroom_ok(min_ratio=0.3): + logger.info("Analysis requeue: LLM budget headroom insufficient, deferring") + return 0 + + count = await ContentRepo(db).reset_local_fallback_for_reanalysis( + limit=limit, + cooldown_minutes=cooldown_minutes, + ) + if count: + # 恢复是用户可感知的质量事件，按 warning 级别留痕（不是静默修复） + logger.warning("Analysis requeue: reset %d local_fallback content item(s) back to PENDING", count) + return count diff --git a/backend/app/services/llm/budget_guard.py b/backend/app/services/llm/budget_guard.py index 6bdef19e..e86c7d25 100644 --- a/backend/app/services/llm/budget_guard.py +++ b/backend/app/services/llm/budget_guard.py @@ -9,6 +9,9 @@ - 计数查询失败时 fail-open（保险丝自身故障不阻断主链路），记 warning。 - 预算拒绝写一行 ``llm_call_logs`` 审计记录（status="BUDGET_REJECTED"， 不参与 DONE 计数），使拒绝在用量看板 ALL 视图可见。 +- 豁免场景（``LLM_BUDGET_EXEMPT_SCENES``，默认日报/周报/月报）：低频 + 核心承诺场景不检查预算——日报生成每天仅数次，不该被熔断硬失败 + （daily_report 预算耗尽会走 ERROR 路径，违背核心承诺）。 """ from __future__ import annotations @@ -36,8 +39,14 @@ def _budget_limits() -> tuple[int, int, int] | None: return limits +def _exempt_scenes() -> set[str]: + return {s.strip() for s in settings.LLM_BUDGET_EXEMPT_SCENES.split(",") if s.strip()} + + async def ensure_llm_budget(scene: str) -> None: """Raise :class:`LlmBudgetExceededError` when a budget window is exhausted.""" + if scene in _exempt_scenes(): + return limits = _budget_limits() if limits is None: return @@ -79,3 +88,39 @@ async def _record_budget_rejection(*, scene: str, detail: str) -> None: ) except Exception as exc: # noqa: BLE001 — 审计写失败不能影响拒绝路径本身 logger.warning("LLM budget rejection audit log skipped: %s", exc) + + +async def budget_headroom_ok(min_ratio: float = 0.3) -> bool: + """最紧启用窗口的剩余比例是否 ≥ ``min_ratio``，供降级内容补分析（requeue）门控。 + + 与 :func:`ensure_llm_budget` 的 fail-open 相反，这里 fail-closed： + requeue 是主动追加的负载，计数读不到时不应发起（宁可晚一轮补分析）。 + 预算闸整体关闭（三窗全 0）时视为余量无限，恒 True。 + """ + limits = _budget_limits() + if limits is None: + return True + + from app.services.llm_usage import count_recent_llm_calls + + try: + calls = await count_recent_llm_calls() + except Exception as exc: # noqa: BLE001 — fail-closed：读不到计数就不补 + logger.info("LLM budget headroom check skipped (count query failed): %s", exc) + return False + + for label, used, limit in ( + ("minute", calls[0], limits[0]), + ("hour", calls[1], limits[1]), + ("day", calls[2], limits[2]), + ): + if limit > 0 and (limit - used) / limit < min_ratio: + logger.info( + "LLM budget headroom below %.0f%% in %s window (%d/%d used), deferring requeue", + min_ratio * 100, + label, + used, + limit, + ) + return False + return True diff --git a/backend/tests/test_analysis_requeue.py b/backend/tests/test_analysis_requeue.py new file mode 100644 index 00000000..d5e1b549 --- /dev/null +++ b/backend/tests/test_analysis_requeue.py @@ -0,0 +1,354 @@ +"""降级内容补分析（requeue）与预算豁免/余量门控的回归测试（#90）。 + +覆盖： +- repo.reset_local_fallback_for_reanalysis：资格（最新分析是 local_fallback）、 + cooldown、skip_analysis、max_age、limit 最旧优先、reset 字段语义； +- requeue_local_fallback_content：预算余量门控； +- budget_guard.budget_headroom_ok：全关、余量足、不足、fail-closed； +- 预算豁免场景（daily_report 等）超限仍放行。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.core.config import settings +from app.core.database import Base +from app.models.analysis import AiAnalysis +from app.models.content import ContentItem, ContentStatus +from app.models.source import Source, SourceStatus, SourceType +from app.repositories.content_repo import ContentRepo +from app.services.analysis_requeue import requeue_local_fallback_content +from app.services.llm.budget_guard import ( + LlmBudgetExceededError, + budget_headroom_ok, + ensure_llm_budget, +) + + +def _set_budget(monkeypatch, minute: int, hour: int, day: int) -> None: + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_MINUTE", minute) + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_HOUR", hour) + monkeypatch.setattr(settings, "LLM_BUDGET_CALLS_PER_DAY", day) + + +def _content( + item_id: int, + *, + now: datetime, + updated_at: datetime | None = None, + created_at: datetime | None = None, + skip_analysis: bool = False, + status: ContentStatus = ContentStatus.ANALYZED, +) -> ContentItem: + return ContentItem( + id=item_id, + title=f"降级样本 {item_id}", + url=f"https://example.com/fallback-{item_id}", + source_id=1, + source_name="降级测试信源", + source_type="RSS", + category="AI", + status=status, + skip_analysis=skip_analysis, + crawled_at=now, + published_at=now, + created_at=created_at or now, + updated_at=updated_at or now - timedelta(minutes=120), + ) + + +async def _seed(db, now: datetime) -> None: + db.add( + Source( + id=1, + name="降级测试信源", + source_type=SourceType.RSS, + url="https://example.com/fallback.xml", + category="AI", + status=SourceStatus.ACTIVE, + enabled=True, + weight=3, + ) + ) + # id=1：最新分析是 local_fallback，超过 cooldown —— 应被回收 + db.add(_content(1, now=now)) + # id=2：最新分析是真实 LLM（fallback 之后又真实分析过）—— 不动 + db.add(_content(2, now=now)) + # id=3：fallback 但 cooldown 内（updated_at 刚刷新）—— 不动 + db.add(_content(3, now=now, updated_at=now - timedelta(minutes=5))) + # id=4：fallback 但显式 skip_analysis —— 不动 + db.add(_content(4, now=now, skip_analysis=True)) + # id=5：fallback 但超过 max_age_days —— 不动 + db.add(_content(5, now=now, created_at=now - timedelta(days=10))) + + analyses = [ + # id=1：一条真实分析（旧）+ 一条 fallback（新）→ 最新是 fallback + AiAnalysis(id=101, content_id=1, curation_score=80, created_at=now - timedelta(hours=3)), + AiAnalysis( + id=102, + content_id=1, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=2), + ), + # id=2：fallback（旧）+ 真实分析（新）→ 最新是真实 + AiAnalysis( + id=103, + content_id=2, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=3), + ), + AiAnalysis(id=104, content_id=2, curation_score=85, created_at=now - timedelta(hours=1)), + # id=3/4/5：最新分析均为 fallback + AiAnalysis( + id=105, + content_id=3, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(minutes=10), + ), + AiAnalysis( + id=106, + content_id=4, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=2), + ), + AiAnalysis( + id=107, + content_id=5, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(days=9), + ), + ] + db.add_all(analyses) + await db.commit() + + +@pytest.mark.asyncio +async def test_reset_local_fallback_eligibility_and_fields(): + """只有「最新分析是 fallback 且过冷却」的内容被回收，reset 字段语义正确。""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime.now(UTC) + async with session_factory() as db: + await _seed(db, now) + repo = ContentRepo(db) + count = await repo.reset_local_fallback_for_reanalysis(limit=10, cooldown_minutes=60, max_age_days=7) + assert count == 1, "只有 id=1 同时满足最新分析为 fallback、过冷却、未跳过、未过期" + + item1 = await db.get(ContentItem, 1) + assert item1 is not None + assert item1.status == ContentStatus.PENDING + assert item1.analysis_attempts == 0 + assert item1.analysis_next_retry_at is None + assert item1.analysis_claim_token is None + assert item1.analysis_lease_expires_at is None + + for unchanged_id in (2, 3, 4, 5): + item = await db.get(ContentItem, unchanged_id) + assert item is not None + assert item.status == ContentStatus.ANALYZED, f"id={unchanged_id} 不应被回收" + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_reset_respects_limit_oldest_first(): + """limit 限速且最旧优先。""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime.now(UTC) + async with session_factory() as db: + db.add( + Source( + id=1, + name="信源", + source_type=SourceType.RSS, + url="https://example.com/x.xml", + category="AI", + status=SourceStatus.ACTIVE, + enabled=True, + weight=3, + ) + ) + # 三条 fallback 内容，created_at 递减（id=10 最旧） + for item_id, age_hours in ((10, 6), (11, 4), (12, 2)): + db.add( + _content( + item_id, + now=now, + created_at=now - timedelta(hours=age_hours), + updated_at=now - timedelta(hours=age_hours), + ) + ) + db.add( + AiAnalysis( + id=200 + item_id, + content_id=item_id, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=age_hours), + ) + ) + await db.commit() + + repo = ContentRepo(db) + count = await repo.reset_local_fallback_for_reanalysis(limit=1, cooldown_minutes=60) + assert count == 1 + pending_ids = [ + int(row[0]) + for row in ( + await db.execute(select(ContentItem.id).where(ContentItem.status == ContentStatus.PENDING)) + ).all() + ] + assert pending_ids == [10], "limit=1 时应回收最旧的 id=10" + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_requeue_service_gated_by_budget_headroom(monkeypatch): + """预算余量不足时不发起回收（不触碰 repo）。""" + from unittest.mock import AsyncMock + + reset_mock = AsyncMock(return_value=5) + monkeypatch.setattr(ContentRepo, "reset_local_fallback_for_reanalysis", reset_mock) + + async def _no_headroom(min_ratio: float = 0.3) -> bool: + return False + + monkeypatch.setattr("app.services.analysis_requeue.budget_headroom_ok", _no_headroom) + + count = await requeue_local_fallback_content(object()) # db 不应被触碰 + assert count == 0 + reset_mock.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_budget_headroom_semantics(monkeypatch): + # 全关（三窗全 0）→ 余量无限 + _set_budget(monkeypatch, 0, 0, 0) + + async def _boom(): + raise AssertionError("budget disabled must not query counts") + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _boom) + assert await budget_headroom_ok() is True + + # 余量充足 + _set_budget(monkeypatch, 100, 2000, 20000) + + async def _counts(): + return 10, 500, 10000 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + assert await budget_headroom_ok() is True + + # 最紧窗口（日窗 10000/20000 = 50% 已用，余量 50% ≥30%）；改为 18000（余量 10%）→ False + async def _tight(): + return 5, 100, 18000 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _tight) + assert await budget_headroom_ok() is False + + # fail-closed：计数查询失败 → False（requeue 不发起） + async def _db_error(): + raise RuntimeError("db unavailable") + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _db_error) + assert await budget_headroom_ok() is False + + +@pytest.mark.asyncio +async def test_exempt_scenes_bypass_budget(monkeypatch): + """日报/周报/月报等豁免场景超限仍放行；非豁免场景照常拒绝。""" + _set_budget(monkeypatch, minute=1, hour=0, day=0) + + async def _counts(): + return 1, 0, 0 # 分钟窗已耗尽 + + monkeypatch.setattr("app.services.llm_usage.count_recent_llm_calls", _counts) + + for scene in ("daily_report", "weekly_digest", "monthly_digest"): + await ensure_llm_budget(scene) # 豁免：不抛 + + with pytest.raises(LlmBudgetExceededError): + await ensure_llm_budget("content_analysis") + + +@pytest.mark.asyncio +async def test_reset_skips_content_with_repeated_fallbacks(): + """已有 ≥2 条 local_fallback 分析的内容不再回收（防确定性降级无限循环）。""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime.now(UTC) + async with session_factory() as db: + db.add( + Source( + id=1, + name="信源", + source_type=SourceType.RSS, + url="https://example.com/y.xml", + category="AI", + status=SourceStatus.ACTIVE, + enabled=True, + weight=3, + ) + ) + # id=20：两条 fallback 分析（requeue 重降级过一次）—— 不再回收 + db.add(_content(20, now=now)) + # id=21：一条 fallback —— 正常回收（对照组） + db.add(_content(21, now=now)) + db.add_all( + [ + AiAnalysis( + id=301, + content_id=20, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=4), + ), + AiAnalysis( + id=302, + content_id=20, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=2), + ), + AiAnalysis( + id=303, + content_id=21, + curation_score=62, + summary_source="local_fallback", + created_at=now - timedelta(hours=2), + ), + ] + ) + await db.commit() + + repo = ContentRepo(db) + count = await repo.reset_local_fallback_for_reanalysis(limit=10) + assert count == 1 + item20 = await db.get(ContentItem, 20) + item21 = await db.get(ContentItem, 21) + assert item20 is not None and item20.status == ContentStatus.ANALYZED + assert item21 is not None and item21.status == ContentStatus.PENDING + + await engine.dispose() From ae59cc4568f657253e878571c70c5c0bae583b0e Mon Sep 17 00:00:00 2001 From: fxbin Date: Wed, 30 Sep 2026 21:11:03 +0800 Subject: [PATCH 06/10] =?UTF-8?q?fix(trending):=20P70=20=E9=97=A8=E6=A7=9B?= =?UTF-8?q?=E6=8E=92=E9=99=A4=E9=99=8D=E7=BA=A7=E5=81=87=E5=88=86=E5=B9=B6?= =?UTF-8?q?=E5=9C=A8=E5=8D=A1=E7=89=87=E6=A0=87=E8=AE=B0=E5=BE=85=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scoring_engine 的百分位门槛只由真实 LLM 评分决定：local_fallback 的 确定性假分（curation 收敛 61-64 窄带）不再拖拽 P70 门槛、挤掉真实 65-70 分内容；全降级批次回退全量计算。降级项仍参与 selected 判定 与展示，由 requeue job 换回真实分析。 summary_source 透传补齐全部构造点：DuckDB 主路径、OLTP fallback、 API 响应 payload（前端徽标的数据源）、周报/月报 digest 路径（P70 同类污染一并修复）。today-picks 卡片显示「本地速览 · 待 AI 复核」 琥珀标记，悬停说明自动恢复机制。P70 排除测试为三组自证明对照 （含守门断言，删掉排除逻辑必红）。 --- backend/app/services/_duckdb_reports_mixin.py | 4 +- backend/app/services/digest_context.py | 1 + backend/app/services/scoring_engine.py | 15 +++++- backend/app/services/scoring_inputs.py | 1 + backend/app/services/today_picks.py | 3 ++ backend/tests/test_duckdb_service.py | 14 ++--- backend/tests/test_scoring_engine.py | 54 +++++++++++++++++++ frontend/src/app/today-picks/_components.tsx | 8 +++ frontend/src/types/index.ts | 2 + 9 files changed, 93 insertions(+), 9 deletions(-) diff --git a/backend/app/services/_duckdb_reports_mixin.py b/backend/app/services/_duckdb_reports_mixin.py index f5e01262..6b85d01a 100644 --- a/backend/app/services/_duckdb_reports_mixin.py +++ b/backend/app/services/_duckdb_reports_mixin.py @@ -80,7 +80,8 @@ def query_content_for_weekly(self, start_date: str, end_date: str) -> list[dict[ COALESCE(f.feedback_score, 0) AS feedback_score, COALESCE(a.curation_score, 0) + LEAST({feedback_max}, GREATEST({feedback_min}, COALESCE(f.feedback_score, 0))) * {feedback_weight} - AS adjusted_score + AS adjusted_score, + a.summary_source FROM oltp_db.content_items c LEFT JOIN latest_analysis a ON a.content_id = c.id LEFT JOIN oltp_db.sources s ON s.id = c.source_id @@ -120,6 +121,7 @@ def query_content_for_weekly(self, start_date: str, end_date: str) -> list[dict[ "source_weight_db": int(row[21]) if row[21] else 3, "feedback_score": float(row[22]) if row[22] else 0, "adjusted_score": round(float(row[23] or 0), 1), + "summary_source": row[24], } for row in results ] diff --git a/backend/app/services/digest_context.py b/backend/app/services/digest_context.py index b13d565c..7304c862 100644 --- a/backend/app/services/digest_context.py +++ b/backend/app/services/digest_context.py @@ -67,6 +67,7 @@ def value_or_default(value, default): category=row.get("category"), source_name=row.get("source_name"), crawled_at=row.get("crawled_at"), + summary_source=row.get("summary_source"), curation_score=value_or_default(row.get("curation_score"), 0), info_density=value_or_default(row.get("info_density"), 50), actionability=value_or_default(row.get("actionability"), 50), diff --git a/backend/app/services/scoring_engine.py b/backend/app/services/scoring_engine.py index a1471bd3..962d974b 100644 --- a/backend/app/services/scoring_engine.py +++ b/backend/app/services/scoring_engine.py @@ -89,15 +89,18 @@ class ScoringInput: "source_weight_db", # Source.weight from DB (1-5) # Feedback signal "feedback_score", # user feedback signal (0+, default 0) + # Analysis provenance + "summary_source", # e.g. "local_fallback" for degraded local analysis (#90) ) def __init__(self, **kwargs): for slot in self.__slots__: if slot == "feedback_score": setattr(self, slot, kwargs.get(slot, 0)) - elif slot in ("published_at", "crawled_at"): + elif slot in ("published_at", "crawled_at", "summary_source"): # datetime 字段默认 None（表示"没有"），不能是 0（int）， # 否则 _compute_time_decay 的 ensure_aware_utc(0) 会崩。 + # summary_source 同理默认 None（表示真实 LLM 分析）。 setattr(self, slot, kwargs.get(slot)) else: setattr(self, slot, kwargs.get(slot, 0)) @@ -442,7 +445,15 @@ def score_items(items: list[ScoringInput]) -> list[tuple[ScoreBreakdown, Scoring if cfg["curation_mode"] == "percentile": # Use final_score ranking: top (100 - percentile)% are selected # e.g. curation_percentile=70 -> top 30% selected, bounded by base quality. - final_scores = [bd.final_score for bd, _ in results] + # P70 门槛只由真实 LLM 评分决定：local_fallback 的确定性假分 + # （curation 收敛 61-64 窄带）不得污染百分位门槛、挤掉真实内容。 + # 降级项仍参与 selected 判定与展示（前端带「本地速览」标记）， + # 由 requeue job 尽快换回真实分析（#90）。全候选皆降级时回退全量， + # 避免 _compute_percentile_threshold 收到空列表。 + real_scores = [ + bd.final_score for bd, item in results if item.summary_source != "local_fallback" + ] + final_scores = real_scores or [bd.final_score for bd, _ in results] actual_threshold = _compute_percentile_threshold( final_scores, cfg["curation_percentile"], diff --git a/backend/app/services/scoring_inputs.py b/backend/app/services/scoring_inputs.py index 38be516c..4a13fbf5 100644 --- a/backend/app/services/scoring_inputs.py +++ b/backend/app/services/scoring_inputs.py @@ -44,6 +44,7 @@ def value_or_default(value: Any, default: float | int) -> float | int: content_id=item.id, title=item.title, category=item.category, + summary_source=analysis.summary_source, source_id=item.source_id, source_name=item.source_name, published_at=item.published_at, diff --git a/backend/app/services/today_picks.py b/backend/app/services/today_picks.py index 4c05e98e..d58aa937 100644 --- a/backend/app/services/today_picks.py +++ b/backend/app/services/today_picks.py @@ -431,6 +431,7 @@ def value_or_default(value, default): content_id=row["id"], title=row.get("title") or "", category=row.get("category"), + summary_source=row.get("summary_source"), source_id=row.get("source_id"), source_name=row.get("source_name"), published_at=row.get("published_at"), @@ -480,6 +481,8 @@ def _row_to_content_payload(row: dict, breakdown: ScoreBreakdown) -> dict: "short_video_plan": _decode_json_value(row.get("short_video_plan")), "risk_notes": _decode_json_value(row.get("risk_notes")), "curation_score": row.get("curation_score") or 0, + # 分析来源：'local_fallback' 时前端显示「本地速览·待 AI 复核」标记（#90） + "summary_source": row.get("summary_source"), "tags": analysis_tags, "recommendation": _clean_optional_text(row.get("recommendation")), "info_density": row.get("info_density") or 0, diff --git a/backend/tests/test_duckdb_service.py b/backend/tests/test_duckdb_service.py index ab9904af..89378075 100644 --- a/backend/tests/test_duckdb_service.py +++ b/backend/tests/test_duckdb_service.py @@ -848,7 +848,8 @@ def test_digest_content_query_uses_latest_analysis_and_feedback_order(monkeypatc tags VARCHAR, recommendation VARCHAR, recommended_reason VARCHAR, - created_at TIMESTAMP + created_at TIMESTAMP, + summary_source VARCHAR ) """) conn.execute(""" @@ -882,15 +883,15 @@ def test_digest_content_query_uses_latest_analysis_and_feedback_order(monkeypatc [now, now], ) conn.execute( - "INSERT INTO oltp_db.ai_analyses VALUES (1, 1, '旧摘要', 95, 95, 95, 95, 95, 10, 99, 95, 95, 95, '[\"旧\"]', '旧推荐', '旧理由', ?)", + "INSERT INTO oltp_db.ai_analyses VALUES (1, 1, '旧摘要', 95, 95, 95, 95, 95, 10, 99, 95, 95, 95, '[\"旧\"]', '旧推荐', '旧理由', ?, NULL)", [now - timedelta(hours=2)], ) conn.execute( - "INSERT INTO oltp_db.ai_analyses VALUES (2, 1, '新摘要', 70, 70, 70, 65, 80, 10, 70, 82, 81, 72, '[\"新\"]', '新推荐', '新理由', ?)", + "INSERT INTO oltp_db.ai_analyses VALUES (2, 1, '新摘要', 70, 70, 70, 65, 80, 10, 70, 82, 81, 72, '[\"新\"]', '新推荐', '新理由', ?, NULL)", [now - timedelta(hours=1)], ) conn.execute( - "INSERT INTO oltp_db.ai_analyses VALUES (3, 2, '无反馈摘要', 72, 72, 72, 60, 75, 10, 72, 74, 73, 50, '[\"AI\"]', '推荐', '理由', ?)", + "INSERT INTO oltp_db.ai_analyses VALUES (3, 2, '无反馈摘要', 72, 72, 72, 60, 75, 10, 72, 74, 73, 50, '[\"AI\"]', '推荐', '理由', ?, NULL)", [now], ) conn.execute("INSERT INTO oltp_db.user_feedback VALUES (1, 1, 1, 20.0, ?)", [now]) @@ -1020,7 +1021,8 @@ def test_digest_content_queries_exclude_ignored_content(monkeypatch): tags VARCHAR, recommendation VARCHAR, recommended_reason VARCHAR, - created_at TIMESTAMP + created_at TIMESTAMP, + summary_source VARCHAR ) """) conn.execute(""" @@ -1049,7 +1051,7 @@ def test_digest_content_queries_exclude_ignored_content(monkeypatch): conn.execute( """ INSERT INTO oltp_db.ai_analyses VALUES ( - ?, ?, '摘要', 90, 90, 90, 90, 90, 10, 90, 90, 90, 90, '["AI"]', '推荐', '理由', ? + ?, ?, '摘要', 90, 90, 90, 90, 90, 10, 90, 90, 90, 90, '["AI"]', '推荐', '理由', ?, NULL ) """, [content_id, content_id, now], diff --git a/backend/tests/test_scoring_engine.py b/backend/tests/test_scoring_engine.py index 83ef106a..d1f2140c 100644 --- a/backend/tests/test_scoring_engine.py +++ b/backend/tests/test_scoring_engine.py @@ -210,3 +210,57 @@ def test_diversity_promotes_other_source_over_same_source_run(): # 第一名仍是分数最高的同源第一条；同源第二条因惩罚被其它来源反超 assert final_ids[0] == 1 assert final_ids.index(99) < final_ids.index(2) + + +def test_p70_threshold_excludes_local_fallback_fake_scores(): + """#90：local_fallback 的确定性假分不得污染 P70 门槛。 + + 三组对照（自证明区分度——守门断言保证若删掉排除逻辑本测试必红）： + - real_only：3 条真实分（90/80/70），门槛 T1 只由真实分决定； + - mixed：再混入 2 条假分（85/75，标记 local_fallback）——真实项门槛 + 仍应为 T1（排除生效）； + - full：同样的 2 条 85/75 但标记为真实分析——门槛 T2 应不同于 T1， + 证明该分数组合确实会移动门槛（若排除逻辑失效，mixed 会退化成 + full 的门槛，测试转红）。 + """ + + def _scored(content_id: int, curation: int, **extra) -> ScoringInput: + return _item(content_id, curation_score=curation, **extra) + + real_items = [_scored(1, 90), _scored(2, 80), _scored(3, 70)] + fallback_items = [ + _scored(98, 85, summary_source="local_fallback"), + _scored(99, 75, summary_source="local_fallback"), + ] + same_scores_as_real = [_scored(98, 85), _scored(99, 75)] # 不带标记 + + baseline = score_items(list(real_items)) + mixed = score_items(real_items + fallback_items) + full = score_items(real_items + same_scores_as_real) + + t_baseline = {bd.threshold_used for bd, _ in baseline} + t_mixed_real = {bd.threshold_used for bd, item in mixed if item.content_id in (1, 2, 3)} + t_full = {bd.threshold_used for bd, _ in full} + + assert len(t_baseline) == 1 and len(t_full) == 1 + assert t_mixed_real == t_baseline, "混入 fallback 假分后真实项门槛不应改变" + assert t_full != t_baseline, "守门断言：85/75 混入真实批应移动门槛——本断言失败说明对照构造失去区分度" + + # fallback 项自身仍参与判定与展示（不静默消失），且用同一门槛判定 + fallback_results = {item.content_id: bd for bd, item in mixed if item.content_id in (98, 99)} + assert len(fallback_results) == 2 + assert {bd.threshold_used for bd in fallback_results.values()} == t_baseline + + +def test_all_fallback_batch_falls_back_to_full_scores(): + """全候选皆降级时回退全量计算，P70 引擎不因空集崩溃。""" + fallback_items = [_item(i, summary_source="local_fallback", curation_score=62) for i in range(1, 5)] + scored = score_items(fallback_items) + assert len(scored) == 4 + assert all(bd.threshold_used is not None for bd, _ in scored) + + +def test_scoring_input_defaults_summary_source_none(): + """未传 summary_source 的旧构造路径默认 None（表示真实 LLM 分析）。""" + item = _item(1) + assert item.summary_source is None diff --git a/frontend/src/app/today-picks/_components.tsx b/frontend/src/app/today-picks/_components.tsx index c2d44d5e..5a562471 100644 --- a/frontend/src/app/today-picks/_components.tsx +++ b/frontend/src/app/today-picks/_components.tsx @@ -630,6 +630,14 @@ export function PickCard({ / {timeAgo(item.published_at || item.crawled_at)} + {analysis?.summary_source === 'local_fallback' && ( + + 本地速览 · 待 AI 复核 + + )} {item.category && {item.category}} {item.content_type && {item.content_type}} {tags.slice(0, 3).map((tag) => ( diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index a27eb3c8..7cc52682 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -340,6 +340,8 @@ export interface ContentAnalysis { creator_score: number; viral_score: number; risk_score: number; + // 分析来源标记：'local_fallback' = 降级本地速览（#90，前端据此显示待复核标记） + summary_source?: string | null; platform_fit?: Record | null; recommended_reason?: string | null; summary?: string | null; From e9fd62021e0839efb9bb9ed1b2e54cea8bb21b11 Mon Sep 17 00:00:00 2001 From: fxbin Date: Thu, 1 Oct 2026 21:49:25 +0800 Subject: [PATCH 07/10] =?UTF-8?q?test(scoring):=20=E5=85=A8=E9=99=8D?= =?UTF-8?q?=E7=BA=A7=E6=89=B9=E6=AC=A1=E6=96=AD=E8=A8=80=E9=97=A8=E6=A7=9B?= =?UTF-8?q?=E5=8F=96=E8=87=AA=E6=9C=AC=E6=89=B9=E5=88=86=E6=95=B0=E8=80=8C?= =?UTF-8?q?=E9=9D=9E=E5=85=A8=E5=B1=80=E9=BB=98=E8=AE=A4=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原断言只检查 threshold_used 非空，而「本批分数」和「全局默认」两条路径 都会给出数字，删掉被测的 `or [bd.final_score ...]` 回退分支后测试照样全绿。 改成断言门槛等于本批分数的 P70、且不等于 CONFIG["curation_threshold"]， 并要求本批至少选出一条。这条回退分支不是防御性冗余：真实分为空时 _compute_percentile_threshold 返回全局默认 55，而降级内容的 final_score 约为 19.94，全部低于 55，缺了这个分支整批一条都选不出来。 变异验证：删掉该分支后本测试转红（55 != 19.94）。 --- backend/tests/test_scoring_engine.py | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/backend/tests/test_scoring_engine.py b/backend/tests/test_scoring_engine.py index d1f2140c..82a14e37 100644 --- a/backend/tests/test_scoring_engine.py +++ b/backend/tests/test_scoring_engine.py @@ -1,6 +1,11 @@ from datetime import UTC, datetime, timedelta -from app.services.scoring_engine import CONFIG, ScoringInput, score_items +from app.services.scoring_engine import ( + CONFIG, + ScoringInput, + _compute_percentile_threshold, + score_items, +) _NOW = datetime(2026, 1, 1, 12, 0, 0) @@ -253,11 +258,27 @@ def _scored(content_id: int, curation: int, **extra) -> ScoringInput: def test_all_fallback_batch_falls_back_to_full_scores(): - """全候选皆降级时回退全量计算，P70 引擎不因空集崩溃。""" + """全候选皆降级时，门槛仍由本批实际分数决定，不退回全局默认阈值。 + + 这个分支不是防御性冗余：真实分为空时若直接把空列表交给 + ``_compute_percentile_threshold``，它会返回 ``CONFIG["curation_threshold"]`` + （55），而降级内容的 final_score 普遍低于该值，结果是整批一条都选不出来。 + """ fallback_items = [_item(i, summary_source="local_fallback", curation_score=62) for i in range(1, 5)] scored = score_items(fallback_items) assert len(scored) == 4 - assert all(bd.threshold_used is not None for bd, _ in scored) + + thresholds = {bd.threshold_used for bd, _ in scored} + assert len(thresholds) == 1 + threshold = thresholds.pop() + + own_scores = [bd.final_score for bd, _ in scored] + assert threshold == _compute_percentile_threshold(own_scores, 70), "门槛应由本批实际分数决定" + + # 守门断言：门槛退回全局默认时本批会全部低于阈值、页面一条都不出。 + # 删掉 score_items 里的 `or [bd.final_score ...]` 回退分支，本测试必红。 + assert threshold != CONFIG["curation_threshold"] + assert any(bd.selected for bd, _ in scored), "门槛来自本批分数时至少应选出一条" def test_scoring_input_defaults_summary_source_none(): From 191de4bda346bf071a4e45e6f60bcd52695d1f24 Mon Sep 17 00:00:00 2001 From: fxbin Date: Thu, 1 Oct 2026 21:49:31 +0800 Subject: [PATCH 08/10] =?UTF-8?q?docs:=20=E5=9B=9E=E5=BD=92=E7=9F=A9?= =?UTF-8?q?=E9=98=B5=E6=9B=B4=E6=96=B0=20#90=20=E4=BF=AE=E5=A4=8D=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E3=80=81=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95=E4=B8=8E?= =?UTF-8?q?=E5=8F=98=E5=BC=82=E9=AA=8C=E8=AF=81=E7=BB=93=E8=AE=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #90 修复已随 wip-llm-budget-guard 分支落地，条目仍写「待修复」和「回归测试 待补」，与实际状态矛盾。补上状态、落地范围、测试条数与变异验证结论。 顺带把条目里的「结疤」换成具体说法，与 AGENTS.md「中文说明怎么写」一致。 --- docs/quality/regression-matrix.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/quality/regression-matrix.md b/docs/quality/regression-matrix.md index b7b4bbbc..b3aaa758 100644 --- a/docs/quality/regression-matrix.md +++ b/docs/quality/regression-matrix.md @@ -120,16 +120,16 @@ - 回归测试：`npx vitest run src/lib/__tests__/admin-nav.test.ts`（13 条）。导航清单与面包屑映射合并到 `src/lib/admin-nav.ts` 单一事实源，映射由清单派生；`nav-checklist` 等价断言（壳内每项都有映射、label 与清单逐项一致）已在其中，漏项即红。顺带修掉一个被测试抓出的旧行为：`/admin/任何未知路径` 曾因壳根前缀匹配被判成「概览」，现已回退兜底。 - owner：#88。 -### LLM 降级内容永久结疤：local_fallback 无恢复路径 — #90 🐛 待修复（2026-09-30） +### LLM 降级内容再也不会被重新分析：local_fallback 无恢复路径 — #90 🔧 修复完成待合并（2026-10-01） - 症状：LLM 降级路径（熔断 `CircuitOpenError` / 内容过滤 `BadRequestError` / 预算 `LlmBudgetExceededError` 三种触发源共用）把内容写为 `ANALYZED` 终态、`AiAnalysis.summary_source='local_fallback'` 落库后，**全代码库没有任何 requeue 消费者**——重分析资格谓词只认 PENDING/僵死 ANALYZING/ERROR，手动分析端点遇已有记录早退。降级内容是确定性假分数（curation 61-64 窄带、risk 恒 28、模板推荐语），today-picks 以 40px 大字渲染且前端不读 `summary_source`，用户无法分辨；窄带假分参与 P70 计算会拖低百分位门槛、挤掉真实 65-70 分内容。 - 环境/前提：任一降级触发源命中即触发；当前部署全走本地模型（`llm_call_logs.total_cost` 全 0），预算闸实际不触发，触发面在切换付费 API 后成为现实。 - 复现：触发降级（如断言 `LlmBudgetExceededError` 路径或熔断 OPEN）→ 确认内容 `status=ANALYZED`、`summary_source='local_fallback'` → 检索全部重分析入口（`_analysis_candidate_condition`、`api/v1/analyses.py` 手动端点、scripts/）确认无路径捡起该内容。 - 期望 vs 实际：期望触发原因消除后（预算窗口滚过/熔断恢复）内容可被重新 LLM 分析；实际永久停留假分形态。 -- 边界：`analysis.py` 降级路径 + `content_repo.py` 重分析资格谓词；修复为 requeue job + P70 排除 + 前端「本地速览·待 AI 复核」标记 + 日报豁免的一揽子（详见 #90）。 -- 严重度：P1（预算日窗最坏锁 ~24h × 实测 ~250 条/时摄入 ≈ 数千条永久假分；当前零账单部署不触发故非 P0）；**复现性**：always（降级发生即结疤）。 -- 回归测试：待 requeue 落地补——造 fallback 内容 → 模拟预算恢复 → 断言内容回 PENDING 并被重新分析；P70 排除与前端标记各有对应测试（教训机器化随修复交付）。 +- 边界：`analysis.py` 降级路径 + `content_repo.py` 重分析资格谓词。已修（分支 `wip-llm-budget-guard`，2 笔提交）：新增 `services/analysis_requeue.py` requeue 服务 + scheduler 每 15 分钟调度，预算余量 ≥30% 且降级满 60 分钟后把 `local_fallback` 内容重置回 PENDING；P70 门槛改为只由真实 LLM 评分决定（`scoring_engine.py`）；`summary_source` 透传补齐 DuckDB 主路径、OLTP fallback、API payload 与周/月报 digest；today-picks 卡片显示「本地速览 · 待 AI 复核」标记；预算闸豁免 daily_report/weekly_digest/monthly_digest。 +- 严重度：P1（预算日窗最坏锁 ~24h × 实测 ~250 条/时摄入 ≈ 数千条永久假分；当前零账单部署不触发故非 P0）；**复现性**：always（降级发生即永久停留假分）。 +- 回归测试：`uv run python -m pytest tests/test_analysis_requeue.py -q`（6 条：资格与字段重置 / limit 最旧优先 / 服务层余量门控 / 余量阈值语义 / 豁免场景绕行 / ≥2 条 fallback 不再回收）；`tests/test_llm_budget_guard.py`（9 条）；`tests/test_scoring_engine.py -k "p70 or all_fallback"`（2 条）。变异验证（2026-10-01 独立复核）：删掉 P70 排除逻辑 → `test_p70_threshold_excludes_local_fallback_fake_scores` 转红；删掉全降级批次的 `or [...]` 回退分支 → `test_all_fallback_batch_falls_back_to_full_scores` 转红（**该分支非防御性冗余**：真实分为空时 `_compute_percentile_threshold` 返回全局默认 55，而降级内容 final_score 普遍低于 55，回退缺失会导致整批选不出内容）。全量后端套件仅 2 条 FAILED，均为 `test_source_url_safety.py` 依赖真实网络的存量红，在 main 上同样 FAILED，非本次引入。 - owner：#90。 -- **存量定性（勿误读）**：非预算闸（`feat(llm): 增加三窗预算熔断闸`）引入——熔断降级早就同样结疤；预算闸放大触发面，圆桌评审三席独立发现。发现方式：预算数值圆桌（2026-09-30，Memory 存 `.vidt/roundtable/budget-fuse-values/`，会话态）。 +- **存量定性（勿误读）**：非预算闸（`feat(llm): 增加三窗预算熔断闸`）引入——熔断降级早就同样留下假分内容；预算闸放大触发面，圆桌评审三席独立发现。发现方式：预算数值圆桌（2026-09-30，Memory 存 `.vidt/roundtable/budget-fuse-values/`，会话态）。 ## 三、关键流程基线（9 项） From 84c5c5528fee224b45d7791257e83fb6a31bb3b1 Mon Sep 17 00:00:00 2001 From: fxbin Date: Thu, 1 Oct 2026 21:49:35 +0800 Subject: [PATCH 09/10] =?UTF-8?q?chore:=20.gitignore=20=E5=BF=BD=E7=95=A5?= =?UTF-8?q?=20.worktrees=20=E6=9C=AC=E5=9C=B0=E5=B7=A5=E4=BD=9C=E5=8C=BA?= =?UTF-8?q?=E7=9B=AE=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent 用来隔离 issue 切片的 git worktree 放在仓库内 .worktrees/，否则 主工作区的 git status 会被它污染。不入库。 --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index e098e89e..f4ace5a5 100644 --- a/.gitignore +++ b/.gitignore @@ -73,3 +73,6 @@ htmlcov/ .brooks-lint-history.json .vidt/ .node-version + +# 独立 worktree 工作区（issue 切片隔离） +.worktrees/ From 237de31086d58e0b20ec56825bd1d74f81308edc Mon Sep 17 00:00:00 2001 From: fxbin Date: Sat, 3 Oct 2026 15:22:34 +0800 Subject: [PATCH 10/10] =?UTF-8?q?fix(scoring):=20scoring=5Fengine=20?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E5=AF=B9=E9=BD=90=20ruff=20=E5=B9=B6?= =?UTF-8?q?=E6=9B=B4=E6=AD=A3=E5=9B=9E=E9=80=80=E5=88=86=E6=94=AF=E6=B3=A8?= =?UTF-8?q?=E9=87=8A=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI backend-lint 阻塞：`ae59cc4` 里那个列表推导被折成 3 行，但并未超长， ruff format 要求单行。本地此前只跑 ruff check、漏了 format --check， 是 CI 抓出来的。 顺带更正上方注释的因果。原文写「避免 _compute_percentile_threshold 收到 空列表」，但空列表在函数内已被处理；真实原因是空列表会被回退到全局默认 阈值 55，而降级内容的 final_score 普遍低于它，那样整批一条都选不出来。 --- backend/app/services/scoring_engine.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/backend/app/services/scoring_engine.py b/backend/app/services/scoring_engine.py index 962d974b..c7a86084 100644 --- a/backend/app/services/scoring_engine.py +++ b/backend/app/services/scoring_engine.py @@ -448,11 +448,10 @@ def score_items(items: list[ScoringInput]) -> list[tuple[ScoreBreakdown, Scoring # P70 门槛只由真实 LLM 评分决定：local_fallback 的确定性假分 # （curation 收敛 61-64 窄带）不得污染百分位门槛、挤掉真实内容。 # 降级项仍参与 selected 判定与展示（前端带「本地速览」标记）， - # 由 requeue job 尽快换回真实分析（#90）。全候选皆降级时回退全量， - # 避免 _compute_percentile_threshold 收到空列表。 - real_scores = [ - bd.final_score for bd, item in results if item.summary_source != "local_fallback" - ] + # 由 requeue job 尽快换回真实分析（#90）。全候选皆降级时改用全量分数： + # 空列表会被 _compute_percentile_threshold 回退到全局默认阈值（55）， + # 而降级内容的 final_score 普遍低于它，那样整批一条都选不出来。 + real_scores = [bd.final_score for bd, item in results if item.summary_source != "local_fallback"] final_scores = real_scores or [bd.final_score for bd, _ in results] actual_threshold = _compute_percentile_threshold( final_scores,