diff --git a/README.md b/README.md index b085590ce6..d2cc32fc60 100644 --- a/README.md +++ b/README.md @@ -2221,6 +2221,10 @@ behavior, and current boundaries. Across sessions, DeerFlow builds a persistent memory of your profile, preferences, and accumulated knowledge. The more you use it, the better it knows you — your writing style, your technical stack, your recurring workflows. Memory is stored locally and stays under your control. +DeerMem and mem0 capture user inputs and final assistant replies. Assistant +turns that request tools, including malformed or provider-raw calls, are +excluded from memory extraction even when they contain visible text. + DeerMem remains the default local backend. An opt-in `mem0` backend is also available for the hosted mem0 Platform API or API-compatible self-hosted servers. Its token-bearing `base_url` must use HTTPS by default; plaintext HTTP diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index 549216e635..23af2255af 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -87,6 +87,12 @@ fact-only import payload supplies empty summary defaults. It captures `user_id` when it enqueues work. This identity survives the background timer boundary. +DeerMem and mem0 exclude assistant tool-call intent in every representation: +`tool_calls`, `invalid_tool_calls`, and provider-raw `tool_calls`/`function_call`. +An empty parsed call list does not make an attempted call a final response. +Keep both portable backend filters aligned without importing host helpers; +test the real queue/HTTP write boundary in `tests/test_memory_tool_call_intent.py`. + `memory.mode: tool` registers the four memory tools. The model chooses when to search or change facts. Tool mode still uses `MemoryMiddleware` for passive writes on supported remote backends. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/message_processing.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/message_processing.py index ce296b7c78..e482cabe1f 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/message_processing.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/message_processing.py @@ -199,12 +199,14 @@ def filter_messages_for_memory(messages: list[Any], *, should_keep_hidden_messag 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) + # Invalid/provider-raw calls are still tool turns, not final replies. + additional_kwargs = getattr(msg, "additional_kwargs", {}) or {} + if getattr(msg, "tool_calls", None) or getattr(msg, "invalid_tool_calls", None) or additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"): + continue + if skip_next_ai: + skip_next_ai = False + continue + filtered.append(msg) return filtered diff --git a/backend/packages/harness/deerflow/agents/memory/backends/mem0/message_filtering.py b/backend/packages/harness/deerflow/agents/memory/backends/mem0/message_filtering.py index 94f3ce634a..18aa638482 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/mem0/message_filtering.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/mem0/message_filtering.py @@ -84,7 +84,9 @@ def filter_messages_for_memory(messages: list[Any]) -> list[Any]: filtered.append(msg) skip_next_ai = False elif msg_type == "ai": - if getattr(msg, "tool_calls", None): + # Invalid/provider-raw calls are still tool turns, not final replies. + additional_kwargs = getattr(msg, "additional_kwargs", {}) or {} + if getattr(msg, "tool_calls", None) or getattr(msg, "invalid_tool_calls", None) or additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"): continue if skip_next_ai: skip_next_ai = False diff --git a/backend/tests/test_memory_tool_call_intent.py b/backend/tests/test_memory_tool_call_intent.py new file mode 100644 index 0000000000..1824ad7453 --- /dev/null +++ b/backend/tests/test_memory_tool_call_intent.py @@ -0,0 +1,129 @@ +"""Tool-call assistant turns must not become final-response memory inputs.""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import httpx +import pytest +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, messages_from_dict, messages_to_dict +from langchain_openai.chat_models.base import _convert_dict_to_message + +from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem +from deerflow.agents.memory.backends.deermem.deermem.core.message_processing import filter_messages_for_memory as filter_deermem +from deerflow.agents.memory.backends.mem0.client import Mem0Client +from deerflow.agents.memory.backends.mem0.mem0_manager import Mem0Manager +from deerflow.agents.memory.backends.mem0.message_filtering import filter_messages_for_memory as filter_mem0 + +_ATTEMPT_TEXT = "I will save the nightly production deployment preference." +_USER_TEXT = "Inspect the deployment settings before changing anything." +_FINAL_TEXT = "The settings were inspected; no deployment preference was saved." + + +def _assistant_attempt(kind: str) -> AIMessage: + if kind == "parsed": + return AIMessage(content=_ATTEMPT_TEXT, tool_calls=[{"name": "memory_add", "args": {}, "id": "call-1"}]) + if kind == "invalid": + return AIMessage(content=_ATTEMPT_TEXT, invalid_tool_calls=[{"name": "memory_add", "args": "{broken", "id": "call-1"}]) + if kind == "legacy": + return _convert_dict_to_message({"role": "assistant", "content": _ATTEMPT_TEXT, "function_call": {"name": "memory_add", "arguments": "{}"}}) + raw = [{"id": "call-1", "type": "function", "function": {"name": "memory_add", "arguments": "{broken"}}] + if kind == "raw": + # Provider-raw fallback with neither structured view (e.g. an older checkpoint). + return AIMessage(content=_ATTEMPT_TEXT).model_copy(update={"additional_kwargs": {"tool_calls": raw}}) + message = _convert_dict_to_message({"role": "assistant", "content": _ATTEMPT_TEXT, "tool_calls": raw}) + assert isinstance(message, AIMessage) + assert not message.tool_calls and message.invalid_tool_calls + # Exercise the message serialization used when persisting/reloading history. + return messages_from_dict(messages_to_dict([message]))[0] + + +@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"]) +@pytest.mark.parametrize("kind", ["parsed", "invalid", "provider-invalid", "raw", "legacy"]) +def test_filter_excludes_all_tool_call_views_without_mutating_history(filter_messages, kind: str) -> None: + attempt = _assistant_attempt(kind) + messages = [HumanMessage(content=_USER_TEXT), attempt, ToolMessage(content="Not executed", tool_call_id="call-1", status="error"), AIMessage(content=_FINAL_TEXT)] + original = messages_to_dict(messages) + + kept = filter_messages(messages) + + assert kept == [messages[0], messages[-1]] + assert kept[0] is messages[0] and kept[1] is messages[-1] + assert messages_to_dict(messages) == original + + +@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"]) +@pytest.mark.parametrize("kwargs", [{}, {"tool_calls": []}, {"function_call": None}, {"tool_calls": [], "function_call": {}}]) +def test_empty_call_views_keep_final_response(filter_messages, kwargs: dict) -> None: + final = AIMessage(content=[{"type": "text", "text": _FINAL_TEXT}], additional_kwargs=kwargs) + assert filter_messages([final]) == [final] + + +@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"]) +def test_invalid_attempt_does_not_consume_upload_ack_filter(filter_messages) -> None: + upload = HumanMessage(content="\nsettings.yaml\n") + attempt = _assistant_attempt("provider-invalid") + ack = AIMessage(content="I see the uploaded settings.") + user = HumanMessage(content=_USER_TEXT) + final = AIMessage(content=_FINAL_TEXT) + assert filter_messages([upload, attempt, ack, user, final]) == [user, final] + + +@pytest.mark.parametrize("method", ["add", "add_nowait"]) +@pytest.mark.parametrize("include_final", [False, True]) +def test_deermem_queue_contains_only_user_and_final_response(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, method: str, include_final: bool) -> None: + manager = DeerMem(backend_config={"storage_path": str(tmp_path)}) + # Keep the real queue deterministic; do not launch background extraction. + monkeypatch.setattr(manager._queue, "_schedule_timer", lambda *args, **kwargs: None) + messages = [HumanMessage(content=_USER_TEXT), _assistant_attempt("provider-invalid")] + if include_final: + messages.append(AIMessage(content=_FINAL_TEXT)) + original = messages_to_dict(messages) + try: + getattr(manager, method)("thread-1", messages, user_id="alice", agent_name="deployment-agent") + contexts = manager._queue._items + if include_final: + assert len(contexts) == 1 + context = contexts[0] + assert context.messages == [messages[0], messages[-1]] + assert (context.thread_id, context.user_id, context.agent_name) == ("thread-1", "alice", "deployment-agent") + else: + # An attempted call is not the assistant reply required for extraction. + assert contexts == [] + assert messages_to_dict(messages) == original + finally: + manager.cancel_by_agent("deployment-agent", user_id="alice") + manager.close() + + +@pytest.mark.parametrize("method", ["add", "aadd"]) +@pytest.mark.parametrize("include_final", [False, True]) +def test_mem0_http_payload_excludes_invalid_attempt(monkeypatch: pytest.MonkeyPatch, method: str, include_final: bool) -> None: + monkeypatch.setenv("MEM0_API_KEY", "offline-test-key") + payloads: list[dict] = [] + + def record(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" and request.url.path == "/v3/memories/add/" + payloads.append(json.loads(request.content)) + return httpx.Response(200, json={"status": "PENDING"}) + + manager = Mem0Manager(backend_config={"startup_policy": "tolerate"}) + manager.close() + manager._client = Mem0Client(base_url="https://mem0.test", api_key="offline-test-key", transport=httpx.MockTransport(record)) + messages = [HumanMessage(content=_USER_TEXT), _assistant_attempt("provider-invalid")] + if include_final: + messages.append(AIMessage(content=_FINAL_TEXT)) + original = messages_to_dict(messages) + try: + result = getattr(manager, method)("thread-1", messages, user_id="alice", agent_name="deployment-agent") + if method == "aadd": + asyncio.run(result) + expected = [{"role": "user", "content": _USER_TEXT}] + if include_final: + expected.append({"role": "assistant", "content": _FINAL_TEXT}) + assert payloads == [{"messages": expected, "user_id": "alice", "agent_id": "deployment-agent", "run_id": "thread-1"}] + assert messages_to_dict(messages) == original + finally: + manager.close()