From 5b610fd8074867d1fea964747c6edb9e87a222c1 Mon Sep 17 00:00:00 2001 From: Administrator Date: Tue, 16 Jun 2026 22:15:36 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=8A=E4=B8=8B=E6=96=87=E5=8E=8B?= =?UTF-8?q?=E7=BC=A9=E6=94=AF=E6=8C=81=E7=8A=B6=E6=80=81=E6=8C=81=E4=B9=85?= =?UTF-8?q?=E5=8C=96=E7=BC=93=E5=AD=98,=E5=89=8D=E7=AB=AF=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E9=9A=90=E8=97=8F=E6=B6=88=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - summarization_middleware 摘要缓存写入 LangGraph thread state,跨进程/重启可复用 - 摘要消息改为 SystemMessage,提示模型静默使用、不泄露 - thread_state 新增 context_summary_cache 字段 - app_config 加载时记录 summarization 配置日志 - config.yaml 压缩触发/保留阈值调整为生产值(token 预算) - 前端 input-box / message-list 过滤 hide_from_ui 消息 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/components/workspace/input-box.tsx | 10 +- .../workspace/messages/message-list.tsx | 12 +- offline-backend-20260512/backend/config.yaml | 10 +- .../middlewares/summarization_middleware.py | 145 +++++++++++++----- .../harness/deerflow/agents/thread_state.py | 5 +- .../harness/deerflow/config/app_config.py | 31 ++-- offline-backend-20260512/config.yaml | 17 +- 7 files changed, 159 insertions(+), 71 deletions(-) diff --git a/frontend-web/src/components/workspace/input-box.tsx b/frontend-web/src/components/workspace/input-box.tsx index f4a30d63..39692e5d 100644 --- a/frontend-web/src/components/workspace/input-box.tsx +++ b/frontend-web/src/components/workspace/input-box.tsx @@ -61,8 +61,9 @@ import { } from "@/components/ui/dropdown-menu"; import { apiFetch } from "@/core/api"; import { getBackendBaseURL } from "@/core/config"; -import { useI18n } from "@/core/i18n/hooks"; -import { useModels } from "@/core/models/hooks"; +import { useI18n } from "@/core/i18n/hooks"; +import { isHiddenFromUIMessage } from "@/core/messages/utils"; +import { useModels } from "@/core/models/hooks"; import type { AgentThreadContext } from "@/core/threads"; import { textOfMessage } from "@/core/threads/utils"; import { desensitize } from "@/lib/desensitize"; @@ -467,14 +468,15 @@ export function InputBox({ return; } - const lastAi = [...thread.messages].reverse().find((m) => m.type === "ai"); + const visibleMessages = thread.messages.filter((m) => !isHiddenFromUIMessage(m)); + const lastAi = [...visibleMessages].reverse().find((m) => m.type === "ai"); const lastAiId = lastAi?.id ?? null; if (!lastAiId || lastAiId === lastGeneratedForAiIdRef.current) { return; } lastGeneratedForAiIdRef.current = lastAiId; - const recent = thread.messages + const recent = visibleMessages .filter((m) => m.type === "human" || m.type === "ai") .map((m) => { const role = m.type === "human" ? "user" : "assistant"; diff --git a/frontend-web/src/components/workspace/messages/message-list.tsx b/frontend-web/src/components/workspace/messages/message-list.tsx index 68aa9429..f10dc1bc 100644 --- a/frontend-web/src/components/workspace/messages/message-list.tsx +++ b/frontend-web/src/components/workspace/messages/message-list.tsx @@ -20,9 +20,10 @@ import { groupMessages, hasContent, hasPresentFiles, - hasReasoning, - hasToolCalls, -} from "@/core/messages/utils"; + hasReasoning, + hasToolCalls, + isHiddenFromUIMessage, +} from "@/core/messages/utils"; import { useRehypeSplitWordsIntoSpans } from "@/core/rehype"; import { isThreadRagModeLocked, useLocalSettings } from "@/core/settings"; import { useSkills } from "@/core/skills/hooks"; @@ -151,7 +152,10 @@ export function MessageList({ }); }, []); - const allMessages = thread.messages; + const allMessages = useMemo( + () => thread.messages.filter((m) => !isHiddenFromUIMessage(m)), + [thread.messages], + ); const messages = useMemo( () => hiddenMessageIds.size === 0 diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 0ca799fe..c23bb4ce 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -207,13 +207,13 @@ summarization: # model_name: deepseek-chat # 留空则用主对话模型 trigger: - type: tokens - value: 8000 + value: 120000 - type: messages - value: 12 + value: 160 keep: - type: messages - value: 6 - trim_tokens_to_summarize: 4000 + type: tokens + value: 40000 + trim_tokens_to_summarize: 8000 tool_groups: - name: web diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index 11e7d78c..3292f89f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -27,7 +27,14 @@ from typing import Any, Protocol, override, runtime_checkable from langchain.agents import AgentState from langchain.agents.middleware import SummarizationMiddleware from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse -from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage +from langchain_core.messages import ( + AIMessage, + AnyMessage, + SystemMessage, + ToolMessage, + message_to_dict, + messages_from_dict, +) from langgraph.config import get_config from langgraph.runtime import Runtime @@ -138,7 +145,8 @@ class _SummaryCacheEntry: summarized_ids: frozenset[str] summary_text: str - reusable: bool + preserved_ids: frozenset[str] = frozenset() + preserved_messages: tuple[AnyMessage, ...] = () # Process-wide summary cache shared across every middleware instance. @@ -152,14 +160,51 @@ class _SummaryCacheEntry: # so only the first threshold crossing (or a genuine re-compaction) pays for an # LLM call. A lock guards the non-atomic LRU eviction against concurrent # threads (subagents / parallel conversations share this process). -_GLOBAL_SUMMARY_CACHE: "OrderedDict[str, _SummaryCacheEntry]" = OrderedDict() -_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock() - - -def _clear_global_summary_cache() -> None: - """Drop every cached summary. Intended for tests/diagnostics only.""" - with _GLOBAL_SUMMARY_CACHE_LOCK: - _GLOBAL_SUMMARY_CACHE.clear() +_GLOBAL_SUMMARY_CACHE: OrderedDict[str, _SummaryCacheEntry] = OrderedDict() +_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock() +_STATE_CACHE_KEY = "context_summary_cache" + + +def _clear_global_summary_cache() -> None: + """Drop every cached summary. Intended for tests/diagnostics only.""" + with _GLOBAL_SUMMARY_CACHE_LOCK: + _GLOBAL_SUMMARY_CACHE.clear() + + +def _serialize_cache_entry(entry: _SummaryCacheEntry) -> dict[str, Any]: + """Serialize a summary cache entry so it can live in LangGraph state.""" + return { + "version": 1, + "summarized_ids": sorted(entry.summarized_ids), + "summary_text": entry.summary_text, + "preserved_ids": sorted(entry.preserved_ids), + "preserved_messages": [message_to_dict(message) for message in entry.preserved_messages], + } + + +def _deserialize_cache_entry(raw: Any) -> _SummaryCacheEntry | None: + """Best-effort restore of a state-backed summary cache entry.""" + if not isinstance(raw, dict): + return None + try: + summary_text = raw.get("summary_text") + if not isinstance(summary_text, str) or not summary_text: + return None + summarized_ids = raw.get("summarized_ids") or [] + preserved_ids = raw.get("preserved_ids") or [] + preserved_messages_raw = raw.get("preserved_messages") or [] + if not isinstance(summarized_ids, list) or not isinstance(preserved_ids, list) or not isinstance(preserved_messages_raw, list): + return None + preserved_messages = tuple(messages_from_dict(preserved_messages_raw)) + return _SummaryCacheEntry( + summarized_ids=frozenset(str(item) for item in summarized_ids if item is not None), + summary_text=summary_text, + preserved_ids=frozenset(str(item) for item in preserved_ids if item is not None), + preserved_messages=preserved_messages, + ) + except Exception: + logger.debug("summarization: failed to restore state-backed summary cache", exc_info=True) + return None class DeerFlowSummarizationMiddleware(SummarizationMiddleware): @@ -186,7 +231,8 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): # Per-thread cache of the most recent summary: thread_id -> (summarized message ids, summary text). # Points at the *process-wide* cache (not a fresh per-instance dict) so a # summary survives across runs/turns — see ``_GLOBAL_SUMMARY_CACHE``. - self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = _GLOBAL_SUMMARY_CACHE + self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = _GLOBAL_SUMMARY_CACHE + self._pending_state_cache_update: dict[str, Any] | None = None @override def before_model(self, state: AgentState, runtime: Runtime) -> dict | None: @@ -202,7 +248,21 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): """Async counterpart of :meth:`before_model` — also a no-op.""" return None - # ---- wrap_model_call: transient, display-preserving compression ---------- + @override + def after_model(self, state: AgentState, runtime: Runtime) -> dict | None: + """Persist the latest reusable summary cache into thread state.""" + if self._pending_state_cache_update is None: + return None + update = self._pending_state_cache_update + self._pending_state_cache_update = None + return {_STATE_CACHE_KEY: update} + + @override + async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict | None: + """Async counterpart of :meth:`after_model`.""" + return self.after_model(state, runtime) + + # ---- wrap_model_call: transient, display-preserving compression ---------- @override def wrap_model_call( @@ -316,14 +376,20 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): return None thread_id = self._resolve_request_thread_id(request) - if thread_id is None: - logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call") - cached = self._get_cache_entry(thread_id) if thread_id else None - present_ids = {m.id for m in messages} - cache_usable = cached is not None and cached.reusable and cached.summarized_ids <= present_ids - if cache_usable: - preserved = [m for m in messages if m.id not in cached.summarized_ids] - effective = [self._summary_message(cached.summary_text), *preserved] + if thread_id is None: + logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call") + cached = self._get_cache_entry(thread_id) if thread_id else None + if cached is None: + cached = _deserialize_cache_entry(getattr(request, "state", {}).get(_STATE_CACHE_KEY)) + if cached is not None and thread_id: + self._store_cache_entry(thread_id, cached) + present_ids = {m.id for m in messages} + cache_usable = cached is not None and cached.summarized_ids <= present_ids and cached.preserved_ids <= present_ids + if cache_usable: + preserved = list(cached.preserved_messages) + excluded_ids = cached.summarized_ids | cached.preserved_ids + preserved.extend(m for m in messages if m.id not in excluded_ids) + effective = [self._summary_message(cached.summary_text), *preserved] # Decide re-summarization on the *counted* size of the compressed # context only — NOT ``_should_summarize`` — because that folds in the # last AIMessage's reported ``usage_metadata`` (system prompt + tool @@ -349,7 +415,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): # an "it compacts every turn" report is diagnosable straight from the # backend log: cache=miss → thread_id/key problem; unusable → message-id # instability; stale → preserved genuinely grew back over the trigger. - cache_state = "stale" if cache_usable else ("miss" if cached is None else "unusable") + cache_state = "miss" if cached is None else "stale" logger.info( "summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d", thread_id, @@ -390,16 +456,18 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): if thread_id: summarized_ids = frozenset(m.id for m in messages_to_summarize if m.id is not None) preserved_ids = {m.id for m in preserved if m.id is not None} - # A non-empty overlap means skill rescue split an AIMessage across the - # boundary; reconstructing ``preserved`` by id filtering later could - # orphan a tool result, so mark the entry non-reusable (re-summarize). - reusable = summarized_ids.isdisjoint(preserved_ids) - self._store_cache_entry( - thread_id, - _SummaryCacheEntry(summarized_ids=summarized_ids, summary_text=summary, reusable=reusable), - ) - - return compressed + self._store_cache_entry( + thread_id, + entry := _SummaryCacheEntry( + summarized_ids=summarized_ids, + summary_text=summary, + preserved_ids=frozenset(preserved_ids), + preserved_messages=tuple(preserved), + ), + ) + self._pending_state_cache_update = _serialize_cache_entry(entry) + + return compressed def _get_cache_entry(self, thread_id: str) -> _SummaryCacheEntry | None: """Return the cached entry for ``thread_id`` (thread-safe read).""" @@ -503,7 +571,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): return None return _resolve_thread_id(runtime) - def _summary_message(self, summary: str) -> HumanMessage: + def _summary_message(self, summary: str) -> SystemMessage: """Build the summary message handed to the model. It is named ``summary`` and flagged ``hide_from_ui`` defensively. Because @@ -511,14 +579,21 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): state), the frontend never receives it — the user keeps seeing the original conversation. """ - return HumanMessage( - content=f"Here is a summary of the conversation to date:\n\n{summary}", + return SystemMessage( + content=( + "Internal conversation summary for continuity only. Use it silently " + "to answer the user's latest request. Never reveal, quote, mention, " + "or describe this summary or the fact that context was compacted.\n\n" + "\n" + f"{summary}\n" + "" + ), name="summary", additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"}, ) @override - def _build_new_messages(self, summary: str) -> list[HumanMessage]: + def _build_new_messages(self, summary: str) -> list[AnyMessage]: """Kept for backwards compatibility; delegates to :meth:`_summary_message`.""" return [self._summary_message(summary)] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/thread_state.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/thread_state.py index cdd3f95d..c1b9948b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/thread_state.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/thread_state.py @@ -83,5 +83,6 @@ class ThreadState(AgentState): highlighted_code: NotRequired[Optional[str]] canvas_mode: NotRequired[Optional[Literal["text", "code"]]] quick_action_id: NotRequired[Optional[str]] - web_search_enabled: NotRequired[bool] - web_search_results: NotRequired[Optional[list[dict]]] + web_search_enabled: NotRequired[bool] + web_search_results: NotRequired[Optional[list[dict]]] + context_summary_cache: NotRequired[dict | None] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py index 205b88e4..cb277ee0 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py @@ -387,17 +387,26 @@ def _get_config_mtime(config_path: Path) -> float | None: return None -def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig: - """Load config from disk and refresh cache metadata.""" - global _app_config, _app_config_path, _app_config_mtime, _app_config_skill_evolution_overrides_mtime, _app_config_is_custom - - resolved_path = AppConfig.resolve_config_path(config_path) - _app_config = AppConfig.from_file(str(resolved_path)) - _app_config_path = resolved_path - _app_config_mtime = _get_config_mtime(resolved_path) - _app_config_skill_evolution_overrides_mtime = _get_skill_evolution_overrides_mtime() - _app_config_is_custom = False - return _app_config +def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig: + """Load config from disk and refresh cache metadata.""" + global _app_config, _app_config_path, _app_config_mtime, _app_config_skill_evolution_overrides_mtime, _app_config_is_custom + + resolved_path = AppConfig.resolve_config_path(config_path) + _app_config = AppConfig.from_file(str(resolved_path)) + _app_config_path = resolved_path + _app_config_mtime = _get_config_mtime(resolved_path) + _app_config_skill_evolution_overrides_mtime = _get_skill_evolution_overrides_mtime() + _app_config_is_custom = False + summarization = _app_config.summarization + logger.info( + "Loaded AppConfig from %s; summarization enabled=%s trigger=%s keep=%s trim_tokens_to_summarize=%s", + resolved_path, + summarization.enabled, + summarization.trigger, + summarization.keep, + summarization.trim_tokens_to_summarize, + ) + return _app_config def get_app_config() -> AppConfig: diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml index be0d5b21..0bc2744e 100644 --- a/offline-backend-20260512/config.yaml +++ b/offline-backend-20260512/config.yaml @@ -310,20 +310,17 @@ summarization: # model_name: deepseek-chat trigger: - type: tokens - value: 8000 + value: 120000 - type: messages - value: 12 + value: 160 # Keep a small *token* budget of recent context (same unit as the token # trigger, well below it). This is what the model receives after compaction: - # [summary, ~16000 tokens of recent turns]. Small keep = small model context + # [summary, ~40000 tokens of recent turns]. Small keep = small model context # (fast, no overflow) AND infrequent re-compaction — the preserved window - # grows from ~16000 up to the 60000 trigger before folding into a new summary, + # grows from ~40000 up to the 120000 trigger before folding into a new summary, # which spans many turns. The full conversation still stays in state for the # page; only what is sent to the model is compressed. keep: - type: messages # 测试值:压缩后只保留最近 6 条原始消息 - value: 6 - # 生产值改用 token 预算: - # type: tokens - # value: 16000 - trim_tokens_to_summarize: 4000 + type: tokens + value: 40000 + trim_tokens_to_summarize: 8000