feat: flush memory before summarization (#2176)

* feat: flush memory before summarization * fix: keep agent-scoped memory on summarization flush * fix: harden summarization hook plumbing * fix: address summarization review feedback * style: format memory middleware
2026-05-24 08:55:59 +00:00 · 2026-04-14 15:01:06 +08:00
parent e4f896e90d
commit 4ba3167f48
10 changed files with 667 additions and 189 deletions
@@ -0,0 +1,109 @@
+"""Shared helpers for turning conversations into memory update inputs."""
+
+from __future__ import annotations
+
+import re
+from copy import copy
+from typing import Any
+
+_UPLOAD_BLOCK_RE = re.compile(r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", re.IGNORECASE)
+_CORRECTION_PATTERNS = (
+    re.compile(r"\bthat(?:'s| is) (?:wrong|incorrect)\b", re.IGNORECASE),
+    re.compile(r"\byou misunderstood\b", re.IGNORECASE),
+    re.compile(r"\btry again\b", re.IGNORECASE),
+    re.compile(r"\bredo\b", re.IGNORECASE),
+    re.compile(r"不对"),
+    re.compile(r"你理解错了"),
+    re.compile(r"你理解有误"),
+    re.compile(r"重试"),
+    re.compile(r"重新来"),
+    re.compile(r"换一种"),
+    re.compile(r"改用"),
+)
+_REINFORCEMENT_PATTERNS = (
+    re.compile(r"\byes[,.]?\s+(?:exactly|perfect|that(?:'s| is) (?:right|correct|it))\b", re.IGNORECASE),
+    re.compile(r"\bperfect(?:[.!?]|$)", re.IGNORECASE),
+    re.compile(r"\bexactly\s+(?:right|correct)\b", re.IGNORECASE),
+    re.compile(r"\bthat(?:'s| is)\s+(?:exactly\s+)?(?:right|correct|what i (?:wanted|needed|meant))\b", re.IGNORECASE),
+    re.compile(r"\bkeep\s+(?:doing\s+)?that\b", re.IGNORECASE),
+    re.compile(r"\bjust\s+(?:like\s+)?(?:that|this)\b", re.IGNORECASE),
+    re.compile(r"\bthis is (?:great|helpful)\b(?:[.!?]|$)", re.IGNORECASE),
+    re.compile(r"\bthis is what i wanted\b(?:[.!?]|$)", re.IGNORECASE),
+    re.compile(r"对[，,]?\s*就是这样(?:[。！？!?.]|$)"),
+    re.compile(r"完全正确(?:[。！？!?.]|$)"),
+    re.compile(r"(?:对[，,]?\s*)?就是这个意思(?:[。！？!?.]|$)"),
+    re.compile(r"正是我想要的(?:[。！？!?.]|$)"),
+    re.compile(r"继续保持(?:[。！？!?.]|$)"),
+)
+
+
+def extract_message_text(message: Any) -> str:
+    """Extract plain text from message content for filtering and signal detection."""
+    content = getattr(message, "content", "")
+    if isinstance(content, list):
+        text_parts: list[str] = []
+        for part in content:
+            if isinstance(part, str):
+                text_parts.append(part)
+            elif isinstance(part, dict):
+                text_val = part.get("text")
+                if isinstance(text_val, str):
+                    text_parts.append(text_val)
+        return " ".join(text_parts)
+    return str(content)
+
+
+def filter_messages_for_memory(messages: list[Any]) -> list[Any]:
+    """Keep only user inputs and final assistant responses for memory updates."""
+    filtered = []
+    skip_next_ai = False
+    for msg in messages:
+        msg_type = getattr(msg, "type", None)
+
+        if msg_type == "human":
+            content_str = extract_message_text(msg)
+            if "<uploaded_files>" in content_str:
+                stripped = _UPLOAD_BLOCK_RE.sub("", content_str).strip()
+                if not stripped:
+                    skip_next_ai = True
+                    continue
+                clean_msg = copy(msg)
+                clean_msg.content = stripped
+                filtered.append(clean_msg)
+                skip_next_ai = False
+            else:
+                filtered.append(msg)
+                skip_next_ai = False
+        elif msg_type == "ai":
+            tool_calls = getattr(msg, "tool_calls", None)
+            if not tool_calls:
+                if skip_next_ai:
+                    skip_next_ai = False
+                    continue
+                filtered.append(msg)
+
+    return filtered
+
+
+def detect_correction(messages: list[Any]) -> bool:
+    """Detect explicit user corrections in recent conversation turns."""
+    recent_user_msgs = [msg for msg in messages[-6:] if getattr(msg, "type", None) == "human"]
+
+    for msg in recent_user_msgs:
+        content = extract_message_text(msg).strip()
+        if content and any(pattern.search(content) for pattern in _CORRECTION_PATTERNS):
+            return True
+
+    return False
+
+
+def detect_reinforcement(messages: list[Any]) -> bool:
+    """Detect explicit positive reinforcement signals in recent conversation turns."""
+    recent_user_msgs = [msg for msg in messages[-6:] if getattr(msg, "type", None) == "human"]
+
+    for msg in recent_user_msgs:
+        content = extract_message_text(msg).strip()
+        if content and any(pattern.search(content) for pattern in _REINFORCEMENT_PATTERNS):
+            return True
+
+    return False
@@ -61,48 +61,88 @@ class MemoryUpdateQueue:
            return

        with self._lock:
-            existing_context = next(
-                (context for context in self._queue if context.thread_id == thread_id),
-                None,
-            )
-            merged_correction_detected = correction_detected or (existing_context.correction_detected if existing_context is not None else False)
-            merged_reinforcement_detected = reinforcement_detected or (existing_context.reinforcement_detected if existing_context is not None else False)
-            context = ConversationContext(
+            self._enqueue_locked(
                thread_id=thread_id,
                messages=messages,
                agent_name=agent_name,
-                correction_detected=merged_correction_detected,
-                reinforcement_detected=merged_reinforcement_detected,
+                correction_detected=correction_detected,
+                reinforcement_detected=reinforcement_detected,
            )
-
-            # Check if this thread already has a pending update
-            # If so, replace it with the newer one
-            self._queue = [c for c in self._queue if c.thread_id != thread_id]
-            self._queue.append(context)
-
-            # Reset or start the debounce timer
            self._reset_timer()

        logger.info("Memory update queued for thread %s, queue size: %d", thread_id, len(self._queue))

+    def add_nowait(
+        self,
+        thread_id: str,
+        messages: list[Any],
+        agent_name: str | None = None,
+        correction_detected: bool = False,
+        reinforcement_detected: bool = False,
+    ) -> None:
+        """Add a conversation and start processing immediately in the background."""
+        config = get_memory_config()
+        if not config.enabled:
+            return
+
+        with self._lock:
+            self._enqueue_locked(
+                thread_id=thread_id,
+                messages=messages,
+                agent_name=agent_name,
+                correction_detected=correction_detected,
+                reinforcement_detected=reinforcement_detected,
+            )
+            self._schedule_timer(0)
+
+        logger.info("Memory update queued for immediate processing on thread %s, queue size: %d", thread_id, len(self._queue))
+
+    def _enqueue_locked(
+        self,
+        *,
+        thread_id: str,
+        messages: list[Any],
+        agent_name: str | None,
+        correction_detected: bool,
+        reinforcement_detected: bool,
+    ) -> None:
+        existing_context = next(
+            (context for context in self._queue if context.thread_id == thread_id),
+            None,
+        )
+        merged_correction_detected = correction_detected or (existing_context.correction_detected if existing_context is not None else False)
+        merged_reinforcement_detected = reinforcement_detected or (existing_context.reinforcement_detected if existing_context is not None else False)
+        context = ConversationContext(
+            thread_id=thread_id,
+            messages=messages,
+            agent_name=agent_name,
+            correction_detected=merged_correction_detected,
+            reinforcement_detected=merged_reinforcement_detected,
+        )
+
+        self._queue = [c for c in self._queue if c.thread_id != thread_id]
+        self._queue.append(context)
+
    def _reset_timer(self) -> None:
        """Reset the debounce timer."""
        config = get_memory_config()
+        self._schedule_timer(config.debounce_seconds)

+        logger.debug("Memory update timer set for %ss", config.debounce_seconds)
+
+    def _schedule_timer(self, delay_seconds: float) -> None:
+        """Schedule queue processing after the provided delay."""
        # Cancel existing timer if any
        if self._timer is not None:
            self._timer.cancel()

-        # Start new timer
        self._timer = threading.Timer(
-            config.debounce_seconds,
+            delay_seconds,
            self._process_queue,
        )
        self._timer.daemon = True
        self._timer.start()

-        logger.debug("Memory update timer set for %ss", config.debounce_seconds)
-
    def _process_queue(self) -> None:
        """Process all queued conversation contexts."""
        # Import here to avoid circular dependency
@@ -110,8 +150,8 @@ class MemoryUpdateQueue:

        with self._lock:
            if self._processing:
-                # Already processing, reschedule
-                self._reset_timer()
+                # Preserve immediate flush semantics even if another worker is active.
+                self._schedule_timer(0)
                return

            if not self._queue:
@@ -164,6 +204,13 @@ class MemoryUpdateQueue:

        self._process_queue()

+    def flush_nowait(self) -> None:
+        """Start queue processing immediately in a background thread."""
+        with self._lock:
+            # Daemon thread: queued messages may be lost if the process exits
+            # before _process_queue completes. Acceptable for best-effort memory updates.
+            self._schedule_timer(0)
+
    def clear(self) -> None:
        """Clear the queue without processing.

@@ -0,0 +1,31 @@
+"""Hooks fired before summarization removes messages from state."""
+
+from __future__ import annotations
+
+from deerflow.agents.memory.message_processing import detect_correction, detect_reinforcement, filter_messages_for_memory
+from deerflow.agents.memory.queue import get_memory_queue
+from deerflow.agents.middlewares.summarization_middleware import SummarizationEvent
+from deerflow.config.memory_config import get_memory_config
+
+
+def memory_flush_hook(event: SummarizationEvent) -> None:
+    """Flush messages about to be summarized into the memory queue."""
+    if not get_memory_config().enabled or not event.thread_id:
+        return
+
+    filtered_messages = filter_messages_for_memory(list(event.messages_to_summarize))
+    user_messages = [message for message in filtered_messages if getattr(message, "type", None) == "human"]
+    assistant_messages = [message for message in filtered_messages if getattr(message, "type", None) == "ai"]
+    if not user_messages or not assistant_messages:
+        return
+
+    correction_detected = detect_correction(filtered_messages)
+    reinforcement_detected = not correction_detected and detect_reinforcement(filtered_messages)
+    queue = get_memory_queue()
+    queue.add_nowait(
+        thread_id=event.thread_id,
+        messages=filtered_messages,
+        agent_name=event.agent_name,
+        correction_detected=correction_detected,
+        reinforcement_detected=reinforcement_detected,
+    )