diff --git a/frontend-web/src/core/messages/utils.ts b/frontend-web/src/core/messages/utils.ts index e030a213..2b6ea765 100644 --- a/frontend-web/src/core/messages/utils.ts +++ b/frontend-web/src/core/messages/utils.ts @@ -434,7 +434,17 @@ export function findToolCallResult( } export function isHiddenFromUIMessage(message: Message) { - return message.additional_kwargs?.hide_from_ui === true; + if (message.additional_kwargs?.hide_from_ui === true) { + return true; + } + // Conversation-summary messages are context for the model only — never shown + // to the user. New summaries are injected transiently (never persisted), but + // threads compacted by older builds still carry a persisted `name: "summary"` + // HumanMessage, so hide those too. + if (message.name === "summary") { + return true; + } + return false; } /** diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index e8944de1..cab36879 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -163,7 +163,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har 6. **GuardrailMiddleware** - Pre-tool-call authorization via pluggable `GuardrailProvider` protocol (optional, if `guardrails.enabled` in config). Evaluates each tool call and returns error ToolMessage on deny. Three provider options: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom providers. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) for setup, usage, and how to implement a provider. 7. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution continues 8. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting -9. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled) +9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` — the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py` 10. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode) 11. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional) 12. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model @@ -175,7 +175,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har 18. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) 8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation — timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills///SKILL.md` path (per the `` prompt block — there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. **Per-run skill merging**: `SqlToolMetricsStore.record` upserts skill rows by `(run_id, skill_name)` instead of inserting one row per call — a skill touched by several tool calls / retries inside one run collapses to a single row. Any success in the run wins (a skill that failed then finally succeeded is recorded as one `success`); a skill that only ever failed yields one `error` row whose `error_message` accumulates every attempt's message; durations are summed. Non-skill tool calls, and skill calls with no `run_id`, keep one row each. The read→merge→write is guarded by a per-event-loop `asyncio.Lock`. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown). 9. **PromptPrefixMiddleware** - Injects the admin-configured prompt prefix into the model request only. The prefix is tagged onto the first human message's `additional_kwargs['prompt_prefix']` by the Gateway (`app.gateway.services`) and prepended to content here at call time, so the persisted/displayed message stays prefix-free -10. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled) +10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` — the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py` 11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode) 12. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional) 13. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model diff --git a/offline-backend-20260512/backend/docs/summarization.md b/offline-backend-20260512/backend/docs/summarization.md index c33a6895..a3037a0f 100644 --- a/offline-backend-20260512/backend/docs/summarization.md +++ b/offline-backend-20260512/backend/docs/summarization.md @@ -4,13 +4,26 @@ DeerFlow includes automatic conversation summarization to handle long conversati ## Overview -The summarization feature uses LangChain's `SummarizationMiddleware` to monitor conversation history and trigger summarization based on configurable thresholds. When activated, it: +The summarization feature uses DeerFlow's `DeerFlowSummarizationMiddleware` (a subclass of LangChain's `SummarizationMiddleware`) to monitor conversation history and trigger summarization based on configurable thresholds. When activated, it: 1. Monitors message token counts in real-time 2. Triggers summarization when thresholds are met 3. Keeps recent messages intact while summarizing older exchanges 4. Maintains AI/Tool message pairs together for context continuity -5. Injects the summary back into the conversation +5. Substitutes the summary for the older messages **in the model request only** + +### Non-destructive (display-preserving) compression + +> **Important:** DeerFlow compresses **transiently**, not destructively. + +LangChain's stock `SummarizationMiddleware` runs in `before_model` and persists a `RemoveMessage(REMOVE_ALL_MESSAGES)` to the checkpoint — which permanently deletes the original Q&A from both the model context *and* the conversation the frontend renders. DeerFlow overrides this: + +- `before_model` / `abefore_model` are **no-ops** (the destructive path is disabled). +- Compression runs in `wrap_model_call` / `awrap_model_call`: for a single model call, the message list is replaced with `[summary, *recent]` via `request.override(messages=...)`. **Nothing is written back to state.** +- The full original messages stay in the checkpoint, so the frontend keeps showing the user's real conversation. The summary message is named `summary` and flagged `hide_from_ui`, and because it only ever lives in the transient request it never reaches the frontend at all. +- The generated summary is cached **per-thread** so we don't pay an extra LLM call on every model invocation. The cached summary is reused (no LLM call) until enough *new* messages accumulate to push the compressed context back over the trigger threshold (a sticky boundary mirroring the stock middleware's post-compaction behavior). The cache is bounded (oldest threads evicted first). + +> **Legacy threads:** conversations compacted by an older (destructive) build already lost their original messages and carry a persisted `summary` message. The frontend hides any `name: "summary"` message, so the stray summary bubble no longer shows — but those already-discarded messages cannot be recovered. ## Configuration @@ -169,10 +182,7 @@ The default LangChain prompt instructs the model to: - Messages to summarize (older messages beyond the `keep` threshold) - Messages to preserve (recent messages within the `keep` threshold) 4. **Summary Generation**: The model generates a concise summary of the older messages -5. **Context Replacement**: The message history is updated: - - All old messages are removed - - A single summary message is added - - Recent messages are preserved +5. **Context Replacement (transient)**: The message list **handed to the model for this call** is rebuilt as `[summary, *recent]`. The persisted state is left untouched — old messages are **not** removed, so the frontend still renders the full original conversation. 6. **AI/Tool Pair Protection**: The system ensures AI messages and their corresponding tool messages stay together 7. **Skill Rescue**: Before the summary is generated, the most recently loaded skill files (tool results whose tool name is in `skill_file_read_tool_names` and whose target path is under `skills.container_path`) are lifted out of the summarization set and prepended to the preserved tail. Selection walks newest-first under three budgets: `preserve_recent_skill_count`, `preserve_recent_skill_tokens`, and `preserve_recent_skill_tokens_per_skill`. The triggering AIMessage and all of its paired ToolMessages move together so tool_call ↔ tool_result pairing stays intact. @@ -298,8 +308,9 @@ The middleware intelligently preserves message context: ### Code Structure - **Configuration**: `packages/harness/deerflow/config/summarization_config.py` -- **Integration**: `packages/harness/deerflow/agents/lead_agent/agent.py` -- **Middleware**: Uses `langchain.agents.middleware.SummarizationMiddleware` +- **Integration**: `packages/harness/deerflow/agents/lead_agent/agent.py` (`_create_summarization_middleware`) +- **Middleware**: `packages/harness/deerflow/agents/middlewares/summarization_middleware.py` — `DeerFlowSummarizationMiddleware`, subclassing `langchain.agents.middleware.SummarizationMiddleware` but overriding compaction to be transient (`wrap_model_call`) instead of destructive (`before_model`) +- **Tests**: `tests/test_summarization_middleware.py` ### Middleware Order @@ -313,9 +324,9 @@ Summarization runs after ThreadData and Sandbox initialization but before Title ### State Management -- Summarization is stateless - configuration is loaded once at startup -- Summaries are added as regular messages in the conversation history -- The checkpointer persists the summarized history automatically +- Configuration is loaded once at startup +- Summarization is **non-destructive**: the summary replaces older messages only in the transient model request (`wrap_model_call`), never in persisted state. The checkpointer therefore keeps the full original conversation +- The per-thread summary cache lives in the middleware instance (in-memory, bounded). It is a cost optimization only — a cold/stale cache simply re-summarizes, which is always correct ## Example Configurations 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 f2443b3f..276f7adf 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 @@ -1,21 +1,41 @@ -"""Summarization middleware extensions for DeerFlow.""" +"""Summarization middleware extensions for DeerFlow. + +Unlike LangChain's stock ``SummarizationMiddleware`` (which runs in +``before_model`` and **persists** a destructive ``RemoveMessage(REMOVE_ALL_MESSAGES)`` +to the checkpoint — wiping the original Q&A from both the model context *and* +the conversation the frontend renders), this middleware compresses the history +**transiently in ``wrap_model_call``**. The full original messages stay in state +(so the UI keeps showing the user's real conversation); only the message list +handed to the model for a single call is replaced with ``[summary, *recent]``. + +The generated summary is cached per-thread so we don't pay an extra LLM call on +every model invocation: once a thread is summarized, the cached summary is reused +until enough *new* messages accumulate to push the compressed context back over +the trigger threshold (sticky boundary, mirroring the stock middleware's +post-compaction behavior). +""" from __future__ import annotations import logging -from collections.abc import Collection +from collections import OrderedDict +from collections.abc import Awaitable, Callable, Collection from dataclasses import dataclass from typing import Any, Protocol, override, runtime_checkable from langchain.agents import AgentState from langchain.agents.middleware import SummarizationMiddleware -from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage, ToolMessage +from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse +from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage from langgraph.config import get_config -from langgraph.graph.message import REMOVE_ALL_MESSAGES from langgraph.runtime import Runtime logger = logging.getLogger(__name__) +# Upper bound on the per-thread summary cache so a long-lived process serving +# many threads can't grow it without limit. Oldest entries are evicted first. +_SUMMARY_CACHE_MAX_THREADS = 512 + @dataclass(frozen=True) class SummarizationEvent: @@ -95,6 +115,22 @@ class _SkillBundle: skill_key: str +@dataclass(frozen=True) +class _SummaryCacheEntry: + """A cached, reusable summary for one thread. + + ``summarized_ids`` are the ids of the original messages folded into + ``summary_text``. The cache is only reusable while every one of those ids is + still present in the live message list and no AIMessage/ToolMessage pair was + split when the boundary was drawn (``reusable``) — otherwise filtering by id + could orphan a tool result, so we fall back to re-summarizing. + """ + + summarized_ids: frozenset[str] + summary_text: str + reusable: bool + + class DeerFlowSummarizationMiddleware(SummarizationMiddleware): """Summarization middleware with pre-compression hook dispatch and skill rescue.""" @@ -116,69 +152,181 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): self._preserve_recent_skill_count = max(0, preserve_recent_skill_count) self._preserve_recent_skill_tokens = max(0, preserve_recent_skill_tokens) self._preserve_recent_skill_tokens_per_skill = max(0, preserve_recent_skill_tokens_per_skill) + # Per-thread cache of the most recent summary: thread_id -> (summarized message ids, summary text). + # Lets us reuse a summary across model calls instead of re-running the LLM every invocation. + self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = OrderedDict() + @override def before_model(self, state: AgentState, runtime: Runtime) -> dict | None: - return self._maybe_summarize(state, runtime) + """Disable the stock middleware's destructive ``before_model`` compaction. + Compression now happens transiently in :meth:`wrap_model_call`, so the + persisted conversation (and the frontend's view of it) is never mutated. + """ + return None + + @override async def abefore_model(self, state: AgentState, runtime: Runtime) -> dict | None: - return await self._amaybe_summarize(state, runtime) + """Async counterpart of :meth:`before_model` — also a no-op.""" + return None - def _maybe_summarize(self, state: AgentState, runtime: Runtime) -> dict | None: - messages = state["messages"] - self._ensure_message_ids(messages) + # ---- wrap_model_call: transient, display-preserving compression ---------- - total_tokens = self.token_counter(messages) - if not self._should_summarize(messages, total_tokens): + @override + def wrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], ModelResponse], + ) -> ModelCallResult: + compressed = self._compressed_messages(request) + if compressed is not None: + request = request.override(messages=compressed) + return handler(request) + + @override + async def awrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], Awaitable[ModelResponse]], + ) -> ModelCallResult: + compressed = await self._acompressed_messages(request) + if compressed is not None: + request = request.override(messages=compressed) + return await handler(request) + + def _compressed_messages(self, request: ModelRequest) -> list[AnyMessage] | None: + """Build the compressed message list for this call (sync summary path). + + Returns ``None`` when no compression is needed (the request is passed + through untouched). + """ + plan = self._plan_compression(request) + if plan is None: return None - - cutoff_index = self._determine_cutoff_index(messages) - if cutoff_index <= 0: - return None - - messages_to_summarize, preserved_messages = self._partition_with_skill_rescue(messages, cutoff_index) - self._fire_hooks(messages_to_summarize, preserved_messages, runtime) + action, payload = plan + if action == "reuse": + return payload # type: ignore[return-value] + # action == "summarize" + messages_to_summarize, preserved, thread_id = payload # type: ignore[misc] + self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None)) summary = self._create_summary(messages_to_summarize) - new_messages = self._build_new_messages(summary) + return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id) - return { - "messages": [ - RemoveMessage(id=REMOVE_ALL_MESSAGES), - *new_messages, - *preserved_messages, - ] - } + async def _acompressed_messages(self, request: ModelRequest) -> list[AnyMessage] | None: + """Async counterpart of :meth:`_compressed_messages`.""" + plan = self._plan_compression(request) + if plan is None: + return None + action, payload = plan + if action == "reuse": + return payload # type: ignore[return-value] + messages_to_summarize, preserved, thread_id = payload # type: ignore[misc] + self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None)) + summary = await self._acreate_summary(messages_to_summarize) + return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id) - async def _amaybe_summarize(self, state: AgentState, runtime: Runtime) -> dict | None: - messages = state["messages"] + def _plan_compression( + self, + request: ModelRequest, + ) -> tuple[str, Any] | None: + """Decide what to do for this model call, doing **no** LLM work. + + Returns one of: + + - ``None`` — pass the request through unchanged (no compression needed). + - ``("reuse", compressed_messages)`` — reuse the cached summary. + - ``("summarize", (messages_to_summarize, preserved, thread_id))`` — the + caller must run the (a)sync summarizer, then call + :meth:`_finish_summarize`. + """ + messages = list(getattr(request, "messages", None) or []) + if not messages: + return None self._ensure_message_ids(messages) + thread_id = self._resolve_request_thread_id(request) - total_tokens = self.token_counter(messages) - if not self._should_summarize(messages, total_tokens): + cached = self._summary_cache.get(thread_id) if thread_id else None + present_ids = {m.id for m in messages} + if cached is not None and cached.reusable and cached.summarized_ids <= present_ids: + preserved = [m for m in messages if m.id not in cached.summarized_ids] + effective = [self._summary_message(cached.summary_text), *preserved] + if not self._should_summarize(effective, self.token_counter(effective)): + # Refresh LRU recency and reuse without an LLM call. + self._summary_cache.move_to_end(thread_id) + return ("reuse", effective) + # New messages have pushed the compressed context back over the + # threshold — fall through and re-summarize a larger prefix. + + if not self._should_summarize(messages, self.token_counter(messages)): return None cutoff_index = self._determine_cutoff_index(messages) if cutoff_index <= 0: return None - messages_to_summarize, preserved_messages = self._partition_with_skill_rescue(messages, cutoff_index) - self._fire_hooks(messages_to_summarize, preserved_messages, runtime) - summary = await self._acreate_summary(messages_to_summarize) - new_messages = self._build_new_messages(summary) + messages_to_summarize, preserved = self._partition_with_skill_rescue(messages, cutoff_index) + if not messages_to_summarize: + return None - return { - "messages": [ - RemoveMessage(id=REMOVE_ALL_MESSAGES), - *new_messages, - *preserved_messages, - ] - } + return ("summarize", (messages_to_summarize, preserved, thread_id)) + + def _finish_summarize( + self, + messages_to_summarize: list[AnyMessage], + preserved: list[AnyMessage], + summary: str, + thread_id: str | None, + ) -> list[AnyMessage]: + """Assemble the compressed message list and update the per-thread cache.""" + compressed = [self._summary_message(summary), *preserved] + + 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 + + def _store_cache_entry(self, thread_id: str, entry: _SummaryCacheEntry) -> None: + """Insert/refresh a cache entry, evicting the oldest thread if over capacity.""" + self._summary_cache[thread_id] = entry + self._summary_cache.move_to_end(thread_id) + while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS: + self._summary_cache.popitem(last=False) + + @staticmethod + def _resolve_request_thread_id(request: ModelRequest) -> str | None: + """Best-effort thread id from the model request's runtime.""" + runtime = getattr(request, "runtime", None) + if runtime is None: + return None + return _resolve_thread_id(runtime) + + def _summary_message(self, summary: str) -> HumanMessage: + """Build the summary message handed to the model. + + It is named ``summary`` and flagged ``hide_from_ui`` defensively. Because + it only ever lives in the transient model request (never in persisted + 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}", + name="summary", + additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"}, + ) @override def _build_new_messages(self, summary: str) -> list[HumanMessage]: - """Override the base implementation to let the human message with the special name 'summary'. - And this message will be ignored to display in the frontend, but still can be used as context for the model. - """ - return [HumanMessage(content=f"Here is a summary of the conversation to date:\n\n{summary}", name="summary")] + """Kept for backwards compatibility; delegates to :meth:`_summary_message`.""" + return [self._summary_message(summary)] def _partition_with_skill_rescue( self,