github/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis.git synced 2026-10-06 15:23:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add #1391 Phase 2 run diagnostic summaries (#1444)

Browse Source 
* fix(issue-1412): [bug]-stock_list格式问题

* fix(review-feedback-1413): preserve exchange hint for dotted A-share inputs

* fix(review-feedback-1413): Keep normalized A-share codes usable by market routing and preserve

* fix(review-feedback-1413): Limit raw dotted codes to fetchers that can parse them

* fix(review-feedback-1413): Keep Tushare daily input normalized for ETF detection

* fix(review-feedback-1413): 澄清结构化检测中的外部模型/API 与运行时配置迁移风险

* fix(review-feedback-1413): 处理或明确确认该失败与本 PR 无关且已有维护者豁免依据

* fix: keep stock list input as bare codes

* docs: add phase-0 run diagnostics contract

* fix(review-feedback-1435): 修正描述并澄清/补齐运行时代码变更的验证证据

* fix(review-feedback-1435): 补齐前缀提示识别，并增加对应回归测试

* fix(review-feedback-1435): 修正

* fix(review-feedback-1435): 解决冲突并更新描述/验证记录后再合入

* fix(review-feedback-1435): 解决冲突，并在最终 head 上重新确认 python -m pytest tests/test a share fetcher code

* fix(review-feedback-1435): 修复并补齐回归覆盖后再复核最终 head

* fix(review-feedback-1435): data provider/baostock fetcher.py 的 convert stock code 只从 .SH/.SS/.SZ

* fix: preserve A-share exchange hints

* fix(review-feedback-1435): 修正 docs/run-diagnostics-p0.md 对 Tushare 本轮范围的矛盾描述

* feat: add phase 1 run diagnostics trace plumbing

* feat: add phase 2 run diagnostic summaries

* fix(review-feedback-1441): 打通 trace id 与数据源运行快照，改动目标明确

* fix(review-feedback-1444): 修复 Agent 模式新报告通过历史诊断 API 返回 unknown 的正确性问题

* fix(review-feedback-1444): Propagate diagnostics lookup errors instead of masking them

* fix(review-feedback-1444): 补对应回归断言

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突

* fix(review-feedback-1444): 解决冲突

* fix(review-feedback-1444): 解决

* fix(review-feedback-1444): 解决

* fix(review-feedback-1444): 解决冲突

* fix(review-feedback-1444): 解决

* fix(review-feedback-1444): 解决

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突

* fix(review-feedback-1444): Derive news diagnostics from retrieval evidence

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突并基于解决后的最终 diff 重新确认 docs/CHANGELOG.md、诊断链路和测试结果

* fix(review-feedback-1444): 解决冲突

* fix(review-feedback-1444): 解决冲突后再合入

* fix(review-feedback-1444): 解决冲突并重新跑阻断型 CI，尤其是 backend-gate 和相关诊断/API/history 回归

* fix(review-feedback-1444): preserve report timestamp when enriching task results

* fix: address run diagnostics review feedback

* fix: redact diagnostic copy text secrets

