wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(memory): Memory V2 P3 - 舍弃后台抽取 + Gateway 端点收口

Browse Source 
P3 在 P1/P2 基础上完成 Memory V2 收尾:

舍弃后台抽取:
- 删除 updater/queue/storage/prompt/summarization_hook 五个模块
- 移除 background_extraction 配置段、扁平字段兼容、storage 字段
- 理由:Hindsight auto_retain 已自动留存每轮对话,后台抽取冗余

Hindsight 召回注入改用 wrap_model_call:
- 由 before_model 改为 wrap_model_call —— 召回只注入单次模型请求、
  不落持久 state,从根上消除 <memory-context> 泄漏给前端的风险
- builtin frozen snapshot 仍走系统提示词

Gateway 端点收口为 /api/memory/v2*:
- 删除旧 facts/history 端点 (/api/memory、/memory/facts 等)
- 新增 entries 增删改 (POST/PUT/DELETE /memory/v2/entries)
- 新增 POST /memory/v2/search —— Hindsight recall/reflect 检索代理
- client.py memory 方法重写为 V2 等价方法

文档同步更新 CLAUDE.md 与 MEMORY_V2_DESIGN_ZH.md。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
xuhdandClaude Opus 4.7 committed 2026-05-15 13:08:18 +08:00
1 parent f9d7610679
commit 21a76f5e7c
14 files changed
+954 -3085
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

