From 56297ac384248cb270e4468415cc7a00083db67f Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Fri, 17 Jul 2026 11:04:20 +0800 Subject: [PATCH] fix: bound historical context snapshots --- docs/design.md | 2 + docs/notes.md | 6 ++ env.example | 2 + tests/test_cli.py | 2 + tests/test_config.py | 31 +++++++++ tests/test_service.py | 72 +++++++++++++++++++- weather_briefing/config.py | 9 +++ weather_briefing/prompts.py | 1 + weather_briefing/service.py | 131 ++++++++++++++++++++++++++++++++---- 9 files changed, 242 insertions(+), 14 deletions(-) diff --git a/docs/design.md b/docs/design.md index d94f3e3..801c466 100644 --- a/docs/design.md +++ b/docs/design.md @@ -78,6 +78,8 @@ SQLite 保存： 每次任务成功结束时，状态存储在重置任务失败计数的同一事务中清理历史数据。已处理文章、已发布简报和上下文快照保留 `HISTORY_HOURS`，预警按独立的 `WARNING_RETENTION_HOURS` 保留；恰好位于阈值时刻的记录仍属于当前窗口。尚未成功发布的待处理文章和使用 upsert 维护的健康状态不参与历史清理。失败任务不清理历史，以免在恢复前丢失诊断与待处理上下文。 +历史上下文快照在进入 LLM 前按 source 折叠连续相同值，并依次保留各 source 的最新值、窗口基线和最近变化节点；每条记录通过 `history_role` 明确标记这三种角色。`LLM_HISTORY_MAX_DOCUMENTS` 限制快照总数，`LLM_HISTORY_MAX_CHARACTERS` 限制 `recent_context_documents` JSON 数组的完整序列化字符数；超过字符预算的完整文档被跳过而不截断正文。最近成功发布的简报仍随 payload 提供，模型因此可以用有界的基线、最新值及变化节点判断相对上次成功发布的累计变化。DEBUG 诊断只记录候选数、入选数和序列化字符数，不记录来源名称、URL 或正文。 + ## 时间模型 应用只接受 Pendulum 的时区感知 `DateTime`，绝不把缺少明确时区的信息解释为时间点，也不依赖进程或服务器的本地时区。内存中的时间保留其明确时区，Pendulum 直接按绝对时间比较，不为比较提前转换。 diff --git a/docs/notes.md b/docs/notes.md index 1dda0ad..fadf864 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -36,6 +36,12 @@ Nominatim 限速使用 `await asyncio.sleep()`，它只挂起当前协程并把 大陆行政区排除规则在载入参考数据和比较 API 返回值时都使用 `casefold()`，不存在两边归一化不一致。空气质量分级数量很小且最后一档无上限，线性扫描比为 `bisect` 拆分阈值与结果更直接。provider UTC offset 接受整分钟的 ±14 小时范围，这是合法时区边界，不应收窄。 +## LLM 历史输入预算的当前范围 + +`LLM_HISTORY_MAX_DOCUMENTS` 和 `LLM_HISTORY_MAX_CHARACTERS` 只限制逐小时积累的历史 API 上下文快照。历史文章与已发布简报仍由 `HISTORY_HOURS` 控制时间范围，不另设数量或字符预算。当前接受这一边界，是因为已配置 RSS 的发布频率较低，普通 briefing 只有值得打扰时才保存正文，而 API 快照是唯一会在每个调度轮次稳定增长的输入类别。 + +应用以 DEBUG 元数据记录完整 payload 的序列化字符数，但不记录正文。若 RSS 发布量或地点数显著增加、历史简报开始接近逐小时产生、payload 字符数持续增长，或再次出现与输入规模相关的模型超时或上下文长度错误，应把文章、简报和快照收敛到带类别最低保留量的统一历史输入预算，而不是继续分别增加当前两个快照上限。 + ## 异步编排中的同步本地持久化 ### 当前选择 diff --git a/env.example b/env.example index b3739a9..bc88fda 100644 --- a/env.example +++ b/env.example @@ -75,3 +75,5 @@ BRIEFING_STATE_PATH=state/weather.sqlite3 RSS_FAILURE_THRESHOLD=3 WARNING_RETENTION_HOURS=12 HISTORY_HOURS=48 +LLM_HISTORY_MAX_DOCUMENTS=8 +LLM_HISTORY_MAX_CHARACTERS=16000 diff --git a/tests/test_cli.py b/tests/test_cli.py index 47aea02..e1e452c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -580,6 +580,8 @@ async def fail_run( rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, + llm_history_max_documents=16, + llm_history_max_characters=16_000, briefing_max_characters=3500, greeting_hour=8, greeting_minute=0, diff --git a/tests/test_config.py b/tests/test_config.py index 81a4842..fd94bf4 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -49,6 +49,37 @@ def test_mainland_weather_providers_default_to_qweather_then_open_meteo(monkeypa assert settings.llm_base_url is None assert settings.qweather_index_types == ("1", "3", "6", "7", "8", "15") assert settings.qweather_jwt_lifetime_seconds == 900 + assert settings.llm_history_max_documents == 8 + assert settings.llm_history_max_characters == 16_000 + + +def test_llm_history_limits_can_be_configured(monkeypatch) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_HISTORY_MAX_DOCUMENTS", "6") + monkeypatch.setenv("LLM_HISTORY_MAX_CHARACTERS", "8000") + + settings = Settings.from_env() + + assert settings.llm_history_max_documents == 6 + assert settings.llm_history_max_characters == 8_000 + + +@pytest.mark.parametrize("value", ("0", "257")) +def test_llm_history_document_limit_rejects_unsafe_values(monkeypatch, value: str) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_HISTORY_MAX_DOCUMENTS", value) + + with pytest.raises(ConfigurationError, match="LLM_HISTORY_MAX_DOCUMENTS"): + Settings.from_env() + + +@pytest.mark.parametrize("value", ("0", "1", "1000001")) +def test_llm_history_character_limit_rejects_unsafe_values(monkeypatch, value: str) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_HISTORY_MAX_CHARACTERS", value) + + with pytest.raises(ConfigurationError, match="LLM_HISTORY_MAX_CHARACTERS"): + Settings.from_env() def test_environment_example_preserves_default_qweather_indices() -> None: diff --git a/tests/test_service.py b/tests/test_service.py index 5b0df30..5a910e6 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,4 +1,6 @@ import asyncio +import json +import logging from dataclasses import dataclass from pathlib import Path from typing import TypeGuard @@ -20,7 +22,7 @@ ) from weather_briefing.publishers import DeliveryProvider from weather_briefing.render import PlainTextRenderer -from weather_briefing.service import BriefingService +from weather_briefing.service import BriefingService, _bounded_context_history, _context_history_candidates from weather_briefing.state import SQLiteStateStore from weather_briefing.weather_context import WeatherContextError @@ -34,6 +36,8 @@ class _TestSettings: rss_failure_threshold: int = 3 warning_retention_hours: int = 12 history_hours: int = 48 + llm_history_max_documents: int = 8 + llm_history_max_characters: int = 16_000 briefing_max_characters: int = 3500 llm_max_attempts: int = 3 @@ -209,6 +213,61 @@ def _location( ) +def _context_document(source_id: str, content: str) -> SourceDocument: + return SourceDocument( + id=source_id, + name=f"Source {source_id}", + url=f"https://example.invalid/{source_id}", + content=content, + ) + + +def test_context_history_keeps_latest_baselines_and_recent_changes() -> None: + documents = ( + _context_document("weather", "weather baseline"), + _context_document("weather", "weather baseline"), + _context_document("weather", "weather changed"), + _context_document("air", "air baseline"), + _context_document("air", "air changed"), + _context_document("weather", "weather latest"), + _context_document("air", "air latest"), + ) + + selected = _context_history_candidates(documents, max_documents=6) + + assert [(candidate.document.id, candidate.document.content, candidate.role) for candidate in selected] == [ + ("air", "air latest", "latest"), + ("weather", "weather latest", "latest"), + ("air", "air baseline", "retention_baseline"), + ("weather", "weather baseline", "retention_baseline"), + ("air", "air changed", "recent_change"), + ("weather", "weather changed", "recent_change"), + ] + + +def test_context_history_enforces_document_and_serialized_character_limits() -> None: + oversized = _context_document("oversized", "private-history" * 20) + compact = _context_document("compact", "current") + compact_payload, compact_characters = _bounded_context_history( + (compact,), + max_documents=1, + max_characters=1_000, + ) + character_limit = compact_characters + + selected, serialized_characters = _bounded_context_history( + (compact, oversized), + max_documents=2, + max_characters=character_limit, + ) + + assert [item["source_id"] for item in selected] == ["compact"] + assert selected[0]["history_role"] == "latest" + assert len(selected) <= 2 + assert serialized_characters == len(json.dumps(selected, ensure_ascii=False, separators=(",", ":"))) + assert serialized_characters <= character_limit + + @pytest.mark.parametrize( ("kind", "historical_ids"), (("briefing", ["historical-verbatim"]), ("forecast", ["historical-verbatim", "historical-summary"])), @@ -551,6 +610,7 @@ async def test_briefing_also_uses_the_llm_provider(tmp_path: Path) -> None: async def test_briefing_api_only_update_can_be_remembered_without_delivery( tmp_path: Path, + caplog: pytest.LogCaptureFixture, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") settings = _TestSettings( @@ -571,6 +631,10 @@ async def test_briefing_api_only_update_can_be_remembered_without_delivery( now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) with SQLiteStateStore(tmp_path / "state.sqlite3") as state: + state.save_context_documents( + (_context_document("weather:test", "sensitive historical body"),), + now.subtract(hours=1), + ) service = BriefingService( settings, _location(), @@ -582,13 +646,17 @@ async def test_briefing_api_only_update_can_be_remembered_without_delivery( delivery, weather_context, ) - result = await service.run("briefing", now) + with caplog.at_level(logging.DEBUG, logger="weather_briefing.service"): + result = await service.run("briefing", now) remembered = state.recent_context_documents(now, 1) assert result is None assert llm.payload is not None assert llm.payload["mode"] == "briefing" assert publisher.messages == [] + assert "input_documents=1 selected_documents=1 serialized_characters=" in caplog.text + assert "LLM payload prepared: serialized_characters=" in caplog.text + assert "sensitive historical body" not in caplog.text assert {document.id for document in remembered} == { "weather:test", "air-quality:test", diff --git a/weather_briefing/config.py b/weather_briefing/config.py index 412c2bf..880e3f6 100644 --- a/weather_briefing/config.py +++ b/weather_briefing/config.py @@ -301,6 +301,8 @@ class Settings: rss_failure_threshold: int warning_retention_hours: int history_hours: int + llm_history_max_documents: int + llm_history_max_characters: int briefing_max_characters: int greeting_hour: int greeting_minute: int @@ -424,6 +426,13 @@ def from_env(cls) -> Settings: rss_failure_threshold=_positive_integer("RSS_FAILURE_THRESHOLD", 3), warning_retention_hours=_positive_integer("WARNING_RETENTION_HOURS", 12), history_hours=_positive_integer("HISTORY_HOURS", 48), + llm_history_max_documents=_bounded_positive_integer("LLM_HISTORY_MAX_DOCUMENTS", 8, 256), + llm_history_max_characters=_bounded_integer( + "LLM_HISTORY_MAX_CHARACTERS", + 16_000, + 2, + 1_000_000, + ), briefing_max_characters=briefing_max_characters, greeting_hour=daily_cron_hour, greeting_minute=daily_cron_minute, diff --git a/weather_briefing/prompts.py b/weather_briefing/prompts.py index 24928ab..f595a12 100644 --- a/weather_briefing/prompts.py +++ b/weather_briefing/prompts.py @@ -53,6 +53,7 @@ 与上次成功发布相比没有值得打扰的累计变化时必须设为 false。forecast 模式必须设为 true。 总结 deferred_articles、historical_articles、recent_context_documents 和 recent_briefings 时， 必须先判断每条信息的时效性。 +recent_context_documents 的 history_role 分别标识各来源的最新值、保留窗口基线和最近变化节点。 对气温、降水、风力、空气质量、短时预报等会过期的信息，始终以时间最新且仍适用于当前时刻的来源为准； 不得因为旧信息尚未发送就保留已经被较新快照取代的数值或结论。 active_warnings 应包含仍有效的预警以维持状态，但预警仅仅持续生效且内容无实质变化时必须设为 false。 diff --git a/weather_briefing/service.py b/weather_briefing/service.py index e37cf57..58dc4a3 100644 --- a/weather_briefing/service.py +++ b/weather_briefing/service.py @@ -3,9 +3,11 @@ from __future__ import annotations import asyncio +import json import logging from collections.abc import Callable -from typing import Protocol +from dataclasses import dataclass +from typing import Literal, Protocol import pendulum @@ -30,6 +32,12 @@ _LOGGER = logging.getLogger("weather_briefing.service") +@dataclass(frozen=True, slots=True) +class _HistoricalContextCandidate: + document: SourceDocument + role: Literal["latest", "retention_baseline", "recent_change"] + + def _serialize_article(article: Article) -> dict[str, object]: return { "source_id": article.id, @@ -80,6 +88,16 @@ def history_hours(self) -> int: """Return the retained briefing context window in hours.""" ... + @property + def llm_history_max_documents(self) -> int: + """Return the maximum historical context snapshots sent to the LLM.""" + ... + + @property + def llm_history_max_characters(self) -> int: + """Return the serialized character budget for historical context.""" + ... + @property def briefing_max_characters(self) -> int: """Return the configured briefing character budget.""" @@ -257,6 +275,17 @@ async def _run( context_items.extend(snapshot_to_documents(weather_context)) context = tuple(context_items) historical_context = self._state.recent_context_documents(now, self._settings.history_hours) + historical_context_payload, historical_context_characters = _bounded_context_history( + historical_context, + max_documents=self._settings.llm_history_max_documents, + max_characters=self._settings.llm_history_max_characters, + ) + _LOGGER.debug( + "Historical context bounded: input_documents=%d selected_documents=%d serialized_characters=%d", + len(historical_context), + len(historical_context_payload), + historical_context_characters, + ) reference_context = _unique_documents((*historical_context, *context)) active_warnings = self._state.active_warnings(now, self._settings.warning_retention_hours) if not unpublished_articles and not context and not active_warnings: @@ -283,13 +312,18 @@ async def _run( deferred_articles, historical_articles, context, - historical_context, + historical_context_payload, active_warnings, ) briefing_limit = self._delivery.briefing_limit(self._settings.briefing_max_characters) payload["output_constraints"] = {"briefing_max_characters": briefing_limit} required_advice_topics = _required_advice_topics(kind, context) payload["required_advice_topics"] = [topic.value for topic in required_advice_topics] + if _LOGGER.isEnabledFor(logging.DEBUG): + _LOGGER.debug( + "LLM payload prepared: serialized_characters=%d", + len(json.dumps(payload, ensure_ascii=False, separators=(",", ":"))), + ) allergen_source_ids = {document.id for document in context if document.has_allergen_information} valid_source_ids = {article.id for article in source_articles} | {document.id for document in reference_context} active_warning_ids = {warning.id for warning in active_warnings} @@ -460,7 +494,7 @@ def _build_payload( deferred_articles: tuple[Article, ...], historical_articles: tuple[Article, ...], context: tuple[SourceDocument, ...], - historical_context: tuple[SourceDocument, ...], + historical_context: list[dict[str, object]], active_warnings: tuple[Warning, ...], ) -> dict[str, object]: location_scope = {"full_name": self._location.name} @@ -484,15 +518,7 @@ def _build_payload( "context_documents": [ {"source_id": item.id, "name": item.name, "url": item.url, "content": item.content} for item in context ], - "recent_context_documents": [ - { - "source_id": item.id, - "name": item.name, - "url": item.url, - "content": item.content, - } - for item in historical_context - ], + "recent_context_documents": historical_context, "recent_briefings": [ { "mode": briefing.kind, @@ -519,6 +545,87 @@ def _unique_articles(articles: tuple[Article, ...]) -> tuple[Article, ...]: return tuple({article.id: article for article in articles}.values()) +def _serialize_context_document( + document: SourceDocument, + *, + history_role: Literal["latest", "retention_baseline", "recent_change"], +) -> dict[str, object]: + return { + "source_id": document.id, + "name": document.name, + "url": document.url, + "content": document.content, + "history_role": history_role, + } + + +def _bounded_context_history( + documents: tuple[SourceDocument, ...], + *, + max_documents: int, + max_characters: int, +) -> tuple[list[dict[str, object]], int]: + candidates = _context_history_candidates(documents, max_documents) + selected: list[dict[str, object]] = [] + serialized_characters = len("[]") + for candidate in candidates: + entry = _serialize_context_document(candidate.document, history_role=candidate.role) + candidate_payload = [*selected, entry] + candidate_characters = len(json.dumps(candidate_payload, ensure_ascii=False, separators=(",", ":"))) + if candidate_characters <= max_characters: + selected = candidate_payload + serialized_characters = candidate_characters + return selected, serialized_characters + + +def _context_history_candidates( + documents: tuple[SourceDocument, ...], + max_documents: int, +) -> tuple[_HistoricalContextCandidate, ...]: + changes_by_source: dict[str, list[tuple[int, SourceDocument]]] = {} + for index, document in enumerate(documents): + source_changes = changes_by_source.setdefault(document.id, []) + previous_document = source_changes[-1][1] if source_changes else None + if previous_document is None or _context_document_value(previous_document) != _context_document_value(document): + source_changes.append((index, document)) + + source_changes = sorted(changes_by_source.values(), key=lambda changes: changes[-1][0], reverse=True) + prioritized: list[tuple[int, _HistoricalContextCandidate]] = [] + selected_indexes: set[int] = set() + + def add( + snapshot: tuple[int, SourceDocument], + role: Literal["latest", "retention_baseline", "recent_change"], + ) -> None: + if snapshot[0] not in selected_indexes: + prioritized.append((snapshot[0], _HistoricalContextCandidate(snapshot[1], role))) + selected_indexes.add(snapshot[0]) + + for changes in source_changes: + add(changes[-1], "latest") + for changes in source_changes: + add(changes[0], "retention_baseline") + + depth = 2 + while len(prioritized) < max_documents: + added_change = False + for changes in source_changes: + if len(changes) > depth: + add(changes[-depth], "recent_change") + added_change = True + if len(prioritized) == max_documents: + break + if not added_change: + break + depth += 1 + + return tuple(candidate for _, candidate in prioritized[:max_documents]) + + +def _context_document_value(document: SourceDocument) -> tuple[str, str, str]: + return document.name, document.url, document.content + + def _required_advice_topics( kind: str, context: tuple[SourceDocument, ...],