mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 08:56:13 +00:00
8145d66a33
* feat(memory): signals-based update pipeline + always-on watermark/trivial filter
Refactor the DeerMem memory update pipeline (message_processing -> queue ->
updater) around a signals frozenset seam, replacing the
(filtered, correction_detected, reinforcement_detected) 3-tuple with
(filtered, signals: frozenset[str]) end to end.
message_processing:
- Externalize signal-detection patterns to YAML (message_patterns/*.yaml).
- Extend signals from correction/reinforcement to a 6-class set
(correction/reinforcement/preference/identity/goal/decision); detect_signals
returns a frozenset aligned with the fact category enum.
- Pure-acknowledgment turns ("ok"/"好的"/...) are always filtered out before
enqueue (whole-message fullmatch), saving an extraction LLM call.
queue (core/queue.py):
- In-memory list + debounce timer, with flush_sync (graceful-shutdown drain
that joins an in-flight worker under a hard timeout) and queue_max_depth
backpressure (signal-bearing updates always admitted; QueueFull otherwise).
- Same-key updates coalesce with a signal union; per-batch success/fail summary.
updater (core/updater.py):
- head500+tail500 message truncation (replaces the 1000-char head chop).
- Always-on per-thread watermark: feed only messages added since the last
extraction. The watermark is in-memory and is not advanced on failure, so a
failed/lost update is re-fed on the next conversation turn.
- [MANUAL] prompt marker for user-authored facts (source.type="manual").
- Post-invoke extraction_callback (host-injected) emitting facts_extracted /
facts_accepted / rejected_low_confidence; the host default logs metrics and
flags >60% rejection.
Confidence filtering remains in _apply_updates (the existing
fact_confidence_threshold check); there is no separate write gate.
Consolidation stays opt-in (lossy). The ABC add/add_nowait signature is
unchanged, so the summarization flush hook and host are unaffected.
Tests: add test_message_processing_signals, test_updater_truncation,
test_updater_watermark; update queue/updater/consolidation/staleness/pluggable
tests for the signals seam.
Co-Authored-By: Claude <noreply@anthropic.com>
* fix(memory): harden update pipeline per PR review
- Catch QueueFull in DeerMem.add/add_nowait so backpressure degrades to
'update skipped' instead of propagating into after_agent /
summarization_hook and breaking the agent run (peer middlewares
self-guard; MemoryMiddleware was the lone exception). Emergency
(add_nowait) always admits under backpressure -- its data cannot be
re-fed next turn.
- Rewrite the watermark from index-based to content/identity-based
(_message_identity + _feed_after_watermark) so it stays correct when
summarization removes the conversation front -- an index watermark
pointed at the wrong message and silently skipped un-extracted tail
turns. The emergency flush bypasses the watermark (bypass_watermark on
ConversationContext, threaded through update_memory) and coexists with
(does not replace) a pending normal update, so a flush cannot drop a
pending update's un-extracted tail.
- Populate facts_accepted / rejected_low_confidence inside _apply_updates
at the real confidence-filter site (passed_threshold) instead of
re-deriving the threshold in _finalize_update -- eliminates metric drift.
- Emit extraction metrics in a finally with an 'attempted' flag so
exception failures (parse error, apply_changes raise after retry) are
observable, not only the happy path.
- Re-detect signals on the post-watermark feed for the extraction hint so
it no longer references turns the LLM cannot see; admission-time signals
still drive backpressure.
- Move the post-batch reschedule inside the queue lock to close a
non-atomic self._timer race with a concurrent add.
Co-Authored-By: Claude <noreply@anthropic.com>
* fix(memory): address follow-up review nits (LRU, metric name, docstring)
- Bound the in-memory watermark cache with a configurable LRU
(watermark_max_keys, default 4096, 0=unbounded). A dropped key re-extracts
one batch on that thread's next turn (the documented restart behavior), so
eviction is safe and preserves the content-identity watermark's
front-removal guarantee. Adds _watermark_get/_watermark_set helpers and a
bounded-LRU regression test.
- Rename the extraction metric facts_accepted -> facts_passed_confidence so
the name matches what the >60% rejection-rate warning assumes (a
confidence-gate signal, not a persisted-fact count); drop the stale
"historical semantics" justification. Brand-new callback, one consumer.
- Fix the stale test_message_processing_signals module docstring: the signals
seam is already swapped to frozenset, and a stale stage-numbering prefix is
removed.
Co-Authored-By: Claude <noreply@anthropic.com>
---------
Co-authored-by: Claude <noreply@anthropic.com>
140 lines
4.9 KiB
Python
140 lines
4.9 KiB
Python
"""Tests for message-processing signal detection and trivial filtering.
|
|
|
|
Pins three behaviors: ``detect_signals`` recognizes all six signal classes
|
|
(correction, reinforcement, preference, identity, goal, decision);
|
|
``filter_trivial`` drops pure-ack turns and their replies while keeping
|
|
substantive turns; and ``_prepare_update`` returns the full signal set as a
|
|
``frozenset`` (not just correction/reinforcement), so every detected class
|
|
flows through to the extraction prompt.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem
|
|
from deerflow.agents.memory.backends.deermem.deermem.core.message_processing import (
|
|
detect_signals,
|
|
extract_message_text,
|
|
filter_trivial,
|
|
)
|
|
|
|
|
|
def _human(text: str) -> HumanMessage:
|
|
return HumanMessage(content=text)
|
|
|
|
|
|
def _ai(text: str) -> AIMessage:
|
|
return AIMessage(content=text)
|
|
|
|
|
|
# ── detect_signals: the 6 signal classes ───────────────────────────────────
|
|
|
|
|
|
def test_detect_signals_correction() -> None:
|
|
assert "correction" in detect_signals([_human("That's wrong, use uv"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_reinforcement() -> None:
|
|
assert "reinforcement" in detect_signals([_human("perfect, exactly right"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_preference() -> None:
|
|
assert "preference" in detect_signals([_human("I prefer uv over pip"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_identity() -> None:
|
|
assert "identity" in detect_signals([_human("I am an engineer"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_goal() -> None:
|
|
assert "goal" in detect_signals([_human("I plan to migrate to uv"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_decision() -> None:
|
|
assert "decision" in detect_signals([_human("let's go with uv"), _ai("ok")])
|
|
|
|
|
|
def test_detect_signals_none_for_substantive_turn() -> None:
|
|
assert detect_signals([_human("what is the weather"), _ai("sunny")]) == set()
|
|
|
|
|
|
def test_detect_signals_multiple_classes_in_one_turn() -> None:
|
|
# A turn that states both a preference and an identity surfaces both.
|
|
signals = detect_signals([_human("I am an engineer and I prefer uv"), _ai("ok")])
|
|
assert "identity" in signals
|
|
assert "preference" in signals
|
|
|
|
|
|
# ── filter_trivial ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_filter_trivial_drops_pure_ack_and_its_reply() -> None:
|
|
msgs = [_human("嗯"), _ai("thanks"), _human("what next"), _ai("let's see")]
|
|
result = filter_trivial(msgs)
|
|
# "嗯" + its AI "thanks" dropped; the substantive pair is kept.
|
|
assert len(result) == 2
|
|
assert extract_message_text(result[0]) == "what next"
|
|
|
|
|
|
def test_filter_trivial_keeps_substantive_message_containing_ok() -> None:
|
|
msgs = [_human("use uv to install, ok?"), _ai("done")]
|
|
result = filter_trivial(msgs)
|
|
assert len(result) == 2 # not dropped: not a whole-message ack
|
|
|
|
|
|
def test_filter_trivial_all_trivial_returns_empty() -> None:
|
|
msgs = [_human("好的"), _ai("嗯")]
|
|
assert filter_trivial(msgs) == []
|
|
|
|
|
|
def test_filter_trivial_tolerates_trailing_punctuation() -> None:
|
|
msgs = [_human("ok."), _ai("ok!")]
|
|
assert filter_trivial(msgs) == []
|
|
|
|
|
|
def test_filter_trivial_no_patterns_keeps_all() -> None:
|
|
msgs = [_human("ok"), _ai("ok")]
|
|
assert filter_trivial(msgs, patterns=[]) == msgs
|
|
|
|
|
|
# ── _prepare_update: seam-stable 3-tuple projection ────────────────────────
|
|
|
|
|
|
def _make_deermem(tmp_path, **overrides) -> DeerMem:
|
|
cfg = {"storage_path": str(tmp_path)}
|
|
cfg.update(overrides)
|
|
return DeerMem(backend_config=cfg)
|
|
|
|
|
|
def test_prepare_update_all_trivial_returns_none(tmp_path) -> None:
|
|
m = _make_deermem(tmp_path)
|
|
assert m._prepare_update([_human("好的"), _ai("嗯")]) is None
|
|
|
|
|
|
def test_prepare_update_returns_correction_signal(tmp_path) -> None:
|
|
m = _make_deermem(tmp_path)
|
|
r = m._prepare_update([_human("That's wrong, use uv"), _ai("ok")])
|
|
assert r is not None and len(r) == 2
|
|
_filtered, signals = r
|
|
assert "correction" in signals
|
|
|
|
|
|
def test_prepare_update_returns_reinforcement_signal(tmp_path) -> None:
|
|
m = _make_deermem(tmp_path)
|
|
r = m._prepare_update([_human("perfect, exactly right"), _ai("ok")])
|
|
assert r is not None and len(r) == 2
|
|
_filtered, signals = r
|
|
assert "reinforcement" in signals
|
|
|
|
|
|
def test_prepare_update_returns_new_signals_after_swap(tmp_path) -> None:
|
|
# After the signals-seam swap, the full signal set flows through (not just
|
|
# correction/reinforcement): a preference turn surfaces "preference".
|
|
m = _make_deermem(tmp_path)
|
|
r = m._prepare_update([_human("I prefer uv over pip"), _ai("ok")])
|
|
assert r is not None and len(r) == 2
|
|
_filtered, signals = r
|
|
assert "preference" in signals
|
|
assert len(_filtered) == 2 # not trivial -> kept
|