From e8f76b43974dd0522cdb73dba45d2b69e19ce33d Mon Sep 17 00:00:00 2001 From: mumu <42829555+ZhuLinsen@users.noreply.github.com> Date: Sun, 5 Apr 2026 10:26:06 +0800 Subject: [PATCH] fix: wrap optional pipeline services in try/except for graceful degradation (#1004) * fix: wrap optional pipeline services in try/except for graceful degradation SearchService and SocialSentimentService init failures previously crashed the entire StockAnalysisPipeline constructor, preventing any analysis from running even when search/sentiment is non-essential. Changes: - Wrap SearchService init in try/except, fallback to None on failure - Wrap SocialSentimentService init in try/except, fallback to None on failure - Add null-safety guards at all 5 downstream call sites - Upgrade progress callback log from debug to warning for better visibility --- docs/CHANGELOG.md | 1 + docs/full-guide.md | 5 + docs/full-guide_EN.md | 5 + src/core/pipeline.py | 76 ++++++++----- ...st_pipeline_optional_service_resilience.py | 101 ++++++++++++++++++ 5 files changed, 163 insertions(+), 25 deletions(-) create mode 100644 tests/test_pipeline_optional_service_resilience.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index cdedc0335..b62a3a3fc 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [修复] 移除 `HistoryItem` 与 `ReportSummary` 响应 Schema 中 `sentiment_score` 的 `ge=0/le=100` 约束（fixes #942）——历史库中存储的超范围负值或大于 100 的情绪评分不再触发 Pydantic ValidationError，历史列表与详情接口恢复正常返回。 - [改进] Agent IntelAgent 新增公司公告搜索维度（上交所/深交所/cninfo）与主力资金流工具（get_capital_flow），修复 Agent 模式下公告和资金流数据经常缺失的问题 - [修复] webui_frontend.py 在 static/index.html 存在但 static/assets/ 缺失时发出明确警告，避免用户因 CSS/JS 资源缺失导致页面元素异常变大却无从排查 +- [修复] `StockAnalysisPipeline` 搜索服务与社交舆情服务改为可选降级初始化：任一服务初始化异常时记录 warning 并以禁用状态继续运行，避免外部依赖抖动阻塞主分析链路与 SSE 进度回调。 - [文档] DEPLOY.md 和 deploy-webui-cloud.md 新增"UI 元素异常变大/布局错乱"排查步骤（重建 Docker 镜像或手动执行 npm run build） - [文档] 补充飞书 Webhook 配置说明：强调 `FEISHU_WEBHOOK_URL` 是群通知必填项、`FEISHU_WEBHOOK_SECRET` 与飞书机器人「签名校验」必须两端同时启用或同时关闭、`FEISHU_APP_SECRET` 仅用于应用/Stream Bot 模式不可替代 Webhook；同步完善英文指南并在 `.env.example` 为相关配置项补充内联说明注释 diff --git a/docs/full-guide.md b/docs/full-guide.md index da37bb903..7f8293588 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -262,12 +262,16 @@ daily_stock_analysis/ | `BOCHA_API_KEYS` | 博查搜索 API Key（中文优化） | 可选 | | `BRAVE_API_KEYS` | Brave Search API Key（美股优化） | 可选 | | `SERPAPI_API_KEYS` | SerpAPI 备用搜索 | 可选 | +| `SOCIAL_SENTIMENT_API_KEY` | Stock Sentiment API Key（Reddit / X / Polymarket，可选） | 可选 | +| `SOCIAL_SENTIMENT_API_URL` | Stock Sentiment API 地址（默认 `https://api.adanos.org`） | 可选 | | `SEARXNG_BASE_URLS` | SearXNG 自建实例（无配额兜底，需在 settings.yml 启用 format: json）；留空时默认自动发现公共实例 | 可选 | | `SEARXNG_PUBLIC_INSTANCES_ENABLED` | 是否在 `SEARXNG_BASE_URLS` 为空时自动从 `searx.space` 获取公共实例（默认 `true`） | 可选 | | `NEWS_STRATEGY_PROFILE` | 新闻策略窗口档位：`ultra_short`(1天)/`short`(3天)/`medium`(7天)/`long`(30天)；实际窗口取与 `NEWS_MAX_AGE_DAYS` 的最小值 | 默认 `short` | | `NEWS_MAX_AGE_DAYS` | 新闻最大时效（天），搜索时限制结果在近期内 | 默认 `3` | | `BIAS_THRESHOLD` | 乖离率阈值（%），超过提示不追高；强势趋势股自动放宽到 1.5 倍 | 默认 `5.0` | +> 行为说明：搜索服务与社交舆情服务为可选增强链路。任一服务初始化失败时，系统会记录 warning 并降级为跳过该服务，仅影响对应环节，不会阻塞技术面主链路和主任务流。 + ### 数据源配置 | 变量名 | 说明 | 默认值 | 必填 | @@ -982,6 +986,7 @@ FastAPI 提供 RESTful API 服务，支持配置管理和触发分析。 > 说明：`POST /api/v1/analysis/analyze` 在 `async_mode=false` 时仅支持单只股票；批量 `stock_codes` 需使用 `async_mode=true`。异步 `202` 响应对单股返回 `task_id`，对批量返回 `accepted` / `duplicates` 汇总结构。 > 进度流说明：`GET /api/v1/analysis/tasks/stream` 除 `task_created / task_started / task_completed / task_failed` 外，新增 `task_progress` 事件。普通分析链路会在“行情准备 / 新闻检索 / 上下文整理 / LLM 生成 / 报告保存”等阶段持续更新 `progress` 与 `message`。LiteLLM 流式返回仅在服务端累积完整文本，最终 JSON 解析成功后才会持久化历史报告；若流式在首个 chunk 前不可用，会自动回退到原非流式调用；若已产生部分 chunk 后失败，系统先尝试同模型非流式重试，失败后再按既有主模型->备用模型顺序继续尝试。 +> 如果任务进度回调异常，主链路不会中断，系统会提升告警为 warning 级别并在服务端日志中输出完整异常，便于排查 SSE 推送断点。 > > 说明：该特性属于运行时 SSE 与回退链路细节，优先记录于完整指南（`full-guide*.md`），不在 `README.md` 中展开详细行为分支。 diff --git a/docs/full-guide_EN.md b/docs/full-guide_EN.md index 4d41194f5..6f165fe36 100644 --- a/docs/full-guide_EN.md +++ b/docs/full-guide_EN.md @@ -238,9 +238,13 @@ Default schedule: Every weekday at **18:00 (Beijing Time)** automatic execution. | `BOCHA_API_KEYS` | Bocha Search API Key (Chinese optimized) | Optional | | `BRAVE_API_KEYS` | Brave Search API Key (US stocks optimized) | Optional | | `SERPAPI_API_KEYS` | SerpAPI Backup search | Optional | +| `SOCIAL_SENTIMENT_API_KEY` | Stock Sentiment API Key (Reddit / X / Polymarket, US stocks optional) | Optional | +| `SOCIAL_SENTIMENT_API_URL` | Stock Sentiment API endpoint (default `https://api.adanos.org`) | Optional | | `SEARXNG_BASE_URLS` | SearXNG self-hosted instances (quota-free fallback, enable format: json in settings.yml); when empty the app auto-discovers public instances | Optional | | `SEARXNG_PUBLIC_INSTANCES_ENABLED` | Auto-discover public SearXNG instances from `searx.space` when `SEARXNG_BASE_URLS` is empty (default `true`) | Optional | +> Behavior note: Search and social sentiment are optional enhancement services. If either service fails to initialize, the system logs a warning and degrades gracefully by skipping that stage without blocking the core analysis flow. + ### Data Source Configuration | Variable | Description | Default | Required | @@ -816,6 +820,7 @@ FastAPI provides RESTful API service for configuration management and triggering > Note: `POST /api/v1/analysis/analyze` supports only one stock when `async_mode=false`; batch `stock_codes` requires `async_mode=true`. The async `202` response returns a single `task_id` for one stock, or an `accepted` / `duplicates` summary for batch requests. > Progress-stream note: `GET /api/v1/analysis/tasks/stream` now emits `task_progress` in addition to `task_created / task_started / task_completed / task_failed`. The regular analysis path updates `progress` and `message` across quote preparation, news retrieval, context assembly, LLM generation, and report persistence. Streaming chunks are accumulated only on the server side; history is persisted only after the final JSON parses successfully. If streaming is unavailable before the first chunk, the system falls back to the previous non-stream request. If a stream fails after partial output has already arrived, the system first retries non-stream for the same model, then continues through existing fallback models in the original order (primary + fallback list). +> If a progress callback fails, the analysis flow continues, and the exception is now logged at warning level to help troubleshoot SSE delivery gaps. > Note: This behavior is documented in the full guide (`full-guide*.md`) because it is detailed runtime SSE/fallback behavior and is therefore kept out of the README. diff --git a/src/core/pipeline.py b/src/core/pipeline.py index 43cd731ac..9c9e25891 100644 --- a/src/core/pipeline.py +++ b/src/core/pipeline.py @@ -102,18 +102,22 @@ class StockAnalysisPipeline: self.notifier = NotificationService(source_message=source_message) self._single_stock_notify_lock = threading.Lock() - # 初始化搜索服务 - self.search_service = SearchService( - bocha_keys=self.config.bocha_api_keys, - tavily_keys=self.config.tavily_api_keys, - brave_keys=self.config.brave_api_keys, - serpapi_keys=self.config.serpapi_keys, - minimax_keys=self.config.minimax_api_keys, - searxng_base_urls=self.config.searxng_base_urls, - searxng_public_instances_enabled=self.config.searxng_public_instances_enabled, - news_max_age_days=self.config.news_max_age_days, - news_strategy_profile=getattr(self.config, "news_strategy_profile", "short"), - ) + # 初始化搜索服务（可选，初始化失败不应阻断主分析流程） + try: + self.search_service = SearchService( + bocha_keys=self.config.bocha_api_keys, + tavily_keys=self.config.tavily_api_keys, + brave_keys=self.config.brave_api_keys, + serpapi_keys=self.config.serpapi_keys, + minimax_keys=self.config.minimax_api_keys, + searxng_base_urls=self.config.searxng_base_urls, + searxng_public_instances_enabled=self.config.searxng_public_instances_enabled, + news_max_age_days=self.config.news_max_age_days, + news_strategy_profile=getattr(self.config, "news_strategy_profile", "short"), + ) + except Exception as exc: + logger.warning("搜索服务初始化失败，将以无搜索模式运行: %s", exc, exc_info=True) + self.search_service = None logger.info(f"调度器初始化完成，最大并发数: {self.max_workers}") logger.info("已启用技术分析引擎（均线/趋势/量价指标）") @@ -126,18 +130,28 @@ class StockAnalysisPipeline: logger.info("筹码分布分析已启用") else: logger.info("筹码分布分析已禁用") - if self.search_service.is_available: + if self.search_service is None: + logger.warning("搜索服务未启用（初始化失败或依赖缺失）") + elif self.search_service.is_available: logger.info("搜索服务已启用") else: logger.warning("搜索服务未启用（未配置搜索能力）") - # 初始化社交舆情服务（仅美股） - self.social_sentiment_service = SocialSentimentService( - api_key=self.config.social_sentiment_api_key, - api_url=self.config.social_sentiment_api_url, - ) - if self.social_sentiment_service.is_available: - logger.info("Social sentiment service enabled (Reddit/X/Polymarket, US stocks only)") + # 初始化社交舆情服务（仅美股，可选） + try: + self.social_sentiment_service = SocialSentimentService( + api_key=self.config.social_sentiment_api_key, + api_url=self.config.social_sentiment_api_url, + ) + if self.social_sentiment_service.is_available: + logger.info("Social sentiment service enabled (Reddit/X/Polymarket, US stocks only)") + except Exception as exc: + logger.warning( + "社交舆情服务初始化失败，将跳过舆情分析: %s", + exc, + exc_info=True, + ) + self.social_sentiment_service = None def _emit_progress(self, progress: int, message: str) -> None: """Best-effort bridge from pipeline stages to task SSE progress.""" @@ -147,7 +161,19 @@ class StockAnalysisPipeline: try: callback(progress, message) except Exception as exc: - logger.debug("[pipeline] progress callback skipped: %s", exc) + query_id = getattr(self, "query_id", None) + logger.warning( + "[pipeline] progress callback failed: %s (progress=%s, message=%r, query_id=%s)", + exc, + progress, + message, + query_id, + extra={ + "progress": progress, + "progress_message": message, + "query_id": query_id, + }, + ) def fetch_and_save_stock_data( self, @@ -349,7 +375,7 @@ class StockAnalysisPipeline: # Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期） news_context = None self._emit_progress(46, f"{stock_name}：正在检索新闻与舆情") - if self.search_service.is_available: + if self.search_service is not None and self.search_service.is_available: logger.info(f"{stock_name}({code}) 开始多维度情报搜索...") # 使用多维度搜索（最多5次搜索） @@ -387,7 +413,7 @@ class StockAnalysisPipeline: logger.info(f"{stock_name}({code}) 搜索服务不可用，跳过情报搜索") # Step 4.5: Social sentiment intelligence (US stocks only) - if self.social_sentiment_service.is_available and is_us_stock_code(code): + if self.social_sentiment_service is not None and self.social_sentiment_service.is_available and is_us_stock_code(code): try: social_context = self.social_sentiment_service.get_social_context(code) if social_context: @@ -749,7 +775,7 @@ class StockAnalysisPipeline: # Agent path: inject social sentiment as news_context so both # executor (_build_user_message) and orchestrator (ctx.set_data) # can consume it through the existing news_context channel - if self.social_sentiment_service.is_available and is_us_stock_code(code): + if self.social_sentiment_service is not None and self.social_sentiment_service.is_available and is_us_stock_code(code): try: social_context = self.social_sentiment_service.get_social_context(code) if social_context: @@ -796,7 +822,7 @@ class StockAnalysisPipeline: # 保存新闻情报到数据库（Agent 工具结果仅用于 LLM 上下文，未持久化，Fixes #396） # 使用 search_stock_news（与 Agent 工具调用逻辑一致），仅 1 次 API 调用，无额外延迟 - if self.search_service.is_available: + if self.search_service is not None and self.search_service.is_available: try: news_response = self.search_service.search_stock_news( stock_code=code, diff --git a/tests/test_pipeline_optional_service_resilience.py b/tests/test_pipeline_optional_service_resilience.py new file mode 100644 index 000000000..7dac73b56 --- /dev/null +++ b/tests/test_pipeline_optional_service_resilience.py @@ -0,0 +1,101 @@ +# -*- coding: utf-8 -*- +"""Regression tests for optional pipeline service degradation logs.""" + +import logging +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from src.core.pipeline import StockAnalysisPipeline + + +def _make_config() -> SimpleNamespace: + return SimpleNamespace( + max_workers=2, + save_context_snapshot=False, + bocha_api_keys=[], + tavily_api_keys=[], + brave_api_keys=[], + serpapi_keys=[], + minimax_api_keys=[], + searxng_base_urls=[], + searxng_public_instances_enabled=False, + news_max_age_days=7, + news_strategy_profile="short", + enable_realtime_quote=False, + realtime_source_priority=[], + enable_chip_distribution=False, + social_sentiment_api_key="", + social_sentiment_api_url="https://example.invalid/social", + ) + + +def _build_pipeline(config: SimpleNamespace) -> StockAnalysisPipeline: + with patch("src.core.pipeline.get_db", return_value=MagicMock()), \ + patch("src.core.pipeline.DataFetcherManager", return_value=MagicMock()), \ + patch("src.core.pipeline.StockTrendAnalyzer", return_value=MagicMock()), \ + patch("src.core.pipeline.GeminiAnalyzer", return_value=MagicMock()), \ + patch("src.core.pipeline.NotificationService", return_value=MagicMock()): + return StockAnalysisPipeline(config=config) + + +def test_search_service_init_failure_logs_traceback_and_failure_state(caplog): + config = _make_config() + social_service = MagicMock() + social_service.is_available = False + + with patch("src.core.pipeline.SearchService", side_effect=RuntimeError("search init boom")), \ + patch("src.core.pipeline.SocialSentimentService", return_value=social_service), \ + caplog.at_level(logging.WARNING, logger="src.core.pipeline"): + pipeline = _build_pipeline(config) + + assert pipeline.search_service is None + + init_failure_records = [ + record for record in caplog.records if "搜索服务初始化失败，将以无搜索模式运行" in record.message + ] + assert len(init_failure_records) == 1 + assert init_failure_records[0].exc_info is not None + assert "搜索服务未启用（初始化失败或依赖缺失）" in caplog.text + assert "搜索服务未启用（未配置搜索能力）" not in caplog.text + + +def test_social_sentiment_init_failure_logs_traceback(caplog): + config = _make_config() + search_service = MagicMock() + search_service.is_available = False + + with patch("src.core.pipeline.SearchService", return_value=search_service), \ + patch("src.core.pipeline.SocialSentimentService", side_effect=RuntimeError("social init boom")), \ + caplog.at_level(logging.WARNING, logger="src.core.pipeline"): + pipeline = _build_pipeline(config) + + assert pipeline.social_sentiment_service is None + + init_failure_records = [ + record for record in caplog.records if "社交舆情服务初始化失败，将跳过舆情分析" in record.message + ] + assert len(init_failure_records) == 1 + assert init_failure_records[0].exc_info is not None + + +def test_emit_progress_logs_context_when_callback_fails(caplog): + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.query_id = "query-123" + + def _fail_callback(progress, message): + raise RuntimeError(f"cannot send {progress}:{message}") + + pipeline.progress_callback = _fail_callback + + with caplog.at_level(logging.WARNING, logger="src.core.pipeline"): + pipeline._emit_progress(55, "fetching news") + + records = [record for record in caplog.records if "progress callback failed" in record.message] + assert len(records) == 1 + record = records[0] + assert "progress=55" in record.message + assert "message='fetching news'" in record.message + assert "query_id=query-123" in record.message + assert record.progress == 55 + assert record.progress_message == "fetching news" + assert record.query_id == "query-123"