mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-31 10:26:04 +00:00
e3e5c73b03
* feat(observability): add trace-id correlation and enhanced logging - add opt-in gateway request trace correlation via X-Trace-Id - enhance logging with configurable trace_id-aware formatting - propagate deerflow_trace_id into runtime context and Langfuse metadata - keep enhanced logging disabled by default to preserve existing behavior * fix: harden trace correlation wiring - Make logging enhancement a restart-required startup snapshot and remove per-request config reads from TraceMiddleware - Restrict trace ids to printable ASCII before writing them to response headers, logs, and Langfuse metadata - Gate implicit DeerFlowClient trace-id creation behind logging.enhance.enabled while preserving explicit caller opt-in - Bind embedded client trace context per stream step to avoid generator ContextVar leaks and cross-context reset errors - Rebind memory update trace ids in Timer/executor worker paths so enhanced logs keep the captured correlation id - Remove unrelated __run_journal context overwrite from the trace-correlation change set * fix(gateway): avoid eager app construction on package import * fix(gateway): avoid config load during app import Keep Gateway app construction import-safe when config.yaml is absent by disabling TraceMiddleware only for that construction-time fallback path. Startup lifespan still performs strict config loading before serving.
308 lines
11 KiB
Python
308 lines
11 KiB
Python
"""Memory update queue with debounce mechanism."""
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from contextlib import nullcontext
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from deerflow.config.memory_config import get_memory_config
|
|
from deerflow.trace_context import request_trace_context
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ConversationContext:
|
|
"""Context for a conversation to be processed for memory update."""
|
|
|
|
thread_id: str
|
|
messages: list[Any]
|
|
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
agent_name: str | None = None
|
|
user_id: str | None = None
|
|
deerflow_trace_id: str | None = None
|
|
correction_detected: bool = False
|
|
reinforcement_detected: bool = False
|
|
|
|
|
|
class MemoryUpdateQueue:
|
|
"""Queue for memory updates with debounce mechanism.
|
|
|
|
This queue collects conversation contexts and processes them after
|
|
a configurable debounce period. Multiple conversations received within
|
|
the debounce window are batched together.
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""Initialize the memory update queue."""
|
|
self._queue: list[ConversationContext] = []
|
|
self._lock = threading.Lock()
|
|
self._timer: threading.Timer | None = None
|
|
self._processing = False
|
|
|
|
@staticmethod
|
|
def _queue_key(
|
|
thread_id: str,
|
|
user_id: str | None,
|
|
agent_name: str | None,
|
|
) -> tuple[str, str | None, str | None]:
|
|
"""Return the debounce identity for a memory update target."""
|
|
return (thread_id, user_id, agent_name)
|
|
|
|
def add(
|
|
self,
|
|
thread_id: str,
|
|
messages: list[Any],
|
|
agent_name: str | None = None,
|
|
user_id: str | None = None,
|
|
deerflow_trace_id: str | None = None,
|
|
correction_detected: bool = False,
|
|
reinforcement_detected: bool = False,
|
|
) -> None:
|
|
"""Add a conversation to the update queue.
|
|
|
|
Args:
|
|
thread_id: The thread ID.
|
|
messages: The conversation messages.
|
|
agent_name: If provided, memory is stored per-agent. If None, uses global memory.
|
|
user_id: The user ID captured at enqueue time. Stored in ConversationContext so it
|
|
survives the threading.Timer boundary (ContextVar does not propagate across
|
|
raw threads).
|
|
deerflow_trace_id: Request trace id captured at enqueue time so the
|
|
later Timer thread can attach it to memory LLM tracing metadata.
|
|
correction_detected: Whether recent turns include an explicit correction signal.
|
|
reinforcement_detected: Whether recent turns include a positive reinforcement signal.
|
|
"""
|
|
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,
|
|
user_id=user_id,
|
|
deerflow_trace_id=deerflow_trace_id,
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
)
|
|
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,
|
|
user_id: str | None = None,
|
|
deerflow_trace_id: 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,
|
|
user_id=user_id,
|
|
deerflow_trace_id=deerflow_trace_id,
|
|
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,
|
|
user_id: str | None,
|
|
deerflow_trace_id: str | None,
|
|
correction_detected: bool,
|
|
reinforcement_detected: bool,
|
|
) -> None:
|
|
queue_key = self._queue_key(thread_id, user_id, agent_name)
|
|
existing_context = next(
|
|
(context for context in self._queue if self._queue_key(context.thread_id, context.user_id, context.agent_name) == queue_key),
|
|
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,
|
|
user_id=user_id,
|
|
deerflow_trace_id=deerflow_trace_id,
|
|
correction_detected=merged_correction_detected,
|
|
reinforcement_detected=merged_reinforcement_detected,
|
|
)
|
|
|
|
self._queue = [context for context in self._queue if self._queue_key(context.thread_id, context.user_id, context.agent_name) != queue_key]
|
|
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()
|
|
|
|
self._timer = threading.Timer(
|
|
delay_seconds,
|
|
self._process_queue,
|
|
)
|
|
self._timer.daemon = True
|
|
self._timer.start()
|
|
|
|
def _process_queue(self) -> None:
|
|
"""Process all queued conversation contexts."""
|
|
# Import here to avoid circular dependency
|
|
from deerflow.agents.memory.updater import MemoryUpdater
|
|
|
|
with self._lock:
|
|
if self._processing:
|
|
# Preserve immediate flush semantics even if another worker is active.
|
|
self._schedule_timer(0)
|
|
return
|
|
|
|
if not self._queue:
|
|
return
|
|
|
|
self._processing = True
|
|
contexts_to_process = self._queue.copy()
|
|
self._queue.clear()
|
|
self._timer = None
|
|
|
|
logger.info("Processing %d queued memory updates", len(contexts_to_process))
|
|
|
|
try:
|
|
updater = MemoryUpdater()
|
|
|
|
for context in contexts_to_process:
|
|
# Rebind the request-trace ContextVar from the value captured at
|
|
# enqueue time so ``TraceContextFilter`` attaches the correct
|
|
# trace id to every log record emitted below (this Timer thread
|
|
# does not inherit the enqueue-thread's ContextVar). Each
|
|
# iteration is scoped independently so id A does not leak into
|
|
# id B's logs.
|
|
trace_ctx = request_trace_context(context.deerflow_trace_id) if context.deerflow_trace_id else nullcontext()
|
|
with trace_ctx:
|
|
try:
|
|
logger.info("Updating memory for thread %s", context.thread_id)
|
|
success = updater.update_memory(
|
|
messages=context.messages,
|
|
thread_id=context.thread_id,
|
|
agent_name=context.agent_name,
|
|
correction_detected=context.correction_detected,
|
|
reinforcement_detected=context.reinforcement_detected,
|
|
user_id=context.user_id,
|
|
deerflow_trace_id=context.deerflow_trace_id,
|
|
)
|
|
if success:
|
|
logger.info("Memory updated successfully for thread %s", context.thread_id)
|
|
else:
|
|
logger.warning("Memory update skipped/failed for thread %s", context.thread_id)
|
|
except Exception as e:
|
|
logger.error("Error updating memory for thread %s: %s", context.thread_id, e)
|
|
|
|
# Small delay between updates to avoid rate limiting
|
|
if len(contexts_to_process) > 1:
|
|
time.sleep(0.5)
|
|
|
|
finally:
|
|
with self._lock:
|
|
self._processing = False
|
|
|
|
def flush(self) -> None:
|
|
"""Force immediate processing of the queue.
|
|
|
|
This is useful for testing or graceful shutdown.
|
|
"""
|
|
with self._lock:
|
|
if self._timer is not None:
|
|
self._timer.cancel()
|
|
self._timer = None
|
|
|
|
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.
|
|
|
|
This is useful for testing.
|
|
"""
|
|
with self._lock:
|
|
if self._timer is not None:
|
|
self._timer.cancel()
|
|
self._timer = None
|
|
self._queue.clear()
|
|
self._processing = False
|
|
|
|
@property
|
|
def pending_count(self) -> int:
|
|
"""Get the number of pending updates."""
|
|
with self._lock:
|
|
return len(self._queue)
|
|
|
|
@property
|
|
def is_processing(self) -> bool:
|
|
"""Check if the queue is currently being processed."""
|
|
with self._lock:
|
|
return self._processing
|
|
|
|
|
|
# Global singleton instance
|
|
_memory_queue: MemoryUpdateQueue | None = None
|
|
_queue_lock = threading.Lock()
|
|
|
|
|
|
def get_memory_queue() -> MemoryUpdateQueue:
|
|
"""Get the global memory update queue singleton.
|
|
|
|
Returns:
|
|
The memory update queue instance.
|
|
"""
|
|
global _memory_queue
|
|
with _queue_lock:
|
|
if _memory_queue is None:
|
|
_memory_queue = MemoryUpdateQueue()
|
|
return _memory_queue
|
|
|
|
|
|
def reset_memory_queue() -> None:
|
|
"""Reset the global memory queue.
|
|
|
|
This is useful for testing.
|
|
"""
|
|
global _memory_queue
|
|
with _queue_lock:
|
|
if _memory_queue is not None:
|
|
_memory_queue.clear()
|
|
_memory_queue = None
|