This commit is contained in:
fengxiangdebaozi committed 2026-06-16 22:18:22 +08:00
commit d8102e8db8
7 files changed
+121 -33

No files matched your search

@@ -62,6 +62,7 @@ import {
import { apiFetch } from "@/core/api"; import { apiFetch } from "@/core/api";
import { getBackendBaseURL } from "@/core/config"; import { getBackendBaseURL } from "@/core/config";
import { useI18n } from "@/core/i18n/hooks"; import { useI18n } from "@/core/i18n/hooks";
import { isHiddenFromUIMessage } from "@/core/messages/utils";
import { useModels } from "@/core/models/hooks"; import { useModels } from "@/core/models/hooks";
import type { AgentThreadContext } from "@/core/threads"; import type { AgentThreadContext } from "@/core/threads";
import { textOfMessage } from "@/core/threads/utils"; import { textOfMessage } from "@/core/threads/utils";
@@ -467,14 +468,15 @@ export function InputBox({
return; 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; const lastAiId = lastAi?.id ?? null;
if (!lastAiId || lastAiId === lastGeneratedForAiIdRef.current) { if (!lastAiId || lastAiId === lastGeneratedForAiIdRef.current) {
return; return;
} }
lastGeneratedForAiIdRef.current = lastAiId; lastGeneratedForAiIdRef.current = lastAiId;
const recent = thread.messages const recent = visibleMessages
.filter((m) => m.type === "human" || m.type === "ai") .filter((m) => m.type === "human" || m.type === "ai")
.map((m) => { .map((m) => {
const role = m.type === "human" ? "user" : "assistant"; const role = m.type === "human" ? "user" : "assistant";
@@ -22,6 +22,7 @@ import {
hasPresentFiles, hasPresentFiles,
hasReasoning, hasReasoning,
hasToolCalls, hasToolCalls,
isHiddenFromUIMessage,
} from "@/core/messages/utils"; } from "@/core/messages/utils";
import { useRehypeSplitWordsIntoSpans } from "@/core/rehype"; import { useRehypeSplitWordsIntoSpans } from "@/core/rehype";
import { isThreadRagModeLocked, useLocalSettings } from "@/core/settings"; import { isThreadRagModeLocked, useLocalSettings } from "@/core/settings";
@@ -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( const messages = useMemo(
() => () =>
hiddenMessageIds.size === 0 hiddenMessageIds.size === 0
+5 -5
View File
@@ -207,13 +207,13 @@ summarization:
# model_name: deepseek-chat # 留空则用主对话模型 # model_name: deepseek-chat # 留空则用主对话模型
trigger: trigger:
- type: tokens - type: tokens
value: 8000 value: 120000
- type: messages - type: messages
value: 12 value: 160
keep: keep:
type: messages type: tokens
value: 6 value: 40000
trim_tokens_to_summarize: 4000 trim_tokens_to_summarize: 8000
tool_groups: tool_groups:
- name: web - name: web
@@ -27,7 +27,14 @@ from typing import Any, Protocol, override, runtime_checkable
from langchain.agents import AgentState from langchain.agents import AgentState
from langchain.agents.middleware import SummarizationMiddleware from langchain.agents.middleware import SummarizationMiddleware
from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse 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.config import get_config
from langgraph.runtime import Runtime from langgraph.runtime import Runtime
@@ -138,7 +145,8 @@ class _SummaryCacheEntry:
summarized_ids: frozenset[str] summarized_ids: frozenset[str]
summary_text: str summary_text: str
reusable: bool preserved_ids: frozenset[str] = frozenset()
preserved_messages: tuple[AnyMessage, ...] = ()
# Process-wide summary cache shared across every middleware instance. # Process-wide summary cache shared across every middleware instance.
@@ -152,8 +160,9 @@ class _SummaryCacheEntry:
# so only the first threshold crossing (or a genuine re-compaction) pays for an # 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 # LLM call. A lock guards the non-atomic LRU eviction against concurrent
# threads (subagents / parallel conversations share this process). # threads (subagents / parallel conversations share this process).
_GLOBAL_SUMMARY_CACHE: "OrderedDict[str, _SummaryCacheEntry]" = OrderedDict() _GLOBAL_SUMMARY_CACHE: OrderedDict[str, _SummaryCacheEntry] = OrderedDict()
_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock() _GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock()
_STATE_CACHE_KEY = "context_summary_cache"
def _clear_global_summary_cache() -> None: def _clear_global_summary_cache() -> None:
@@ -162,6 +171,42 @@ def _clear_global_summary_cache() -> None:
_GLOBAL_SUMMARY_CACHE.clear() _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): class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
"""Summarization middleware with pre-compression hook dispatch and skill rescue.""" """Summarization middleware with pre-compression hook dispatch and skill rescue."""
@@ -187,6 +232,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
# Points at the *process-wide* cache (not a fresh per-instance dict) so a # Points at the *process-wide* cache (not a fresh per-instance dict) so a
# summary survives across runs/turns — see ``_GLOBAL_SUMMARY_CACHE``. # 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 @override
def before_model(self, state: AgentState, runtime: Runtime) -> dict | None: def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:
@@ -202,6 +248,20 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
"""Async counterpart of :meth:`before_model` — also a no-op.""" """Async counterpart of :meth:`before_model` — also a no-op."""
return None return None
@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 ---------- # ---- wrap_model_call: transient, display-preserving compression ----------
@override @override
@@ -319,10 +379,16 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
if thread_id is None: if thread_id is None:
logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call") 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 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} present_ids = {m.id for m in messages}
cache_usable = cached is not None and cached.reusable and cached.summarized_ids <= present_ids cache_usable = cached is not None and cached.summarized_ids <= present_ids and cached.preserved_ids <= present_ids
if cache_usable: if cache_usable:
preserved = [m for m in messages if m.id not in cached.summarized_ids] 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] effective = [self._summary_message(cached.summary_text), *preserved]
# Decide re-summarization on the *counted* size of the compressed # Decide re-summarization on the *counted* size of the compressed
# context only — NOT ``_should_summarize`` — because that folds in the # context only — NOT ``_should_summarize`` — because that folds in the
@@ -349,7 +415,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
# an "it compacts every turn" report is diagnosable straight from the # an "it compacts every turn" report is diagnosable straight from the
# backend log: cache=miss → thread_id/key problem; unusable → message-id # backend log: cache=miss → thread_id/key problem; unusable → message-id
# instability; stale → preserved genuinely grew back over the trigger. # 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( logger.info(
"summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d", "summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d",
thread_id, thread_id,
@@ -390,14 +456,16 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
if thread_id: if thread_id:
summarized_ids = frozenset(m.id for m in messages_to_summarize if m.id is not None) 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} 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( self._store_cache_entry(
thread_id, thread_id,
_SummaryCacheEntry(summarized_ids=summarized_ids, summary_text=summary, reusable=reusable), 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 return compressed
@@ -503,7 +571,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
return None return None
return _resolve_thread_id(runtime) 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. """Build the summary message handed to the model.
It is named ``summary`` and flagged ``hide_from_ui`` defensively. Because 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 state), the frontend never receives it — the user keeps seeing the
original conversation. original conversation.
""" """
return HumanMessage( return SystemMessage(
content=f"Here is a summary of the conversation to date:\n\n{summary}", 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", name="summary",
additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"}, additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"},
) )
@override @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`.""" """Kept for backwards compatibility; delegates to :meth:`_summary_message`."""
return [self._summary_message(summary)] return [self._summary_message(summary)]
@@ -85,3 +85,4 @@ class ThreadState(AgentState):
quick_action_id: NotRequired[Optional[str]] quick_action_id: NotRequired[Optional[str]]
web_search_enabled: NotRequired[bool] web_search_enabled: NotRequired[bool]
web_search_results: NotRequired[Optional[list[dict]]] web_search_results: NotRequired[Optional[list[dict]]]
context_summary_cache: NotRequired[dict | None]
@@ -397,6 +397,15 @@ def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
_app_config_mtime = _get_config_mtime(resolved_path) _app_config_mtime = _get_config_mtime(resolved_path)
_app_config_skill_evolution_overrides_mtime = _get_skill_evolution_overrides_mtime() _app_config_skill_evolution_overrides_mtime = _get_skill_evolution_overrides_mtime()
_app_config_is_custom = False _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 return _app_config
+7 -10
View File
@@ -310,20 +310,17 @@ summarization:
# model_name: deepseek-chat # model_name: deepseek-chat
trigger: trigger:
- type: tokens - type: tokens
value: 8000 value: 120000
- type: messages - type: messages
value: 12 value: 160
# Keep a small *token* budget of recent context (same unit as the token # 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: # 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 # (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 # which spans many turns. The full conversation still stays in state for the
# page; only what is sent to the model is compressed. # page; only what is sent to the model is compressed.
keep: keep:
type: messages # 测试值:压缩后只保留最近 6 条原始消息 type: tokens
value: 6 value: 40000
# 生产值改用 token 预算: trim_tokens_to_summarize: 8000
# type: tokens
# value: 16000
trim_tokens_to_summarize: 4000