mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
* 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.
153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
"""Assistants compatibility endpoints.
|
|
|
|
Provides LangGraph Platform-compatible assistants API backed by the
|
|
``langgraph.json`` graph registry and ``config.yaml`` agent definitions.
|
|
|
|
This is a minimal stub that satisfies the ``useStream`` React hook's
|
|
initialization requirements (``assistants.search()`` and ``assistants.get()``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/assistants", tags=["assistants-compat"])
|
|
|
|
|
|
class AssistantResponse(BaseModel):
|
|
assistant_id: str
|
|
graph_id: str
|
|
name: str
|
|
config: dict[str, Any] = Field(default_factory=dict)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
description: str | None = None
|
|
created_at: str = ""
|
|
updated_at: str = ""
|
|
version: int = 1
|
|
|
|
|
|
class AssistantSearchRequest(BaseModel):
|
|
graph_id: str | None = None
|
|
name: str | None = None
|
|
metadata: dict[str, Any] | None = None
|
|
limit: int = 10
|
|
offset: int = 0
|
|
|
|
|
|
def _get_default_assistant() -> AssistantResponse:
|
|
"""Return the default lead_agent assistant."""
|
|
now = datetime.now(UTC).isoformat()
|
|
return AssistantResponse(
|
|
assistant_id="lead_agent",
|
|
graph_id="lead_agent",
|
|
name="lead_agent",
|
|
config={},
|
|
metadata={"created_by": "system"},
|
|
description="DeerFlow lead agent",
|
|
created_at=now,
|
|
updated_at=now,
|
|
version=1,
|
|
)
|
|
|
|
|
|
def _list_assistants() -> list[AssistantResponse]:
|
|
"""List all available assistants from config."""
|
|
assistants = [_get_default_assistant()]
|
|
|
|
# Also include custom agents from config.yaml agents directory
|
|
try:
|
|
from deerflow.config.agents_config import list_custom_agents
|
|
|
|
for agent_cfg in list_custom_agents():
|
|
now = datetime.now(UTC).isoformat()
|
|
assistants.append(
|
|
AssistantResponse(
|
|
assistant_id=agent_cfg.name,
|
|
graph_id="lead_agent", # All agents use the same graph
|
|
name=agent_cfg.name,
|
|
config={},
|
|
metadata={"created_by": "user"},
|
|
description=agent_cfg.description or "",
|
|
created_at=now,
|
|
updated_at=now,
|
|
version=1,
|
|
)
|
|
)
|
|
except Exception:
|
|
logger.debug("Could not load custom agents for assistants list")
|
|
|
|
return assistants
|
|
|
|
|
|
@router.post("/search", response_model=list[AssistantResponse])
|
|
async def search_assistants(body: AssistantSearchRequest | None = None) -> list[AssistantResponse]:
|
|
"""Search assistants.
|
|
|
|
Returns all registered assistants (lead_agent + custom agents from config).
|
|
"""
|
|
# _list_assistants reads custom agents through the sync agent store (file IO
|
|
# or a DB round trip), so keep it off the event loop.
|
|
assistants = await asyncio.to_thread(_list_assistants)
|
|
|
|
if body and body.graph_id:
|
|
assistants = [a for a in assistants if a.graph_id == body.graph_id]
|
|
if body and body.name:
|
|
assistants = [a for a in assistants if body.name.lower() in a.name.lower()]
|
|
|
|
offset = body.offset if body else 0
|
|
limit = body.limit if body else 10
|
|
return assistants[offset : offset + limit]
|
|
|
|
|
|
@router.get("/{assistant_id}", response_model=AssistantResponse)
|
|
async def get_assistant_compat(assistant_id: str) -> AssistantResponse:
|
|
"""Get an assistant by ID."""
|
|
for a in await asyncio.to_thread(_list_assistants):
|
|
if a.assistant_id == assistant_id:
|
|
return a
|
|
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
|
|
|
|
|
|
@router.get("/{assistant_id}/graph")
|
|
async def get_assistant_graph(assistant_id: str) -> dict:
|
|
"""Get the graph structure for an assistant.
|
|
|
|
Returns a minimal graph description. Full graph introspection is
|
|
not supported in the Gateway — this stub satisfies SDK validation.
|
|
"""
|
|
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
|
|
if not found:
|
|
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
|
|
|
|
return {
|
|
"graph_id": "lead_agent",
|
|
"nodes": [],
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@router.get("/{assistant_id}/schemas")
|
|
async def get_assistant_schemas(assistant_id: str) -> dict:
|
|
"""Get JSON schemas for an assistant's input/output/state.
|
|
|
|
Returns empty schemas — full introspection not supported in Gateway.
|
|
"""
|
|
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
|
|
if not found:
|
|
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
|
|
|
|
return {
|
|
"graph_id": "lead_agent",
|
|
"input_schema": {},
|
|
"output_schema": {},
|
|
"state_schema": {},
|
|
"config_schema": {},
|
|
}
|