diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 3714e7e7..2480abab 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -1,6 +1,7 @@ import asyncio import logging import os +import sys from collections.abc import AsyncGenerator from contextlib import asynccontextmanager @@ -268,6 +269,93 @@ async def _run_db_migrations() -> None: logger.exception("Database migration failed — schema may be out of date") +# Skill curator 调度参数。 +# - 检查间隔:每隔这么久检查一次"需不需要进化";真正是否执行由 should_run_now() 判定。 +# - 启动延迟:启动后先等这么久再首次检查,避开启动高峰。 +_CURATOR_CHECK_INTERVAL_SECONDS = 3600 # 1 小时 +_CURATOR_STARTUP_DELAY_SECONDS = 300 # 5 分钟 + + +def _acquire_singleton_lock(lock_path): + """跨平台非阻塞独占文件锁,用于多 worker 下的 leader 选举。 + + 成功 -> 返回持有锁的文件句柄(调用方须保持其存活,直到进程退出/主动释放); + 抢不到锁(已有其它 worker 持有)或打开失败 -> 返回 None。 + 进程异常退出时 OS 会自动释放该锁。 + """ + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = open(lock_path, "a+") + except OSError: + logger.exception("Curator scheduler: 打开锁文件失败") + return None + try: + if sys.platform == "win32": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + handle.close() + return None + return handle + + +def _release_singleton_lock(handle) -> None: + """释放并关闭 _acquire_singleton_lock 拿到的锁文件句柄。""" + try: + if sys.platform == "win32": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + except OSError: + pass + finally: + try: + handle.close() + except OSError: + pass + + +async def _curator_scheduler_loop() -> None: + """常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。 + + 多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正 + 运行调度,其余 worker 直接退出,避免 curator 被重复触发、重复调用 LLM。 + 被 cancel 时 CancelledError 会穿过 ``except Exception``(它是 BaseException + 子类,不会被捕获),协程随之干净退出并在 finally 里释放锁。 + """ + from deerflow.config.runtime_paths import runtime_home + from deerflow.skills.curator import maybe_run_curator + + lock_handle = _acquire_singleton_lock(runtime_home() / ".curator_scheduler.lock") + if lock_handle is None: + logger.info("Curator scheduler: 另一个 worker 已持有调度锁,本 worker 跳过") + return + + logger.info("Curator scheduler: 取得调度锁,本 worker 负责 curator 调度") + try: + # 启动后延迟一会儿再首次检查,避开启动高峰 + await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS) + while True: + try: + await maybe_run_curator() + except Exception: + logger.exception("Curator scheduler tick failed (non-fatal)") + await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS) + finally: + _release_singleton_lock(lock_handle) + + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """Application lifespan handler.""" @@ -302,12 +390,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Skill ownership migration failed (non-fatal)") - # Kick off skill curator in background if the run interval has elapsed + # 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化 + # (per-user interval_hours 仍是真正决定要不要跑进化的闸)。 try: - from deerflow.skills.curator import maybe_run_curator - asyncio.create_task(maybe_run_curator()) + app.state.curator_scheduler_task = asyncio.create_task(_curator_scheduler_loop()) except Exception: - logger.exception("Skill curator startup task failed (non-fatal)") + logger.exception("Skill curator scheduler failed to start (non-fatal)") # Start IM channel service if any channels are configured try: @@ -335,6 +423,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: yield + # 停止 curator 调度器 + curator_task = getattr(app.state, "curator_scheduler_task", None) + if curator_task is not None: + curator_task.cancel() + try: + await curator_task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Curator scheduler shutdown error") + try: scheduler_service = getattr(app.state, "scheduled_task_service", None) if scheduler_service is not None: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py index b40c269c..f491ed4f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py @@ -264,7 +264,7 @@ def _eligible_for_curation( 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_cfg = _get_curator_config() + curator_cfg = _get_curator_config(user_id) if curator_cfg is None or not getattr(curator_cfg, "enabled", True): return False state = _read_curator_state(user_id) @@ -297,7 +297,7 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di if now is None: now = datetime.now(UTC) - curator_cfg = _get_curator_config() + 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 @@ -590,8 +590,72 @@ def _make_skill_archive_tool(user_id: str): return _skill_archive -def _run_llm_review(candidate_list: str, *, user_id: str) -> str: - """Run the curator consolidation pass as a React agent with skill tools bound.""" +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: @@ -601,10 +665,10 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str: 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 - 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(user_id)] + # 用 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 @@ -618,25 +682,26 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str: result = agent.invoke({"messages": [HumanMessage(content=prompt)]}) 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 + 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 + return text, tool_actions curator_log.record("warning", "LLM 评审:本次未产出最终回复", user_id=user_id) - return "" + 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 "" + return "", [] def _parse_structured_summary(text: str) -> dict: @@ -653,6 +718,44 @@ def _parse_structured_summary(text: str) -> dict: 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. @@ -684,16 +787,23 @@ def run_curator_review(user_id: str) -> dict: "info", f"进入 LLM 评审的候选技能:共 {candidate_line_count} 个", user_id=user_id ) - llm_text = _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, - ) + 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 []: @@ -718,15 +828,26 @@ def run_curator_review(user_id: str) -> dict: 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'])} 项" + f"LLM:建议合并 {len(suggestions['consolidations'])} 项、实际工具调用 {len(tool_actions)} 次" ) - state["last_consolidation_suggestions"] = suggestions + 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"]) @@ -768,10 +889,11 @@ def _curator_thread_target(user_id: str) -> None: async def maybe_run_curator() -> None: - """Async startup entry — kicks off a thread per user whose interval elapsed. + """周期调度入口 —— 给每个 interval 已到期的用户起一个后台策展线程。 - Skips users with no DB-tracked entries (a fresh deployment) and never - blocks lifespan startup on failure. + 由 app 的 curator 调度器周期性调用(也用于手动触发)。无候选技能的用户,其 + LLM 评审会在 run_curator_review 内被跳过 —— 只做不耗费 LLM 的年龄流转。 + 任何失败都不会阻断调用方。 """ try: from sqlalchemy import select @@ -819,7 +941,7 @@ def get_curator_status(user_id: str) -> dict: s = entry.get("state", STATE_ACTIVE) counts[s] = counts.get(s, 0) + 1 - cfg = _get_curator_config() + cfg = _get_curator_config(user_id) return { "enabled": cfg is not None and getattr(cfg, "enabled", True), "run_count": state.get("run_count", 0), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py index b36f7f4b..79abe58c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py @@ -6,6 +6,7 @@ import json import logging import os import tempfile +import threading from datetime import UTC, datetime from pathlib import Path @@ -13,6 +14,11 @@ logger = logging.getLogger(__name__) _USAGE_FILENAME = ".usage.json" +# 串行化对 .usage.json 的"读-改-写",避免并发(尤其 curator 多线程)丢更新。 +# .usage.json 是全局单文件;此锁覆盖进程内并发。leader 选举已使 curator 只在 +# 单进程跑,故进程内锁即可保证状态流转不丢。 +_usage_lock = threading.Lock() + STATE_ACTIVE = "active" STATE_STALE = "stale" STATE_ARCHIVED = "archived" @@ -75,13 +81,20 @@ def _now_iso() -> str: return datetime.now(UTC).isoformat() -def _update_entry(name: str, updates: dict) -> None: +def _update_entry(name: str, updates) -> None: + """更新一个 usage 条目。``updates`` 可以是 dict,或 ``(entry) -> dict`` 回调。 + + 用回调形式可让"读当前值 -> 算增量"也落在锁内(例如计数自增), + 避免并发下的 lost update。 + """ try: - data = _read_all() - entry = {**_DEFAULT_ENTRY, **data.get(name, {})} - entry.update(updates) - data[name] = entry - _write_all(data) + with _usage_lock: + data = _read_all() + entry = {**_DEFAULT_ENTRY, **data.get(name, {})} + resolved = updates(entry) if callable(updates) else updates + entry.update(resolved) + data[name] = entry + _write_all(data) except Exception: logger.debug("Failed to update usage entry for %r", name, exc_info=True) @@ -103,18 +116,16 @@ def agent_created_entries() -> dict: 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()}) + # 用回调:读当前计数 + 自增在同一把锁内完成,避免并发丢计数。 + _update_entry(name, lambda e: {"view_count": e.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()}) + _update_entry(name, lambda e: {"use_count": e.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()}) + _update_entry(name, lambda e: {"patch_count": e.get("patch_count", 0) + 1, "last_patched_at": _now_iso()}) def set_state(name: str, state: str) -> None: @@ -132,9 +143,10 @@ def set_pinned(name: str, pinned: bool) -> None: def forget(name: str) -> None: try: - data = _read_all() - data.pop(name, None) - _write_all(data) + with _usage_lock: + 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)