feat(skills): 自我进化后端 — curator 配置可编辑 + 运行日志可读
后端骨架： - curator.py / usage.py / skill_tools.py：基于使用计数的生命周期管理 + LLM 合并/归档建议（同事已有原型，集中提交） - skill_evolution_config.py：CuratorConfig 嵌套模型 - app.py：启动时按间隔触发 curator 后台线程 - tools.py：开启自我进化时一并注册 skill_list / skill_view 工具 - skill_manage_tool.py：agent 创建技能时标记 source=agent 本次新增： - skill_evolution_runtime.py：skill_evolution.json 读写，原子写入 - app_config.py：把 overrides 深合并到 AppConfig.skill_evolution， 并把 overrides 的 mtime 纳入缓存失效条件，前端 PUT 后热生效 - curator_log.py：进程内 ring buffer (maxlen=200)，所有 record/get 操作内部兜异常，保证 logging 永远不会让 curator 跑挂 - curator.py 关键节点插桩：run_started / 自动转换计数 / 候选数 / LLM agent invoke / 解析结果 / archive 动作 / run complete - routers/skills.py：GET/PUT /api/curator/config（写到 skill_evolution.json + reset_app_config）+ GET /api/curator/logs 测试： - tests/test_skill_evolution_runtime.py：overrides 往返、损坏 JSON 容错、AppConfig 合并、ring buffer 线程安全、record 异常吞咽 杂项： - .gitignore：忽略 .deer-flow/* 与 skills/* 等运行时目录 - uv.lock：跟随依赖锁更新 - 清理已废弃的 agent runtime 数据 SOUL.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
xuhdandClaude Opus 4.7 committed
1 parent
230eec1baf
commit
9b6b94d919
14 files changed
+3981
-2698
No files matched your search
@@ -64,3 +64,10 @@ offline-backend-20260512/backend/.deer-flow/*
|
||||
offline-backend-20260512/start_backend.sh
|
||||
offline-backend-20260512/stop_backend.sh
|
||||
offline-backend-20260512/check_backend.sh
|
||||
|
||||
hermes-agent-main/
|
||||
|
||||
|
||||
skills/*
|
||||
|
||||
offline-backend-20260512/backend/.deer-flow/*
|
||||
@@ -228,6 +228,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
# Must run AFTER langgraph_runtime so app.state.store is available for thread migration
|
||||
await _ensure_admin_user(app)
|
||||
|
||||
# Kick off skill curator in background if the run interval has elapsed
|
||||
try:
|
||||
from deerflow.skills.curator import maybe_run_curator
|
||||
asyncio.create_task(maybe_run_curator())
|
||||
except Exception:
|
||||
logger.exception("Skill curator startup task failed (non-fatal)")
|
||||
|
||||
# Start IM channel service if any channels are configured
|
||||
try:
|
||||
from app.channels.service import start_channel_service
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import json
|
||||
import logging
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
|
||||
from fastapi import APIRouter, Body, Depends, File, HTTPException, UploadFile
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.gateway.deps import get_config
|
||||
@@ -30,6 +31,13 @@ class SkillResponse(BaseModel):
|
||||
license: str | None = Field(None, description="License information")
|
||||
category: SkillCategory = Field(..., description="Category of the skill (public or custom)")
|
||||
enabled: bool = Field(default=True, description="Whether this skill is enabled")
|
||||
use_count: int = Field(default=0, description="Number of times the skill has been used")
|
||||
view_count: int = Field(default=0, description="Number of times the skill has been viewed")
|
||||
patch_count: int = Field(default=0, description="Number of times the skill has been patched")
|
||||
last_activity_at: datetime | None = Field(default=None, description="Most recent activity timestamp")
|
||||
state: str = Field(default="active", description="Lifecycle state: active, stale, or archived")
|
||||
pinned: bool = Field(default=False, description="Whether this skill is pinned (excluded from curation)")
|
||||
source: str = Field(default="upload", description="Origin: 'agent' for agent-created, 'upload' for user-uploaded")
|
||||
|
||||
|
||||
class SkillsListResponse(BaseModel):
|
||||
@@ -91,12 +99,29 @@ class SkillRollbackRequest(BaseModel):
|
||||
|
||||
def _skill_to_response(skill: Skill) -> SkillResponse:
|
||||
"""Convert a Skill object to a SkillResponse."""
|
||||
from deerflow.skills.usage import get_entry, latest_activity_at
|
||||
|
||||
entry = get_entry(skill.name)
|
||||
last_at_str = latest_activity_at(skill.name)
|
||||
last_at: datetime | None = None
|
||||
if last_at_str:
|
||||
try:
|
||||
last_at = datetime.fromisoformat(last_at_str)
|
||||
except Exception:
|
||||
pass
|
||||
return SkillResponse(
|
||||
name=skill.name,
|
||||
description=skill.description,
|
||||
license=skill.license,
|
||||
category=skill.category,
|
||||
enabled=skill.enabled,
|
||||
use_count=entry.get("use_count", 0),
|
||||
view_count=entry.get("view_count", 0),
|
||||
patch_count=entry.get("patch_count", 0),
|
||||
last_activity_at=last_at,
|
||||
state=entry.get("state", "active"),
|
||||
pinned=entry.get("pinned", False),
|
||||
source=entry.get("source", "upload"),
|
||||
)
|
||||
|
||||
|
||||
@@ -202,6 +227,11 @@ async def install_skill_upload(
|
||||
try:
|
||||
storage = get_or_new_skill_storage(app_config=config)
|
||||
result = await storage.ainstall_skill_from_archive(tmp_path, overwrite=overwrite, require_skill_suffix=False)
|
||||
try:
|
||||
from deerflow.skills.usage import mark_uploaded
|
||||
mark_uploaded(result["skill_name"])
|
||||
except Exception:
|
||||
pass
|
||||
await refresh_skills_system_prompt_cache_async()
|
||||
return SkillInstallResponse(**result)
|
||||
except FileNotFoundError as e:
|
||||
@@ -229,6 +259,11 @@ async def install_skill(request: SkillInstallRequest, config: AppConfig = Depend
|
||||
try:
|
||||
skill_file_path = resolve_thread_virtual_path(request.thread_id, request.path)
|
||||
result = await get_or_new_skill_storage(app_config=config).ainstall_skill_from_archive(skill_file_path)
|
||||
try:
|
||||
from deerflow.skills.usage import mark_uploaded
|
||||
mark_uploaded(result["skill_name"])
|
||||
except Exception:
|
||||
pass
|
||||
await refresh_skills_system_prompt_cache_async()
|
||||
return SkillInstallResponse(**result)
|
||||
except FileNotFoundError as e:
|
||||
@@ -294,6 +329,11 @@ async def update_custom_skill(skill_name: str, request: CustomSkillUpdateRequest
|
||||
"scanner": {"decision": scan.decision, "reason": scan.reason},
|
||||
},
|
||||
)
|
||||
try:
|
||||
from deerflow.skills.usage import bump_patch
|
||||
bump_patch(skill_name)
|
||||
except Exception:
|
||||
pass
|
||||
await refresh_skills_system_prompt_cache_async()
|
||||
return await get_custom_skill(skill_name, config)
|
||||
except HTTPException:
|
||||
@@ -324,6 +364,11 @@ async def delete_custom_skill(skill_name: str, config: AppConfig = Depends(get_c
|
||||
"scanner": {"decision": "allow", "reason": "Deletion requested."},
|
||||
},
|
||||
)
|
||||
try:
|
||||
from deerflow.skills.usage import forget
|
||||
forget(skill_name)
|
||||
except Exception:
|
||||
pass
|
||||
await refresh_skills_system_prompt_cache_async()
|
||||
return {"success": True}
|
||||
except FileNotFoundError as e:
|
||||
@@ -469,3 +514,168 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update skill {skill_name}: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to update skill: {str(e)}")
|
||||
|
||||
|
||||
class SkillPinRequest(BaseModel):
|
||||
pinned: bool = Field(..., description="Whether to pin or unpin the skill")
|
||||
|
||||
|
||||
@router.post("/skills/custom/{skill_name}/pin", response_model=SkillResponse, summary="Pin or Unpin Skill")
|
||||
async def pin_skill(skill_name: str, request: SkillPinRequest, config: AppConfig = Depends(get_config)) -> SkillResponse:
|
||||
"""Pin a skill to prevent it from being automatically archived by the curator."""
|
||||
try:
|
||||
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
|
||||
storage = get_or_new_skill_storage(app_config=config)
|
||||
if not storage.custom_skill_exists(skill_name):
|
||||
raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
|
||||
from deerflow.skills.usage import set_pinned
|
||||
set_pinned(skill_name, request.pinned)
|
||||
skills = storage.load_skills(enabled_only=False)
|
||||
skill = next((s for s in skills if s.name == skill_name), None)
|
||||
if skill is None:
|
||||
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
|
||||
return _skill_to_response(skill)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to pin skill %s: %s", skill_name, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to pin skill: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/curator/status", summary="Curator Status")
|
||||
async def curator_status() -> dict:
|
||||
"""Return the current curator state and per-lifecycle skill counts."""
|
||||
try:
|
||||
from deerflow.skills.curator import get_curator_status
|
||||
return get_curator_status()
|
||||
except Exception as e:
|
||||
logger.error("Failed to get curator status: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get curator status: {str(e)}")
|
||||
|
||||
|
||||
@router.post("/curator/run", summary="Trigger Curator Run")
|
||||
async def run_curator() -> dict:
|
||||
"""Manually trigger an immediate curator review pass (bypasses interval check, runs in background thread)."""
|
||||
try:
|
||||
import threading
|
||||
from deerflow.skills.curator import _curator_thread_target
|
||||
thread = threading.Thread(target=_curator_thread_target, daemon=True, name="skill-curator-manual")
|
||||
thread.start()
|
||||
return {"triggered": True, "message": "Curator run started in background"}
|
||||
except Exception as e:
|
||||
logger.error("Failed to trigger curator: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to trigger curator: {str(e)}")
|
||||
|
||||
|
||||
class CuratorSubConfig(BaseModel):
|
||||
"""Nested curator configuration mirroring CuratorConfig."""
|
||||
|
||||
enabled: bool = Field(default=True, description="Whether the curator runs automatically.")
|
||||
interval_hours: int = Field(default=168, ge=1, description="Hours between curator runs.")
|
||||
stale_after_days: int = Field(default=30, ge=1, description="Days of inactivity before a skill is marked stale.")
|
||||
archive_after_days: int = Field(default=90, ge=1, description="Days of inactivity before a stale skill is archived.")
|
||||
|
||||
|
||||
class CuratorConfigResponse(BaseModel):
|
||||
"""Response model exposing the effective skill_evolution config."""
|
||||
|
||||
enabled: bool
|
||||
moderation_model_name: str | None
|
||||
curator: CuratorSubConfig
|
||||
|
||||
|
||||
class CuratorConfigRequest(BaseModel):
|
||||
"""Payload for updating skill_evolution overrides.
|
||||
|
||||
All fields are required so the frontend always sends a complete config;
|
||||
backend writes the entire payload to skill_evolution.json atomically.
|
||||
"""
|
||||
|
||||
enabled: bool = Field(..., description="Whether the agent can create and modify skills under skills/custom.")
|
||||
moderation_model_name: str | None = Field(default=None, description="Optional model name for skill security moderation.")
|
||||
curator: CuratorSubConfig
|
||||
|
||||
|
||||
class CuratorLogRecord(BaseModel):
|
||||
ts: str
|
||||
level: str
|
||||
message: str
|
||||
|
||||
|
||||
class CuratorLogsResponse(BaseModel):
|
||||
records: list[CuratorLogRecord]
|
||||
run_count: int
|
||||
last_run_at: str | None
|
||||
|
||||
|
||||
@router.get("/curator/config", response_model=CuratorConfigResponse, summary="Get Skill Evolution Config")
|
||||
async def get_curator_config(config: AppConfig = Depends(get_config)) -> CuratorConfigResponse:
|
||||
"""Return the current effective skill_evolution config (yaml defaults + skill_evolution.json overrides)."""
|
||||
try:
|
||||
evo = config.skill_evolution
|
||||
return CuratorConfigResponse(
|
||||
enabled=evo.enabled,
|
||||
moderation_model_name=evo.moderation_model_name,
|
||||
curator=CuratorSubConfig(
|
||||
enabled=evo.curator.enabled,
|
||||
interval_hours=evo.curator.interval_hours,
|
||||
stale_after_days=evo.curator.stale_after_days,
|
||||
archive_after_days=evo.curator.archive_after_days,
|
||||
),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("Failed to read curator config: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to read curator config: {str(e)}")
|
||||
|
||||
|
||||
@router.put("/curator/config", response_model=CuratorConfigResponse, summary="Update Skill Evolution Config")
|
||||
async def update_curator_config(body: CuratorConfigRequest) -> CuratorConfigResponse:
|
||||
"""Persist overrides to skill_evolution.json and force AppConfig to reload on next read.
|
||||
|
||||
The payload is validated against the SkillEvolutionConfig shape; unknown
|
||||
fields are rejected by the Pydantic model_validate call.
|
||||
"""
|
||||
try:
|
||||
from deerflow.config.skill_evolution_config import SkillEvolutionConfig
|
||||
from deerflow.config.skill_evolution_runtime import save_overrides
|
||||
|
||||
payload = body.model_dump()
|
||||
# Round-trip through SkillEvolutionConfig to ensure persisted shape stays valid.
|
||||
SkillEvolutionConfig.model_validate(payload)
|
||||
save_overrides(payload)
|
||||
|
||||
# Force the next get_app_config() call to pick up the new overrides.
|
||||
# mtime check inside get_app_config() will trigger reload automatically,
|
||||
# but resetting here makes the response below reflect the new state.
|
||||
from deerflow.config.app_config import reset_app_config
|
||||
reset_app_config()
|
||||
|
||||
return await get_curator_config()
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to update curator config: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to update curator config: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/curator/logs", response_model=CuratorLogsResponse, summary="Recent Curator Logs")
|
||||
async def get_curator_logs() -> CuratorLogsResponse:
|
||||
"""Return in-memory log records from the most recent curator run.
|
||||
|
||||
The buffer is process-local and cleared when curator starts a new run or
|
||||
when the backend restarts. Records are ordered oldest → newest.
|
||||
"""
|
||||
try:
|
||||
from deerflow.skills import curator_log
|
||||
from deerflow.skills.curator import get_curator_status
|
||||
|
||||
records = curator_log.get_recent()
|
||||
status = get_curator_status()
|
||||
return CuratorLogsResponse(
|
||||
records=[CuratorLogRecord(**r) for r in records],
|
||||
run_count=int(status.get("run_count", 0)),
|
||||
last_run_at=status.get("last_run_at"),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("Failed to read curator logs: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to read curator logs: {str(e)}")
|
||||
@@ -20,6 +20,8 @@ from deerflow.config.run_events_config import RunEventsConfig
|
||||
from deerflow.config.runtime_paths import existing_project_file
|
||||
from deerflow.config.sandbox_config import SandboxConfig
|
||||
from deerflow.config.skill_evolution_config import SkillEvolutionConfig
|
||||
from deerflow.config.skill_evolution_runtime import get_overrides_mtime as _get_skill_evolution_overrides_mtime
|
||||
from deerflow.config.skill_evolution_runtime import load_overrides as _load_skill_evolution_overrides
|
||||
from deerflow.config.skills_config import SkillsConfig
|
||||
from deerflow.config.stream_bridge_config import StreamBridgeConfig, load_stream_bridge_config_from_dict
|
||||
from deerflow.config.subagents_config import SubagentsAppConfig, load_subagents_config_from_dict
|
||||
@@ -54,6 +56,22 @@ def _legacy_config_candidates() -> tuple[Path, ...]:
|
||||
return (backend_dir / "config.yaml", repo_root / "config.yaml")
|
||||
|
||||
|
||||
def _deep_merge_dict(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Deep-merge ``overrides`` onto a copy of ``base`` and return the result.
|
||||
|
||||
Nested dicts are merged recursively; any other value type in ``overrides``
|
||||
replaces the corresponding value in ``base`` outright.
|
||||
"""
|
||||
merged: dict[str, Any] = dict(base)
|
||||
for key, value in overrides.items():
|
||||
existing = merged.get(key)
|
||||
if isinstance(existing, dict) and isinstance(value, dict):
|
||||
merged[key] = _deep_merge_dict(existing, value)
|
||||
else:
|
||||
merged[key] = value
|
||||
return merged
|
||||
|
||||
|
||||
def logging_level_from_config(name: str | None) -> int:
|
||||
"""Map ``config.yaml`` ``log_level`` string to a :mod:`logging` level constant."""
|
||||
mapping = logging.getLevelNamesMapping()
|
||||
@@ -204,6 +222,15 @@ class AppConfig(BaseModel):
|
||||
extensions_config = ExtensionsConfig.from_file()
|
||||
config_data["extensions"] = extensions_config.model_dump()
|
||||
|
||||
# Apply skill_evolution runtime overrides from skill_evolution.json
|
||||
# (file is optional; load_overrides returns None if missing/invalid)
|
||||
overrides = _load_skill_evolution_overrides()
|
||||
if overrides:
|
||||
yaml_section = config_data.get("skill_evolution") or {}
|
||||
if not isinstance(yaml_section, dict):
|
||||
yaml_section = {}
|
||||
config_data["skill_evolution"] = _deep_merge_dict(yaml_section, overrides)
|
||||
|
||||
result = cls.model_validate(config_data)
|
||||
return result
|
||||
|
||||
@@ -329,6 +356,7 @@ class AppConfig(BaseModel):
|
||||
_app_config: AppConfig | None = None
|
||||
_app_config_path: Path | None = None
|
||||
_app_config_mtime: float | None = None
|
||||
_app_config_skill_evolution_overrides_mtime: float | None = None
|
||||
_app_config_is_custom = False
|
||||
_current_app_config: ContextVar[AppConfig | None] = ContextVar("deerflow_current_app_config", default=None)
|
||||
_current_app_config_stack: ContextVar[tuple[AppConfig | None, ...]] = ContextVar("deerflow_current_app_config_stack", default=())
|
||||
@@ -344,12 +372,13 @@ def _get_config_mtime(config_path: Path) -> float | None:
|
||||
|
||||
def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
|
||||
"""Load config from disk and refresh cache metadata."""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_skill_evolution_overrides_mtime, _app_config_is_custom
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path(config_path)
|
||||
_app_config = AppConfig.from_file(str(resolved_path))
|
||||
_app_config_path = resolved_path
|
||||
_app_config_mtime = _get_config_mtime(resolved_path)
|
||||
_app_config_skill_evolution_overrides_mtime = _get_skill_evolution_overrides_mtime()
|
||||
_app_config_is_custom = False
|
||||
return _app_config
|
||||
|
||||
@@ -373,8 +402,14 @@ def get_app_config() -> AppConfig:
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path()
|
||||
current_mtime = _get_config_mtime(resolved_path)
|
||||
current_overrides_mtime = _get_skill_evolution_overrides_mtime()
|
||||
|
||||
should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_mtime != current_mtime
|
||||
should_reload = (
|
||||
_app_config is None
|
||||
or _app_config_path != resolved_path
|
||||
or _app_config_mtime != current_mtime
|
||||
or _app_config_skill_evolution_overrides_mtime != current_overrides_mtime
|
||||
)
|
||||
if should_reload:
|
||||
if _app_config_path == resolved_path and _app_config_mtime is not None and current_mtime is not None and _app_config_mtime != current_mtime:
|
||||
logger.info(
|
||||
@@ -382,6 +417,12 @@ def get_app_config() -> AppConfig:
|
||||
_app_config_mtime,
|
||||
current_mtime,
|
||||
)
|
||||
elif _app_config_skill_evolution_overrides_mtime != current_overrides_mtime:
|
||||
logger.info(
|
||||
"skill_evolution.json overrides changed (mtime: %s -> %s), reloading AppConfig",
|
||||
_app_config_skill_evolution_overrides_mtime,
|
||||
current_overrides_mtime,
|
||||
)
|
||||
_load_and_cache_app_config(str(resolved_path))
|
||||
return _app_config
|
||||
|
||||
@@ -409,10 +450,11 @@ def reset_app_config() -> None:
|
||||
`get_app_config()` to reload from file. Useful for testing
|
||||
or when switching between different configurations.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_skill_evolution_overrides_mtime, _app_config_is_custom
|
||||
_app_config = None
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_skill_evolution_overrides_mtime = None
|
||||
_app_config_is_custom = False
|
||||
|
||||
|
||||
@@ -424,10 +466,11 @@ def set_app_config(config: AppConfig) -> None:
|
||||
Args:
|
||||
config: The AppConfig instance to use.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_skill_evolution_overrides_mtime, _app_config_is_custom
|
||||
_app_config = config
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_skill_evolution_overrides_mtime = None
|
||||
_app_config_is_custom = True
|
||||
|
||||
|
||||
|
||||
+13
@@ -1,6 +1,15 @@
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class CuratorConfig(BaseModel):
|
||||
"""Configuration for the background skill curator."""
|
||||
|
||||
enabled: bool = Field(default=True, description="Whether the curator runs automatically.")
|
||||
interval_hours: int = Field(default=168, description="Hours between curator runs (default: 7 days).")
|
||||
stale_after_days: int = Field(default=30, description="Days of inactivity before a skill is marked stale.")
|
||||
archive_after_days: int = Field(default=90, description="Days of inactivity before a stale skill is archived.")
|
||||
|
||||
|
||||
class SkillEvolutionConfig(BaseModel):
|
||||
"""Configuration for agent-managed skill evolution."""
|
||||
|
||||
@@ -12,3 +21,7 @@ class SkillEvolutionConfig(BaseModel):
|
||||
default=None,
|
||||
description="Optional model name for skill security moderation. Defaults to the primary chat model.",
|
||||
)
|
||||
curator: CuratorConfig = Field(
|
||||
default_factory=CuratorConfig,
|
||||
description="Configuration for the background skill curator.",
|
||||
)
|
||||
+115
@@ -0,0 +1,115 @@
|
||||
"""Runtime overrides for `skill_evolution` config, stored in skill_evolution.json.
|
||||
|
||||
The file format mirrors the ``skill_evolution`` YAML section — its content is
|
||||
deep-merged onto the defaults loaded from ``config.yaml`` when ``AppConfig`` is
|
||||
instantiated. The file is optional; 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
|
||||
@@ -0,0 +1,440 @@
|
||||
"""Skill curator — automatic lifecycle transitions and LLM consolidation pass."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
import threading
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CURATOR_STATE_FILENAME = ".curator_state"
|
||||
|
||||
|
||||
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() -> Path:
|
||||
return _get_skills_root() / _CURATOR_STATE_FILENAME
|
||||
|
||||
|
||||
def _read_curator_state() -> dict:
|
||||
path = _curator_state_path()
|
||||
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) -> None:
|
||||
path = _curator_state_path()
|
||||
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 _get_curator_config():
|
||||
try:
|
||||
from deerflow.config import get_app_config
|
||||
|
||||
config = get_app_config()
|
||||
skill_evo = getattr(config, "skill_evolution", None)
|
||||
return getattr(skill_evo, "curator", None)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def should_run_now() -> bool:
|
||||
"""Return True if the curator interval has elapsed and it is not paused."""
|
||||
curator_cfg = _get_curator_config()
|
||||
if curator_cfg is None or not getattr(curator_cfg, "enabled", True):
|
||||
return False
|
||||
state = _read_curator_state()
|
||||
if state.get("paused"):
|
||||
return False
|
||||
last_run = state.get("last_run_at")
|
||||
if not last_run:
|
||||
return True
|
||||
interval_hours = getattr(curator_cfg, "interval_hours", 168)
|
||||
try:
|
||||
last_dt = datetime.fromisoformat(last_run)
|
||||
return datetime.now(UTC) >= last_dt + timedelta(hours=interval_hours)
|
||||
except Exception:
|
||||
return True
|
||||
|
||||
|
||||
def apply_automatic_transitions(now: datetime | None = None) -> dict:
|
||||
"""Pure age-based state transitions — no LLM involved.
|
||||
|
||||
Only processes skills with source=='agent'. Skips pinned skills.
|
||||
Returns counters: {marked_stale, archived, checked}.
|
||||
"""
|
||||
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()
|
||||
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
|
||||
|
||||
entries = agent_created_entries()
|
||||
counters = {"marked_stale": 0, "archived": 0, "checked": len(entries)}
|
||||
|
||||
for name, entry in entries.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
|
||||
elif days_idle >= stale_days and current_state == STATE_ACTIVE:
|
||||
set_state(name, STATE_STALE)
|
||||
counters["marked_stale"] += 1
|
||||
|
||||
return counters
|
||||
|
||||
|
||||
def _render_candidate_list() -> str:
|
||||
from deerflow.skills.usage import STATE_ARCHIVED, all_entries
|
||||
|
||||
entries = {k: v for k, v in all_entries().items() if v.get("source") == "agent" and v.get("state") != STATE_ARCHIVED}
|
||||
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 (moving the skill's directory "
|
||||
"into the .archive/ folder under the skills root) is the maximum "
|
||||
"destructive action. Archives are recoverable; deletion is not.\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 moved to .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():
|
||||
"""Curator-only tool: move a skill directory to .archive/ (recoverable)."""
|
||||
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) -> str:
|
||||
"""Move an agent-created skill to the .archive/ directory (recoverable). Use this instead of delete.
|
||||
|
||||
Args:
|
||||
name: The skill name to archive.
|
||||
"""
|
||||
from deerflow.skills.usage import STATE_ARCHIVED, get_entry, set_state
|
||||
|
||||
storage = get_or_new_skill_storage()
|
||||
if not storage.custom_skill_exists(name):
|
||||
curator_log.record("warning", f"skill_archive: '{name}' not found or already archived")
|
||||
return f"Skill '{name}' not found or already archived."
|
||||
if get_entry(name).get("pinned"):
|
||||
curator_log.record("info", f"skill_archive: '{name}' is pinned — skipping")
|
||||
return f"Skill '{name}' is pinned — skipping."
|
||||
skill_dir = storage.get_custom_skill_dir(name)
|
||||
archive_root = Path(storage.get_skills_root_path()) / "custom" / ".archive"
|
||||
archive_root.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
|
||||
dest = archive_root / f"{name}-{stamp}"
|
||||
try:
|
||||
skill_dir.rename(dest)
|
||||
except Exception as e:
|
||||
curator_log.record("error", f"skill_archive: archive failed for '{name}': {e}")
|
||||
return f"Archive failed: {e}"
|
||||
set_state(name, STATE_ARCHIVED)
|
||||
curator_log.record("info", f"Archived '{name}' → custom/.archive/{dest.name}")
|
||||
return f"Archived '{name}' → custom/.archive/{dest.name}."
|
||||
|
||||
return _skill_archive
|
||||
|
||||
|
||||
def _run_llm_review(candidate_list: str) -> str:
|
||||
"""Run the curator consolidation pass as a React agent with skill tools bound."""
|
||||
from deerflow.skills import curator_log
|
||||
|
||||
try:
|
||||
from langchain_core.messages import HumanMessage
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
|
||||
from deerflow.models import create_chat_model
|
||||
from deerflow.tools.builtins.skill_tools import skill_list_tool, skill_view_tool
|
||||
from deerflow.tools.skill_manage_tool import skill_manage_tool
|
||||
|
||||
model = create_chat_model()
|
||||
tools = [skill_list_tool, skill_view_tool, skill_manage_tool, _make_skill_archive_tool()]
|
||||
agent = create_react_agent(model, tools)
|
||||
|
||||
curator_log.record("info", "LLM review: invoking React agent")
|
||||
prompt = _CURATOR_REVIEW_PROMPT + f"\n\nSkill candidates for this pass:\n{candidate_list}"
|
||||
result = agent.invoke({"messages": [HumanMessage(content=prompt)]})
|
||||
|
||||
messages = result.get("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 review: agent returned final answer")
|
||||
return content
|
||||
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 review: agent returned final answer")
|
||||
return text
|
||||
curator_log.record("warning", "LLM review: no final assistant message produced")
|
||||
return ""
|
||||
except Exception as e:
|
||||
logger.warning("Curator LLM review failed: %s", e)
|
||||
curator_log.record("warning", f"LLM review failed: {e}")
|
||||
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 run_curator_review() -> dict:
|
||||
"""Main curator run: automatic transitions → LLM consolidation → write state."""
|
||||
from deerflow.skills import curator_log
|
||||
|
||||
curator_log.start_new_run()
|
||||
|
||||
now = datetime.now(UTC)
|
||||
logger.info("Curator: starting run at %s", now.isoformat())
|
||||
curator_log.record("info", f"Curator starting run at {now.isoformat()}")
|
||||
|
||||
transition_counters = apply_automatic_transitions(now)
|
||||
logger.info("Curator: transitions=%s", transition_counters)
|
||||
curator_log.record(
|
||||
"info",
|
||||
(
|
||||
f"Age-based transitions: checked={transition_counters['checked']}, "
|
||||
f"marked_stale={transition_counters['marked_stale']}, "
|
||||
f"archived={transition_counters['archived']}"
|
||||
),
|
||||
)
|
||||
|
||||
candidate_list = _render_candidate_list()
|
||||
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"Candidates for LLM review: {candidate_line_count} skill(s)")
|
||||
|
||||
llm_text = _run_llm_review(candidate_list)
|
||||
suggestions = _parse_structured_summary(llm_text)
|
||||
curator_log.record(
|
||||
"info",
|
||||
(
|
||||
f"LLM suggestions parsed: consolidations={len(suggestions['consolidations'])}, "
|
||||
f"prunings={len(suggestions['prunings'])}"
|
||||
),
|
||||
)
|
||||
|
||||
state = _read_curator_state()
|
||||
state["last_run_at"] = now.isoformat()
|
||||
state["run_count"] = state.get("run_count", 0) + 1
|
||||
state["last_run_summary"] = (
|
||||
f"auto: {transition_counters['marked_stale']} marked stale, "
|
||||
f"{transition_counters['archived']} archived; "
|
||||
f"llm: {len(suggestions['consolidations'])} consolidations suggested"
|
||||
)
|
||||
state["last_consolidation_suggestions"] = suggestions
|
||||
_write_curator_state(state)
|
||||
|
||||
logger.info("Curator: run complete — %s", state["last_run_summary"])
|
||||
curator_log.record("info", f"Curator run complete — {state['last_run_summary']}")
|
||||
return {"transitions": transition_counters, "suggestions": suggestions}
|
||||
|
||||
|
||||
def _curator_thread_target() -> None:
|
||||
from deerflow.skills import curator_log
|
||||
|
||||
try:
|
||||
run_curator_review()
|
||||
except Exception as exc:
|
||||
logger.exception("Curator background run failed")
|
||||
curator_log.record("error", f"Curator background run failed: {exc}")
|
||||
|
||||
|
||||
async def maybe_run_curator() -> None:
|
||||
"""Async entry called at startup — spawns a daemon thread if the interval has elapsed."""
|
||||
if not should_run_now():
|
||||
return
|
||||
logger.info("Curator: interval elapsed, spawning background thread")
|
||||
thread = threading.Thread(target=_curator_thread_target, daemon=True, name="skill-curator")
|
||||
thread.start()
|
||||
|
||||
|
||||
def get_curator_status() -> dict:
|
||||
"""Return curator state for the API status endpoint."""
|
||||
from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries
|
||||
|
||||
state = _read_curator_state()
|
||||
entries = agent_created_entries()
|
||||
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
|
||||
|
||||
return {
|
||||
"enabled": _get_curator_config() is not None and getattr(_get_curator_config(), "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],
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
"""In-memory ring buffer holding the most recent curator run's execution log.
|
||||
|
||||
The buffer is intentionally process-local and non-persistent — restarting the
|
||||
server clears it. The Gateway exposes the buffer over ``GET /api/curator/logs``
|
||||
so the frontend can show what the latest curator pass actually did.
|
||||
|
||||
All public functions swallow internal exceptions so a logging failure can never
|
||||
break the curator run itself.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from collections import deque
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_MAX_RECORDS = 200
|
||||
|
||||
_lock = threading.Lock()
|
||||
_buffer: deque[dict[str, Any]] = deque(maxlen=_MAX_RECORDS)
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
return datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
def start_new_run() -> None:
|
||||
"""Clear the buffer and emit a ``run_started`` record.
|
||||
|
||||
Called once at the top of each curator run so the frontend sees a fresh
|
||||
log stream rather than the previous run's tail.
|
||||
"""
|
||||
try:
|
||||
with _lock:
|
||||
_buffer.clear()
|
||||
_buffer.append({"ts": _now_iso(), "level": "info", "message": "run_started"})
|
||||
except Exception:
|
||||
logger.debug("curator_log.start_new_run failed", exc_info=True)
|
||||
|
||||
|
||||
def record(level: str, message: str, **fields: Any) -> None:
|
||||
"""Append a single log record. Never raises.
|
||||
|
||||
``level`` is one of ``info``/``warning``/``error``; unknown values are
|
||||
accepted and rendered verbatim by the frontend. Extra keyword arguments
|
||||
are merged into the record so callers can attach structured context.
|
||||
"""
|
||||
try:
|
||||
entry: dict[str, Any] = {
|
||||
"ts": _now_iso(),
|
||||
"level": str(level or "info"),
|
||||
"message": str(message),
|
||||
}
|
||||
if fields:
|
||||
entry.update({k: v for k, v in fields.items() if k not in entry})
|
||||
with _lock:
|
||||
_buffer.append(entry)
|
||||
except Exception:
|
||||
logger.debug("curator_log.record failed", exc_info=True)
|
||||
|
||||
|
||||
def get_recent() -> list[dict[str, Any]]:
|
||||
"""Return a snapshot copy of the buffer contents. Never raises."""
|
||||
try:
|
||||
with _lock:
|
||||
return list(_buffer)
|
||||
except Exception:
|
||||
logger.debug("curator_log.get_recent failed", exc_info=True)
|
||||
return []
|
||||
|
||||
|
||||
def clear() -> None:
|
||||
"""Drop all buffered records. Used by tests."""
|
||||
try:
|
||||
with _lock:
|
||||
_buffer.clear()
|
||||
except Exception:
|
||||
logger.debug("curator_log.clear failed", exc_info=True)
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Skill usage sidecar — per-skill telemetry stored in .usage.json."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_USAGE_FILENAME = ".usage.json"
|
||||
|
||||
STATE_ACTIVE = "active"
|
||||
STATE_STALE = "stale"
|
||||
STATE_ARCHIVED = "archived"
|
||||
|
||||
SOURCE_AGENT = "agent"
|
||||
SOURCE_UPLOAD = "upload"
|
||||
|
||||
_DEFAULT_ENTRY: dict = {
|
||||
"use_count": 0,
|
||||
"view_count": 0,
|
||||
"patch_count": 0,
|
||||
"last_used_at": None,
|
||||
"last_viewed_at": None,
|
||||
"last_patched_at": None,
|
||||
"created_at": None,
|
||||
"state": STATE_ACTIVE,
|
||||
"pinned": False,
|
||||
"archived_at": None,
|
||||
"source": SOURCE_UPLOAD,
|
||||
}
|
||||
|
||||
|
||||
def _usage_path() -> Path:
|
||||
from deerflow.skills.storage import get_or_new_skill_storage
|
||||
|
||||
return Path(get_or_new_skill_storage().get_skills_root_path()) / _USAGE_FILENAME
|
||||
|
||||
|
||||
def _read_all() -> dict:
|
||||
path = _usage_path()
|
||||
if not path.exists():
|
||||
return {}
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
logger.debug("Failed to read %s", path, exc_info=True)
|
||||
return {}
|
||||
|
||||
|
||||
def _write_all(data: dict) -> None:
|
||||
path = _usage_path()
|
||||
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(data, 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 %s", path, exc_info=True)
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
return datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
def _update_entry(name: str, updates: dict) -> None:
|
||||
try:
|
||||
data = _read_all()
|
||||
entry = {**_DEFAULT_ENTRY, **data.get(name, {})}
|
||||
entry.update(updates)
|
||||
data[name] = entry
|
||||
_write_all(data)
|
||||
except Exception:
|
||||
logger.debug("Failed to update usage entry for %r", name, exc_info=True)
|
||||
|
||||
|
||||
def get_entry(name: str) -> dict:
|
||||
"""Return the usage entry for *name*, filling missing fields with defaults."""
|
||||
data = _read_all()
|
||||
return {**_DEFAULT_ENTRY, **data.get(name, {})}
|
||||
|
||||
|
||||
def all_entries() -> dict:
|
||||
"""Return all usage entries keyed by skill name."""
|
||||
return _read_all()
|
||||
|
||||
|
||||
def agent_created_entries() -> dict:
|
||||
"""Return only entries where source == 'agent' (eligible for curation)."""
|
||||
return {k: v for k, v in _read_all().items() if v.get("source") == SOURCE_AGENT}
|
||||
|
||||
|
||||
def bump_view(name: str) -> None:
|
||||
entry = get_entry(name)
|
||||
_update_entry(name, {"view_count": entry.get("view_count", 0) + 1, "last_viewed_at": _now_iso()})
|
||||
|
||||
|
||||
def bump_use(name: str) -> None:
|
||||
entry = get_entry(name)
|
||||
_update_entry(name, {"use_count": entry.get("use_count", 0) + 1, "last_used_at": _now_iso()})
|
||||
|
||||
|
||||
def bump_patch(name: str) -> None:
|
||||
entry = get_entry(name)
|
||||
_update_entry(name, {"patch_count": entry.get("patch_count", 0) + 1, "last_patched_at": _now_iso()})
|
||||
|
||||
|
||||
def set_state(name: str, state: str) -> None:
|
||||
updates: dict = {"state": state}
|
||||
if state == STATE_ARCHIVED:
|
||||
updates["archived_at"] = _now_iso()
|
||||
elif state == STATE_ACTIVE:
|
||||
updates["archived_at"] = None
|
||||
_update_entry(name, updates)
|
||||
|
||||
|
||||
def set_pinned(name: str, pinned: bool) -> None:
|
||||
_update_entry(name, {"pinned": pinned})
|
||||
|
||||
|
||||
def forget(name: str) -> None:
|
||||
try:
|
||||
data = _read_all()
|
||||
data.pop(name, None)
|
||||
_write_all(data)
|
||||
except Exception:
|
||||
logger.debug("Failed to forget usage entry for %r", name, exc_info=True)
|
||||
|
||||
|
||||
def mark_agent_created(name: str) -> None:
|
||||
"""Record a skill as agent-created; sets source='agent' and initialises created_at."""
|
||||
_update_entry(name, {"source": SOURCE_AGENT, "created_at": _now_iso(), "state": STATE_ACTIVE})
|
||||
|
||||
|
||||
def mark_uploaded(name: str) -> None:
|
||||
"""Record a skill as user-uploaded; sets source='upload'."""
|
||||
_update_entry(name, {"source": SOURCE_UPLOAD, "created_at": _now_iso()})
|
||||
|
||||
|
||||
def latest_activity_at(name: str) -> str | None:
|
||||
"""Return the most recent activity timestamp, falling back to created_at."""
|
||||
entry = get_entry(name)
|
||||
candidates = [entry.get("last_used_at"), entry.get("last_viewed_at"), entry.get("last_patched_at")]
|
||||
valid = [ts for ts in candidates if ts]
|
||||
if valid:
|
||||
return max(valid)
|
||||
return entry.get("created_at")
|
||||
+83
@@ -0,0 +1,83 @@
|
||||
"""skill_view and skill_list — read-only tools for agent skill introspection."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from langchain.tools import tool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@tool("skill_view")
|
||||
def skill_view_tool(name: str) -> str:
|
||||
"""Read the full SKILL.md content of a skill by name.
|
||||
|
||||
Args:
|
||||
name: The skill name (hyphen-case, e.g. 'my-skill').
|
||||
"""
|
||||
from deerflow.skills.storage import get_or_new_skill_storage
|
||||
from deerflow.skills.usage import bump_view
|
||||
|
||||
storage = get_or_new_skill_storage()
|
||||
try:
|
||||
if storage.custom_skill_exists(name):
|
||||
content = storage.read_custom_skill(name)
|
||||
elif storage.public_skill_exists(name):
|
||||
skill_file = storage.get_skills_root_path()
|
||||
from pathlib import Path
|
||||
from deerflow.skills.types import SKILL_MD_FILE
|
||||
|
||||
path = Path(skill_file) / "public" / name / SKILL_MD_FILE
|
||||
if not path.exists():
|
||||
return f"Skill '{name}' not found."
|
||||
content = path.read_text(encoding="utf-8")
|
||||
else:
|
||||
return f"Skill '{name}' not found."
|
||||
except Exception as e:
|
||||
logger.debug("skill_view failed for %r: %s", name, e)
|
||||
return f"Error reading skill '{name}': {e}"
|
||||
|
||||
try:
|
||||
bump_view(name)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return content
|
||||
|
||||
|
||||
@tool("skill_list")
|
||||
def skill_list_tool() -> str:
|
||||
"""List all available skills with metadata and usage stats. Does NOT read SKILL.md content.
|
||||
|
||||
Returns a JSON array with fields: name, description, category, enabled, state,
|
||||
use_count, view_count, patch_count, last_activity_at, pinned, source.
|
||||
"""
|
||||
from deerflow.skills.storage import get_or_new_skill_storage
|
||||
from deerflow.skills.usage import all_entries, latest_activity_at
|
||||
|
||||
try:
|
||||
storage = get_or_new_skill_storage()
|
||||
skills = storage.load_skills(enabled_only=False)
|
||||
usage = all_entries()
|
||||
result = []
|
||||
for skill in skills:
|
||||
entry = usage.get(skill.name, {})
|
||||
result.append({
|
||||
"name": skill.name,
|
||||
"description": skill.description,
|
||||
"category": skill.category.value if hasattr(skill.category, "value") else str(skill.category),
|
||||
"enabled": skill.enabled,
|
||||
"state": entry.get("state", "active"),
|
||||
"use_count": entry.get("use_count", 0),
|
||||
"view_count": entry.get("view_count", 0),
|
||||
"patch_count": entry.get("patch_count", 0),
|
||||
"last_activity_at": latest_activity_at(skill.name),
|
||||
"pinned": entry.get("pinned", False),
|
||||
"source": entry.get("source", "upload"),
|
||||
})
|
||||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||||
except Exception as e:
|
||||
logger.debug("skill_list failed: %s", e)
|
||||
return f"Error listing skills: {e}"
|
||||
@@ -104,6 +104,11 @@ async def _skill_manage_impl(
|
||||
name,
|
||||
_history_record(action="create", file_path=SKILL_MD_FILE, prev_content=None, new_content=content, thread_id=thread_id, scanner=scan),
|
||||
)
|
||||
try:
|
||||
from deerflow.skills.usage import mark_agent_created
|
||||
await _to_thread(mark_agent_created, name)
|
||||
except Exception:
|
||||
pass
|
||||
await refresh_skills_system_prompt_cache_async()
|
||||
return f"Created custom skill '{name}'."
|
||||
|
||||
|
||||
@@ -86,9 +86,10 @@ def get_available_tools(
|
||||
builtin_tools.append(scheduled_task_tool)
|
||||
skill_evolution_config = getattr(config, "skill_evolution", None)
|
||||
if getattr(skill_evolution_config, "enabled", False):
|
||||
from deerflow.tools.builtins.skill_tools import skill_list_tool, skill_view_tool
|
||||
from deerflow.tools.skill_manage_tool import skill_manage_tool
|
||||
|
||||
builtin_tools.append(skill_manage_tool)
|
||||
builtin_tools.extend([skill_list_tool, skill_view_tool, skill_manage_tool])
|
||||
|
||||
# Add subagent tools only if enabled via runtime parameter
|
||||
if subagent_enabled:
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.config import skill_evolution_runtime
|
||||
from deerflow.skills import curator_log
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def overrides_path(tmp_path, monkeypatch):
|
||||
"""Redirect skill_evolution_runtime.resolve_path at a fresh temp file."""
|
||||
target = tmp_path / "skill_evolution.json"
|
||||
monkeypatch.setenv("DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH", str(target))
|
||||
yield target
|
||||
|
||||
|
||||
def test_load_overrides_returns_none_when_missing(overrides_path):
|
||||
assert not overrides_path.exists()
|
||||
assert skill_evolution_runtime.load_overrides() is None
|
||||
|
||||
|
||||
def test_save_and_load_overrides_roundtrip(overrides_path):
|
||||
payload = {
|
||||
"enabled": True,
|
||||
"moderation_model_name": "moderation-mini",
|
||||
"curator": {
|
||||
"enabled": True,
|
||||
"interval_hours": 12,
|
||||
"stale_after_days": 7,
|
||||
"archive_after_days": 30,
|
||||
},
|
||||
}
|
||||
written = skill_evolution_runtime.save_overrides(payload)
|
||||
assert written == overrides_path
|
||||
assert overrides_path.exists()
|
||||
assert skill_evolution_runtime.load_overrides() == payload
|
||||
|
||||
|
||||
def test_load_overrides_handles_invalid_json(overrides_path):
|
||||
overrides_path.write_text("{not valid json", encoding="utf-8")
|
||||
assert skill_evolution_runtime.load_overrides() is None
|
||||
|
||||
|
||||
def test_load_overrides_rejects_non_object_root(overrides_path):
|
||||
overrides_path.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
|
||||
assert skill_evolution_runtime.load_overrides() is None
|
||||
|
||||
|
||||
def test_app_config_merges_overrides_on_from_file(overrides_path, monkeypatch, tmp_path):
|
||||
yaml_cfg = tmp_path / "config.yaml"
|
||||
yaml_cfg.write_text(
|
||||
"""
|
||||
sandbox:
|
||||
use: deerflow.sandbox.local:LocalSandboxProvider
|
||||
skill_evolution:
|
||||
enabled: false
|
||||
curator:
|
||||
interval_hours: 168
|
||||
stale_after_days: 30
|
||||
""".strip(),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
# Overrides flip enabled and change interval_hours.
|
||||
skill_evolution_runtime.save_overrides(
|
||||
{
|
||||
"enabled": True,
|
||||
"curator": {"interval_hours": 24},
|
||||
}
|
||||
)
|
||||
|
||||
from deerflow.config.app_config import AppConfig
|
||||
|
||||
cfg = AppConfig.from_file(str(yaml_cfg))
|
||||
assert cfg.skill_evolution.enabled is True
|
||||
assert cfg.skill_evolution.curator.interval_hours == 24
|
||||
# Untouched yaml values remain.
|
||||
assert cfg.skill_evolution.curator.stale_after_days == 30
|
||||
|
||||
|
||||
def test_curator_log_start_new_run_resets_and_seeds_record():
|
||||
curator_log.clear()
|
||||
curator_log.record("info", "old record")
|
||||
assert len(curator_log.get_recent()) == 1
|
||||
|
||||
curator_log.start_new_run()
|
||||
records = curator_log.get_recent()
|
||||
assert len(records) == 1
|
||||
assert records[0]["message"] == "run_started"
|
||||
assert records[0]["level"] == "info"
|
||||
|
||||
|
||||
def test_curator_log_record_is_thread_safe():
|
||||
curator_log.clear()
|
||||
|
||||
def writer(idx: int) -> None:
|
||||
for i in range(20):
|
||||
curator_log.record("info", f"thread-{idx}-{i}")
|
||||
|
||||
threads = [threading.Thread(target=writer, args=(i,)) for i in range(5)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
records = curator_log.get_recent()
|
||||
# 5 threads * 20 records each = 100, well under the 200 ring-buffer cap.
|
||||
assert len(records) == 100
|
||||
assert all("message" in r and "ts" in r for r in records)
|
||||
|
||||
|
||||
def test_curator_log_record_ignores_internal_errors(monkeypatch):
|
||||
# Even if the timestamp helper blows up, record() must not raise.
|
||||
monkeypatch.setattr(curator_log, "_now_iso", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
||||
curator_log.record("info", "should not raise")
|
||||
Generated
+2692
-2692
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user