Files
deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_evolution_runtime.py
T
xuhdandClaude Opus 4.7 4ef96c7930 feat(skills): 多租户阶段 3 — Curator + 配置 per-user
curator_log.py
- ring buffer 从单一 deque 改为 dict[user_id -> deque],每个用户
  自己的运行日志互不可见
- start_new_run / record / get_recent / clear 全部接 user_id
  关键字参数;未传时落入 __default__ 桶兼容旧调用
- 新增 clear_all() 用于跨用户清理(仅测试用)

curator.py
- 状态文件路径从 skills/.curator_state 改为
  .deer-flow/users/{user_id}/.curator_state(per-user 隔离)
- 所有公开 API 接受 user_id 参数:should_run_now /
  apply_automatic_transitions / _render_candidate_list /
  run_curator_review / _curator_thread_target /
  get_curator_status / _make_skill_archive_tool / _run_llm_review
- 新增 _list_user_owned_skill_names(user_id):通过 SkillStore
  查询当前用户拥有的 skill 名集合,用于过滤 transitions / LLM
  candidates,保证 user A 跑 curator 不会动 user B 的 skill;
  SkillStore 不可达时退化为"全部 agent-created skill"(兼容旧部署)
- _get_curator_config(user_id) 通过 load_skill_evolution_config_for_user
  叠加 per-user override,让阈值/间隔/moderation_model 也是 per-user
- maybe_run_curator 启动时枚举 UserRow,按用户检查 should_run_now
  并各起一个 daemon 线程跑

skill_evolution_runtime.py
- 全局 backend/skill_evolution.json 语义保留作为 fallback
- 新增 per-user API:resolve_user_path / load_user_overrides /
  save_user_overrides / get_user_overrides_mtime,文件落到
  .deer-flow/users/{user_id}/skill_evolution.json
- 新增 load_skill_evolution_config_for_user(user_id):读全局
  SkillEvolutionConfig 作为 baseline,浅合并 per-user JSON,
  返回 Pydantic 实例。验证失败时回落到 baseline

routers/skills.py
- /api/curator/status / run / config GET+PUT / logs 全部接
  _current_user_id 注入运行上下文
- update_curator_config 改为 per-user:保存到
  users/{user_id}/skill_evolution.json,不再 reset_app_config
- _build_curator_config_response 接受 SkillEvolutionConfig 实例
  (而不是 AppConfig)以适配 per-user 合并结果

旧 19 个测试继续通过;未引入新依赖。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 17:09:05 +08:00

217 lines
7.4 KiB
Python

