From 9ccc473982070b501b388ceac823b41b87270553 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 27 May 2026 22:23:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/app/gateway/app.py | 18 +++ .../backend/app/gateway/routers/agents.py | 122 ++++++++++++++++-- .../harness/deerflow/config/agents_config.py | 54 +++++++- 3 files changed, 179 insertions(+), 15 deletions(-) diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index fba742f5..15287a9d 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -502,6 +502,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Skill ownership migration failed (non-fatal)") + # 一次性把 .deer-flow/agents/{id}/ 目录里历史遗留的 agent 导入 agents 表。 + # 路由层原本每次请求都跑一遍这个 sync(扫文件系统 + N 个 agent 各一次 DB + # SELECT/UPDATE),把它挪到启动期 + 路由层只保留 TTL 兜底,显著降低 + # /api/agents 列表延迟。 + try: + from app.gateway.routers.agents import ( + _mark_legacy_sync_fresh, + _sync_legacy_agents, + ) + + agent_store = getattr(app.state, "agent_store", None) + if agent_store is not None: + await _sync_legacy_agents(agent_store) + _mark_legacy_sync_fresh() + logger.info("Legacy folder-backed agents synced at startup") + except Exception: + logger.exception("Legacy agent sync failed at startup (non-fatal)") + # 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化 # (per-user interval_hours 仍是真正决定要不要跑进化的闸)。 try: diff --git a/offline-backend-20260512/backend/app/gateway/routers/agents.py b/offline-backend-20260512/backend/app/gateway/routers/agents.py index 800c4364..4672265a 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/agents.py +++ b/offline-backend-20260512/backend/app/gateway/routers/agents.py @@ -2,8 +2,10 @@ from __future__ import annotations +import asyncio import logging import shutil +import time from datetime import datetime from typing import Any from uuid import uuid4 @@ -192,6 +194,11 @@ async def _sync_legacy_agents(store: AgentStore) -> None: This preserves old deployments where agents already existed under ``.deer-flow/agents/{name}`` before the agents table was introduced. + + Heavy: scans the agents directory + parses each ``config.yaml`` + issues a + SELECT-then-INSERT/UPDATE per agent. Call once at startup via the lifespan, + and rely on :func:`_maybe_sync_legacy_agents` (TTL-gated) inside request + handlers as a safety net for environments that bypassed the lifespan hook. """ try: for agent_cfg in list_custom_agents(): @@ -210,6 +217,44 @@ async def _sync_legacy_agents(store: AgentStore) -> None: logger.exception("Failed to sync legacy folder-backed agents") +# Throttle the per-request legacy-sync fallback. The real sync runs once at +# startup (lifespan); request handlers used to call ``_sync_legacy_agents`` on +# every hit, which scanned the filesystem + did N DB upserts per agent each +# time. After the first successful sync we skip the work for ``_LEGACY_SYNC_TTL_SECONDS`` +# — long enough to be effectively "startup-only" in steady state, short enough +# that a freshly dropped agent directory gets picked up on the next bucket. +_LEGACY_SYNC_TTL_SECONDS = 3600.0 +_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None} + + +async def _maybe_sync_legacy_agents(store: AgentStore) -> None: + """TTL-gated wrapper around :func:`_sync_legacy_agents`. + + Becomes a no-op in steady state — the heavy lifting happens in the lifespan + hook. Inside the TTL window we don't even take the lock; outside it we + serialize callers so a thundering herd doesn't fan out into N filesystem + scans + N*M DB queries. + """ + now = time.monotonic() + if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS: + return + lock = _legacy_sync_state["lock"] + if lock is None: + lock = asyncio.Lock() + _legacy_sync_state["lock"] = lock + async with lock: + # Re-check under the lock — a sibling coroutine may have just refreshed. + if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS: + return + await _sync_legacy_agents(store) + _legacy_sync_state["last_at"] = time.monotonic() + + +def _mark_legacy_sync_fresh() -> None: + """Called by the lifespan hook after running :func:`_sync_legacy_agents`.""" + _legacy_sync_state["last_at"] = time.monotonic() + + def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None: """Lazily restore an agent's on-disk artifacts from its DB record. @@ -247,14 +292,29 @@ def _record_to_response( record: dict[str, Any], *, include_soul: bool = False, + include_extras: bool = True, ) -> AgentResponse: - _ensure_agent_dir_from_record(record) + """Convert a DB row to an API response. + + Args: + include_soul: Read SOUL.md from disk and embed in ``soul``. Set False + in list endpoints — the frontend (gallery / recommend / selector) + never reads ``soul``, only the edit dialog does. Skipping this + avoids one synchronous file read per agent. + include_extras: Self-heal a missing agent dir and read deprecated + ``model`` / ``tool_groups`` / ``skills`` from ``config.yaml``. Set + False in list endpoints to skip a stat + yaml parse per agent — + those fields are unused by current frontends and the dir restore + can happen lazily when the agent is actually opened. + """ soul: str | None = None + extras: dict[str, Any] = {"skills": None, "model": None, "tool_groups": None} + if include_extras: + _ensure_agent_dir_from_record(record) + extras = _load_config_extras(record["id"]) if include_soul: soul = load_agent_soul(record["id"]) or "" - extras = _load_config_extras(record["id"]) - return AgentResponse( id=record["id"], name=record["name"], @@ -272,6 +332,25 @@ def _record_to_response( ) +async def _record_to_response_async( + record: dict[str, Any], + *, + include_soul: bool = False, + include_extras: bool = True, +) -> AgentResponse: + """Async wrapper: when we actually touch disk, hop to a worker thread so + we don't block the asyncio loop. List endpoints (no soul, no extras) skip + this overhead and stay on the calling task.""" + if not include_soul and not include_extras: + return _record_to_response(record, include_soul=False, include_extras=False) + return await asyncio.to_thread( + _record_to_response, + record, + include_soul=include_soul, + include_extras=include_extras, + ) + + async def _get_owned_or_raise(store: AgentStore, agent_id: str, user_id: str) -> dict[str, Any]: owned = await store.get_owned(agent_id, user_id) if owned is not None: @@ -321,10 +400,17 @@ async def list_agents( try: store = get_agent_store(request) tag_store: TagStore = get_tag_store(request) - await _sync_legacy_agents(store) + await _maybe_sync_legacy_agents(store) user_id = _current_user_id(request) agents = await store.list_visible(user_id) - responses = [_record_to_response(a, include_soul=True) for a in agents] + # List endpoint stays lean: skip per-agent SOUL.md read, config.yaml + # parse, and dir self-heal — none of those feed UI fields the gallery / + # recommend dialog / selector actually render. Detail endpoints (single + # agent + create/update responses) still load them. + responses = [ + _record_to_response(a, include_soul=False, include_extras=False) + for a in agents + ] assignments = await tag_store.list_assignments("agent", [r.id for r in responses]) for resp in responses: @@ -368,11 +454,18 @@ async def get_agent_selector(request: Request) -> AgentSelectorResponse: _require_agents_api_enabled() try: store = get_agent_store(request) - await _sync_legacy_agents(store) + await _maybe_sync_legacy_agents(store) user_id = _current_user_id(request) records = await store.list_featured(user_id) - return AgentSelectorResponse(agents=[_record_to_response(r) for r in records]) + # Selector dropdown only renders name/description/featured_order — same + # rationale as list_agents: skip soul + extras + self-heal. + return AgentSelectorResponse( + agents=[ + _record_to_response(r, include_soul=False, include_extras=False) + for r in records + ] + ) except HTTPException: raise except Exception as e: @@ -392,12 +485,15 @@ async def get_agent(agent_id: str, request: Request) -> AgentResponse: try: store = get_agent_store(request) - await _sync_legacy_agents(store) + await _maybe_sync_legacy_agents(store) user_id = _current_user_id(request) record = await store.get_visible(agent_id, user_id) if record is None: raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found") - return _record_to_response(record, include_soul=True) + # Detail view legitimately reads SOUL.md + config.yaml — hop to a + # worker thread so the asyncio loop stays responsive under concurrent + # detail fetches. + return await _record_to_response_async(record, include_soul=True, include_extras=True) except HTTPException: raise except Exception as e: @@ -449,7 +545,7 @@ async def create_agent_endpoint(request: AgentCreateRequest, http_request: Reque } ) logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir) - return _record_to_response(record, include_soul=True) + return await _record_to_response_async(record, include_soul=True, include_extras=True) except HTTPException: raise @@ -518,7 +614,7 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request: raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found") logger.info("Updated agent '%s' (%s)", record["name"], agent_id) - return _record_to_response(record, include_soul=True) + return await _record_to_response_async(record, include_soul=True, include_extras=True) except HTTPException: raise @@ -642,7 +738,7 @@ async def set_agent_featured( try: store = get_agent_store(request) - await _sync_legacy_agents(store) + await _maybe_sync_legacy_agents(store) record = await store.get_any(agent_id) if record is None: raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found") @@ -654,7 +750,7 @@ async def set_agent_featured( if updated is None: raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found") logger.info("Set agent '%s' featured_order=%s by admin", agent_id, body.order) - return _record_to_response(updated, include_soul=True) + return await _record_to_response_async(updated, include_soul=True, include_extras=True) except HTTPException: raise except Exception as e: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py index 63358dd7..3fddcd22 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py @@ -1,7 +1,9 @@ """Configuration and loaders for custom agents.""" import logging +import os import re +import threading from typing import Any import yaml @@ -13,6 +15,25 @@ logger = logging.getLogger(__name__) SOUL_FILENAME = "SOUL.md" AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$") + +# Per-file mtime caches. These are read on every /api/agents list/detail request +# (one read per agent) — caching keyed by (path, mtime_ns, size) is correct +# because any edit to the file updates mtime, and an external tool re-writing +# the same bytes doesn't actually change the result we'd return. Cache size is +# unbounded but lookup is O(1) and entries are tiny (one parsed dict / string +# per agent file); for the order of N≤10⁴ agents the memory cost is negligible. +_config_cache: dict[str, tuple[tuple[int, int], "AgentConfig | None"]] = {} +_soul_cache: dict[str, tuple[tuple[int, int], str | None]] = {} +_cache_lock = threading.Lock() + + +def _file_stat_key(path) -> tuple[int, int] | None: + """Return (mtime_ns, size) for cache invalidation, or None when missing.""" + try: + st = os.stat(path) + except OSError: + return None + return (st.st_mtime_ns, st.st_size) # Backward-compatible alias. Historically this validated the display name; # after the DB-backed agent table, display names can be Chinese and duplicated. # The pattern now applies only to storage/runtime ids. @@ -94,6 +115,18 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None: ) return None + # mtime-based cache. The config file is rewritten every time the agent is + # edited via PUT /api/agents/{id}, which bumps mtime — so the next read + # picks up the change automatically. Missing config_file = key None below, + # bypassing the cache (cheap path: just return AgentConfig with id/name). + cache_path_key = str(config_file) + stat_key = _file_stat_key(config_file) if config_file.exists() else None + if stat_key is not None: + with _cache_lock: + cached = _config_cache.get(cache_path_key) + if cached is not None and cached[0] == stat_key: + return cached[1] + data: dict[str, Any] = {} if config_file.exists(): try: @@ -112,7 +145,11 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None: known_fields = set(AgentConfig.model_fields.keys()) data = {k: v for k, v in data.items() if k in known_fields} - return AgentConfig(**data) + result = AgentConfig(**data) + if stat_key is not None: + with _cache_lock: + _config_cache[cache_path_key] = (stat_key, result) + return result def load_agent_soul(agent_id: str | None) -> str | None: @@ -131,8 +168,21 @@ def load_agent_soul(agent_id: str | None) -> str | None: soul_path = agent_dir / SOUL_FILENAME if not soul_path.exists(): return None + # mtime-based cache, same rationale as load_agent_config: SOUL.md is + # rewritten on edit (PUT /api/agents/{id}), which bumps mtime. + cache_path_key = str(soul_path) + stat_key = _file_stat_key(soul_path) + if stat_key is not None: + with _cache_lock: + cached = _soul_cache.get(cache_path_key) + if cached is not None and cached[0] == stat_key: + return cached[1] content = soul_path.read_text(encoding="utf-8").strip() - return content or None + result = content or None + if stat_key is not None: + with _cache_lock: + _soul_cache[cache_path_key] = (stat_key, result) + return result def list_custom_agents() -> list[AgentConfig]: