Merge branch 'xuhd' into 'master'

feat: 记忆配置改为数据库存储，用户只可配置召回模式 See merge request knowledge-base-bj/deerflow-server!35
2026-05-21 09:34:48 +00:00
commit a514b2fe79
@@ -442,6 +442,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
        # Run Alembic migrations before anything else touches the schema.
        await _run_db_migrations()

+        # 预热 recall_budget cache（从 DB 批量加载,使 get_effective_memory_config 同步可读）
+        try:
+            from deerflow.config.memory_config import load_recall_budget_cache
+
+            _mc_store = getattr(app.state, "memory_config_store", None)
+            if _mc_store is not None:
+                _recall_budgets = await _mc_store.load_all()
+                load_recall_budget_cache(_recall_budgets)
+                logger.info("Memory config cache warmed: %d user override(s)", len(_recall_budgets))
+        except Exception:
+            logger.exception("Memory config cache warm-up failed (non-fatal)")
+
        # Ensure admin user exists (auto-create on first boot)
        # Must run AFTER langgraph_runtime so app.state.store is available for thread migration
        await _ensure_admin_user(app)
@@ -144,6 +144,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:

        app.state.article_type_store = make_article_type_store(sf)

+        from deerflow.persistence.memory_config import make_user_memory_config_store
+
+        app.state.memory_config_store = make_user_memory_config_store(sf)
+
        # Run event store (has its own factory with config-driven backend selection)
        run_events_config = getattr(config, "run_events", None)
        app.state.run_event_store = make_run_event_store(run_events_config)
@@ -267,6 +271,14 @@ def get_article_type_store(request: Request):
    return getattr(request.app.state, "article_type_store", None)


+def get_memory_config_store(request: Request):
+    """Return the per-user memory config store."""
+    val = getattr(request.app.state, "memory_config_store", None)
+    if val is None:
+        raise HTTPException(status_code=503, detail="Memory config store not available")
+    return val
+
+
 def get_run_context(request: Request) -> RunContext:
    """Build a :class:`RunContext` from ``app.state`` singletons.

@@ -21,15 +21,16 @@ from __future__ import annotations

 from typing import Any

-from fastapi import APIRouter, HTTPException, Query
+from fastapi import APIRouter, HTTPException, Query, Request
 from pydantic import BaseModel, Field

+from app.gateway.deps import get_memory_config_store
 from deerflow.config.agents_config import AGENT_NAME_PATTERN
 from deerflow.config.memory_config import (
    USER_WRITABLE_FIELDS,
-    delete_user_memory_override,
+    delete_user_memory_override_db,
    get_effective_memory_config,
-    save_user_memory_override,
+    save_user_memory_override_db,
 )
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
@@ -295,14 +296,17 @@ async def get_memory_v2_config() -> MemoryV2ConfigResponse:
    "/memory/v2/config",
    response_model=MemoryV2ConfigResponse,
    summary="Patch Memory V2 Config",
-    description="把白名单字段写入当前用户的 per-user 配置覆盖。非白名单字段被忽略。",
+    description="把 hindsight.recall_budget 写入当前用户的 per-user 配置覆盖 (DB 存储)。",
 )
-async def patch_memory_v2_config(request: MemoryConfigPatchRequest) -> MemoryV2ConfigResponse:
-    """更新当前用户的 per-user 记忆配置覆盖。"""
+async def patch_memory_v2_config(request: MemoryConfigPatchRequest, req: Request) -> MemoryV2ConfigResponse:
+    """更新当前用户的 per-user 记忆配置覆盖 (仅 hindsight.recall_budget)。"""
    user_id = get_effective_user_id()
+    store = get_memory_config_store(req)
    try:
-        save_user_memory_override(user_id, request.overrides)
-    except OSError as exc:
+        await save_user_memory_override_db(user_id, request.overrides, store)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except Exception as exc:  # noqa: BLE001
        raise HTTPException(status_code=500, detail="Failed to save memory config override.") from exc

    config = get_effective_memory_config(user_id)
@@ -317,18 +321,20 @@ async def patch_memory_v2_config(request: MemoryConfigPatchRequest) -> MemoryV2C
    response_model=MemoryV2ConfigResponse,
    summary="Delete Memory V2 Config Override",
    description=(
-        "撤销当前用户的 per-user 配置覆盖。传 ?field=injection.enabled 只撤销该字段;"
-        "不传 field 则清空整个 per-user 覆盖。"
+        "撤销当前用户的 per-user 配置覆盖。传 ?field=hindsight.recall_budget 只撤销该字段;"
+        "不传 field 则清空全部覆盖。"
    ),
 )
 async def delete_memory_v2_config(
+    req: Request,
    field: str | None = Query(None, description="要撤销的点号路径字段;省略则清空全部覆盖"),
 ) -> MemoryV2ConfigResponse:
    """撤销当前用户的 per-user 记忆配置覆盖。"""
    user_id = get_effective_user_id()
+    store = get_memory_config_store(req)
    try:
-        delete_user_memory_override(user_id, field)
-    except OSError as exc:
+        await delete_user_memory_override_db(user_id, field, store)
+    except Exception as exc:  # noqa: BLE001
        raise HTTPException(status_code=500, detail="Failed to delete memory config override.") from exc

    config = get_effective_memory_config(user_id)
@@ -12,21 +12,21 @@
      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. per-user 覆盖   数据库 ``user_memory_configs`` 表 (经进程内 cache 读取)
+     (有 per-user 记录时自动强制 enabled=True)
+  2. config.yaml     ``memory:`` 段 (即 ``get_memory_config()``)
+  3. Pydantic 默认值

-第 1、2 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
+第 1 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
+cache 由 Gateway lifespan 启动时预热,写入时同步更新,保持同步读取路径。
 """

 from __future__ import annotations

 import copy
