wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 上下文压缩支持状态持久化缓存,前端过滤隐藏消息

Browse Source 
- 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) <noreply@anthropic.com>
This commit is contained in:
AdministratorandClaude Opus 4.8 committed 2026-06-16 22:16:06 +08:00
1 parent 5f39879e4d
commit 5b610fd807
7 files changed
+159 -71
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend-web/src/components/workspace/input-box.tsx
+6 -4
View File
@@ -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";
frontend-web/src/components/workspace/messages/message-list.tsx
+8 -4
View File
@@ -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
offline-backend-20260512/backend/config.yaml
+5 -5
View File
@@ -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
offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py
+110 -35
View File
@@ -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"
"<internal_context_summary>\n"
f"{summary}\n"
"</internal_context_summary>"
),
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)]
offline-backend-20260512/backend/packages/harness/deerflow/agents/thread_state.py
+3 -2
View File
@@ -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]
offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py
+20 -11
View File
@@ -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:
offline-backend-20260512/config.yaml
+7 -10
View File
@@ -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