智能体查询优化

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:

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:

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]: