xuhdandClaude Opus 4.7
39def1c20b
调度健壮性:
- 技能进化总开关 skill_evolution.enabled 真正生效;CuratorConfig
与 CuratorSubConfig 默认 enabled 改为 false(显式 opt-in)
- 新增 per-user 运行锁(.curator_state 的 running/started_at),
防定时器与手动触发并发重复运行;6h stale 检测防进程残留死锁
- maybe_run_curator 改用 ThreadPoolExecutor 限并发(上限 5),
防首次部署所有用户同时到期的惊群
- 首次运行持续失败时按 6h 退避重试,不再每个 tick 烧 LLM 配额
- should_run_now 时间解析容忍 naive 时间串;损坏的 last_run_at
触发重跑而非永久死锁
- SkillStore 新增 list_published_names 批量接口,curator 一轮
只查一次 published,取代逐技能查库
LLM 评审超时:
- _run_llm_review 给 agent.invoke 套整体超时
(curator.review_timeout_seconds,默认 600s),超时中止本次评审
并在 curator 日志输出,curator 正常收尾、释放运行锁
- copy_context 把 user 上下文传入工作线程
模型可配置:
- 新增 curator.model_name;curator 评审与技能安全审核共用该模型,
未配置则回退主模型
- 移除独立的 moderation_model_name 配置
- 前端进化面板新增模型下拉选择器
手动触发体验:
- /api/curator/run 派发后台线程前先预检,返回 {triggered, reason},
不再无论开关状态都提示"已开始"
- 前端:总开关/curator 开关未开时禁用"立即执行"按钮,
被预检拦下时按原因(开关未开 / 已有任务在跑)给出准确提示
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1193 lines
51 KiB
Python
1193 lines
51 KiB
Python
"""Skill curator — automatic lifecycle transitions and LLM consolidation pass.
|
|
|
|
Per-user since the multitenancy upgrade: every public function takes a
|
|
``user_id`` argument and the on-disk state lives under
|
|
``{base_dir}/users/{user_id}/.curator_state`` (alongside the per-user
|
|
memory.json and threads/ dirs).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import concurrent.futures
|
|
import contextvars
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import tempfile
|
|
import threading
|
|
from collections.abc import Coroutine
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any, TypeVar
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_CURATOR_STATE_FILENAME = ".curator_state"
|
|
|
|
# 周期调度一轮里同时执行的 curator 数量上限。到期用户多时(尤其首次部署所有用户一起
|
|
# 到期)用线程池排队消费,避免一次性起 N 个线程齐调 LLM 打爆 provider 速率限制。
|
|
_CURATOR_MAX_CONCURRENCY = 5
|
|
|
|
_T = TypeVar("_T")
|
|
|
|
|
|
def _run_async_blocking(coro: Coroutine[Any, Any, _T]) -> _T:
|
|
"""Run an async coroutine to completion from sync curator code.
|
|
|
|
Curator is invoked from a background thread, but ``agent.invoke()`` inside
|
|
``_run_llm_review`` (LangChain 1.x ``create_agent``) spins up its own
|
|
asyncio loop which can leave the calling thread in an "already has a
|
|
running loop" state. In that case raw ``asyncio.run(coro)`` raises
|
|
``RuntimeError`` *after* the coroutine object has been materialised but
|
|
before anyone awaits it — producing the ``coroutine was never awaited``
|
|
RuntimeWarning that pollutes our logs.
|
|
|
|
This helper detects a running loop in the caller's thread and offloads
|
|
to a dedicated worker thread; otherwise it falls back to ``asyncio.run``.
|
|
Either way it guarantees the coroutine is closed on the failure path,
|
|
so no RuntimeWarning ever escapes.
|
|
"""
|
|
try:
|
|
asyncio.get_running_loop()
|
|
in_loop = True
|
|
except RuntimeError:
|
|
in_loop = False
|
|
|
|
if in_loop:
|
|
try:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
|
return pool.submit(asyncio.run, coro).result()
|
|
except BaseException:
|
|
# ``asyncio.run`` ran in the worker (it consumes the coroutine on
|
|
# success or surfaces the inner exception); but if the executor
|
|
# itself failed to submit, close to be safe.
|
|
try:
|
|
coro.close()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
try:
|
|
return asyncio.run(coro)
|
|
except BaseException:
|
|
try:
|
|
coro.close()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
|
|
def _get_skills_root() -> Path:
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
|
|
return Path(get_or_new_skill_storage().get_skills_root_path())
|
|
|
|
|
|
def _curator_state_path(user_id: str | None) -> Path:
|
|
"""Return the state file path. Per-user when user_id is known; falls back
|
|
to the global ``skills/.curator_state`` for legacy callers."""
|
|
if not user_id:
|
|
return _get_skills_root() / _CURATOR_STATE_FILENAME
|
|
from deerflow.config.paths import get_paths
|
|
|
|
return get_paths().user_dir(user_id) / _CURATOR_STATE_FILENAME
|
|
|
|
|
|
def _read_curator_state(user_id: str | None) -> dict:
|
|
path = _curator_state_path(user_id)
|
|
if not path.exists():
|
|
return {"run_count": 0, "paused": False}
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {"run_count": 0, "paused": False}
|
|
|
|
|
|
def _write_curator_state(state: dict, user_id: str | None) -> None:
|
|
path = _curator_state_path(user_id)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
|
|
try:
|
|
with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
|
|
json.dump(state, f, indent=2, default=str)
|
|
os.replace(tmp_path, path)
|
|
except Exception:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except Exception:
|
|
pass
|
|
raise
|
|
except Exception:
|
|
logger.debug("Failed to write curator state", exc_info=True)
|
|
|
|
|
|
def _parse_state_time(value: str | None) -> datetime | None:
|
|
"""解析 ``.curator_state`` 里的 ISO 时间字符串,返回 aware datetime(UTC)。
|
|
|
|
naive(无时区)字符串按 UTC 补全 —— 与 ``apply_automatic_transitions`` 的处理
|
|
一致,避免 naive 时间串与 ``datetime.now(UTC)`` 直接比较时抛异常。无法解析时
|
|
返回 None,由调用方决定降级行为。
|
|
"""
|
|
if not value:
|
|
return None
|
|
try:
|
|
dt = datetime.fromisoformat(value)
|
|
except Exception:
|
|
return None
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=UTC)
|
|
return dt
|
|
|
|
|
|
# Curator 运行锁:防止同一用户的 curator 并发执行(定时器 vs 手动触发、连点手动触发)。
|
|
# 锁状态写进 .curator_state 的 running/started_at 字段,跨 worker 可见(手动请求可能
|
|
# 落到非调度 worker)。started_at 超过下面的阈值视为进程残留,允许接管以免永久死锁。
|
|
_CURATOR_STALE_LOCK_HOURS = 6
|
|
# 上一轮"尝试"过但未成功(last_run_at 始终为空)时的重试间隔 —— 比正常 interval_hours
|
|
# 短得多,但也不至于每个调度 tick(1h)都重试、每次都烧一份 LLM 配额。
|
|
_CURATOR_RETRY_HOURS = 6
|
|
# 进程内串行化运行锁的"读-改-写",保证同进程内的抢锁是原子的(跨进程的极小竞态窗口
|
|
# 可接受 —— curator 触发频率极低)。
|
|
_run_lock_guard = threading.Lock()
|
|
|
|
|
|
def _run_lock_active(state: dict, now: datetime) -> bool:
|
|
"""判断 ``.curator_state`` 里的运行锁当前是否有效(running 且未过期)。
|
|
|
|
``started_at`` 缺失或不可解析、或距今已超过 ``_CURATOR_STALE_LOCK_HOURS`` 时,
|
|
视为进程残留留下的死锁,返回 False 以便后续触发能接管。
|
|
"""
|
|
if not state.get("running"):
|
|
return False
|
|
started_dt = _parse_state_time(state.get("started_at"))
|
|
if started_dt is None:
|
|
return False
|
|
return now < started_dt + timedelta(hours=_CURATOR_STALE_LOCK_HOURS)
|
|
|
|
|
|
def _try_acquire_run_lock(user_id: str, now: datetime) -> bool:
|
|
"""尝试为 ``user_id`` 抢占 curator 运行锁。
|
|
|
|
抢到锁(已把 running=True / started_at 写回)返回 True;已有未过期的运行时返回
|
|
False。读-改-写在 ``_run_lock_guard`` 下进行,保证同进程内原子。
|
|
"""
|
|
with _run_lock_guard:
|
|
state = _read_curator_state(user_id)
|
|
if _run_lock_active(state, now):
|
|
return False
|
|
state["running"] = True
|
|
state["started_at"] = now.isoformat()
|
|
_write_curator_state(state, user_id)
|
|
return True
|
|
|
|
|
|
def _release_run_lock(user_id: str) -> None:
|
|
"""释放 curator 运行锁并记录本次尝试时间。无论运行成功或异常都应调用。
|
|
|
|
除了把 running 置回 False,还写入 ``last_attempt_at`` —— 用于失败退避:首次运行
|
|
若一直崩溃,``last_run_at`` 永远为空,``should_run_now`` 改用
|
|
``last_attempt_at + _CURATOR_RETRY_HOURS`` 判定,避免每个 tick 都重试。
|
|
成功跑完后 ``last_run_at`` 已被写入,``should_run_now`` 不再看 ``last_attempt_at``。
|
|
"""
|
|
with _run_lock_guard:
|
|
state = _read_curator_state(user_id)
|
|
state["running"] = False
|
|
state["last_attempt_at"] = datetime.now(UTC).isoformat()
|
|
_write_curator_state(state, user_id)
|
|
|
|
|
|
def _get_curator_config(user_id: str | None = None):
|
|
"""Return the curator config object for ``user_id``.
|
|
|
|
Reads ``config.yaml.skill_evolution.curator`` as the baseline and merges
|
|
the per-user override file (``users/{user_id}/skill_evolution.json``) on
|
|
top. With ``user_id=None`` falls back to the global default.
|
|
"""
|
|
try:
|
|
from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user
|
|
|
|
skill_evo = load_skill_evolution_config_for_user(user_id)
|
|
return getattr(skill_evo, "curator", None)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _skill_evolution_enabled(user_id: str | None = None) -> bool:
|
|
"""返回技能进化总开关 ``skill_evolution.enabled``(已合并 per-user 覆盖)。
|
|
|
|
这是整套技能进化功能的总闸:关闭时 curator 完全不动作。任何读取异常都按
|
|
"关闭"处理(fail-safe)—— 宁可不跑,也不在配置不可读时偷偷运行。
|
|
"""
|
|
try:
|
|
from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user
|
|
|
|
skill_evo = load_skill_evolution_config_for_user(user_id)
|
|
return bool(getattr(skill_evo, "enabled", False))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _list_user_owned_skill_names(user_id: str) -> set[str] | None:
|
|
"""Return skill names owned by ``user_id`` (used to filter transitions).
|
|
|
|
Returns None if the SkillStore is unreachable — curator then degrades to
|
|
"process every on-disk agent-created skill" (legacy behaviour).
|
|
"""
|
|
try:
|
|
from deerflow.persistence.engine import get_session_factory
|
|
from deerflow.persistence.skills import make_skill_store
|
|
|
|
factory = get_session_factory()
|
|
if factory is None:
|
|
return None
|
|
store = make_skill_store(factory)
|
|
|
|
async def _run() -> list[dict]:
|
|
return await store.list_visible(user_id)
|
|
|
|
records = _run_async_blocking(_run())
|
|
return {r["name"] for r in records if r.get("owner_user_id") == user_id}
|
|
except Exception:
|
|
logger.debug("_list_user_owned_skill_names failed", exc_info=True)
|
|
return None
|
|
|
|
|
|
def _collect_referenced_skill_names() -> set[str]:
|
|
"""Scan every agent's config.yaml and return the union of their ``skills``
|
|
fields.
|
|
|
|
Used as the "do not touch" set during curator transitions and LLM
|
|
candidate selection — any skill currently bound to *any* agent
|
|
(including the owner's own) is exempt from automatic stale/archive
|
|
and from being suggested for merging.
|
|
|
|
Returns an empty set on filesystem errors (best-effort, non-fatal).
|
|
"""
|
|
import yaml
|
|
|
|
from deerflow.config.paths import get_paths
|
|
|
|
referenced: set[str] = set()
|
|
try:
|
|
agents_dir = get_paths().agents_dir
|
|
except Exception:
|
|
return referenced
|
|
if not agents_dir.exists():
|
|
return referenced
|
|
|
|
for entry in agents_dir.iterdir():
|
|
if not entry.is_dir():
|
|
continue
|
|
config_file = entry / "config.yaml"
|
|
if not config_file.exists():
|
|
continue
|
|
try:
|
|
with config_file.open("r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
except Exception:
|
|
logger.debug("_collect_referenced_skill_names: failed to parse %s", config_file, exc_info=True)
|
|
continue
|
|
skills = data.get("skills")
|
|
if isinstance(skills, list):
|
|
for s in skills:
|
|
if isinstance(s, str):
|
|
referenced.add(s)
|
|
return referenced
|
|
|
|
|
|
def _is_skill_published(skill_name: str) -> bool:
|
|
"""Return True if ``skill_name`` is marked ``published=True`` in SkillStore.
|
|
|
|
Best-effort — returns False on any error so a transient DB hiccup never
|
|
causes a published skill to be wrongly archived (the caller treats
|
|
"published=True" as exempt, so False = "fail open to processing", which
|
|
only puts a skill *back* into curator scope when DB is unavailable; in
|
|
that degraded mode curator legacy behaviour applies).
|
|
"""
|
|
try:
|
|
from deerflow.persistence.engine import get_session_factory
|
|
from deerflow.persistence.skills import make_skill_store
|
|
|
|
factory = get_session_factory()
|
|
if factory is None:
|
|
return False
|
|
store = make_skill_store(factory)
|
|
|
|
async def _run() -> dict | None:
|
|
return await store.get_any(skill_name)
|
|
|
|
record = _run_async_blocking(_run())
|
|
return bool(record and record.get("published", False))
|
|
except Exception:
|
|
logger.debug("_is_skill_published(%s) failed", skill_name, exc_info=True)
|
|
return False
|
|
|
|
|
|
def _list_published_skill_names() -> set[str]:
|
|
"""一次性返回所有 ``published=True`` 的技能名集合。
|
|
|
|
取代"每个技能调一次 ``_is_skill_published``"的逐条查库 —— curator 一轮里会遍历
|
|
所有候选技能,逐条查会产生 ≈2N 次独立 event loop + DB 往返。出错时返回空集
|
|
(fail-open:与 ``_is_skill_published`` 出错返回 False 语义一致,即 DB 不可用时
|
|
所有技能都视为未发布,回退到 curator 的 legacy 行为)。
|
|
"""
|
|
try:
|
|
from deerflow.persistence.engine import get_session_factory
|
|
from deerflow.persistence.skills import make_skill_store
|
|
|
|
factory = get_session_factory()
|
|
if factory is None:
|
|
return set()
|
|
store = make_skill_store(factory)
|
|
|
|
async def _run() -> set[str]:
|
|
return await store.list_published_names()
|
|
|
|
return _run_async_blocking(_run())
|
|
except Exception:
|
|
logger.debug("_list_published_skill_names failed", exc_info=True)
|
|
return set()
|
|
|
|
|
|
def _eligible_for_curation(
|
|
name: str,
|
|
entry: dict,
|
|
*,
|
|
owned_set: set[str] | None,
|
|
referenced_set: set[str],
|
|
published_set: set[str],
|
|
) -> bool:
|
|
"""Three-condition gate for whether curator may touch a skill.
|
|
|
|
Curator only acts on skills that ALL of:
|
|
1. source == "agent" (caller is expected to have already filtered, but
|
|
we double-check here for safety)
|
|
2. NOT referenced by any agent's skills config
|
|
3. NOT published
|
|
|
|
``owned_set`` is the set of skill names owned by the running user. When
|
|
None, the legacy "process every agent-created skill" fallback applies
|
|
for ownership only (the other two conditions still hold).
|
|
|
|
``published_set`` is the set of published skill names, pre-fetched once by
|
|
the caller via ``_list_published_skill_names()`` — avoids a per-skill DB
|
|
round-trip.
|
|
"""
|
|
if entry.get("source") != "agent":
|
|
return False
|
|
if owned_set is not None and name not in owned_set:
|
|
return False
|
|
if name in referenced_set:
|
|
return False
|
|
if name in published_set:
|
|
return False
|
|
return True
|
|
|
|
|
|
def should_run_now(user_id: str) -> bool:
|
|
"""Return True if the curator interval has elapsed for ``user_id`` and it
|
|
is not paused."""
|
|
# 总开关:技能进化整体关闭时,curator 一律不跑。
|
|
if not _skill_evolution_enabled(user_id):
|
|
return False
|
|
curator_cfg = _get_curator_config(user_id)
|
|
if curator_cfg is None or not getattr(curator_cfg, "enabled", False):
|
|
return False
|
|
state = _read_curator_state(user_id)
|
|
if state.get("paused"):
|
|
return False
|
|
# 运行锁:已有 curator 在跑(且锁未过期)时不再触发新的一轮。
|
|
if _run_lock_active(state, datetime.now(UTC)):
|
|
return False
|
|
last_run = state.get("last_run_at")
|
|
if not last_run:
|
|
# 从未成功跑完过。若此前已尝试过(last_attempt_at 存在),按较短的重试间隔
|
|
# 退避 —— 避免首次运行持续失败时每个 tick(1h)都重试、每次烧一份 LLM 配额。
|
|
attempt_dt = _parse_state_time(state.get("last_attempt_at"))
|
|
if attempt_dt is None:
|
|
return True
|
|
return datetime.now(UTC) >= attempt_dt + timedelta(hours=_CURATOR_RETRY_HOURS)
|
|
interval_hours = getattr(curator_cfg, "interval_hours", 168)
|
|
last_dt = _parse_state_time(last_run)
|
|
if last_dt is None:
|
|
# 时间串损坏 —— 视同"从未跑过",允许重跑并重写正确时间戳,而不是像旧逻辑那样
|
|
# except 返回 False 导致该用户永久不再触发。
|
|
return True
|
|
return datetime.now(UTC) >= last_dt + timedelta(hours=interval_hours)
|
|
|
|
|
|
def manual_trigger_blocked_reason(user_id: str) -> str | None:
|
|
"""手动触发 curator 的预检。返回 None 表示可以跑;否则返回阻塞原因代码:
|
|
``evolution_disabled`` / ``curator_disabled`` / ``already_running``。
|
|
|
|
供 ``POST /api/curator/run`` 在派发后台线程**前**同步判定 —— 后台线程
|
|
(``_curator_thread_target``)本身也会再查一遍,这里只是为了让前端拿到准确反馈,
|
|
而不是无论开关状态如何都提示"已开始"。
|
|
"""
|
|
if not _skill_evolution_enabled(user_id):
|
|
return "evolution_disabled"
|
|
curator_cfg = _get_curator_config(user_id)
|
|
if curator_cfg is None or not getattr(curator_cfg, "enabled", False):
|
|
return "curator_disabled"
|
|
state = _read_curator_state(user_id)
|
|
if _run_lock_active(state, datetime.now(UTC)):
|
|
return "already_running"
|
|
return None
|
|
|
|
|
|
def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> dict:
|
|
"""Pure age-based state transitions — no LLM involved.
|
|
|
|
Only processes agent-created skills that are ALL of:
|
|
- owned by ``user_id``
|
|
- not referenced by any agent's ``skills`` config
|
|
- not currently published
|
|
|
|
Skips pinned skills. Returns counters: {marked_stale, archived, checked,
|
|
skipped_referenced, skipped_published}.
|
|
"""
|
|
from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries, latest_activity_at, set_state
|
|
|
|
if now is None:
|
|
now = datetime.now(UTC)
|
|
|
|
curator_cfg = _get_curator_config(user_id)
|
|
stale_days = getattr(curator_cfg, "stale_after_days", 30) if curator_cfg else 30
|
|
archive_days = getattr(curator_cfg, "archive_after_days", 90) if curator_cfg else 90
|
|
|
|
owned = _list_user_owned_skill_names(user_id)
|
|
referenced = _collect_referenced_skill_names()
|
|
published = _list_published_skill_names() # 一次性查出,避免逐技能查库
|
|
raw_entries = agent_created_entries()
|
|
|
|
skipped_referenced = 0
|
|
skipped_published = 0
|
|
eligible: dict[str, dict] = {}
|
|
for name, entry in raw_entries.items():
|
|
if owned is not None and name not in owned:
|
|
continue
|
|
if name in referenced:
|
|
skipped_referenced += 1
|
|
continue
|
|
if name in published:
|
|
skipped_published += 1
|
|
continue
|
|
eligible[name] = entry
|
|
|
|
counters = {
|
|
"marked_stale": 0,
|
|
"archived": 0,
|
|
"checked": len(eligible),
|
|
"skipped_referenced": skipped_referenced,
|
|
"skipped_published": skipped_published,
|
|
}
|
|
|
|
from deerflow.skills import curator_log
|
|
|
|
for name, entry in eligible.items():
|
|
if entry.get("pinned"):
|
|
continue
|
|
last_activity_str = latest_activity_at(name)
|
|
if not last_activity_str:
|
|
continue
|
|
try:
|
|
last_dt = datetime.fromisoformat(last_activity_str)
|
|
if last_dt.tzinfo is None:
|
|
last_dt = last_dt.replace(tzinfo=UTC)
|
|
except Exception:
|
|
continue
|
|
|
|
days_idle = (now - last_dt).days
|
|
current_state = entry.get("state", STATE_ACTIVE)
|
|
|
|
if days_idle >= archive_days and current_state != STATE_ARCHIVED:
|
|
set_state(name, STATE_ARCHIVED)
|
|
counters["archived"] += 1
|
|
curator_log.record(
|
|
"info",
|
|
f" 自动归档 [{name}]:已闲置 {days_idle} 天(阈值 {archive_days} 天)",
|
|
user_id=user_id,
|
|
)
|
|
elif days_idle >= stale_days and current_state == STATE_ACTIVE:
|
|
set_state(name, STATE_STALE)
|
|
counters["marked_stale"] += 1
|
|
curator_log.record(
|
|
"info",
|
|
f" 标记过时 [{name}]:已闲置 {days_idle} 天(阈值 {stale_days} 天)",
|
|
user_id=user_id,
|
|
)
|
|
|
|
return counters
|
|
|
|
|
|
def _render_candidate_list(user_id: str) -> str:
|
|
"""List of skill candidates handed to the LLM for merge/archive review.
|
|
|
|
Same three-condition gate as ``apply_automatic_transitions``: only
|
|
agent-created, not-referenced-by-any-agent, not-published skills owned
|
|
by ``user_id`` show up here. The LLM literally cannot see (and thus
|
|
cannot suggest action on) anything outside that set.
|
|
"""
|
|
from deerflow.skills.usage import STATE_ARCHIVED, all_entries
|
|
|
|
owned = _list_user_owned_skill_names(user_id)
|
|
referenced = _collect_referenced_skill_names()
|
|
published = _list_published_skill_names() # 一次性查出,避免逐技能查库
|
|
entries: dict[str, dict] = {}
|
|
for k, v in all_entries().items():
|
|
if not _eligible_for_curation(k, v, owned_set=owned, referenced_set=referenced, published_set=published):
|
|
continue
|
|
if v.get("state") == STATE_ARCHIVED:
|
|
continue
|
|
entries[k] = v
|
|
if not entries:
|
|
return "No agent-created active skills found."
|
|
lines = []
|
|
for name, entry in sorted(entries.items()):
|
|
state = entry.get("state", "active")
|
|
use = entry.get("use_count", 0)
|
|
view = entry.get("view_count", 0)
|
|
patch = entry.get("patch_count", 0)
|
|
last = entry.get("last_used_at") or entry.get("created_at") or "unknown"
|
|
lines.append(f"- {name} [state={state}, use={use}, view={view}, patches={patch}, last_active={last}]")
|
|
return "\n".join(lines)
|
|
|
|
|
|
_CURATOR_REVIEW_PROMPT = (
|
|
"You are running as DeerFlow's background skill CURATOR. This is an "
|
|
"UMBRELLA-BUILDING consolidation pass, not a passive audit and not a "
|
|
"duplicate-finder.\n\n"
|
|
"The goal of the skill collection is a LIBRARY OF CLASS-LEVEL "
|
|
"INSTRUCTIONS AND EXPERIENTIAL KNOWLEDGE. A collection of hundreds of "
|
|
"narrow skills where each one captures one session's specific bug is "
|
|
"a FAILURE of the library — not a feature. An agent searching skills "
|
|
"matches on descriptions, not on exact names; one broad umbrella "
|
|
"skill with labeled subsections beats five narrow siblings for "
|
|
"discoverability, not the other way around.\n\n"
|
|
"The right target shape is CLASS-LEVEL skills with rich SKILL.md "
|
|
"bodies + `references/`, `templates/`, and `scripts/` subfiles for "
|
|
"session-specific detail — not one-session-one-skill micro-entries.\n\n"
|
|
"Hard rules — do not violate:\n"
|
|
"1. DO NOT touch public (built-in) skills. The candidate list below "
|
|
"is already filtered to agent-created custom skills only.\n"
|
|
"2. DO NOT delete any skill. Archiving (a soft state flip via the "
|
|
"skill_archive tool — the file stays in custom/<name>/ and the user "
|
|
"can restore it from the Web UI) is the maximum action you may take. "
|
|
"Archives are recoverable; deletion is not. "
|
|
"When you call skill_archive(name, reason) you MUST pass a `reason` "
|
|
"string written **in Chinese**, in plain language a human operator "
|
|
"would read — NOT English jargon or terse debug notes. Examples of "
|
|
"acceptable reasons: '已合并到 code-formatter,该技能同时覆盖 Python 与 JS', "
|
|
"'与 foo-bar 内容重复,保留 foo-bar', '重构后过时,被 api-v2-search 替代', "
|
|
"'一次性 POC 实验,无复用价值'. Bad reasons (do NOT use): 'merged', "
|
|
"'duplicate', 'obsolete', 'consolidated'. The reason is shown in the "
|
|
"Web UI run-history for user audit, so write it as you would explain "
|
|
"it to the user themselves.\n"
|
|
"3. DO NOT touch skills shown as pinned=yes. Skip them entirely.\n"
|
|
"4. DO NOT use usage counters as a reason to skip consolidation. The "
|
|
"counters are new and often mostly zero. Judge overlap on CONTENT, "
|
|
"not on use_count. 'use=0' is not evidence a skill is valuable; it's "
|
|
"absence of evidence either way.\n"
|
|
"5. DO NOT reject consolidation on the grounds that 'each skill has "
|
|
"a distinct trigger'. Pairwise distinctness is the wrong bar. The "
|
|
"right bar is: 'would a human maintainer write this as N separate "
|
|
"skills, or as one skill with N labeled subsections?' When the "
|
|
"answer is the latter, merge.\n\n"
|
|
"How to work — not optional:\n"
|
|
"1. Scan the full candidate list. Identify PREFIX CLUSTERS (skills "
|
|
"sharing a first word or domain keyword). Examples you are likely "
|
|
"to find: search-*, code-*, data-*, api-*, file-*, analysis-*, "
|
|
"workflow-*, report-*, monitor-*, debug-*, python-*, security-*, etc. "
|
|
"Expect 10-25 clusters.\n"
|
|
"2. For each cluster with 2+ members, do NOT ask 'are these pairs "
|
|
"overlapping?' — ask 'what is the UMBRELLA CLASS these skills all "
|
|
"serve? Would a maintainer name that class and write one skill for "
|
|
"it?' If yes, pick (or create) the umbrella and absorb the siblings "
|
|
"into it.\n"
|
|
"3. Three ways to consolidate — use the right one per cluster:\n"
|
|
" a. MERGE INTO EXISTING UMBRELLA — one skill in the cluster is "
|
|
"already broad enough to be the umbrella. Patch it to add a labeled "
|
|
"section for each sibling's unique insight, then archive the "
|
|
"siblings.\n"
|
|
" b. CREATE A NEW UMBRELLA SKILL.md — no existing member is broad "
|
|
"enough. Create a new class-level skill whose SKILL.md covers the "
|
|
"shared workflow and has short labeled subsections. Archive the "
|
|
"now-absorbed narrow siblings.\n"
|
|
" c. DEMOTE TO REFERENCES/TEMPLATES/SCRIPTS — a sibling has "
|
|
"narrow-but-valuable session-specific content. Move it into the "
|
|
"umbrella's appropriate support directory:\n"
|
|
" • `references/<topic>.md` for session-specific detail OR "
|
|
"condensed knowledge banks (quoted research, API docs excerpts, "
|
|
"domain notes, provider quirks, reproduction recipes)\n"
|
|
" • `templates/<name>.<ext>` for starter files meant to be "
|
|
"copied and modified\n"
|
|
" • `scripts/<name>.<ext>` for statically re-runnable actions "
|
|
"(verification scripts, fixture generators, probes)\n"
|
|
" Then archive the old sibling.\n"
|
|
"4. Also flag skills whose NAME is too narrow (contains a specific "
|
|
"issue number, a feature codename, a specific error string, an "
|
|
"'audit' / 'diagnosis' / 'salvage' session artifact). These almost "
|
|
"always belong as a subsection or support file under a class-level "
|
|
"umbrella.\n"
|
|
"5. Iterate. After one consolidation round, scan the remaining set "
|
|
"and look for the NEXT umbrella opportunity. Don't stop after 3 "
|
|
"merges.\n\n"
|
|
"'keep' is a legitimate decision ONLY when the skill is already a "
|
|
"class-level umbrella and none of the proposed merges would improve "
|
|
"discoverability. 'This is narrow but distinct from its siblings' "
|
|
"is NOT a reason to keep — it's a reason to move it under an "
|
|
"umbrella as a subsection or support file.\n\n"
|
|
"Expected output: real umbrella-ification. Process every obvious "
|
|
"cluster. If you end the pass with fewer than 10 archives, you "
|
|
"stopped too early — go back and look at the clusters you left "
|
|
"alone.\n\n"
|
|
"When done, write a human summary AND a structured machine-readable "
|
|
"block so downstream tooling can distinguish consolidation from "
|
|
"pruning. Format EXACTLY:\n\n"
|
|
"## Structured summary (required)\n"
|
|
"```yaml\n"
|
|
"consolidations:\n"
|
|
" - from: <old-skill-name>\n"
|
|
" into: <umbrella-skill-name>\n"
|
|
" reason: <one short sentence — why merged, not just 'similar'>\n"
|
|
"prunings:\n"
|
|
" - name: <skill-name>\n"
|
|
" reason: <one short sentence — why archived with no merge target>\n"
|
|
"```\n\n"
|
|
"Every skill you archived via skill_archive MUST appear in exactly one "
|
|
"of the two lists. If you consolidated X into umbrella Y (patched Y, wrote "
|
|
"a references file to Y, or created Y with X's content absorbed), X "
|
|
"goes under `consolidations` with `into: Y`. If you archived X with "
|
|
"no absorption — truly stale, irrelevant, or obsolete — X goes under "
|
|
"`prunings`. Leave a list empty (`consolidations: []`) if none. Do "
|
|
"not omit the block. The block comes AFTER your human-readable "
|
|
"summary of clusters processed, patches made, and decisions left alone."
|
|
)
|
|
|
|
|
|
def _make_skill_archive_tool(user_id: str):
|
|
"""Curator-only tool: soft-archive a skill (sets state, files stay put)."""
|
|
from langchain.tools import tool as lc_tool
|
|
|
|
from deerflow.skills import curator_log
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
|
|
@lc_tool("skill_archive")
|
|
def _skill_archive(name: str, reason: str) -> str:
|
|
"""Soft-archive an agent-created skill (sets state=archived; files stay in custom/).
|
|
|
|
The user can recover the skill from the Web UI's archive drawer.
|
|
This is intentionally non-destructive — the on-disk SKILL.md and any
|
|
supporting files are preserved so nothing is lost.
|
|
|
|
Args:
|
|
name: The skill name to archive.
|
|
reason: 必填。用**中文**写一句话,清晰说明为什么归档这个技能 —
|
|
直接给读它的运维人员看,不要写英文术语 / 调试信息。例如:
|
|
- "已合并到 code-formatter,该技能同时覆盖 Python 与 JS"
|
|
- "重构后过时,被 api-v2-search 替代"
|
|
- "与现有 foo-bar 技能重复,内容完全一致"
|
|
- "一次性 POC 实验,无复用价值"
|
|
- "命名过窄(含具体 issue 号),已并入通用伞技能 X"
|
|
此原因会展示在策展日志和 Web UI 运行历史中,供用户审计。
|
|
"""
|
|
from deerflow.skills.usage import STATE_ARCHIVED, get_entry, set_state
|
|
|
|
# Defensive: tolerate empty/None reason but log a clear placeholder
|
|
# rather than crashing the curator run.
|
|
reason_text = (reason or "").strip() or "未提供原因"
|
|
|
|
storage = get_or_new_skill_storage()
|
|
if not storage.custom_skill_exists(name):
|
|
curator_log.record("warning", f"归档:未找到技能 [{name}],或它已被归档", user_id=user_id)
|
|
return f"Skill '{name}' not found or already archived."
|
|
entry = get_entry(name)
|
|
if entry.get("pinned"):
|
|
curator_log.record("info", f"归档:技能 [{name}] 已被钉选,跳过", user_id=user_id)
|
|
return f"Skill '{name}' is pinned — skipping."
|
|
if entry.get("state") == STATE_ARCHIVED:
|
|
curator_log.record("info", f"归档:技能 [{name}] 已是归档状态,跳过", user_id=user_id)
|
|
return f"Skill '{name}' is already archived."
|
|
|
|
# Safety net: even if the LLM hallucinated a name not in the
|
|
# candidate list, refuse to archive anything that is published or
|
|
# currently referenced by an agent.
|
|
if _is_skill_published(name):
|
|
curator_log.record(
|
|
"warning",
|
|
f"归档:已拒绝处理 [{name}],该技能已发布(策展不允许操作已发布的技能)",
|
|
user_id=user_id,
|
|
)
|
|
return f"Refused: skill '{name}' is published; manual unpublish is required first."
|
|
referenced = _collect_referenced_skill_names()
|
|
if name in referenced:
|
|
curator_log.record(
|
|
"warning",
|
|
f"归档:已拒绝处理 [{name}],该技能正被某个 Agent 引用",
|
|
user_id=user_id,
|
|
)
|
|
return f"Refused: skill '{name}' is still referenced by one or more agents."
|
|
|
|
# Soft archive: flip the lifecycle state. The skill directory stays in
|
|
# ``custom/<name>/`` untouched so the user can recover it from the
|
|
# Web UI's archive drawer (which lists skills by state, not by location).
|
|
try:
|
|
set_state(name, STATE_ARCHIVED)
|
|
except Exception as e:
|
|
curator_log.record("error", f"归档失败:技能 [{name}] — {e}", user_id=user_id)
|
|
return f"Archive failed: {e}"
|
|
curator_log.record(
|
|
"info",
|
|
f"已归档(软):[{name}] — 原因:{reason_text} — 文件保留在 custom/{name}/",
|
|
user_id=user_id,
|
|
)
|
|
return f"Archived '{name}' (reason: {reason_text}; files preserved in custom/{name}/)."
|
|
|
|
return _skill_archive
|
|
|
|
|
|
def _make_curator_skill_manage_tool():
|
|
"""curator 受限版 ``skill_manage``:在工具层屏蔽 delete 动作。
|
|
|
|
curator 做 umbrella 合并需要 create / patch / edit / write_file,但绝不能删整个
|
|
技能 —— prompt 硬规则也写明"归档(skill_archive)是允许的最大操作"。通用
|
|
skill_manage 的 delete 走 ``_assert_agent_may_delete``,**绕过了** skill_archive
|
|
对"已发布 / 被 agent 引用"技能的拒绝保护。只靠 prompt 文字约束不可靠,这里在
|
|
工具层直接拒掉 delete,curator 的 LLM 无论如何都删不掉技能。
|
|
"""
|
|
from langchain.tools import tool as lc_tool
|
|
|
|
from deerflow.tools.skill_manage_tool import _skill_manage_impl
|
|
|
|
@lc_tool("skill_manage")
|
|
def _curator_skill_manage(
|
|
action: str,
|
|
name: str,
|
|
content: str | None = None,
|
|
path: str | None = None,
|
|
find: str | None = None,
|
|
replace: str | None = None,
|
|
expected_count: int | None = None,
|
|
) -> str:
|
|
"""管理 skills/custom/ 下的自定义技能(curator 受限版)。
|
|
|
|
支持 action:create / patch / edit / write_file / remove_file / list / view。
|
|
**不支持 delete** —— curator 不得删除技能;如需下架请改用 skill_archive。
|
|
|
|
Args:
|
|
action: 操作类型(见上;delete 会被拒绝)。
|
|
name: 连字符命名的技能名。
|
|
content: create / edit / write_file 的新文件内容。
|
|
path: write_file / remove_file 的支持文件路径。
|
|
find: patch 要替换的原文本。
|
|
replace: patch 的替换文本。
|
|
expected_count: patch 的预期替换次数(可选)。
|
|
"""
|
|
if action == "delete":
|
|
return (
|
|
"Refused: curator 不允许删除技能。归档是允许的最大操作 —— "
|
|
"请改用 skill_archive(name, reason) 软下架(文件保留,用户可在 Web UI 恢复)。"
|
|
)
|
|
# curator 不绑定到具体会话线程,runtime 传 None(_get_thread_id 兼容 None)。
|
|
return _run_async_blocking(
|
|
_skill_manage_impl(
|
|
runtime=None,
|
|
action=action,
|
|
name=name,
|
|
content=content,
|
|
path=path,
|
|
find=find,
|
|
replace=replace,
|
|
expected_count=expected_count,
|
|
)
|
|
)
|
|
|
|
return _curator_skill_manage
|
|
|
|
|
|
def _run_llm_review(candidate_list: str, *, user_id: str) -> tuple[str, list[dict]]:
|
|
"""Run the curator consolidation pass as a React agent with skill tools bound.
|
|
|
|
返回 ``(llm_text, tool_actions)``:
|
|
- ``llm_text`` —— LLM 最终回复文字(它"自述"做了什么);
|
|
- ``tool_actions`` —— 从消息流提取的**真实**工具调用记录(它"实际"做了什么)。
|
|
"""
|
|
from deerflow.skills import curator_log
|
|
|
|
try:
|
|
from langchain.agents import create_agent
|
|
from langchain_core.messages import HumanMessage
|
|
|
|
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
|
|
from deerflow.models import create_chat_model
|
|
from deerflow.tools.builtins.skill_tools import skill_list_tool, skill_view_tool
|
|
|
|
curator_cfg = _get_curator_config(user_id)
|
|
# curator 评审用的模型:per-user 可配 curator.model_name(对应 config.yaml
|
|
# models[].name);为空则用主模型。
|
|
model_name = getattr(curator_cfg, "model_name", None) if curator_cfg else None
|
|
model = create_chat_model(name=model_name)
|
|
# 用 curator 受限版 skill_manage(屏蔽 delete),不用通用版 —— 见 _make_curator_skill_manage_tool。
|
|
tools = [skill_list_tool, skill_view_tool, _make_curator_skill_manage_tool(), _make_skill_archive_tool(user_id)]
|
|
# Reuse the lead-agent error middleware: when the LLM mis-calls a tool
|
|
# (wrong action / missing required args), the exception is converted
|
|
# into a ToolMessage(status="error") so the LLM can read the error and
|
|
# retry, instead of aborting the entire review pass.
|
|
# NOTE: ``create_react_agent`` (langgraph.prebuilt) does NOT accept a
|
|
# middleware kwarg — use LangChain 1.x's ``create_agent`` which does.
|
|
agent = create_agent(model=model, tools=tools, middleware=[ToolErrorHandlingMiddleware()])
|
|
|
|
timeout_seconds = getattr(curator_cfg, "review_timeout_seconds", 600) if curator_cfg else 600
|
|
curator_log.record(
|
|
"info",
|
|
f"LLM 评审:开始调用 React Agent(模型 {model_name or '主模型'},整体超时 {timeout_seconds}s)",
|
|
user_id=user_id,
|
|
)
|
|
prompt = _CURATOR_REVIEW_PROMPT + f"\n\nSkill candidates for this pass:\n{candidate_list}"
|
|
|
|
# 给整个 React Agent 评审套一个总超时:agent.invoke 内部可能多轮调 LLM,模型层的
|
|
# 单次 request_timeout 兜不住整体。超时后本 curator 线程能正常收尾、释放运行锁,
|
|
# 而不是无限期挂起。注意:Python 无法强杀线程 —— 超时后那个 agent.invoke 线程会
|
|
# 成为孤儿(最终被模型层 request_timeout 或自然返回所终结),但不再拖住 curator。
|
|
# copy_context():把 current_user 等 ContextVar 传进工作线程,否则工具调用会把
|
|
# user_id 误解析为 DEFAULT_USER_ID。
|
|
ctx = contextvars.copy_context()
|
|
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="curator-llm")
|
|
future = pool.submit(ctx.run, agent.invoke, {"messages": [HumanMessage(content=prompt)]})
|
|
try:
|
|
result = future.result(timeout=timeout_seconds)
|
|
except concurrent.futures.TimeoutError:
|
|
curator_log.record(
|
|
"warning",
|
|
f"LLM 评审超时(超过 {timeout_seconds}s),已中止本次评审",
|
|
user_id=user_id,
|
|
)
|
|
logger.warning("Curator[%s] LLM review timed out after %ss", user_id, timeout_seconds)
|
|
return "", []
|
|
finally:
|
|
# wait=False:超时时不阻塞等待挂死的 invoke 线程。
|
|
pool.shutdown(wait=False)
|
|
|
|
messages = result.get("messages", [])
|
|
tool_actions = _extract_tool_actions(messages)
|
|
for msg in reversed(messages):
|
|
if not getattr(msg, "tool_calls", None):
|
|
content = getattr(msg, "content", "")
|
|
if isinstance(content, str) and content.strip():
|
|
curator_log.record("info", "LLM 评审:Agent 已返回最终结论", user_id=user_id)
|
|
return content, tool_actions
|
|
if isinstance(content, list):
|
|
text = " ".join(
|
|
block.get("text", "") if isinstance(block, dict) else str(block)
|
|
for block in content
|
|
).strip()
|
|
curator_log.record("info", "LLM 评审:Agent 已返回最终结论", user_id=user_id)
|
|
return text, tool_actions
|
|
curator_log.record("warning", "LLM 评审:本次未产出最终回复", user_id=user_id)
|
|
return "", tool_actions
|
|
except Exception as e:
|
|
logger.warning("Curator LLM review failed: %s", e)
|
|
curator_log.record("warning", f"LLM 评审失败:{e}", user_id=user_id)
|
|
return "", []
|
|
|
|
|
|
def _parse_structured_summary(text: str) -> dict:
|
|
pattern = re.compile(r"```yaml\s*(.*?)```", re.DOTALL)
|
|
match = pattern.search(text)
|
|
if not match:
|
|
return {"consolidations": [], "prunings": []}
|
|
try:
|
|
import yaml
|
|
|
|
data = yaml.safe_load(match.group(1)) or {}
|
|
return {"consolidations": data.get("consolidations") or [], "prunings": data.get("prunings") or []}
|
|
except Exception:
|
|
return {"consolidations": [], "prunings": []}
|
|
|
|
|
|
def _extract_tool_actions(messages: list) -> list[dict]:
|
|
"""从 agent 消息流里提取**真实发生的**工具调用 + 结果。
|
|
|
|
与 _parse_structured_summary(解析 LLM 自述的 YAML)互补:这里记录的是 curator
|
|
实际调了哪些工具、入参是什么、成功还是失败 —— 用于审计"嘴上说的"与"实际做的"
|
|
是否一致。每条形如 ``{tool, args, status, result}``。
|
|
"""
|
|
# 先按 tool_call_id 收集每次调用的工具名与入参(来自 AIMessage.tool_calls)
|
|
calls: dict[str, dict] = {}
|
|
for msg in messages:
|
|
for tc in getattr(msg, "tool_calls", None) or []:
|
|
tc_id = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
|
|
if not tc_id:
|
|
continue
|
|
calls[tc_id] = {
|
|
"tool": tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None),
|
|
"args": (tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", None)) or {},
|
|
}
|
|
# 再按 ToolMessage(带 tool_call_id)收集真实结果
|
|
actions: list[dict] = []
|
|
for msg in messages:
|
|
tc_id = getattr(msg, "tool_call_id", None)
|
|
if not tc_id:
|
|
continue
|
|
call = calls.get(tc_id, {})
|
|
raw = getattr(msg, "content", "")
|
|
result = raw if isinstance(raw, str) else str(raw)
|
|
actions.append(
|
|
{
|
|
"tool": call.get("tool") or getattr(msg, "name", None),
|
|
"args": call.get("args", {}),
|
|
"status": getattr(msg, "status", None) or "success",
|
|
"result": result[:500],
|
|
}
|
|
)
|
|
return actions
|
|
|
|
|
|
def run_curator_review(user_id: str) -> dict:
|
|
"""Main curator run for ``user_id``: transitions → LLM consolidation → state.
|
|
|
|
Only skills owned by ``user_id`` are touched.
|
|
"""
|
|
from deerflow.skills import curator_log
|
|
|
|
curator_log.start_new_run(user_id=user_id)
|
|
|
|
now = datetime.now(UTC)
|
|
logger.info("Curator[%s]: starting run at %s", user_id, now.isoformat())
|
|
curator_log.record("info", f"本次策展开始,启动时间 {now.isoformat()}", user_id=user_id)
|
|
|
|
transition_counters = apply_automatic_transitions(user_id, now)
|
|
logger.info("Curator[%s]: transitions=%s", user_id, transition_counters)
|
|
curator_log.record(
|
|
"info",
|
|
(
|
|
f"按年龄自动流转完成:已检查 {transition_counters['checked']} 个,"
|
|
f"标记陈旧 {transition_counters['marked_stale']} 个,"
|
|
f"自动归档 {transition_counters['archived']} 个"
|
|
),
|
|
user_id=user_id,
|
|
)
|
|
|
|
candidate_list = _render_candidate_list(user_id)
|
|
candidate_line_count = 0 if not candidate_list.strip() else len([line for line in candidate_list.splitlines() if line.startswith("- ")])
|
|
curator_log.record(
|
|
"info", f"进入 LLM 评审的候选技能:共 {candidate_line_count} 个", user_id=user_id
|
|
)
|
|
|
|
if candidate_line_count == 0:
|
|
# 没有候选技能 —— 跳过 LLM 评审,省掉无意义的模型调用。
|
|
# 年龄流转(上面的 apply_automatic_transitions)已完成,且不依赖 LLM。
|
|
curator_log.record("info", "无候选技能,跳过 LLM 评审", user_id=user_id)
|
|
suggestions = {"consolidations": [], "prunings": []}
|
|
tool_actions: list[dict] = []
|
|
else:
|
|
llm_text, tool_actions = _run_llm_review(candidate_list, user_id=user_id)
|
|
suggestions = _parse_structured_summary(llm_text)
|
|
curator_log.record(
|
|
"info",
|
|
(
|
|
f"LLM 建议已解析:建议合并 {len(suggestions['consolidations'])} 项,"
|
|
f"建议删减 {len(suggestions['prunings'])} 项"
|
|
),
|
|
user_id=user_id,
|
|
)
|
|
# Surface per-item reasoning so operators can audit *why* each action
|
|
# was taken, not just the count.
|
|
for item in suggestions.get("consolidations", []) or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = item.get("from", "?")
|
|
into = item.get("into", "?")
|
|
reason = (item.get("reason") or "无说明").strip()
|
|
curator_log.record(
|
|
"info",
|
|
f" 合并建议 [{name}] → [{into}]:{reason}",
|
|
user_id=user_id,
|
|
)
|
|
for item in suggestions.get("prunings", []) or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = item.get("name", "?")
|
|
reason = (item.get("reason") or "无说明").strip()
|
|
curator_log.record(
|
|
"info",
|
|
f" 删减建议 [{name}]:{reason}",
|
|
user_id=user_id,
|
|
)
|
|
|
|
# 上面是 LLM"自述"的建议;下面记录它"实际"调用了哪些工具,供对照审计。
|
|
curator_log.record("info", f"LLM 实际调用工具 {len(tool_actions)} 次", user_id=user_id)
|
|
for act in tool_actions:
|
|
target = act.get("args", {}).get("name", "")
|
|
curator_log.record(
|
|
"info",
|
|
f" 实际调用 [{act.get('tool')}]{f' [{target}]' if target else ''} → {act.get('status')}",
|
|
user_id=user_id,
|
|
)
|
|
|
|
state = _read_curator_state(user_id)
|
|
state["last_run_at"] = now.isoformat()
|
|
state["run_count"] = state.get("run_count", 0) + 1
|
|
state["last_run_summary"] = (
|
|
f"自动流转:标记陈旧 {transition_counters['marked_stale']} 个、"
|
|
f"归档 {transition_counters['archived']} 个;"
|
|
f"LLM:建议合并 {len(suggestions['consolidations'])} 项、实际工具调用 {len(tool_actions)} 次"
|
|
)
|
|
state["last_consolidation_suggestions"] = suggestions # LLM 自述
|
|
state["last_actual_actions"] = tool_actions # 真实工具调用记录
|
|
_write_curator_state(state, user_id)
|
|
|
|
logger.info("Curator[%s]: run complete — %s", user_id, state["last_run_summary"])
|
|
curator_log.record("info", f"本次策展已完成 — {state['last_run_summary']}", user_id=user_id)
|
|
return {"transitions": transition_counters, "suggestions": suggestions}
|
|
|
|
|
|
class _CuratorUser:
|
|
"""Minimal ``CurrentUser`` protocol implementation for background runs.
|
|
|
|
Curator threads are spawned via ``threading.Thread`` (not an asyncio task),
|
|
so the ``deerflow.runtime.user_context`` ContextVar starts unset. Without
|
|
a stand-in, any tool the LLM review invokes (e.g. ``skill_manage(create)``)
|
|
would call ``get_effective_user_id()`` and silently fall back to the
|
|
``DEFAULT_USER_ID`` sentinel — causing newly-created skills to be owned
|
|
by ``"default"`` instead of the user the curator is running for.
|
|
"""
|
|
|
|
__slots__ = ("id",)
|
|
|
|
def __init__(self, user_id: str) -> None:
|
|
self.id = user_id
|
|
|
|
|
|
def _curator_thread_target(user_id: str) -> None:
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
from deerflow.skills import curator_log
|
|
|
|
# 总开关 + curator 子开关:任一关闭则直接跳过。手动触发 /api/curator/run 也经由
|
|
# 本函数,因此手动触发同样受总开关与 curator.enabled 约束。
|
|
if not _skill_evolution_enabled(user_id):
|
|
logger.info("Curator[%s]: 技能进化总开关关闭,跳过", user_id)
|
|
return
|
|
curator_cfg = _get_curator_config(user_id)
|
|
if curator_cfg is None or not getattr(curator_cfg, "enabled", False):
|
|
logger.info("Curator[%s]: curator 未启用,跳过", user_id)
|
|
return
|
|
|
|
# 运行锁:防止同一用户的 curator 并发(定时器 vs 手动触发、连点手动触发)。抢不到
|
|
# 锁说明已有一轮在跑,直接跳过本次触发。
|
|
if not _try_acquire_run_lock(user_id, datetime.now(UTC)):
|
|
logger.info("Curator[%s]: 已有 curator 正在运行,跳过本次触发", user_id)
|
|
curator_log.record("info", "已有 curator 正在运行,跳过本次触发", user_id=user_id)
|
|
return
|
|
|
|
# Bind the per-thread user context for the entire curator run so every
|
|
# downstream tool / repository call resolves the correct user_id.
|
|
token = set_current_user(_CuratorUser(user_id))
|
|
try:
|
|
run_curator_review(user_id)
|
|
except Exception as exc:
|
|
logger.exception("Curator[%s] background run failed", user_id)
|
|
curator_log.record("error", f"策展后台任务失败:{exc}", user_id=user_id)
|
|
finally:
|
|
reset_current_user(token)
|
|
# 无论成功或异常,必定释放运行锁。
|
|
_release_run_lock(user_id)
|
|
|
|
|
|
async def maybe_run_curator() -> None:
|
|
"""周期调度入口 —— 给每个 interval 已到期的用户起一个后台策展线程。
|
|
|
|
由 app 的 curator 调度器周期性调用(也用于手动触发)。无候选技能的用户,其
|
|
LLM 评审会在 run_curator_review 内被跳过 —— 只做不耗费 LLM 的年龄流转。
|
|
任何失败都不会阻断调用方。
|
|
"""
|
|
try:
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.engine import get_session_factory
|
|
from deerflow.persistence.user.model import UserRow
|
|
|
|
factory = get_session_factory()
|
|
if factory is None:
|
|
return
|
|
async with factory() as session:
|
|
result = await session.execute(select(UserRow.id))
|
|
user_ids = [str(row[0]) for row in result.all()]
|
|
except Exception:
|
|
logger.exception("Curator: failed to enumerate users at startup")
|
|
return
|
|
|
|
due_users: list[str] = []
|
|
for user_id in user_ids:
|
|
try:
|
|
if should_run_now(user_id):
|
|
due_users.append(user_id)
|
|
except Exception:
|
|
logger.exception("Curator: should_run_now check failed for user %s", user_id)
|
|
_dispatch_curator_runs(due_users)
|
|
|
|
|
|
def _dispatch_curator_runs(due_users: list[str]) -> None:
|
|
"""用线程池(并发上限 ``_CURATOR_MAX_CONCURRENCY``)派发到期用户的 curator 执行。
|
|
|
|
到期用户排队消费,而非一次性全部起线程齐调 LLM —— 防首次部署所有用户同时到期时
|
|
的惊群。``shutdown(wait=False)``:提交后立即返回,线程池在所有任务跑完后自行回收。
|
|
抽成独立函数便于对并发上限做单元测试。
|
|
"""
|
|
if not due_users:
|
|
return
|
|
|
|
logger.info(
|
|
"Curator: %d 个用户到期,以并发上限 %d 派发执行",
|
|
len(due_users),
|
|
_CURATOR_MAX_CONCURRENCY,
|
|
)
|
|
executor = concurrent.futures.ThreadPoolExecutor(
|
|
max_workers=_CURATOR_MAX_CONCURRENCY,
|
|
thread_name_prefix="skill-curator",
|
|
)
|
|
try:
|
|
for user_id in due_users:
|
|
executor.submit(_curator_thread_target, user_id)
|
|
except Exception:
|
|
logger.exception("Curator: failed to submit curator tasks")
|
|
finally:
|
|
executor.shutdown(wait=False)
|
|
|
|
|
|
def get_curator_status(user_id: str) -> dict:
|
|
"""Return curator state for ``user_id``."""
|
|
from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries
|
|
|
|
state = _read_curator_state(user_id)
|
|
owned = _list_user_owned_skill_names(user_id)
|
|
entries = agent_created_entries()
|
|
if owned is not None:
|
|
entries = {name: entry for name, entry in entries.items() if name in owned}
|
|
counts: dict[str, int] = {STATE_ACTIVE: 0, STATE_STALE: 0, STATE_ARCHIVED: 0}
|
|
for entry in entries.values():
|
|
s = entry.get("state", STATE_ACTIVE)
|
|
counts[s] = counts.get(s, 0) + 1
|
|
|
|
cfg = _get_curator_config(user_id)
|
|
return {
|
|
"enabled": cfg is not None and getattr(cfg, "enabled", True),
|
|
"run_count": state.get("run_count", 0),
|
|
"last_run_at": state.get("last_run_at"),
|
|
"last_run_summary": state.get("last_run_summary"),
|
|
"paused": state.get("paused", False),
|
|
"agent_created_skills": {
|
|
"total": len(entries),
|
|
"active": counts[STATE_ACTIVE],
|
|
"stale": counts[STATE_STALE],
|
|
"archived": counts[STATE_ARCHIVED],
|
|
},
|
|
# 上次运行的两份记录:LLM 自述的建议 + 实际发生的工具调用,供前端对照展示。
|
|
"last_consolidation_suggestions": state.get("last_consolidation_suggestions")
|
|
or {"consolidations": [], "prunings": []},
|
|
"last_actual_actions": state.get("last_actual_actions") or [],
|
|
}
|