* fix(review-feedback-1444): 补一条多渠道部分失败的回归测试
This commit is contained in:
mumu
2026-05-26 00:05:24 +08:00
committed by GitHub
parent 4610dcbb95
commit 8a23124d01
17 changed files with 2021 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/v1/endpoints/analysis.py
+25 -1
View File
@@ -70,6 +70,7 @@ from src.services.task_queue import (
DuplicateTaskError,
TaskStatus as TaskStatusEnum,
)
from src.services.run_diagnostics import build_run_diagnostic_summary
from src.utils.data_processing import (
normalize_model_used,
parse_json_field,
@@ -465,6 +466,7 @@ def _handle_sync_analysis(
stock_code=result.get("stock_code", stock_code),
stock_name=result.get("stock_name"),
report=report.model_dump() if report else None,
diagnostic_summary=result.get("diagnostic_summary"),
created_at=datetime.now().isoformat()
)
@@ -728,6 +730,18 @@ def _extract_report_created_at(payload: Dict[str, Any]) -> Optional[str]:
return _datetime_to_iso(meta.get("created_at"))
def _prepare_report_for_task_enrichment(
report_data: Dict[str, Any],
created_at: Optional[str],
) -> Dict[str, Any]:
enriched_report = dict(report_data)
meta = dict(enriched_report.get("meta") or {})
if created_at and not _datetime_to_iso(meta.get("created_at")):
meta["created_at"] = created_at
enriched_report["meta"] = meta
return enriched_report
def _build_task_analysis_result(task: Any) -> AnalysisResultResponse:
"""
Normalize an in-memory completed task result to the public API contract.
@@ -766,7 +780,10 @@ def _build_task_analysis_result(task: Any) -> AnalysisResultResponse:
if context_snapshot is not None or fundamental_snapshot is not None:
try:
report = _build_analysis_report(
report_data,
_prepare_report_for_task_enrichment(
report_data,
payload.get("created_at"),
),
query_id,
stock_code,
payload.get("stock_name") or getattr(task, "stock_name", None),
@@ -960,6 +977,13 @@ def get_analysis_status(task_id: str) -> TaskStatus:
stock_code=record.code,
stock_name=stock_name,
report=report_dict,
diagnostic_summary=build_run_diagnostic_summary(
context_snapshot=context_snapshot,
raw_result=raw_result,
report_saved=True,
query_id=task_id,
stock_code=record.code,
),
created_at=record.created_at.isoformat() if record.created_at else datetime.now().isoformat()
),
error=None,
api/v1/endpoints/history.py
+44
View File
@@ -28,6 +28,7 @@ from api.v1.schemas.history import (
ReportStrategy,
ReportDetails,
MarkdownReportResponse,
RunDiagnosticSummaryResponse,
)
from api.v1.schemas.common import ErrorResponse
from src.storage import DatabaseManager
@@ -329,6 +330,49 @@ def get_history_detail(
)
@router.get(
"/{record_id}/diagnostics",
response_model=RunDiagnosticSummaryResponse,
responses={
200: {"description": "运行诊断摘要"},
404: {"description": "报告不存在", "model": ErrorResponse},
500: {"description": "服务器错误", "model": ErrorResponse},
},
summary="获取历史报告运行诊断摘要",
description="根据分析历史记录 ID 或 query_id 获取用户可读诊断摘要和脱敏复制文本。",
)
def get_history_diagnostics(
record_id: str,
db_manager: DatabaseManager = Depends(get_database_manager),
) -> RunDiagnosticSummaryResponse:
"""
获取历史报告运行诊断摘要。
"""
try:
service = HistoryService(db_manager)
summary = service.resolve_and_get_diagnostics(record_id)
if summary is None:
raise HTTPException(
status_code=404,
detail={
"error": "not_found",
"message": f"未找到 id/query_id={record_id} 的分析记录",
},
)
return RunDiagnosticSummaryResponse.model_validate(summary)
except HTTPException:
raise
except Exception as e:
logger.error(f"查询运行诊断摘要失败: {e}", exc_info=True)
raise HTTPException(
status_code=500,
detail={
"error": "internal_error",
"message": f"查询运行诊断摘要失败: {str(e)}",
},
)
@router.get(
"/{record_id}/news",
response_model=NewsIntelResponse,
api/v1/schemas/analysis.py
+1
View File
@@ -126,6 +126,7 @@ class AnalysisResultResponse(BaseModel):
stock_code: str = Field(..., description="股票代码")
stock_name: Optional[str] = Field(None, description="股票名称")
report: Optional[Any] = Field(None, description="分析报告")
diagnostic_summary: Optional[Any] = Field(None, description="运行诊断摘要")
created_at: str = Field(..., description="创建时间")
model_config = ConfigDict(json_schema_extra={
api/v1/schemas/history.py
+39 -1
View File
@@ -9,7 +9,7 @@
2. 定义分析报告完整模型
"""
from typing import Optional, List, Any
from typing import Optional, List, Any, Dict
from pydantic import BaseModel, ConfigDict, Field
@@ -200,3 +200,41 @@ class MarkdownReportResponse(BaseModel):
"content": "# 📊 贵州茅台 (600519) 分析报告\n\n> 分析日期：**2024-01-01**\n\n..."
}
})
class RunDiagnosticComponent(BaseModel):
"""单个运行诊断组件摘要。"""
key: str = Field(..., description="组件键")
label: str = Field(..., description="组件显示名称")
status: str = Field(..., description="组件状态：ok/degraded/failed/unknown/not_configured/skipped")
message: str = Field(..., description="用户可读摘要")
details: Optional[Dict[str, Any]] = Field(None, description="折叠展示的诊断细节")
class RunDiagnosticSummaryResponse(BaseModel):
"""历史报告运行诊断摘要。"""
trace_id: Optional[str] = Field(None, description="诊断 trace ID")
task_id: Optional[str] = Field(None, description="任务 ID")
query_id: Optional[str] = Field(None, description="分析 query ID")
stock_code: Optional[str] = Field(None, description="股票代码")
trigger_source: Optional[str] = Field(None, description="触发来源")
status: str = Field(..., description="总体状态：normal/degraded/failed/unknown")
status_label: str = Field(..., description="总体状态中文标签")
reason: str = Field(..., description="最主要的诊断原因")
components: Dict[str, RunDiagnosticComponent] = Field(default_factory=dict, description="关键链路诊断组件")
copy_text: str = Field(..., description="可复制的脱敏排障文本")
model_config = ConfigDict(json_schema_extra={
"example": {
"trace_id": "task_abc123",
"query_id": "task_abc123",
"stock_code": "600519",
"status": "degraded",
"status_label": "部分降级",
"reason": "实时行情失败：timeout",
"components": {},
"copy_text": "trace_id: task_abc123\nstock_code: 600519\n...",
}
})
docs/CHANGELOG.md
+2
View File
@@ -29,6 +29,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
- [修复] macOS 桌面端将运行时配置迁移到用户数据目录，并在旧 `.app` 包内文件仍可访问时迁移 `.env`、数据库和日志，避免后续替换升级后重新配置。
- [改进] 新增运行态市场阶段上下文构造与降级测试。
- [文档] 新增 AnalysisContextPack P0 上下文盘点，明确字段质量状态、现有状态映射和首版 pack 边界。
- [新功能] 落地 #1391 Phase 2 运行诊断摘要：生成用户可读 RunDiagnosticSummary，提供历史报告诊断 API 与脱敏复制文本。
- [文档] 明确 #1391 Phase 2 的结构化检测告警为非配置迁移信号：`agent_max_steps`/`agent_orchestrator_timeout_s` 非法值会 fallback 至默认并产生日志告警，新增诊断链路仅新增 `context_snapshot`/`RunDiagnosticSummary` 读写字段，不改写 `litellm_model`、`agent_litellm_model`、`openai_base_url`、LLM channel 路由或配置迁移语义。
- [新功能] 新增 AnalysisContextPack P1 内部契约与脱敏序列化测试。
- [修复] 恢复 Agent/历史兼容快照中的关联板块与板块联动字段提取，修复新版首页报告缺少“板块联动”的回归问题。
- [修复] 修复 Web 中文设置页在数据源、通知、系统与 Agent 区域的配置标题、说明和关键下拉选项漏翻问题。
docs/full-guide.md
+4 -1
View File
@@ -211,7 +211,8 @@ daily_stock_analysis/
### AI 模型配置
> 完整说明见 [LLM 配置指南](LLM_CONFIG_GUIDE.md)（三层配置、渠道模式、Vision、Agent、排错）；常用服务商预设、Actions 变量对照和错误排障见 [LLM 服务商配置指南](llm-providers.md)。
> 兼容性说明（Issue #1306）：本次改动只复用已有历史写入链路展示大盘复盘结果，不修改模型名、provider、Base URL、`LiteLLM` 清理/兼容语义。回退路径为回滚本版本。兼容验证来源见 `requirements.txt`（`litellm` 版本约束）、`docs/LLM_CONFIG_GUIDE*.md`，以及回归用例 `tests/test_analysis_api_contract.py`、`tests/test_analysis_history.py`、`tests/test_market_review.py`；官方源参考：[LiteLLM OpenAI-compatible](https://docs.litellm.ai/docs/providers/openai_compatible)、[OpenAI Chat Completion API](https://platform.openai.com/docs/api-reference/chat)。
> 兼容性说明（Issue #1306/#1391）：本次改动只复用已有历史写入链路展示大盘复盘结果，不修改模型名、provider、Base URL、`LiteLLM` 清理/兼容语义。回退路径为回滚本版本。兼容验证来源见 `requirements.txt`（`litellm` 版本约束）、`docs/LLM_CONFIG_GUIDE*.md`，以及回归用例 `tests/test_analysis_api_contract.py`、`tests/test_analysis_history.py`、`tests/test_market_review.py`；官方源参考：[LiteLLM OpenAI-compatible](https://docs.litellm.ai/docs/providers/openai_compatible)、[OpenAI Chat Completion API](https://platform.openai.com/docs/api-reference/chat)。
> #1391 Phase 2 的结构化检测风险来自 `src/agent/factory.py` 的 `agent_max_steps` / `agent_orchestrator_timeout_s` int 安全兜底，属于配置读取侧的类型兼容增强，不会改写 `litellm_model`、`agent_litellm_model`、`openai_base_url` 或 `LLM_*` 路由状态；回归可复核 `tests/test_agent_pipeline.py::TestAgentConfig::test_build_agent_executor_does_not_mutate_llm_route_config` 与 `tests/test_agent_pipeline.py::TestAgentConfig::test_build_agent_executor_multi_arch_does_not_mutate_llm_route_config`。当配置值非法（如非数字）时，`src.agent.factory` 会记录 warning 并回退到默认值，便于排障与避免误判配置已生效。
> 本节仅同步模型/渠道配置清单，不额外引入新的外部 provider / Base URL 兼容约定；兼容语义以当前仓库 `requirements.txt` 依赖约束和相关测试为准，历史回退路径见上述两份文档中“回退/恢复”说明。
| 变量名 | 说明 | 默认值 | 必填 |
@@ -1232,6 +1233,7 @@ FastAPI 提供 RESTful API 服务，支持配置管理和触发分析。
| `/api/v1/analysis/tasks/stream` | GET (SSE) | 订阅任务实时状态流 |
| `/api/v1/analysis/status/{task_id}` | GET | 查询任务状态 |
| `/api/v1/history` | GET | 查询分析历史 |
| `/api/v1/history/{record_id}/diagnostics` | GET | 查询历史报告运行诊断摘要与脱敏复制文本 |
| `/api/v1/usage/summary?period=today|month|all` | GET | 按调用类型与模型维度汇总 LLM 调用次数和 Token 用量 |
| `/api/v1/backtest/run` | POST | 触发回测 |
| `/api/v1/backtest/results` | GET | 查询回测结果（分页） |
@@ -1249,6 +1251,7 @@ FastAPI 提供 RESTful API 服务，支持配置管理和触发分析。
> 审计依据：优先级与回退语义以 `src/config.py` 的 `Config._load_from_env()` 为准（`LITELLM_CONFIG` > `LLM_CHANNELS` > legacy）。配套回归见 `tests/test_llm_channel_config.py`（配置源解析）与 `tests/test_market_review_runtime.py`（共享装配路径）。该接口当前仅提供单进程/单机级防重复能力，若为多实例部署需通过外部任务队列或分布式锁补齐全局幂等。
> 说明：`POST /api/v1/analysis/market-review` 触发后，报告会以 `report_type=market_review` 写入历史库；你可直接查询 `/api/v1/history` 或 `/api/v1/history/{record_id}` 获取历史 Markdown，避免再次触发分析重算。
> 说明：该端点若返回 `task_id`，WebUI 会轮询 `GET /api/v1/analysis/status/{task_id}` 展示状态。状态为 `completed` 时给出完成提示（报告已生成并按配置推送），状态为 `failed` 时在前端错误区域显示 `error` 原因。
> 说明：`GET /api/v1/history/{record_id}/diagnostics` 支持历史记录主键 ID 或 `query_id`，返回 `normal/degraded/failed/unknown` 摘要、关键链路组件和可复制的脱敏 `copy_text`；旧报告缺少诊断快照时返回 `unknown`，不影响报告读取。
> 兼容性审计证据：
> - 官方来源：LiteLLM OpenAI-compatible provider 文档 <https://docs.litellm.ai/docs/providers/openai_compatible>；OpenAI Chat API 文档 <https://platform.openai.com/docs/api-reference/chat/create>；DeepSeek API 文档 <https://api-docs.deepseek.com/>。
docs/full-guide_EN.md
+2
View File
@@ -1072,6 +1072,7 @@ FastAPI provides RESTful API service for configuration management and triggering
| `/api/v1/analysis/tasks/stream` | GET (SSE) | Subscribe to realtime task updates |
| `/api/v1/analysis/status/{task_id}` | GET | Query task status |
| `/api/v1/history` | GET | Query analysis history |
| `/api/v1/history/{record_id}/diagnostics` | GET | Query a historical report run diagnostic summary and sanitized copy text |
| `/api/v1/usage/summary?period=today|month|all` | GET | Query LLM call counts and token usage grouped by call type and model |
| `/api/v1/backtest/run` | POST | Trigger backtest |
| `/api/v1/backtest/results` | GET | Query backtest results (paginated) |
@@ -1087,6 +1088,7 @@ FastAPI provides RESTful API service for configuration management and triggering
> Audit note: priority and fallback are defined by `Config._load_from_env()` in `src/config.py` (`LITELLM_CONFIG` > `LLM_CHANNELS` > legacy). Regression coverage is in `tests/test_llm_channel_config.py` (configuration source parsing) and `tests/test_market_review_runtime.py` (shared runtime assembly). The endpoint lock is process/host-level only; multi-instance deployments still need external distributed idempotency controls.
> Note: Once `/api/v1/analysis/market-review` completes, the report is persisted with `report_type=market_review`; open `/api/v1/history` and `/api/v1/history/{record_id}` (or Markdown history endpoints) to view it directly without re-running analysis.
> Note: when `/api/v1/analysis/market-review` returns a `task_id`, the WebUI polls `GET /api/v1/analysis/status/{task_id}`. The UI renders clear `pending/processing` progress, shows completion feedback when status becomes `completed`, and surfaces `error` content on `failed`.
> Note: `GET /api/v1/history/{record_id}/diagnostics` accepts either the history primary key ID or `query_id`, and returns a `normal/degraded/failed/unknown` summary, key pipeline components, and sanitized `copy_text`. Older reports without `context_snapshot.diagnostics` return `unknown` without affecting normal report reads.
> Compatibility audit evidence:
> - Official references: LiteLLM OpenAI-compatible provider documentation <https://docs.litellm.ai/docs/providers/openai_compatible>, OpenAI Chat API <https://platform.openai.com/docs/api-reference/chat/create>, and DeepSeek API docs <https://api-docs.deepseek.com/>.
docs/run-diagnostics-p2.md
+67
View File
@@ -0,0 +1,67 @@
# 运行诊断与数据可靠性 1.0（Phase 2）
本文档记录 #1391 Phase 2 的后端落地范围：基于 Phase 1 的 `trace_id` 与 provider run 记录，生成用户可读的运行诊断摘要，并提供可复制的脱敏排障文本。
## 本轮范围
- 新增 `RunDiagnosticSummary` 聚合逻辑，输出总体状态：
- `normal` / 正常
- `degraded` / 部分降级
- `failed` / 失败
- `unknown` / 未知
- 摘要覆盖以下关键链路：
- 实时行情
- 日线数据
- 新闻搜索
- LLM
- 通知
- 历史保存
- `AnalysisService` 同步/异步任务结果追加可选 `diagnostic_summary`。
- 新增历史报告诊断 API：
```http
GET /api/v1/history/{record_id}/diagnostics
```
`record_id` 支持历史记录主键 ID 或 `query_id`，返回诊断摘要与 `copy_text`。
## 复制排障信息
`copy_text` 是面向 issue/排障的纯文本，包含：
- `trace_id`
- `query_id`
- `stock_code`
- `trigger_source`
- 总体 `data_status`
- 实时行情、日线、新闻、LLM、通知、历史保存的简短状态
- 首要原因
生成前会复用运行诊断脱敏规则，避免输出 token、API key、Authorization、Cookie、webhook URL、邮箱密码、代理凭据等敏感信息。
## 兼容性边界
- 本轮不新增配置项，不改变数据源优先级，不改变 fallback 策略。
- 本轮不改变任何 LLM/provider/Base URL/配置迁移语义，仅新增历史快照中的诊断字段与查询接口。
- API 只追加可选字段和新增只读接口；旧客户端可忽略。
- 旧报告没有 `context_snapshot.diagnostics` 时返回 `unknown`，不报错。
- 通知诊断在当前任务上下文中记录；历史报告如果保存时尚无通知证据，会在摘要中显示通知结果未知。
- 诊断摘要生成失败不得影响报告读取或分析主流程。
### 结构化检测告警澄清
- 自动化检测命中的“模型/provider/base URL 兼容风险”来源是：`src/agent/factory.py` 新增了 `agent_max_steps` 与 `agent_orchestrator_timeout_s` 的 **数字安全兜底**（`_coerce_config_int`），因此扫描可能将其误识别为配置敏感路径；该命中属于测试与路由保护触发，不是运行时配置或兼容语义变更。
- 当数值配置存在非法值时，系统会记录 `warning` 到 `src.agent.factory` 日志（示例：`[AgentFactory] Invalid value for agent_max_steps...`），并回退到默认值；日志用于定位“参数未生效”类问题，与模型/provider/base URL 兼容性独立。
- 本轮确认无静默迁移/清空/改写：
- `src/core/pipeline.py` 与 `src/services/analysis_service.py` 仅新增诊断记录，不修改 `Config` 中任何 `litellm_model`、`agent_litellm_model`、`openai_base_url` 或 channel `LLM_*` 字段。
- `src/agent/factory.py` 的 `_coerce_config_int` 只在构建执行参数时计算 `max_steps` 与 `timeout_seconds`，并且不写回到 `config` 对象；`litellm_model`、`agent_litellm_model`、`openai_base_url` 原值在构造链路中完整透传。
- 本轮不触发 `Config` 的运行时清理、持久化回写或迁移流程，因此不存在写回导致运行时配置被重写的风险。
- 回归验证：`tests/test_agent_pipeline.py::TestAgentConfig::test_build_agent_executor_does_not_mutate_llm_route_config` 与 `tests/test_agent_pipeline.py::TestAgentConfig::test_build_agent_executor_multi_arch_does_not_mutate_llm_route_config` 明确断言上述字段在 `build_agent_executor` 后保持原值。
- 回退路径：如需恢复到旧行为，移除本轮相关提交；或将 `diag_*` 字段从 `context_snapshot`/`RunDiagnosticSummary` 的反序列化链路中移除。主链路与模型/provider 配置无需额外迁移或修复。
## 验证建议
```bash
python -m pytest tests/test_run_diagnostics_p2.py tests/test_run_diagnostics_p1.py
python -m py_compile src/services/run_diagnostics.py src/services/history_service.py api/v1/endpoints/history.py api/v1/schemas/history.py
```
src/agent/factory.py
+42 -3
View File
@@ -59,6 +59,29 @@ class SkillPromptState:
technical_skill_policy: str
def _coerce_config_int(raw_value: object, default: int, *, field_name: str | None = None) -> int:
"""Coerce optional numeric config values to int with a fallback default.
This protects test doubles and incomplete config objects from propagating
mock-like values (e.g., MagicMock attributes) into strict numeric paths.
This function is side-effect free: it only returns a parsed int fallback value
and intentionally never mutates source config attributes.
"""
try:
return int(raw_value)
except (TypeError, ValueError, OverflowError):
if field_name:
logger.warning(
"[AgentFactory] Invalid value for %s: %r, fallback to default %s",
field_name,
raw_value,
default,
)
return default
def _normalize_skill_ids(
skill_ids: Optional[List[str]],
*,
@@ -319,14 +342,26 @@ def build_agent_executor(config=None, skills: Optional[List[str]] = None):
)
from src.agent.executor import AgentExecutor
# Intentionally do not mutate config routing fields here. We only coerce
# execution params (max_steps/timeout_seconds) from config values; provider,
# model, base URL and channel routes stay unchanged and are consumed by
# downstream adapter logic as-is.
return AgentExecutor(
tool_registry=registry,
llm_adapter=llm_adapter,
skill_instructions=prompt_state.skill_instructions,
default_skill_policy=prompt_state.default_skill_policy,
use_legacy_default_prompt=prompt_state.use_legacy_default_prompt,
max_steps=getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
timeout_seconds=getattr(config, "agent_orchestrator_timeout_s", 0),
max_steps=_coerce_config_int(
getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
AGENT_MAX_STEPS_DEFAULT,
field_name="agent_max_steps",
),
timeout_seconds=_coerce_config_int(
getattr(config, "agent_orchestrator_timeout_s", 0),
0,
field_name="agent_orchestrator_timeout_s",
),
)
@@ -346,7 +381,11 @@ def _build_orchestrator(config, registry, llm_adapter, skill_manager, *, technic
llm_adapter=llm_adapter,
skill_instructions=skill_manager.get_skill_instructions(),
technical_skill_policy=technical_skill_policy,
max_steps=getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
max_steps=_coerce_config_int(
getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
AGENT_MAX_STEPS_DEFAULT,
field_name="agent_max_steps",
),
mode=mode,
skill_manager=skill_manager,
config=config,
src/core/pipeline.py
+280 -44
View File
@@ -50,6 +50,9 @@ from src.services.run_diagnostics import (
activate_run_diagnostic_context,
current_diagnostic_snapshot,
get_current_diagnostic_context,
record_history_run,
record_llm_run,
record_notification_run,
reset_run_diagnostic_context,
)
from src.enums import ReportType
@@ -422,6 +425,7 @@ class StockAnalysisPipeline:
# Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期）
news_context = None
news_result_count: Optional[int] = None
self._emit_progress(46, f"{stock_name}：正在检索新闻与舆情")
if self.search_service is not None and self.search_service.is_available:
logger.info(f"{stock_name}({code}) 开始多维度情报搜索...")
@@ -439,6 +443,7 @@ class StockAnalysisPipeline:
total_results = sum(
len(r.results) for r in intel_results.values() if r.success
)
news_result_count = total_results
logger.info(f"{stock_name}({code}) 情报搜索完成: 共 {total_results} 条结果")
logger.debug(f"{stock_name}({code}) 情报搜索结果:\n{news_context}")
@@ -516,12 +521,41 @@ class StockAnalysisPipeline:
)
self._emit_progress(64, f"{stock_name}：正在请求 LLM 生成报告")
result = self.analyzer.analyze(
enhanced_context,
news_context=news_context,
progress_callback=self._emit_progress,
stream_progress_callback=_on_llm_stream,
)
llm_started_at = time.monotonic()
try:
result = self.analyzer.analyze(
enhanced_context,
news_context=news_context,
progress_callback=self._emit_progress,
stream_progress_callback=_on_llm_stream,
)
llm_duration_ms = int((time.monotonic() - llm_started_at) * 1000)
record_llm_run(
success=bool(result and getattr(result, "success", True)),
model=getattr(result, "model_used", None) if result else None,
call_type="analysis",
duration_ms=llm_duration_ms,
error_type=(
None
if result and getattr(result, "success", True)
else "AnalysisResultError"
),
error_message=(
getattr(result, "error_message", None)
if result and not getattr(result, "success", True)
else ("LLM returned empty result" if result is None else None)
),
)
except Exception as exc:
record_llm_run(
success=False,
model=getattr(self.config, "litellm_model", None),
call_type="analysis",
duration_ms=int((time.monotonic() - llm_started_at) * 1000),
error_type=type(exc).__name__,
error_message=exc,
)
raise
# Step 7.5: 填充分析时的价格信息到 result
if result:
@@ -549,10 +583,12 @@ class StockAnalysisPipeline:
context_snapshot = self._build_context_snapshot(
enhanced_context=enhanced_context,
news_content=news_context,
news_result_count=news_result_count,
realtime_quote=realtime_quote,
chip_data=chip_data
)
self.db.save_analysis_history(
result.diagnostic_context_snapshot = context_snapshot
saved_count = self.db.save_analysis_history(
result=result,
query_id=query_id,
report_type=report_type.value,
@@ -560,7 +596,16 @@ class StockAnalysisPipeline:
context_snapshot=context_snapshot,
save_snapshot=self.save_context_snapshot
)
record_history_run(
report_saved=bool(saved_count),
metadata_saved=bool(saved_count),
)
except Exception as e:
record_history_run(
report_saved=False,
metadata_saved=False,
error_message=e,
)
logger.warning(f"{stock_name}({code}) 保存分析历史失败: {e}")
return result
@@ -896,7 +941,19 @@ class StockAnalysisPipeline:
message = f"Analyze stock {code} ({stock_name}) and return the full decision dashboard JSON in English."
else:
message = f"请分析股票 {code} ({stock_name})，并生成决策仪表盘报告。"
agent_result = executor.run(message, context=initial_context)
llm_started_at = time.monotonic()
try:
agent_result = executor.run(message, context=initial_context)
except Exception as exc:
record_llm_run(
success=False,
model=getattr(self.config, "agent_litellm_model", None),
call_type="agent_analysis",
duration_ms=int((time.monotonic() - llm_started_at) * 1000),
error_type=type(exc).__name__,
error_message=exc,
)
raise
# 转换为 AnalysisResult
result = self._agent_result_to_analysis_result(
@@ -907,6 +964,22 @@ class StockAnalysisPipeline:
query_id,
trend_result=trend_result,
)
record_llm_run(
success=bool(result and getattr(result, "success", True)),
model=getattr(result, "model_used", None) if result else getattr(agent_result, "model", None),
call_type="agent_analysis",
duration_ms=int((time.monotonic() - llm_started_at) * 1000),
error_type=(
None
if result and getattr(result, "success", True)
else "AgentResultError"
),
error_message=(
getattr(result, "error_message", None)
if result and not getattr(result, "success", True)
else ("Agent returned empty result" if result is None else None)
),
)
if result:
result.query_id = query_id
# Agent weak integrity: placeholder fill only, no LLM retry
@@ -972,7 +1045,9 @@ class StockAnalysisPipeline:
realtime_quote=realtime_quote,
chip_data=chip_data,
)
self.db.save_analysis_history(
result.diagnostic_context_snapshot = agent_context_snapshot
agent_context_snapshot["stock_name"] = resolved_stock_name
saved_count = self.db.save_analysis_history(
result=result,
query_id=query_id,
report_type=report_type.value,
@@ -980,7 +1055,20 @@ class StockAnalysisPipeline:
context_snapshot=agent_context_snapshot,
save_snapshot=self.save_context_snapshot,
)
record_history_run(
report_saved=bool(saved_count),
metadata_saved=bool(saved_count),
)
latest_diagnostic_snapshot = current_diagnostic_snapshot()
if latest_diagnostic_snapshot is not None:
agent_context_snapshot["diagnostics"] = latest_diagnostic_snapshot
result.diagnostic_context_snapshot = agent_context_snapshot
except Exception as e:
record_history_run(
report_saved=False,
metadata_saved=False,
error_message=e,
)
logger.warning(f"[{code}] 保存 Agent 分析历史失败: {e}")
return result
@@ -1586,7 +1674,8 @@ class StockAnalysisPipeline:
enhanced_context: Dict[str, Any],
news_content: Optional[str],
realtime_quote: Any,
chip_data: Optional[ChipDistribution]
chip_data: Optional[ChipDistribution],
news_result_count: Optional[int] = None,
) -> Dict[str, Any]:
"""
构建分析上下文快照
@@ -1597,6 +1686,10 @@ class StockAnalysisPipeline:
"realtime_quote_raw": self._safe_to_dict(realtime_quote),
"chip_distribution_raw": self._safe_to_dict(chip_data),
}
if news_content is not None:
snapshot["news_retrieval_content"] = news_content
if news_result_count is not None:
snapshot["news_result_count"] = news_result_count
diagnostic_snapshot = current_diagnostic_snapshot()
if diagnostic_snapshot is not None:
snapshot["diagnostics"] = diagnostic_snapshot
@@ -1964,6 +2057,12 @@ class StockAnalysisPipeline:
) -> None:
"""发送单股通知，供直接单股入口和批量串行推送共用。"""
if not self.notifier.is_available():
record_notification_run(
channel="report",
status="not_configured",
success=False,
attempts=0,
)
return
stock_code = getattr(result, "code", None) or fallback_code or "unknown"
@@ -1987,18 +2086,30 @@ class StockAnalysisPipeline:
report_content = self.notifier.generate_single_stock_report(result)
logger.info(f"[{stock_code}] 使用精简报告格式")
if self.notifier.send(
sent = self.notifier.send(
report_content,
email_stock_codes=[stock_code],
route_type="report",
severity="info",
dedup_key=f"report:single:{stock_code}:{report_type.value}",
cooldown_key=f"report:single:{stock_code}:{report_type.value}",
):
)
record_notification_run(
channel="report",
status="success" if sent else "failed",
success=sent,
)
if sent:
logger.info(f"[{stock_code}] 单股推送成功")
else:
logger.warning(f"[{stock_code}] 单股推送失败")
except Exception as e:
record_notification_run(
channel="report",
status="failed",
success=False,
error_message=e,
)
logger.error(f"[{stock_code}] 单股推送异常: {e}")
def _save_local_report(
@@ -2037,13 +2148,19 @@ class StockAnalysisPipeline:
# 跳过推送（单股推送模式 / 合并模式：报告已由 _save_local_report 保存）
if skip_push:
record_notification_run(
channel="report",
status="skipped",
success=False,
attempts=0,
)
return
# 推送通知
if self.notifier.is_available():
channels = self.notifier.get_available_channels()
channels = self.notifier.get_channels_for_route("report", channels=channels)
context_success = self.notifier.send_to_context(report)
send_context = self.notifier.send_to_context(report)
if channels and hasattr(self.notifier, "evaluate_noise_control"):
report_type_key = report_type.value if isinstance(report_type, ReportType) else str(report_type)
codes_key = ",".join(
@@ -2058,6 +2175,12 @@ class StockAnalysisPipeline:
cooldown_key=noise_key,
)
if not noise_decision.should_send:
record_notification_run(
channel="report",
status="skipped",
success=False,
attempts=0,
)
logger.info(noise_decision.message)
return
@@ -2083,16 +2206,31 @@ class StockAnalysisPipeline:
else "wkhtmltopdf (apt install wkhtmltopdf / brew install wkhtmltopdf)"
)
def _send_channel_safely(channel_label: str, send_func: Callable[[], bool]) -> bool:
def _send_channel_safely(
channel_label: str,
send_func: Callable[[], bool],
) -> tuple[bool, Optional[Exception]]:
try:
return bool(send_func())
return bool(send_func()), None
except Exception as e:
logger.exception(
"通知渠道 %s 推送异常，继续尝试其他渠道: %s",
channel_label,
e,
)
return False
return False, e
def _record_channel_result(
channel_label: str,
success: bool,
error_message: Optional[Exception] = None,
) -> None:
record_notification_run(
channel=channel_label,
status="success" if success else "failed",
success=success,
error_message=error_message,
)
image_bytes = None
if non_wechat_channels_needing_image:
@@ -2138,10 +2276,15 @@ class StockAnalysisPipeline:
return self.notifier._send_wechat_image(wechat_image_bytes)
return self.notifier.send_to_wechat(dashboard_content)
wechat_success = _send_channel_safely(
wechat_success, wechat_error = _send_channel_safely(
NotificationChannel.WECHAT.value,
_send_wechat_report,
)
_record_channel_result(
NotificationChannel.WECHAT.value,
wechat_success,
wechat_error,
)
# 其他渠道：发完整报告（避免自定义 Webhook 被 wechat 截断逻辑污染）
non_wechat_success = False
@@ -2150,10 +2293,16 @@ class StockAnalysisPipeline:
if channel == NotificationChannel.WECHAT:
continue
if channel == NotificationChannel.FEISHU:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_feishu(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.TELEGRAM:
def _send_telegram_report() -> bool:
use_image = self.notifier._should_use_image_for_channel(
@@ -2163,10 +2312,16 @@ class StockAnalysisPipeline:
return self.notifier._send_telegram_photo(image_bytes)
return self.notifier.send_to_telegram(report)
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
_send_telegram_report,
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.EMAIL:
if stock_email_groups:
code_to_emails: Dict[str, Optional[List[str]]] = {}
@@ -2212,10 +2367,16 @@ class StockAnalysisPipeline:
f"{channel.value}:{','.join(receivers)}"
if receivers else f"{channel.value}:default"
)
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
email_label,
_send_email_group,
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
email_label,
channel_success,
channel_error,
)
else:
def _send_email_report() -> bool:
use_image = self.notifier._should_use_image_for_channel(
@@ -2225,10 +2386,16 @@ class StockAnalysisPipeline:
return self.notifier._send_email_with_inline_image(image_bytes)
return self.notifier.send_to_email(report)
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
_send_email_report,
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.CUSTOM:
def _send_custom_report() -> bool:
use_image = self.notifier._should_use_image_for_channel(
@@ -2240,45 +2407,93 @@ class StockAnalysisPipeline:
)
return self.notifier.send_to_custom(report)
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
_send_custom_report,
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.PUSHPLUS:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_pushplus(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.SERVERCHAN3:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_serverchan3(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.DISCORD:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_discord(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.PUSHOVER:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_pushover(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.NTFY:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_ntfy(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.GOTIFY:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_gotify(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.ASTRBOT:
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
lambda: self.notifier.send_to_astrbot(report),
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
elif channel == NotificationChannel.SLACK:
def _send_slack_report() -> bool:
use_image = self.notifier._should_use_image_for_channel(
@@ -2290,14 +2505,23 @@ class StockAnalysisPipeline:
)
return self.notifier.send_to_slack(report)
non_wechat_success = _send_channel_safely(
channel_success, channel_error = _send_channel_safely(
channel.value,
_send_slack_report,
) or non_wechat_success
)
non_wechat_success = channel_success or non_wechat_success
_record_channel_result(
channel.value,
channel_success,
channel_error,
)
else:
logger.warning(f"未知通知渠道: {channel}")
success = wechat_success or non_wechat_success or context_success
has_targeted_channels = bool(channels)
success = wechat_success or non_wechat_success or (
not has_targeted_channels and send_context
)
if (
(wechat_success or non_wechat_success)
and noise_decision is not None
@@ -2316,9 +2540,21 @@ class StockAnalysisPipeline:
else:
logger.warning("决策仪表盘推送失败")
else:
record_notification_run(
channel="report",
status="not_configured",
success=False,
attempts=0,
)
logger.info("通知渠道未配置，跳过推送")
except Exception as e:
record_notification_run(
channel="report",
status="failed",
success=False,
error_message=e,
)
if (
noise_decision is not None
and not noise_finalized
src/services/analysis_service.py
+19
View File
@@ -24,6 +24,7 @@ from src.report_language import (
)
from src.services.run_diagnostics import (
activate_run_diagnostic_context,
build_run_diagnostic_summary,
get_current_diagnostic_context,
reset_run_diagnostic_context,
)
@@ -162,6 +163,23 @@ class AnalysisService:
stock_name = get_localized_stock_name(getattr(result, "name", None), result.code, report_language)
diagnostic_context = get_current_diagnostic_context()
trace_id = diagnostic_context.trace_id if diagnostic_context is not None else query_id
diagnostic_snapshot = diagnostic_context.snapshot() if diagnostic_context is not None else None
diagnostic_context_snapshot = getattr(result, "diagnostic_context_snapshot", None)
if isinstance(diagnostic_context_snapshot, dict):
context_snapshot = dict(diagnostic_context_snapshot)
if diagnostic_snapshot is not None:
context_snapshot["diagnostics"] = diagnostic_snapshot
elif diagnostic_snapshot is not None:
context_snapshot = {"diagnostics": diagnostic_snapshot}
else:
context_snapshot = None
diagnostic_summary = build_run_diagnostic_summary(
context_snapshot=context_snapshot,
raw_result=result.to_dict() if hasattr(result, "to_dict") else None,
report_saved=True,
query_id=query_id,
stock_code=result.code,
)
# 构建报告结构
report = {
@@ -203,4 +221,5 @@ class AnalysisService:
"stock_code": result.code,
"stock_name": stock_name,
"report": report,
"diagnostic_summary": diagnostic_summary,
}
src/services/history_service.py
+41
View File
@@ -30,6 +30,7 @@ from src.report_language import (
normalize_report_language,
)
from src.storage import DatabaseManager
from src.services.run_diagnostics import build_run_diagnostic_summary
from src.utils.data_processing import normalize_model_used, parse_json_field
if TYPE_CHECKING:
@@ -199,6 +200,46 @@ class HistoryService:
logger.error(f"resolve_and_get_news failed for {record_id}: {e}", exc_info=True)
return []
def resolve_and_get_diagnostics(self, record_id: str) -> Optional[Dict[str, Any]]:
"""
Resolve record_id and return a user-facing run diagnostic summary.
Legacy records without diagnostic snapshots return an ``unknown``
summary instead of failing. Storage and JSON parsing errors are
propagated so callers can surface backend failures accurately.
"""
record = self._resolve_record(record_id)
if not record:
return None
return build_run_diagnostic_summary(
context_snapshot=self._parse_diagnostic_json_field(
getattr(record, "context_snapshot", None),
"context_snapshot",
),
raw_result=self._parse_diagnostic_json_field(
getattr(record, "raw_result", None),
"raw_result",
),
report_saved=True,
query_id=getattr(record, "query_id", None),
stock_code=getattr(record, "code", None),
)
@staticmethod
def _parse_diagnostic_json_field(value: Any, field_name: str) -> Any:
"""Strict JSON parser for persisted diagnostic inputs."""
if value is None:
return None
if isinstance(value, str):
if not value.strip():
return None
try:
return json.loads(value)
except (json.JSONDecodeError, TypeError, ValueError) as exc:
raise ValueError(f"invalid {field_name} JSON") from exc
return value
def get_history_detail_by_id(self, record_id: int) -> Optional[Dict[str, Any]]:
"""
Get history report detail.
src/services/run_diagnostics.py
+651 -13
View File
@@ -23,13 +23,45 @@ _CURRENT_CONTEXT: ContextVar[Optional["RunDiagnosticContext"]] = ContextVar(
default=None,
)
_SECRET_PATTERNS = (
re.compile(
r"(?i)\b(api[_-]?key|access[_-]?token|token|secret|password|passwd|cookie|authorization)"
r"\s*[:=]\s*([^\s,&]+)"
_SECRET_REDACTIONS = (
(
re.compile(r"(?i)\b(authorization)\s*[:=]\s*(?:(?:Bearer|Basic|Token)\s+)?[^\s,&;]+"),
lambda match: f"{match.group(1)}=<redacted>",
),
(
re.compile(r"(https?://)([^/\s:@]+):([^@\s/]+)@"),
r"\1<redacted>:<redacted>@",
),
(
re.compile(r"https?://[^\s]+?(?:token|key|secret|webhook)[^\s]*", re.IGNORECASE),
"<redacted-url>",
),
(
re.compile(
r"(?i)([\"']?)"
r"([A-Z0-9_]*?(?:api[_-]?key|access[_-]?token|token|secret|password|passwd|cookie))"
r"\1\s*:\s*([\"'])([^\"']+)\3"
),
lambda match: f"{match.group(1)}{match.group(2)}{match.group(1)}: {match.group(3)}<redacted>{match.group(3)}",
),
(
re.compile(
r"(?i)\b([A-Z0-9_]*?(?:api[_-]?key|access[_-]?token|token|secret|password|passwd|cookie))"
r"\s*=\s*([^\s,&;]+)"
),
lambda match: f"{match.group(1)}=<redacted>",
),
(
re.compile(
r"(?i)\b(api[_-]?key|access[_-]?token|token|secret|password|passwd|cookie)"
r"\s*:\s*([^\s,&;]+)"
),
lambda match: f"{match.group(1)}=<redacted>",
),
(
re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+"),
"Bearer <redacted>",
),
re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+"),
re.compile(r"https?://[^\s]+?(?:token|key|secret|webhook)[^\s]*", re.IGNORECASE),
)
@@ -47,13 +79,8 @@ def sanitize_diagnostic_text(value: Any, *, max_length: int = 300) -> Optional[s
if not text:
return None
for pattern in _SECRET_PATTERNS:
if "Bearer" in pattern.pattern:
text = pattern.sub("Bearer <redacted>", text)
elif "https?" in pattern.pattern:
text = pattern.sub("<redacted-url>", text)
else:
text = pattern.sub(lambda match: f"{match.group(1)}=<redacted>", text)
for pattern, replacement in _SECRET_REDACTIONS:
text = pattern.sub(replacement, text)
if len(text) > max_length:
return f"{text[:max_length].rstrip()}..."
@@ -99,6 +126,141 @@ class ProviderRun:
return {key: value for key, value in payload.items() if value is not None}
@dataclass
class LLMRun:
"""One LLM call result in a trace."""
trace_id: str
provider: Optional[str] = None
model: Optional[str] = None
call_type: str = "analysis"
success: bool = True
tokens: Optional[int] = None
duration_ms: Optional[int] = None
fallback_model: Optional[str] = None
error_type: Optional[str] = None
error_message_sanitized: Optional[str] = None
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> Dict[str, Any]:
payload = {
"trace_id": self.trace_id,
"provider": self.provider,
"model": self.model,
"call_type": self.call_type,
"success": self.success,
"tokens": self.tokens,
"duration_ms": self.duration_ms,
"fallback_model": self.fallback_model,
"error_type": self.error_type,
"error_message_sanitized": self.error_message_sanitized,
"created_at": self.created_at,
}
return {key: value for key, value in payload.items() if value is not None}
@dataclass
class NotificationRun:
"""Notification dispatch result in a trace."""
trace_id: str
channel: str
status: str
success: bool
attempts: int = 1
error_message_sanitized: Optional[str] = None
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> Dict[str, Any]:
payload = {
"trace_id": self.trace_id,
"channel": self.channel,
"status": self.status,
"success": self.success,
"attempts": self.attempts,
"error_message_sanitized": self.error_message_sanitized,
"created_at": self.created_at,
}
return {key: value for key, value in payload.items() if value is not None}
@dataclass
class HistoryRun:
"""History persistence result in a trace."""
trace_id: str
report_saved: bool
metadata_saved: Optional[bool] = None
analysis_history_id: Optional[int] = None
error_message_sanitized: Optional[str] = None
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> Dict[str, Any]:
payload = {
"trace_id": self.trace_id,
"report_saved": self.report_saved,
"metadata_saved": self.metadata_saved,
"analysis_history_id": self.analysis_history_id,
"error_message_sanitized": self.error_message_sanitized,
"created_at": self.created_at,
}
return {key: value for key, value in payload.items() if value is not None}
@dataclass
class RunDiagnosticComponent:
"""User-facing status for one diagnostic component."""
key: str
label: str
status: str
message: str
details: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
payload = {
"key": self.key,
"label": self.label,
"status": self.status,
"message": self.message,
"details": self.details,
}
return {key: value for key, value in payload.items() if value not in (None, {}, [])}
@dataclass
class RunDiagnosticSummary:
"""User-facing diagnostic summary for one analysis run."""
status: str
status_label: str
reason: str
trace_id: Optional[str] = None
task_id: Optional[str] = None
query_id: Optional[str] = None
stock_code: Optional[str] = None
trigger_source: Optional[str] = None
components: Dict[str, RunDiagnosticComponent] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
payload = {
"trace_id": self.trace_id,
"task_id": self.task_id,
"query_id": self.query_id,
"stock_code": self.stock_code,
"trigger_source": self.trigger_source,
"status": self.status,
"status_label": self.status_label,
"reason": self.reason,
"components": {
key: component.to_dict()
for key, component in self.components.items()
},
}
payload["copy_text"] = format_copyable_diagnostics(payload)
return {key: value for key, value in payload.items() if value is not None}
@dataclass
class RunDiagnosticContext:
"""Diagnostic state for one analysis run."""
@@ -109,10 +271,22 @@ class RunDiagnosticContext:
stock_code: Optional[str] = None
trigger_source: Optional[str] = None
provider_runs: List[ProviderRun] = field(default_factory=list)
llm_runs: List[LLMRun] = field(default_factory=list)
notification_runs: List[NotificationRun] = field(default_factory=list)
history_runs: List[HistoryRun] = field(default_factory=list)
def record_provider_run(self, provider_run: ProviderRun) -> None:
self.provider_runs.append(provider_run)
def record_llm_run(self, llm_run: LLMRun) -> None:
self.llm_runs.append(llm_run)
def record_notification_run(self, notification_run: NotificationRun) -> None:
self.notification_runs.append(notification_run)
def record_history_run(self, history_run: HistoryRun) -> None:
self.history_runs.append(history_run)
def snapshot(self) -> Dict[str, Any]:
return {
"trace_id": self.trace_id,
@@ -121,6 +295,9 @@ class RunDiagnosticContext:
"stock_code": self.stock_code,
"trigger_source": self.trigger_source,
"provider_runs": [run.to_dict() for run in self.provider_runs],
"llm_runs": [run.to_dict() for run in self.llm_runs],
"notification_runs": [run.to_dict() for run in self.notification_runs],
"history_runs": [run.to_dict() for run in self.history_runs],
}
@@ -207,3 +384,464 @@ def record_provider_run(
)
except Exception as exc: # pragma: no cover - defensive fail-open guard
logger.warning("provider diagnostic record failed: %s", exc)
def record_llm_run(
*,
success: bool,
provider: Optional[str] = None,
model: Optional[str] = None,
call_type: str = "analysis",
tokens: Optional[int] = None,
duration_ms: Optional[int] = None,
fallback_model: Optional[str] = None,
error_type: Optional[str] = None,
error_message: Optional[Any] = None,
) -> None:
"""Append an LLM call result to the active context without affecting callers."""
context = get_current_diagnostic_context()
if context is None:
return
try:
context.record_llm_run(
LLMRun(
trace_id=context.trace_id,
provider=provider,
model=model,
call_type=call_type,
success=success,
tokens=tokens,
duration_ms=duration_ms,
fallback_model=fallback_model,
error_type=error_type,
error_message_sanitized=sanitize_diagnostic_text(error_message),
)
)
except Exception as exc: # pragma: no cover - defensive fail-open guard
logger.warning("llm diagnostic record failed: %s", exc)
def record_notification_run(
*,
channel: str,
status: str,
success: bool,
attempts: int = 1,
error_message: Optional[Any] = None,
) -> None:
"""Append a notification result to the active context without affecting callers."""
context = get_current_diagnostic_context()
if context is None:
return
try:
context.record_notification_run(
NotificationRun(
trace_id=context.trace_id,
channel=channel,
status=status,
success=success,
attempts=attempts,
error_message_sanitized=sanitize_diagnostic_text(error_message),
)
)
except Exception as exc: # pragma: no cover - defensive fail-open guard
logger.warning("notification diagnostic record failed: %s", exc)
def record_history_run(
*,
report_saved: bool,
metadata_saved: Optional[bool] = None,
analysis_history_id: Optional[int] = None,
error_message: Optional[Any] = None,
) -> None:
"""Append a history persistence result to the active context without affecting callers."""
context = get_current_diagnostic_context()
if context is None:
return
try:
context.record_history_run(
HistoryRun(
trace_id=context.trace_id,
report_saved=report_saved,
metadata_saved=metadata_saved,
analysis_history_id=analysis_history_id,
error_message_sanitized=sanitize_diagnostic_text(error_message),
)
)
except Exception as exc: # pragma: no cover - defensive fail-open guard
logger.warning("history diagnostic record failed: %s", exc)
_SUMMARY_STATUS_LABELS = {
"normal": "正常",
"degraded": "部分降级",
"failed": "失败",
"unknown": "未知",
}
def _as_dict(value: Any) -> Dict[str, Any]:
return value if isinstance(value, dict) else {}
def _as_list(value: Any) -> List[Any]:
return value if isinstance(value, list) else []
def _component(
key: str,
label: str,
status: str,
message: str,
details: Optional[Dict[str, Any]] = None,
) -> RunDiagnosticComponent:
clean_details = {
key: value
for key, value in (details or {}).items()
if value is not None
}
return RunDiagnosticComponent(
key=key,
label=label,
status=status,
message=message,
details=clean_details,
)
def _provider_component(
*,
key: str,
label: str,
data_type: str,
provider_runs: List[Dict[str, Any]],
) -> RunDiagnosticComponent:
runs = [
run for run in provider_runs
if isinstance(run, dict) and run.get("data_type") == data_type
]
if not runs:
return _component(key, label, "unknown", f"{label}未记录诊断信息")
successes = [run for run in runs if run.get("success") is True]
failures = [run for run in runs if run.get("success") is False]
last_run = runs[-1]
if successes:
success_run = successes[-1]
provider = success_run.get("provider") or "unknown"
record_count = success_run.get("record_count")
details = {
"provider": provider,
"attempts": len(runs),
"record_count": record_count,
"fallback_to": next(
(run.get("fallback_to") for run in failures if run.get("fallback_to")),
None,
),
}
details = {key: value for key, value in details.items() if value is not None}
if failures:
return _component(
key,
label,
"degraded",
f"{label}{provider} 成功，前置数据源失败后已继续",
details,
)
return _component(
key,
label,
"ok",
f"{label}{provider} 成功",
details,
)
message = (
last_run.get("error_message_sanitized")
or last_run.get("error_type")
or "所有数据源尝试失败"
)
return _component(
key,
label,
"failed",
f"{label}失败：{message}",
{
"attempts": len(runs),
"provider": last_run.get("provider"),
"error_type": last_run.get("error_type"),
},
)
def _news_component(context_snapshot: Dict[str, Any], raw_result: Dict[str, Any]) -> RunDiagnosticComponent:
label = "新闻搜索"
has_retrieval_news = "news_retrieval_content" in context_snapshot
has_snapshot_news = has_retrieval_news or "news_content" in context_snapshot
news_result_count = context_snapshot.get("news_result_count")
if isinstance(news_result_count, int):
if news_result_count > 0:
return _component(
"news",
label,
"ok",
f"新闻检索返回 {news_result_count} 条结果",
{"record_count": news_result_count},
)
return _component("news", label, "degraded", "新闻搜索无结果", {"record_count": 0})
if has_snapshot_news and not has_retrieval_news:
return _component("news", label, "unknown", "新闻检索未记录原始证据，可能未尝试或未启用")
return _component("news", label, "unknown", "新闻搜索未记录诊断信息")
def _llm_component(diagnostics: Dict[str, Any], raw_result: Dict[str, Any]) -> RunDiagnosticComponent:
label = "LLM"
runs = [
run for run in _as_list(diagnostics.get("llm_runs"))
if isinstance(run, dict)
]
if runs:
successes = [run for run in runs if run.get("success") is True]
failures = [run for run in runs if run.get("success") is False]
last_run = runs[-1]
if successes:
success_run = successes[-1]
model = success_run.get("model") or raw_result.get("model_used") or "unknown"
status = "degraded" if failures or success_run.get("fallback_model") else "ok"
message = f"LLM {model} 成功"
if status == "degraded":
message = f"LLM {model} 成功，期间发生过失败或模型切换"
return _component(
"llm",
label,
status,
message,
{
"model": model,
"tokens": success_run.get("tokens"),
"duration_ms": success_run.get("duration_ms"),
"fallback_model": success_run.get("fallback_model"),
},
)
return _component(
"llm",
label,
"failed",
f"LLM 失败：{last_run.get('error_message_sanitized') or last_run.get('error_type') or '未知错误'}",
{"model": last_run.get("model"), "error_type": last_run.get("error_type")},
)
if raw_result:
if raw_result.get("success") is False:
return _component(
"llm",
label,
"failed",
f"LLM 失败：{sanitize_diagnostic_text(raw_result.get('error_message')) or '未知错误'}",
)
model = raw_result.get("model_used")
if model:
return _component("llm", label, "ok", f"LLM {model} 成功", {"model": model})
if raw_result.get("analysis_summary"):
return _component("llm", label, "ok", "LLM 成功，模型未记录")
return _component("llm", label, "unknown", "LLM 未记录诊断信息")
def _notification_component(diagnostics: Dict[str, Any]) -> RunDiagnosticComponent:
label = "通知"
runs = [
run for run in _as_list(diagnostics.get("notification_runs"))
if isinstance(run, dict)
]
if not runs:
return _component("notification", label, "unknown", "通知结果未记录")
skipped = [run for run in runs if run.get("status") in {"skipped", "not_configured"}]
successes = [run for run in runs if run.get("success") is True]
failures = [run for run in runs if run.get("success") is False and run not in skipped]
channels = [run.get("channel") for run in runs if run.get("channel")]
if successes and failures:
return _component(
"notification",
label,
"degraded",
"部分通知渠道失败，其余渠道已发送",
{"channels": channels, "failed": [run.get("channel") for run in failures]},
)
if successes:
return _component(
"notification",
label,
"ok",
"通知发送成功",
{"channels": channels},
)
if skipped and not failures:
status = "not_configured" if any(run.get("status") == "not_configured" for run in skipped) else "skipped"
return _component(
"notification",
label,
status,
"通知未配置或本次跳过",
{"channels": channels},
)
last_failure = failures[-1] if failures else runs[-1]
return _component(
"notification",
label,
"failed",
f"通知失败：{last_failure.get('error_message_sanitized') or last_failure.get('status') or '未知错误'}",
{"channels": channels},
)
def _history_component(
diagnostics: Dict[str, Any],
report_saved: Optional[bool],
) -> RunDiagnosticComponent:
label = "历史保存"
runs = [
run for run in _as_list(diagnostics.get("history_runs"))
if isinstance(run, dict)
]
if runs:
last_run = runs[-1]
if last_run.get("report_saved") is True:
return _component(
"history",
label,
"ok",
"报告历史已保存",
{"analysis_history_id": last_run.get("analysis_history_id")},
)
return _component(
"history",
label,
"failed",
f"报告历史保存失败：{last_run.get('error_message_sanitized') or '未知错误'}",
)
if report_saved is True:
return _component("history", label, "ok", "报告历史已保存")
if report_saved is False:
return _component("history", label, "failed", "报告历史保存失败")
return _component("history", label, "unknown", "历史保存未记录诊断信息")
def build_run_diagnostic_summary(
*,
context_snapshot: Optional[Any] = None,
raw_result: Optional[Any] = None,
report_saved: Optional[bool] = None,
query_id: Optional[str] = None,
stock_code: Optional[str] = None,
) -> Dict[str, Any]:
"""Build a user-facing diagnostic summary from persisted or in-memory evidence."""
snapshot = _as_dict(context_snapshot)
raw = _as_dict(raw_result)
diagnostics = _as_dict(snapshot.get("diagnostics"))
provider_runs = [
run for run in _as_list(diagnostics.get("provider_runs"))
if isinstance(run, dict)
]
components = {
"realtime_quote": _provider_component(
key="realtime_quote",
label="实时行情",
data_type="realtime_quote",
provider_runs=provider_runs,
),
"daily_data": _provider_component(
key="daily_data",
label="日线数据",
data_type="daily_data",
provider_runs=provider_runs,
),
"news": _news_component(snapshot, raw),
"llm": _llm_component(diagnostics, raw),
"notification": _notification_component(diagnostics),
"history": _history_component(diagnostics, report_saved),
}
has_evidence = bool(snapshot or raw or diagnostics or report_saved is not None)
if not has_evidence or not diagnostics:
status = "unknown"
elif components["llm"].status == "failed" or components["history"].status == "failed":
status = "failed"
elif any(component.status in {"failed", "degraded"} for component in components.values()):
status = "degraded"
elif all(component.status == "unknown" for component in components.values()):
status = "unknown"
else:
status = "normal"
if status == "unknown":
reason = "旧报告或诊断证据不足，无法判断本次运行状态"
else:
reason = next(
(
component.message
for component in components.values()
if component.status == "failed"
),
next(
(
component.message
for component in components.values()
if component.status == "degraded"
),
_SUMMARY_STATUS_LABELS[status],
),
)
trace_id = diagnostics.get("trace_id") or snapshot.get("trace_id") or raw.get("trace_id")
resolved_query_id = query_id or diagnostics.get("query_id") or snapshot.get("query_id") or raw.get("query_id")
resolved_stock_code = (
stock_code
or diagnostics.get("stock_code")
or snapshot.get("stock_code")
or raw.get("code")
or raw.get("stock_code")
)
return RunDiagnosticSummary(
trace_id=trace_id,
task_id=diagnostics.get("task_id"),
query_id=resolved_query_id,
stock_code=resolved_stock_code,
trigger_source=diagnostics.get("trigger_source") or snapshot.get("trigger_source"),
status=status,
status_label=_SUMMARY_STATUS_LABELS[status],
reason=reason,
components=components,
).to_dict()
def format_copyable_diagnostics(summary: Dict[str, Any]) -> str:
"""Format a sanitized plain-text diagnostic payload for issue reports."""
components = _as_dict(summary.get("components"))
def _component_line(key: str) -> str:
component = _as_dict(components.get(key))
message = sanitize_diagnostic_text(component.get("message"), max_length=160) or "unknown"
return f"{key}: {component.get('status', 'unknown')} - {message}"
lines = [
f"trace_id: {summary.get('trace_id') or 'unknown'}",
f"query_id: {summary.get('query_id') or 'unknown'}",
f"stock_code: {summary.get('stock_code') or 'unknown'}",
f"trigger_source: {summary.get('trigger_source') or 'unknown'}",
f"data_status: {summary.get('status', 'unknown')}",
_component_line("realtime_quote"),
_component_line("daily_data"),
_component_line("news"),
_component_line("llm"),
_component_line("notification"),
_component_line("history"),
f"reason: {sanitize_diagnostic_text(summary.get('reason'), max_length=160) or 'unknown'}",
]
return "\n".join(lines)
tests/test_agent_pipeline.py
+269
View File
@@ -139,6 +139,198 @@ class TestAgentConfig(unittest.TestCase):
Config._instance = None
def test_build_agent_executor_does_not_mutate_llm_route_config(self) -> None:
"""Agent factory should not rewrite model/base_url/runtime routing fields."""
provided_config = SimpleNamespace(
agent_arch="single",
agent_skills=["bull_trend"],
agent_max_steps="10",
agent_orchestrator_timeout_s="120",
litellm_model="openai/gpt-5",
agent_litellm_model="anthropic/claude-3-7-sonnet-20250219",
openai_base_url="https://api.openai.com/v1",
)
captured: Dict[str, Any] = {}
def _mock_llm_adapter(cfg):
captured["cfg"] = cfg
return MagicMock()
fake_llm_module = types.ModuleType("src.agent.llm_adapter")
fake_llm_module.LLMToolAdapter = _mock_llm_adapter
fake_executor_module = types.ModuleType("src.agent.executor")
fake_executor_cls = MagicMock(return_value=MagicMock())
fake_executor_module.AgentExecutor = fake_executor_cls
skill_manager = MagicMock()
skill_manager.list_skills.return_value = [
SimpleNamespace(
name="bull_trend",
display_name="bull_trend",
description="bull_trend desc",
instructions="测试指令",
default_active=True,
default_router=True,
default_priority=100,
user_invocable=True,
source="builtin",
)
]
skill_manager.get_skill_instructions.return_value = "测试指令"
with patch.dict(sys.modules, {
"litellm": MagicMock(),
"src.agent.llm_adapter": fake_llm_module,
"src.agent.executor": fake_executor_module,
}):
factory_module = importlib.import_module("src.agent.factory")
with patch.object(factory_module, "get_skill_manager", return_value=skill_manager), \
patch.object(factory_module, "get_tool_registry", return_value=MagicMock()):
factory_module.build_agent_executor(provided_config)
adapter_cfg = captured.get("cfg")
self.assertIs(adapter_cfg, provided_config)
self.assertEqual(provided_config.agent_max_steps, "10")
self.assertEqual(provided_config.agent_orchestrator_timeout_s, "120")
self.assertEqual(provided_config.litellm_model, "openai/gpt-5")
self.assertEqual(provided_config.agent_litellm_model, "anthropic/claude-3-7-sonnet-20250219")
self.assertEqual(provided_config.openai_base_url, "https://api.openai.com/v1")
fake_executor_cls.assert_called_once()
kwargs = fake_executor_cls.call_args.kwargs
self.assertEqual(kwargs["max_steps"], 10)
self.assertEqual(kwargs["timeout_seconds"], 120)
def test_build_agent_executor_multi_arch_does_not_mutate_llm_route_config(self) -> None:
"""Multi-arch path should keep provider/base_url/runtime fields unchanged."""
provided_config = SimpleNamespace(
agent_arch="multi",
agent_skills=["bull_trend"],
agent_max_steps="10",
agent_orchestrator_timeout_s="120",
litellm_model="openai/gpt-5",
agent_litellm_model="anthropic/claude-3-7-sonnet-20250219",
openai_base_url="https://api.openai.com/v1",
agent_orchestrator_mode="standard",
)
captured: Dict[str, Any] = {}
def _mock_llm_adapter(cfg):
captured["cfg"] = cfg
return MagicMock()
fake_llm_module = types.ModuleType("src.agent.llm_adapter")
fake_llm_module.LLMToolAdapter = _mock_llm_adapter
fake_orchestrator_module = types.ModuleType("src.agent.orchestrator")
fake_orchestrator_cls = MagicMock(return_value=MagicMock())
fake_orchestrator_module.AgentOrchestrator = fake_orchestrator_cls
skill_manager = MagicMock()
skill_manager.list_skills.return_value = [
SimpleNamespace(
name="bull_trend",
display_name="bull_trend",
description="bull_trend desc",
instructions="测试指令",
default_active=True,
default_router=True,
default_priority=100,
user_invocable=True,
source="builtin",
)
]
skill_manager.get_skill_instructions.return_value = "测试指令"
with patch.dict(sys.modules, {
"litellm": MagicMock(),
"src.agent.llm_adapter": fake_llm_module,
"src.agent.orchestrator": fake_orchestrator_module,
"src.agent.executor": MagicMock(),
}):
factory_module = importlib.import_module("src.agent.factory")
with patch.object(factory_module, "get_skill_manager", return_value=skill_manager), \
patch.object(factory_module, "get_tool_registry", return_value=MagicMock()):
factory_module.build_agent_executor(provided_config)
adapter_cfg = captured.get("cfg")
self.assertIs(adapter_cfg, provided_config)
self.assertEqual(provided_config.agent_max_steps, "10")
self.assertEqual(provided_config.agent_orchestrator_timeout_s, "120")
self.assertEqual(provided_config.litellm_model, "openai/gpt-5")
self.assertEqual(provided_config.agent_litellm_model, "anthropic/claude-3-7-sonnet-20250219")
self.assertEqual(provided_config.openai_base_url, "https://api.openai.com/v1")
fake_orchestrator_cls.assert_called_once()
kwargs = fake_orchestrator_cls.call_args.kwargs
self.assertEqual(kwargs["max_steps"], 10)
self.assertIs(kwargs["config"], provided_config)
def test_invalid_numeric_config_values_fallback_to_defaults_with_warning(self) -> None:
"""Invalid agent_max_steps / agent_orchestrator_timeout_s should fallback and emit warning."""
provided_config = SimpleNamespace(
agent_arch="single",
agent_skills=["bull_trend"],
agent_max_steps="invalid-steps",
agent_orchestrator_timeout_s="invalid-timeout",
litellm_model="openai/gpt-5",
agent_litellm_model="anthropic/claude-3-7-sonnet-20250219",
openai_base_url="https://api.openai.com/v1",
)
captured: Dict[str, Any] = {}
def _mock_llm_adapter(cfg):
captured["cfg"] = cfg
return MagicMock()
fake_llm_module = types.ModuleType("src.agent.llm_adapter")
fake_llm_module.LLMToolAdapter = _mock_llm_adapter
fake_executor_module = types.ModuleType("src.agent.executor")
fake_executor_cls = MagicMock(return_value=MagicMock())
fake_executor_module.AgentExecutor = fake_executor_cls
skill_manager = MagicMock()
skill_manager.list_skills.return_value = [
SimpleNamespace(
name="bull_trend",
display_name="bull_trend",
description="bull_trend desc",
instructions="测试指令",
default_active=True,
default_router=True,
default_priority=100,
user_invocable=True,
source="builtin",
)
]
skill_manager.get_skill_instructions.return_value = "测试指令"
with self.assertLogs("src.agent.factory", level="WARNING") as logs:
with patch.dict(sys.modules, {
"litellm": MagicMock(),
"src.agent.llm_adapter": fake_llm_module,
"src.agent.executor": fake_executor_module,
}):
factory_module = importlib.import_module("src.agent.factory")
with patch.object(factory_module, "get_skill_manager", return_value=skill_manager), \
patch.object(factory_module, "get_tool_registry", return_value=MagicMock()):
factory_module.build_agent_executor(provided_config)
adapter_cfg = captured.get("cfg")
self.assertIs(adapter_cfg, provided_config)
self.assertEqual(provided_config.litellm_model, "openai/gpt-5")
self.assertEqual(provided_config.agent_litellm_model, "anthropic/claude-3-7-sonnet-20250219")
self.assertEqual(provided_config.openai_base_url, "https://api.openai.com/v1")
log_output = "\n".join(logs.output)
self.assertIn("[AgentFactory] Invalid value for agent_max_steps", log_output)
self.assertIn("[AgentFactory] Invalid value for agent_orchestrator_timeout_s", log_output)
kwargs = fake_executor_cls.call_args.kwargs
from src.config import AGENT_MAX_STEPS_DEFAULT
self.assertEqual(kwargs["max_steps"], AGENT_MAX_STEPS_DEFAULT)
self.assertEqual(kwargs["timeout_seconds"], 0)
class TestAgentFactorySkillBaseline(unittest.TestCase):
"""Ensure explicit skill selection does not silently re-apply the default bull-trend baseline."""
@@ -1538,6 +1730,83 @@ class TestAnalyzeWithAgentStockName(unittest.TestCase):
self.assertEqual(dp["chip_structure"]["concentration"], "0.00%")
self.assertNotIn("chip_unavailable_reason", dp)
def test_analyze_with_agent_history_context_includes_diagnostic_snapshot(self):
"""Agent 分析入库存档时应保留 diagnostics 快照，避免历史诊断返回 unknown。"""
with patch('src.core.pipeline.get_config') as mock_config, \
patch('src.core.pipeline.get_db'), \
patch('src.core.pipeline.DataFetcherManager'), \
patch('src.core.pipeline.GeminiAnalyzer'), \
patch('src.core.pipeline.NotificationService'), \
patch('src.core.pipeline.SearchService'), \
patch('src.core.pipeline.fill_price_position_if_needed'), \
patch('src.core.pipeline.stabilize_decision_with_structure'), \
patch('src.core.pipeline.current_diagnostic_snapshot') as mock_diagnostic_snapshot:
mock_cfg = MagicMock()
mock_cfg.max_workers = 2
mock_cfg.agent_mode = True
mock_cfg.agent_max_steps = 10
mock_cfg.agent_skills = []
mock_cfg.bocha_api_keys = []
mock_cfg.tavily_api_keys = []
mock_cfg.anspire_api_keys = []
mock_cfg.brave_api_keys = []
mock_cfg.serpapi_keys = []
mock_cfg.searxng_base_urls = []
mock_cfg.searxng_public_instances_enabled = False
mock_cfg.news_max_age_days = 7
mock_cfg.enable_realtime_quote = True
mock_cfg.enable_chip_distribution = True
mock_cfg.realtime_source_priority = []
mock_cfg.save_context_snapshot = True
mock_cfg.report_language = "zh"
mock_cfg.report_integrity_enabled = False
mock_config.return_value = mock_cfg
from src.core.pipeline import StockAnalysisPipeline
from src.enums import ReportType
pipeline = StockAnalysisPipeline(config=mock_cfg)
pipeline.search_service.is_available = False
pipeline._ensure_agent_history = MagicMock()
pipeline._agent_result_to_analysis_result = MagicMock(
return_value=SimpleNamespace(
success=True,
code="588200",
name="科创芯片ETF",
model_used="agent-model",
sentiment_score=70,
operation_advice="持有",
trend_prediction="震荡",
analysis_summary="测试摘要",
)
)
mock_executor = MagicMock()
mock_executor.run.return_value = SimpleNamespace(
success=True,
provider="agent-provider",
dashboard={"stock_name": "科创芯片ETF"},
)
with patch('src.agent.factory.build_agent_executor', return_value=mock_executor):
mock_diagnostic_snapshot.return_value = {"trace_id": "trace-1391", "query_id": "q-1391"}
pipeline.db.save_analysis_history = MagicMock(return_value=1)
result = pipeline._analyze_with_agent(
code="588200",
report_type=ReportType.SIMPLE,
query_id="q-1391",
stock_name="科创芯片ETF",
realtime_quote=None,
chip_data=None,
)
self.assertIsNotNone(result)
call_kwargs = pipeline.db.save_analysis_history.call_args.kwargs
history_context = call_kwargs["context_snapshot"]
self.assertIn("diagnostics", history_context)
self.assertEqual(history_context["diagnostics"]["trace_id"], "trace-1391")
self.assertEqual(history_context["stock_name"], "科创芯片ETF")
# ============================================================
# Agent construction chain (real objects, mocked LLM)
tests/test_analysis_api_contract.py
+68
View File
@@ -320,6 +320,48 @@ class AnalysisApiContractTestCase(unittest.TestCase):
"summary",
)
def test_get_analysis_status_preserves_queue_report_created_at_when_enriching(self) -> None:
if get_analysis_status is None or analysis_endpoint_module is None:
self.skipTest("analysis endpoint helpers unavailable in this environment")
created_at = datetime(2026, 5, 21, 17, 40, 0)
queue = MagicMock()
queue.get_task.return_value = SimpleNamespace(
task_id="task-queue-2",
stock_code="600519",
stock_name="贵州茅台",
status=analysis_endpoint_module.TaskStatusEnum.COMPLETED,
progress=100,
result={
"stock_code": "600519",
"stock_name": "贵州茅台",
"report": {
"meta": {"query_id": "task-queue-2", "stock_code": "600519"},
"summary": {"analysis_summary": "summary"},
},
},
error=None,
original_query=None,
selection_source=None,
created_at=created_at,
completed_at=datetime(2026, 5, 21, 17, 45, 0),
)
with patch("api.v1.endpoints.analysis.get_task_queue", return_value=queue), \
patch(
"api.v1.endpoints.analysis._load_sync_fundamental_sources",
return_value=({}, None),
):
status = get_analysis_status("task-queue-2")
self.assertEqual(status.status, "completed")
self.assertIsNotNone(status.result)
self.assertEqual(status.result.created_at, created_at.isoformat())
self.assertEqual(
status.result.report["meta"]["created_at"],
created_at.isoformat(),
)
def test_run_market_review_background_raises_when_report_is_empty(self) -> None:
if analysis_endpoint_module is None:
self.skipTest("analysis endpoint helpers unavailable in this environment")
@@ -694,6 +736,32 @@ class AnalysisApiContractTestCase(unittest.TestCase):
self.assertEqual(result["stock_name"], "Unnamed Stock")
self.assertEqual(result["report"]["meta"]["stock_name"], "Unnamed Stock")
def test_build_analysis_response_does_not_use_model_news_summary_as_retrieval_evidence(self) -> None:
service = AnalysisService()
result = service._build_analysis_response(
SimpleNamespace(
code="600519",
name="贵州茅台",
current_price=1234.56,
change_pct=1.23,
model_used="test-model",
analysis_summary="summary",
operation_advice="hold",
trend_prediction="up",
sentiment_score=80,
news_summary="model generated news summary",
technical_analysis="tech",
fundamental_analysis="fundamental",
risk_warning="risk",
get_sniper_points=lambda: {},
),
"q1",
report_type="full",
)
news_component = result["diagnostic_summary"]["components"]["news"]
self.assertEqual(news_component["status"], "unknown")
def test_build_analysis_report_extracts_fundamental_fields_from_snapshot(self) -> None:
if _build_analysis_report is None:
self.skipTest("analysis endpoint helpers unavailable in this environment")
tests/test_pipeline_notification_image_routing.py
+43
View File
@@ -16,6 +16,12 @@ from tests.litellm_stub import ensure_litellm_stub
ensure_litellm_stub()
from src.core.pipeline import StockAnalysisPipeline, NotificationChannel
from src.services.run_diagnostics import (
activate_run_diagnostic_context,
build_run_diagnostic_summary,
current_diagnostic_snapshot,
reset_run_diagnostic_context,
)
from src.enums import ReportType
@@ -344,6 +350,43 @@ class TestPipelineReportRouteFiltering(unittest.TestCase):
pipeline.notifier.record_noise_control.assert_not_called()
pipeline.notifier.release_noise_control.assert_called_once()
def test_notification_summary_degraded_when_only_partial_channels_fail(self):
pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
pipeline.notifier = _FakeRoutedNotifier([NotificationChannel.WECHAT, NotificationChannel.TELEGRAM])
pipeline.config = SimpleNamespace(stock_email_groups=[])
pipeline.notifier.send_to_wechat.return_value = True
pipeline.notifier.send_to_telegram.return_value = False
results = [SimpleNamespace(code="000001")]
token = activate_run_diagnostic_context(
trace_id="trace-notify",
query_id="query-notify",
stock_code="000001",
trigger_source="api",
)
try:
pipeline._send_notifications(results, ReportType.SIMPLE)
snapshot = current_diagnostic_snapshot()
finally:
reset_run_diagnostic_context(token)
self.assertEqual(snapshot["notification_runs"][0]["channel"], "wechat")
self.assertEqual(snapshot["notification_runs"][0]["success"], True)
self.assertEqual(snapshot["notification_runs"][1]["channel"], "telegram")
self.assertEqual(snapshot["notification_runs"][1]["success"], False)
summary = build_run_diagnostic_summary(
context_snapshot={"diagnostics": snapshot},
raw_result={"success": True, "model_used": "deepseek-chat"},
report_saved=True,
)
self.assertEqual(summary["components"]["notification"]["status"], "degraded")
self.assertIn(
"telegram",
summary["components"]["notification"]["details"]["failed"],
)
if __name__ == "__main__":
unittest.main()
tests/test_run_diagnostics_p2.py
+424
View File
@@ -0,0 +1,424 @@
# -*- coding: utf-8 -*-
"""Regression tests for #1391 Phase 2 run diagnostic summaries."""
from __future__ import annotations
import json
import os
import sys
import unittest
from datetime import datetime
from types import SimpleNamespace
from fastapi import HTTPException
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from api.v1.endpoints.history import get_history_diagnostics
from src.services.history_service import HistoryService
from src.services.run_diagnostics import build_run_diagnostic_summary, sanitize_diagnostic_text
def _diagnostic_snapshot() -> dict:
return {
"trace_id": "trace-p2",
"task_id": "task-p2",
"query_id": "query-p2",
"stock_code": "600519",
"trigger_source": "api",
"provider_runs": [
{
"trace_id": "trace-p2",
"data_type": "realtime_quote",
"provider": "FirstQuote",
"operation": "get_realtime_quote",
"success": False,
"error_type": "TimeoutError",
"error_message_sanitized": "token=<redacted>",
"fallback_to": "SecondQuote",
},
{
"trace_id": "trace-p2",
"data_type": "realtime_quote",
"provider": "SecondQuote",
"operation": "get_realtime_quote",
"success": True,
},
{
"trace_id": "trace-p2",
"data_type": "daily_data",
"provider": "DailyFetcher",
"operation": "get_daily_data",
"success": True,
"record_count": 30,
},
],
"llm_runs": [
{
"trace_id": "trace-p2",
"model": "deepseek-chat",
"call_type": "analysis",
"success": True,
"tokens": 1234,
}
],
"notification_runs": [
{
"trace_id": "trace-p2",
"channel": "wechat",
"status": "success",
"success": True,
}
],
"history_runs": [
{
"trace_id": "trace-p2",
"report_saved": True,
"metadata_saved": True,
}
],
}
def _history_record(*, context_snapshot: dict | None) -> SimpleNamespace:
return SimpleNamespace(
id=1,
query_id="query-p2",
code="600519",
name="贵州茅台",
report_type="detailed",
created_at=datetime(2026, 5, 24, 12, 0, 0),
raw_result=json.dumps(
{
"success": True,
"model_used": "deepseek-chat",
"analysis_summary": "测试摘要",
"news_summary": "新闻摘要",
},
ensure_ascii=False,
),
context_snapshot=(
json.dumps(context_snapshot, ensure_ascii=False)
if context_snapshot is not None
else None
),
sentiment_score=60,
operation_advice="持有",
trend_prediction="看多",
analysis_summary="测试摘要",
news_content="新闻摘要",
ideal_buy=None,
secondary_buy=None,
stop_loss=None,
take_profit=None,
)
class _FakeHistoryDb:
def __init__(self, record: SimpleNamespace | None):
self.record = record
def get_analysis_history_by_id(self, record_id: int):
return self.record if record_id == 1 else None
def get_latest_analysis_by_query_id(self, query_id: str):
return self.record if query_id == "query-p2" else None
class _FailingHistoryDb:
def get_analysis_history_by_id(self, record_id: int):
raise RuntimeError("database unavailable")
def get_latest_analysis_by_query_id(self, query_id: str):
raise RuntimeError("database unavailable")
class RunDiagnosticsP2TestCase(unittest.TestCase):
def test_news_diagnostics_use_retrieval_evidence_not_model_summary(self) -> None:
diagnostics = _diagnostic_snapshot()
diagnostics["provider_runs"] = [
{
"trace_id": "trace-p2",
"data_type": "realtime_quote",
"provider": "QuoteFetcher",
"operation": "get_realtime_quote",
"success": True,
},
{
"trace_id": "trace-p2",
"data_type": "daily_data",
"provider": "DailyFetcher",
"operation": "get_daily_data",
"success": True,
"record_count": 30,
},
]
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": None,
},
raw_result={
"success": True,
"model_used": "deepseek-chat",
"analysis_summary": "测试摘要",
"news_summary": "模型生成的新闻摘要",
},
report_saved=True,
)
self.assertEqual(summary["components"]["news"]["status"], "unknown")
self.assertEqual(summary["status"], "normal")
def test_news_summary_string_is_not_treated_as_retrieval_evidence(self) -> None:
diagnostics = _diagnostic_snapshot()
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": "模型生成的新闻摘要",
},
raw_result={
"success": True,
"model_used": "deepseek-chat",
"analysis_summary": "测试摘要",
"news_summary": "模型生成的新闻摘要",
},
report_saved=True,
)
self.assertEqual(summary["components"]["news"]["status"], "unknown")
def test_news_result_count_zero_is_degraded_even_with_formatted_text(self) -> None:
diagnostics = _diagnostic_snapshot()
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": "【贵州茅台 情报搜索结果】\n 未找到相关信息",
"news_result_count": 0,
},
raw_result={
"success": True,
"model_used": "deepseek-chat",
"analysis_summary": "测试摘要",
"news_summary": "模型生成的新闻摘要",
},
report_saved=True,
)
self.assertEqual(summary["components"]["news"]["status"], "degraded")
self.assertEqual(summary["components"]["news"]["details"]["record_count"], 0)
def test_summary_classifies_provider_fallback_as_degraded_and_copy_text_is_sanitized(self) -> None:
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": _diagnostic_snapshot(),
"news_content": "新闻摘要",
},
raw_result={
"success": True,
"model_used": "deepseek-chat",
"analysis_summary": "测试摘要",
},
report_saved=True,
)
self.assertEqual(summary["status"], "degraded")
self.assertEqual(summary["status_label"], "部分降级")
self.assertEqual(summary["components"]["realtime_quote"]["status"], "degraded")
self.assertEqual(summary["components"]["daily_data"]["status"], "ok")
self.assertEqual(summary["components"]["llm"]["status"], "ok")
self.assertEqual(summary["components"]["notification"]["status"], "ok")
self.assertIn("trace_id: trace-p2", summary["copy_text"])
self.assertNotIn("secret", summary["copy_text"])
def test_summary_marks_llm_failure_as_failed(self) -> None:
diagnostics = _diagnostic_snapshot()
diagnostics["llm_runs"] = [
{
"trace_id": "trace-p2",
"model": "deepseek-chat",
"success": False,
"error_type": "RuntimeError",
"error_message_sanitized": "api_key=<redacted>",
}
]
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": "新闻摘要",
},
raw_result={"success": False, "error_message": "api_key=secret-value"},
report_saved=True,
)
self.assertEqual(summary["status"], "failed")
self.assertEqual(summary["components"]["llm"]["status"], "failed")
self.assertIn("LLM 失败", summary["reason"])
self.assertNotIn("secret-value", summary["copy_text"])
def test_copy_text_redacts_authorization_bearer_tokens(self) -> None:
diagnostics = _diagnostic_snapshot()
diagnostics["llm_runs"] = [
{
"trace_id": "trace-p2",
"model": "deepseek-chat",
"success": False,
"error_type": "Unauthorized",
"error_message_sanitized": (
"request failed Authorization: Bearer sk-live-token-abc123"
),
}
]
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": "新闻摘要",
},
raw_result={
"success": False,
"error_message": "Authorization: Bearer sk-raw-token-xyz789",
},
report_saved=True,
)
self.assertEqual(summary["status"], "failed")
self.assertIn("authorization=<redacted>", summary["copy_text"].lower())
self.assertNotIn("sk-live-token-abc123", summary["copy_text"])
self.assertNotIn("sk-raw-token-xyz789", summary["copy_text"])
self.assertNotIn("Bearer sk-", summary["copy_text"])
def test_copy_text_redacts_env_json_and_proxy_credentials(self) -> None:
diagnostics = _diagnostic_snapshot()
diagnostics["llm_runs"] = [
{
"trace_id": "trace-p2",
"model": "deepseek-chat",
"success": False,
"error_type": "ProxyError",
"error_message_sanitized": (
"OPENAI_API_KEY=sk-env-secret "
"\"api_key\": \"sk-json-secret\" "
"proxy http://proxy_user:proxy_pass@proxy.example.com"
),
}
]
summary = build_run_diagnostic_summary(
context_snapshot={
"diagnostics": diagnostics,
"news_content": "news summary",
},
raw_result={
"success": False,
"error_message": (
"DEEPSEEK_API_KEY=sk-raw-secret "
"'access_token': 'raw-token-secret' "
"http://raw_user:raw_pass@proxy.internal"
),
},
report_saved=True,
)
copy_text = summary["copy_text"]
self.assertIn("OPENAI_API_KEY=<redacted>", copy_text)
self.assertIn("\"api_key\": \"<redacted>\"", copy_text)
self.assertIn("http://<redacted>:<redacted>@proxy.example.com", copy_text)
for leaked in (
"sk-env-secret",
"sk-json-secret",
"proxy_user",
"proxy_pass",
):
self.assertNotIn(leaked, copy_text)
def test_sanitize_diagnostic_text_redacts_common_secret_shapes(self) -> None:
text = (
"OPENAI_API_KEY=sk-env-secret "
"\"api_key\": \"sk-json-secret\" "
"'access_token': 'raw-token-secret' "
"http://proxy_user:proxy_pass@proxy.example.com "
"Authorization: Bearer sk-auth-secret"
)
sanitized = sanitize_diagnostic_text(text)
self.assertIsNotNone(sanitized)
self.assertIn("OPENAI_API_KEY=<redacted>", sanitized)
self.assertIn("\"api_key\": \"<redacted>\"", sanitized)
self.assertIn("'access_token': '<redacted>'", sanitized)
self.assertIn("http://<redacted>:<redacted>@proxy.example.com", sanitized)
self.assertIn("Authorization=<redacted>", sanitized)
for leaked in (
"sk-env-secret",
"sk-json-secret",
"sk-raw-secret",
"raw-token-secret",
"proxy_user",
"proxy_pass",
"sk-auth-secret",
):
self.assertNotIn(leaked, sanitized)
def test_legacy_report_without_diagnostics_returns_unknown(self) -> None:
summary = build_run_diagnostic_summary(
context_snapshot={"news_content": "legacy news"},
raw_result={"success": True, "model_used": "deepseek-chat"},
report_saved=True,
query_id="legacy-query",
stock_code="600519",
)
self.assertEqual(summary["status"], "unknown")
self.assertEqual(summary["status_label"], "未知")
self.assertEqual(summary["query_id"], "legacy-query")
def test_history_service_and_endpoint_return_diagnostic_summary(self) -> None:
context_snapshot = {
"diagnostics": _diagnostic_snapshot(),
"news_content": "新闻摘要",
}
db = _FakeHistoryDb(_history_record(context_snapshot=context_snapshot))
service_summary = HistoryService(db).resolve_and_get_diagnostics("1")
endpoint_summary = get_history_diagnostics("1", db_manager=db)
self.assertIsNotNone(service_summary)
self.assertEqual(service_summary["trace_id"], "trace-p2")
self.assertEqual(endpoint_summary.trace_id, "trace-p2")
self.assertIn("realtime_quote", endpoint_summary.components)
def test_history_service_returns_unknown_for_legacy_record(self) -> None:
db = _FakeHistoryDb(_history_record(context_snapshot=None))
summary = HistoryService(db).resolve_and_get_diagnostics("1")
self.assertIsNotNone(summary)
self.assertEqual(summary["status"], "unknown")
self.assertIn("copy_text", summary)
def test_history_diagnostics_endpoint_surfaces_lookup_errors(self) -> None:
with self.assertRaises(HTTPException) as ctx:
get_history_diagnostics("1", db_manager=_FailingHistoryDb())
self.assertEqual(ctx.exception.status_code, 500)
def test_history_diagnostics_endpoint_surfaces_malformed_payloads(self) -> None:
record = _history_record(context_snapshot=None)
record.context_snapshot = "{invalid-json"
db = _FakeHistoryDb(record)
with self.assertRaises(ValueError):
HistoryService(db).resolve_and_get_diagnostics("1")
with self.assertRaises(HTTPException) as ctx:
get_history_diagnostics("1", db_manager=db)
self.assertEqual(ctx.exception.status_code, 500)
if __name__ == "__main__":
unittest.main()