diff --git a/.gitignore b/.gitignore index 68e1c08..410b088 100644 --- a/.gitignore +++ b/.gitignore @@ -41,6 +41,11 @@ reasonix.toml .env .env.local .env.*.local +# 迁移/切换服务商时留下的 .env 备份(含密钥,禁止入库) +.env* +!.env.example +*.bak +*.bak* *.key *.pem diff --git a/configs/llm_models.yaml b/configs/llm_models.yaml index 0957755..4e21c50 100644 --- a/configs/llm_models.yaml +++ b/configs/llm_models.yaml @@ -18,7 +18,8 @@ # LLM_MODEL);若全部缺失则直接报错,绝不静默使用内置默认模型。 # · api_key_env / base_url_env 为可选字段,填写存放 API Key / 服务地址的 # 环境变量名;API Key 一律放 .env,禁止写入本文件(安全规范)。 -# · 修改后无需重启常驻服务即可生效(每次调用重新读取;如需热更新缓存可重启)。 +# · 修改后无需重启常驻服务即可生效:configs/loader.py 以 (mtime, size) 失效缓存, +# 保存后下一次调用即读到新值;.env 的改动由 configs/runtime_env.py 在约 2s 内热更新。 # ============================================================================= # ---- 全局默认参数(各场景可覆盖;低于 .env,高于代码内置默认)---- @@ -49,9 +50,9 @@ scenes: # 建议模型: deepseek-v4-flash(生产实测) / deepseek-chat / qwen-plus / qwen-max event_extraction: provider: qwen # 建议 deepseek | qwen;留空则回退 .env 的 LLM_PROVIDER - model: qwen3.7-flash # 留空则回退 .env(DEEPSEEK_MODEL → LLM_MODEL) - api_key_env: DASHSCOPE_API_KEY # 例如: DEEPSEEK_API_KEY / QWEN_API_KEY / DASHSCOPE_API_KEY - base_url_env: QWEN_BASE_URL # 例如: DEEPSEEK_BASE_URL / QWEN_BASE_URL + model: qwen3.6-flash # Token Plan 模型;注意 Token Plan 无 qwen3.7-flash + api_key_env: QWEN_API_KEY # Token Plan 计费账号(sk-sp-…);勿用 DASHSCOPE_API_KEY + base_url_env: QWEN_BASE_URL # .env 指向 token-plan.*.maas.aliyuncs.com temperature: 0.1 timeout_sec: 60 max_attempts: 3 # 单篇解析失败的最大重试次数 @@ -66,20 +67,23 @@ scenes: # 使用方式:无需手动触发,定时任务自动执行;失败自动降级(日报留空,不影响入库)。 # 对模型的要求: # · OpenAI 兼容 chat 接口(不需要 JSON 输出); - # · 输出长度 ≥ 1500 tokens(max_tokens=1500,输出超长会被截断并记 WARNING); + # · 输出长度 ≥ max_tokens 配置值(见下,输出超长会被截断并记 WARNING); # · 中文摘要能力强、要点化输出稳定(每条一行,以 "- " 开头); # · 上下文窗口 ≥ 8K tokens(素材按 3000 字符/块分块,多块先分段再合并); # · temperature 0.3 左右,兼顾稳定与表达;网络失败按指数退避重试 3 次。 - # · 输出长度需求:分段摘要约 800 tokens、合并摘要约 1500 tokens(代码内置, - # 不在本文件配置),模型应能稳定输出 1500+ tokens 的中文要点。 + # · max_tokens 说明:推理模型(deepseek-v4.1-flash 等)的 reasoning token 与 + # 正文共用该预算;预算过小时"思考"会占满配额导致正文为空 + # (finish_reason=length、0 字符,日报因此没有 AI 摘要)。代码兜底见 + # scheduler/reporter.py: 正文为空时自动加倍预算重试(最多 2 次,上限 16000)。 # 建议模型: deepseek-v4-flash(生产实测) / deepseek-chat / qwen-plus daily_report: - provider: # 建议 deepseek | qwen;留空则回退 .env 的 LLM_PROVIDER - model: - api_key_env: - base_url_env: + provider: qwen # Token Plan 计费账号 + model: deepseek-v4.1-flash + api_key_env: QWEN_API_KEY + base_url_env: QWEN_BASE_URL temperature: 0.3 timeout_sec: 60 + max_tokens: 4000 # 单块/合并摘要输出预算(需为 reasoning token 预留余量) # ------------------------------------------------------------------------- # # 场景 3: 个股 AI 要点分析 @@ -97,10 +101,10 @@ scenes: # · 输出长度需求:约 500 tokens(代码内置,不在本文件配置)。 # 建议模型: deepseek-v4-flash(生产实测) / deepseek-chat / qwen-plus stock_report: - provider: # 建议 deepseek | qwen;留空则回退 .env 的 LLM_PROVIDER - model: - api_key_env: - base_url_env: + provider: qwen # Token Plan（个股日报当前禁用，配置好以防将来启用时漏计费） + model: qwen3.6-flash + api_key_env: QWEN_API_KEY + base_url_env: QWEN_BASE_URL temperature: 0.3 timeout_sec: 60 diff --git a/configs/loader.py b/configs/loader.py index 3179869..d1e3295 100644 --- a/configs/loader.py +++ b/configs/loader.py @@ -8,33 +8,75 @@ 3. 环境变量 / .env(LLM_PROVIDER、DEEPSEEK_MODEL 等,向后兼容) 4. 代码内置默认值 +热加载: 缓存以 ``(mtime_ns, size)`` 为准 —— 改完 YAML 保存后，下一次读取即生效， +常驻进程（调度器 / MCP server）无需重启。 + 说明:API Key 一律放 .env,本文件只保存环境变量名(api_key_env),禁止写密钥。 """ from __future__ import annotations -from functools import lru_cache +import os +import threading from pathlib import Path from loguru import logger -DEFAULT_CONFIG_PATH = Path("configs/llm_models.yaml") +#: 指定替代的模型配置文件路径（测试 / 多环境部署用） +MODELS_CONFIG_OVERRIDE = "A_SHARE_MODELS_CONFIG" + +#: 默认模型配置文件（绝对路径，不依赖当前工作目录） +DEFAULT_CONFIG_PATH = Path(__file__).resolve().parents[1] / "configs" / "llm_models.yaml" + +_lock = threading.Lock() +_cache: dict[Path, tuple[tuple[int, int] | None, dict]] = {} + + +def config_path() -> Path: + """返回当前使用的 ``llm_models.yaml`` 路径。""" + override = os.environ.get(MODELS_CONFIG_OVERRIDE) + if override: + return Path(override).expanduser() + return DEFAULT_CONFIG_PATH + + +def _signature(path: Path) -> tuple[int, int] | None: + """返回 ``(mtime_ns, size)``；文件不存在时返回 None。""" + try: + st = path.stat() + except OSError: + return None + return (st.st_mtime_ns, st.st_size) -@lru_cache(maxsize=8) def _load_yaml(path: Path) -> dict: - """读取 YAML 文件为 dict;文件缺失或解析失败返回空 dict(走兜底配置)。""" + """读取 YAML 为 dict;文件缺失或解析失败返回空 dict(走兜底配置)。 + + 按 ``(mtime_ns, size)`` 失效缓存:文件一旦变化，下次调用即重新解析。 + """ + sig = _signature(path) + + with _lock: + cached = _cache.get(path) + if cached is not None and cached[0] == sig: + return cached[1] + if not path.is_file(): logger.debug("配置文件不存在,使用内置/环境变量兜底: {}", path) - return {} - try: - import yaml + data: dict = {} + else: + try: + import yaml - data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} - except Exception as e: # noqa: BLE001 - YAML 语法错误等 - logger.error("解析 {} 失败: {}", path, e) - return {} - return data if isinstance(data, dict) else {} + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + data = raw if isinstance(raw, dict) else {} + except Exception as e: # noqa: BLE001 - YAML 语法错误等 + logger.error("解析 {} 失败: {}", path, e) + data = {} + + with _lock: + _cache[path] = (sig, data) + return data def load_scene_config(scene: str) -> dict: @@ -45,7 +87,7 @@ def load_scene_config(scene: str) -> dict: """ if not scene: return {} - data = _load_yaml(DEFAULT_CONFIG_PATH) + data = _load_yaml(config_path()) scenes = data.get("scenes") or {} cfg = scenes.get(scene) if cfg is None: @@ -59,11 +101,12 @@ def load_scene_config(scene: str) -> dict: def load_defaults() -> dict: """读取 llm_models.yaml 顶层 defaults(全局默认参数)。""" - data = _load_yaml(DEFAULT_CONFIG_PATH) + data = _load_yaml(config_path()) d = data.get("defaults") or {} return d if isinstance(d, dict) else {} def clear_cache() -> None: - """清空 YAML 缓存(测试或热更新配置时使用)。""" - _load_yaml.cache_clear() + """清空 YAML 缓存(测试或强制重载时使用；正常热加载无需调用)。""" + with _lock: + _cache.clear() diff --git a/configs/runtime_env.py b/configs/runtime_env.py new file mode 100644 index 0000000..deab3e5 --- /dev/null +++ b/configs/runtime_env.py @@ -0,0 +1,188 @@ +"""运行期配置热加载：改 ``.env`` 后无需重启进程即生效。 + +为什么需要它 +------------ +常驻进程（``scripts/run_scheduler.py``、``mcp_server``）启动时把 ``.env`` 读进 +``os.environ``，之后再改 ``.env`` 不会生效——子进程虽然会 ``load_dotenv()``，但 +它继承的是父进程那份旧环境，而 python-dotenv 默认不覆盖已存在的键，于是 +"改了配置却没反应"。 + +做法 +---- +- :func:`ensure_env_loaded`：先比对 ``.env`` 的 ``(mtime_ns, size)``。文件没变时 + 只做一次 ``stat``；变了才重新解析并同步到 ``os.environ``。 +- 删除语义：上一轮由 ``.env`` 带入、这一轮已从文件里删掉的键会被清除， + 保证"文件即事实源"，而不是只能加不能减。 +- :func:`env_get`：先热加载再读取，供各模块统一取配置（空字符串视为未设置）。 +- :func:`start_env_watcher`：守护线程周期性刷新，照顾那些仍直接读 + ``os.environ`` 的历史代码路径。 + +优先级（从高到低） +------------------ + 显式参数 / CLI > configs/llm_models.yaml 场景 > 进程环境（shell / systemd） + > .env 文件 > 内置默认值 + +其中「进程环境」与「.env」的关系是： + +- 进程环境里**显式设置且与文件不同**的键优先，热加载不会覆盖它 + （例如 ``LLM_PROVIDER=qwen python -m a_share_cli`` 这种一次性覆盖）； +- 其余键由本模块托管，跟随 ``.env`` 文件变化即时更新； +- 从 ``.env`` 里删掉的托管键，会同步从进程环境移除。 + +测试或部署可用 ``A_SHARE_ENV_FILE`` 指定其它 ``.env`` 路径。 +""" + +from __future__ import annotations + +import os +import threading +import time +from pathlib import Path + +from dotenv import dotenv_values +from loguru import logger + +#: 指定替代的 .env 路径（测试 / 多环境部署用） +ENV_FILE_OVERRIDE = "A_SHARE_ENV_FILE" + +#: watcher 轮询间隔（秒） +WATCH_INTERVAL_SEC = 2.0 + +_lock = threading.Lock() +_signature: tuple[int, int] | None = None +#: 由本模块写入 os.environ 的键 -> 写入值；用于识别"外部显式覆盖" +_managed: dict[str, str] = {} + + +def dotenv_path() -> Path: + """返回当前使用的 ``.env`` 路径。""" + override = os.environ.get(ENV_FILE_OVERRIDE) + if override: + return Path(override).expanduser() + return Path(__file__).resolve().parents[1] / ".env" + + +def file_signature(path: Path) -> tuple[int, int] | None: + """返回 ``(mtime_ns, size)``；文件不存在时返回 None。""" + try: + st = path.stat() + except OSError: + return None + return (st.st_mtime_ns, st.st_size) + + +def _release_all() -> None: + """撤下所有仍由本模块托管的键（.env 消失时）。""" + for key, managed_value in list(_managed.items()): + if os.environ.get(key) == managed_value: + os.environ.pop(key, None) + del _managed[key] + + +def _apply(values: dict[str, str]) -> None: + """把文件值同步到 ``os.environ``，尊重外部显式覆盖。""" + # 1) 已从文件移除的托管键 → 同步删除 + for key in list(_managed): + if key in values: + continue + if os.environ.get(key) == _managed[key]: + os.environ.pop(key, None) + del _managed[key] + + # 2) 应用文件中的键 + for key, value in values.items(): + current = os.environ.get(key) + if current is None or current == value: + # 未设置，或与文件一致 → 交给文件托管（后续可热更新） + os.environ[key] = value + _managed[key] = value + elif _managed.get(key) == current: + # 当前值正是本模块上一轮写入的 → 跟随文件热更新 + os.environ[key] = value + _managed[key] = value + else: + # 进程环境里显式设置且与文件不同 → 外部优先，不接管 + _managed.pop(key, None) + + +def ensure_env_loaded(force: bool = False) -> bool: + """确保 ``os.environ`` 与 ``.env`` 文件一致。 + + Args: + force: 忽略签名缓存，强制重新解析（首次加载 / 测试用）。 + + Returns: + 本次是否真的重新加载了文件。 + """ + global _signature + + path = dotenv_path() + sig = file_signature(path) + + with _lock: + if not force and sig == _signature: + return False + + if sig is None: + removed = len(_managed) + _release_all() + _signature = None + if removed: + logger.warning("{} 不可读，已回退 {} 个环境变量", path, removed) + return True + + values = {k: v for k, v in dotenv_values(path).items() if v is not None} + _apply(values) + _signature = sig + + logger.debug("已加载/热更新 {}（{} 项）", path, len(values)) + return True + + +def env_get(key: str, default: str | None = None) -> str | None: + """读取配置项（读取前自动热加载 ``.env``）；空字符串视为未设置。""" + ensure_env_loaded() + value = os.environ.get(key) + if value is None or value.strip() == "": + return default + return value.strip() + + +def env_raw(key: str, default: str | None = None) -> str | None: + """读取配置项原始值（读取前自动热加载 ``.env``）。 + + 与 :func:`env_get` 的区别：**不把空字符串当作未设置**。 + 用于"显式留空表示禁用"这类开关，例如 ``STOCK_REPORT_TIME=``。 + """ + ensure_env_loaded() + value = os.environ.get(key) + if value is None: + return default + return value.strip() + + +def start_env_watcher(interval: float = WATCH_INTERVAL_SEC) -> threading.Thread: + """启动守护线程：周期性检查 ``.env``，变了就热更新 ``os.environ``。 + + 只对**常驻进程**有意义；短命的一次性脚本按需读取即可。 + """ + def _loop() -> None: + while True: + try: + ensure_env_loaded() + except Exception: # noqa: BLE001 - 热加载失败不应拖垮主进程 + logger.exception("热加载 .env 失败") + time.sleep(interval) + + thread = threading.Thread(target=_loop, name="env-watcher", daemon=True) + thread.start() + logger.info("已启动 .env 热加载监听（每 {:.0f}s 检查一次）", interval) + return thread + + +def reset_cache() -> None: + """仅供测试：撤下托管键并清空签名缓存。""" + global _signature + with _lock: + _release_all() + _signature = None diff --git a/continuation.md b/continuation.md index 4e1dd54..bc03b83 100644 --- a/continuation.md +++ b/continuation.md @@ -4,6 +4,44 @@ --- +## 本次完成 (2026-09-25) — 日报 AI 摘要为空修复（推理模型 reasoning 占满 max_tokens） + +**现象**:用户反馈 2026-09-25 日报没有 AI 摘要。`news_report` 中 `id=357`(2026-09-25, finance) `ai_summary` 为 `NULL`;当天 07:13:55 日志: + +``` +WARNING | scheduler.reporter:_llm_call - AI 摘要可能被截断: max_tokens=1500 finish_reason=length 实际输出 0 字符 +``` + +**根因(证据链闭合)**: +- 日报场景(daily_report)调用 `deepseek-v4.1-flash`(Token Plan),这是**推理模型**:`reasoning_content` 的 token 与正文**共用** `max_tokens` 预算 +- 用当天真实素材原样复现:**`reasoning_tokens=1500` / `text_tokens=0` / `content=0 字符` / `finish_reason=length`** —— 预算被"思考"全部吃掉,正文为空 +- 同素材把预算提到 4000:`finish_reason=stop`、reasoning 938 + text 337、摘要 523 字 ✓ +- 代码缺陷:`scheduler/reporter.py:_llm_call` 只取 `message.content`,空内容**不抛异常** → `_generate_ai_summary` 返回 `""` → `ai_summary=None` 入库,pipeline 仍标 report ✅(**静默失败、无重试**) +- **非本次 Token Plan 迁移引入**:历史同为空的还有 9-12 / 9-13 / 9-15 / 9-18,当时用的是 `deepseek-v4-flash`(provider=deepseek),同样是推理类模型 → 长期间歇性缺陷 + +**修复(方案 B:配置化 + 代码兜底)**: +1. `configs/llm_models.yaml`:`daily_report` 新增 `max_tokens: 4000`(注释说明 reasoning 共用预算);YAML 保存即热生效 +2. `llm/client.py`:`LLMConfig` 新增可选字段 `max_tokens`;`load_llm_config` 用新增的 `_pick_optional_int` 读取场景配置(未配置 = `None`,调用方回退内置默认) +3. `scheduler/reporter.py`: + - 新增常量 `DEFAULT_SUMMARY_MAX_TOKENS=4000` / `DEFAULT_SUMMARY_CHUNK_MAX_TOKENS=2000` / `MAX_SUMMARY_MAX_TOKENS=16000` / `_MAX_BUDGET_ESCALATIONS=2` + - 新增 `_summary_max_tokens()`(场景配置 > 内置默认)、`_chunk_max_tokens()`(不超过合并预算) + - `_llm_call` 拆出 `_call_once`;保持网络异常指数退避重试语义不变;新增**空正文 + finish_reason=length 时自动加倍预算重试**(上限 16000),用尽后返回空串降级(不抛异常) + - 分块预算 800 → 2000;合并预算 1500 → 配置值(4000) + +**验证**: +- 线上复现 → 修复后回归:`uv run a-share report --date 20260925` 重跑,**无截断告警**,`report_id=357` 原地更新(幂等 upsert),`ai_summary` 466 字 ✓ +- 测试:新增 `TestReasoningBudgetEscalation`(升级恢复 / 场景值优先 / 用尽降级返回空 / 上限)+ `test_llm.py` 的 `max_tokens` 场景配置、`_pick_optional_int`、真实 YAML 预算 ≥4000 回归保护 → `tests/test_report_builder.py tests/test_llm.py` **56 passed** +- `ruff` 改动文件无新增问题(5 条 N806/SIM115 为 reporter.py 既有);`mypy` 仅剩 `llm/client.py:86` 既有告警 + +**运维动作**:`report` 步骤由调度器**进程内**执行,已 `sudo systemctl restart a-share-research`(10:55,重启后任务同步正常),明早 07:00 起生效。 + +**遗留与后续**: +- `scheduler/stock_reporter.py:285` 个股日报 `max_tokens=500`,配置模型 `qwen3.6-flash` 亦属推理类,**同类隐患**(当前个股日报禁用);启用前建议一并按本方案处理 +- 空摘要目前只降级为"无摘要",未做告警;可考虑连续 N 天为空时推送通知 +- 本次改动尚未 git commit(工作区还混有 9-22 Token Plan 迁移的未提交改动,避免混提) + +--- + ## 本次完成 (2026-09-10) — cninfo 抓取压穿内存导致整机冻结的修复 **现象**:2026-09-06 / 09-08 / 09-10 连续三次早上 06:0x 整机冻结,看门狗(硬件 2min)硬复位。 diff --git a/docs/architecture.md b/docs/architecture.md index 5d3ccbc..53a69c1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -543,9 +543,15 @@ ReportData (Pydantic) | 文件 | 格式 | 用途 | 热更新 | |------|------|------|--------| | `configs/sources.yaml` | YAML | 14 个新闻源配置 | 每次抓取重读 | -| `configs/llm_models.yaml` | YAML | 4 个 LLM 场景配置 | 每次调用重读 | +| `configs/llm_models.yaml` | YAML | 4 个 LLM 场景配置 | 每次调用重读（mtime 缓存失效） | | `configs/watchlist.yaml` | YAML | cninfo 公告关注列表 | 每次操作重读 | -| `.env` | dotenv | API Key + 调度/超时/DB 配置 | 重启服务生效 | +| `.env` | dotenv | API Key + 调度/超时/DB 配置 | ~2s 内自动生效（常驻进程无需重启） | + +> 热加载实现见 `configs/runtime_env.py`：常驻进程（调度器 / MCP server）启动后 +> 会以 2s 周期比对 `.env` 的 `(mtime, size)`，变化即写入 `os.environ`； +> `configs/loader.py` 对 YAML 做同样的 mtime 失效。 +> 进程环境里**显式设置且与文件不同**的变量优先（如 `LLM_PROVIDER=qwen ...`）， +> `.env` 中删除的键也会同步从环境中移除。 ### 5.2 配置优先级(LLM 场景) diff --git a/docs/user-guide.md b/docs/user-guide.md index 2b1c08e..32f552e 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -507,17 +507,25 @@ tail -f logs/scheduler.log # 文件日志 | 18:00 | crawler→xwlb→extractor→dedup→llm→embedding→qdrant | | 22:00 | crawler→xwlb→extractor→dedup→llm→embedding→qdrant | -### 7.4 修改调度时间 +### 7.4 修改调度时间（无需重启） ```bash # 编辑 .env 中的 SCHEDULE_TIMES,格式: HH:MM,HH:MM,... nano /home/pi/news/.env # 例: SCHEDULE_TIMES=08:00,14:00,20:00 -# 重启生效 -sudo systemctl restart a-share-research +# 保存即生效:常驻调度器每 30s 比对一次,自动重新注册定时任务 +# 日志确认:grep "定时任务已同步" /home/pi/news/logs/scheduler.log ``` +> **配置热加载**：`.env`（模型 / Key / 端点 / 超时 / DB / 调度时间）与 +> `configs/llm_models.yaml`（各场景 provider / model / 温度）都是**改完保存即生效**， +> 不需要 `systemctl restart`。调度时间最长 30s 生效，其余配置约 2s 生效。 +> +> 只有两种情况需要重启： +> 1. 部署/更新了 Python 代码本身； +> 2. 正在执行中的那一步（子进程）会继续用旧配置跑完，下一步才用新配置。 + ### 7.5 前台守护模式(调试) ```bash @@ -703,6 +711,16 @@ uv run a-share stock-report 检查 `.env` 中 `DASHSCOPE_API_KEY` 是否填写。可用 `--provider qwen` 切换到百炼测试。模型名缺失时直接报错,检查 `configs/llm_models.yaml` 中 `event_extraction` 场景的 `model` 字段。 +**Q: 日报没有 AI 摘要(ai_summary 为空)？** + +先查 `logs/scheduler.log` 是否有 `AI 摘要可能被截断: ... finish_reason=length 实际输出 0 字符`。根因通常是**推理模型的 reasoning token 与正文共用 `max_tokens`**:预算过小时"思考"占满配额,正文一个字都没有。处理办法: + +1. 调大 `configs/llm_models.yaml` 中 `daily_report.max_tokens`(默认 4000,YAML 保存即热生效,无需重启); +2. 代码已内置兜底:正文为空时自动加倍预算重试(上限 16000),仍失败才降级为无摘要; +3. 补生成某天摘要:`uv run a-share report --date `(按 `(report_date, report_type, file_name)` 幂等 upsert,不会新增记录)。 + +注意:`report` 步骤由调度器**进程内**执行(`scheduler/pipeline.py`),改动 Python 代码后需 `sudo systemctl restart a-share-research` 才会生效;只改 YAML / `.env` 则无需重启。 + **Q: Qdrant 搜索不到结果？** ```bash diff --git a/embedding/factory.py b/embedding/factory.py index 10f7c8f..f53a750 100644 --- a/embedding/factory.py +++ b/embedding/factory.py @@ -5,9 +5,8 @@ from __future__ import annotations -import os - from configs.loader import load_scene_config +from configs.runtime_env import env_get from .base import AsyncEmbeddingProvider, EmbeddingProvider from .models import EmbeddingError, EmbeddingProviderType @@ -18,10 +17,8 @@ from .remote import ( def _read_env(key: str, default: str | None = None) -> str | None: - val = os.environ.get(key) - if val is None or val.strip() == "": - return default - return val.strip() + """读取环境变量（先热加载 .env，改文件后无需重启进程）。""" + return env_get(key, default) def resolve_provider_type(provider: str | None = None) -> EmbeddingProviderType: diff --git a/embedding/local.py b/embedding/local.py index 4aff44f..9ceaff2 100644 --- a/embedding/local.py +++ b/embedding/local.py @@ -10,7 +10,6 @@ from __future__ import annotations import asyncio -import os from typing import TYPE_CHECKING from loguru import logger @@ -22,16 +21,15 @@ if TYPE_CHECKING: from sentence_transformers import SentenceTransformer from configs.loader import load_scene_config +from configs.runtime_env import env_get LOCAL_DEFAULT_MODEL = "BAAI/bge-m3" LOCAL_DEFAULT_DIM = 1024 def _read_env(key: str, default: str | None = None) -> str | None: - val = os.environ.get(key) - if val is None or val.strip() == "": - return default - return val.strip() + """读取环境变量（先热加载 .env，改文件后无需重启进程）。""" + return env_get(key, default) def _try_import_st() -> type[SentenceTransformer]: diff --git a/embedding/remote.py b/embedding/remote.py index 36d1b6e..47442b5 100644 --- a/embedding/remote.py +++ b/embedding/remote.py @@ -18,12 +18,12 @@ from __future__ import annotations import asyncio -import os from loguru import logger from openai import AsyncOpenAI, OpenAI from configs.loader import load_scene_config +from configs.runtime_env import env_get from .base import AsyncEmbeddingProvider, EmbeddingProvider from .models import EmbeddingError @@ -43,10 +43,8 @@ SCENE_EMBEDDING = "embedding" def _read_env(key: str, default: str | None = None) -> str | None: - val = os.environ.get(key) - if val is None or val.strip() == "": - return default - return val.strip() + """读取环境变量（先热加载 .env，改文件后无需重启进程）。""" + return env_get(key, default) def _scene() -> dict: diff --git a/llm/client.py b/llm/client.py index 7f60ec5..f1a3ba8 100644 --- a/llm/client.py +++ b/llm/client.py @@ -19,13 +19,13 @@ from __future__ import annotations -import os from dataclasses import dataclass from loguru import logger from openai import AsyncOpenAI, OpenAI from configs.loader import load_defaults, load_scene_config +from configs.runtime_env import env_get # 默认基址 _DEEPSEEK_DEFAULT_BASE = "https://api.deepseek.com" @@ -53,6 +53,9 @@ class LLMConfig: timeout_sec: float = DEFAULT_TIMEOUT_SEC temperature: float = DEFAULT_TEMPERATURE max_attempts: int = DEFAULT_MAX_ATTEMPTS # 单次任务失败重试次数 + # 单次输出预算(可选,场景配置 scenes..max_tokens);None = 调用方用内置默认。 + # 注意:推理模型(deepseek-v4.1-flash 等)的 reasoning token 与正文共用该预算。 + max_tokens: int | None = None def __post_init__(self) -> None: if not self.api_key: @@ -60,10 +63,8 @@ class LLMConfig: def _read_env(key: str, default: str | None = None) -> str | None: - val = os.environ.get(key) - if val is None or val.strip() == "": - return default - return val.strip() + """读取环境变量（先热加载 .env，改文件后无需重启进程）。""" + return env_get(key, default) def _first_env(keys: list[str | None]) -> str | None: @@ -147,6 +148,7 @@ def load_llm_config( timeout = _pick_float(sc, dflt, "timeout_sec", "LLM_TIMEOUT_SEC", DEFAULT_TIMEOUT_SEC) temperature = _pick_float(sc, dflt, "temperature", "LLM_TEMPERATURE", DEFAULT_TEMPERATURE) max_attempts = _pick_int(sc, "max_attempts", DEFAULT_MAX_ATTEMPTS) + max_tokens = _pick_optional_int(sc, "max_tokens") return LLMConfig( provider=p, @@ -156,6 +158,7 @@ def load_llm_config( timeout_sec=timeout, temperature=temperature, max_attempts=max_attempts, + max_tokens=max_tokens, ) @@ -182,6 +185,12 @@ def _pick_int(sc: dict, sc_key: str, default: int) -> int: return int(v) if v is not None else default +def _pick_optional_int(sc: dict, sc_key: str) -> int | None: + """可选整数场景配置;未配置或非法时返回 None(调用方回退各自内置默认)。""" + v = _num(sc.get(sc_key)) + return int(v) if v is not None else None + + def make_sync_client(config: LLMConfig) -> OpenAI: """构造同步 OpenAI 客户端(指向 DeepSeek/Qwen 兼容端点)。""" logger.debug( diff --git a/mcp_server/tools.py b/mcp_server/tools.py index cc8b7a7..dadfb62 100644 --- a/mcp_server/tools.py +++ b/mcp_server/tools.py @@ -9,15 +9,22 @@ from __future__ import annotations from dataclasses import dataclass from typing import Any -from dotenv import load_dotenv from loguru import logger from mcp.server.fastmcp import FastMCP +from configs.loader import config_path +from configs.runtime_env import ( + dotenv_path, + ensure_env_loaded, + file_signature, + start_env_watcher, +) from embedding import make_sync_provider from vectorstore import SearchFilter, VectorStore, make_qdrant_client -# 加载 .env(API key 等) -load_dotenv() +# 加载 .env 并开启热加载：MCP server 是常驻进程，改配置无需重启 +ensure_env_loaded() +start_env_watcher() # --------------------------------------------------------------------------- # # 单例(模块加载时初始化,所有工具共用) @@ -29,17 +36,27 @@ class _Backend: vector_store: VectorStore _backend: _Backend | None = None +_backend_sig: tuple[Any, ...] | None = None + + +def _config_signature() -> tuple[Any, ...]: + """(.env, llm_models.yaml) 的 mtime 签名；变化即表示需要重建后端。""" + return (file_signature(dotenv_path()), file_signature(config_path())) def _get_backend() -> _Backend: - global _backend - if _backend is None: + global _backend, _backend_sig + sig = _config_signature() + if _backend is None or sig != _backend_sig: + if _backend is not None: + logger.info("检测到配置变化, 重建 MCP 后端") emb = make_sync_provider() # 读取 EMBEDDING_PROVIDER 环境变量 logger.info("MCP embedder 就绪: dim={}", emb.dim) client = make_qdrant_client() store = VectorStore(client) logger.info("MCP vector_store 就绪: count={}", store.count()) _backend = _Backend(embedder=emb, vector_store=store) + _backend_sig = sig return _backend diff --git a/scheduler/pipeline.py b/scheduler/pipeline.py index e40a6f5..9e2ac00 100644 --- a/scheduler/pipeline.py +++ b/scheduler/pipeline.py @@ -12,7 +12,6 @@ from __future__ import annotations import json -import os import subprocess import time from dataclasses import dataclass, field @@ -21,6 +20,8 @@ from pathlib import Path from loguru import logger +from configs.runtime_env import env_get, env_raw + from .timeutil import today_str # 断点状态文件(按日期隔离,记录每步骤结果) @@ -159,20 +160,16 @@ def _llm_scene_desc(scene: str) -> str | None: try: from configs.loader import load_scene_config - def _env(key: str) -> str | None: - v = os.environ.get(key) - return v.strip() if v else None - sc = load_scene_config(scene) - p = (sc.get("provider") or _env("LLM_PROVIDER") or "deepseek").lower() + p = (sc.get("provider") or env_get("LLM_PROVIDER") or "deepseek").lower() if p in ("qwen", "dashscope"): p = "qwen" model = sc.get("model") if not model: if p == "qwen": - model = _env("QWEN_MODEL") or _env("LLM_MODEL") + model = env_get("QWEN_MODEL") or env_get("LLM_MODEL") else: - model = _env("DEEPSEEK_MODEL") or _env("LLM_MODEL") + model = env_get("DEEPSEEK_MODEL") or env_get("LLM_MODEL") if not model: return None return f"provider={p}, model={model}" @@ -192,9 +189,9 @@ def _embedding_desc() -> str | None: model = sc.get("model") if not model: if pt == "dashscope": - model = os.environ.get("DASHSCOPE_EMBEDDING_MODEL") or "text-embedding-v3" + model = env_get("DASHSCOPE_EMBEDDING_MODEL") or "text-embedding-v3" else: - model = os.environ.get("LOCAL_EMBEDDING_MODEL") or "BAAI/bge-m3" + model = env_get("LOCAL_EMBEDDING_MODEL") or "BAAI/bge-m3" return f"provider={pt}, model={model}" except Exception as e: # noqa: BLE001 logger.debug("embedding 描述解析失败: {}", e) @@ -281,14 +278,14 @@ def run_step(name: str, date_str: str) -> StepResult: # 2. PIPELINE_STEP_TIMEOUT 环境变量 (全局兜底, 覆盖硬编码) # 3. STEP_TIMEOUTS 硬编码字典 (代码内默认值) # 4. 1800s (最终兜底) - import os specific_key = f"TIMEOUT_{name.upper()}" - if specific_key in os.environ: - timeout = int(os.environ[specific_key]) - elif "PIPELINE_STEP_TIMEOUT" in os.environ: - timeout = int(os.environ["PIPELINE_STEP_TIMEOUT"]) - else: - timeout = STEP_TIMEOUTS.get(name, 1800) + timeout_raw = env_raw(specific_key) or env_raw("PIPELINE_STEP_TIMEOUT") + default_timeout = STEP_TIMEOUTS.get(name, 1800) + try: + timeout = int(timeout_raw) if timeout_raw else default_timeout + except ValueError: + logger.warning("超时配置 {!r} 非法,回退默认 {}s", timeout_raw, default_timeout) + timeout = default_timeout started = datetime.now() logger.info("步骤 {} 开始: {}", name, " ".join(full_cmd)) diff --git a/scheduler/reporter.py b/scheduler/reporter.py index 70d5bb6..e848d5f 100644 --- a/scheduler/reporter.py +++ b/scheduler/reporter.py @@ -10,7 +10,6 @@ from __future__ import annotations import json -import os as _os import re as _re import subprocess import time @@ -22,14 +21,11 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from llm.client import LLMConfig -from dotenv import load_dotenv from loguru import logger +from configs.runtime_env import env_get from report_db.models import EventRow, ReportData # noqa: F401 - 供 _build_report_data 注解使用 -# 确保 .env 已加载（模块级常量依赖环境变量） -load_dotenv() - # --------------------------------------------------------------------------- # # 配置 # --------------------------------------------------------------------------- # @@ -37,14 +33,64 @@ load_dotenv() UPLOAD_HOST = "simon@doorcome.cn" UPLOAD_BASE = "/var/www/html/echart/research" -CNINFO_DAYS_BACK = int(_os.environ.get("STOCK_REPORT_DAYS", "15")) # 与个股日报共用参数, 默认值保持一致 NEWS_DAYS_BACK = 1 # 新闻回溯天数 _MAX_HIGH_EVENTS = 20 -# LLM 摘要调用重试参数（环境变量可覆盖） -_LLM_RETRY_TIMES = int(_os.environ.get("LLM_RETRY_TIMES", "3")) -_LLM_RETRY_BACKOFF_SEC = float(_os.environ.get("LLM_RETRY_BACKOFF_SEC", "2.0")) +# AI 摘要输出预算（token）。 +# 推理模型（deepseek-v4.1-flash 等）的 reasoning token 与正文共用 max_tokens： +# 预算过小时"思考"会占满配额，正文为空（finish_reason=length、0 字符）， +# 日报就会没有 AI 摘要。默认值需为 reasoning 预留余量。 +DEFAULT_SUMMARY_MAX_TOKENS = 4000 # 单块 / 合并摘要默认预算 +DEFAULT_SUMMARY_CHUNK_MAX_TOKENS = 2000 # 分块摘要默认预算 +MAX_SUMMARY_MAX_TOKENS = 16000 # 正文为空时预算升级上限 +_MAX_BUDGET_ESCALATIONS = 2 # 正文为空时最多升级预算次数 + + +def _env_int(key: str, default: int) -> int: + """读取整数环境变量（热加载 .env）；非法值回退默认。""" + try: + return int(env_get(key, str(default)) or default) + except ValueError: + logger.warning("环境变量 {} 不是整数, 回退默认 {}", key, default) + return default + + +def _env_float(key: str, default: float) -> float: + """读取浮点环境变量（热加载 .env）；非法值回退默认。""" + try: + return float(env_get(key, str(default)) or default) + except ValueError: + logger.warning("环境变量 {} 不是数字, 回退默认 {}", key, default) + return default + + +def _cninfo_days_back() -> int: + """公告/调研回溯天数（与个股日报共用 STOCK_REPORT_DAYS）。""" + return _env_int("STOCK_REPORT_DAYS", 15) + + +def _llm_retry_times() -> int: + return _env_int("LLM_RETRY_TIMES", 3) + + +def _llm_retry_backoff_sec() -> float: + return _env_float("LLM_RETRY_BACKOFF_SEC", 2.0) + + +def _summary_max_tokens(config: LLMConfig | None = None) -> int: + """摘要输出预算：场景配置 max_tokens（热更新）> 代码内置默认。""" + value = getattr(config, "max_tokens", None) + try: + return int(value) if value else DEFAULT_SUMMARY_MAX_TOKENS + except (TypeError, ValueError): + logger.warning("场景 max_tokens 非法({!r}), 回退默认 {}", value, DEFAULT_SUMMARY_MAX_TOKENS) + return DEFAULT_SUMMARY_MAX_TOKENS + + +def _chunk_max_tokens(config: LLMConfig | None = None) -> int: + """分块摘要预算：不超过单块/合并预算，避免小块调用过度消耗。""" + return min(_summary_max_tokens(config), DEFAULT_SUMMARY_CHUNK_MAX_TOKENS) # 日报新闻回溯窗口（小时）：07:00 生成当日日报时覆盖昨日全天至今晨的新闻 _NEWS_LOOKBACK_HOURS = 30 @@ -181,12 +227,14 @@ def _collect_news_events(day_str: str) -> dict[str, Any]: } -def _collect_cninfo_events(today_str: str, days_back: int = CNINFO_DAYS_BACK) -> dict[str, Any]: +def _collect_cninfo_events(today_str: str, days_back: int | None = None) -> dict[str, Any]: """收集近 N 日 cninfo 公告/调研/互动（直接从 processed 数据读取，不依赖 M4 事件抽取）。 cninfo 公告/调研数据已结构化（stock_code/name/title/time/type）， 无需经过 LLM 事件抽取即可直接用于日报。 """ + if days_back is None: + days_back = _cninfo_days_back() today = datetime.strptime(today_str, "%Y%m%d") since_str = (today - timedelta(days=days_back)).strftime("%Y-%m-%d") wl_codes = _load_watchlist_codes() @@ -514,7 +562,7 @@ def _generate_ai_summary(news: dict, cninfo: dict, day_str: str, # 公告/调研 if cninfo["high"]: - lines.append(f"## 近 {CNINFO_DAYS_BACK} 日重要公告/调研 ({len(cninfo['high'])} 条)") + lines.append(f"## 近 {_cninfo_days_back()} 日重要公告/调研 ({len(cninfo['high'])} 条)") for e in cninfo["high"][:8]: ev = e.get("event", {}) lines.append(f"- [{ev.get('event_type', '公司公告')}] {e['title']}") @@ -570,7 +618,7 @@ def _llm_summarize(client, config: LLMConfig, lines: list[str], day_str: str) -> {chr(10).join(chunk)} 直接输出要点列表：""" - result = _llm_call(client, config, prompt, max_tokens=800) + result = _llm_call(client, config, prompt, max_tokens=_chunk_max_tokens(config)) if result: partials.append(result) logger.info("AI 摘要: 分块 {}/{} 完成 ({} 字)", i, len(chunks), len(result)) @@ -589,14 +637,14 @@ def _llm_summarize(client, config: LLMConfig, lines: list[str], day_str: str) -> 请合并为要点总结，每条一行以 "- " 开头，要求： 1. 前 3 条为影响最大的事件，说明为什么重要 -2. 汇总近 {CNINFO_DAYS_BACK} 日公司公告/调研核心信息 +2. 汇总近 {_cninfo_days_back()} 日公司公告/调研核心信息 3. 市场情绪基调（利好/利空/中性） 4. 值得持续关注的行业或主题 5. 纯要点，不要开场白/结束语 6. 总字数 500 字以内 直接输出要点列表：""" - return _llm_call(client, config, merge_prompt, max_tokens=1500) + return _llm_call(client, config, merge_prompt, max_tokens=_summary_max_tokens(config)) def _build_prompt(lines: list[str], day_str: str) -> str: @@ -607,7 +655,7 @@ def _build_prompt(lines: list[str], day_str: str) -> str: 请用要点总结，每条一行，以 "- " 开头，要求： 1. 前 3 条为过去 24 小时影响最大的事件（优先参考新闻联播中的重大政策信号），说明为什么重要 -2. 汇总近 {CNINFO_DAYS_BACK} 日重要公司公告/调研的核心信息 +2. 汇总近 {_cninfo_days_back()} 日重要公司公告/调研的核心信息 3. 市场情绪基调（利好/利空/中性） 4. 值得持续关注的行业或主题 5. 纯要点，不要开场白/结束语/标题 @@ -616,45 +664,70 @@ def _build_prompt(lines: list[str], day_str: str) -> str: 直接输出要点列表：""" -def _llm_call(client, config: LLMConfig, prompt: str, max_tokens: int = 1500) -> str: - """单次 LLM 调用（带重试），返回 strip 后的文本。 +def _call_once(client, config: LLMConfig, prompt: str, max_tokens: int) -> tuple[str, str | None]: + """单次 LLM 调用，返回 (正文, finish_reason)。异常由调用方处理。""" + resp = client.chat.completions.create( + model=config.model, + messages=[ + {"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"}, + {"role": "user", "content": prompt}, + ], + temperature=config.temperature, + max_tokens=max_tokens, + ) + choice = resp.choices[0] + content = (choice.message.content or "").strip() + return content, getattr(choice, "finish_reason", None) - config 为 llm.client.LLMConfig(daily_report 场景),提供 model / temperature。 - 失败按指数退避重试 `_LLM_RETRY_TIMES` 次（默认 3），全部失败则抛出最后一次异常。 - 若 finish_reason 为 'length' 则说明达到 max_tokens 上限被截断。 + +def _llm_call(client, config: LLMConfig, prompt: str, max_tokens: int | None = None) -> str: + """单次 LLM 调用（带重试 + 空正文预算升级），返回 strip 后的文本。 + + config 为 llm.client.LLMConfig(daily_report 场景),提供 model / temperature / + max_tokens（未配置 max_tokens 时用 DEFAULT_SUMMARY_MAX_TOKENS）。 + 网络等异常按指数退避重试 `LLM_RETRY_TIMES` 次（默认 3），全部失败则抛出最后一次异常。 + 若 finish_reason 为 'length' 且正文为空（推理模型 reasoning 占满预算）， + 自动加倍预算重试，最多 `_MAX_BUDGET_ESCALATIONS` 次、上限 MAX_SUMMARY_MAX_TOKENS； + 此时返回空字符串而非抛异常，由调用方降级。 """ + retry_times = _llm_retry_times() + backoff_sec = _llm_retry_backoff_sec() + budget = max_tokens or _summary_max_tokens(config) last_exc: Exception | None = None - for attempt in range(_LLM_RETRY_TIMES): + escalations = 0 + + for attempt in range(retry_times): try: - resp = client.chat.completions.create( - model=config.model, - messages=[ - {"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"}, - {"role": "user", "content": prompt}, - ], - temperature=config.temperature, - max_tokens=max_tokens, - ) - content = (resp.choices[0].message.content or "").strip() - finish = getattr(resp.choices[0], "finish_reason", None) - if finish == "length": - logger.warning( - "AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符", - max_tokens, len(content), - ) - return content + content, finish = _call_once(client, config, prompt, budget) except Exception as e: last_exc = e - if attempt < _LLM_RETRY_TIMES - 1: - wait = _LLM_RETRY_BACKOFF_SEC * (2 ** attempt) + if attempt < retry_times - 1: + wait = backoff_sec * (2 ** attempt) logger.warning( "AI 摘要 LLM 调用失败(第 {}/{} 次): {}; {} 秒后重试", - attempt + 1, _LLM_RETRY_TIMES, e, round(wait, 2), + attempt + 1, retry_times, e, round(wait, 2), ) time.sleep(wait) - logger.error("AI 摘要 LLM 调用重试 {} 次仍失败: {}", _LLM_RETRY_TIMES, last_exc) - assert last_exc is not None - raise last_exc + continue + + if finish == "length": + logger.warning( + "AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符", + budget, len(content), + ) + # 正文为空 = 推理占满预算；加倍预算重试（不改动调用方传入的显式预算以外逻辑） + if not content and escalations < _MAX_BUDGET_ESCALATIONS: + budget = min(budget * 2, MAX_SUMMARY_MAX_TOKENS) + escalations += 1 + logger.warning("AI 摘要正文为空(推理占满预算), 提升 max_tokens 至 {} 重试", budget) + continue + return content + + if last_exc is not None: + logger.error("AI 摘要 LLM 调用重试 {} 次仍失败: {}", retry_times, last_exc) + raise last_exc + logger.warning("AI 摘要 {} 次尝试仍未获得正文, 返回空(日报降级为无 AI 摘要)", retry_times) + return "" # --------------------------------------------------------------------------- # @@ -964,7 +1037,7 @@ def _render_html(news: dict, cninfo: dict, pipeline: dict, news_table=news_table, cninfo_high_count=len(cninfo["high"]), cninfo_threshold=cninfo.get("hi_threshold", 4), - cninfo_days=CNINFO_DAYS_BACK, + cninfo_days=_cninfo_days_back(), cninfo_table=cninfo_table, raw_total=pipeline["raw_total"], raw_total_24h=pipeline.get("raw_total_24h", pipeline["raw_total"]), @@ -1065,7 +1138,7 @@ def generate_report(day_str: str | None = None, *, upload: bool = True) -> int | # 收集数据 try: news = _collect_news_events(day_str) - cninfo = _collect_cninfo_events(day_str, days_back=CNINFO_DAYS_BACK) + cninfo = _collect_cninfo_events(day_str, days_back=_cninfo_days_back()) pipeline = _collect_pipeline_stats(day_str) xwlb = _collect_xwlb(day_str) except Exception as e: diff --git a/scripts/run_scheduler.py b/scripts/run_scheduler.py index 29b99ae..564183e 100644 --- a/scripts/run_scheduler.py +++ b/scripts/run_scheduler.py @@ -14,14 +14,15 @@ from __future__ import annotations import argparse +import json import signal import sys from pathlib import Path from typing import Any -from dotenv import load_dotenv from loguru import logger +from configs.runtime_env import ensure_env_loaded, env_raw, start_env_watcher from scheduler import DEFAULT_NEWS_STEPS, run_pipeline from scheduler.stock_reporter import generate_all_stock_reports from scheduler.timeutil import now as tz_now @@ -59,6 +60,131 @@ def _parse_schedule_times(raw: str) -> list[tuple[int, int]]: return out +def _parse_hhmm(raw: str, default: tuple[int, int]) -> tuple[int, int]: + """解析 "HH:MM"；非法或越界时记警告并返回 default。""" + parts = raw.split(":") + try: + hour = int(parts[0]) + minute = int(parts[1]) if len(parts) > 1 else 0 + except (ValueError, IndexError): + logger.warning("时间格式错误: {!r},改用默认 {:02d}:{:02d}", raw, *default) + return default + if not (0 <= hour <= 23 and 0 <= minute <= 59): + logger.warning("时间越界: {!r},改用默认 {:02d}:{:02d}", raw, *default) + return default + return hour, minute + + +def _scheduled_pipeline(steps: list[str] | None = None) -> None: + """定时触发的全链路：每次触发时按调度时区重新计算日期。""" + run_pipeline(today_str(), steps=steps) + + +def _desired_jobs() -> dict[str, dict[str, Any]]: + """按当前环境变量算出「期望的」定时任务集合。 + + 每次调用都通过 env_get 读取，所以改 .env 后无需重启即可反映到 _sync_jobs。 + """ + jobs: dict[str, dict[str, Any]] = {} + + times = _parse_schedule_times( + env_raw("SCHEDULE_TIMES", "07:00,12:00,18:00,22:00") or "" + ) + if times: + first = min(times) + for hour, minute in times: + with_report = (hour, minute) == first + jobs[f"pipeline_{hour:02d}{minute:02d}"] = { + "kind": "pipeline", + "hour": hour, + "minute": minute, + "steps": list(DEFAULT_NEWS_STEPS) + (["report"] if with_report else []), + "name": f"全链路 {'+日报' if with_report else ''} {hour:02d}:{minute:02d}", + } + + cninfo_h, cninfo_m = _parse_hhmm( + env_raw("CNINFO_SCHEDULE_TIME", "06:30") or "06:30", (6, 30) + ) + jobs["pipeline_cninfo"] = { + "kind": "pipeline", + "hour": cninfo_h, + "minute": cninfo_m, + "steps": ["cninfo_crawl", "cninfo_extract", "cninfo_pdf", + "dedup", "llm", "embedding", "qdrant"], + "name": f"cninfo 公告管道 {cninfo_h:02d}:{cninfo_m:02d}", + } + + # STOCK_REPORT_TIME 缺省 07:30；显式留空表示禁用 + stock_raw = env_raw("STOCK_REPORT_TIME") + if stock_raw is None: + stock_raw = "07:30" + if stock_raw: + stock_h, stock_m = _parse_hhmm(stock_raw, (7, 30)) + jobs["stock_report"] = { + "kind": "stock", + "hour": stock_h, + "minute": stock_m, + "name": f"个股日报 {stock_h:02d}:{stock_m:02d}", + } + return jobs + + +_jobs_sig: str | None = None + + +def _sync_jobs(scheduler: Any) -> bool: + """把「期望任务」同步到 APScheduler；只在配置变化时增删。返回是否变更。 + + 以 ``_`` 开头的内部任务（如配置热同步自身）不参与增删。 + """ + global _jobs_sig + + from apscheduler.triggers.cron import CronTrigger # noqa: E402 + + desired = _desired_jobs() + if not any(jid.startswith("pipeline_") for jid in desired): + logger.error("SCHEDULE_TIMES 为空或全部非法,保留现有定时任务不改动") + return False + + sig = json.dumps(desired, sort_keys=True, ensure_ascii=False) + if sig == _jobs_sig: + return False + + for jid, spec in desired.items(): + trigger = CronTrigger( + hour=spec["hour"], minute=spec["minute"], timezone=str(schedule_tz()) + ) + if spec["kind"] == "pipeline": + scheduler.add_job( + _scheduled_pipeline, + trigger=trigger, + kwargs={"steps": spec["steps"]}, + id=jid, + name=spec["name"], + replace_existing=True, + ) + else: + scheduler.add_job( + generate_all_stock_reports, + trigger=trigger, + id=jid, + name=spec["name"], + replace_existing=True, + ) + + for job in scheduler.get_jobs(): + if job.id.startswith("_") or job.id in desired: + continue + scheduler.remove_job(job.id) + + _jobs_sig = sig + logger.info( + "定时任务已同步: {}", + ", ".join(f"{s['name']}" for s in desired.values()), + ) + return True + + def _once(args: argparse.Namespace) -> int: """单次执行模式。 @@ -76,77 +202,20 @@ def _once(args: argparse.Namespace) -> int: def _daemon(args: argparse.Namespace) -> int: """守护进程模式(APScheduler)。""" - import os - from apscheduler.schedulers.background import BackgroundScheduler # noqa: E402 - from apscheduler.triggers.cron import CronTrigger # noqa: E402 - times_raw = os.environ.get("SCHEDULE_TIMES", "07:00,12:00,18:00,22:00") - times = _parse_schedule_times(times_raw) + times = _parse_schedule_times( + env_raw("SCHEDULE_TIMES", "07:00,12:00,18:00,22:00") or "" + ) if not times: logger.error("SCHEDULE_TIMES 为空或全部非法,无法启动定时任务") return 2 - # 找出最早的时间(当天首次运行),仅该次追加日报步骤 sorted_times = sorted(times) - first_hour, first_minute = sorted_times[0] if sorted_times else (0, 0) scheduler = BackgroundScheduler() - - # 包装函数:每次触发时按调度时区重新计算日期(P1-3), - # 避免 date.today() 在注册时冻结或与 cron 时区不一致。 - def _scheduled_pipeline(steps: list[str] | None = None) -> None: - run_pipeline(today_str(), steps=steps) - - for hour, minute in times: - trigger = CronTrigger(hour=hour, minute=minute, timezone=str(schedule_tz())) - is_first = (hour == first_hour and minute == first_minute) - job_kwargs: dict | None = None - if is_first: - job_kwargs = { - "steps": list(DEFAULT_NEWS_STEPS) + ["report"] - } - scheduler.add_job( - _scheduled_pipeline, - trigger=trigger, - kwargs=job_kwargs, - id=f"pipeline_{hour:02d}{minute:02d}", - name=f"全链路 {'+日报' if is_first else ''} {hour:02d}:{minute:02d}", - ) - logger.info("已注册定时任务: {}每天 {:02d}:{:02d}{}", trigger, hour, minute, - " (含日报)" if is_first else "") - - # cninfo 公告管道(可配置,默认 06:30) - cninfo_raw = os.environ.get("CNINFO_SCHEDULE_TIME", "06:30") - cninfo_parts = cninfo_raw.split(":") - cninfo_h, cninfo_m = int(cninfo_parts[0]), int(cninfo_parts[1]) if len(cninfo_parts) > 1 else 0 - cninfo_trigger = CronTrigger(hour=cninfo_h, minute=cninfo_m, timezone=str(schedule_tz())) - cninfo_steps = ["cninfo_crawl", "cninfo_extract", "cninfo_pdf", - "dedup", "llm", "embedding", "qdrant"] - scheduler.add_job( - _scheduled_pipeline, - trigger=cninfo_trigger, - kwargs={"steps": cninfo_steps}, - id="pipeline_cninfo", - name=f"cninfo 公告管道 {cninfo_h:02d}:{cninfo_m:02d}", - ) - logger.info("已注册定时任务: cninfo 公告管道 每天 {:02d}:{:02d}", cninfo_h, cninfo_m) - - # 个股日报(可配置,默认 07:30, 设为空可禁用) - stock_raw = os.environ.get("STOCK_REPORT_TIME", "07:30") - if stock_raw: - stock_parts = stock_raw.split(":") - stock_h, stock_m = int(stock_parts[0]), int(stock_parts[1]) if len(stock_parts) > 1 else 0 - stock_trigger = CronTrigger(hour=stock_h, minute=stock_m, timezone=str(schedule_tz())) - scheduler.add_job( - generate_all_stock_reports, - trigger=stock_trigger, - id="stock_report", - name=f"个股日报 {stock_h:02d}:{stock_m:02d}", - ) - logger.info("已注册定时任务: 个股日报 每天 {:02d}:{:02d}", stock_h, stock_m) - else: - logger.info("STOCK_REPORT_TIME 为空, 已禁用个股日报") + # 首次注册；此后由 _config_watch 每 30s 热同步，改 .env 无需重启 + _sync_jobs(scheduler) # 优雅退出 def _shutdown(signum: int, frame: Any) -> None: @@ -158,7 +227,19 @@ def _daemon(args: argparse.Namespace) -> int: signal.signal(signal.SIGTERM, _shutdown) scheduler.start() - logger.info("调度器已启动,等待触发... (按 Ctrl+C 退出)") + # 配置热同步：每 30s 重新计算 SCHEDULE_TIMES / CNINFO_SCHEDULE_TIME / STOCK_REPORT_TIME + scheduler.add_job( + _sync_jobs, + "interval", + seconds=30, + args=[scheduler], + id="_config_watch", + name="配置热同步", + replace_existing=True, + ) + # .env 热加载监听：provider / model / key / base_url 等改动无需重启 + start_env_watcher() + logger.info("调度器已启动,等待触发... (按 Ctrl+C 退出；改 .env 无需重启)") # 启动时检查是否有因重启/宕机错过的定时任务,30 分钟内补跑 now = tz_now() @@ -203,7 +284,8 @@ def main() -> int: args = parser.parse_args() _setup_logger(args.log_level) - load_dotenv() + # .env 热加载（改文件后常驻进程无需重启；--once 也会即时读取最新配置） + ensure_env_loaded() if args.once: return _once(args) diff --git a/tests/test_hot_reload.py b/tests/test_hot_reload.py new file mode 100644 index 0000000..f5ee606 --- /dev/null +++ b/tests/test_hot_reload.py @@ -0,0 +1,217 @@ +"""配置热加载测试：改 .env / llm_models.yaml 后无需重启即生效。 + +覆盖三层： + - configs/loader.py YAML 场景（(mtime, size) 失效缓存） + - configs/runtime_env .env（热更新 / 删除语义 / 外部显式覆盖优先） + - scripts/run_scheduler 定时任务热同步（改 SCHEDULE_TIMES 等） +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from configs import loader, runtime_env + + +@pytest.fixture(autouse=True) +def _clean_caches() -> None: + """每个用例前后都清干净，避免托管的环境变量污染其它测试。""" + loader.clear_cache() + runtime_env.reset_cache() + yield + loader.clear_cache() + runtime_env.reset_cache() + + +def _use_env_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, content: str) -> Path: + """把 runtime_env 指向临时 .env，返回其路径。""" + env = tmp_path / ".env" + env.write_text(content, encoding="utf-8") + monkeypatch.setenv(runtime_env.ENV_FILE_OVERRIDE, str(env)) + runtime_env.reset_cache() + return env + + +# --------------------------------------------------------------------------- # +# YAML 场景热加载 +# --------------------------------------------------------------------------- # + +def test_yaml_scene_hot_reload(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """改 llm_models.yaml 后，下一次 load_scene_config 立即读到新值。""" + cfg = tmp_path / "llm_models.yaml" + cfg.write_text( + "scenes:\n daily_report:\n provider: qwen\n model: qwen3.6-flash\n", + encoding="utf-8", + ) + monkeypatch.setenv(loader.MODELS_CONFIG_OVERRIDE, str(cfg)) + + assert loader.load_scene_config("daily_report")["model"] == "qwen3.6-flash" + + # 内容长度不同，保证 (mtime, size) 签名一定变化 + cfg.write_text( + "scenes:\n daily_report:\n provider: qwen\n model: deepseek-v4.1-flash\n", + encoding="utf-8", + ) + # 不重启、不 clear_cache 也应生效 + assert loader.load_scene_config("daily_report")["model"] == "deepseek-v4.1-flash" + + +def test_yaml_scene_removed_falls_back(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """场景被删掉后返回空 dict（由调用方回退 .env）。""" + cfg = tmp_path / "llm_models.yaml" + cfg.write_text("scenes:\n stock_report:\n model: x\n", encoding="utf-8") + monkeypatch.setenv(loader.MODELS_CONFIG_OVERRIDE, str(cfg)) + assert loader.load_scene_config("stock_report")["model"] == "x" + + cfg.write_text("scenes: {}\n", encoding="utf-8") + assert loader.load_scene_config("stock_report") == {} + + +def test_yaml_missing_file_returns_empty(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(loader.MODELS_CONFIG_OVERRIDE, str(tmp_path / "nope.yaml")) + assert loader.load_scene_config("daily_report") == {} + assert loader.load_defaults() == {} + + +# --------------------------------------------------------------------------- # +# .env 热加载 +# --------------------------------------------------------------------------- # + +def test_env_hot_reload(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """改 .env 后 env_get 立即读到新值。""" + env = _use_env_file(tmp_path, monkeypatch, "HOT_RELOAD_KEY=one\n") + monkeypatch.delenv("HOT_RELOAD_KEY", raising=False) + + assert runtime_env.env_get("HOT_RELOAD_KEY") == "one" + + env.write_text("HOT_RELOAD_KEY=two-longer-value\n", encoding="utf-8") + assert runtime_env.env_get("HOT_RELOAD_KEY") == "two-longer-value" + + +def test_env_removed_key_is_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """从 .env 删掉的托管键要同步从 os.environ 移除。""" + env = _use_env_file(tmp_path, monkeypatch, "HOT_REMOVE_KEY=abc\n") + monkeypatch.delenv("HOT_REMOVE_KEY", raising=False) + assert runtime_env.env_get("HOT_REMOVE_KEY") == "abc" + assert os.environ["HOT_REMOVE_KEY"] == "abc" + + env.write_text("OTHER_KEY=1\n", encoding="utf-8") + assert runtime_env.env_get("HOT_REMOVE_KEY") is None + assert "HOT_REMOVE_KEY" not in os.environ + + +def test_env_empty_value_is_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _use_env_file(tmp_path, monkeypatch, "HOT_EMPTY_KEY=\n") + monkeypatch.delenv("HOT_EMPTY_KEY", raising=False) + assert runtime_env.env_get("HOT_EMPTY_KEY", "fallback") == "fallback" + + +def test_explicit_process_env_wins(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """进程环境显式设置且与文件不同 → 不被 .env 覆盖（一次性覆盖语义）。""" + env = _use_env_file(tmp_path, monkeypatch, "HOT_OVERRIDE_KEY=from-file\n") + monkeypatch.setenv("HOT_OVERRIDE_KEY", "from-shell") + + assert runtime_env.env_get("HOT_OVERRIDE_KEY") == "from-shell" + + env.write_text("HOT_OVERRIDE_KEY=from-file-changed\n", encoding="utf-8") + assert runtime_env.env_get("HOT_OVERRIDE_KEY") == "from-shell" + + +def test_llm_config_follows_env_hot_reload( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """端到端：改 .env 里的 QWEN_MODEL，load_llm_config 立即用新模型。""" + from llm.client import load_llm_config + + contents = ( + "LLM_PROVIDER=qwen\n" + "QWEN_API_KEY=sk-test\n" + "QWEN_BASE_URL=https://token-plan.example/compatible-mode/v1\n" + "QWEN_MODEL=qwen3.6-flash\n" + ) + env = _use_env_file(tmp_path, monkeypatch, contents) + for key in ("LLM_PROVIDER", "QWEN_API_KEY", "QWEN_BASE_URL", "QWEN_MODEL"): + monkeypatch.delenv(key, raising=False) + + cfg = load_llm_config(scene="") + assert cfg.provider == "qwen" + assert cfg.model == "qwen3.6-flash" + + env.write_text(contents.replace("qwen3.6-flash", "deepseek-v4.1-flash"), encoding="utf-8") + assert load_llm_config(scene="").model == "deepseek-v4.1-flash" + + +# --------------------------------------------------------------------------- # +# 调度任务热同步 +# --------------------------------------------------------------------------- # + +def test_desired_jobs_tracks_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """SCHEDULE_TIMES / CNINFO_SCHEDULE_TIME / STOCK_REPORT_TIME 改完立即反映。""" + from scripts.run_scheduler import _desired_jobs + + _use_env_file( + tmp_path, + monkeypatch, + "SCHEDULE_TIMES=09:15\nCNINFO_SCHEDULE_TIME=05:45\nSTOCK_REPORT_TIME=\n", + ) + + jobs = _desired_jobs() + assert set(jobs) == {"pipeline_0915", "pipeline_cninfo"} + # 当天唯一次数 → 附带日报步骤 + assert jobs["pipeline_0915"]["steps"][-1] == "report" + assert jobs["pipeline_cninfo"]["hour"] == 5 + assert jobs["pipeline_cninfo"]["minute"] == 45 + + +def test_sync_jobs_adds_and_removes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """_sync_jobs 只按需增删：加时间点、启停个股日报、幂等。""" + from apscheduler.schedulers.background import BackgroundScheduler + + import scripts.run_scheduler as rs + + env = _use_env_file( + tmp_path, + monkeypatch, + "SCHEDULE_TIMES=09:15\nCNINFO_SCHEDULE_TIME=05:45\nSTOCK_REPORT_TIME=\n", + ) + rs._jobs_sig = None + + sched = BackgroundScheduler() + assert rs._sync_jobs(sched) is True + assert {j.id for j in sched.get_jobs()} == {"pipeline_0915", "pipeline_cninfo"} + + # 新增一个时间点 + 启用个股日报 + env.write_text( + "SCHEDULE_TIMES=09:15,16:40\nCNINFO_SCHEDULE_TIME=05:45\nSTOCK_REPORT_TIME=07:30\n", + encoding="utf-8", + ) + assert rs._sync_jobs(sched) is True + ids = {j.id for j in sched.get_jobs()} + assert ids == {"pipeline_0915", "pipeline_1640", "pipeline_cninfo", "stock_report"} + + # 配置未变 → 不重复变更 + assert rs._sync_jobs(sched) is False + + +def test_parse_hhmm_falls_back_on_bad_input() -> None: + from scripts.run_scheduler import _parse_hhmm + + assert _parse_hhmm("06:30", (0, 0)) == (6, 30) + assert _parse_hhmm("99:99", (6, 30)) == (6, 30) + assert _parse_hhmm("garbage", (7, 0)) == (7, 0) + + +# --------------------------------------------------------------------------- # +# 其它模块复用热读取 +# --------------------------------------------------------------------------- # + +def test_reporter_days_back_is_lazy(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """日报的 STOCK_REPORT_DAYS 走懒读取，不再冻结在模块常量里。""" + from scheduler import reporter + + _use_env_file(tmp_path, monkeypatch, "STOCK_REPORT_DAYS=15\n") + monkeypatch.delenv("STOCK_REPORT_DAYS", raising=False) + assert reporter._cninfo_days_back() == 15 diff --git a/tests/test_llm.py b/tests/test_llm.py index dd90824..a29dbe7 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -356,7 +356,23 @@ def test_load_llm_config_deepseek_from_env(monkeypatch: pytest.MonkeyPatch) -> N assert "deepseek" in cfg.base_url -def test_load_llm_config_qwen_from_env(monkeypatch: pytest.MonkeyPatch) -> None: +def test_load_llm_config_qwen_from_env( + monkeypatch: pytest.MonkeyPatch, tmp_path +) -> None: + """qwen 的 key 兜底链:QWEN_API_KEY 缺失时回退 DASHSCOPE_API_KEY。 + + 真实 .env 里带有 QWEN_API_KEY，会盖过 DASHSCOPE_API_KEY，所以本用例先把 + 热加载指向一个空 .env，测完再切回真实 .env。 + """ + from configs import runtime_env + + empty = tmp_path / ".env" + empty.write_text("", encoding="utf-8") + monkeypatch.setenv(runtime_env.ENV_FILE_OVERRIDE, str(empty)) + runtime_env.reset_cache() + monkeypatch.delenv("QWEN_API_KEY", raising=False) + monkeypatch.delenv("QWEN_BASE_URL", raising=False) + monkeypatch.setenv("LLM_PROVIDER", "qwen") monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-qwen") monkeypatch.setenv("QWEN_MODEL", "qwen-plus") @@ -367,6 +383,11 @@ def test_load_llm_config_qwen_from_env(monkeypatch: pytest.MonkeyPatch) -> None: assert cfg.model == "qwen-plus" assert "dashscope" in cfg.base_url or "aliyuncs" in cfg.base_url + # 切回真实 .env，避免影响后续用例（DB / Qdrant 等直接读 os.environ 的测试） + monkeypatch.delenv(runtime_env.ENV_FILE_OVERRIDE, raising=False) + runtime_env.reset_cache() + runtime_env.ensure_env_loaded(force=True) + def test_load_llm_config_unknown_provider_raises(monkeypatch: pytest.MonkeyPatch) -> None: with pytest.raises(ValueError): @@ -498,3 +519,39 @@ def test_load_llm_config_scene_max_attempts_zero() -> None: assert _pick_int({"max_attempts": 0}, "max_attempts", 3) == 0 assert _pick_int({"max_attempts": ""}, "max_attempts", 3) == 3 assert _pick_int({}, "max_attempts", 3) == 3 + + +# --------------------------------------------------------------------------- # +# max_tokens 场景配置（推理模型 reasoning 与正文共用预算,回归 9-25 无 AI 摘要） +# --------------------------------------------------------------------------- # + +def test_load_llm_config_scene_max_tokens(monkeypatch: pytest.MonkeyPatch) -> None: + """scenes..max_tokens 应写入 LLMConfig;未配置则为 None(调用方走默认)。""" + monkeypatch.setenv("QWEN_API_KEY", "sk-qwen") + _patch_scene(monkeypatch, {"provider": "qwen", "model": "deepseek-v4.1-flash", + "max_tokens": 4000}) + cfg = load_llm_config(scene="daily_report") + assert cfg.max_tokens == 4000 + + _patch_scene(monkeypatch, {"provider": "qwen", "model": "deepseek-v4.1-flash"}) + assert load_llm_config(scene="daily_report").max_tokens is None + + +def test_pick_optional_int() -> None: + """_pick_optional_int:未配置/空值/非法 → None;数字字符串可解析。""" + from llm.client import _pick_optional_int + + assert _pick_optional_int({"max_tokens": 4000}, "max_tokens") == 4000 + assert _pick_optional_int({"max_tokens": "4000"}, "max_tokens") == 4000 + assert _pick_optional_int({"max_tokens": ""}, "max_tokens") is None + assert _pick_optional_int({"max_tokens": "abc"}, "max_tokens") is None + assert _pick_optional_int({}, "max_tokens") is None + + +def test_real_yaml_daily_report_max_tokens_has_reasoning_headroom() -> None: + """真实配置的日报预算必须大于旧的 1500(为 reasoning token 留余量)。""" + from configs.loader import load_scene_config + + scene = load_scene_config("daily_report") + max_tokens = int(scene.get("max_tokens") or 0) + assert max_tokens >= 4000, f"daily_report.max_tokens 过小: {max_tokens}" diff --git a/tests/test_report_builder.py b/tests/test_report_builder.py index 1916a41..96d346b 100644 --- a/tests/test_report_builder.py +++ b/tests/test_report_builder.py @@ -132,8 +132,8 @@ class TestLlmCallRetry: def test_retry_then_success(self, monkeypatch) -> None: import scheduler.reporter as rep - monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 3) - monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) + monkeypatch.setattr(rep, "_llm_retry_times", lambda: 3) + monkeypatch.setattr(rep, "_llm_retry_backoff_sec", lambda: 0.01) client, n = self._fake_client(2) # 前 2 次失败，第 3 次成功 out = rep._llm_call(client, self._cfg(), "p") assert out == "今日要点摘要" @@ -141,8 +141,8 @@ class TestLlmCallRetry: def test_exhausts_retries_raises(self, monkeypatch) -> None: import scheduler.reporter as rep - monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 2) - monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) + monkeypatch.setattr(rep, "_llm_retry_times", lambda: 2) + monkeypatch.setattr(rep, "_llm_retry_backoff_sec", lambda: 0.01) client, n = self._fake_client(99) # 一直失败 with pytest.raises(ConnectionError): rep._llm_call(client, self._cfg(), "p") @@ -171,6 +171,84 @@ class TestLlmCallRetry: assert n["count"] == 3 # 2 块 + 1 次合并 +class TestReasoningBudgetEscalation: + """推理模型占满 max_tokens 导致正文为空时的预算升级（回归 9-25 无 AI 摘要）。""" + + @staticmethod + def _empty_then_ok_client(empty_times: int): + """前 empty_times 次返回空正文 + finish_reason=length，之后返回正常摘要。""" + from types import SimpleNamespace + + seen: list[int] = [] + + class Completions: + def create(self, **kwargs): + seen.append(kwargs.get("max_tokens")) + if len(seen) <= empty_times: + return SimpleNamespace( + choices=[SimpleNamespace( + message=SimpleNamespace(content=""), + finish_reason="length", + )] + ) + return SimpleNamespace( + choices=[SimpleNamespace( + message=SimpleNamespace(content="恢复后的摘要"), + finish_reason="stop", + )] + ) + + return SimpleNamespace(chat=SimpleNamespace(completions=Completions())), seen + + @staticmethod + def _cfg(max_tokens: int | None = None): + from llm.client import LLMConfig + + return LLMConfig( + provider="qwen", model="deepseek-v4.1-flash", + api_key="sk-test", base_url="https://example.invalid/v1", + temperature=0.3, max_tokens=max_tokens, + ) + + def test_empty_content_escalates_and_recovers(self, monkeypatch) -> None: + """正文为空时自动加倍预算并最终拿到摘要（不再静默返回空）。""" + import scheduler.reporter as rep + monkeypatch.setattr(rep, "_llm_retry_times", lambda: 3) + monkeypatch.setattr(rep, "_llm_retry_backoff_sec", lambda: 0.01) + client, seen = self._empty_then_ok_client(1) + out = rep._llm_call(client, self._cfg(4000), "p") + assert out == "恢复后的摘要" + assert seen == [4000, 8000] # 首次失败后预算翻倍 + + def test_scene_max_tokens_wins_over_default(self, monkeypatch) -> None: + """场景 max_tokens 生效；未配置时回退代码默认 4000。""" + import scheduler.reporter as rep + monkeypatch.setattr(rep, "_llm_retry_times", lambda: 1) + client, seen = self._empty_then_ok_client(0) + rep._llm_call(client, self._cfg(6000), "p") + assert seen == [6000] + client, seen = self._empty_then_ok_client(0) + rep._llm_call(client, self._cfg(), "p") + assert seen == [rep.DEFAULT_SUMMARY_MAX_TOKENS] + + def test_all_empty_returns_blank_without_raising(self, monkeypatch) -> None: + """预算升级用尽仍为空时降级返回空串（日报仍可入库，不抛异常）。""" + import scheduler.reporter as rep + monkeypatch.setattr(rep, "_llm_retry_times", lambda: 3) + monkeypatch.setattr(rep, "_llm_retry_backoff_sec", lambda: 0.01) + client, seen = self._empty_then_ok_client(99) + out = rep._llm_call(client, self._cfg(4000), "p") + assert out == "" + # 4000 → 8000 → 16000（受 MAX_SUMMARY_MAX_TOKENS 上限约束） + assert seen == [4000, 8000, 16000] + + def test_budget_never_exceeds_cap(self) -> None: + """升级预算不超过 MAX_SUMMARY_MAX_TOKENS，避免无限放大。""" + import scheduler.reporter as rep + assert rep.MAX_SUMMARY_MAX_TOKENS == 16000 + assert rep.DEFAULT_SUMMARY_MAX_TOKENS > rep.DEFAULT_SUMMARY_CHUNK_MAX_TOKENS + + class TestCollectXwlb: """_collect_xwlb 取数逻辑：应查询日报前一日(已播出的联播)，并跳过内容提要。""" diff --git a/vectorstore/client.py b/vectorstore/client.py index 1cf69a4..cadaf5d 100644 --- a/vectorstore/client.py +++ b/vectorstore/client.py @@ -10,7 +10,6 @@ from __future__ import annotations -import os import uuid from datetime import datetime from pathlib import Path @@ -30,6 +29,8 @@ from qdrant_client.http.models import ( VectorParams, ) +from configs.runtime_env import env_get + from .models import CollectionInfo, SearchFilter, SearchResult # 默认配置 @@ -52,10 +53,8 @@ def url_hash_to_uuid(url_hash: str) -> str: def _read_env(key: str, default: str | None = None) -> str | None: - val = os.environ.get(key) - if val is None or val.strip() == "": - return default - return val.strip() + """读取环境变量（先热加载 .env，改文件后无需重启进程）。""" + return env_get(key, default) def make_qdrant_client(