-import json
 import logging
 from typing import Any

@@ -129,31 +129,26 @@ class MemoryConfig(BaseModel):


 # ===========================================================================
-# per-user 运行时覆盖
+# per-user 运行时覆盖 — 白名单
 # ===========================================================================

-# 用户可在运行时覆盖的字段白名单 (点号路径)。
-# 不在此表中的字段必须改 config.yaml 并重启 (例如 provider / api_key / 字符配额)。
+# 用户唯一可覆盖的字段:召回深度。
+# injection.* / builtin.* / enabled 始终由 config.yaml 控制,用户不可关闭。
 USER_WRITABLE_FIELDS: frozenset[str] = frozenset(
    {
-        # injection.enabled and injection.include_builtin are always ON;
-        # removing them from the whitelist makes _filter_to_whitelist drop
-        # any per-user override for these fields, so stored "false" values
-        # in existing memory_config.json files are silently ignored.
-        "injection.max_tokens",
-        "injection.include_hindsight",
-        # hindsight.memory_mode is fixed to "hybrid"; not user-writable.
-        # hindsight.bank_mission is fixed; not user-writable.
        "hindsight.recall_budget",
    }
 )

+_VALID_RECALL_BUDGETS: frozenset[str] = frozenset({"low", "mid", "high"})
+

 def _filter_to_whitelist(overrides: dict[str, Any]) -> dict[str, Any]:
    """把覆盖 dict 过滤为只含白名单字段的嵌套 dict。

-    输入既支持嵌套形式 ``{"injection": {"enabled": false}}``,也支持
-    点号扁平形式 ``{"injection.enabled": false}``。非白名单字段被丢弃并 log。
+    输入支持嵌套形式 ``{"hindsight": {"recall_budget": "high"}}`` 或
+    点号扁平形式 ``{"hindsight.recall_budget": "high"}``。
+    非白名单字段被丢弃并 log。
    """
    result: dict[str, Any] = {}

@@ -184,21 +179,6 @@ def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any
    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 {}
-
-
 # ===========================================================================
 # 全局单例 + 加载入口
 # ===========================================================================
@@ -223,82 +203,88 @@ def load_memory_config_from_dict(config_dict: dict) -> None:
    _memory_config = MemoryConfig(**config_dict)


+# ===========================================================================
+# per-user recall_budget 进程内 cache
+#
+# Gateway lifespan 启动时调用 load_recall_budget_cache() 从 DB 批量加载。
+# API 写入时同步更新 cache,使 get_effective_memory_config() 保持同步读取。
+# ===========================================================================
+
+_recall_budget_cache: dict[str, str] = {}
+
+
+def load_recall_budget_cache(data: dict[str, str]) -> None:
+    """用 DB 批量加载结果替换整个 cache (启动时调用一次)。"""
+    global _recall_budget_cache
+    _recall_budget_cache = dict(data)
+    logger.info("recall_budget cache loaded: %d user(s)", len(_recall_budget_cache))
+
+
+def set_recall_budget_cache(user_id: str, value: str) -> None:
+    """写入单个用户的 recall_budget cache。"""
+    _recall_budget_cache[user_id] = value
+
+
+def delete_recall_budget_cache(user_id: str) -> None:
+    """从 cache 中移除某用户的 recall_budget。"""
+    _recall_budget_cache.pop(user_id, None)
+
+
+def get_recall_budget_from_cache(user_id: str) -> str | None:
+    """返回 cache 中某用户的 recall_budget,未设置时返回 None。"""
+    return _recall_budget_cache.get(user_id)
+
+
 def get_effective_memory_config(user_id: str) -> MemoryConfig:
