This commit is contained in:
吉鑫升 committed 2026-06-04 14:28:22 +08:00
commit 9704eff9f7
42 files changed
+1973 -10517

No files matched your search

@@ -261,12 +261,22 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
**HTML page tasks (`task_kind == "html_page"`)**: a second task kind that renders a full HTML document each run instead of a Markdown report. The kind + its config (`html_config`: `page_prompt`, `negative_prompt`, `reference_favorite_id`, `model_name`) live inside `execution_context` — no schema change to `scheduled_tasks`. Data gathering is delegated to the agent's own search skills (driven by `page_prompt`), so there is no explicit search-query/data-source field; the render model is user-selectable (`model_name`, default `zai-org/GLM-5-FP8`). `_run_task_attempt()` dispatches by kind; `_run_html_page_attempt()` runs the pipeline: gather real data via the lead agent (its search tools) → optionally load a reference favorite's reusable style/layout → render with the chosen model (thinking off) → fault-tolerant JSON parse → sanitize (BeautifulSoup, strips `<script>`/`<iframe>`/`on*`/`javascript:`) → structural validate → one GLM5 repair pass if invalid → **completeness gate**: if the HTML is still not a full valid document (missing doctype/html/head/body, empty body, or residual scripts) after the repair pass, raise `HtmlGenerationError` so the run is marked **failed** (a page only counts as successful when it is a complete, previewable HTML document) → best-effort Playwright screenshot (no-op when Playwright absent) → archive `index.html` → persist a `scheduled_html_pages` row → return `result` with `format: "html"` and the full HTML inline. Pure helpers live in `deerflow/runtime/scheduler/html_page.py`. The frontend previews the HTML in a sandboxed `iframe` (no `allow-scripts`) on the run-detail and public pages; a user can bookmark a page (style summary extracted via GLM5) and later pick it as a style reference. New tables `scheduled_html_pages` + `html_page_favorites` (`deerflow/persistence/html_pages/`, wired as `app.state.html_page_store`) are created by `Base.metadata.create_all` with no migration. Tests: `tests/test_html_page_tasks.py`.
**Unified output switches (`require_html` + `require_markdown`).** The task form no longer splits "普通任务 / HTML 页面" into separate kinds. `execution_context` now carries two **independent** booleans that may both be on: `require_html` (必须生成 HTML) and `require_markdown` (必须生成 Markdown). `_task_requires_html()` returns true for `require_html` **or** legacy `task_kind == "html_page"` (so old tasks need no data migration), and `_run_task_attempt()` dispatches to `_run_html_page_attempt` when it is true, else the regular agent attempt. When **both** switches are on, `_gather_html_data()` returns `(text, files)` — the gather agent run is additionally told to save a `.md` file — and `_run_html_page_attempt` validates it (`MarkdownRequirementError` on failure) and surfaces it alongside the generated page, so the run delivers both artifacts. The single main `prompt` doubles as the HTML page requirement (`html_config.page_prompt`); there is no separate page-prompt field. Tests: `tests/test_scheduled_task_require_html.py` + the markdown-coexistence cases in `tests/test_html_page_tasks.py`.
### Per-user Prompts & UI Preferences
**Custom prompts can be scoped to one agent.** `user_custom_prompts` gains a nullable `agent_id` column (migration `20260603_02`): `NULL` = global (the chip shows in every chat), otherwise the prompt only surfaces when that agent is selected. The visibility filter is applied in the frontend `CustomPromptBar` (it receives the current agent id from `ChatPage`'s selector or `AgentChatPage`'s `assistantId`); the backend (`/api/user-prompts` CRUD) just stores/returns/clears `agent_id` (update accepts an explicit `null` to clear). Tests: `tests/test_user_prompts_agent_id.py`.
**Per-user UI preferences** (`deerflow/persistence/user_preferences/`, wired as `app.state.user_preferences_store`; migration `20260603_03`, also auto-created by `create_all`): one JSON-blob row per user so new toggles need no schema change. Router `/api/user-preferences` (`GET`/`PUT`, account-scoped). Current key: `hide_admin_prompt_prefix` — when true the chat input hides the admin-configured prompt-prefix switch and never applies the prefix (the user uses only their own prompts; enforced in `ChatPage` for both `shouldApplyPrefix` and `promptPrefixProp`). Toggled from the "管理常用提示词" dialog. Tests: `tests/test_user_preferences.py`.
**HTML skills (tools the render step uses).** Two offline skills under `skills/public/` shape the generated HTML and are toggled in the skills settings page (enabled by default): `html-page-builder` (a self-contained inline-CSS design system — colors/cards/tables/KPI/responsive) and `html-charts` (inline-SVG charts that render in the no-`allow-scripts` preview iframe; optional local ECharts documented in its `assets/README.md`). A skill opts into HTML injection via `metadata.kind: html_page` in its SKILL.md frontmatter; `_collect_html_skill_instructions()` reads the bodies of **enabled** such skills and `build_generation_prompt(skill_instructions=…)` injects them, so the (raw-model) render step honors the user's skill selection. The data-gathering step and the render/repair/style passes all use the task's selected `model_name` (no silent fallback to the default model — and there is **no hardcoded GLM-5**; an unset selection falls back to config `models[0]`). To stop auxiliary LLM calls from leaking onto the default model, **`TitleMiddleware` is skipped when `is_scheduled_run` is set** (`_build_middlewares` in `lead_agent/agent.py`) — scheduled/HTML runs use throwaway threads whose titles are never shown, and `config.title.model_name` is usually unset (→ would call `models[0]`, not the chosen model). The completeness gate (`validate_html`) also requires a non-empty `<title>` and forbids external stylesheets, on top of the doctype/html/head/body + non-empty-body + no-script checks.
### Custom Agent Persistence
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
**Listing cache + pagination** — to keep `GET /api/agents` fast, the gallery's per-card extras (skills + tool_groups) are denormalized into two cache tables instead of reading every agent's `config.yaml` on each request: `agent_skills` (relational, one row per (agent, skill) with `position`) and `agent_extras` (one row per agent: `tool_groups` JSON + `model` + `skills_synced_at`). `config.yaml` remains authoritative for the agent runtime; these tables are a read cache, write-through on create/update and backfilled from disk by `_sync_legacy_agents` (mtime-vs-`skills_synced_at` drift check). The list endpoint enriches via `AgentStore.get_extras_for([ids])` (two batched `IN` queries). Both tables are auto-created by `Base.metadata.create_all` (no migration needed). `GET /api/agents` supports **opt-in server-side pagination**: `scope` (`mine` = owned + favorited / `square` = built-in/published public), `page`, `page_size`, returning `{agents, total, page, page_size}`; omitting `scope`/`page` preserves the legacy "return everything" behavior. Name `search` matches the raw OR desensitized name (the frontend's static term→abbrev map is mirrored server-side in `deerflow/config/agent_name_desensitize.py`); `tag_ids` is an SQL OR-filter. `POST /api/agents/private-square/list` takes the same pagination params. Repository entry point: `AgentStore.list_paginated(...)`.
**Homepage selector (agent picker on the new-chat page)** — single admin-curated list backed by `agents.featured_order` (nullable BIGINT). NULL = not featured. Only `published = true` agents can be featured (router enforces on writes); any published agent is eligible regardless of ownership. Cleared with `order: null`. The list rendered to end users is `WHERE featured_order IS NOT NULL` + visibility-filter, sorted by `featured_order ASC`. Admins reorder by sending the desired ids to `PUT /api/agents/featured/order` (assigns indices 0..N-1). BIGINT is used rather than INT so the management UI can seed initial values from `Date.now()` (overflows INT on MySQL).
### Portable Column Types (`deerflow/persistence/types.py`)
@@ -57,6 +57,7 @@ from app.gateway.routers import (
thread_shares,
threads,
uploads,
user_preferences,
user_prompts,
writing,
)
@@ -857,6 +858,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
app.include_router(user_prompts.router)
# Per-user UI preferences at /api/user-preferences
app.include_router(user_preferences.router)
# Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
app.include_router(roundtable_drafts.router)
@@ -160,6 +160,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.user_prompt_store = make_user_prompt_store(sf)
from deerflow.persistence.user_preferences import make_user_preferences_store
app.state.user_preferences_store = make_user_preferences_store(sf)
from deerflow.persistence.roundtable_drafts import make_roundtable_draft_store
app.state.roundtable_draft_store = make_roundtable_draft_store(sf)
@@ -66,9 +66,18 @@ class AgentResponse(BaseModel):
class AgentsListResponse(BaseModel):
"""Response model for listing all visible agents."""
"""Response model for listing agents.
``total`` / ``page`` / ``page_size`` describe server-side pagination. When
a caller does not opt into pagination they default to "everything on one
page" (``total == len(agents)``, ``page == 1``), so existing unpaginated
callers keep working unchanged.
"""
agents: list[AgentResponse]
total: int = Field(default=0, description="Total agents matching the query across all pages")
page: int = Field(default=1, description="1-based page index of this response")
page_size: int = Field(default=0, description="Page size; 0 means the result was not paginated")
class AgentCreateRequest(BaseModel):
@@ -215,6 +224,7 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
handlers as a safety net for environments that bypassed the lifespan hook.
"""
try:
pairs: list[tuple[str, Any]] = []
for agent_cfg in list_custom_agents():
agent_id = agent_cfg.id or agent_cfg.name
if not agent_id:
@@ -227,10 +237,41 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
"published": True,
}
)
pairs.append((agent_id, agent_cfg))
await _backfill_listing_cache(store, pairs)
except Exception:
logger.exception("Failed to sync legacy folder-backed agents")
async def _backfill_listing_cache(store: AgentStore, pairs: list[tuple[str, Any]]) -> None:
"""Mirror each agent's config.yaml skills/tool_groups/model into the DB cache.
The list endpoint reads skills/tool_groups from these cache rows instead of
stat+parsing every ``config.yaml`` per request. We fetch the current cache
in one batch and only re-write rows whose values actually changed, so a
steady-state sync (every ``_LEGACY_SYNC_TTL_SECONDS``) is cheap and does not
churn DELETE/INSERTs. For the cache, ``None`` and ``[]`` skills are
equivalent ("no skills") — the tri-state only matters to the agent runtime,
which reads config.yaml directly.
"""
if not pairs:
return
cached = await store.get_extras_for([aid for aid, _ in pairs])
for agent_id, cfg in pairs:
want_skills = [s for s in (cfg.skills or []) if isinstance(s, str)]
want_tool_groups = cfg.tool_groups
want_model = cfg.model
existing = cached.get(agent_id)
if existing is None or existing.get("skills") != want_skills:
await store.replace_skills(agent_id, want_skills)
if (
existing is None
or existing.get("tool_groups") != want_tool_groups
or existing.get("model") != want_model
):
await store.set_extras(agent_id, tool_groups=want_tool_groups, model=want_model)
# 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
@@ -403,6 +444,38 @@ async def _get_editable_or_raise(
raise HTTPException(status_code=403, detail="You can only modify agents that you created")
async def _build_list_responses(
store: AgentStore,
tag_store: TagStore,
user_id: str,
records: list[dict[str, Any]],
) -> list[AgentResponse]:
"""Build gallery responses for a set of agent rows WITHOUT touching disk.
skills / tool_groups / model come from the denormalized DB cache
(``get_extras_for`` — one batched query) instead of stat+parsing every
agent's ``config.yaml`` per request, which was the list-endpoint hot spot.
SOUL.md is skipped (only the edit dialog reads it). An agent absent from
the cache (never mirrored) keeps its ``None`` extras until the next sync.
"""
responses = [
_record_to_response(a, include_soul=False, include_extras=False) for a in records
]
ids = [r.id for r in responses]
extras = await store.get_extras_for(ids)
assignments = await tag_store.list_assignments("agent", ids)
favorite_ids = await store.list_favorite_ids(user_id)
for resp in responses:
cached = extras.get(resp.id)
if cached is not None:
resp.skills = cached.get("skills")
resp.tool_groups = cached.get("tool_groups")
resp.model = cached.get("model")
resp.tags = assignments.get(resp.id, [])
resp.is_favorite = resp.id in favorite_ids
return responses
@router.get(
"/agents",
response_model=AgentsListResponse,
@@ -411,8 +484,15 @@ async def _get_editable_or_raise(
)
async def list_agents(
request: Request,
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on agent name"),
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on agent name (raw or desensitized display form)"),
tag_ids: list[str] | None = Query(default=None, description="Keep agents carrying any of these tags (OR)"),
scope: str | None = Query(
default=None,
pattern="^(mine|square)$",
description="Pagination scope. 'mine' = owned + favorited; 'square' = built-in/published public. Omit for the default combined visibility list.",
),
page: int | None = Query(default=None, ge=1, description="1-based page index. Providing this (or 'scope') enables server-side pagination."),
page_size: int = Query(default=20, ge=1, le=200, description="Items per page when paginating."),
admin: bool = Query(
default=False,
description=(
@@ -429,38 +509,36 @@ async def list_agents(
tag_store: TagStore = get_tag_store(request)
await _maybe_sync_legacy_agents(store)
user_id = _current_user_id(request)
if admin:
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can list all agents")
agents = await store.list_all()
else:
agents = await store.list_visible(user_id)
# Load lightweight extras (skills / model / tool_groups from config.yaml)
# so the gallery cards can render skill badges. SOUL.md is still skipped
# (only the edit dialog reads it). The per-agent config.yaml reads run in
# worker threads concurrently, so the event loop stays responsive even
# with many agents.
responses = list(
await asyncio.gather(
*(
_record_to_response_async(a, include_soul=False, include_extras=True)
for a in agents
)
)
)
if admin and not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can list all agents")
assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
favorite_ids = await store.list_favorite_ids(user_id)
for resp in responses:
resp.tags = assignments.get(resp.id, [])
resp.is_favorite = resp.id in favorite_ids
paginate = scope is not None or page is not None
if paginate:
eff_page = page or 1
rows, total = await store.list_paginated(
user_id=user_id,
scope=scope,
search=search,
tag_ids=tag_ids,
page=eff_page,
page_size=page_size,
admin=admin,
)
responses = await _build_list_responses(store, tag_store, user_id, rows)
return AgentsListResponse(agents=responses, total=total, page=eff_page, page_size=page_size)
# Legacy unpaginated path — returns everything (admin console, callers
# that have not migrated to pagination). Extras now come from the DB
# cache too, so this path no longer reads config.yaml per agent.
agents = await store.list_all() if admin else await store.list_visible(user_id)
responses = await _build_list_responses(store, tag_store, user_id, agents)
if search:
needle = search.strip().lower()
responses = [r for r in responses if needle in r.name.lower()]
if tag_ids:
wanted = set(tag_ids)
responses = [r for r in responses if any(t.get("id") in wanted for t in r.tags)]
return AgentsListResponse(agents=responses)
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
except HTTPException:
raise
except Exception as e:
@@ -587,6 +665,10 @@ async def create_agent_endpoint(request: AgentCreateRequest, http_request: Reque
"is_private": request.is_private,
}
)
# Mirror skills/tool_groups/model into the listing cache so the gallery
# can render this agent's badges without reading its config.yaml.
await store.replace_skills(agent_id, request.skills)
await store.set_extras(agent_id, tool_groups=request.tool_groups, model=request.model)
logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir)
return await _record_to_response_async(record, include_soul=True, include_extras=True)
@@ -653,6 +735,9 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
model=next_model,
tool_groups=next_tool_groups,
)
# Keep the listing cache in lock-step with config.yaml on every write.
await store.replace_skills(agent_id, next_skills)
await store.set_extras(agent_id, tool_groups=next_tool_groups, model=next_model)
if request.soul is not None:
_write_agent_soul(agent_id, request.soul)
@@ -766,6 +851,10 @@ async def delete_agent(agent_id: str, request: Request) -> None:
await store.delete_favorites_for_agent(agent_id)
except Exception: # noqa: BLE001 — favorite cleanup must not block deletion
pass
try:
await store.delete_extras_for_agent(agent_id)
except Exception: # noqa: BLE001 — cache cleanup must not block deletion
pass
logger.info("Deleted agent '%s' from %s", agent_id, agent_dir)
except HTTPException:
raise
@@ -868,6 +957,10 @@ class PrivateSquareListRequest(BaseModel):
"""Request body for listing all private-square agents."""
password: str = Field(..., description="Plain-text private-square password. Verified against the admin-configured hash before the listing is returned.")
search: str | None = Field(default=None, description="Case-insensitive fuzzy match on agent name (raw or desensitized).")
tag_ids: list[str] | None = Field(default=None, description="Keep agents carrying any of these tags (OR).")
page: int | None = Field(default=None, ge=1, description="1-based page index. Providing this enables server-side pagination.")
page_size: int = Field(default=20, ge=1, le=200, description="Items per page when paginating.")
class AgentPrivateUpdateRequest(BaseModel):
@@ -899,14 +992,26 @@ async def list_private_square(body: PrivateSquareListRequest, request: Request)
try:
store = get_agent_store(request)
tag_store: TagStore = get_tag_store(request)
await _sync_legacy_agents(store)
records = await store.list_private()
responses = [_record_to_response(r, include_soul=True) for r in records]
await _maybe_sync_legacy_agents(store)
# user_id is unused by the "private" scope filter (the password is the
# gate), but list_paginated requires it for its signature.
user_id = _current_user_id(request)
assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
for resp in responses:
resp.tags = assignments.get(resp.id, [])
return AgentsListResponse(agents=responses)
if body.page is not None:
rows, total = await store.list_paginated(
user_id=user_id,
scope="private",
search=body.search,
tag_ids=body.tag_ids,
page=body.page,
page_size=body.page_size,
)
responses = await _build_list_responses(store, tag_store, user_id, rows)
return AgentsListResponse(agents=responses, total=total, page=body.page, page_size=body.page_size)
records = await store.list_private()
responses = await _build_list_responses(store, tag_store, user_id, records)
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
except HTTPException:
raise
except Exception as e:
@@ -0,0 +1,66 @@
"""Per-user UI preferences (account-bound, follow the user across devices).
Currently exposes a single toggle: ``hide_admin_prompt_prefix`` — when true the
chat input hides the admin-configured prompt-prefix switch so the user only uses
their own prompts. Kept as a small key-value (JSON) store so new toggles do not
each need a schema change.
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/user-preferences", tags=["user-preferences"])
class UserPreferencesResponse(BaseModel):
hide_admin_prompt_prefix: bool = False
class UserPreferencesUpdateRequest(BaseModel):
hide_admin_prompt_prefix: bool | None = Field(default=None)
def _current_user_id(request: Request) -> str:
user = getattr(request.state, "user", None)
if user is not None:
return str(user.id)
return get_effective_user_id()
def _get_store(request: Request):
store = getattr(request.app.state, "user_preferences_store", None)
if store is None:
raise HTTPException(status_code=503, detail="User preferences store not available")
return store
def _to_response(prefs: dict) -> UserPreferencesResponse:
return UserPreferencesResponse(
hide_admin_prompt_prefix=bool(prefs.get("hide_admin_prompt_prefix", False)),
)
@router.get("", response_model=UserPreferencesResponse)
async def get_user_preferences(request: Request) -> UserPreferencesResponse:
store = _get_store(request)
user_id = _current_user_id(request)
prefs = await store.get_for_user(user_id)
return _to_response(prefs)
@router.put("", response_model=UserPreferencesResponse)
async def update_user_preferences(request: Request, body: UserPreferencesUpdateRequest) -> UserPreferencesResponse:
store = _get_store(request)
user_id = _current_user_id(request)
patch = body.model_dump(exclude_unset=True, exclude_none=True)
if not patch:
return _to_response(await store.get_for_user(user_id))
prefs = await store.merge_for_user(user_id, patch)
return _to_response(prefs)
@@ -24,6 +24,8 @@ class UserPromptResponse(BaseModel):
title: str = ""
content: str
sort_order: int = 0
# Optional single-agent association. NULL = global (shown for every agent).
agent_id: str | None = None
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
@@ -36,12 +38,16 @@ class UserPromptCreateRequest(BaseModel):
title: str = Field(default="", max_length=255)
content: str = Field(..., min_length=1)
sort_order: int = Field(default=0)
agent_id: str | None = Field(default=None, max_length=64)
class UserPromptUpdateRequest(BaseModel):
title: str | None = Field(default=None, max_length=255)
content: str | None = Field(default=None, min_length=1)
sort_order: int | None = None
# Sent on every save; an empty string / null clears the association. Use a
# sentinel so "field omitted" (don't touch) differs from "set to null".
agent_id: str | None = Field(default=None, max_length=64)
def _current_user_id(request: Request) -> str:
@@ -77,6 +83,7 @@ async def create_user_prompt(request: Request, body: UserPromptCreateRequest) ->
"title": (body.title or "").strip(),
"content": body.content,
"sort_order": body.sort_order,
"agent_id": (body.agent_id or "").strip() or None,
},
)
return UserPromptResponse(**row)
@@ -93,6 +100,11 @@ async def update_user_prompt(request: Request, prompt_id: str, body: UserPromptU
data["content"] = body.content
if body.sort_order is not None:
data["sort_order"] = body.sort_order
# ``agent_id`` is nullable: only act when the caller actually sent the field
# (``model_fields_set``), so an explicit null clears the association while an
# omitted field leaves it untouched.
if "agent_id" in body.model_fields_set:
data["agent_id"] = (body.agent_id or "").strip() or None
row = await store.update(prompt_id, user_id, data)
if row is None:
raise HTTPException(status_code=404, detail="Prompt not found")
@@ -0,0 +1,92 @@
"""Server-side mirror of the frontend agent-name desensitization map.
The web UI displays agent names through a static term→abbreviation map
(``frontend-web/src/lib/desensitize.ts``) — e.g. the stored name "态势分析" is
shown as "TS分析". The gallery search box therefore matches against the
*desensitized* display form, not the raw stored name.
When search moves server-side (paginated listing), the backend must match the
same way, so this module mirrors that static map verbatim. The frontend map is
never extended at runtime (``addDesensitizeEntries`` is defined but never
called), so a static copy stays in sync.
Used by the SQL listing query to match "raw name OR desensitized name".
"""
from __future__ import annotations
# Mirror of DESENSITIZE_MAP in frontend-web/src/lib/desensitize.ts.
# Keep entries identical to the frontend so server-side and client-side search
# behave the same. Order does not affect membership tests.
DESENSITIZE_MAP: dict[str, str] = {
"态势": "TS",
"你好": "nh",
"滨海作战体系": "第一张网",
"滨海作战部队": "第一张网",
"Taiwan Air Force Flight Units": "第二张网",
"台海空军飞行部队": "第二张网",
"空军": "KJ",
"海军": "HJ",
"陆军": "LJ",
"陆战队": "LZD",
"海巡": "HX",
"憲兵": "XB",
"資通電軍": "ZTDJ",
"情报": "QB",
"指挥": "ZH",
"信息准备": "XX准备",
"行为体": "XWT",
"驱动因素": "QDYS",
"叙事": "XS",
"前三定": "QSD",
"受众目标": "SZMB",
"作战": "ZZ",
"目的": "MD",
"关键传播节点": "GJCBJD",
"战法": "ZF",
"行动": "XD",
"打击": "DJ",
"目标": "MB",
"战力": "ZL",
"筹划": "COH",
"策划": "CEH",
"模拟": "MN",
"推演": "TY",
"认知": "RZ",
"触及": "CJ",
"军": "J",
"美军": "MJ",
"台军": "TJ",
}
def desensitize_name(name: str) -> str:
"""Return ``name`` with every mapped term replaced by its abbreviation.
Mirrors the frontend ``desensitize()`` — a sequence of literal replacements.
"""
out = name
for term, abbr in DESENSITIZE_MAP.items():
if term in out:
out = out.replace(term, abbr)
return out
def name_matches(name: str, needle: str) -> bool:
"""True when ``needle`` (case-insensitive) is in the raw OR desensitized name."""
n = needle.strip().lower()
if not n:
return True
return n in name.lower() or n in desensitize_name(name).lower()
def source_terms_for_needle(needle: str) -> list[str]:
"""Return the raw source terms whose abbreviation contains ``needle``.
Lets the SQL search add ``name LIKE %term%`` clauses so a user typing an
abbreviation (e.g. "TS") still matches the stored raw term (e.g. "态势").
"""
n = needle.strip().lower()
if not n:
return []
return [term for term, abbr in DESENSITIZE_MAP.items() if n in abbr.lower()]
@@ -4,13 +4,20 @@ from __future__ import annotations
from deerflow.persistence.agents.base import AgentStore
from deerflow.persistence.agents.memory import MemoryAgentStore
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.agents.sql import AgentRepository
__all__ = [
"AgentExtrasRow",
"AgentFavoriteRow",
"AgentRepository",
"AgentRow",
"AgentSkillRow",
"AgentStore",
"MemoryAgentStore",
"make_agent_store",
@@ -118,6 +118,68 @@ class AgentStore(ABC):
"""
raise NotImplementedError
@abstractmethod
async def list_paginated(
self,
*,
user_id: str,
scope: str | None,
search: str | None,
tag_ids: list[str] | None,
page: int,
page_size: int,
admin: bool = False,
) -> tuple[list[dict[str, Any]], int]:
"""Return one page of agents plus the total matching count.
``scope``:
- ``"mine"`` — owned by ``user_id`` OR favorited by ``user_id``.
- ``"square"`` — built-in or published, and not private-square.
- ``"private"``— flagged ``is_private=True`` (caller gates access).
- ``None`` — the default listing visibility (``list_visible``).
``admin=True`` bypasses the visibility/scope filter entirely (every
agent), but still honors ``search`` / ``tag_ids``. Caller authorizes.
``search`` matches the raw name OR its desensitized display form.
``tag_ids`` keeps agents carrying ANY of the tags (OR). Both filters,
the COUNT, and LIMIT/OFFSET all execute in SQL.
Returns ``(rows, total)`` where ``rows`` is at most ``page_size`` items
for the 1-based ``page``.
"""
raise NotImplementedError
# --- Listing cache (skills / tool_groups / model) ---
@abstractmethod
async def replace_skills(self, agent_id: str, skills: list[str] | None) -> None:
"""Replace the cached skill list for ``agent_id``.
``None`` and ``[]`` both clear the cache to "no skills" (mirrors the
tri-state semantics of ``AgentConfig.skills``).
"""
raise NotImplementedError
@abstractmethod
async def set_extras(self, agent_id: str, *, tool_groups: list[str] | None, model: str | None) -> None:
"""Upsert the cached ``tool_groups`` / ``model`` for ``agent_id``."""
raise NotImplementedError
@abstractmethod
async def get_extras_for(self, agent_ids: list[str]) -> dict[str, dict[str, Any]]:
"""Batch-fetch cached listing extras for ``agent_ids``.
Returns ``{agent_id: {"skills": list[str] | None, "tool_groups":
list[str] | None, "model": str | None}}``. An agent that was never
mirrored is absent from the mapping (caller treats it as unknown).
"""
raise NotImplementedError
@abstractmethod
async def delete_extras_for_agent(self, agent_id: str) -> None:
"""Drop the cached skills + extras rows for ``agent_id`` (on delete)."""
raise NotImplementedError
# --- Favorites ("收藏") ---
@abstractmethod
@@ -5,6 +5,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from deerflow.config.agent_name_desensitize import name_matches
from deerflow.persistence.agents.base import AgentStore
@@ -22,6 +23,9 @@ class MemoryAgentStore(AgentStore):
def __init__(self) -> None:
self._agents: dict[str, dict[str, Any]] = {}
self._favorites: set[tuple[str, str]] = set()
# Listing cache mirrors (see AgentExtrasRow / AgentSkillRow):
self._skills: dict[str, list[str]] = {}
self._extras: dict[str, dict[str, Any]] = {}
@staticmethod
def _is_visible(agent: dict[str, Any], user_id: str) -> bool:
@@ -166,6 +170,73 @@ class MemoryAgentStore(AgentStore):
self._agents.pop(agent_id, None)
return True
# --- Paginated listing ---
def _matches_scope(self, agent: dict[str, Any], *, user_id: str, scope: str | None, admin: bool) -> bool:
if admin:
return True
if scope == "square":
return (agent.get("user_id") is None or bool(agent.get("published"))) and not bool(agent.get("is_private"))
if scope == "mine":
owned = agent.get("user_id") is not None and agent.get("user_id") == user_id
favorited = (user_id, agent["id"]) in self._favorites
return owned or favorited
if scope == "private":
return bool(agent.get("is_private"))
return self._is_visible(agent, user_id)
async def list_paginated(
self,
*,
user_id: str,
scope: str | None,
search: str | None,
tag_ids: list[str] | None,
page: int,
page_size: int,
admin: bool = False,
) -> tuple[list[dict[str, Any]], int]:
# NOTE: the in-memory store has no tag-assignment data, so ``tag_ids``
# is ignored here. Tag-filter behavior is covered by the SQL repo tests.
matched = [
a
for a in self._agents.values()
if self._matches_scope(a, user_id=user_id, scope=scope, admin=admin)
and (not search or not search.strip() or name_matches(a.get("name") or "", search))
]
matched.sort(key=lambda a: (a.get("created_at") or "", a.get("id") or ""))
total = len(matched)
start = max(page - 1, 0) * page_size
page_rows = matched[start : start + page_size]
return [_copy(a) for a in page_rows], total
# --- Listing cache (skills / tool_groups / model) ---
async def replace_skills(self, agent_id: str, skills: list[str] | None) -> None:
self._skills[agent_id] = [s for s in (skills or []) if isinstance(s, str)]
async def set_extras(self, agent_id: str, *, tool_groups: list[str] | None, model: str | None) -> None:
self._extras[agent_id] = {"tool_groups": tool_groups, "model": model}
async def get_extras_for(self, agent_ids: list[str]) -> dict[str, dict[str, Any]]:
out: dict[str, dict[str, Any]] = {}
for agent_id in agent_ids:
has_extras = agent_id in self._extras
has_skills = agent_id in self._skills
if not has_extras and not has_skills:
continue
extras = self._extras.get(agent_id, {})
out[agent_id] = {
"skills": list(self._skills.get(agent_id, [])),
"tool_groups": extras.get("tool_groups"),
"model": extras.get("model"),
}
return out
async def delete_extras_for_agent(self, agent_id: str) -> None:
self._skills.pop(agent_id, None)
self._extras.pop(agent_id, None)
# --- Favorites ("收藏") ---
async def add_favorite(self, user_id: str, agent_id: str) -> None:
@@ -4,11 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import BigInteger, Boolean, String, UniqueConstraint
from sqlalchemy import BigInteger, Boolean, Integer, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class AgentRow(Base):
@@ -52,3 +52,47 @@ class AgentFavoriteRow(Base):
user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
class AgentSkillRow(Base):
"""Denormalized cache of a custom agent's selected skills.
``config.yaml`` under ``.deer-flow/agents/{id}/`` stays the source of truth
for the agent runtime (the lead-agent prompt reads it directly). This table
is a read cache so the gallery list endpoint can render skill badges with a
single batched ``IN`` query instead of stat+YAML-parsing every agent's
``config.yaml`` on each request. Written through on create/update and
backfilled from disk by ``_sync_legacy_agents``.
One row per (agent, skill); ``position`` preserves the configured order.
"""
__tablename__ = "agent_skills"
__table_args__ = (UniqueConstraint("agent_id", "skill_name", name="uq_agent_skills_agent_skill"),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
skill_name: Mapped[str] = mapped_column(String(191), nullable=False, index=True)
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class AgentExtrasRow(Base):
"""Denormalized cache of a custom agent's ``tool_groups`` / ``model``.
Companion to :class:`AgentSkillRow` — same rationale (avoid per-agent
``config.yaml`` reads in the list path). One row per agent. The presence of
a row also marks "this agent has been mirrored": the list path treats a
missing row as "unknown / not yet backfilled" (``skills=None``) so it can
be told apart from an explicitly empty skill list.
``skills_synced_at`` records when the cache was last written from disk so
the backfill can skip agents whose ``config.yaml`` has not changed.
"""
__tablename__ = "agent_extras"
agent_id: Mapped[str] = mapped_column(String(64), primary_key=True)
# list[str] | None — the agent's tool_groups whitelist from config.yaml.
tool_groups: Mapped[list | None] = mapped_column(PortableJSON(), nullable=True, default=None)
model: Mapped[str | None] = mapped_column(String(191), nullable=True, default=None)
skills_synced_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -10,8 +10,15 @@ from sqlalchemy import and_, delete, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.config.agent_name_desensitize import source_terms_for_needle
from deerflow.persistence.agents.base import AgentStore
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.tags.model import TagAssignmentRow
def _row_to_dict(row: AgentRow) -> dict[str, Any]:
@@ -233,6 +240,152 @@ class AgentRepository(AgentStore):
await session.commit()
return True
# --- Paginated listing ---
def _scope_filter(self, *, user_id: str, scope: str | None, admin: bool):
"""Build the visibility predicate for a paginated listing."""
if admin:
return None # no visibility filter — every agent
if scope == "square":
return and_(
or_(AgentRow.user_id.is_(None), AgentRow.published.is_(True)),
AgentRow.is_private.is_(False),
)
if scope == "mine":
# owned OR favorited. A correlated IN-subquery (not a JOIN) keeps a
# single row per agent even when it is both owned and favorited.
fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
return or_(AgentRow.user_id == user_id, AgentRow.id.in_(fav_subq))
if scope == "private":
return AgentRow.is_private.is_(True)
return self._listing_filter(user_id)
@staticmethod
def _search_clause(search: str):
"""Match the raw name OR (via the desensitize mirror) its display form."""
needle = search.strip().lower()
clauses = [func.lower(AgentRow.name).like(f"%{needle}%")]
for term in source_terms_for_needle(needle):
clauses.append(AgentRow.name.like(f"%{term}%"))
return or_(*clauses)
@staticmethod
def _tag_clause(tag_ids: list[str]):
subq = select(TagAssignmentRow.target_id).where(
TagAssignmentRow.target_type == "agent",
TagAssignmentRow.tag_id.in_(tag_ids),
)
return AgentRow.id.in_(subq)
async def list_paginated(
self,
*,
user_id: str,
scope: str | None,
search: str | None,
tag_ids: list[str] | None,
page: int,
page_size: int,
admin: bool = False,
) -> tuple[list[dict[str, Any]], int]:
filters = []
scope_filter = self._scope_filter(user_id=user_id, scope=scope, admin=admin)
if scope_filter is not None:
filters.append(scope_filter)
if search and search.strip():
filters.append(self._search_clause(search))
if tag_ids:
filters.append(self._tag_clause(tag_ids))
count_stmt = select(func.count()).select_from(AgentRow)
page_stmt = (
select(AgentRow)
# ``id`` is a stable tiebreaker on top of ``created_at`` so OFFSET
# paging is deterministic when bulk-seeded built-ins share a timestamp.
.order_by(AgentRow.created_at.asc(), AgentRow.id.asc())
.limit(page_size)
.offset(max(page - 1, 0) * page_size)
)
if filters:
where = and_(*filters)
count_stmt = count_stmt.where(where)
page_stmt = page_stmt.where(where)
async with self._sf() as session:
total = await session.scalar(count_stmt) or 0
result = await session.execute(page_stmt)
rows = [_row_to_dict(row) for row in result.scalars()]
return rows, total
# --- Listing cache (skills / tool_groups / model) ---
async def replace_skills(self, agent_id: str, skills: list[str] | None) -> None:
async with self._sf() as session:
await session.execute(delete(AgentSkillRow).where(AgentSkillRow.agent_id == agent_id))
for position, name in enumerate(skills or []):
if not isinstance(name, str):
continue
session.add(
AgentSkillRow(
id=uuid4().hex,
agent_id=agent_id,
skill_name=name,
position=position,
)
)
await session.commit()
async def set_extras(self, agent_id: str, *, tool_groups: list[str] | None, model: str | None) -> None:
async with self._sf() as session:
row = await session.get(AgentExtrasRow, agent_id)
now = datetime.now(UTC)
if row is None:
session.add(
AgentExtrasRow(
agent_id=agent_id,
tool_groups=tool_groups,
model=model,
skills_synced_at=now,
)
)
else:
row.tool_groups = tool_groups
row.model = model
row.skills_synced_at = now
await session.commit()
async def get_extras_for(self, agent_ids: list[str]) -> dict[str, dict[str, Any]]:
if not agent_ids:
return {}
out: dict[str, dict[str, Any]] = {}
async with self._sf() as session:
extras = await session.execute(
select(AgentExtrasRow).where(AgentExtrasRow.agent_id.in_(agent_ids))
)
for row in extras.scalars():
out[row.agent_id] = {
"skills": [], # default; filled below if skill rows exist
"tool_groups": row.tool_groups,
"model": row.model,
}
skill_rows = await session.execute(
select(AgentSkillRow.agent_id, AgentSkillRow.skill_name)
.where(AgentSkillRow.agent_id.in_(agent_ids))
.order_by(AgentSkillRow.agent_id.asc(), AgentSkillRow.position.asc())
)
for agent_id, skill_name in skill_rows.all():
entry = out.setdefault(
agent_id, {"skills": [], "tool_groups": None, "model": None}
)
entry["skills"].append(skill_name)
return out
async def delete_extras_for_agent(self, agent_id: str) -> None:
async with self._sf() as session:
await session.execute(delete(AgentSkillRow).where(AgentSkillRow.agent_id == agent_id))
await session.execute(delete(AgentExtrasRow).where(AgentExtrasRow.agent_id == agent_id))
await session.commit()
# --- Favorites ("收藏") ---
async def add_favorite(self, user_id: str, agent_id: str) -> None:
@@ -0,0 +1,45 @@
"""Add ``agent_id`` to ``user_custom_prompts`` for per-agent prompt scoping.
NULL = global (shown for every agent); otherwise the prompt only surfaces when
that agent is selected.
Revision ID: 20260603_02
Revises: 20260603_01
Create Date: 2026-06-03
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260603_02"
down_revision = "20260603_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
inspector = inspect(op.get_bind())
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_table("user_custom_prompts"):
return
if not _has_column("user_custom_prompts", "agent_id"):
op.add_column("user_custom_prompts", sa.Column("agent_id", sa.String(64), nullable=True))
op.create_index("ix_user_custom_prompts_agent_id", "user_custom_prompts", ["agent_id"])
def downgrade() -> None:
if _has_table("user_custom_prompts") and _has_column("user_custom_prompts", "agent_id"):
op.drop_index("ix_user_custom_prompts_agent_id", table_name="user_custom_prompts")
op.drop_column("user_custom_prompts", "agent_id")
@@ -0,0 +1,42 @@
"""Add ``user_preferences`` for per-user UI toggles (account-bound).
Holds a single JSON blob per user — e.g. ``hide_admin_prompt_prefix`` so a user
can choose to use only their own prompts and hide the admin-configured prefix.
Revision ID: 20260603_03
Revises: 20260603_02
Create Date: 2026-06-03
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260603_03"
down_revision = "20260603_02"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def upgrade() -> None:
if not _has_table("user_preferences"):
op.create_table(
"user_preferences",
sa.Column("user_id", sa.String(64), primary_key=True),
sa.Column("preferences", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
def downgrade() -> None:
if _has_table("user_preferences"):
op.drop_table("user_preferences")
@@ -15,7 +15,12 @@ its storage implementation lives in ``deerflow.runtime.events.store.db`` and
there is no matching entity directory.
"""
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.memory_config.model import UserMemoryConfigRow
from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
@@ -42,12 +47,15 @@ from deerflow.persistence.embed_sessions.model import EmbedSessionRow
from deerflow.persistence.thread_shares.model import ThreadShareRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
from deerflow.persistence.user.model import UserRow
from deerflow.persistence.user_preferences.model import UserPreferencesRow
from deerflow.persistence.user_prompts.model import UserCustomPromptRow
__all__ = [
"FeedbackRow",
"AgentExtrasRow",
"AgentFavoriteRow",
"AgentRow",
"AgentSkillRow",
"LlmCallMetricRow",
"NotificationRow",
"RecommendedQuestionRow",
@@ -71,5 +79,6 @@ __all__ = [
"ToolCallMetricRow",
"UserCustomPromptRow",
"UserMemoryConfigRow",
"UserPreferencesRow",
"UserRow",
]
@@ -0,0 +1,18 @@
"""Per-user UI preferences persistence."""
from __future__ import annotations
from deerflow.persistence.user_preferences.model import UserPreferencesRow
from deerflow.persistence.user_preferences.sql import UserPreferencesRepository
__all__ = [
"UserPreferencesRepository",
"UserPreferencesRow",
"make_user_preferences_store",
]
def make_user_preferences_store(session_factory):
if session_factory is None:
return None
return UserPreferencesRepository(session_factory)
@@ -0,0 +1,26 @@
"""ORM model for per-user UI preferences.
A single key-value (JSON) row per user, holding small UI-level toggles that
should follow the account across devices — e.g. whether to hide the
admin-configured prompt prefix in the chat input. Kept generic (one JSON blob)
so new toggles do not each require an ALTER TABLE.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class UserPreferencesRow(Base):
__tablename__ = "user_preferences"
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
preferences: Mapped[dict] = mapped_column(PortableJSON(), nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -0,0 +1,41 @@
"""SQLAlchemy-backed per-user preferences repository."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.user_preferences.model import UserPreferencesRow
class UserPreferencesRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
async def get_for_user(self, user_id: str) -> dict[str, Any]:
"""Return the user's stored preferences (empty dict when none)."""
async with self._sf() as session:
row = await session.get(UserPreferencesRow, user_id)
if row is None or not isinstance(row.preferences, dict):
return {}
return dict(row.preferences)
async def merge_for_user(self, user_id: str, patch: dict[str, Any]) -> dict[str, Any]:
"""Shallow-merge ``patch`` into the user's preferences and return the result."""
now = datetime.now(UTC)
async with self._sf() as session:
row = await session.get(UserPreferencesRow, user_id)
if row is None:
merged = dict(patch)
row = UserPreferencesRow(user_id=user_id, preferences=merged, created_at=now, updated_at=now)
session.add(row)
else:
merged = dict(row.preferences or {})
merged.update(patch)
# Reassign so SQLAlchemy detects the change on the JSON column.
row.preferences = merged
row.updated_at = now
await session.commit()
return dict(merged)
@@ -19,5 +19,8 @@ class UserCustomPromptRow(Base):
title: Mapped[str] = mapped_column(String(255), nullable=False, default="")
content: Mapped[str] = mapped_column(Text, nullable=False)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# Optional association to a single agent. NULL = global (shown everywhere);
# otherwise the prompt only surfaces when that agent is selected.
agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -42,6 +42,7 @@ class UserCustomPromptRepository:
title=data.get("title") or "",
content=data["content"],
sort_order=data.get("sort_order", 0),
agent_id=(data.get("agent_id") or None),
created_at=now,
updated_at=now,
)
@@ -59,6 +60,10 @@ class UserCustomPromptRepository:
for key in ("title", "content", "sort_order"):
if key in data:
setattr(row, key, data[key])
# ``agent_id`` is explicitly nullable: an empty/None value clears the
# association (back to global), so accept None as a real update.
if "agent_id" in data:
row.agent_id = data["agent_id"] or None
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
@@ -219,6 +219,23 @@ def _task_requires_markdown(task: dict[str, Any]) -> bool:
return bool(isinstance(ctx, dict) and ctx.get("require_markdown"))
def _task_requires_html(task: dict[str, Any]) -> bool:
"""Whether a task must produce an HTML page (the ``必须生成 HTML`` switch).
The flag lives inside ``execution_context.require_html``. For backward
compatibility, tasks created under the old two-tab UI (which stored
``task_kind == "html_page"``) are still treated as HTML tasks, so no data
migration is needed. ``require_html`` and ``require_markdown`` are
independent — both may be on at once.
"""
ctx = task.get("execution_context")
if not isinstance(ctx, dict):
return False
if ctx.get("require_html"):
return True
return ctx.get("task_kind") == "html_page"
def _has_valid_markdown_file(result_files: list[dict[str, Any]] | None) -> bool:
"""True when the run produced a ``.md`` file whose text exceeds the minimum.
@@ -905,15 +922,18 @@ class ScheduledTaskService:
scheduled_for: datetime,
run_id: str,
) -> dict[str, Any]:
"""Dispatch a single execution attempt by task kind.
"""Dispatch a single execution attempt by the task's output requirements.
``html_page`` tasks render an HTML document via GLM5; everything else
runs the regular lead-agent attempt. The kind lives in
``execution_context.task_kind`` so no schema migration is needed.
A task carries two independent switches in ``execution_context``:
``require_html`` (必须生成 HTML) and ``require_markdown`` (必须生成
Markdown). When ``require_html`` is set the run goes through the HTML
render pipeline (which also enforces the Markdown requirement when both
switches are on); otherwise it runs the regular lead-agent attempt.
Both flags live in ``execution_context`` so no schema migration is
needed, and old ``task_kind == "html_page"`` tasks still resolve to the
HTML pipeline via :func:`_task_requires_html`.
"""
ctx = task.get("execution_context")
kind = ctx.get("task_kind") if isinstance(ctx, dict) else None
if kind == "html_page":
if _task_requires_html(task):
return await self._run_html_page_attempt(
task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id
)
@@ -956,17 +976,21 @@ class ScheduledTaskService:
runtime_thread_id: str,
scheduled_for: datetime,
run_id: str,
) -> str:
) -> tuple[str, list[dict[str, Any]]]:
"""Run the lead agent to collect source data for the HTML page.
Reuses the regular agent-run primitives (with the task's search query
folded into the prompt) and returns the agent's final text, which feeds
the GLM5 render step as ``本次数据``. Returns an empty string if the
agent produced nothing.
folded into the prompt) and returns ``(text, files)`` — the agent's
final text (feeds the render step as ``本次数据``) plus any files it
produced. The files are only captured/needed when the task also has the
Markdown switch on (both 必须生成 HTML and 必须生成 Markdown), so the run
can validate and surface the ``.md`` file alongside the HTML page;
otherwise the list is empty.
"""
cfg = self._html_config(task)
page_prompt = str(cfg.get("page_prompt") or "").strip()
negative_prompt = str(cfg.get("negative_prompt") or "").strip()
also_markdown = _task_requires_markdown(task)
gather_instruction = (
"请运用你自身的检索/搜索等技能,围绕下述页面需求获取最新、真实的资料,"
"并整理成结构化的中文要点(可用列表/小标题),供后续生成 HTML 页面使用,本步骤不要生成 HTML:\n"
@@ -982,12 +1006,22 @@ class ScheduledTaskService:
# which sources/data the agent collects, otherwise it silently leaks in.
if negative_prompt:
parts.append(f"限制条件(最高优先级,必须严格遵守):{negative_prompt}")
gather_task = {**task, "prompt": "\n".join(parts)}
# Never apply the markdown gate to the gather pass.
gather_prompt = "\n".join(parts)
# When the Markdown switch is also on, ask this same gather pass to save a
# .md file (the length gate is enforced in code, not surfaced here) so the
# task produces both deliverables in one agent run.
if also_markdown:
gather_prompt += _MARKDOWN_INSTRUCTION
gather_task = {**task, "prompt": gather_prompt}
gather_ctx = dict(task.get("execution_context") or {})
gather_ctx.pop("require_markdown", None)
# Strip the HTML-specific flags so the gather pass is a plain agent run.
gather_ctx.pop("require_html", None)
gather_ctx.pop("task_kind", None)
gather_ctx.pop("html_config", None)
# Keep ``require_markdown`` only when both switches are on, so the agent
# is told (via _start_agent_run) to produce the .md file.
if not also_markdown:
gather_ctx.pop("require_markdown", None)
# Use the task's selected model for the data-gathering agent too, so the
# whole HTML pipeline honors ``html_config.model_name`` instead of
# silently falling back to the default model for this step.
@@ -997,13 +1031,22 @@ class ScheduledTaskService:
gather_task["execution_context"] = gather_ctx
await self._ensure_thread_meta(task["user_id"], runtime_thread_id)
files: list[dict[str, Any]] = []
async with self._agent_run_lock:
output_snapshot = self._snapshot_result_files(runtime_thread_id, task["user_id"]) if also_markdown else None
record = await self._start_agent_run(gather_task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id)
await self.store.update_run(run_id, {"agent_run_id": record.run_id})
if record.task is not None:
await record.task
text = await self._extract_latest_ai_text(runtime_thread_id)
return text or ""
if also_markdown:
files = await self._extract_result_files(
runtime_thread_id,
task["user_id"],
run_id=run_id,
output_snapshot=output_snapshot,
)
return text or "", files
def _collect_html_skill_instructions(self) -> str:
"""Concatenate the bodies of enabled skills marked for HTML pages.
@@ -1072,10 +1115,17 @@ class ScheduledTaskService:
negative_prompt = str(cfg.get("negative_prompt") or "").strip()
reference_favorite_id = cfg.get("reference_favorite_id") or None
# 1) Gather source data (real, via the agent's search tools).
data_text = await self._gather_html_data(
# 1) Gather source data (real, via the agent's search tools). When the
# Markdown switch is also on, the gather pass additionally saves the .md
# file, returned here so we can validate + surface it next to the page.
also_markdown = _task_requires_markdown(task)
data_text, gathered_files = await self._gather_html_data(
task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id
)
if also_markdown and not _has_valid_markdown_file(gathered_files):
raise MarkdownRequirementError(
f"未满足输出要求:需生成 Markdown(.md) 文件,且文件正文不少于 {_MARKDOWN_MIN_CHARS} 字"
)
# 2) Load reference style/layout from a favorite if one was selected.
reference_style = None
@@ -1158,6 +1208,13 @@ class ScheduledTaskService:
archive_path = self._archive_html_index(run_id, html)
files: list[dict[str, Any]] = []
# When both switches are on, surface the gathered Markdown file(s)
# alongside the generated page so the run delivers both artifacts.
if also_markdown:
for item in gathered_files:
name = str(item.get("name") or "") if isinstance(item, dict) else ""
if name.lower().endswith((".md", ".markdown")):
files.append(item)
if archive_path is not None:
files.append(
{
@@ -3,9 +3,14 @@ from __future__ import annotations
import pytest
import yaml
from app.gateway.routers.agents import _load_config_extras, _write_agent_config
from app.gateway.routers.agents import (
_backfill_listing_cache,
_load_config_extras,
_write_agent_config,
)
from app.gateway.services import build_run_config
from deerflow.config import paths as paths_module
from deerflow.config.agents_config import AgentConfig
from deerflow.persistence.agents import MemoryAgentStore
@@ -198,3 +203,46 @@ def test_write_agent_config_omits_skills_when_none(isolated_paths) -> None:
assert "skills" not in raw
assert "model" not in raw
assert "tool_groups" not in raw
@pytest.mark.asyncio
async def test_backfill_listing_cache_mirrors_config_values() -> None:
store = MemoryAgentStore()
cfg = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=["g1"], model="m1")
await _backfill_listing_cache(store, [("a1", cfg)])
extras = await store.get_extras_for(["a1"])
assert extras["a1"]["skills"] == ["s1", "s2"]
assert extras["a1"]["tool_groups"] == ["g1"]
assert extras["a1"]["model"] == "m1"
@pytest.mark.asyncio
async def test_backfill_listing_cache_is_idempotent_and_skips_unchanged(monkeypatch) -> None:
store = MemoryAgentStore()
cfg = AgentConfig(id="a1", name="Agent", skills=["s1"], tool_groups=None, model=None)
await _backfill_listing_cache(store, [("a1", cfg)])
# On a second run with identical config, no re-write should occur.
calls = {"replace": 0, "set": 0}
orig_replace = store.replace_skills
orig_set = store.set_extras
async def _replace(agent_id, skills):
calls["replace"] += 1
await orig_replace(agent_id, skills)
async def _set(agent_id, *, tool_groups, model):
calls["set"] += 1
await orig_set(agent_id, tool_groups=tool_groups, model=model)
monkeypatch.setattr(store, "replace_skills", _replace)
monkeypatch.setattr(store, "set_extras", _set)
await _backfill_listing_cache(store, [("a1", cfg)])
assert calls == {"replace": 0, "set": 0}
# Changing the config triggers a re-write.
cfg2 = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=None, model=None)
await _backfill_listing_cache(store, [("a1", cfg2)])
assert calls["replace"] == 1