"""Runtime overrides for `skill_evolution` config, stored in skill_evolution.json.
Two scopes exist:
* **Global** (back-compat): ``backend/skill_evolution.json`` deep-merged onto
``config.yaml.skill_evolution`` at AppConfig load time. Used as a fallback
for users who never set a personal override.
* **Per-user**: ``.deer-flow/users/{user_id}/skill_evolution.json``. Read by
the curator on each run and by the ``GET/PUT /api/curator/config`` route
with the active user id. Curator merges this on top of the global config
via :func:`load_curator_config_for_user`.
The file is optional in both scopes; absence (or a malformed payload) is
treated as "no overrides" so the system keeps the YAML-only behaviour.
"""
from __future__ import annotations
import json
import logging
import os
import tempfile
from pathlib import Path
from typing import Any
from deerflow.config.runtime_paths import existing_project_file
logger = logging.getLogger(__name__)
_FILENAME = "skill_evolution.json"
def resolve_path() -> Path:
"""Resolve the on-disk location of ``skill_evolution.json``.
Priority:
1. ``DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH`` environment variable
2. ``skill_evolution.json`` in the project root (current and parents)
3. Legacy ``backend/`` or repo-root sibling locations
4. Falls back to ``backend/skill_evolution.json`` (does not need to exist)
"""
override = os.getenv("DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH")
if override:
return Path(override)
project_file = existing_project_file((_FILENAME,))
if project_file is not None:
return project_file
backend_dir = Path(__file__).resolve().parents[4]
repo_root = backend_dir.parent
for candidate in (backend_dir / _FILENAME, repo_root / _FILENAME):
if candidate.exists():
return candidate
return backend_dir / _FILENAME
def load_overrides() -> dict[str, Any] | None:
"""Read overrides from disk.
Returns ``None`` when the file does not exist, is empty, contains
non-dict JSON, or fails to parse. Errors are logged at WARNING level
so production stays silent in the common "no overrides" case.
"""
path = resolve_path()
if not path.exists():
return None
try:
raw = path.read_text(encoding="utf-8").strip()
if not raw:
return None
data = json.loads(raw)
except OSError as exc:
logger.warning("Failed to read %s: %s", path, exc)
return None
except json.JSONDecodeError as exc:
logger.warning("Skill evolution overrides at %s are not valid JSON: %s", path, exc)
return None
if not isinstance(data, dict):
logger.warning("Skill evolution overrides at %s must be a JSON object, got %s", path, type(data).__name__)
return None
return data
def save_overrides(payload: dict[str, Any]) -> Path:
"""Atomically persist ``payload`` to ``skill_evolution.json``.
The caller is responsible for shape validation (e.g. via the
``SkillEvolutionConfig`` Pydantic model). Returns the path that was
written so callers can log it.
"""
if not isinstance(payload, dict):
raise TypeError("skill_evolution overrides payload must be a dict")
path = resolve_path()
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path_str = tempfile.mkstemp(prefix=".skill-evolution-", suffix=".tmp", dir=str(path.parent))
tmp_path = Path(tmp_path_str)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
os.replace(tmp_path, path)
except Exception:
tmp_path.unlink(missing_ok=True)
raise
logger.info("Wrote skill_evolution overrides to %s", path)
return path
def get_overrides_mtime() -> float | None:
"""Return the mtime of ``skill_evolution.json`` if it exists, else None.
Used by the AppConfig cache to decide whether to reload — when the
file appears, disappears, or is modified, the mtime tuple changes
and ``get_app_config()`` rebuilds.
"""
path = resolve_path()
try:
return path.stat().st_mtime
except OSError:
return None
# ---------------------------------------------------------------------------
# Per-user overrides
# ---------------------------------------------------------------------------
def resolve_user_path(user_id: str) -> Path:
"""Return ``{base_dir}/users/{user_id}/skill_evolution.json``."""
from deerflow.config.paths import get_paths
return get_paths().user_dir(user_id) / _FILENAME
def load_user_overrides(user_id: str) -> dict[str, Any] | None:
"""Read the per-user overrides file for ``user_id`` (None if missing/invalid)."""
path = resolve_user_path(user_id)
if not path.exists():
return None
try:
raw = path.read_text(encoding="utf-8").strip()
if not raw:
return None
data = json.loads(raw)
except OSError as exc:
logger.warning("Failed to read %s: %s", path, exc)
return None
except json.JSONDecodeError as exc:
logger.warning("User skill_evolution overrides at %s are not valid JSON: %s", path, exc)
return None
if not isinstance(data, dict):
logger.warning("User skill_evolution overrides at %s must be a JSON object", path)
return None
return data
def save_user_overrides(user_id: str, payload: dict[str, Any]) -> Path:
"""Atomically persist ``payload`` to the per-user overrides file."""
if not isinstance(payload, dict):
raise TypeError("skill_evolution overrides payload must be a dict")
path = resolve_user_path(user_id)
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path_str = tempfile.mkstemp(prefix=".skill-evolution-", suffix=".tmp", dir=str(path.parent))
tmp_path = Path(tmp_path_str)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
os.replace(tmp_path, path)
except Exception:
tmp_path.unlink(missing_ok=True)
raise
logger.info("Wrote per-user skill_evolution overrides to %s", path)
return path
def get_user_overrides_mtime(user_id: str) -> float | None:
path = resolve_user_path(user_id)
try:
return path.stat().st_mtime
except OSError:
return None
def load_skill_evolution_config_for_user(user_id: str | None):
"""Return a ``SkillEvolutionConfig`` instance with per-user overrides merged
on top of the global AppConfig defaults.
Used by the curator + the per-user ``GET/PUT /api/curator/config`` route.
When ``user_id`` is None or the per-user file is absent, returns the global
config unchanged.
"""
from deerflow.config import get_app_config
from deerflow.config.skill_evolution_config import SkillEvolutionConfig
base = get_app_config().skill_evolution
if not user_id:
return base
overrides = load_user_overrides(user_id)
if not overrides:
return base
merged = base.model_dump()
for key, value in overrides.items():
existing = merged.get(key)
if isinstance(existing, dict) and isinstance(value, dict):
merged[key] = {**existing, **value}
else:
merged[key] = value
try:
return SkillEvolutionConfig.model_validate(merged)
except Exception:
logger.warning("Per-user skill_evolution overrides for %s failed validation, ignoring", user_id, exc_info=True)
return base