mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 00:17:53 +00:00
0d4d0cb17d
* feat(agents): database-backed storage for custom agent definitions Add an agent_storage.backend switch (default file, behaviour-unchanged) with a db backend that stores each custom agent as a row in the shared SQL persistence layer, so a multi-instance deployment sees the same agents on every node (#4331, #4357). Introduces an AgentStore interface routing all read/write surfaces, an agents table + migration 0006, startup validation, and a file->db importer. Follows the thread_meta store / run_events backend-switch / 0003_scheduled_tasks migration patterns; no new dependency. * fix(agents): make db storage path production-ready (review round 1) Addresses review feedback on the db/sync agent-storage path: - sql.py: mirror the async engine's per-connection SQLite PRAGMAs on the sync engine (busy_timeout=30000, synchronous=NORMAL, foreign_keys=ON, WAL) so both engines behave identically against the shared DB; guard the engine cache with a lock (double-checked) so concurrent first-touch cannot build duplicate engines or register the connect listener twice. - routers/agents.py + routers/assistants_compat.py: offload the sync-store reads that ran on the event loop (list/get/check, update's pre-read + legacy guard + refresh, and assistants_compat's four list routes) via asyncio.to_thread — on db+postgres each was a network round trip stalling the loop. Writes were already offloaded. - file.py: translate the create() mkdir(exist_ok=False) race FileExistsError into AgentExistsError (router 409, matching SqlAgentStore's IntegrityError path); correct the _write docstring — per-file atomic replace, two commits sequential not transactional. Tests: sync-engine PRAGMA + engine-cache reuse assertions; file create-race -> AgentExistsError; strict Blockbuster anchor over the read endpoints so a regression back onto the loop fails CI. * fix(agents): address round-2 review on the db store path - update_agent tool: align the docstring/inline comment with FileAgentStore._write. Cross-field write atomicity is db-only; the file backend commits config then soul via two sequential os.replace (a crash between them can leave a fresh config.yaml beside a stale SOUL.md). The dropped partial-write *reporting* is an intentional tradeoff — the stage-then-replace safety is preserved (test_update_agent_soul_failure_does_not_replace_config still holds). - SqlAgentStore.update(): true upsert. Catch IntegrityError on the insert-on-missing branch, re-fetch and apply, so two concurrent first-time writes (e.g. two setup_agent handshakes) converge instead of surfacing a raw UNIQUE(user_id, name) violation as a 500. Symmetric with create(). - get_agent_store(): document the graph-subprocess config-resolution invariant (the except->file fallback is a genuine no-config path, not a mask for a misconfigured graph process) and pin it with two tests driving the real get_app_config() file resolution: db resolves from an on-disk config.yaml, file fallback when config is unresolvable. * test(agents): cover SqlAgentStore.update() write-race upsert recovery Mandatory-TDD test for the round-2 fix in 0680340a: two concurrent first-time update()s where the loser's insert hits UNIQUE(user_id, name). Deterministically forces the IntegrityError recovery path by making the first _row probe miss the committed winner, and asserts last-writer-wins instead of a surfaced 500.
95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Custom agent definition persistence — abstract store + file/db backends.
|
|
|
|
The public entry point is :func:`get_agent_store`, which the free functions in
|
|
:mod:`deerflow.config.agents_config` dispatch to. ``file`` (default) preserves
|
|
today's on-disk layout; ``db`` shares definitions across nodes via the SQL
|
|
persistence layer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from deerflow.persistence.agents.base import (
|
|
AgentDeleteOutcome,
|
|
AgentExistsError,
|
|
AgentStore,
|
|
parse_agent_config,
|
|
)
|
|
from deerflow.persistence.agents.model import AgentRow
|
|
|
|
if TYPE_CHECKING:
|
|
from deerflow.config.app_config import AppConfig
|
|
|
|
__all__ = [
|
|
"AgentDeleteOutcome",
|
|
"AgentExistsError",
|
|
"AgentRow",
|
|
"AgentStore",
|
|
"get_agent_store",
|
|
"make_agent_store",
|
|
"parse_agent_config",
|
|
]
|
|
|
|
_file_store_singleton: AgentStore | None = None
|
|
|
|
|
|
def make_agent_store(config: AppConfig) -> AgentStore:
|
|
"""Build (or reuse) the store selected by ``config.agent_storage.backend``.
|
|
|
|
``db`` requires ``database.backend`` to be ``sqlite`` or ``postgres``; a
|
|
``memory`` database has no durable URL and is rejected here (the gateway
|
|
also fails fast at startup, but this guard covers the graph-process path).
|
|
"""
|
|
if config.agent_storage.backend == "db":
|
|
db_backend = config.database.backend
|
|
if db_backend not in ("sqlite", "postgres"):
|
|
raise ValueError(
|
|
f"agent_storage.backend='db' requires database.backend to be 'sqlite' or 'postgres', "
|
|
f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot "
|
|
"share agent definitions across nodes; set database.backend accordingly or use "
|
|
"agent_storage.backend='file'."
|
|
)
|
|
from deerflow.persistence.agents.sql import SqlAgentStore
|
|
|
|
return SqlAgentStore(config.database.app_sync_sqlalchemy_url)
|
|
|
|
return _file_store()
|
|
|
|
|
|
def get_agent_store() -> AgentStore:
|
|
"""Return the store for the current process's configuration.
|
|
|
|
Defaults to the file backend when no app config can be resolved — the free
|
|
functions in ``agents_config`` must keep working in lightweight contexts
|
|
(CLI, tests, tools) that never load a full ``config.yaml``. Only an
|
|
explicit ``agent_storage.backend: db`` diverges from the file default.
|
|
|
|
Cross-process invariant (the ``db`` backend's whole point): the per-run
|
|
agent build runs in the **graph subprocess**, a different process from the
|
|
gateway. Its cross-node guarantee holds only because ``get_app_config()``
|
|
resolves ``config.yaml`` there too and returns ``backend: db`` — so the read
|
|
path in the graph process sees the same shared table the gateway wrote. The
|
|
``except`` below is a genuine *no resolvable config* fallback (CLI/tests),
|
|
**not** a mask for a misconfigured graph process: if ``config.yaml`` is
|
|
reachable there (it is, same working tree), ``db`` is honoured, not silently
|
|
downgraded to node-local ``file``. Pinned by
|
|
``test_get_agent_store_resolves_db_backend_from_on_disk_config``.
|
|
"""
|
|
from deerflow.config.app_config import get_app_config
|
|
|
|
try:
|
|
config = get_app_config()
|
|
except Exception: # noqa: BLE001 — no resolvable config → file default
|
|
return _file_store()
|
|
return make_agent_store(config)
|
|
|
|
|
|
def _file_store() -> AgentStore:
|
|
global _file_store_singleton
|
|
if _file_store_singleton is None:
|
|
from deerflow.persistence.agents.file import FileAgentStore
|
|
|
|
_file_store_singleton = FileAgentStore()
|
|
return _file_store_singleton
|