-    """按三层覆盖优先级,返回某用户的生效记忆配置。
+    """返回某用户的生效记忆配置 (同步,从进程内 cache 读取 per-user 覆盖)。

-    优先级 (高 -> 低):per-user 覆盖 > 全局覆盖 > config.yaml > 默认值。
-    第 1、2 层只应用 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。
-
-    覆盖文件读取失败不会抛异常 —— 退化为只用 config.yaml 级别配置。
+    优先级 (高 -> 低):per-user DB 覆盖 (经 cache) > config.yaml > 默认值。
+    有 per-user 记录时自动强制 enabled=True。
    """
    base = _memory_config.model_dump()
-
+    recall_budget = _recall_budget_cache.get(user_id)
+    if recall_budget is not None:
+        base.setdefault("hindsight", {})["recall_budget"] = recall_budget
+        base["enabled"] = True
    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)
+        return MemoryConfig(**base)
    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。
+# ===========================================================================
+# Gateway 调用的异步 DB 对接函数
+# ===========================================================================
+
+
+async def save_user_memory_override_db(user_id: str, patch: dict[str, Any], store: Any) -> dict[str, Any]:
+    """把白名单字段写入 DB store 并更新 cache,返回生效配置 dict。

    ``patch`` 支持嵌套或点号扁平形式;非白名单字段被忽略。
+    ``store`` 为 ``UserMemoryConfigStore`` 实例 (避免循环 import,用 Any)。
    """
-    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
+    recall_budget = (filtered.get("hindsight") or {}).get("recall_budget")
+    if recall_budget is not None:
+        if recall_budget not in _VALID_RECALL_BUDGETS:
+            raise ValueError(f"recall_budget 必须为 {sorted(_VALID_RECALL_BUDGETS)} 之一,收到:{recall_budget!r}")
+        await store.set_recall_budget(user_id, recall_budget)
+        set_recall_budget_cache(user_id, recall_budget)
+    return get_effective_memory_config(user_id).model_dump()


-def delete_user_memory_override(user_id: str, field: str | None = None) -> dict[str, Any]:
-    """撤销某用户的 per-user 覆盖。
+async def delete_user_memory_override_db(user_id: str, field: str | None, store: Any) -> dict[str, Any]:
+    """撤销某用户的 per-user 覆盖并更新 cache,返回生效配置 dict。

-    ``field`` 为点号路径 (如 ``injection.enabled``) 时只删该字段;
-    为 ``None`` 时清空整个 per-user 覆盖文件。返回剩余的覆盖 dict。
+    ``field=None`` 清空全部;``field="hindsight.recall_budget"`` 只清该字段。
    """
-    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)
+    if field is None or field == "hindsight.recall_budget":
+        await store.delete(user_id)
+        delete_recall_budget_cache(user_id)
    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