offline-backend-20260512/backend/CLAUDE.md
+36 -33
View File
@@ -31,7 +31,7 @@ deer-flow/
│ │ ├── agents/ # LangGraph agent system
│ │ │ ├── lead_agent/ # Main agent (factory + system prompt)
│ │ │ ├── middlewares/ # 10 middleware components
│ │ │ ├── memory/ # Memory extraction, queue, prompts
│ │ │ ├── memory/ # Memory V2: provider/manager/builtin/hindsight
│ │ │ └── thread_state.py # ThreadState schema
│ │ ├── sandbox/ # Sandbox execution system
│ │ │ ├── local/ # Local filesystem provider
@@ -167,7 +167,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
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** - Auto-generates thread title after first complete exchange and normalizes structured message content before prompting the title model
13. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses)
13. **MemoryMiddleware** - Memory V2: `wrap_model_call` injects Hindsight recall transiently; `after_agent` auto-retains the turn to Hindsight
14. **ViewImageMiddleware** - Injects base64 image data before LLM call (conditional on vision support)
15. **DeferredToolFilterMiddleware** - Hides deferred tool schemas from the bound model until tool search is enabled (optional)
16. **SubagentLimitMiddleware** - Truncates excess `task` tool calls from model response to enforce `MAX_CONCURRENT_SUBAGENTS` limit (optional, if `subagent_enabled`)
@@ -214,7 +214,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details |
| **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) |
| **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
| **Memory V2** (`/api/memory/v2`) | `GET /` - builtin data (USER.md/MEMORY.md); `POST/PUT/DELETE /entries` - add/replace/remove entry; `GET/PATCH/DELETE /config` - effective config + per-user overrides; `POST /search` - Hindsight recall/reflect proxy |
| **Agents** (`/api/agents`) | `GET /` - list built-in, owned, and published agents; `POST /` - create user-owned agent; `GET/PUT/DELETE /{id}` - read/update/delete by stable agent id; `GET /check` - validate non-empty display name (duplicates allowed) |
| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); `GET /list` - list; `DELETE /{filename}` - delete |
| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
@@ -348,43 +348,46 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, DingTalk) to the
- Per-channel configs: `feishu` (app_id, app_secret), `slack` (bot_token, app_token), `telegram` (bot_token), `dingtalk` (client_id, client_secret, optional `card_template_id` for AI Card streaming)
### Memory System (`packages/harness/deerflow/agents/memory/`)
### Memory System — Memory V2 (`packages/harness/deerflow/agents/memory/`)
Pluggable memory architecture (Hermes-style). See `docs/MEMORY_V2_DESIGN_ZH.md`
and `docs/HINDSIGHT_SETUP_ZH.md` for the full design.
**Components**:
- `updater.py` - LLM-based memory updates with fact extraction, whitespace-normalized fact deduplication (trims leading/trailing whitespace before comparing), and atomic file I/O
- `queue.py` - Debounced update queue (per-thread deduplication, configurable wait time); captures `user_id` at enqueue time so it survives the `threading.Timer` boundary
- `prompt.py` - Prompt templates for memory updates
- `storage.py` - File-based storage with per-user isolation; cache keyed by `(user_id, agent_name)` tuple
- `provider.py` - `MemoryProvider` abstract base (12 lifecycle/hook methods)
- `manager.py` - `MemoryManager` orchestrates builtin + ≤1 external provider; `build_memory_manager()` is the construction entry point
- `providers/builtin.py` - `BuiltinFileProvider`: local USER.md / MEMORY.md, § delimiter, char quotas, file locking, atomic writes, frozen snapshot
- `providers/hindsight.py` - `HindsightProvider`: HTTP client to a self-hosted Hindsight Docker instance, dual-bank routing
- `security.py` - injection/exfiltration scan + invisible-unicode block for memory content
- `message_processing.py` - `filter_messages_for_memory` (last-turn extraction for Hindsight retain)
**Per-User Isolation**:
- Memory is stored per-user at `{base_dir}/users/{user_id}/memory.json`
- Per-agent per-user memory at `{base_dir}/users/{user_id}/agents/{agent_name}/memory.json`
- `user_id` is resolved via `get_effective_user_id()` from `deerflow.runtime.user_context`
- In no-auth mode, `user_id` defaults to `"default"` (constant `DEFAULT_USER_ID`)
- Absolute `storage_path` in config opts out of per-user isolation
- **Migration**: Run `PYTHONPATH=. python scripts/migrate_user_isolation.py` to move legacy `memory.json` and `threads/` into per-user layout; supports `--dry-run`
**Two stores (binary classification)**:
- `USER.md` at `{base_dir}/users/{user_id}/USER.md` — user profile, shared across agents
- `MEMORY.md` at `{base_dir}/users/{user_id}/agents/{agent_id}/MEMORY.md` — agent-private working notes
- `user_id` resolved via `get_effective_user_id()`; `"default"` in no-auth mode
**Data Structure** (stored in `{base_dir}/users/{user_id}/memory.json`):
- **User Context**: `workContext`, `personalContext`, `topOfMind` (1-3 sentence summaries)
- **History**: `recentMonths`, `earlierContext`, `longTermBackground`
- **Facts**: Discrete facts with `id`, `content`, `category` (preference/knowledge/context/behavior/goal), `confidence` (0-1), `createdAt`, `source`
**Write paths**:
- LLM actively calls the `memory` tool (`tools/builtins/memory_tool.py`) → writes USER.md/MEMORY.md, mirrors to Hindsight
- LLM calls `hindsight_recall`/`reflect`/`retain` tools (`tools/builtins/hindsight_tools.py`) — only in `tools`/`hybrid` memory_mode
- Hindsight `auto_retain` — every turn auto-stored to the work bank (via `MemoryMiddleware.after_agent`)
**Workflow**:
1. `MemoryMiddleware` filters messages (user inputs + final AI responses), captures `user_id` via `get_effective_user_id()`, and queues conversation with the captured `user_id`
2. Queue debounces (30s default), batches updates, deduplicates per-thread
3. Background thread invokes LLM to extract context updates and facts, using the stored `user_id` (not the contextvar, which is unavailable on timer threads)
4. Applies updates atomically (temp file + rename) with cache invalidation, skipping duplicate fact content before append
5. Next interaction injects top 15 facts + context into `<memory>` tags in system prompt
**Injection**:
- builtin frozen snapshot → injected into the system prompt via `lead_agent/prompt.py` `_get_memory_context`
- Hindsight dynamic recall → injected per model call via `MemoryMiddleware.wrap_model_call` (transient — not persisted to state, so it never leaks to the frontend)
Focused regression coverage for the updater lives in `backend/tests/test_memory_updater.py`.
**Hindsight dual-bank isolation** (B-scheme):
- global user bank `deerflow-user-{user_id}` — cross-agent
- agent work bank `deerflow-work-{user_id}-{agent_id}` — per-agent
- `hindsight-client` is an optional dependency; when missing, `HindsightProvider.is_available()` returns False and the system falls back to builtin-only
**Configuration** (`config.yaml` → `memory`):
- `enabled` / `injection_enabled` - Master switches
- `storage_path` - Path to memory.json (absolute path opts out of per-user isolation)
- `debounce_seconds` - Wait time before processing (default: 30)
- `model_name` - LLM for updates (null = default model)
- `max_facts` / `fact_confidence_threshold` - Fact storage limits (100 / 0.7)
- `max_injection_tokens` - Token limit for prompt injection (2000)
**Migration**: `PYTHONPATH=. python scripts/migrate_memory_to_v2.py [--all|--user-id X] [--dry-run]` converts legacy `memory.json` into USER.md/MEMORY.md.
**Configuration** (`config.yaml` → `memory`, nested): `enabled`, `provider` (builtin/hindsight),
`builtin.{memory_char_limit,user_char_limit,...}`, `hindsight.{mode,api_url,api_key,bank_id_template,work_bank_id_template,memory_mode,...}`,
`injection.{enabled,max_tokens,include_builtin,include_hindsight,context_tag}`, `security.{scan_content,block_invisible_unicode}`.
Per-user runtime overrides (whitelisted fields) live at `{base_dir}/users/{user_id}/memory_config.json` — see `get_effective_memory_config()`.
**Gateway API** — see the Memory router below (`/api/memory/v2*`).
### Reflection System (`packages/harness/deerflow/reflection/`)
offline-backend-20260512/backend/app/gateway/routers/memory.py
+385 -675
View File
File diff suppressed because it is too large. Load diff
offline-backend-20260512/backend/docs/MEMORY_V2_DESIGN_ZH.md
+28 -2
View File
@@ -1,18 +1,44 @@
# Memory V2 设计文档(Hermes 兼容 + Hindsight 集成)
> **状态**: Draft v1 · 待 review
> **状态**: 已实施 (P1/P2/P3 完成)
> **作者**: xuhd
> **日期**: 2026-05-15
> **目标版本**: DeerFlow 内部 1.x
本设计文档描述以 Hermes Agent 的记忆架构为骨架,重写 DeerFlow 的 memory 子系统的方案。重写后:
- **保留** DeerFlow 当前的"后台 LLM 抽取 facts"能力(可选,默认关闭)。
- **引入** Hermes 的 `MemoryProvider` 可插拔抽象 + `MemoryManager` 编排层 + LLM 主动调 `memory` 工具的能力。
- **新增** Hindsight Docker 实例作为唯一支持的外部 provider(语义检索 + 知识图谱)。
- **统一** USER / MEMORY 二元分类,并按 user / agent 维度做双 bank 隔离。
---
## 0. 实施变更记录(与下文原计划的偏差)
下文 §1~§18 是规划阶段的原始设计。实施过程中有以下偏差,**以本节为准**:
1. **后台抽取已舍弃**(原计划保留为可选)。`updater.py` / `queue.py` /
`storage.py` / `prompt.py` / `summarization_hook.py` 全部删除,`background_extraction`
配置段移除。理由:Hindsight 的 `auto_retain` 已把每轮对话自动存入 work bank,
"不漏记"场景完全覆盖,后台抽取冗余且白费 token。下文 §6 的"后台抽取"路径、
§9 的 `background_extraction` 配置、§16 P3 的后台抽取重接入均作废。
2. **Hindsight 动态召回走 `wrap_model_call` 而非 `before_model`**(原计划见 Appendix B.1)。
`wrap_model_call` 把召回内容只注入到单次模型请求、不落持久 state —— 因此不会泄漏
给前端、也不会在历史里堆积。builtin 的 frozen snapshot 仍走系统提示词
(`_get_memory_context`)。
3. **streaming scrubber 未实现**(原计划见 §12.3)。`wrap_model_call` 的瞬态注入已从
结构上消除 `<memory-context>` 泄漏,scrubber 无必要。
4. **Gateway 端点最终形态**:全部用 `/api/memory/v2*` 前缀,旧 facts/history 端点
(`/api/memory`、`/memory/facts` 等) 已删除。端点见 §11 修订 + 新增
`POST /api/memory/v2/search`(Hindsight recall/reflect 代理)。
5. **`client.py` memory 方法**重写为 V2 等价方法(`get_memory` / `add_memory_entry` /
`replace_memory_entry` / `remove_memory_entry` / `get_memory_config` / `search_memory`)。
---
## 1. 设计目标
### 1.1 业务目标
offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py
+8 -10
View File
@@ -5,12 +5,11 @@ from langchain.agents.middleware import AgentMiddleware
from langchain_core.runnables import RunnableConfig
from deerflow.agents.lead_agent.prompt import apply_prompt_template
from deerflow.agents.memory.summarization_hook import memory_flush_hook
from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware
from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
from deerflow.agents.middlewares.setup_agent_approval_middleware import SetupAgentApprovalMiddleware
from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
from deerflow.agents.middlewares.setup_agent_approval_middleware import SetupAgentApprovalMiddleware
from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
from deerflow.agents.middlewares.summarization_middleware import BeforeSummarizationHook, DeerFlowSummarizationMiddleware
from deerflow.agents.middlewares.title_middleware import TitleMiddleware
from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
@@ -91,9 +90,8 @@ def _create_summarization_middleware(*, app_config: AppConfig | None = None) ->
if config.summary_prompt is not None:
kwargs["summary_prompt"] = config.summary_prompt
# Memory V2:不再有后台抽取队列,无需 summarization 前的 flush hook。
hooks: list[BeforeSummarizationHook] = []
if resolved_app_config.memory.enabled:
hooks.append(memory_flush_hook)
# The logic below relies on two assumptions holding true: this factory is
# the sole entry point for DeerFlowSummarizationMiddleware, and the runtime
@@ -304,10 +302,10 @@ def _build_middlewares(
if custom_middlewares:
middlewares.extend(custom_middlewares)
middlewares.append(SetupAgentApprovalMiddleware())
# ClarificationMiddleware should always be last
middlewares.append(ClarificationMiddleware())
middlewares.append(SetupAgentApprovalMiddleware())
# ClarificationMiddleware should always be last
middlewares.append(ClarificationMiddleware())
return middlewares
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/__init__.py
+37 -80
View File
@@ -1,80 +1,37 @@
"""DeerFlow 记忆模块 (Memory V2)。
V2 架构 (详见 docs/MEMORY_V2_DESIGN_ZH.md):
- :class:`MemoryProvider` 可插拔记忆 provider 抽象基类
- :class:`MemoryManager` 编排内建 + 至多 1 个外部 provider
- :class:`BuiltinFileProvider` 本地 USER.md / MEMORY.md provider
- :mod:`deerflow.agents.memory.security` 记忆内容安全扫描
旧版组件 (queue / storage / updater) 仍保留 —— 它们服务于后台抽取
(默认关闭),将在 P3 期重写。
"""
# --- Memory V2 ---
from deerflow.agents.memory.manager import MemoryManager, build_memory_manager
# --- 旧版组件 (后台抽取,P3 期重写) ---
from deerflow.agents.memory.prompt import (
FACT_EXTRACTION_PROMPT,
MEMORY_UPDATE_PROMPT,
format_conversation_for_update,
format_memory_for_injection,
)
from deerflow.agents.memory.provider import MemoryProvider
from deerflow.agents.memory.providers.builtin import BuiltinFileProvider
from deerflow.agents.memory.queue import (
ConversationContext,
MemoryUpdateQueue,
get_memory_queue,
reset_memory_queue,
)
from deerflow.agents.memory.security import (
MemoryContentBlocked,
assert_memory_content_safe,
scan_memory_content,
)
from deerflow.agents.memory.storage import (
FileMemoryStorage,
MemoryStorage,
get_memory_storage,
)
from deerflow.agents.memory.updater import (
MemoryUpdater,
clear_memory_data,
delete_memory_fact,
get_memory_data,
reload_memory_data,
update_memory_from_conversation,
)
__all__ = [
# Memory V2
"MemoryProvider",
"MemoryManager",
"build_memory_manager",
"BuiltinFileProvider",
"MemoryContentBlocked",
"scan_memory_content",
"assert_memory_content_safe",
# Prompt utilities
"MEMORY_UPDATE_PROMPT",
"FACT_EXTRACTION_PROMPT",
"format_memory_for_injection",
"format_conversation_for_update",
# Queue
"ConversationContext",
"MemoryUpdateQueue",
"get_memory_queue",
"reset_memory_queue",
# Storage
"MemoryStorage",
"FileMemoryStorage",
"get_memory_storage",
# Updater
"MemoryUpdater",
"clear_memory_data",
"delete_memory_fact",
"get_memory_data",
"reload_memory_data",
"update_memory_from_conversation",
]
"""DeerFlow 记忆模块 (Memory V2)。
架构 (详见 docs/MEMORY_V2_DESIGN_ZH.md):
- :class:`MemoryProvider` 可插拔记忆 provider 抽象基类
- :class:`MemoryManager` 编排内建 + 至多 1 个外部 provider
- :func:`build_memory_manager` 按生效配置组装 manager 的统一入口
- :class:`BuiltinFileProvider` 本地 USER.md / MEMORY.md provider
- :class:`HindsightProvider` Hindsight HTTP 外部 provider
- :mod:`deerflow.agents.memory.security` 记忆内容安全扫描
记忆有两条自动写入路径:LLM 主动调 ``memory`` / ``hindsight_*`` 工具,
以及 Hindsight 的 auto-retain。注入走系统提示词 (builtin frozen snapshot)
与 ``MemoryMiddleware.wrap_model_call`` (Hindsight 动态召回)。
"""
from deerflow.agents.memory.manager import MemoryManager, build_memory_manager
from deerflow.agents.memory.message_processing import filter_messages_for_memory
from deerflow.agents.memory.provider import MemoryProvider
from deerflow.agents.memory.providers.builtin import BuiltinFileProvider
from deerflow.agents.memory.providers.hindsight import HindsightProvider
from deerflow.agents.memory.security import (
MemoryContentBlocked,
assert_memory_content_safe,
scan_memory_content,
)
__all__ = [
"MemoryProvider",
"MemoryManager",
"build_memory_manager",
"BuiltinFileProvider",
"HindsightProvider",
"MemoryContentBlocked",
"scan_memory_content",
"assert_memory_content_safe",
"filter_messages_for_memory",
]
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/message_processing.py
+5 -53
View File
@@ -1,4 +1,8 @@
"""Shared helpers for turning conversations into memory update inputs."""
"""把对话整理成记忆写入所需输入的共享辅助函数。
Memory V2 说明:后台抽取已舍弃,这里只保留 ``filter_messages_for_memory``
(供 MemoryMiddleware 取出最后一轮对话用于 Hindsight retain)。
"""
from __future__ import annotations
@@ -7,34 +11,6 @@ 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:
@@ -83,27 +59,3 @@ def filter_messages_for_memory(messages: list[Any]) -> list[Any]:
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
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/prompt.py
-363
View File
@@ -1,363 +0,0 @@
"""Prompt templates for memory update and injection."""
import math
import re
from typing import Any
try:
import tiktoken
TIKTOKEN_AVAILABLE = True
except ImportError:
TIKTOKEN_AVAILABLE = False
# Prompt template for updating memory based on conversation
MEMORY_UPDATE_PROMPT = """You are a memory management system. Your task is to analyze a conversation and update the user's memory profile.
Current Memory State:
<current_memory>
{current_memory}
</current_memory>
New Conversation to Process:
<conversation>
{conversation}
</conversation>
Instructions:
1. Analyze the conversation for important information about the user
2. Extract relevant facts, preferences, and context with specific details (numbers, names, technologies)
3. Update the memory sections as needed following the detailed length guidelines below
Before extracting facts, perform a structured reflection on the conversation:
1. Error/Retry Detection: Did the agent encounter errors, require retries, or produce incorrect results?
If yes, record the root cause and correct approach as a high-confidence fact with category "correction".
2. User Correction Detection: Did the user correct the agent's direction, understanding, or output?
If yes, record the correct interpretation or approach as a high-confidence fact with category "correction".
Include what went wrong in "sourceError" only when category is "correction" and the mistake is explicit in the conversation.
3. Project Constraint Discovery: Were any project-specific constraints discovered during the conversation?
If yes, record them as facts with the most appropriate category and confidence.
{correction_hint}
Memory Section Guidelines:
**User Context** (Current state - concise summaries):
- workContext: Professional role, company, key projects, main technologies (2-3 sentences)
Example: Core contributor, project names with metrics (16k+ stars), technical stack
- personalContext: Languages, communication preferences, key interests (1-2 sentences)
Example: Bilingual capabilities, specific interest areas, expertise domains
- topOfMind: Multiple ongoing focus areas and priorities (3-5 sentences, detailed paragraph)
Example: Primary project work, parallel technical investigations, ongoing learning/tracking
Include: Active implementation work, troubleshooting issues, market/research interests
Note: This captures SEVERAL concurrent focus areas, not just one task
**History** (Temporal context - rich paragraphs):
- recentMonths: Detailed summary of recent activities (4-6 sentences or 1-2 paragraphs)
Timeline: Last 1-3 months of interactions
Include: Technologies explored, projects worked on, problems solved, interests demonstrated
- earlierContext: Important historical patterns (3-5 sentences or 1 paragraph)
Timeline: 3-12 months ago
Include: Past projects, learning journeys, established patterns
- longTermBackground: Persistent background and foundational context (2-4 sentences)
Timeline: Overall/foundational information
Include: Core expertise, longstanding interests, fundamental working style
**Facts Extraction**:
- Extract specific, quantifiable details (e.g., "16k+ GitHub stars", "200+ datasets")
- Include proper nouns (company names, project names, technology names)
- Preserve technical terminology and version numbers
- Categories:
* preference: Tools, styles, approaches user prefers/dislikes
* knowledge: Specific expertise, technologies mastered, domain knowledge
* context: Background facts (job title, projects, locations, languages)
* behavior: Working patterns, communication habits, problem-solving approaches
* goal: Stated objectives, learning targets, project ambitions
* correction: Explicit agent mistakes or user corrections, including the correct approach
- Confidence levels:
* 0.9-1.0: Explicitly stated facts ("I work on X", "My role is Y")
* 0.7-0.8: Strongly implied from actions/discussions
* 0.5-0.6: Inferred patterns (use sparingly, only for clear patterns)
**What Goes Where**:
- workContext: Current job, active projects, primary tech stack
- personalContext: Languages, personality, interests outside direct work tasks
- topOfMind: Multiple ongoing priorities and focus areas user cares about recently (gets updated most frequently)
Should capture 3-5 concurrent themes: main work, side explorations, learning/tracking interests
- recentMonths: Detailed account of recent technical explorations and work
- earlierContext: Patterns from slightly older interactions still relevant
- longTermBackground: Unchanging foundational facts about the user
**Multilingual Content**:
- Preserve original language for proper nouns and company names
- Keep technical terms in their original form (DeepSeek, LangGraph, etc.)
- Note language capabilities in personalContext
Output Format (JSON):
{{
"user": {{
"workContext": {{ "summary": "...", "shouldUpdate": true/false }},
"personalContext": {{ "summary": "...", "shouldUpdate": true/false }},
"topOfMind": {{ "summary": "...", "shouldUpdate": true/false }}
}},
"history": {{
"recentMonths": {{ "summary": "...", "shouldUpdate": true/false }},
"earlierContext": {{ "summary": "...", "shouldUpdate": true/false }},
"longTermBackground": {{ "summary": "...", "shouldUpdate": true/false }}
}},
"newFacts": [
{{ "content": "...", "category": "preference|knowledge|context|behavior|goal|correction", "confidence": 0.0-1.0 }}
],
"factsToRemove": ["fact_id_1", "fact_id_2"]
}}
Important Rules:
- Only set shouldUpdate=true if there's meaningful new information
- Follow length guidelines: workContext/personalContext are concise (1-3 sentences), topOfMind and history sections are detailed (paragraphs)
- Include specific metrics, version numbers, and proper nouns in facts
- Only add facts that are clearly stated (0.9+) or strongly implied (0.7+)
- Use category "correction" for explicit agent mistakes or user corrections; assign confidence >= 0.95 when the correction is explicit
- Include "sourceError" only for explicit correction facts when the prior mistake or wrong approach is clearly stated; omit it otherwise
- Remove facts that are contradicted by new information
- When updating topOfMind, integrate new focus areas while removing completed/abandoned ones
Keep 3-5 concurrent focus themes that are still active and relevant
- For history sections, integrate new information chronologically into appropriate time period
- Preserve technical accuracy - keep exact names of technologies, companies, projects
- Focus on information useful for future interactions and personalization
- IMPORTANT: Do NOT record file upload events in memory. Uploaded files are
session-specific and ephemeral — they will not be accessible in future sessions.
Recording upload events causes confusion in subsequent conversations.
Return ONLY valid JSON, no explanation or markdown."""
# Prompt template for extracting facts from a single message
FACT_EXTRACTION_PROMPT = """Extract factual information about the user from this message.
Message:
{message}
Extract facts in this JSON format:
{{
"facts": [
{{ "content": "...", "category": "preference|knowledge|context|behavior|goal|correction", "confidence": 0.0-1.0 }}
]
}}
Categories:
- preference: User preferences (likes/dislikes, styles, tools)
- knowledge: User's expertise or knowledge areas
- context: Background context (location, job, projects)
- behavior: Behavioral patterns
- goal: User's goals or objectives
- correction: Explicit corrections or mistakes to avoid repeating
Rules:
- Only extract clear, specific facts
- Confidence should reflect certainty (explicit statement = 0.9+, implied = 0.6-0.8)
- Skip vague or temporary information
Return ONLY valid JSON."""
def _count_tokens(text: str, encoding_name: str = "cl100k_base") -> int:
"""Count tokens in text using tiktoken.
Args:
text: The text to count tokens for.
encoding_name: The encoding to use (default: cl100k_base for GPT-4/3.5).
Returns:
The number of tokens in the text.
"""
if not TIKTOKEN_AVAILABLE:
# Fallback to character-based estimation if tiktoken is not available
return len(text) // 4
try:
encoding = tiktoken.get_encoding(encoding_name)
return len(encoding.encode(text))
except Exception:
# Fallback to character-based estimation on error
return len(text) // 4
def _coerce_confidence(value: Any, default: float = 0.0) -> float:
"""Coerce a confidence-like value to a bounded float in [0, 1].
Non-finite values (NaN, inf, -inf) are treated as invalid and fall back
to the default before clamping, preventing them from dominating ranking.
The ``default`` parameter is assumed to be a finite value.
"""
try:
confidence = float(value)
except (TypeError, ValueError):
return max(0.0, min(1.0, default))
if not math.isfinite(confidence):
return max(0.0, min(1.0, default))
return max(0.0, min(1.0, confidence))
def format_memory_for_injection(memory_data: dict[str, Any], max_tokens: int = 2000) -> str:
"""Format memory data for injection into system prompt.
Args:
memory_data: The memory data dictionary.
max_tokens: Maximum tokens to use (counted via tiktoken for accuracy).
Returns:
Formatted memory string for system prompt injection.
"""
if not memory_data:
return ""
sections = []
# Format user context
user_data = memory_data.get("user", {})
if user_data:
user_sections = []
work_ctx = user_data.get("workContext", {})
if work_ctx.get("summary"):
user_sections.append(f"Work: {work_ctx['summary']}")
personal_ctx = user_data.get("personalContext", {})
if personal_ctx.get("summary"):
user_sections.append(f"Personal: {personal_ctx['summary']}")
top_of_mind = user_data.get("topOfMind", {})
if top_of_mind.get("summary"):
user_sections.append(f"Current Focus: {top_of_mind['summary']}")
if user_sections:
sections.append("User Context:\n" + "\n".join(f"- {s}" for s in user_sections))
# Format history
history_data = memory_data.get("history", {})
if history_data:
history_sections = []
recent = history_data.get("recentMonths", {})
if recent.get("summary"):
history_sections.append(f"Recent: {recent['summary']}")
earlier = history_data.get("earlierContext", {})
if earlier.get("summary"):
history_sections.append(f"Earlier: {earlier['summary']}")
background = history_data.get("longTermBackground", {})
if background.get("summary"):
history_sections.append(f"Background: {background['summary']}")
if history_sections:
sections.append("History:\n" + "\n".join(f"- {s}" for s in history_sections))
# Format facts (sorted by confidence; include as many as token budget allows)
facts_data = memory_data.get("facts", [])
if isinstance(facts_data, list) and facts_data:
ranked_facts = sorted(
(f for f in facts_data if isinstance(f, dict) and isinstance(f.get("content"), str) and f.get("content").strip()),
key=lambda fact: _coerce_confidence(fact.get("confidence"), default=0.0),
reverse=True,
)
# Compute token count for existing sections once, then account
# incrementally for each fact line to avoid full-string re-tokenization.
base_text = "\n\n".join(sections)
base_tokens = _count_tokens(base_text) if base_text else 0
# Account for the separator between existing sections and the facts section.
facts_header = "Facts:\n"
separator_tokens = _count_tokens("\n\n" + facts_header) if base_text else _count_tokens(facts_header)
running_tokens = base_tokens + separator_tokens
fact_lines: list[str] = []
for fact in ranked_facts:
content_value = fact.get("content")
if not isinstance(content_value, str):
continue
content = content_value.strip()
if not content:
continue
category = str(fact.get("category", "context")).strip() or "context"
confidence = _coerce_confidence(fact.get("confidence"), default=0.0)
source_error = fact.get("sourceError")
if category == "correction" and isinstance(source_error, str) and source_error.strip():
line = f"- [{category} | {confidence:.2f}] {content} (avoid: {source_error.strip()})"
else:
line = f"- [{category} | {confidence:.2f}] {content}"
# Each additional line is preceded by a newline (except the first).
line_text = ("\n" + line) if fact_lines else line
line_tokens = _count_tokens(line_text)
if running_tokens + line_tokens <= max_tokens:
fact_lines.append(line)
running_tokens += line_tokens
else:
break
if fact_lines:
sections.append("Facts:\n" + "\n".join(fact_lines))
if not sections:
return ""
result = "\n\n".join(sections)
# Use accurate token counting with tiktoken
token_count = _count_tokens(result)
if token_count > max_tokens:
# Truncate to fit within token limit
# Estimate characters to remove based on token ratio
char_per_token = len(result) / token_count
target_chars = int(max_tokens * char_per_token * 0.95) # 95% to leave margin
result = result[:target_chars] + "\n..."
return result
def format_conversation_for_update(messages: list[Any]) -> str:
"""Format conversation messages for memory update prompt.
Args:
messages: List of conversation messages.
Returns:
Formatted conversation string.
"""
lines = []
for msg in messages:
role = getattr(msg, "type", "unknown")
content = getattr(msg, "content", str(msg))
# Handle content that might be a list (multimodal)
if isinstance(content, list):
text_parts = []
for p in content:
if isinstance(p, str):
text_parts.append(p)
elif isinstance(p, dict):
text_val = p.get("text")
if isinstance(text_val, str):
text_parts.append(text_val)
content = " ".join(text_parts) if text_parts else str(content)
# Strip uploaded_files tags from human messages to avoid persisting
# ephemeral file path info into long-term memory. Skip the turn entirely
# when nothing remains after stripping (upload-only message).
if role == "human":
content = re.sub(r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", "", str(content)).strip()
if not content:
continue
# Truncate very long messages
if len(str(content)) > 1000:
content = str(content)[:1000] + "..."
if role == "human":
lines.append(f"User: {content}")
elif role == "ai":
lines.append(f"Assistant: {content}")
return "\n\n".join(lines)
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/queue.py
-277
View File
@@ -1,277 +0,0 @@
"""Memory update queue with debounce mechanism."""
import logging
import threading
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from deerflow.config.memory_config import get_memory_config
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
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
def add(
self,
thread_id: str,
messages: list[Any],
agent_name: str | None = None,
user_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).
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,
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,
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,
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,
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,
user_id=user_id,
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()
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:
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,
)
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
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/storage.py
-231
View File
@@ -1,231 +0,0 @@
"""Memory storage providers."""
import abc
import json
import logging
import threading
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from deerflow.config.agents_config import AGENT_NAME_PATTERN
from deerflow.config.memory_config import get_memory_config
from deerflow.config.paths import get_paths
logger = logging.getLogger(__name__)
def utc_now_iso_z() -> str:
"""Current UTC time as ISO-8601 with ``Z`` suffix (matches prior naive-UTC output)."""
return datetime.now(UTC).isoformat().removesuffix("+00:00") + "Z"
def create_empty_memory() -> dict[str, Any]:
"""Create an empty memory structure."""
return {
"version": "1.0",
"lastUpdated": utc_now_iso_z(),
"user": {
"workContext": {"summary": "", "updatedAt": ""},
"personalContext": {"summary": "", "updatedAt": ""},
"topOfMind": {"summary": "", "updatedAt": ""},
},
"history": {
"recentMonths": {"summary": "", "updatedAt": ""},
"earlierContext": {"summary": "", "updatedAt": ""},
"longTermBackground": {"summary": "", "updatedAt": ""},
},
"facts": [],
}
class MemoryStorage(abc.ABC):
"""Abstract base class for memory storage providers."""
@abc.abstractmethod
def load(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Load memory data for the given agent."""
pass
@abc.abstractmethod
def reload(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Force reload memory data for the given agent."""
pass
@abc.abstractmethod
def save(self, memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> bool:
"""Save memory data for the given agent."""
pass
class FileMemoryStorage(MemoryStorage):
"""File-based memory storage provider."""
def __init__(self):
"""Initialize the file memory storage."""
# Per-user/agent memory cache: keyed by (user_id, agent_name) tuple (None = global)
# Value: (memory_data, file_mtime)
self._memory_cache: dict[tuple[str | None, str | None], tuple[dict[str, Any], float | None]] = {}
# Guards all reads and writes to _memory_cache across concurrent callers.
self._cache_lock = threading.Lock()
def _validate_agent_name(self, agent_name: str) -> None:
"""Validate that the agent name is safe to use in filesystem paths.
Uses the repository's established AGENT_NAME_PATTERN to ensure consistency
across the codebase and prevent path traversal or other problematic characters.
"""
if not agent_name:
raise ValueError("Agent name must be a non-empty string.")
if not AGENT_NAME_PATTERN.match(agent_name):
raise ValueError(f"Invalid agent name {agent_name!r}: names must match {AGENT_NAME_PATTERN.pattern}")
def _get_memory_file_path(self, agent_name: str | None = None, *, user_id: str | None = None) -> Path:
"""Get the path to the memory file."""
if user_id is not None:
if agent_name is not None:
self._validate_agent_name(agent_name)
return get_paths().user_agent_memory_file(user_id, agent_name)
config = get_memory_config()
if config.storage_path and Path(config.storage_path).is_absolute():
return Path(config.storage_path)
return get_paths().user_memory_file(user_id)
# Legacy: no user_id
if agent_name is not None:
self._validate_agent_name(agent_name)
return get_paths().agent_memory_file(agent_name)
config = get_memory_config()
if config.storage_path:
p = Path(config.storage_path)
return p if p.is_absolute() else get_paths().base_dir / p
return get_paths().memory_file
def _load_memory_from_file(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Load memory data from file."""
file_path = self._get_memory_file_path(agent_name, user_id=user_id)
if not file_path.exists():
return create_empty_memory()
try:
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
return data
except (json.JSONDecodeError, OSError) as e:
logger.warning("Failed to load memory file: %s", e)
return create_empty_memory()
@staticmethod
def _cache_key(agent_name: str | None = None, *, user_id: str | None = None) -> tuple[str | None, str | None]:
return (user_id, agent_name)
def load(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Load memory data (cached with file modification time check)."""
file_path = self._get_memory_file_path(agent_name, user_id=user_id)
cache_key = self._cache_key(agent_name, user_id=user_id)
try:
current_mtime = file_path.stat().st_mtime if file_path.exists() else None
except OSError:
current_mtime = None
with self._cache_lock:
cached = self._memory_cache.get(cache_key)
if cached is not None and cached[1] == current_mtime:
return cached[0]
memory_data = self._load_memory_from_file(agent_name, user_id=user_id)
with self._cache_lock:
self._memory_cache[cache_key] = (memory_data, current_mtime)
return memory_data
def reload(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Reload memory data from file, forcing cache invalidation."""
file_path = self._get_memory_file_path(agent_name, user_id=user_id)
memory_data = self._load_memory_from_file(agent_name, user_id=user_id)
cache_key = self._cache_key(agent_name, user_id=user_id)
try:
mtime = file_path.stat().st_mtime if file_path.exists() else None
except OSError:
mtime = None
with self._cache_lock:
self._memory_cache[cache_key] = (memory_data, mtime)
return memory_data
def save(self, memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> bool:
"""Save memory data to file and update cache."""
file_path = self._get_memory_file_path(agent_name, user_id=user_id)
cache_key = self._cache_key(agent_name, user_id=user_id)
try:
file_path.parent.mkdir(parents=True, exist_ok=True)
# Shallow-copy before adding lastUpdated so the caller's dict is not
# mutated as a side-effect, and the cache reference is not silently
# updated before the file write succeeds.
memory_data = {**memory_data, "lastUpdated": utc_now_iso_z()}
temp_path = file_path.with_suffix(f".{uuid.uuid4().hex}.tmp")
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(memory_data, f, indent=2, ensure_ascii=False)
temp_path.replace(file_path)
try:
mtime = file_path.stat().st_mtime
except OSError:
mtime = None
with self._cache_lock:
self._memory_cache[cache_key] = (memory_data, mtime)
logger.info("Memory saved to %s", file_path)
return True
except OSError as e:
logger.error("Failed to save memory file: %s", e)
return False
_storage_instance: MemoryStorage | None = None
_storage_lock = threading.Lock()
def get_memory_storage() -> MemoryStorage:
"""Get the configured memory storage instance."""
global _storage_instance
if _storage_instance is not None:
return _storage_instance
with _storage_lock:
if _storage_instance is not None:
return _storage_instance
config = get_memory_config()
storage_class_path = config.storage_class
try:
module_path, class_name = storage_class_path.rsplit(".", 1)
import importlib
module = importlib.import_module(module_path)
storage_class = getattr(module, class_name)
# Validate that the configured storage is a MemoryStorage implementation
if not isinstance(storage_class, type):
raise TypeError(f"Configured memory storage '{storage_class_path}' is not a class: {storage_class!r}")
if not issubclass(storage_class, MemoryStorage):
raise TypeError(f"Configured memory storage '{storage_class_path}' is not a subclass of MemoryStorage")
_storage_instance = storage_class()
except Exception as e:
logger.error(
"Failed to load memory storage %s, falling back to FileMemoryStorage: %s",
storage_class_path,
e,
)
_storage_instance = FileMemoryStorage()
return _storage_instance
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/summarization_hook.py
-31
View File
@@ -1,31 +0,0 @@
"""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,
)
offline-backend-20260512/backend/packages/harness/deerflow/agents/memory/updater.py
-723
View File
@@ -1,723 +0,0 @@
"""Memory updater for reading, writing, and updating memory data."""
import asyncio
import atexit
import concurrent.futures
import copy
import json
import logging
import math
import re
import uuid
from typing import Any
from deerflow.agents.memory.prompt import (
MEMORY_UPDATE_PROMPT,
format_conversation_for_update,
)
from deerflow.agents.memory.storage import (
create_empty_memory,
get_memory_storage,
utc_now_iso_z,
)
from deerflow.config.memory_config import get_memory_config
from deerflow.models import create_chat_model
logger = logging.getLogger(__name__)
# Thread pool for offloading sync memory updates when called from an async
# context. Unlike the previous asyncio.run() approach, this runs *sync*
# model.invoke() calls — no event loop is created, so the langchain async
# httpx client pool (globally cached via @lru_cache) is never touched and
# cross-loop connection reuse is impossible.
_SYNC_MEMORY_UPDATER_EXECUTOR = concurrent.futures.ThreadPoolExecutor(
max_workers=4,
thread_name_prefix="memory-updater-sync",
)
atexit.register(lambda: _SYNC_MEMORY_UPDATER_EXECUTOR.shutdown(wait=False))
def _create_empty_memory() -> dict[str, Any]:
"""Backward-compatible wrapper around the storage-layer empty-memory factory."""
return create_empty_memory()
def _save_memory_to_file(memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> bool:
"""Backward-compatible wrapper around the configured memory storage save path."""
return get_memory_storage().save(memory_data, agent_name, user_id=user_id)
def get_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Get the current memory data via storage provider."""
return get_memory_storage().load(agent_name, user_id=user_id)
def reload_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Reload memory data via storage provider."""
return get_memory_storage().reload(agent_name, user_id=user_id)
def import_memory_data(memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Persist imported memory data via storage provider.
Args:
memory_data: Full memory payload to persist.
agent_name: If provided, imports into per-agent memory.
user_id: If provided, scopes memory to a specific user.
Returns:
The saved memory data after storage normalization.
Raises:
OSError: If persisting the imported memory fails.
"""
storage = get_memory_storage()
if not storage.save(memory_data, agent_name, user_id=user_id):
raise OSError("Failed to save imported memory data")
return storage.load(agent_name, user_id=user_id)
def clear_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Clear all stored memory data and persist an empty structure."""
cleared_memory = create_empty_memory()
if not _save_memory_to_file(cleared_memory, agent_name, user_id=user_id):
raise OSError("Failed to save cleared memory data")
return cleared_memory
def _validate_confidence(confidence: float) -> float:
"""Validate persisted fact confidence so stored JSON stays standards-compliant."""
if not math.isfinite(confidence) or confidence < 0 or confidence > 1:
raise ValueError("confidence")
return confidence
def create_memory_fact(
content: str,
category: str = "context",
confidence: float = 0.5,
agent_name: str | None = None,
*,
user_id: str | None = None,
) -> dict[str, Any]:
"""Create a new fact and persist the updated memory data."""
normalized_content = content.strip()
if not normalized_content:
raise ValueError("content")
normalized_category = category.strip() or "context"
validated_confidence = _validate_confidence(confidence)
now = utc_now_iso_z()
memory_data = get_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
facts = list(memory_data.get("facts", []))
facts.append(
{
"id": f"fact_{uuid.uuid4().hex[:8]}",
"content": normalized_content,
"category": normalized_category,
"confidence": validated_confidence,
"createdAt": now,
"source": "manual",
}
)
updated_memory["facts"] = facts
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
raise OSError("Failed to save memory data after creating fact")
return updated_memory
def delete_memory_fact(fact_id: str, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Delete a fact by its id and persist the updated memory data."""
memory_data = get_memory_data(agent_name, user_id=user_id)
facts = memory_data.get("facts", [])
updated_facts = [fact for fact in facts if fact.get("id") != fact_id]
if len(updated_facts) == len(facts):
raise KeyError(fact_id)
updated_memory = dict(memory_data)
updated_memory["facts"] = updated_facts
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
raise OSError(f"Failed to save memory data after deleting fact '{fact_id}'")
return updated_memory
def update_memory_fact(
fact_id: str,
content: str | None = None,
category: str | None = None,
confidence: float | None = None,
agent_name: str | None = None,
*,
user_id: str | None = None,
) -> dict[str, Any]:
"""Update an existing fact and persist the updated memory data."""
memory_data = get_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
updated_facts: list[dict[str, Any]] = []
found = False
for fact in memory_data.get("facts", []):
if fact.get("id") == fact_id:
found = True
updated_fact = dict(fact)
if content is not None:
normalized_content = content.strip()
if not normalized_content:
raise ValueError("content")
updated_fact["content"] = normalized_content
if category is not None:
updated_fact["category"] = category.strip() or "context"
if confidence is not None:
updated_fact["confidence"] = _validate_confidence(confidence)
updated_facts.append(updated_fact)
else:
updated_facts.append(fact)
if not found:
raise KeyError(fact_id)
updated_memory["facts"] = updated_facts
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
raise OSError(f"Failed to save memory data after updating fact '{fact_id}'")
return updated_memory
def _extract_text(content: Any) -> str:
"""Extract plain text from LLM response content (str or list of content blocks).
Modern LLMs may return structured content as a list of blocks instead of a
plain string, e.g. [{"type": "text", "text": "..."}]. Using str() on such
content produces Python repr instead of the actual text, breaking JSON
parsing downstream.
String chunks are concatenated without separators to avoid corrupting
chunked JSON/text payloads. Dict-based text blocks are treated as full text
blocks and joined with newlines for readability.
"""
if isinstance(content, str):
return content
if isinstance(content, list):
pieces: list[str] = []
pending_str_parts: list[str] = []
def flush_pending_str_parts() -> None:
if pending_str_parts:
pieces.append("".join(pending_str_parts))
pending_str_parts.clear()
for block in content:
if isinstance(block, str):
pending_str_parts.append(block)
elif isinstance(block, dict):
flush_pending_str_parts()
text_val = block.get("text")
if isinstance(text_val, str):
pieces.append(text_val)
flush_pending_str_parts()
return "\n".join(pieces)
return str(content)
# Matches sentences that describe a file-upload *event* rather than general
# file-related work. Deliberately narrow to avoid removing legitimate facts
# such as "User works with CSV files" or "prefers PDF export".
_UPLOAD_SENTENCE_RE = re.compile(
r"[^.!?]*\b(?:"
r"upload(?:ed|ing)?(?:\s+\w+){0,3}\s+(?:file|files?|document|documents?|attachment|attachments?)"
r"|file\s+upload"
r"|/mnt/user-data/uploads/"
r"|<uploaded_files>"
r")[^.!?]*[.!?]?\s*",
re.IGNORECASE,
)
def _strip_upload_mentions_from_memory(memory_data: dict[str, Any]) -> dict[str, Any]:
"""Remove sentences about file uploads from all memory summaries and facts.
Uploaded files are session-scoped; persisting upload events in long-term
memory causes the agent to search for non-existent files in future sessions.
"""
# Scrub summaries in user/history sections
for section in ("user", "history"):
section_data = memory_data.get(section, {})
for _key, val in section_data.items():
if isinstance(val, dict) and "summary" in val:
cleaned = _UPLOAD_SENTENCE_RE.sub("", val["summary"]).strip()
cleaned = re.sub(r" +", " ", cleaned)
val["summary"] = cleaned
# Also remove any facts that describe upload events
facts = memory_data.get("facts", [])
if facts:
memory_data["facts"] = [f for f in facts if not _UPLOAD_SENTENCE_RE.search(f.get("content", ""))]
return memory_data
def _fact_content_key(content: Any) -> str | None:
if not isinstance(content, str):
return None
stripped = content.strip()
if not stripped:
return None
return stripped.casefold()
# Match a fenced code block, optionally with an info string (e.g. ```json).
# We deliberately use a non-greedy DOTALL pattern so that *only* the contents
# are kept — no leading prose like "Here is the updated memory:" and no
# trailing chatter that would break ``json.loads``.
_FENCED_BLOCK_RE = re.compile(r"```[a-zA-Z0-9_-]*\s*\n?(.*?)```", re.DOTALL)
# Match common "thinking" wrappers that some Chinese-domestic LLMs emit even
# when the system prompt asks for JSON only (GLM, Qwen, MindIE-served models,
# and others). Stripping these *before* JSON extraction prevents the body of
# the reasoning text from being mistakenly parsed as the payload.
_THINK_BLOCK_RE = re.compile(
r"<\s*(think|thinking|reasoning|analysis)\s*>.*?<\s*/\s*\1\s*>",
re.DOTALL | re.IGNORECASE,
)
def _extract_first_json_object(text: str) -> str | None:
"""Return the substring spanning the first balanced ``{...}`` JSON object.
Walks the text once tracking brace depth, ignoring braces that appear
inside double-quoted strings (with backslash-escape awareness). This is
sufficient for the JSON dialect the prompt asks for; we deliberately do
not handle single-quoted "JSON" because that would silently accept
invalid responses we'd rather reject and retry on.
"""
start = text.find("{")
if start == -1:
return None
depth = 0
in_string = False
escape = False
for index in range(start, len(text)):
char = text[index]
if in_string:
if escape:
escape = False
continue
if char == "\\":
escape = True
continue
if char == '"':
in_string = False
continue
if char == '"':
in_string = True
continue
if char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return text[start : index + 1]
return None
def _parse_memory_update_response(response_text: str) -> dict[str, Any]:
"""Best-effort extraction of the JSON payload from a memory-update response.
Models served by ``vllm`` / ``MindIE`` (e.g. the GLM_5 endpoint at
``http://10.106.57.108:8010/GLM_5/v1/chat/completions`` users have hit
in production) routinely:
1. Wrap the JSON in a fenced code block (\\`\\`\\`json ... \\`\\`\\`)
even though the prompt forbids it.
2. Emit a ``<think>...</think>`` reasoning block before the answer.
3. Add prose ("Here is the updated memory:") before/after the JSON.
4. Return an empty body when the upstream model produced only reasoning
and no answer (the original cause of "Expecting value: line 1
column 1 (char 0)" reports).
This helper normalizes all four cases and surfaces a clear error
message containing a truncated response snippet so future failures are
diagnosable from the log alone.
"""
if not isinstance(response_text, str):
raise json.JSONDecodeError("Empty memory update response (non-string content)", "", 0)
cleaned = _THINK_BLOCK_RE.sub("", response_text).strip()
if not cleaned:
raise json.JSONDecodeError("Empty memory update response (only reasoning content)", response_text, 0)
# 1. Try fenced code block first — favours the model's explicit framing.
fenced_match = _FENCED_BLOCK_RE.search(cleaned)
if fenced_match:
candidate = fenced_match.group(1).strip()
if candidate:
try:
return json.loads(candidate)
except json.JSONDecodeError:
# Fall through to balanced-brace extraction; the fenced block
# may have contained truncated JSON or interleaved prose.
pass
# 2. Try balanced-brace JSON extraction anywhere in the text.
balanced = _extract_first_json_object(cleaned)
if balanced is not None:
try:
return json.loads(balanced)
except json.JSONDecodeError:
pass
# 3. Last resort: try parsing the whole cleaned text directly. This
# covers responses that already are pure JSON.
try:
return json.loads(cleaned)
except json.JSONDecodeError as exc:
snippet = response_text.strip()
if len(snippet) > 400:
snippet = snippet[:400] + "...[truncated]"
raise json.JSONDecodeError(
f"{exc.msg}; raw response head: {snippet!r}",
exc.doc,
exc.pos,
) from exc
class MemoryUpdater:
"""Updates memory using LLM based on conversation context."""
def __init__(self, model_name: str | None = None):
"""Initialize the memory updater.
Args:
model_name: Optional model name to use. If None, uses config or default.
"""
self._model_name = model_name
def _get_model(self):
"""Get the model for memory updates."""
config = get_memory_config()
model_name = self._model_name or config.model_name
return create_chat_model(name=model_name, thinking_enabled=False)
def _build_correction_hint(
self,
correction_detected: bool,
reinforcement_detected: bool,
) -> str:
"""Build optional prompt hints for correction and reinforcement signals."""
correction_hint = ""
if correction_detected:
correction_hint = (
"IMPORTANT: Explicit correction signals were detected in this conversation. "
"Pay special attention to what the agent got wrong, what the user corrected, "
"and record the correct approach as a fact with category "
'"correction" and confidence >= 0.95 when appropriate.'
)
if reinforcement_detected:
reinforcement_hint = (
"IMPORTANT: Positive reinforcement signals were detected in this conversation. "
"The user explicitly confirmed the agent's approach was correct or helpful. "
"Record the confirmed approach, style, or preference as a fact with category "
'"preference" or "behavior" and confidence >= 0.9 when appropriate.'
)
correction_hint = (correction_hint + "\n" + reinforcement_hint).strip() if correction_hint else reinforcement_hint
return correction_hint
def _prepare_update_prompt(
self,
messages: list[Any],
agent_name: str | None,
correction_detected: bool,
reinforcement_detected: bool,
user_id: str | None = None,
) -> tuple[dict[str, Any], str] | None:
"""Load memory and build the update prompt for a conversation."""
config = get_memory_config()
if not config.enabled or not messages:
return None
current_memory = get_memory_data(agent_name, user_id=user_id)
conversation_text = format_conversation_for_update(messages)
if not conversation_text.strip():
return None
correction_hint = self._build_correction_hint(
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
)
prompt = MEMORY_UPDATE_PROMPT.format(
current_memory=json.dumps(current_memory, indent=2),
conversation=conversation_text,
correction_hint=correction_hint,
)
return current_memory, prompt
def _finalize_update(
self,
current_memory: dict[str, Any],
response_content: Any,
thread_id: str | None,
agent_name: str | None,
user_id: str | None = None,
) -> bool:
"""Parse the model response, apply updates, and persist memory."""
response_text = _extract_text(response_content)
update_data = _parse_memory_update_response(response_text)
# Deep-copy before in-place mutation so a subsequent save() failure
# cannot corrupt the still-cached original object reference.
updated_memory = self._apply_updates(copy.deepcopy(current_memory), update_data, thread_id)
updated_memory = _strip_upload_mentions_from_memory(updated_memory)
return get_memory_storage().save(updated_memory, agent_name, user_id=user_id)
async def aupdate_memory(
self,
messages: list[Any],
thread_id: str | None = None,
agent_name: str | None = None,
correction_detected: bool = False,
reinforcement_detected: bool = False,
user_id: str | None = None,
) -> bool:
"""Update memory asynchronously by delegating to the sync path.
Uses ``asyncio.to_thread`` to run the *sync* ``model.invoke()`` path
in a worker thread so no second event loop is created and the
langchain async httpx client pool (shared with the lead agent) is
never touched. This eliminates the cross-loop connection-reuse bug
described in issue #2615.
"""
return await asyncio.to_thread(
self._do_update_memory_sync,
messages=messages,
thread_id=thread_id,
agent_name=agent_name,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
user_id=user_id,
)
def _do_update_memory_sync(
self,
messages: list[Any],
thread_id: str | None = None,
agent_name: str | None = None,
correction_detected: bool = False,
reinforcement_detected: bool = False,
user_id: str | None = None,
) -> bool:
"""Pure-sync memory update using ``model.invoke()``.
Uses the *sync* LLM call path so no event loop is created. This
guarantees that the langchain provider's globally cached async
httpx ``AsyncClient`` / connection pool (the one shared with the
lead agent) is never touched — no cross-loop connection reuse is
possible.
"""
try:
prepared = self._prepare_update_prompt(
messages=messages,
agent_name=agent_name,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
user_id=user_id,
)
if prepared is None:
return False
current_memory, prompt = prepared
model = self._get_model()
response = model.invoke(prompt, config={"run_name": "memory_agent"})
return self._finalize_update(
current_memory=current_memory,
response_content=response.content,
thread_id=thread_id,
agent_name=agent_name,
user_id=user_id,
)
except json.JSONDecodeError as e:
logger.warning("Failed to parse LLM response for memory update: %s", e)
return False
except Exception as e:
logger.exception("Memory update failed: %s", e)
return False
def update_memory(
self,
messages: list[Any],
thread_id: str | None = None,
agent_name: str | None = None,
correction_detected: bool = False,
reinforcement_detected: bool = False,
user_id: str | None = None,
) -> bool:
"""Synchronously update memory using the sync LLM path.
Uses ``model.invoke()`` (sync HTTP) which operates on a completely
separate connection pool from the async ``AsyncClient`` shared by
the lead agent. This eliminates the cross-loop connection-reuse
bug described in issue #2615.
When called from within a running event loop (e.g. from a LangGraph
node), the blocking sync call is offloaded to a thread pool so the
caller's loop is not blocked.
Args:
messages: List of conversation messages.
thread_id: Optional thread ID for tracking source.
agent_name: If provided, updates per-agent memory. If None, updates global memory.
correction_detected: Whether recent turns include an explicit correction signal.
reinforcement_detected: Whether recent turns include a positive reinforcement signal.
user_id: If provided, scopes memory to a specific user.
Returns:
True if update was successful, False otherwise.
"""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if loop is not None and loop.is_running():
try:
future = _SYNC_MEMORY_UPDATER_EXECUTOR.submit(
self._do_update_memory_sync,
messages=messages,
thread_id=thread_id,
agent_name=agent_name,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
user_id=user_id,
)
return future.result()
except Exception:
logger.exception("Failed to offload memory update to executor")
return False
return self._do_update_memory_sync(
messages=messages,
thread_id=thread_id,
agent_name=agent_name,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
user_id=user_id,
)
def _apply_updates(
self,
current_memory: dict[str, Any],
update_data: dict[str, Any],
thread_id: str | None = None,
) -> dict[str, Any]:
"""Apply LLM-generated updates to memory.
Args:
current_memory: Current memory data.
update_data: Updates from LLM.
thread_id: Optional thread ID for tracking.
Returns:
Updated memory data.
"""
config = get_memory_config()
now = utc_now_iso_z()
# Update user sections
user_updates = update_data.get("user", {})
for section in ["workContext", "personalContext", "topOfMind"]:
section_data = user_updates.get(section, {})
if section_data.get("shouldUpdate") and section_data.get("summary"):
current_memory["user"][section] = {
"summary": section_data["summary"],
"updatedAt": now,
}
# Update history sections
history_updates = update_data.get("history", {})
for section in ["recentMonths", "earlierContext", "longTermBackground"]:
section_data = history_updates.get(section, {})
if section_data.get("shouldUpdate") and section_data.get("summary"):
current_memory["history"][section] = {
"summary": section_data["summary"],
"updatedAt": now,
}
# Remove facts
facts_to_remove = set(update_data.get("factsToRemove", []))
if facts_to_remove:
current_memory["facts"] = [f for f in current_memory.get("facts", []) if f.get("id") not in facts_to_remove]
# Add new facts
existing_fact_keys = {fact_key for fact_key in (_fact_content_key(fact.get("content")) for fact in current_memory.get("facts", [])) if fact_key is not None}
new_facts = update_data.get("newFacts", [])
for fact in new_facts:
confidence = fact.get("confidence", 0.5)
if confidence >= config.fact_confidence_threshold:
raw_content = fact.get("content", "")
if not isinstance(raw_content, str):
continue
normalized_content = raw_content.strip()
fact_key = _fact_content_key(normalized_content)
if fact_key is not None and fact_key in existing_fact_keys:
continue
fact_entry = {
"id": f"fact_{uuid.uuid4().hex[:8]}",
"content": normalized_content,
"category": fact.get("category", "context"),
"confidence": confidence,
"createdAt": now,
"source": thread_id or "unknown",
}
source_error = fact.get("sourceError")
if isinstance(source_error, str):
normalized_source_error = source_error.strip()
if normalized_source_error:
fact_entry["sourceError"] = normalized_source_error
current_memory["facts"].append(fact_entry)
if fact_key is not None:
existing_fact_keys.add(fact_key)
# Enforce max facts limit
if len(current_memory["facts"]) > config.max_facts:
# Sort by confidence and keep top ones
current_memory["facts"] = sorted(
current_memory["facts"],
key=lambda f: f.get("confidence", 0),
reverse=True,
)[: config.max_facts]
return current_memory
def update_memory_from_conversation(
messages: list[Any],
thread_id: str | None = None,
agent_name: str | None = None,
correction_detected: bool = False,
reinforcement_detected: bool = False,
user_id: str | None = None,
) -> bool:
"""Convenience function to update memory from a conversation.
Args:
messages: List of conversation messages.
thread_id: Optional thread ID.
agent_name: If provided, updates per-agent memory. If None, updates global memory.
correction_detected: Whether recent turns include an explicit correction signal.
reinforcement_detected: Whether recent turns include a positive reinforcement signal.
user_id: If provided, scopes memory to a specific user.
Returns:
True if successful, False otherwise.
"""
updater = MemoryUpdater()
return updater.update_memory(messages, thread_id, agent_name, correction_detected, reinforcement_detected, user_id=user_id)
offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py
+70 -104
View File
@@ -2,31 +2,32 @@
本中间件负责两件事:
1. **before_model** —— Hindsight 动态召回注入。
1. **wrap_model_call** —— Hindsight 动态召回注入。
每次 LLM 调用前,用最近一条用户消息作为 query 召回 Hindsight 记忆,
注入成一条带 ``<memory-context>`` 标签的消息。仅在 ``provider="hindsight"``
且配置允许时生效;builtin 的 frozen snapshot 走系统提示词,不在这里。
作为一条带 ``<memory-context>`` 标签的消息**临时**追加到本次模型请求里。
关键:用 ``wrap_model_call`` 而非 ``before_model`` —— 注入只作用于这一次
模型调用,不写入持久 state,因此不会泄漏给前端、也不会在历史里堆积。
仅在 ``provider="hindsight"`` 且配置允许时生效;builtin 的 frozen
snapshot 走系统提示词,不在这里。
2. **after_agent** —— Hindsight 自动 retain + 后台抽取排队。
- Hindsight ``auto_retain`` 开启时,把本轮对话写入工作 bank。
- 后台抽取 (``background_extraction.enabled``,默认关) 开启时,把对话
排队给后台 LLM 抽取。
2. **after_agent** —— Hindsight 自动 retain。
Hindsight ``auto_retain`` 开启时,把本轮对话写入工作 bank。
LLM 主动写入走 ``memory`` / ``hindsight_*`` 工具,不经过本中间件。
"""
import asyncio
import logging
from typing import TYPE_CHECKING, Any, override
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.config import get_config
from langgraph.runtime import Runtime
from deerflow.agents.memory.manager import build_memory_manager
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.memory.message_processing import filter_messages_for_memory
from deerflow.config.memory_config import get_effective_memory_config, get_memory_config
from deerflow.runtime.user_context import get_effective_user_id
@@ -35,9 +36,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
#: 注入的 Hindsight 召回消息的固定 id —— 每轮先删旧的再加新的,保证只有一条且总在末尾。
_RECALL_MESSAGE_ID = "deerflow-memory-recall"
def _coerce_text(content: Any) -> str:
"""把消息 content (可能是 str 或多模态 block 列表) 压成纯文本。"""
@@ -57,7 +55,7 @@ def _coerce_text(content: Any) -> str:
def _latest_human_text(messages: list) -> str:
"""返回最近一条用户消息的纯文本。"""
for msg in reversed(messages):
if isinstance(msg, HumanMessage) and getattr(msg, "id", None) != _RECALL_MESSAGE_ID:
if isinstance(msg, HumanMessage):
return _coerce_text(msg.content).strip()
return ""
@@ -83,12 +81,11 @@ class MemoryMiddlewareState(AgentState):
class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
"""记忆中间件:before_model 注入 Hindsight 召回,after_agent 写入与排队。
"""记忆中间件:wrap_model_call 注入 Hindsight 召回,after_agent 自动 retain。
Memory V2 说明:
- builtin 本地记忆的 frozen snapshot 走系统提示词 (``_get_memory_context``)。
- Hindsight 动态召回走本中间件 ``before_model``。
- 后台抽取默认关闭。
- Hindsight 动态召回走本中间件 ``wrap_model_call`` —— 临时注入,不落 state。
"""
state_schema = MemoryMiddlewareState
@@ -106,9 +103,11 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
# ---- 运行时上下文解析 -------------------------------------------------
def _resolve_thread_id(self, runtime: Runtime) -> str | None:
def _resolve_thread_id(self, runtime: Runtime | None) -> str | None:
"""从 runtime context 取 thread_id,回退到 LangGraph configurable。"""
thread_id = runtime.context.get("thread_id") if runtime.context else None
thread_id = None
if runtime is not None and getattr(runtime, "context", None):
thread_id = runtime.context.get("thread_id")
if thread_id:
return thread_id
try:
@@ -116,18 +115,24 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
except RuntimeError:
return None
def _resolve_agent_id(self, runtime: Runtime) -> str:
def _resolve_agent_id(self, runtime: Runtime | None) -> str:
"""解析 agent_id;优先 middleware 构造时的 agent_name,再 runtime context。"""
if self._agent_name:
return self._agent_name
agent_id = runtime.context.get("agent_id") if runtime.context else None
return str(agent_id) if agent_id else "default"
if runtime is not None and getattr(runtime, "context", None):
agent_id = runtime.context.get("agent_id")
if agent_id:
return str(agent_id)
return "default"
# ---- before_model:Hindsight 动态召回注入 ----------------------------
# ---- wrap_model_call:Hindsight 动态召回注入 -------------------------
@override
def before_model(self, state: MemoryMiddlewareState, runtime: Runtime) -> dict | None:
"""LLM 调用前,把 Hindsight 召回结果注入成一条记忆上下文消息。"""
def _build_recall_message(self, request: Any) -> HumanMessage | None:
"""按本次模型请求构建一条 Hindsight 召回消息;无召回内容时返回 None。
召回结果包在 ``<memory-context>`` 标签里并带系统提示,告诉模型这是
背景信息而非用户的新输入。
"""
config = self._memory_config or get_memory_config()
if not config.enabled:
return None
@@ -136,17 +141,16 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
effective_config = get_effective_memory_config(user_id)
if not effective_config.enabled or not effective_config.injection.enabled:
return None
# 只有 Hindsight 提供动态召回;builtin 走系统提示词,无需在此注入。
# 只有 Hindsight 提供动态召回;builtin 走系统提示词。
if effective_config.provider != "hindsight" or not effective_config.injection.include_hindsight:
return None
messages = state.get("messages", [])
if not messages:
return None
messages = getattr(request, "messages", None) or []
query = _latest_human_text(messages)
if not query:
return None
runtime = getattr(request, "runtime", None)
thread_id = self._resolve_thread_id(runtime)
try:
manager = build_memory_manager(
@@ -160,28 +164,42 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
logger.debug("Hindsight 召回注入失败 (非致命):%s", exc)
return None
has_prior = any(getattr(m, "id", None) == _RECALL_MESSAGE_ID for m in messages)
updates: list = []
if has_prior:
# 先删掉上一轮注入的召回消息,保证只有一条且总在末尾
updates.append(RemoveMessage(id=_RECALL_MESSAGE_ID))
if recall and recall.strip():
tag = effective_config.injection.context_tag
fenced = (
f"<{tag}>\n"
"[系统提示:以下是召回的记忆上下文,不是用户的新输入,作为背景信息参考。]\n\n"
f"{recall}\n"
f"</{tag}>"
)
updates.append(HumanMessage(content=fenced, id=_RECALL_MESSAGE_ID))
if not recall or not recall.strip():
return None
return {"messages": updates} if updates else None
tag = effective_config.injection.context_tag
fenced = (
f"<{tag}>\n"
"[系统提示:以下是召回的记忆上下文,不是用户的新输入,作为背景信息参考。]\n\n"
f"{recall}\n"
f"</{tag}>"
)
return HumanMessage(content=fenced)
# ---- after_agent:Hindsight retain + 后台抽取排队 --------------------
@override
def wrap_model_call(self, request: Any, handler: Any) -> Any:
"""模型调用前临时注入 Hindsight 召回 (同步路径)。"""
recall_msg = self._build_recall_message(request)
if recall_msg is not None:
request = request.override(messages=[*request.messages, recall_msg])
return handler(request)
@override
async def awrap_model_call(self, request: Any, handler: Any) -> Any:
"""模型调用前临时注入 Hindsight 召回 (异步路径)。
召回过程是同步阻塞 (含 Hindsight HTTP),放到线程池里跑以免阻塞事件循环。
"""
recall_msg = await asyncio.to_thread(self._build_recall_message, request)
if recall_msg is not None:
request = request.override(messages=[*request.messages, recall_msg])
return await handler(request)
# ---- after_agent:Hindsight 自动 retain -------------------------------
@override
def after_agent(self, state: MemoryMiddlewareState, runtime: Runtime) -> dict | None:
"""agent 完成后写入 Hindsight 并按需排队后台抽取。
"""agent 完成后把本轮对话写入 Hindsight 工作 bank。
Returns:
None (本中间件不改状态)。
@@ -192,51 +210,18 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
user_id = get_effective_user_id()
effective_config = get_effective_memory_config(user_id)
if effective_config.provider != "hindsight" or not effective_config.hindsight.auto_retain:
return None
messages = state.get("messages", [])
if not messages:
return None
filtered_messages = filter_messages_for_memory(messages)
user_messages = [m for m in filtered_messages if getattr(m, "type", None) == "human"]
assistant_messages = [m for m in filtered_messages if getattr(m, "type", None) == "ai"]
if not user_messages or not assistant_messages:
user_text, assistant_text = _last_exchange(filtered_messages)
if not user_text or not assistant_text:
return None
thread_id = self._resolve_thread_id(runtime)
# 1. Hindsight 自动 retain
if effective_config.provider == "hindsight" and effective_config.hindsight.auto_retain:
self._hindsight_retain(
user_id=user_id,
runtime=runtime,
thread_id=thread_id,
effective_config=effective_config,
filtered_messages=filtered_messages,
)
# 2. 后台抽取排队 (默认关闭)
if effective_config.background_extraction.enabled and thread_id:
self._queue_background_extraction(
thread_id=thread_id,
user_id=user_id,
filtered_messages=filtered_messages,
)
return None
def _hindsight_retain(
self,
*,
user_id: str,
runtime: Runtime,
thread_id: str | None,
effective_config: "MemoryConfig",
filtered_messages: list,
) -> None:
"""把最后一轮对话写入 Hindsight 工作 bank。"""
user_text, assistant_text = _last_exchange(filtered_messages)
if not user_text or not assistant_text:
return
try:
manager = build_memory_manager(
user_id=user_id,
@@ -248,23 +233,4 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
except Exception as exc: # noqa: BLE001 - 写入失败不能阻断对话
logger.warning("Hindsight retain 失败:%s", exc)
def _queue_background_extraction(
self,
*,
thread_id: str,
user_id: str,
filtered_messages: list,
) -> None:
"""把对话排队给后台 LLM 抽取。"""
# user_id 在此显式捕获:threading.Timer 在别的线程触发,ContextVar 不传播。
correction_detected = detect_correction(filtered_messages)
reinforcement_detected = not correction_detected and detect_reinforcement(filtered_messages)
queue = get_memory_queue()
queue.add(
thread_id=thread_id,
messages=filtered_messages,
agent_name=self._agent_name,
user_id=user_id,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
)
return None
offline-backend-20260512/backend/packages/harness/deerflow/client.py
+70 -85
View File
@@ -765,27 +765,22 @@ class DeerFlowClient:
]
}
def get_memory(self) -> dict:
"""Get current memory data.
def get_memory(self, agent: str | None = None) -> dict:
"""读取 Memory V2 本地记忆 (USER.md / MEMORY.md)。
Args:
agent: 可选 agent 名;省略表示默认 agent。
Returns:
Memory data dict (see src/agents/memory/updater.py for structure).
``{user_entries, memory_entries, user_usage, memory_usage}``。
"""
from deerflow.agents.memory.updater import get_memory_data
return get_memory_data(user_id=get_effective_user_id())
def export_memory(self) -> dict:
"""Export current memory data for backup or transfer."""
from deerflow.agents.memory.updater import get_memory_data
return get_memory_data(user_id=get_effective_user_id())
def import_memory(self, memory_data: dict) -> dict:
"""Import and persist full memory data."""
from deerflow.agents.memory.updater import import_memory_data
return import_memory_data(memory_data, user_id=get_effective_user_id())
provider = self._build_memory_provider(agent)
return {
"user_entries": provider.read("user"),
"memory_entries": provider.read("memory"),
"user_usage": provider.usage("user"),
"memory_usage": provider.usage("memory"),
}
def get_model(self, name: str) -> dict | None:
"""Get a specific model's configuration by name.
@@ -949,83 +944,73 @@ class DeerFlowClient:
return get_or_new_skill_storage().install_skill_from_archive(skill_path)
# ------------------------------------------------------------------
# Public API — memory management
# Public API — memory management (Memory V2)
# ------------------------------------------------------------------
def reload_memory(self) -> dict:
"""Reload memory data from file, forcing cache invalidation.
def _build_memory_provider(self, agent: str | None = None):
"""按生效配置构建并初始化一个 V2 builtin provider。"""
from deerflow.agents.memory.providers.builtin import BuiltinFileProvider
from deerflow.config.memory_config import get_effective_memory_config
from deerflow.config.paths import get_paths
Returns:
The reloaded memory data dict.
"""
from deerflow.agents.memory.updater import reload_memory_data
return reload_memory_data(user_id=get_effective_user_id())
def clear_memory(self) -> dict:
"""Clear all persisted memory data."""
from deerflow.agents.memory.updater import clear_memory_data
return clear_memory_data(user_id=get_effective_user_id())
def create_memory_fact(self, content: str, category: str = "context", confidence: float = 0.5) -> dict:
"""Create a single fact manually."""
from deerflow.agents.memory.updater import create_memory_fact
return create_memory_fact(content=content, category=category, confidence=confidence)
def delete_memory_fact(self, fact_id: str) -> dict:
"""Delete a single fact from memory by fact id."""
from deerflow.agents.memory.updater import delete_memory_fact
return delete_memory_fact(fact_id)
def update_memory_fact(
self,
fact_id: str,
content: str | None = None,
category: str | None = None,
confidence: float | None = None,
) -> dict:
"""Update a single fact manually, preserving omitted fields."""
from deerflow.agents.memory.updater import update_memory_fact
return update_memory_fact(
fact_id=fact_id,
content=content,
category=category,
confidence=confidence,
user_id = get_effective_user_id()
config = get_effective_memory_config(user_id)
provider = BuiltinFileProvider(
memory_char_limit=config.builtin.memory_char_limit,
user_char_limit=config.builtin.user_char_limit,
deduplicate_on_load=config.builtin.deduplicate_on_load,
security=config.security,
)
provider.initialize(
user_id=user_id,
agent_id=agent or "default",
base_dir=str(get_paths().base_dir),
)
return provider
def add_memory_entry(self, target: str, content: str, agent: str | None = None) -> dict:
"""新增一条记忆条目。``target`` 为 'user'(用户画像) 或 'memory'(工作笔记)。"""
return self._build_memory_provider(agent).add(target, content)
def replace_memory_entry(self, target: str, old_text: str, content: str, agent: str | None = None) -> dict:
"""用 ``old_text`` 短串定位并替换一条记忆条目。"""
return self._build_memory_provider(agent).replace(target, old_text, content)
def remove_memory_entry(self, target: str, old_text: str, agent: str | None = None) -> dict:
"""用 ``old_text`` 短串定位并删除一条记忆条目。"""
return self._build_memory_provider(agent).remove(target, old_text)
def get_memory_config(self) -> dict:
"""Get memory system configuration.
"""读取当前用户的生效记忆配置 (含 per-user 覆盖)。"""
from deerflow.config.memory_config import get_effective_memory_config
return get_effective_memory_config(get_effective_user_id()).model_dump()
def search_memory(self, query: str, mode: str = "recall", agent: str | None = None) -> dict:
"""代理 Hindsight 检索。
Args:
query: 检索查询。
mode: ``recall`` 返回原始片段;``reflect`` 返回 LLM 综合结论。
agent: 可选 agent 名。
Returns:
Memory config dict.
``{available, results, text}``;Hindsight 不可用时 available=False。
"""
from deerflow.config.memory_config import get_memory_config
from deerflow.agents.memory.providers.hindsight import HindsightProvider
from deerflow.config.memory_config import get_effective_memory_config
config = get_memory_config()
return {
"enabled": config.enabled,
"storage_path": config.storage_path,
"debounce_seconds": config.debounce_seconds,
"max_facts": config.max_facts,
"fact_confidence_threshold": config.fact_confidence_threshold,
"injection_enabled": config.injection_enabled,
"max_injection_tokens": config.max_injection_tokens,
}
def get_memory_status(self) -> dict:
"""Get memory status: config + current data.
Returns:
Dict with "config" and "data" keys.
"""
return {
"config": self.get_memory_config(),
"data": self.get_memory(),
}
user_id = get_effective_user_id()
config = get_effective_memory_config(user_id)
if not config.enabled or config.provider != "hindsight":
return {"available": False, "results": [], "text": ""}
provider = HindsightProvider(config.hindsight)
if not provider.is_available():
return {"available": False, "results": [], "text": ""}
provider.initialize(user_id=user_id, agent_id=agent or "default", thread_id="", base_dir="")
if mode == "reflect":
return {"available": True, "results": [], "text": provider.reflect(query)}
return {"available": True, "results": provider.recall(query), "text": ""}
# ------------------------------------------------------------------
# Public API — file uploads
offline-backend-20260512/backend/packages/harness/deerflow/config/memory_config.py
+315 -418
View File
@@ -1,418 +1,315 @@
"""Memory V2 - 记忆子系统配置。
配置结构 (详见 docs/MEMORY_V2_DESIGN_ZH.md §9)
==============================================
``MemoryConfig`` 采用嵌套结构::
memory:
enabled: true
provider: hindsight # 外部 provider 选哪个 (builtin 始终启用)
builtin: {...} # 本地 USER.md / MEMORY.md
hindsight: {...} # Hindsight HTTP 客户端
background_extraction: {...} # 后台 LLM 抽取 (默认关)
injection: {...} # 系统提示词注入
security: {...} # 内容安全扫描
旧版扁平 yaml (``debounce_seconds`` / ``max_facts`` / ``injection_enabled`` 等)
仍可加载:``_migrate_flat_keys`` 校验器会把它们迁移进嵌套结构。
三层配置覆盖
============
``get_effective_memory_config(user_id)`` 按优先级合并 (高 -> 低):
1. per-user 覆盖 ``{base_dir}/users/{user_id}/memory_config.json``
2. 全局覆盖 ``{base_dir}/memory_user_overrides.json``
3. config.yaml ``memory:`` 段 (即 ``get_memory_config()``)
4. Pydantic 默认值
第 1、2 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
"""
from __future__ import annotations
import copy
import json
import logging
from typing import Any
from pydantic import BaseModel, Field, model_validator
logger = logging.getLogger(__name__)
# ===========================================================================
# 嵌套子配置
# ===========================================================================
class BuiltinMemoryConfig(BaseModel):
"""本地文件 provider (USER.md / MEMORY.md) 配置。"""
enabled: bool = Field(default=True, description="是否启用 builtin 本地记忆 provider")
memory_char_limit: int = Field(
default=2200, ge=200, le=20000, description="MEMORY.md 字符配额"
)
user_char_limit: int = Field(
default=1375, ge=200, le=20000, description="USER.md 字符配额"
)
deduplicate_on_load: bool = Field(
default=True, description="加载时是否对条目去重"
)
class HindsightMemoryConfig(BaseModel):
"""Hindsight 外部 provider 配置 (P2 期接入)。"""
mode: str = Field(
default="local_external",
description="连接模式:cloud / local_embedded / local_external",
)
api_url: str = Field(default="http://localhost:8765", description="Hindsight API 地址")
api_key: str = Field(default="", description="API key,建议用 $HINDSIGHT_API_KEY 环境变量")
bank_id_template: str = Field(
default="deerflow-user-{user_id}",
description="全局 user bank 命名模板,占位符 {user_id}/{agent_id}/{thread_id}",
)
work_bank_id_template: str = Field(
default="deerflow-work-{user_id}-{agent_id}",
description="agent 私有工作 bank 命名模板",
)
memory_mode: str = Field(
default="context",
description="工具暴露策略:context(仅自动) / tools(仅工具) / hybrid(两者)",
)
bank_mission: str = Field(default="", description="reflect 推理的身份/框架描述")
bank_retain_mission: str = Field(default="", description="引导抽取内容的 mission")
recall_budget: str = Field(default="mid", description="召回深度:low / mid / high")
recall_max_tokens: int = Field(default=4096, ge=256, le=32768, description="召回结果 token 上限")
recall_max_input_chars: int = Field(default=800, ge=64, le=8000, description="召回 query 长度上限")
auto_recall: bool = Field(default=True, description="每轮对话前是否自动召回")
auto_retain: bool = Field(default=True, description="是否自动写入对话")
retain_async: bool = Field(default=True, description="retain 是否在 Hindsight 服务端异步处理")
retain_every_n_turns: int = Field(default=1, ge=1, le=50, description="每 N 轮 retain 一次")
retain_user_prefix: str = Field(default="User", description="自动 retain 时用户发言前缀")
retain_assistant_prefix: str = Field(default="Assistant", description="自动 retain 时助手发言前缀")
class BackgroundExtractionConfig(BaseModel):
"""后台 LLM 抽取 facts 配置。默认关闭,作为 LLM 漏记的兜底。"""
enabled: bool = Field(default=False, description="是否启用后台抽取 (默认关)")
debounce_seconds: int = Field(
default=30, ge=1, le=300, description="抽取前的 debounce 等待秒数"
)
model_name: str | None = Field(default=None, description="抽取用的模型名 (None = 默认模型)")
fact_confidence_threshold: float = Field(
default=0.7, ge=0.0, le=1.0, description="fact 入库的最低置信度"
)
max_facts: int = Field(default=100, ge=10, le=500, description="存储的 fact 数量上限")
max_facts_per_extraction: int = Field(
default=10, ge=1, le=50, description="单次抽取产出的 fact 上限"
)
correction_detection: bool = Field(default=True, description="是否检测用户纠错信号")
reinforcement_detection: bool = Field(default=True, description="是否检测正反馈信号")
class InjectionConfig(BaseModel):
"""系统提示词记忆注入配置。"""
enabled: bool = Field(default=True, description="是否往系统提示词注入记忆")
max_tokens: int = Field(default=2000, ge=100, le=8000, description="注入 token 上限")
include_builtin: bool = Field(default=True, description="是否注入 builtin 本地记忆")
include_hindsight: bool = Field(default=True, description="是否注入 Hindsight 召回结果")
context_tag: str = Field(
default="memory-context", description="包裹外部召回内容的标签名"
)
class SecurityConfig(BaseModel):
"""记忆内容安全配置。"""
scan_content: bool = Field(default=True, description="是否扫描注入/外渗模式")
block_invisible_unicode: bool = Field(default=True, description="是否阻断不可见 unicode")
streaming_scrubber: bool = Field(
default=True, description="SSE 流出前是否清洗 <memory-context> 标签 (P3)"
)
# ===========================================================================
# 顶层配置
# ===========================================================================
# 旧版扁平 key -> (嵌套子段, 嵌套字段名) 的迁移映射。
_FLAT_KEY_MIGRATION: dict[str, tuple[str, str]] = {
"debounce_seconds": ("background_extraction", "debounce_seconds"),
"model_name": ("background_extraction", "model_name"),
"max_facts": ("background_extraction", "max_facts"),
"fact_confidence_threshold": ("background_extraction", "fact_confidence_threshold"),
"injection_enabled": ("injection", "enabled"),
"max_injection_tokens": ("injection", "max_tokens"),
}
class MemoryConfig(BaseModel):
"""记忆子系统的全局配置 (config.yaml 的 ``memory:`` 段)。"""
enabled: bool = Field(default=True, description="记忆子系统总开关")
provider: str = Field(
default="builtin",
description="启用的外部 provider 名:builtin(无外部) 或 hindsight",
)
builtin: BuiltinMemoryConfig = Field(default_factory=BuiltinMemoryConfig)
hindsight: HindsightMemoryConfig = Field(default_factory=HindsightMemoryConfig)
background_extraction: BackgroundExtractionConfig = Field(
default_factory=BackgroundExtractionConfig
)
injection: InjectionConfig = Field(default_factory=InjectionConfig)
security: SecurityConfig = Field(default_factory=SecurityConfig)
# --- 旧版扁平字段 (legacy storage.py 专用,V2 不使用,P3 移除) ---
storage_path: str = Field(
default="",
description="[旧版] 一体化 memory.json 路径,仅 FileMemoryStorage 使用",
)
storage_class: str = Field(
default="deerflow.agents.memory.storage.FileMemoryStorage",
description="[旧版] 记忆存储类路径,仅 legacy storage 使用",
)
@model_validator(mode="before")
@classmethod
def _migrate_flat_keys(cls, data: Any) -> Any:
"""把旧版扁平 yaml 字段迁移进嵌套结构,让旧 config.yaml 仍可加载。"""
if not isinstance(data, dict):
return data
data = dict(data) # 不修改调用方的 dict
for flat_key, (sub_section, sub_field) in _FLAT_KEY_MIGRATION.items():
if flat_key not in data:
continue
value = data.pop(flat_key)
section = data.get(sub_section)
if not isinstance(section, dict):
section = dict(section) if section else {}
# 嵌套结构里已显式给出的值优先,不被扁平值覆盖
section.setdefault(sub_field, value)
data[sub_section] = section
return data
# --- 兼容属性:让 legacy updater/queue/router/client/prompt 不用改 ---
# P3 期 background extraction 重写后,这些 @property 连同调用方一起移除。
@property
def debounce_seconds(self) -> int:
"""[兼容] 等价于 ``background_extraction.debounce_seconds``。"""
return self.background_extraction.debounce_seconds
@property
def model_name(self) -> str | None:
"""[兼容] 等价于 ``background_extraction.model_name``。"""
return self.background_extraction.model_name
@property
def max_facts(self) -> int:
"""[兼容] 等价于 ``background_extraction.max_facts``。"""
return self.background_extraction.max_facts
@property
def fact_confidence_threshold(self) -> float:
"""[兼容] 等价于 ``background_extraction.fact_confidence_threshold``。"""
return self.background_extraction.fact_confidence_threshold
@property
def injection_enabled(self) -> bool:
"""[兼容] 等价于 ``injection.enabled``。"""
return self.injection.enabled
@property
def max_injection_tokens(self) -> int:
"""[兼容] 等价于 ``injection.max_tokens``。"""
return self.injection.max_tokens
# ===========================================================================
# per-user 运行时覆盖
# ===========================================================================
# 用户可在运行时覆盖的字段白名单 (点号路径)。
# 不在此表中的字段必须改 config.yaml 并重启 (例如 provider / api_key / 字符配额)。
USER_WRITABLE_FIELDS: frozenset[str] = frozenset(
{
"injection.enabled",
"injection.max_tokens",
"injection.include_builtin",
"injection.include_hindsight",
"hindsight.memory_mode",
"hindsight.recall_budget",
"hindsight.bank_mission",
"background_extraction.enabled",
}
)
def _filter_to_whitelist(overrides: dict[str, Any]) -> dict[str, Any]:
"""把覆盖 dict 过滤为只含白名单字段的嵌套 dict。
输入既支持嵌套形式 ``{"injection": {"enabled": false}}``,也支持
点号扁平形式 ``{"injection.enabled": false}``。非白名单字段被丢弃并 log。
"""
result: dict[str, Any] = {}
def _set_nested(path: str, value: Any) -> None:
if path not in USER_WRITABLE_FIELDS:
logger.warning("忽略非白名单的记忆配置覆盖字段:%s", path)
return
section, field = path.split(".", 1)
result.setdefault(section, {})[field] = value
for key, value in overrides.items():
if isinstance(value, dict):
for sub_key, sub_value in value.items():
_set_nested(f"{key}.{sub_key}", sub_value)
else:
_set_nested(key, value)
return result
def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
"""递归合并两个 dict,``override`` 优先。返回新 dict,不修改入参。"""
merged = copy.deepcopy(base)
for key, value in override.items():
if isinstance(value, dict) and isinstance(merged.get(key), dict):
merged[key] = _deep_merge(merged[key], value)
else:
merged[key] = value
return merged
def _load_json_file(path: Any) -> dict[str, Any]:
"""读取一个 JSON 覆盖文件;不存在或解析失败时返回空 dict。"""
try:
from pathlib import Path
p = Path(path)
if not p.exists():
return {}
data = json.loads(p.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except (OSError, json.JSONDecodeError) as exc:
logger.warning("读取记忆配置覆盖文件失败 %s: %s", path, exc)
return {}
# ===========================================================================
# 全局单例 + 加载入口
# ===========================================================================
_memory_config: MemoryConfig = MemoryConfig()
def get_memory_config() -> MemoryConfig:
"""返回 config.yaml 级别的记忆配置 (不含 per-user 覆盖)。"""
return _memory_config
def set_memory_config(config: MemoryConfig) -> None:
"""设置 config.yaml 级别的记忆配置 (测试与运行时重载用)。"""
global _memory_config
_memory_config = config
def load_memory_config_from_dict(config_dict: dict) -> None:
"""从 dict 加载记忆配置 (支持旧扁平 yaml 与新嵌套 yaml)。"""
global _memory_config
_memory_config = MemoryConfig(**config_dict)
def get_effective_memory_config(user_id: str) -> MemoryConfig:
"""按三层覆盖优先级,返回某用户的生效记忆配置。
优先级 (高 -> 低):per-user 覆盖 > 全局覆盖 > config.yaml > 默认值。
第 1、2 层只应用 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
覆盖文件读取失败不会抛异常 —— 退化为只用 config.yaml 级别配置。
"""
base = _memory_config.model_dump()
try:
from deerflow.config.paths import get_paths
paths = get_paths()
global_overrides = _filter_to_whitelist(
_load_json_file(paths.memory_global_overrides_file)
)
user_overrides = _filter_to_whitelist(
_load_json_file(paths.user_memory_config_file(user_id))
)
except Exception as exc: # noqa: BLE001 - 配置覆盖失败必须降级,不能阻断对话
logger.warning("加载 per-user 记忆配置覆盖失败,退回 config.yaml: %s", exc)
return _memory_config
merged = _deep_merge(_deep_merge(base, global_overrides), user_overrides)
try:
return MemoryConfig(**merged)
except Exception as exc: # noqa: BLE001 - 覆盖产生非法配置时降级
logger.warning("合并后的记忆配置非法,退回 config.yaml: %s", exc)
return _memory_config
def save_user_memory_override(user_id: str, patch: dict[str, Any]) -> dict[str, Any]:
"""把白名单字段写入某用户的 per-user 覆盖文件,返回写入后的完整覆盖 dict。
``patch`` 支持嵌套或点号扁平形式;非白名单字段被忽略。
"""
from deerflow.config.paths import get_paths
path = get_paths().user_memory_config_file(user_id)
current = _load_json_file(path)
filtered = _filter_to_whitelist(patch)
merged = _deep_merge(current, filtered)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
return merged
def delete_user_memory_override(user_id: str, field: str | None = None) -> dict[str, Any]:
"""撤销某用户的 per-user 覆盖。
``field`` 为点号路径 (如 ``injection.enabled``) 时只删该字段;
为 ``None`` 时清空整个 per-user 覆盖文件。返回剩余的覆盖 dict。
"""
from deerflow.config.paths import get_paths
path = get_paths().user_memory_config_file(user_id)
if field is None:
if path.exists():
path.unlink()
return {}
current = _load_json_file(path)
if "." in field:
section, sub_field = field.split(".", 1)
if isinstance(current.get(section), dict):
current[section].pop(sub_field, None)
if not current[section]:
current.pop(section, None)
else:
current.pop(field, None)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8")
return current
__all__ = [
"BuiltinMemoryConfig",
"HindsightMemoryConfig",
"BackgroundExtractionConfig",
"InjectionConfig",
"SecurityConfig",
"MemoryConfig",
"USER_WRITABLE_FIELDS",
"get_memory_config",
"set_memory_config",
"load_memory_config_from_dict",
"get_effective_memory_config",
"save_user_memory_override",
"delete_user_memory_override",
]
"""Memory V2 - 记忆子系统配置。
配置结构 (详见 docs/MEMORY_V2_DESIGN_ZH.md §9)
==============================================
``MemoryConfig`` 采用嵌套结构::
memory:
enabled: true
provider: hindsight # 外部 provider 选哪个 (builtin 始终启用)
builtin: {...} # 本地 USER.md / MEMORY.md
hindsight: {...} # Hindsight HTTP 客户端
injection: {...} # 系统提示词注入
security: {...} # 内容安全扫描
三层配置覆盖
============
``get_effective_memory_config(user_id)`` 按优先级合并 (高 -> 低):
1. per-user 覆盖 ``{base_dir}/users/{user_id}/memory_config.json``
2. 全局覆盖 ``{base_dir}/memory_user_overrides.json``
3. config.yaml ``memory:`` 段 (即 ``get_memory_config()``)
4. Pydantic 默认值
第 1、2 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
"""
from __future__ import annotations
import copy
import json
import logging
from typing import Any
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
# ===========================================================================
# 嵌套子配置
# ===========================================================================
class BuiltinMemoryConfig(BaseModel):
"""本地文件 provider (USER.md / MEMORY.md) 配置。"""
enabled: bool = Field(default=True, description="是否启用 builtin 本地记忆 provider")
memory_char_limit: int = Field(
default=2200, ge=200, le=20000, description="MEMORY.md 字符配额"
)
user_char_limit: int = Field(
default=1375, ge=200, le=20000, description="USER.md 字符配额"
)
deduplicate_on_load: bool = Field(
default=True, description="加载时是否对条目去重"
)
class HindsightMemoryConfig(BaseModel):
"""Hindsight 外部 provider 配置。"""
mode: str = Field(
default="local_external",
description="连接模式:cloud / local_embedded / local_external",
)
api_url: str = Field(default="http://localhost:8765", description="Hindsight API 地址")
api_key: str = Field(default="", description="API key,建议用 $HINDSIGHT_API_KEY 环境变量")
bank_id_template: str = Field(
default="deerflow-user-{user_id}",
description="全局 user bank 命名模板,占位符 {user_id}/{agent_id}/{thread_id}",
)
work_bank_id_template: str = Field(
default="deerflow-work-{user_id}-{agent_id}",
description="agent 私有工作 bank 命名模板",
)
memory_mode: str = Field(
default="context",
description="工具暴露策略:context(仅自动) / tools(仅工具) / hybrid(两者)",
)
bank_mission: str = Field(default="", description="reflect 推理的身份/框架描述")
bank_retain_mission: str = Field(default="", description="引导抽取内容的 mission")
recall_budget: str = Field(default="mid", description="召回深度:low / mid / high")
recall_max_tokens: int = Field(default=4096, ge=256, le=32768, description="召回结果 token 上限")
recall_max_input_chars: int = Field(default=800, ge=64, le=8000, description="召回 query 长度上限")
auto_recall: bool = Field(default=True, description="每轮对话前是否自动召回")
auto_retain: bool = Field(default=True, description="是否自动写入对话")
retain_async: bool = Field(default=True, description="retain 是否在 Hindsight 服务端异步处理")
retain_every_n_turns: int = Field(default=1, ge=1, le=50, description="每 N 轮 retain 一次")
retain_user_prefix: str = Field(default="User", description="自动 retain 时用户发言前缀")
retain_assistant_prefix: str = Field(default="Assistant", description="自动 retain 时助手发言前缀")
class InjectionConfig(BaseModel):
"""系统提示词记忆注入配置。"""
enabled: bool = Field(default=True, description="是否往系统提示词注入记忆")
max_tokens: int = Field(default=2000, ge=100, le=8000, description="注入 token 上限")
include_builtin: bool = Field(default=True, description="是否注入 builtin 本地记忆")
include_hindsight: bool = Field(default=True, description="是否注入 Hindsight 召回结果")
context_tag: str = Field(
default="memory-context", description="包裹外部召回内容的标签名"
)
class SecurityConfig(BaseModel):
"""记忆内容安全配置。"""
scan_content: bool = Field(default=True, description="是否扫描注入/外渗模式")
block_invisible_unicode: bool = Field(default=True, description="是否阻断不可见 unicode")
# ===========================================================================
# 顶层配置
# ===========================================================================
class MemoryConfig(BaseModel):
"""记忆子系统的全局配置 (config.yaml 的 ``memory:`` 段)。"""
enabled: bool = Field(default=True, description="记忆子系统总开关")
provider: str = Field(
default="builtin",
description="启用的外部 provider 名:builtin(无外部) 或 hindsight",
)
builtin: BuiltinMemoryConfig = Field(default_factory=BuiltinMemoryConfig)
hindsight: HindsightMemoryConfig = Field(default_factory=HindsightMemoryConfig)
injection: InjectionConfig = Field(default_factory=InjectionConfig)
security: SecurityConfig = Field(default_factory=SecurityConfig)
# ===========================================================================
# per-user 运行时覆盖
# ===========================================================================
# 用户可在运行时覆盖的字段白名单 (点号路径)。
# 不在此表中的字段必须改 config.yaml 并重启 (例如 provider / api_key / 字符配额)。
USER_WRITABLE_FIELDS: frozenset[str] = frozenset(
{
"injection.enabled",
"injection.max_tokens",
"injection.include_builtin",
"injection.include_hindsight",
"hindsight.memory_mode",
"hindsight.recall_budget",
"hindsight.bank_mission",
}
)
def _filter_to_whitelist(overrides: dict[str, Any]) -> dict[str, Any]:
"""把覆盖 dict 过滤为只含白名单字段的嵌套 dict。
输入既支持嵌套形式 ``{"injection": {"enabled": false}}``,也支持
点号扁平形式 ``{"injection.enabled": false}``。非白名单字段被丢弃并 log。
"""
result: dict[str, Any] = {}
def _set_nested(path: str, value: Any) -> None:
if path not in USER_WRITABLE_FIELDS:
logger.warning("忽略非白名单的记忆配置覆盖字段:%s", path)
return
section, field = path.split(".", 1)
result.setdefault(section, {})[field] = value
for key, value in overrides.items():
if isinstance(value, dict):
for sub_key, sub_value in value.items():
_set_nested(f"{key}.{sub_key}", sub_value)
else:
_set_nested(key, value)
return result
def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
"""递归合并两个 dict,``override`` 优先。返回新 dict,不修改入参。"""
merged = copy.deepcopy(base)
for key, value in override.items():
if isinstance(value, dict) and isinstance(merged.get(key), dict):
merged[key] = _deep_merge(merged[key], value)
else:
merged[key] = value
return merged
def _load_json_file(path: Any) -> dict[str, Any]:
"""读取一个 JSON 覆盖文件;不存在或解析失败时返回空 dict。"""
try:
from pathlib import Path
p = Path(path)
if not p.exists():
return {}
data = json.loads(p.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except (OSError, json.JSONDecodeError) as exc:
logger.warning("读取记忆配置覆盖文件失败 %s: %s", path, exc)
return {}
# ===========================================================================
# 全局单例 + 加载入口
# ===========================================================================
_memory_config: MemoryConfig = MemoryConfig()
def get_memory_config() -> MemoryConfig:
"""返回 config.yaml 级别的记忆配置 (不含 per-user 覆盖)。"""
return _memory_config
def set_memory_config(config: MemoryConfig) -> None:
"""设置 config.yaml 级别的记忆配置 (测试与运行时重载用)。"""
global _memory_config
_memory_config = config
def load_memory_config_from_dict(config_dict: dict) -> None:
"""从 dict 加载记忆配置。"""
global _memory_config
_memory_config = MemoryConfig(**config_dict)
def get_effective_memory_config(user_id: str) -> MemoryConfig:
"""按三层覆盖优先级,返回某用户的生效记忆配置。
优先级 (高 -> 低):per-user 覆盖 > 全局覆盖 > config.yaml > 默认值。
第 1、2 层只应用 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
覆盖文件读取失败不会抛异常 —— 退化为只用 config.yaml 级别配置。
"""
base = _memory_config.model_dump()
try:
from deerflow.config.paths import get_paths
paths = get_paths()
global_overrides = _filter_to_whitelist(
_load_json_file(paths.memory_global_overrides_file)
)
user_overrides = _filter_to_whitelist(
_load_json_file(paths.user_memory_config_file(user_id))
)
except Exception as exc: # noqa: BLE001 - 配置覆盖失败必须降级,不能阻断对话
logger.warning("加载 per-user 记忆配置覆盖失败,退回 config.yaml: %s", exc)
return _memory_config
merged = _deep_merge(_deep_merge(base, global_overrides), user_overrides)
try:
return MemoryConfig(**merged)
except Exception as exc: # noqa: BLE001 - 覆盖产生非法配置时降级
logger.warning("合并后的记忆配置非法,退回 config.yaml: %s", exc)
return _memory_config
def save_user_memory_override(user_id: str, patch: dict[str, Any]) -> dict[str, Any]:
"""把白名单字段写入某用户的 per-user 覆盖文件,返回写入后的完整覆盖 dict。
``patch`` 支持嵌套或点号扁平形式;非白名单字段被忽略。
"""
from deerflow.config.paths import get_paths
path = get_paths().user_memory_config_file(user_id)
current = _load_json_file(path)
filtered = _filter_to_whitelist(patch)
merged = _deep_merge(current, filtered)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
return merged
def delete_user_memory_override(user_id: str, field: str | None = None) -> dict[str, Any]:
"""撤销某用户的 per-user 覆盖。
``field`` 为点号路径 (如 ``injection.enabled``) 时只删该字段;
为 ``None`` 时清空整个 per-user 覆盖文件。返回剩余的覆盖 dict。
"""
from deerflow.config.paths import get_paths
path = get_paths().user_memory_config_file(user_id)
if field is None:
if path.exists():
path.unlink()
return {}
current = _load_json_file(path)
if "." in field:
section, sub_field = field.split(".", 1)
if isinstance(current.get(section), dict):
current[section].pop(sub_field, None)
if not current[section]:
current.pop(section, None)
else:
current.pop(field, None)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8")
return current
__all__ = [
"BuiltinMemoryConfig",
"HindsightMemoryConfig",
"InjectionConfig",
"SecurityConfig",
"MemoryConfig",
"USER_WRITABLE_FIELDS",
"get_memory_config",
"set_memory_config",
"load_memory_config_from_dict",
"get_effective_memory_config",
"save_user_memory_override",
"delete_user_memory_override",
]