+        logger.warning("忽略删除非白名单的记忆配置覆盖字段: %s", field)
+    return get_effective_memory_config(user_id).model_dump()


 __all__ = [
@@ -312,6 +298,10 @@ __all__ = [
    "set_memory_config",
    "load_memory_config_from_dict",
    "get_effective_memory_config",
-    "save_user_memory_override",
-    "delete_user_memory_override",
+    "load_recall_budget_cache",
+    "set_recall_budget_cache",
+    "delete_recall_budget_cache",
+    "get_recall_budget_from_cache",
+    "save_user_memory_override_db",
+    "delete_user_memory_override_db",
 ]
@@ -0,0 +1,23 @@
+"""Per-user memory configuration override persistence."""
+
+from __future__ import annotations
+
+from deerflow.persistence.memory_config.base import UserMemoryConfigStore
+from deerflow.persistence.memory_config.memory import MemoryUserMemoryConfigStore
+from deerflow.persistence.memory_config.model import UserMemoryConfigRow
+from deerflow.persistence.memory_config.sql import SqlUserMemoryConfigRepository
+
+
+def make_user_memory_config_store(session_factory) -> UserMemoryConfigStore:
+    if session_factory is None:
+        return MemoryUserMemoryConfigStore()
+    return SqlUserMemoryConfigRepository(session_factory)
+
+
+__all__ = [
+    "UserMemoryConfigStore",
+    "UserMemoryConfigRow",
+    "MemoryUserMemoryConfigStore",
+    "SqlUserMemoryConfigRepository",
+    "make_user_memory_config_store",
+]
@@ -0,0 +1,32 @@
+"""Abstract persistence API for per-user memory config overrides."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class UserMemoryConfigStore(ABC):
+
+    @abstractmethod
+    async def get_recall_budget(self, user_id: str) -> str | None:
+        """Return the user's recall_budget override, or None if not set."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def set_recall_budget(self, user_id: str, value: str) -> None:
+        """Upsert the user's recall_budget override."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def delete(self, user_id: str) -> None:
+        """Remove all per-user overrides for the given user."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def load_all(self) -> dict[str, str]:
+        """Bulk-load all user recall_budget overrides.
+
+        Returns {user_id: recall_budget} for all rows where recall_budget IS NOT NULL.
+        Called once at startup to warm the in-memory cache.
+        """
+        raise NotImplementedError
@@ -0,0 +1,23 @@
+"""In-memory user memory config store (used when database.backend=memory)."""
+
+from __future__ import annotations
+
+from deerflow.persistence.memory_config.base import UserMemoryConfigStore
+
+
+class MemoryUserMemoryConfigStore(UserMemoryConfigStore):
+
+    def __init__(self) -> None:
+        self._data: dict[str, str] = {}
+
+    async def get_recall_budget(self, user_id: str) -> str | None:
+        return self._data.get(user_id)
+
+    async def set_recall_budget(self, user_id: str, value: str) -> None:
+        self._data[user_id] = value
+
+    async def delete(self, user_id: str) -> None:
+        self._data.pop(user_id, None)
+
+    async def load_all(self) -> dict[str, str]:
+        return dict(self._data)
@@ -0,0 +1,29 @@
+"""ORM model for per-user memory configuration overrides."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
+
+
+class UserMemoryConfigRow(Base):
+    __tablename__ = "user_memory_configs"
+
+    # One row per user; upserted on write.
+    user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+
+    # The only user-writable field. NULL means "use global default from config.yaml".
+    # Allowed values: "low" | "mid" | "high".
+    recall_budget: Mapped[str | None] = mapped_column(String(16), nullable=True)
+
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )
@@ -0,0 +1,50 @@
+"""SQLAlchemy-backed user memory config repository."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.memory_config.base import UserMemoryConfigStore
+from deerflow.persistence.memory_config.model import UserMemoryConfigRow
+
+
+class SqlUserMemoryConfigRepository(UserMemoryConfigStore):
+
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def get_recall_budget(self, user_id: str) -> str | None:
+        async with self._sf() as session:
+            row = await session.get(UserMemoryConfigRow, user_id)
+            return row.recall_budget if row is not None else None
+
+    async def set_recall_budget(self, user_id: str, value: str) -> None:
+        async with self._sf() as session:
+            row = await session.get(UserMemoryConfigRow, user_id)
+            if row is None:
+                row = UserMemoryConfigRow(
+                    user_id=user_id,
+                    recall_budget=value,
+                    updated_at=datetime.now(UTC),
+                )
+                session.add(row)
+            else:
+                row.recall_budget = value
+                row.updated_at = datetime.now(UTC)
+            await session.commit()
+
+    async def delete(self, user_id: str) -> None:
+        async with self._sf() as session:
+            row = await session.get(UserMemoryConfigRow, user_id)
+            if row is not None:
+                await session.delete(row)
+                await session.commit()
+
+    async def load_all(self) -> dict[str, str]:
+        stmt = select(UserMemoryConfigRow).where(UserMemoryConfigRow.recall_budget.is_not(None))
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return {row.user_id: row.recall_budget for row in result.scalars() if row.recall_budget is not None}
@@ -0,0 +1,44 @@
+"""Create user_memory_configs table.
+
+Replaces the per-user file-based memory config override
+({base_dir}/users/{user_id}/memory_config.json) with a DB row.
+Stores only hindsight.recall_budget — the single user-writable field
+after the memory config simplification.
+
+Revision ID: 20260521_01
+Revises: 20260519_02
+Create Date: 2026-05-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260521_01"
+down_revision = "20260519_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if _has_table("user_memory_configs"):
+        return
+    op.create_table(
+        "user_memory_configs",
+        sa.Column("user_id", sa.String(64), primary_key=True),
+        sa.Column("recall_budget", sa.String(16), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    if _has_table("user_memory_configs"):
+        op.drop_table("user_memory_configs")
@@ -16,6 +16,7 @@ there is no matching entity directory.
 """

 from deerflow.persistence.agents.model import AgentRow
+from deerflow.persistence.memory_config.model import UserMemoryConfigRow
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
 from deerflow.persistence.models.run_event import RunEventRow
@@ -55,5 +56,6 @@ __all__ = [
    "ThreadMetaRow",
    "ThreadShareRow",
    "ToolCallMetricRow",
+    "UserMemoryConfigRow",
    "UserRow",
 ]