feat: 知识库文件夹/模板、技能压缩设置与摘要窗口调优

- 知识库新增文件夹与模板支持(模型/仓储/服务/迁移),前端知识管理组件
- 岗位管理模型与接口完善
- 技能新增压缩设置页与相关存储/工具能力
- 系统设置扩展配置项
- 摘要中间件 keep 窗口改用 token 预算(16000),减少重复压缩

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-16 15:23:54 +08:00
1 parent e6662cf673
commit 5ebbb82d66
54 files changed
+2658 -100

No files matched your search

+54 -6
View File
@@ -163,7 +163,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
6. **GuardrailMiddleware** - Pre-tool-call authorization via pluggable `GuardrailProvider` protocol (optional, if `guardrails.enabled` in config). Evaluates each tool call and returns error ToolMessage on deny. Three provider options: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom providers. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) for setup, usage, and how to implement a provider.
7. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution continues
8. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting
9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. **Stickiness headroom** (`_apply_stickiness_headroom`, `_KEEP_HEADROOM_FRACTION=0.75`): after the `keep` policy picks a cutoff, the kept suffix is forced below ~75% of *every* trigger before summarizing. Without it a `keep` in *messages* (e.g. 20) can itself exceed a *token* trigger (e.g. 60000) when recent messages are large (tool outputs / pasted docs), leaving the "compressed" context still over the trigger so it re-summarizes on every follow-up question; the headroom makes the sticky boundary actually stick (re-compaction only after genuinely new content). When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
10. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
11. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
12. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model
@@ -175,7 +175,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
18. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation ? timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills/<category>/<name>/SKILL.md` path (per the `<skill_system>` prompt block ? there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. **Per-run skill merging**: `SqlToolMetricsStore.record` upserts skill rows by `(run_id, skill_name)` instead of inserting one row per call ? a skill touched by several tool calls / retries inside one run collapses to a single row. Any success in the run wins (a skill that failed then finally succeeded is recorded as one `success`); a skill that only ever failed yields one `error` row whose `error_message` accumulates every attempt's message; durations are summed. Non-skill tool calls, and skill calls with no `run_id`, keep one row each. The read?merge?write is guarded by a per-event-loop `asyncio.Lock`. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown).
9. **PromptPrefixMiddleware** - Injects the admin-configured prompt prefix into the model request only. The prefix is tagged onto the first human message's `additional_kwargs['prompt_prefix']` by the Gateway (`app.gateway.services`) and prepended to content here at call time, so the persisted/displayed message stays prefix-free
10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. **Stickiness headroom** (`_apply_stickiness_headroom`, `_KEEP_HEADROOM_FRACTION=0.75`): after the `keep` policy picks a cutoff, the kept suffix is forced below ~75% of *every* trigger before summarizing. Without it a `keep` in *messages* (e.g. 20) can itself exceed a *token* trigger (e.g. 60000) when recent messages are large (tool outputs / pasted docs), leaving the "compressed" context still over the trigger so it re-summarizes on every follow-up question; the headroom makes the sticky boundary actually stick (re-compaction only after genuinely new content). When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
12. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
13. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model
@@ -252,7 +252,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` ? HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
| **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
| **Light Apps** (`/api/light-apps`) | Light-app registry (????? / ????), **global/shared** (no user scoping; `created_by`/`updated_by` audit-only). Two kinds: `window_open` (launched in a new tab from the ???? card grid) and `iframe` (embedded under a parent sidebar menu via `mount_location` and dynamically injected into the sidebar). Both carry `route_params` ? a list of `{key, paramType: login\|theme, source, customValue, themeValue}` descriptors the **frontend** resolves at launch time (login params read from `localStorage.userInfo`; theme params use a literal value) and appends to `url` as query string. `GET /` - list (filters `app_type`/`mount_location`/`enabled_only`/`search`); `GET /{app_id}` - read one (both open to any authenticated user); `POST /`, `PUT /{app_id}`, `DELETE /{app_id}` - **admin only**; iframe apps must carry a `mount_location` (400 otherwise); the unique `name` collides as HTTP 409. Backed by `deerflow.persistence.light_apps` (table `light_apps`, auto-created by `create_all`); store wired in `deps.py` as `app.state.light_app_store`. Tests: `tests/test_light_apps.py`. |
| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` ? `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "???????". The "??????" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/file` - **import an uploaded file** (multipart: `file` + optional `title`/`tags`/`status`/`model_name`) into wiki note(s): the file is converted to Markdown (md/txt read directly, pdf/word/ppt/excel via `convert_file_to_markdown`), distilled by the same wiki-capture LLM pipeline as thread capture (`source_type="file"`), and falls back to a verbatim **draft** note when the model returns nothing usable (`llm_fallback:true`); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector?keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities, graph-blacklist filtered); `GET/POST /graph/blacklist` + `DELETE /graph/blacklist/{id}` - graph node blacklist (labels = entity names / note titles, case-insensitive match; blacklisted nodes and their edges are dropped from `/graph` and the wiki-export artifacts; table `knowledge_graph_blacklist`, auto-created); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[?]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). Backed by the `deerflow.knowledge` module (see below). |
| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` ? `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "???????". The "??????" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/file` - **import an uploaded file** (multipart: `file` + optional `title`/`tags`/`status`/`model_name`) into wiki note(s): the file is converted to Markdown (md/txt read directly, pdf/word/ppt/excel via `convert_file_to_markdown`), distilled by the same wiki-capture LLM pipeline as thread capture (`source_type="file"`), and falls back to a verbatim **draft** note when the model returns nothing usable (`llm_fallback:true`); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector?keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities, graph-blacklist filtered); `GET/POST /graph/blacklist` + `DELETE /graph/blacklist/{id}` - graph node blacklist (labels = entity names / note titles, case-insensitive match; blacklisted nodes and their edges are dropped from `/graph` and the wiki-export artifacts; table `knowledge_graph_blacklist`, auto-created); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[?]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). `GET/POST /folders` + `PUT/DELETE /folders/{id}` - user-managed directories (notes carry a `folder` path; `GET /notes?folder=` filters a dir + sub-dirs); `GET/POST /extract-templates` + `GET/PUT/DELETE /extract-templates/{id}` - editable extraction-prompt templates (thread/document ingest accept `template_id`; capture/PUT-note accept `folder`). Backed by the `deerflow.knowledge` module (see below). |
Proxied through nginx: `/api/langgraph/*` ? LangGraph, all other `/api/*` ? Gateway.
@@ -316,6 +316,14 @@ Self-contained product knowledge base. **All knowledge capability lives in this
- **Phase 3 ? governance**: `sensitive.py` redacts secrets/PII before persistence (`sensitive_redaction_enabled`); every update snapshots `knowledge_note_versions` (`GET /notes/{id}/versions`); `GET /duplicates` + `POST /merge` for dedup; background `AutoIngestQueue` (wired in `deps.py`, started/stopped in the gateway lifespan) auto-sediments finished conversations **off the chat path** when `auto_ingest_enabled` ? enqueued from `KnowledgeRagMiddleware.after_agent`, value-judged by min-chars + `skip_existing`, landing as `draft`.
- **Phase 4 ? graph**: LLM extraction also yields entities/relations ? `knowledge_entities`/`knowledge_relations`; `exporter.build_graph` adds entity nodes + note?entity/entity?entity edges (reflected in `graph.html`/`graph.json`); `GET /graph` returns product graph data.
**User directories + editable extraction templates + display polish (knowledge UX overhaul):**
- **Custom directories (`knowledge_notes.folder` + `knowledge_folders`)** ? notes carry a free-form, slash-joined `folder` path (e.g. `投研/行业`) independent of the wiki-capture `category` and of `vault_path`. `knowledge_folders` is the manageable directory list (supports empty folders + rename/move that re-points descendant folders **and** filed notes). Router CRUD under `/api/knowledge/folders` (`GET`/`POST`/`PUT/{id}`/`DELETE/{id}?reassign_to=`); `GET /notes?folder=` filters a directory + its sub-folders (`folder=""` ? unfiled). Capture (`capture_thread`/`capture_file`/`create_manual_note`) and `PUT /notes/{id}` accept `folder` (the note PUT uses `model_fields_set` so an absent key leaves the folder untouched, a present `null` clears it). Frontend `buildKnowledgeTree(notes, folderPaths)` builds the left tree from `folder` (no longer from `vault_path`); folder management dialog + `FolderSelect` in the create/import dialogs and the note detail header.
- **Editable extraction prompts (`knowledge_extract_templates`)** ? the hard-coded distillation prompts are seeded as editable templates on startup (`KnowledgeService.seed_default_templates`, idempotent — `默认 · 对话提炼` / `默认 · 文档提炼`, each `is_default` for its scope). `scope ∈ {thread, document, both}`; capture resolves an explicit `template_id` → the scope default → the built-in constant, and passes it to `llm_extractor.extract_thread_llm_multi` / `extract_document_llm_multi` as `system_prompt` (used verbatim). Router CRUD under `/api/knowledge/extract-templates` (409 on duplicate name). Frontend: extraction-template management dialog + `TemplateSelect` in the batch/file import dialogs. Editing a template is the supported way to fix "提炼太精简" (loosen the prompt to keep more detail) and to add alternate extraction logics.
- **Single source list** ? `extractor._related_section_md` no longer emits a `### Sources` sub-section into the body; the note detail shows sources once via the structured `来源` block (`note.sources`). `## Related` stays pure wikilinks so the frontend extracts it cleanly into the related-docs module.
- **Chinese section headings (display-only)** ? the frontend `localizeHeadings` (wikiUtils) maps `Context/Finding / Decision/Reasoning/Implications/Related/Sources` to 中文 at render time (applied to the body fed to both `extractToc` and the viewer, so anchors stay consistent). No data migration — historical English-heading notes render in Chinese too.
- **Historical-data cleanup** ? `scripts/cleanup_knowledge_related.py` (one-off, idempotent, `--dry-run`/`--note-id`) strips legacy `## Related` structural links (vault scaffold: `index`/`_meta/taxonomy`/`projects/…`, any slash-bearing target) and the old `### Sources` sub-section from existing notes (DB `content_md` + vault files, via `KnowledgeService.update_note`).
- Migration `20260616_02` + idempotent `scripts/sql/column_additions_mysql.sql` cover `knowledge_notes.folder` (+ index); the two new tables are auto-created by `create_all` / `_ensure_orm_columns_sync` on startup. Tests: `tests/test_knowledge.py`.
### 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.
@@ -376,7 +384,11 @@ default gets their grants materialized).
position CRUD; `GET/PUT /{id}/tag-bindings`; `GET/POST/DELETE /{id}/members`
(batch assign/unassign → triggers sync); `POST /{id}/resync`;
`GET /users` (all users + their position); `PUT /users/{user_id}` (set one
user's position). Related changes: `GET /api/recommended-questions` is now
user's position). **`GET /me` is the one non-admin route** — it returns the
caller's own effective position (explicit assignment, else the default position
unassigned users inherit, else `null`); powers the top-bar 岗位 label
(`strategy-components/Header.tsx` user cluster, via `useMyPosition`). Related
changes: `GET /api/recommended-questions` is now
filtered by the caller's position (`?all=1` bypasses for admin management) and
admin can tag questions via `/api/tags/assignments/recommended_question/{id}`;
`GET /api/skills?scope=mine` applies skill visibility. Frontend: `core/positions`
@@ -469,11 +481,47 @@ Operators upgrading an existing DB run the whole file (`USE <db>;` then execute)
- **Location**: `deer-flow/skills/{public,custom}/`
- **Format**: Directory with `SKILL.md` (YAML frontmatter: name, description, license, allowed-tools)
- **Loading**: `load_skills()` recursively scans `skills/{public,custom}` for `SKILL.md`, parses metadata, and reads enabled state from extensions_config.json
- **Loading**: `load_skills()` recursively scans `skills/{public,custom}` for `SKILL.md`, parses metadata, and reads enabled state from extensions_config.json. **A directory containing `SKILL.md` is a self-contained leaf** — discovery does not descend into its subdirectories, so a skill *package* that bundles its own `examples/`/`references/` sub-skills (each with a SKILL.md) surfaces as **one** skill, not a swarm of phantom entries.
- **Discovery↔delete consistency**: a skill is keyed for discovery by its frontmatter `name`, but its on-disk directory name may differ (hand-dropped / mis-extracted package). `LocalSkillStorage._resolve_skill_dir(name, category)` resolves the *discovered* directory (canonical `<category>/<name>` fast-path, else scan-by-frontmatter-name); `custom_skill_exists`/`read_custom_skill`/`delete_custom_skill` all go through it, so **anything that gets listed can also be read and deleted** regardless of directory name (the old code assumed `custom/<name>/` and 404'd "Custom skill 'x' not found." on any non-canonical layout). Delete is path-guarded to never escape the `custom/` root. Tests: `tests/test_skill_storage_delete.py`.
- **Injection**: Enabled skills listed in agent system prompt with container paths
- **Installation**: `POST /api/skills/install` extracts .skill ZIP archive to custom/ directory
- **Chinese name (`name_zh`)**: a Chinese display name stored in the **`skills` DB table** (`SkillRow.name_zh`, `String(128)`, migration `20260609_01`) ? never written to the on-disk SKILL.md, so it works for built-in skills too. Threaded through `SkillStore` create/ensure_legacy/update/update_admin exactly like `detail`; surfaced on `SkillResponse.name_zh` from the ownership record in `_skill_to_response`. Authorization mirrors `detail`: custom skill ? owner or admin; built-in/public ? admin only (row materialized on demand via `ensure_legacy`). Endpoints (in `app/gateway/routers/skills.py`): `PUT /api/skills/{name}/name-zh` (set/clear), `POST /api/skills/{name}/name-zh/generate` (LLM generates one), `POST /api/skills/name-zh/generate-missing` (????: **admin-only** bulk-generate for every skill that lacks one, bounded-concurrency `asyncio.gather`). **Optional model selection**: both generate endpoints accept an optional JSON body `{model_name}` ? an explicit name wins, else the curator model setting, else the main model (`_resolve_skill_name_zh_model_name`). **Model-error surfacing**: `_generate_skill_name_zh` raises `SkillNameZhModelError` when the model itself fails (build/timeout/API error) ? distinct from the model returning an unusable name (? `None`). The single endpoint maps that to `502 ?????,?????????:?`; the bulk endpoint fails fast on a model-build error (502) and otherwise carries a representative `model_error` string in `SkillNameZhBulkResponse` so the UI can say "N ?????????". The frontend (`skill-settings-page.tsx` list + `skill-detail-dialog.tsx` editor, plus the agent skill pickers/badges in `NewAgentPage.tsx` + `agent-card.tsx`) shows `name_zh || name`, offers per-skill "AI ??" plus an **admin-only** header "???????" button, and a compact model dropdown (`components/workspace/model-name-select.tsx`, default = configured model) beside each action.
#### Skill Compression (技能压缩)
When the skill catalogue grows large, injecting every enabled skill full-text
into the lead agent's system prompt bloats it and wastes tokens. **Skill
compression** is an admin-gated mode that keeps only *常驻(免压缩)* skills in the
prompt body and surfaces the rest on demand via keyword retrieval.
- **Master switch** — `system_settings.json → skill_compression`
(`SkillCompressionSettings`: `enabled` default **false**, `search_top_n`
default 5, `keep_index_in_prompt` default true). Read/written by
`GET`/`PUT /api/system-settings/skill-compression` (GET open to any user — the
skill detail page reads it; PUT admin-only, refreshes the prompt cache).
- **Per-skill 常驻 flag** — `extensions_config.json → skills.{name}.always_on`
(`SkillStateConfig.always_on`, alongside `enabled`). Surfaced on
`Skill.always_on` (populated in `load_skills`) and `SkillResponse.always_on`;
toggled via `PUT /api/skills/{name}/always-on`. `update_skill` /
`update_skill_display` preserve it when rewriting the config entry.
- **Prompt split** — `get_skills_prompt_section` (`lead_agent/prompt.py`) reads
the compression settings (skipped for restricted custom agents — they already
have a curated allowlist). When **off**, behaviour is unchanged (all enabled
skills full-text). When **on**, `always_on` skills render as full `<skill>`
blocks; the rest are dropped from the body, replaced by a `search_skills`
usage note plus (when `keep_index_in_prompt`) a name-only `<compressed_skills>`
index. The `_get_cached_skills_prompt_section` cache key includes the
`always_on` flag + compression flags so changes invalidate correctly.
- **Retrieval tool** — `search_skills` (`tools/builtins/skill_tools.py`,
`search_skills_impl`) keyword-scores `name + description` over enabled,
visible, non-archived skills (same visibility filter as `skill_list`) and
returns top-N `{name, description, category, location}`. It is registered in
`get_available_tools` **only when compression is enabled** (offline,
deterministic, zero external deps). The agent then `read_file`s the returned
`location`.
Tests: `tests/test_skill_compression.py`.
#### Skill Curator (`packages/harness/deerflow/skills/curator.py`)
Background skill lifecycle manager. The Gateway `lifespan` starts a resident
@@ -640,7 +688,7 @@ Both endpoints are whitelisted in `auth_middleware._PUBLIC_EXACT_PATHS`. The sha
**`extensions_config.json`**:
- `mcpServers` - Map of server name ? config (enabled, type, command, args, env, url, headers, oauth, description)
- `skills` - Map of skill name ? state (enabled)
- `skills` - Map of skill name ? state (enabled, always_on)
Both can be modified at runtime via Gateway API endpoints or `DeerFlowClient` methods.
@@ -184,6 +184,14 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.knowledge_service = make_knowledge_service(sf, config.knowledge)
# Seed the built-in extraction prompts as editable templates so operators
# can immediately view/edit them (idempotent; never overwrites edits).
if app.state.knowledge_service is not None:
try:
await app.state.knowledge_service.seed_default_templates()
except Exception:
logger.exception("Failed to seed knowledge extraction templates")
# Background auto-sedimentation queue (phase 3) — drains finished
# conversations off the chat path. Worker only does work when
# ``knowledge.auto_ingest_enabled`` is set; the queue is always wired so
@@ -54,6 +54,8 @@ class IngestThreadRequest(BaseModel):
description="When true, sediment the thread asynchronously in a background task and return immediately.",
)
model_name: str | None = Field(default=None, description="可选:指定用于知识提炼的大模型名称;留空使用默认配置")
template_id: str | None = Field(default=None, description="可选:抽取提示词模板 id;留空使用对话默认模板")
folder: str | None = Field(default=None, description="可选:沉淀结果归入的目录(斜杠分隔路径)")
class BatchIngestRequest(BaseModel):
@@ -64,6 +66,8 @@ class BatchIngestRequest(BaseModel):
skip_existing: bool = True
mode: str = "summary"
model_name: str | None = Field(default=None, description="可选:指定用于知识提炼的大模型名称;留空使用默认配置")
template_id: str | None = Field(default=None, description="可选:抽取提示词模板 id;留空使用对话默认模板")
folder: str | None = Field(default=None, description="可选:沉淀结果归入的目录(斜杠分隔路径)")
class SearchResultItem(BaseModel):
@@ -90,6 +94,7 @@ class CreateNoteRequest(BaseModel):
summary: str | None = None
tags: list[str] = Field(default_factory=list)
status: str = "approved"
folder: str | None = None
class UpdateNoteRequest(BaseModel):
@@ -98,6 +103,34 @@ class UpdateNoteRequest(BaseModel):
summary: str | None = None
tags: list[str] | None = None
status: str | None = None
# Present (even as null) ⇒ (re)assign the note's directory; absent ⇒ leave as-is.
folder: str | None = None
class CreateFolderRequest(BaseModel):
path: str = Field(description="目录全路径,斜杠分隔,如 投研/行业")
class RenameFolderRequest(BaseModel):
path: str = Field(description="新的目录全路径")
class ExtractTemplateRequest(BaseModel):
name: str
system_prompt: str
scope: str = Field(default="both", description="thread | document | both")
description: str | None = None
is_default: bool = False
enabled: bool = True
class UpdateExtractTemplateRequest(BaseModel):
name: str | None = None
system_prompt: str | None = None
scope: str | None = None
description: str | None = None
is_default: bool | None = None
enabled: bool | None = None
class SearchRequest(BaseModel):
@@ -187,6 +220,8 @@ async def _ingest_thread_background(service, checkpointer, thread_id: str, body:
reference_batches=batches,
created_by=created_by,
model_name=body.model_name,
template_id=body.template_id,
folder=body.folder,
)
except EmptyThreadError:
logger.warning("Background knowledge ingest skipped: thread %s has no sediment-able content", sanitize_log_param(thread_id))
@@ -237,6 +272,8 @@ async def ingest_thread(thread_id: str, body: IngestThreadRequest, request: Requ
reference_batches=batches,
created_by=get_effective_user_id(),
model_name=body.model_name,
template_id=body.template_id,
folder=body.folder,
)
except EmptyThreadError:
raise HTTPException(status_code=400, detail="Thread has no sediment-able content")
@@ -285,6 +322,8 @@ async def ingest_threads_batch(body: BatchIngestRequest, request: Request) -> di
created_by=get_effective_user_id(),
skip_existing=body.skip_existing,
model_name=body.model_name,
template_id=body.template_id,
folder=body.folder,
)
if result.get("llm_fallback"):
llm_fallback += 1
@@ -352,6 +391,8 @@ async def ingest_file(
tags: str | None = Form(default=None),
status: str = Form(default="approved"),
model_name: str | None = Form(default=None),
template_id: str | None = Form(default=None),
folder: str | None = Form(default=None),
) -> dict[str, Any]:
"""Import an uploaded file into the knowledge base as wiki note(s).
@@ -371,6 +412,8 @@ async def ingest_file(
status=status,
created_by=get_effective_user_id(),
model_name=(model_name or None),
template_id=(template_id or None),
folder=(folder or None),
)
except EmptyThreadError:
raise HTTPException(status_code=422, detail="文件内容无可沉淀的知识")
@@ -413,6 +456,7 @@ async def create_note(body: CreateNoteRequest, request: Request) -> dict[str, An
summary=body.summary,
tags=body.tags,
status=body.status,
folder=body.folder,
created_by=get_effective_user_id(),
)
return {"note": note}
@@ -427,13 +471,14 @@ async def list_notes(
entity: str | None = None,
source_type: str | None = None,
status: str | None = None,
folder: str | None = None,
limit: int = 20,
offset: int = 0,
) -> dict[str, Any]:
service = get_knowledge_service(request)
limit = max(1, min(limit, 100))
offset = max(0, offset)
return await service.list_notes(keyword=keyword, tag=tag, entity=entity, source_type=source_type, status=status, limit=limit, offset=offset)
return await service.list_notes(keyword=keyword, tag=tag, entity=entity, source_type=source_type, status=status, folder=folder, limit=limit, offset=offset)
@router.get("/notes/{note_id}")
@@ -451,6 +496,9 @@ async def get_note(note_id: str, request: Request) -> dict[str, Any]:
@require_auth
async def update_note(note_id: str, body: UpdateNoteRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
# ``folder`` present in the request body (even as null) ⇒ (re)assign the
# directory; absent ⇒ leave it untouched.
folder_provided = "folder" in body.model_fields_set
note = await service.update_note(
note_id,
title=body.title,
@@ -458,6 +506,8 @@ async def update_note(note_id: str, body: UpdateNoteRequest, request: Request) -
summary=body.summary,
tags=body.tags,
status=body.status,
folder=body.folder,
set_folder=folder_provided,
updated_by=get_effective_user_id(),
)
if note is None:
@@ -634,3 +684,119 @@ async def export_file(filename: str, request: Request) -> FileResponse:
raise HTTPException(status_code=404, detail="Export file not found")
media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
return FileResponse(path, media_type=media_type)
# --------------------------------------------------------------------------- #
# Folders (user-managed directories)
# --------------------------------------------------------------------------- #
@router.get("/folders")
@require_auth
async def list_folders(request: Request) -> dict[str, Any]:
"""List every managed knowledge directory (for the directory tree + picker)."""
service = get_knowledge_service(request)
return {"items": await service.list_folders()}
@router.post("/folders")
@require_auth
async def create_folder(body: CreateFolderRequest, request: Request) -> dict[str, Any]:
"""Create a directory (parents auto-created). Notes are filed into it later."""
service = get_knowledge_service(request)
try:
folder = await service.create_folder(body.path, created_by=get_effective_user_id())
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"folder": folder}
@router.put("/folders/{folder_id}")
@require_auth
async def rename_folder(folder_id: str, body: RenameFolderRequest, request: Request) -> dict[str, Any]:
"""Rename/move a directory; descendants + filed notes follow the new prefix."""
service = get_knowledge_service(request)
try:
folder = await service.rename_folder(folder_id, body.path)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
if folder is None:
raise HTTPException(status_code=404, detail="Folder not found")
return {"folder": folder}
@router.delete("/folders/{folder_id}")
@require_auth
async def delete_folder(folder_id: str, request: Request, reassign_to: str | None = None) -> dict[str, Any]:
"""Delete a directory (and sub-dirs); notes inside move to ``reassign_to`` or unfiled."""
service = get_knowledge_service(request)
if not await service.delete_folder(folder_id, reassign_to=reassign_to):
raise HTTPException(status_code=404, detail="Folder not found")
return {"ok": True}
# --------------------------------------------------------------------------- #
# Extraction prompt templates
# --------------------------------------------------------------------------- #
@router.get("/extract-templates")
@require_auth
async def list_extract_templates(request: Request, scope: str | None = None, enabled_only: bool = False) -> dict[str, Any]:
"""List editable extraction-prompt templates (optionally filtered by scope)."""
service = get_knowledge_service(request)
return {"items": await service.list_extract_templates(scope=scope, enabled_only=enabled_only)}
@router.get("/extract-templates/{template_id}")
@require_auth
async def get_extract_template(template_id: str, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
tpl = await service.get_extract_template(template_id)
if tpl is None:
raise HTTPException(status_code=404, detail="Template not found")
return {"template": tpl}
@router.post("/extract-templates")
@require_auth
async def create_extract_template(body: ExtractTemplateRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
if await service.template_name_taken(body.name):
raise HTTPException(status_code=409, detail="同名模板已存在")
tpl = await service.create_extract_template(
name=body.name,
system_prompt=body.system_prompt,
scope=body.scope,
description=body.description,
is_default=body.is_default,
enabled=body.enabled,
created_by=get_effective_user_id(),
)
return {"template": tpl}
@router.put("/extract-templates/{template_id}")
@require_auth
async def update_extract_template(template_id: str, body: UpdateExtractTemplateRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
if body.name is not None and await service.template_name_taken(body.name, exclude_id=template_id):
raise HTTPException(status_code=409, detail="同名模板已存在")
tpl = await service.update_extract_template(
template_id,
name=body.name,
system_prompt=body.system_prompt,
scope=body.scope,
description=body.description,
is_default=body.is_default,
enabled=body.enabled,
updated_by=get_effective_user_id(),
)
if tpl is None:
raise HTTPException(status_code=404, detail="Template not found")
return {"template": tpl}
@router.delete("/extract-templates/{template_id}")
@require_auth
async def delete_extract_template(template_id: str, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
if not await service.delete_extract_template(template_id):
raise HTTPException(status_code=404, detail="Template not found")
return {"ok": True}
@@ -110,6 +110,10 @@ class PositionUsersResponse(BaseModel):
users: list[PositionUserResponse]
class MyPositionResponse(BaseModel):
position: PositionResponse | None = None
# ----------------------------------------------------------------- position CRUD
@@ -127,6 +131,28 @@ async def list_position_users(request: Request, store: PositionStore = Depends(g
return PositionUsersResponse(users=[PositionUserResponse(**u) for u in users])
@router.get("/me", response_model=MyPositionResponse, summary="Get My Effective Position")
async def get_my_position(request: Request, store: PositionStore = Depends(get_position_store)) -> MyPositionResponse:
"""The caller's own position — open to any authenticated user (not admin).
Returns the explicitly-assigned position, falling back to the default
position (which unassigned users inherit). ``null`` when neither exists or
the caller is anonymous. Powers the top-bar 岗位 label.
"""
user = await get_optional_user_from_request(request)
if user is None:
return MyPositionResponse(position=None)
record = None
pid = await store.get_user_position_id(str(user.id))
if pid:
record = await store.get_position(pid)
if record is None:
default_id = await store.get_default_position_id()
if default_id:
record = await store.get_position(default_id)
return MyPositionResponse(position=PositionResponse(**record) if record else None)
@router.post("", response_model=PositionResponse, status_code=201, summary="Create Position")
async def create_position(request: Request, body: PositionCreateRequest, store: PositionStore = Depends(get_position_store)) -> PositionResponse:
await _require_admin(request)
@@ -154,6 +154,7 @@ class SkillResponse(BaseModel):
license: str | None = Field(None, description="License information")
category: SkillCategory = Field(..., description="Category of the skill (public or custom)")
enabled: bool = Field(default=True, description="Whether this skill is enabled")
always_on: bool = Field(default=False, description="常驻(免压缩):开启技能压缩后仍全文注入系统提示词")
use_count: int = Field(default=0, description="Number of times the skill has been used")
view_count: int = Field(default=0, description="Number of times the skill has been viewed")
patch_count: int = Field(default=0, description="Number of times the skill has been patched")
@@ -186,6 +187,12 @@ class SkillUpdateRequest(BaseModel):
enabled: bool = Field(..., description="Whether to enable or disable the skill")
class SkillAlwaysOnUpdateRequest(BaseModel):
"""Request model for updating a skill's 常驻(免压缩) flag."""
always_on: bool = Field(..., description="Whether this skill bypasses skill compression (full-text injected)")
class SkillDisplayUpdateRequest(BaseModel):
"""Request model for updating a skill's frontend display adapter."""
@@ -334,6 +341,7 @@ def _skill_to_response(
license=skill.license,
category=skill.category,
enabled=skill.enabled,
always_on=skill.always_on,
use_count=entry.get("use_count", 0),
view_count=entry.get("view_count", 0),
patch_count=entry.get("patch_count", 0),
@@ -2186,6 +2194,7 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=request.enabled,
always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
display=current_skill_config.display if current_skill_config else None,
)
@@ -2214,6 +2223,55 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
raise HTTPException(status_code=500, detail=f"Failed to update skill: {str(e)}")
@router.put(
"/skills/{skill_name}/always-on",
response_model=SkillResponse,
summary="Update Skill 常驻(免压缩) Flag",
description="Mark a skill as 常驻(免压缩) so it is injected full-text into the system prompt even when skill compression is enabled.",
)
async def update_skill_always_on(skill_name: str, request: SkillAlwaysOnUpdateRequest, config: AppConfig = Depends(get_config)) -> SkillResponse:
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
skill = next((s for s in skills if s.name == skill_name), None)
if skill is None:
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
config_path = ExtensionsConfig.resolve_config_path()
if config_path is None:
config_path = Path.cwd().parent / "extensions_config.json"
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
extensions_config = get_extensions_config()
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
always_on=request.always_on,
display=current_skill_config.display if current_skill_config else None,
)
with open(config_path, "w", encoding="utf-8") as f:
json.dump(_dump_extensions_config(extensions_config), f, indent=2)
reload_extensions_config()
await refresh_skills_system_prompt_cache_async()
skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
updated_skill = next((s for s in skills if s.name == skill_name), None)
if updated_skill is None:
raise HTTPException(status_code=500, detail=f"Failed to reload skill '{skill_name}' after update")
logger.info(f"Skill '{skill_name}' always_on updated to {request.always_on}")
return _skill_to_response(updated_skill)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to update skill always_on {skill_name}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to update skill: {str(e)}")
@router.put(
"/skills/{skill_name}/display",
response_model=SkillResponse,
@@ -2242,6 +2300,7 @@ async def update_skill_display(
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
display=request.display,
)
@@ -16,6 +16,7 @@ from deerflow.config.system_settings import (
PromptPrefixSettings,
PublishSquare,
PublishSquaresSettings,
SkillCompressionSettings,
SystemSettings,
list_publish_squares,
load_system_settings,
@@ -113,6 +114,64 @@ async def update_prompt_prefix(body: PromptPrefixUpdateRequest, request: Request
return _to_response(saved.prompt_prefix)
class SkillCompressionResponse(BaseModel):
enabled: bool = Field(..., description="Whether skill compression is active")
search_top_n: int = Field(..., description="Max skills returned by the search_skills tool")
keep_index_in_prompt: bool = Field(..., description="Whether compressed skills keep a compact name index in the prompt")
class SkillCompressionUpdateRequest(BaseModel):
enabled: bool | None = None
search_top_n: int | None = Field(default=None, ge=1, le=50)
keep_index_in_prompt: bool | None = None
def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse:
return SkillCompressionResponse(
enabled=settings.enabled,
search_top_n=settings.search_top_n,
keep_index_in_prompt=settings.keep_index_in_prompt,
)
@router.get("/skill-compression", response_model=SkillCompressionResponse)
async def get_skill_compression() -> SkillCompressionResponse:
"""Return the skill-compression configuration (readable by any user).
The frontend reads this to decide whether to surface the per-skill 常驻
(免压缩) toggle on the skill detail page.
"""
return _skill_compression_to_response(load_system_settings().skill_compression)
@router.put("/skill-compression", response_model=SkillCompressionResponse)
async def update_skill_compression(body: SkillCompressionUpdateRequest, request: Request) -> SkillCompressionResponse:
"""Update the skill-compression configuration. Admin only."""
await _require_admin(request)
current = load_system_settings()
fields = body.model_dump(exclude_unset=True)
update: dict[str, object] = {}
if "enabled" in fields and fields["enabled"] is not None:
update["enabled"] = bool(fields["enabled"])
if "keep_index_in_prompt" in fields and fields["keep_index_in_prompt"] is not None:
update["keep_index_in_prompt"] = bool(fields["keep_index_in_prompt"])
if "search_top_n" in fields and fields["search_top_n"] is not None:
update["search_top_n"] = max(1, min(int(fields["search_top_n"]), 50))
compression = current.skill_compression.model_copy(update=update)
saved = save_system_settings(current.model_copy(update={"skill_compression": compression}))
# 压缩开关/参数变化会改变系统提示词的技能注入,清掉提示词缓存让其立即生效。
try:
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
await refresh_skills_system_prompt_cache_async()
except Exception:
logger.debug("Failed to refresh skills prompt cache after compression update", exc_info=True)
return _skill_compression_to_response(saved.skill_compression)
def _leaderboard_to_response(settings: LeaderboardSettings) -> LeaderboardSettingsResponse:
return LeaderboardSettingsResponse(
exclude_user_ids=list(dict.fromkeys([uid for uid in settings.exclude_user_ids if uid])),
@@ -23,6 +23,7 @@ LangChain's stock `SummarizationMiddleware` runs in `before_model` and persists
- The full original messages stay in the checkpoint, so the frontend keeps showing the user's real conversation. The summary message is named `summary` and flagged `hide_from_ui`, and because it only ever lives in the transient request it never reaches the frontend at all.
- The generated summary is cached **per-thread** so we don't pay an extra LLM call on every model invocation. The cached summary is reused (no LLM call) until enough *new* messages accumulate to push the compressed context back over the trigger threshold (a sticky boundary mirroring the stock middleware's post-compaction behavior). The cache is bounded (oldest threads evicted first).
- **Threshold-first**: while the full current conversation is still under the trigger, the request is passed through untouched — no cache lookup, no summary substitution. Compression only ever engages once the real context crosses the configured trigger.
- **Stickiness headroom**: after `keep` picks a cutoff, the kept window is forced below ~75% of *every* trigger before summarizing (`_apply_stickiness_headroom`). This prevents the every-turn re-compaction that happens when `keep` is in *messages* (e.g. 20) but a trigger is in *tokens* (e.g. 60000) and the recent messages are large — without headroom the "compressed" context would still sit over the trigger and re-summarize on every follow-up question.
- When the genuine (blocking) summarize path actually runs an LLM call, the middleware emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel so the chat UI can show a transient "正在压缩" toast (the cheap cache-reuse path stays silent). Frontend handler: `onCustomEvent` in `frontend-web/src/core/threads/hooks.ts`.
> **Legacy threads:** conversations compacted by an older (destructive) build already lost their original messages and carry a persisted `summary` message. The frontend hides any `name: "summary"` message, so the stray summary bubble no longer shows — but those already-discarded messages cannot be recovered.
@@ -758,51 +758,83 @@ def _get_memory_context(agent_name: str | None = None, *, app_config: AppConfig
from deerflow.config.memory_config import get_effective_memory_config
from deerflow.runtime.user_context import get_effective_user_id
user_id = get_effective_user_id()
config = get_effective_memory_config(user_id)
manager = build_memory_manager(
user_id=user_id,
agent_id=agent_name or "default",
config=config,
app_config=app_config,
)
user_id = get_effective_user_id()
config = get_effective_memory_config(user_id)
manager = build_memory_manager(
user_id=user_id,
agent_id=agent_name or "default",
config=config,
app_config=app_config,
)
# 只取静态块 (builtin 的 frozen snapshot + Hindsight 的模式说明)。
# Hindsight 的动态召回需要当前 query,走 MemoryMiddleware.before_model。
context_content = manager.build_system_prompt()
if not context_content.strip():
return ""
return f"{context_content}\n"
context_content = manager.build_system_prompt()
if not context_content.strip():
return ""
return f"{context_content}\n"
except Exception:
logger.exception("Failed to load memory context")
return ""
def _render_skill_block(name: str, description: str, category: str, location: str, display_hint: str) -> str:
return f" <skill>\n <name>{name}</name>\n <description>{description} {_skill_mutability_label(category)}</description>\n <location>{location}</location>{display_hint}\n </skill>"
@lru_cache(maxsize=32)
def _get_cached_skills_prompt_section(
skill_signature: tuple[tuple[str, str, str, str], ...],
skill_signature: tuple[tuple[str, str, str, str, bool], ...],
display_signature: tuple[tuple[str, str, str], ...],
available_skills_key: tuple[str, ...] | None,
container_base_path: str,
skill_evolution_section: str,
compression_enabled: bool,
keep_index: bool,
) -> str:
display_by_skill = {name: (mode, custom_instruction) for name, mode, custom_instruction in display_signature}
filtered = [(name, description, category, location) for name, description, category, location in skill_signature if available_skills_key is None or name in available_skills_key]
filtered = [(name, description, category, location, always_on) for name, description, category, location, always_on in skill_signature if available_skills_key is None or name in available_skills_key]
def _display_hint(name: str) -> str:
display_mode, custom_instruction = display_by_skill.get(name, ("", ""))
if display_mode:
return f"\n <display mode=\"{display_mode}\">{custom_instruction}</display>"
return ""
skills_list = ""
compression_note = ""
if filtered:
items = []
for name, description, category, location in filtered:
display_mode, custom_instruction = display_by_skill.get(name, ("", ""))
display_hint = ""
if display_mode:
display_hint = f"\n <display mode=\"{display_mode}\">{custom_instruction}</display>"
items.append(
f" <skill>\n <name>{name}</name>\n <description>{description} {_skill_mutability_label(category)}</description>\n <location>{location}</location>{display_hint}\n </skill>"
)
skill_items = "\n".join(items)
skills_list = f"<available_skills>\n{skill_items}\n</available_skills>"
if compression_enabled:
# 压缩开启:常驻(always_on)技能全文注入;其余技能从提示词正文剔除,
# 仅保留可选的"名字索引",由 agent 通过 search_skills 工具按需检索。
resident = [s for s in filtered if s[4]]
compressed = [s for s in filtered if not s[4]]
blocks: list[str] = []
for name, description, category, location, _ in resident:
blocks.append(_render_skill_block(name, description, category, location, _display_hint(name)))
if blocks:
skills_list = "<available_skills>\n" + "\n".join(blocks) + "\n</available_skills>"
if compressed:
index_block = ""
if keep_index:
index_items = "\n".join(f" <skill><name>{name}</name></skill>" for name, _, _, _, _ in compressed)
index_block = f"\n<compressed_skills>\n{index_items}\n</compressed_skills>"
compression_note = (
"\n**Additional skills are available but not listed in full above to keep this prompt compact.**\n"
"When a user query might match a skill that is NOT fully described above, call the `search_skills` "
"tool with a short keyword query to discover matching skills (it returns each match's description and "
f"`location`), then `read_file` the returned `location` and follow the skill. There are {len(compressed)} "
"such compressed skill(s)."
f"{index_block}\n"
)
else:
items = [
_render_skill_block(name, description, category, location, _display_hint(name))
for name, description, category, location, _ in filtered
]
skills_list = "<available_skills>\n" + "\n".join(items) + "\n</available_skills>"
# Restricted mode: agent has an explicit skill allowlist.
# We deliberately omit the "Skills are located at: <path>" hint and the
@@ -842,7 +874,7 @@ You have access to skills that provide optimized workflows for specific tasks. E
4. Load referenced resources only when needed during execution
5. Follow the skill's instructions precisely
{display_pattern}
{location_hint}{evolution_block}
{location_hint}{evolution_block}{compression_note}
{skills_list}
</skill_system>"""
@@ -877,13 +909,27 @@ def get_skills_prompt_section(
is_restricted = available_skills is not None
effective_skill_evolution_enabled = skill_evolution_enabled and not is_restricted
# 技能压缩仅作用于默认 lead agent 的全局技能集;自定义 agent 已有显式白名单
# (技能数量本就受控),不参与压缩。
compression_enabled = False
keep_index = True
if not is_restricted:
try:
from deerflow.config.system_settings import get_skill_compression_settings
compression_settings = get_skill_compression_settings()
compression_enabled = compression_settings.enabled
keep_index = compression_settings.keep_index_in_prompt
except Exception:
logger.debug("Failed to read skill compression settings for prompt injection", exc_info=True)
if not skills and not effective_skill_evolution_enabled:
return ""
if available_skills is not None and not any(skill.name in available_skills for skill in skills):
return ""
skill_signature = tuple((skill.name, skill.description, skill.category, skill.get_container_file_path(container_base_path)) for skill in skills)
skill_signature = tuple((skill.name, skill.description, skill.category, skill.get_container_file_path(container_base_path), skill.always_on) for skill in skills)
display_signature: tuple[tuple[str, str, str], ...] = ()
if display_enabled:
try:
@@ -907,7 +953,7 @@ def get_skills_prompt_section(
if not skill_signature and available_key is not None:
return ""
skill_evolution_section = _build_skill_evolution_section(effective_skill_evolution_enabled)
return _get_cached_skills_prompt_section(skill_signature, display_signature, available_key, container_base_path, skill_evolution_section)
return _get_cached_skills_prompt_section(skill_signature, display_signature, available_key, container_base_path, skill_evolution_section, compression_enabled, keep_index)
def get_agent_soul(agent_id: str | None) -> str:
@@ -37,6 +37,15 @@ logger = logging.getLogger(__name__)
# many threads can't grow it without limit. Oldest entries are evicted first.
_SUMMARY_CACHE_MAX_THREADS = 512
# After compaction the preserved window must land *comfortably* below every
# trigger threshold, otherwise the "compressed" context is still over the
# trigger and re-summarizes on the very next turn (the sticky boundary never
# sticks). We force the kept suffix down to this fraction of each trigger so a
# re-compaction only happens once meaningfully *new* content accumulates — not
# on every follow-up question. ``keep`` only sets a *target*; this is the hard
# headroom guarantee on top of it.
_KEEP_HEADROOM_FRACTION = 0.75
@dataclass(frozen=True)
class SummarizationEvent:
@@ -319,6 +328,15 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
if cutoff_index <= 0:
return None
# Force the preserved window below every trigger (with headroom) so the
# post-compaction context doesn't immediately re-trigger. Without this,
# a ``keep`` expressed in *messages* (e.g. 20) can itself blow past a
# *token* trigger (e.g. 60000) when recent messages are large (tool
# outputs, pasted docs) — and then every follow-up question re-summarizes.
cutoff_index = self._apply_stickiness_headroom(messages, cutoff_index)
if cutoff_index <= 0:
return None
messages_to_summarize, preserved = self._partition_with_skill_rescue(messages, cutoff_index)
if not messages_to_summarize:
return None
@@ -368,6 +386,62 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
self._summary_cache.popitem(last=False)
def _apply_stickiness_headroom(self, messages: list[AnyMessage], cutoff_index: int) -> int:
"""Advance ``cutoff_index`` (keep fewer recent messages) so the preserved
suffix lands below every trigger threshold with headroom.
Only ever *raises* the cutoff (summarizes more / keeps less); never keeps
more than the caller's ``keep`` policy already chose. This is what makes
the sticky boundary actually stick: after compaction the kept context is
guaranteed under ~``_KEEP_HEADROOM_FRACTION`` of each trigger, so a
re-compaction waits for genuinely new content instead of firing every turn.
"""
n = len(messages)
if n <= 1:
return cutoff_index
for kind, value in getattr(self, "_trigger_conditions", []):
if kind == "messages":
keep_cap = max(1, int(int(value) * _KEEP_HEADROOM_FRACTION))
cutoff_index = max(cutoff_index, n - keep_cap)
elif kind == "tokens":
target = max(1, int(int(value) * _KEEP_HEADROOM_FRACTION))
cutoff_index = max(cutoff_index, self._suffix_token_cutoff(messages, target))
# "fraction" triggers depend on model profile limits; the message/token
# caps above (when present) already bound the window, so skip them here.
# Always keep at least the final message, and re-snap to a safe AI/Tool
# boundary so we never orphan a tool result.
cutoff_index = min(cutoff_index, n - 1)
return self._find_safe_cutoff_point(messages, cutoff_index)
def _suffix_token_cutoff(self, messages: list[AnyMessage], target_tokens: int) -> int:
"""Smallest index ``i`` such that ``token_counter(messages[i:]) <= target_tokens``.
Binary search mirroring the stock token-based cutoff; returns 0 when the
whole list already fits, and never returns the full length (keeps ≥1).
"""
n = len(messages)
if n == 0:
return 0
if self.token_counter(messages) <= target_tokens:
return 0
left, right = 0, n
cutoff = n
for _ in range(n.bit_length() + 1):
if left >= right:
break
mid = (left + right) // 2
if self.token_counter(messages[mid:]) <= target_tokens:
cutoff = mid
right = mid
else:
left = mid + 1
if cutoff == n:
cutoff = left
return min(cutoff, n - 1)
@staticmethod
def _resolve_request_thread_id(request: ModelRequest) -> str | None:
"""Best-effort thread id from the model request's runtime."""
@@ -840,13 +840,13 @@ class DeerFlowClient:
current_config = get_extensions_config()
config_data = {
"mcpServers": mcp_servers,
"skills": {
name: skill.model_dump(by_alias=True, exclude_none=True)
for name, skill in current_config.skills.items()
},
}
config_data = {
"mcpServers": mcp_servers,
"skills": {
name: skill.model_dump(by_alias=True, exclude_none=True)
for name, skill in current_config.skills.items()
},
}
self._atomic_write_json(config_path, config_data)
@@ -906,20 +906,21 @@ class DeerFlowClient:
if config_path is None:
raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
extensions_config = get_extensions_config()
current_skill_config = extensions_config.skills.get(name)
extensions_config.skills[name] = SkillStateConfig(
enabled=enabled,
display=current_skill_config.display if current_skill_config else None,
)
config_data = {
"mcpServers": {n: s.model_dump(by_alias=True) for n, s in extensions_config.mcp_servers.items()},
"skills": {
n: sc.model_dump(by_alias=True, exclude_none=True)
for n, sc in extensions_config.skills.items()
},
}
extensions_config = get_extensions_config()
current_skill_config = extensions_config.skills.get(name)
extensions_config.skills[name] = SkillStateConfig(
enabled=enabled,
always_on=current_skill_config.always_on if current_skill_config else False,
display=current_skill_config.display if current_skill_config else None,
)
config_data = {
"mcpServers": {n: s.model_dump(by_alias=True) for n, s in extensions_config.mcp_servers.items()},
"skills": {
n: sc.model_dump(by_alias=True, exclude_none=True)
for n, sc in extensions_config.skills.items()
},
}
self._atomic_write_json(config_path, config_data)
@@ -161,6 +161,10 @@ class SkillStateConfig(BaseModel):
"""Configuration for a single skill's state."""
enabled: bool = Field(default=True, description="Whether this skill is enabled")
always_on: bool = Field(
default=False,
description="常驻(免压缩):开启技能压缩后,本技能仍以全文形式注入系统提示词,而非降级为索引+search_skills 检索",
)
display: SkillDisplayConfig | None = Field(default=None, description="Optional frontend display adapter")
@@ -313,6 +317,21 @@ class ExtensionsConfig(BaseModel):
return skill_category in ("public", "custom")
return skill_config.enabled
def is_skill_always_on(self, skill_name: str) -> bool:
"""Check if a skill is marked 常驻(免压缩).
Args:
skill_name: Name of the skill
Returns:
True if the skill should bypass skill compression, False otherwise
(default for skills without an explicit config entry).
"""
skill_config = self.skills.get(skill_name)
if skill_config is None:
return False
return skill_config.always_on
_extensions_config: ExtensionsConfig | None = None
@@ -170,6 +170,26 @@ class LeaderboardSettings(BaseModel):
model_config = ConfigDict(extra="ignore")
class SkillCompressionSettings(BaseModel):
"""Admin-configurable skill compression (技能压缩).
When ``enabled``, only skills explicitly marked 常驻(免压缩) (``always_on``)
are injected full-text into the lead agent's system prompt. Every other
enabled skill is "compressed": kept out of the prompt body and surfaced to
the agent on demand via the ``search_skills`` keyword-retrieval tool (and a
compact name index when ``keep_index_in_prompt`` is true). When disabled the
behaviour is unchanged — every enabled skill is injected full-text.
"""
enabled: bool = Field(default=False, description="Whether skill compression is active (master switch)")
search_top_n: int = Field(default=5, ge=1, le=50, description="Max skills returned by the search_skills tool")
keep_index_in_prompt: bool = Field(
default=True,
description="Whether compressed skills still appear as a compact name-only index in the prompt",
)
model_config = ConfigDict(extra="ignore")
class SystemSettings(BaseModel):
"""Top-level system settings document."""
@@ -179,6 +199,7 @@ class SystemSettings(BaseModel):
appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings)
skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings)
model_config = ConfigDict(extra="ignore")
@@ -204,6 +225,15 @@ def load_system_settings() -> SystemSettings:
return SystemSettings()
def get_skill_compression_settings() -> SkillCompressionSettings:
"""Convenience accessor for the skill-compression block (safe defaults on error)."""
try:
return load_system_settings().skill_compression
except Exception:
logger.debug("Failed to read skill compression settings; using defaults", exc_info=True)
return SkillCompressionSettings()
def save_system_settings(settings: SystemSettings) -> SystemSettings:
"""Persist the settings document atomically, returning the saved value."""
path = _settings_path()
@@ -120,14 +120,18 @@ def related_links(project_name: str | None) -> list[str]: # noqa: ARG001
_related_links = related_links
def _related_section_md(related_lines: list[str], source_lines: list[str]) -> str:
"""Render the ``## Related`` section, or ``""`` when there is nothing to show."""
if not related_lines and not source_lines:
def _related_section_md(related_lines: list[str], source_lines: list[str] | None = None) -> str: # noqa: ARG001
"""Render the ``## Related`` section, or ``""`` when there is nothing to show.
Sources are intentionally **not** rendered into the body anymore. The note
detail page already shows every source once via the structured ``来源`` block
(``note.sources``); emitting a ``### Sources`` sub-section here duplicated that
list (and broke the frontend's clean ``## Related`` extraction). The
``source_lines`` argument is kept for backward compatibility and ignored.
"""
if not related_lines:
return ""
inner = "\n".join(related_lines)
if source_lines:
sources_block = "### Sources\n\n" + "\n".join(source_lines)
inner = (inner + "\n\n" + sources_block) if inner else sources_block
return f"## Related\n\n{inner}\n"
@@ -283,13 +283,16 @@ async def extract_thread_llm_multi(
max_input_chars: int = 12000,
allow_multi: bool = True,
max_notes: int = 4,
system_prompt: str | None = None,
) -> list[NoteDraft]:
"""Distill a thread into one *or several* cross-linkable knowledge notes.
The LLM may split a conversation that covers multiple independent topics into
several declarative notes. Returns ``[]`` when there is nothing to sediment
or the LLM path fails — callers should fall back to the rule-based extractor.
When ``allow_multi`` is false only the first note is kept.
When ``allow_multi`` is false only the first note is kept. ``system_prompt``
overrides the built-in distillation prompt with an operator-edited template
(used as-is, since the template is expected to be self-contained).
"""
turns = extractor.split_turns(messages, max_messages=max_messages)
if not turns:
@@ -303,8 +306,9 @@ async def extract_thread_llm_multi(
model = create_chat_model(name=model_name or None, thinking_enabled=False)
transcript = _build_transcript(turns, sources, max_chars=max_input_chars)
system_text = (system_prompt or "").strip() or (_SYSTEM_PROMPT + "\n\n" + WIKI_CAPTURE_GUIDANCE)
prompt = [
SystemMessage(content=_SYSTEM_PROMPT + "\n\n" + WIKI_CAPTURE_GUIDANCE),
SystemMessage(content=system_text),
HumanMessage(content=f"以下是一次对话,请按规范提炼为知识 JSON:\n\n{transcript}"),
]
response = await model.ainvoke(prompt)
@@ -375,6 +379,7 @@ async def extract_document_llm_multi(
max_input_chars: int = 12000,
allow_multi: bool = True,
max_notes: int = 4,
system_prompt: str | None = None,
) -> list[NoteDraft]:
"""Distill an imported **document** into one *or several* knowledge notes.
@@ -397,8 +402,9 @@ async def extract_document_llm_multi(
model = create_chat_model(name=model_name or None, thinking_enabled=False)
body = text if len(text) <= max_input_chars else text[:max_input_chars] + "\n…(已截断)"
header = f"文档标题:{doc_title}\n\n" if doc_title else ""
system_text = (system_prompt or "").strip() or (_DOC_SYSTEM_PROMPT + "\n\n" + WIKI_CAPTURE_GUIDANCE)
prompt = [
SystemMessage(content=_DOC_SYSTEM_PROMPT + "\n\n" + WIKI_CAPTURE_GUIDANCE),
SystemMessage(content=system_text),
HumanMessage(content=f"以下是一篇文档,请按规范提炼为知识 JSON:\n\n{header}{body}"),
]
response = await model.ainvoke(prompt)
@@ -33,6 +33,7 @@ class KnowledgeNoteRow(Base):
__table_args__ = (
Index("ix_knowledge_notes_vault_path", "vault_path", mysql_length=191),
Index("ix_knowledge_notes_source_id", "source_id", mysql_length=191),
Index("ix_knowledge_notes_folder", "folder", mysql_length=191),
)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
@@ -40,6 +41,10 @@ class KnowledgeNoteRow(Base):
summary: Mapped[str | None] = mapped_column(Text)
content_md: Mapped[str] = mapped_column(PortableLongText())
vault_path: Mapped[str | None] = mapped_column(String(1024))
# User-assigned directory (slash-joined path, e.g. "投研/行业"). Independent
# of the wiki-capture ``category`` and of ``vault_path``; drives the product
# directory tree. NULL = unfiled (shown under a fallback group).
folder: Mapped[str | None] = mapped_column(String(512))
source_type: Mapped[str] = mapped_column(String(64), index=True)
source_id: Mapped[str | None] = mapped_column(String(256))
status: Mapped[str] = mapped_column(String(32), default="approved", index=True)
@@ -170,3 +175,49 @@ class KnowledgeRelationRow(Base):
relation_type: Mapped[str] = mapped_column(String(128))
weight: Mapped[float] = mapped_column(Float, default=1.0)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
class KnowledgeFolderRow(Base):
"""User-managed knowledge directory (mirrors ``knowledge_folders``).
A folder is identified by its full slash-joined ``path`` (e.g. ``投研/行业``);
nesting is derived by splitting on ``/`` on the client. Folders exist
independently of notes so an operator can create an empty directory and file
notes into it later (``knowledge_notes.folder`` stores the chosen path).
Global/shared like the rest of the KB; ``created_by`` is audit-only.
"""
__tablename__ = "knowledge_folders"
__table_args__ = (Index("uq_knowledge_folders_path", "path", unique=True, mysql_length=191),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
path: Mapped[str] = mapped_column(String(512))
name: Mapped[str] = mapped_column(String(256))
sort_order: Mapped[int] = mapped_column(Integer, default=0)
created_by: Mapped[str | None] = mapped_column(String(128))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
class KnowledgeExtractTemplateRow(Base):
"""Editable knowledge-extraction prompt template (mirrors ``knowledge_extract_templates``).
Lets an operator open up, edit and add multiple distillation prompts instead
of the hard-coded constants. ``scope`` is ``thread`` / ``document`` / ``both``;
at capture time the service resolves the chosen template (or the default for
that scope) and feeds its ``system_prompt`` to the extractor. Global/shared.
"""
__tablename__ = "knowledge_extract_templates"
__table_args__ = (Index("uq_knowledge_extract_templates_name", "name", unique=True, mysql_length=191),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(256))
description: Mapped[str | None] = mapped_column(Text)
scope: Mapped[str] = mapped_column(String(16), default="both", index=True)
system_prompt: Mapped[str] = mapped_column(PortableLongText())
is_default: Mapped[bool] = mapped_column(default=False, index=True)
enabled: Mapped[bool] = mapped_column(default=True)
created_by: Mapped[str | None] = mapped_column(String(128))
updated_by: Mapped[str | None] = mapped_column(String(128))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True)
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -18,6 +18,8 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.knowledge.models import (
KnowledgeEmbeddingRow,
KnowledgeEntityRow,
KnowledgeExtractTemplateRow,
KnowledgeFolderRow,
KnowledgeGraphBlacklistRow,
KnowledgeNoteRow,
KnowledgeNoteTagRow,
@@ -127,6 +129,7 @@ class KnowledgeRepository:
status: str,
confidence: float = 0.0,
vault_path: str | None = None,
folder: str | None = None,
tags: list[str] | None = None,
created_by: str | None = None,
) -> dict[str, Any]:
@@ -138,6 +141,7 @@ class KnowledgeRepository:
summary=summary,
content_md=content_md,
vault_path=vault_path,
folder=(folder or None),
source_type=source_type,
source_id=source_id,
status=status,
@@ -175,6 +179,8 @@ class KnowledgeRepository:
status: str | None = None,
tags: list[str] | None = None,
vault_path: str | None = None,
folder: str | None = None,
set_folder: bool = False,
updated_by: str | None = None,
) -> dict[str, Any] | None:
async with self._sf() as session:
@@ -191,6 +197,10 @@ class KnowledgeRepository:
row.status = status
if vault_path is not None:
row.vault_path = vault_path
# ``set_folder`` lets the caller assign *or clear* (folder=None) the
# directory; a bare ``folder=None`` without the flag leaves it as-is.
if set_folder:
row.folder = (folder or None)
if updated_by is not None:
row.updated_by = updated_by
row.updated_at = datetime.now(UTC)
@@ -279,6 +289,7 @@ class KnowledgeRepository:
entity: str | None = None,
source_type: str | None = None,
status: str | None = None,
folder: str | None = None,
include_archived: bool = False,
limit: int = 20,
offset: int = 0,
@@ -294,6 +305,12 @@ class KnowledgeRepository:
conditions.append(KnowledgeNoteRow.status != "archived")
if source_type:
conditions.append(KnowledgeNoteRow.source_type == source_type)
if folder is not None:
# Exact folder, plus any nested sub-folders (``folder/…``).
if folder == "":
conditions.append(KnowledgeNoteRow.folder.is_(None))
else:
conditions.append(or_(KnowledgeNoteRow.folder == folder, KnowledgeNoteRow.folder.like(f"{folder}/%")))
if keyword:
like = f"%{keyword}%"
conditions.append(or_(KnowledgeNoteRow.title.like(like), KnowledgeNoteRow.summary.like(like), KnowledgeNoteRow.content_md.like(like)))
@@ -581,3 +598,184 @@ class KnowledgeRepository:
)
)
await session.commit()
# -- folders ----------------------------------------------------------
async def list_folders(self) -> list[dict[str, Any]]:
"""Return every managed directory, ordered by sort_order then path."""
async with self._sf() as session:
stmt = select(KnowledgeFolderRow).order_by(KnowledgeFolderRow.sort_order, KnowledgeFolderRow.path)
return [{"id": r.id, "path": r.path, "name": r.name, "sort_order": r.sort_order} for r in (await session.execute(stmt)).scalars()]
async def create_folder(self, path: str, *, created_by: str | None = None) -> dict[str, Any]:
"""Create a directory by its full slash-joined path (idempotent on path).
Parent directories are auto-created so ``投研/行业/寿险`` materializes all
three levels. Returns the leaf folder row.
"""
segments = [s.strip() for s in (path or "").split("/") if s.strip()]
if not segments:
raise ValueError("folder path is empty")
async with self._sf() as session:
acc: list[str] = []
leaf: KnowledgeFolderRow | None = None
for i, seg in enumerate(segments):
acc.append(seg)
full = "/".join(acc)
row = await self._folder_by_path(session, full)
if row is None:
row = KnowledgeFolderRow(id=_new_id("kbdir"), path=full, name=seg, sort_order=i, created_by=created_by, created_at=datetime.now(UTC))
session.add(row)
await session.flush()
leaf = row
# Capture before commit: the async session expires attributes on
# commit, and re-reading them outside an awaited load would error.
result = {"id": leaf.id, "path": leaf.path, "name": leaf.name, "sort_order": leaf.sort_order}
await session.commit()
return result
@staticmethod
async def _folder_by_path(session: AsyncSession, path: str) -> KnowledgeFolderRow | None:
return (await session.execute(select(KnowledgeFolderRow).where(KnowledgeFolderRow.path == path))).scalars().first()
async def rename_folder(self, folder_id: str, new_path: str) -> dict[str, Any] | None:
"""Rename/move a folder and re-point every descendant folder + note.
Both the folder rows under ``old/…`` and notes whose ``folder`` is ``old``
or ``old/…`` are rewritten to the new prefix, so the move is atomic.
"""
new_path = "/".join(s.strip() for s in (new_path or "").split("/") if s.strip())
if not new_path:
raise ValueError("folder path is empty")
async with self._sf() as session:
row = await session.get(KnowledgeFolderRow, folder_id)
if row is None:
return None
old = row.path
if new_path == old:
return {"id": row.id, "path": row.path, "name": row.name, "sort_order": row.sort_order}
for f in (await session.execute(select(KnowledgeFolderRow).where(or_(KnowledgeFolderRow.path == old, KnowledgeFolderRow.path.like(f"{old}/%"))))).scalars():
f.path = new_path + f.path[len(old):]
f.name = f.path.split("/")[-1]
for n in (await session.execute(select(KnowledgeNoteRow).where(or_(KnowledgeNoteRow.folder == old, KnowledgeNoteRow.folder.like(f"{old}/%"))))).scalars():
n.folder = new_path + (n.folder[len(old):] if n.folder else "")
await session.commit()
refreshed = await session.get(KnowledgeFolderRow, folder_id)
return {"id": refreshed.id, "path": refreshed.path, "name": refreshed.name, "sort_order": refreshed.sort_order}
async def delete_folder(self, folder_id: str, *, reassign_to: str | None = None) -> bool:
"""Delete a folder (and its sub-folders); notes inside move to ``reassign_to`` (or unfiled)."""
async with self._sf() as session:
row = await session.get(KnowledgeFolderRow, folder_id)
if row is None:
return False
old = row.path
target = (reassign_to or "").strip() or None
for n in (await session.execute(select(KnowledgeNoteRow).where(or_(KnowledgeNoteRow.folder == old, KnowledgeNoteRow.folder.like(f"{old}/%"))))).scalars():
n.folder = target
await session.execute(delete(KnowledgeFolderRow).where(or_(KnowledgeFolderRow.path == old, KnowledgeFolderRow.path.like(f"{old}/%"))))
await session.commit()
return True
# -- extraction templates ---------------------------------------------
@staticmethod
def _template_to_dict(row: KnowledgeExtractTemplateRow) -> dict[str, Any]:
d = row.to_dict()
d["created_at"] = _iso(d.get("created_at"))
d["updated_at"] = _iso(d.get("updated_at"))
return d
async def list_templates(self, *, scope: str | None = None, enabled_only: bool = False) -> list[dict[str, Any]]:
async with self._sf() as session:
stmt = select(KnowledgeExtractTemplateRow)
if scope:
stmt = stmt.where(or_(KnowledgeExtractTemplateRow.scope == scope, KnowledgeExtractTemplateRow.scope == "both"))
if enabled_only:
stmt = stmt.where(KnowledgeExtractTemplateRow.enabled.is_(True))
stmt = stmt.order_by(KnowledgeExtractTemplateRow.is_default.desc(), KnowledgeExtractTemplateRow.created_at)
return [self._template_to_dict(r) for r in (await session.execute(stmt)).scalars()]
async def get_template(self, template_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(KnowledgeExtractTemplateRow, template_id)
return self._template_to_dict(row) if row else None
async def get_default_template(self, scope: str) -> dict[str, Any] | None:
"""Return the default enabled template applicable to ``scope`` (thread/document)."""
async with self._sf() as session:
stmt = (
select(KnowledgeExtractTemplateRow)
.where(KnowledgeExtractTemplateRow.enabled.is_(True))
.where(KnowledgeExtractTemplateRow.is_default.is_(True))
.where(or_(KnowledgeExtractTemplateRow.scope == scope, KnowledgeExtractTemplateRow.scope == "both"))
.limit(1)
)
row = (await session.execute(stmt)).scalars().first()
return self._template_to_dict(row) if row else None
async def template_name_taken(self, name: str, *, exclude_id: str | None = None) -> bool:
async with self._sf() as session:
stmt = select(KnowledgeExtractTemplateRow.id).where(func.lower(KnowledgeExtractTemplateRow.name) == (name or "").strip().lower())
if exclude_id:
stmt = stmt.where(KnowledgeExtractTemplateRow.id != exclude_id)
return (await session.execute(stmt)).first() is not None
async def create_template(self, *, name: str, system_prompt: str, scope: str = "both", description: str | None = None, is_default: bool = False, enabled: bool = True, created_by: str | None = None) -> dict[str, Any]:
async with self._sf() as session:
if is_default:
await self._clear_default(session, scope)
row = KnowledgeExtractTemplateRow(
id=_new_id("kbtpl"),
name=name.strip()[:256],
description=description,
scope=scope,
system_prompt=system_prompt,
is_default=is_default,
enabled=enabled,
created_by=created_by,
updated_by=created_by,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
session.add(row)
await session.commit()
await session.refresh(row)
return self._template_to_dict(row)
async def update_template(self, template_id: str, *, name: str | None = None, system_prompt: str | None = None, scope: str | None = None, description: str | None = None, is_default: bool | None = None, enabled: bool | None = None, updated_by: str | None = None) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(KnowledgeExtractTemplateRow, template_id)
if row is None:
return None
if name is not None:
row.name = name.strip()[:256]
if system_prompt is not None:
row.system_prompt = system_prompt
if scope is not None:
row.scope = scope
if description is not None:
row.description = description
if enabled is not None:
row.enabled = enabled
if is_default:
await self._clear_default(session, scope or row.scope)
row.is_default = True
elif is_default is False:
row.is_default = False
row.updated_by = updated_by
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return self._template_to_dict(row)
async def delete_template(self, template_id: str) -> bool:
async with self._sf() as session:
result = await session.execute(delete(KnowledgeExtractTemplateRow).where(KnowledgeExtractTemplateRow.id == template_id))
await session.commit()
return bool(result.rowcount)
@staticmethod
async def _clear_default(session: AsyncSession, scope: str) -> None:
"""Unset is_default on any template that would collide with a new default for ``scope``."""
scopes = {scope, "both"} if scope != "both" else {"thread", "document", "both"}
for row in (await session.execute(select(KnowledgeExtractTemplateRow).where(KnowledgeExtractTemplateRow.is_default.is_(True)).where(KnowledgeExtractTemplateRow.scope.in_(scopes)))).scalars():
row.is_default = False
@@ -48,6 +48,8 @@ class NoteDraft:
source_id: str | None = None
status: str = "approved"
confidence: float = 0.0
# User-assigned directory (slash-joined path); None = unfiled.
folder: str | None = None
tags: list[str] = field(default_factory=list)
key_points: list[str] = field(default_factory=list)
actions: list[str] = field(default_factory=list)
@@ -145,6 +145,7 @@ class KnowledgeService:
source_id=draft.source_id,
status=draft.status,
confidence=draft.confidence,
folder=draft.folder,
tags=draft.tags,
created_by=created_by,
)
@@ -223,15 +224,21 @@ class KnowledgeService:
created_by: str | None = None,
skip_existing: bool = False,
model_name: str | None = None,
template_id: str | None = None,
folder: str | None = None,
) -> dict[str, Any]:
"""Sediment a thread into a knowledge note.
``model_name`` optionally overrides the configured extraction model for
this one capture (falls back to ``config.extract_model_name``).
``template_id`` selects an operator-edited extraction prompt template
(falls back to the ``thread`` default, then the built-in prompt).
``folder`` files the resulting note(s) under a user directory.
Returns ``{"note": <note>, "created": bool}``. Raises
:class:`EmptyThreadError` when there is nothing worth keeping.
"""
extract_prompt = await self._resolve_extract_prompt(template_id, "thread")
# Always-on entry log: proves a capture was triggered for this thread,
# regardless of how it ends (success / skip / fallback). Lets you tell
# "capture never ran" apart from "logs not visible" via one grep.
@@ -276,6 +283,7 @@ class KnowledgeService:
model_name=model_name or self.config.extract_model_name,
max_input_chars=self.config.extract_max_input_chars,
allow_multi=self.config.multi_note_enabled,
system_prompt=extract_prompt,
)
except llm_extractor.LLMRejectedIngest as exc:
# The model judged this conversation worthless (greeting / meta
@@ -324,6 +332,8 @@ class KnowledgeService:
normalized = normalize_status(effective_status)
for draft in drafts:
draft.status = normalized
if folder:
draft.folder = folder
# Cross-link sibling notes from the same conversation by title, so the
# split pages reference each other (resolved once all are persisted).
if len(drafts) > 1:
@@ -370,6 +380,7 @@ class KnowledgeService:
tags: list[str] | None = None,
status: str = "approved",
summary: str | None = None,
folder: str | None = None,
created_by: str | None = None,
) -> dict[str, Any]:
"""Create a note directly from product-edited Markdown."""
@@ -388,6 +399,7 @@ class KnowledgeService:
source_type="manual",
source_id=None,
status=normalize_status(status),
folder=(folder or None),
tags=base_tags,
confidence=0.5,
related=extractor.related_links(self.project_name),
@@ -431,6 +443,8 @@ class KnowledgeService:
created_by: str | None = None,
model_name: str | None = None,
skip_existing: bool = False,
template_id: str | None = None,
folder: str | None = None,
) -> dict[str, Any]:
"""Import a file (already converted to Markdown/text) into knowledge notes.
@@ -468,6 +482,7 @@ class KnowledgeService:
logger.info("[knowledge] file '%s' already imported (note %s); skipping", filename, existing.get("id"))
return {"note": existing, "created": False, "notes": [existing], "llm_fallback": False}
extract_prompt = await self._resolve_extract_prompt(template_id, "document")
drafts: list[NoteDraft] = []
llm_fallback = False
if self.config.llm_extract_enabled:
@@ -482,6 +497,7 @@ class KnowledgeService:
model_name=model_name or self.config.extract_model_name,
max_input_chars=self.config.extract_max_input_chars,
allow_multi=self.config.multi_note_enabled,
system_prompt=extract_prompt,
)
except llm_extractor.LLMRejectedIngest as exc:
raise EmptyThreadError(f"File {filename} judged not worth ingesting: {exc}") from exc
@@ -496,6 +512,8 @@ class KnowledgeService:
normalized = normalize_status(effective_status)
for draft in drafts:
draft.status = normalized
if folder:
draft.folder = folder
# Cross-link sibling notes split from the same document (resolved by title
# once all are persisted), mirroring multi-note thread capture.
@@ -526,6 +544,7 @@ class KnowledgeService:
entity: str | None = None,
source_type: str | None = None,
status: str | None = None,
folder: str | None = None,
limit: int = 20,
offset: int = 0,
) -> dict[str, Any]:
@@ -535,6 +554,7 @@ class KnowledgeService:
entity=entity,
source_type=normalize_source_type(source_type) if source_type else None,
status=status,
folder=folder,
limit=limit,
offset=offset,
)
@@ -555,6 +575,8 @@ class KnowledgeService:
tags: list[str] | None = None,
status: str | None = None,
summary: str | None = None,
folder: str | None = None,
set_folder: bool = False,
updated_by: str | None = None,
) -> dict[str, Any] | None:
current = await self.repo.get_note(note_id, with_sources=False)
@@ -576,6 +598,8 @@ class KnowledgeService:
content_md=body,
status=normalize_status(status) if status else None,
tags=tags,
folder=folder,
set_folder=set_folder,
updated_by=updated_by,
)
if note is None:
@@ -604,6 +628,84 @@ class KnowledgeService:
async def list_versions(self, note_id: str) -> list[dict[str, Any]]:
return await self.repo.list_versions(note_id)
# -- folders ----------------------------------------------------------
async def list_folders(self) -> list[dict[str, Any]]:
return await self.repo.list_folders()
async def create_folder(self, path: str, *, created_by: str | None = None) -> dict[str, Any]:
return await self.repo.create_folder(path, created_by=created_by)
async def rename_folder(self, folder_id: str, new_path: str) -> dict[str, Any] | None:
return await self.repo.rename_folder(folder_id, new_path)
async def delete_folder(self, folder_id: str, *, reassign_to: str | None = None) -> bool:
return await self.repo.delete_folder(folder_id, reassign_to=reassign_to)
# -- extraction templates ---------------------------------------------
async def _resolve_extract_prompt(self, template_id: str | None, scope: str) -> str | None:
"""Return the system prompt for a capture, or ``None`` for the built-in default.
Resolution: explicit ``template_id`` → the scope's default template →
``None`` (the extractor then uses its hard-coded prompt). Best-effort: any
lookup failure degrades to the built-in prompt so capture never breaks.
"""
try:
if template_id:
tpl = await self.repo.get_template(template_id)
if tpl and tpl.get("enabled") and (tpl.get("system_prompt") or "").strip():
return tpl["system_prompt"]
default = await self.repo.get_default_template(scope)
if default and (default.get("system_prompt") or "").strip():
return default["system_prompt"]
except Exception:
logger.exception("Failed to resolve knowledge extraction template (id=%s, scope=%s)", template_id, scope)
return None
async def list_extract_templates(self, *, scope: str | None = None, enabled_only: bool = False) -> list[dict[str, Any]]:
return await self.repo.list_templates(scope=scope, enabled_only=enabled_only)
async def get_extract_template(self, template_id: str) -> dict[str, Any] | None:
return await self.repo.get_template(template_id)
async def create_extract_template(self, **kwargs: Any) -> dict[str, Any]:
return await self.repo.create_template(**kwargs)
async def update_extract_template(self, template_id: str, **kwargs: Any) -> dict[str, Any] | None:
return await self.repo.update_template(template_id, **kwargs)
async def delete_extract_template(self, template_id: str) -> bool:
return await self.repo.delete_template(template_id)
async def template_name_taken(self, name: str, *, exclude_id: str | None = None) -> bool:
return await self.repo.template_name_taken(name, exclude_id=exclude_id)
async def seed_default_templates(self, *, created_by: str | None = None) -> int:
"""Seed the two built-in distillation prompts as editable templates (idempotent).
Runs once on startup so an operator immediately sees — and can edit — the
prompts that drive 对话 / 文档 extraction. Existing templates with the same
name are left untouched. Returns the number of templates created.
"""
from deerflow.knowledge import llm_extractor
guidance = llm_extractor.WIKI_CAPTURE_GUIDANCE
seeds = [
("默认 · 对话提炼", "thread", llm_extractor._SYSTEM_PROMPT + "\n\n" + guidance, "对话沉淀的默认提炼提示词(wiki-capture 规范)。"),
("默认 · 文档提炼", "document", llm_extractor._DOC_SYSTEM_PROMPT + "\n\n" + guidance, "文档导入的默认提炼提示词(wiki-capture 规范)。"),
]
created = 0
for name, scope, prompt, desc in seeds:
try:
if await self.repo.template_name_taken(name):
continue
await self.repo.create_template(name=name, system_prompt=prompt, scope=scope, description=desc, is_default=True, enabled=True, created_by=created_by)
created += 1
except Exception:
logger.exception("Failed to seed knowledge extraction template '%s'", name)
if created:
logger.info("[knowledge] seeded %d default extraction template(s)", created)
return created
async def delete_note(self, note_id: str, *, updated_by: str | None = None, hard: bool = False) -> bool:
"""Soft delete (archive) by default; hard delete removes the vault file."""
current = await self.repo.get_note(note_id, with_sources=False)
@@ -0,0 +1,84 @@
"""Knowledge directories + editable extraction templates.
New column ``knowledge_notes.folder``; new tables ``knowledge_folders`` and
``knowledge_extract_templates``.
Note: the running app also auto-adds the column + tables via
``_ensure_orm_columns_sync`` / ``create_all`` on startup; this migration keeps
the alembic history consistent per the repo convention (see backend/CLAUDE.md →
Schema Migrations).
Revision ID: 20260616_02
Revises: 20260616_01
Create Date: 2026-06-16
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
from sqlalchemy.dialects import mysql
revision = "20260616_02"
down_revision = "20260616_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
return table_name in inspect(op.get_bind()).get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
return any(col["name"] == column_name for col in inspect(op.get_bind()).get_columns(table_name))
def upgrade() -> None:
if _has_table("knowledge_notes") and not _has_column("knowledge_notes", "folder"):
op.add_column("knowledge_notes", sa.Column("folder", sa.String(512), nullable=True))
op.create_index("ix_knowledge_notes_folder", "knowledge_notes", ["folder"], mysql_length=191)
if not _has_table("knowledge_folders"):
op.create_table(
"knowledge_folders",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("path", sa.String(512), nullable=False),
sa.Column("name", sa.String(256), nullable=False),
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
sa.Column("created_by", sa.String(128), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("uq_knowledge_folders_path", "knowledge_folders", ["path"], unique=True, mysql_length=191)
if not _has_table("knowledge_extract_templates"):
op.create_table(
"knowledge_extract_templates",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("name", sa.String(256), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("scope", sa.String(16), nullable=False, server_default="both"),
sa.Column("system_prompt", sa.Text().with_variant(mysql.LONGTEXT(), "mysql"), nullable=False),
sa.Column("is_default", sa.Boolean(), nullable=False, server_default=sa.text("0")),
sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.text("1")),
sa.Column("created_by", sa.String(128), nullable=True),
sa.Column("updated_by", sa.String(128), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_knowledge_extract_templates_scope", "knowledge_extract_templates", ["scope"])
op.create_index("ix_knowledge_extract_templates_is_default", "knowledge_extract_templates", ["is_default"])
op.create_index("ix_knowledge_extract_templates_created_at", "knowledge_extract_templates", ["created_at"])
op.create_index("uq_knowledge_extract_templates_name", "knowledge_extract_templates", ["name"], unique=True, mysql_length=191)
def downgrade() -> None:
if _has_table("knowledge_extract_templates"):
op.drop_table("knowledge_extract_templates")
if _has_table("knowledge_folders"):
op.drop_table("knowledge_folders")
if _has_column("knowledge_notes", "folder"):
op.drop_column("knowledge_notes", "folder")
@@ -28,6 +28,8 @@ from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.knowledge.models import (
KnowledgeEmbeddingRow,
KnowledgeEntityRow,
KnowledgeExtractTemplateRow,
KnowledgeFolderRow,
KnowledgeGraphBlacklistRow,
KnowledgeNoteRow,
KnowledgeNoteTagRow,
@@ -78,6 +80,8 @@ __all__ = [
"BusinessMappingRow",
"KnowledgeEmbeddingRow",
"KnowledgeEntityRow",
"KnowledgeExtractTemplateRow",
"KnowledgeFolderRow",
"KnowledgeGraphBlacklistRow",
"KnowledgeNoteRow",
"KnowledgeNoteTagRow",
@@ -1,8 +1,8 @@
"""ORM models for 岗位 (positions) and their per-resource tag bindings.
A **position** (岗位) is an admin-managed aggregation layer that decides which
agents / skills / scheduled tasks / custom prompts / recommended questions a
user is subscribed to. Each user has at most one position (the link lives on
agents / skills / scheduled tasks / recommended questions a user is subscribed
to. Each user has at most one position (the link lives on
``users.position_id``, a 1:1 relation — see ``UserRow``).
Two tables here:
@@ -26,8 +26,9 @@ from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
# Resource types a position can grant. The first three reuse the existing
# typed-tag namespace (``deerflow.persistence.tags``); ``user_prompt`` and
# ``recommended_question`` are added by the same change (see TAG_TYPES).
# typed-tag namespace (``deerflow.persistence.tags``); ``recommended_question``
# was added alongside (see TAG_TYPES). The removed ``user_prompt`` type is no
# longer accepted — ``get_tag_bindings`` drops any stale rows of that type.
POSITION_RESOURCE_TYPES: tuple[str, ...] = (
"agent",
"skill",
@@ -11,7 +11,7 @@ from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.positions.base import PositionStore
from deerflow.persistence.positions.model import PositionRow, PositionTagBindingRow
from deerflow.persistence.positions.model import POSITION_RESOURCE_TYPES, PositionRow, PositionTagBindingRow
from deerflow.persistence.user.model import UserRow
@@ -134,6 +134,11 @@ class PositionRepository(PositionStore):
)
out: dict[str, list[str]] = {}
for r in rows:
# Drop bindings for resource types no longer supported (e.g. the
# removed ``user_prompt`` type) so stale rows can't poison the
# editor or be echoed back into a 422 on save.
if r.resource_type not in POSITION_RESOURCE_TYPES:
continue
out.setdefault(r.resource_type, []).append(r.tag_id)
return out
@@ -54,11 +54,42 @@ class LocalSkillStorage(SkillStorage):
return self._host_root
def custom_skill_exists(self, name: str) -> bool:
return self.get_custom_skill_file(name).exists()
return self._resolve_skill_dir(name, SkillCategory.CUSTOM) is not None
def public_skill_exists(self, name: str) -> bool:
return self._resolve_skill_dir(name, SkillCategory.PUBLIC) is not None
def _resolve_skill_dir(self, name: str, category: SkillCategory) -> Path | None:
"""Return the on-disk directory of the skill identified by *name*.
Discovery (``_iter_skill_files``/``load_skills``) keys a skill by the
``name`` in its SKILL.md frontmatter, but a hand-dropped or mis-extracted
skill can live at a directory whose name differs from that frontmatter
name. Locating a skill by the canonical ``<category>/<name>`` path alone
would then make it *listable but not deletable/readable*.
Fast path: the canonical ``<category>/<name>`` layout. Fallback: scan the
discovered SKILL.md files and match on the parsed frontmatter ``name`` so
delete/read/editability stay consistent with what discovery surfaces.
Returns ``None`` when no such skill is discoverable.
"""
from deerflow.skills.parser import parse_skill_file
normalized_name = self.validate_skill_name(name)
return (self._host_root / SkillCategory.PUBLIC.value / normalized_name / SKILL_MD_FILE).exists()
canonical = self._host_root / category.value / normalized_name
if (canonical / SKILL_MD_FILE).is_file():
return canonical
for found_category, category_root, md_path in self._iter_skill_files():
if found_category != category:
continue
skill = parse_skill_file(
md_path,
category=found_category,
relative_path=md_path.parent.relative_to(category_root),
)
if skill and skill.name == normalized_name:
return md_path.parent
return None
def _iter_skill_files(self) -> Iterable[tuple[SkillCategory, Path, Path]]:
if not self._host_root.exists():
@@ -71,12 +102,19 @@ class LocalSkillStorage(SkillStorage):
dir_names[:] = sorted(name for name in dir_names if not name.startswith("."))
if SKILL_MD_FILE not in file_names:
continue
# A directory containing SKILL.md *is* a self-contained skill;
# treat it as a leaf and do not descend into its subdirectories.
# This keeps bundled assets that ship their own SKILL.md (e.g.
# ``examples/`` or ``references/`` inside a skill package) from
# being surfaced as separate, undeletable top-level skills.
dir_names[:] = []
yield category, category_path, Path(current_root) / SKILL_MD_FILE
def read_custom_skill(self, name: str) -> str:
if not self.custom_skill_exists(name):
skill_dir = self._resolve_skill_dir(name, SkillCategory.CUSTOM)
if skill_dir is None:
raise FileNotFoundError(f"Custom skill '{name}' not found.")
return (self.get_custom_skill_dir(name) / SKILL_MD_FILE).read_text(encoding="utf-8")
return (skill_dir / SKILL_MD_FILE).read_text(encoding="utf-8")
def write_custom_skill(self, name: str, relative_path: str, content: str) -> None:
target = self.validate_relative_path(relative_path, self.get_custom_skill_dir(name))
@@ -251,7 +289,17 @@ class LocalSkillStorage(SkillStorage):
def delete_custom_skill(self, name: str, *, history_meta: dict | None = None) -> None:
self.validate_skill_name(name)
self.ensure_custom_skill_is_editable(name)
target = self.get_custom_skill_dir(name)
# Delete the skill's *discovered* directory, not the assumed
# ``custom/<name>`` path — a skill placed at a non-canonical or nested
# location must still be removable now that discovery surfaced it.
target = self._resolve_skill_dir(name, SkillCategory.CUSTOM)
if target is None:
raise FileNotFoundError(f"Custom skill '{name}' not found.")
# Safety: never escape the custom root, and never remove the root itself.
custom_root = (self._host_root / SkillCategory.CUSTOM.value).resolve()
resolved_target = target.resolve()
if resolved_target == custom_root or custom_root not in resolved_target.parents:
raise ValueError(f"Refusing to delete skill '{name}': resolved path '{resolved_target}' is outside the custom skills directory.")
if history_meta is not None:
prev_content = self.read_custom_skill(name)
try:
@@ -247,6 +247,7 @@ class SkillStorage(ABC):
extensions_config = ExtensionsConfig.from_file()
for skill in skills:
skill.enabled = extensions_config.is_skill_enabled(skill.name, skill.category)
skill.always_on = extensions_config.is_skill_always_on(skill.name)
except Exception as e:
logger.warning("Failed to load extensions config: %s", e)
@@ -28,6 +28,7 @@ class Skill:
relative_path: Path # Relative path from category root to skill directory
category: SkillCategory # 'public' or 'custom'
enabled: bool = False # Whether this skill is enabled
always_on: bool = False # 常驻(免压缩):开启压缩后仍把本技能全文注入系统提示词
@property
def skill_path(self) -> str:
@@ -176,6 +176,113 @@ def skill_view_tool(name: str, runtime: ToolRuntime) -> str:
return skill_view_impl(name, _resolve_active_agent_id(runtime))
def _resolve_skills_container_base_path() -> str:
"""容器内技能挂载根路径(用于回传 location,供 agent read_file)。"""
try:
from deerflow.config import get_app_config
return get_app_config().skills.container_path
except Exception:
return "/mnt/skills"
def _score_skill(query_terms: list[str], haystack: str) -> int:
"""对单个技能按关键词命中计分(纯关键词、离线、确定性)。"""
score = 0
for term in query_terms:
if term and term in haystack:
score += 1
return score
def search_skills_impl(query: str, agent_id: str | None, top_n: int = 5) -> str:
"""按关键词检索可用技能,返回 top-N 的 name/description/category/location。
与 ``skill_list`` 共享可见性过滤(自定义 agent 走白名单;默认 lead agent 只
暴露未归档且内置或属于当前用户的技能),并只检索**已启用**的技能 —— 它是
"技能压缩"开启后,agent 调出被压缩(非常驻)技能的主要入口。匹配字段为技能
名与描述。
"""
import re
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.types import SkillCategory
from deerflow.skills.usage import STATE_ARCHIVED, all_entries
query = (query or "").strip()
if not query:
return "search_skills: empty query. Provide a short keyword query describing the task."
terms = [t for t in re.findall(r"\w+", query.lower()) if t]
# 整串也作为一个匹配项,兼顾中文等不被 \w+ 切分出有意义 token 的查询。
whole = query.lower()
if whole and whole not in terms:
terms.append(whole)
allowed = _load_agent_skill_whitelist(agent_id)
owned: set[str] | None = None
if allowed is None:
owned = _load_owned_skill_names(_resolve_effective_user_id())
try:
top_n = max(1, min(int(top_n or 5), 50))
except (TypeError, ValueError):
top_n = 5
try:
storage = get_or_new_skill_storage()
skills = storage.load_skills(enabled_only=True)
usage = all_entries()
base_path = _resolve_skills_container_base_path()
scored: list[tuple[int, dict]] = []
for skill in skills:
entry = usage.get(skill.name, {})
if allowed is not None:
if skill.name not in allowed:
continue
else:
if entry.get("state") == STATE_ARCHIVED:
continue
if skill.category == SkillCategory.CUSTOM and owned is not None and skill.name not in owned:
continue
haystack = f"{skill.name} {skill.description}".lower()
score = _score_skill(terms, haystack)
if score <= 0:
continue
scored.append((
score,
{
"name": skill.name,
"description": skill.description,
"category": skill.category.value if hasattr(skill.category, "value") else str(skill.category),
"location": skill.get_container_file_path(base_path),
},
))
scored.sort(key=lambda x: x[0], reverse=True)
results = [item for _, item in scored[:top_n]]
if not results:
return f"No skills matched '{query}'. Try a broader keyword, or call skill_list to browse everything available."
return json.dumps(results, ensure_ascii=False, indent=2)
except Exception as e:
logger.debug("search_skills failed: %s", e)
return f"Error searching skills: {e}"
@tool("search_skills")
def search_skills_tool(query: str, runtime: ToolRuntime) -> str:
"""Search available skills by keyword and return the best matches.
Use this to discover skills that are not already described in full in your
system prompt. Returns a JSON array of matches, each with `name`,
`description`, `category`, and `location` — then `read_file` the `location`
of the most relevant match and follow its SKILL.md instructions.
Args:
query: A short keyword query describing the task (e.g. 'pdf table extract').
"""
return search_skills_impl(query, _resolve_active_agent_id(runtime))
@tool("skill_list")
def skill_list_tool(runtime: ToolRuntime) -> str:
"""List skills available to this agent, with metadata and usage stats.
@@ -139,6 +139,18 @@ def get_available_tools(
builtin_tools.extend([skill_list_tool, skill_view_tool])
# search_skills —— 仅当管理员开启"技能压缩"时提供。压缩开启后,非常驻技能不再
# 全文注入系统提示词,agent 通过该关键词检索工具按需调出(纯关键词、离线、确定性)。
try:
from deerflow.config.system_settings import get_skill_compression_settings
if get_skill_compression_settings().enabled:
from deerflow.tools.builtins.skill_tools import search_skills_tool
builtin_tools.append(search_skills_tool)
except Exception: # noqa: BLE001 - 读取设置失败不能阻断 agent 构建
logger.debug("Failed to evaluate skill compression setting for search_skills tool", exc_info=True)
# read_peer_delivery —— 圆桌席位「按需读取前序席位完整交付」的只读工具(从 run context
# 取网关注入的全文,非圆桌场景为空 → 工具自述「无」)。始终提供(只读、无副作用);圆桌
# 总控/报告角色由 run policy 显式排除,避免干扰它们。
@@ -0,0 +1,182 @@
"""一次性清洗:修正历史知识笔记里遗留的 `## Related` 结构链接与 `### Sources`。
背景
====
早期沉淀的每条笔记都会在 `## Related` 段自动塞入“结构性” wikilink
(`[[index]]` / `[[_meta/taxonomy]]` / `[[projects/<project>/…]]` / 项目 hub),
并在 `## Related` 下嵌套一个 `### Sources` 子段。现在:
- 真·相关链接只用裸标题 `[[Title]]`(由 service 层按共享标签注入);
- 来源统一由结构化的「来源」区块(`note.sources`)展示,不再写进正文。
但历史笔记的 `content_md`(数据库 + vault 文件)里仍焊死着这些遗留内容,
导致详情页出现噪声链接、以及 sources/来源 重复。本脚本把它们清理掉。
清洗规则(保守,只删明确匹配项)
================================
对每条笔记的正文:
1. 删除 `## Related` 段内的 `### Sources` 子段(标题 + 其下条目);
2. 删除 `## Related` 段内的“结构性”链接条目 —— 即 wikilink 目标包含 `/`、
或等于 `index` / `taxonomy` 之类的脚手架页(真·相关链接是裸标题,予以保留);
3. 若清理后 `## Related` 段已无任何内容条目,则连同 `## Related` 标题一并删除;
4. 折叠多余空行。
裸标题的 `[[Title]]` 相关链接、其它正文内容一律保留(无数据丢失)。
用法
====
::
# 预览(不写盘),打印每条将要发生的改动
PYTHONPATH=. python scripts/cleanup_knowledge_related.py --dry-run
# 执行(写回数据库 + 重写 vault 文件)
PYTHONPATH=. python scripts/cleanup_knowledge_related.py
# 只处理单条
PYTHONPATH=. python scripts/cleanup_knowledge_related.py --note-id kb_xxx
脚本幂等:已经干净的笔记会被跳过。写回走 `KnowledgeService.update_note`,
所以每次改动都会落一份版本快照(可在「编辑历史」回滚)。
"""
from __future__ import annotations
import argparse
import asyncio
import logging
import re
import sys
logger = logging.getLogger("cleanup_knowledge_related")
_RELATED_HEADER_RE = re.compile(r"^##\s+(Related|相关知识|相关文档|相关)\s*$", re.IGNORECASE)
_SOURCES_HEADER_RE = re.compile(r"^###\s+(Sources|来源)\s*$", re.IGNORECASE)
_H12_RE = re.compile(r"^#{1,2}\s+")
_H123_RE = re.compile(r"^#{1,3}\s+")
_BULLET_LINK_RE = re.compile(r"^[-*]\s*\[\[([^\]\n|]+)(?:\|[^\]\n]+)?\]\]\s*$")
def _is_structural_target(target: str) -> bool:
"""A `## Related` wikilink that points at vault scaffolding (not a real note).
Real related cross-links are bare note titles (`[[标题]]`) with no slashes.
The legacy structural links were full vault paths / meta pages.
"""
t = (target or "").strip().strip("/").lower()
if not t:
return True
if "/" in t: # projects/<p>/…, _meta/taxonomy, … — all structural
return True
return t in {"index", "taxonomy", "hot", "log", "_meta"}
def clean_note_body(md: str) -> str:
"""Strip the legacy `### Sources` subsection + structural `## Related` links.
Pure function (no I/O) so it can be unit-tested. Returns the cleaned body;
identical to the input when there is nothing to clean.
"""
lines = md.split("\n")
n = len(lines)
out: list[str] = []
i = 0
in_fence = False
while i < n:
line = lines[i]
if re.match(r"^\s*(```|~~~)", line):
in_fence = not in_fence
out.append(line)
i += 1
continue
if in_fence or not _RELATED_HEADER_RE.match(line.strip()):
out.append(line)
i += 1
continue
# --- inside a `## Related` section -------------------------------
header = line
j = i + 1
kept: list[str] = [] # surviving content lines of this section
while j < n:
cur = lines[j]
stripped = cur.strip()
if _H12_RE.match(cur): # next h1/h2 ends the Related section
break
if _SOURCES_HEADER_RE.match(stripped):
# Drop the `### Sources` header and everything until the next header.
j += 1
while j < n and not _H123_RE.match(lines[j]):
j += 1
continue
m = _BULLET_LINK_RE.match(stripped)
if m and _is_structural_target(m.group(1)):
j += 1 # drop the structural link bullet
continue
kept.append(cur)
j += 1
meaningful = [ln for ln in kept if ln.strip()]
if meaningful:
out.append(header)
out.extend(kept)
# else: drop the now-empty `## Related` section entirely.
i = j
cleaned = "\n".join(out)
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).rstrip() + "\n"
return cleaned
async def _run(dry_run: bool, only_note_id: str | None) -> int:
from deerflow.config import get_app_config
from deerflow.knowledge import make_knowledge_service
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
config = get_app_config()
await init_engine_from_config(config.database)
sf = get_session_factory()
if sf is None:
logger.error("No database session factory — is the database backend configured?")
return 1
service = make_knowledge_service(sf, config.knowledge)
if service is None:
logger.error("Knowledge base disabled (knowledge.enabled=false) or memory backend; nothing to do.")
return 1
notes = await service.repo.list_all(include_archived=True)
if only_note_id:
notes = [n for n in notes if n.get("id") == only_note_id]
scanned = changed = 0
for note in notes:
scanned += 1
body = note.get("content_md") or ""
cleaned = clean_note_body(body)
if cleaned == body:
continue
changed += 1
nid = note.get("id")
title = note.get("title") or nid
if dry_run:
logger.info("[dry-run] would clean %s — %s (%d → %d chars)", nid, title, len(body), len(cleaned))
else:
await service.update_note(nid, content_md=cleaned, updated_by="cleanup_script")
logger.info("cleaned %s — %s", nid, title)
await close_engine()
logger.info("Done. scanned=%d, %s=%d", scanned, "would-change" if dry_run else "changed", changed)
return 0
def main() -> int:
logging.basicConfig(level=logging.INFO, format="%(message)s")
parser = argparse.ArgumentParser(description="Clean legacy `## Related` structural links + `### Sources` from knowledge notes.")
parser.add_argument("--dry-run", action="store_true", help="Preview changes without writing.")
parser.add_argument("--note-id", default=None, help="Only process this note id.")
args = parser.parse_args()
return asyncio.run(_run(args.dry_run, args.note_id))
if __name__ == "__main__":
sys.exit(main())
@@ -94,6 +94,10 @@ CALL _add_column_if_missing('agent_favorites', 'position_id', 'VARCHAR(64) NULL
CALL _add_column_if_missing('scheduled_task_subscriptions', 'origin', "VARCHAR(16) NOT NULL DEFAULT 'user'");
CALL _add_column_if_missing('scheduled_task_subscriptions', 'position_id', 'VARCHAR(64) NULL DEFAULT NULL');
-- knowledge_notes.folder: 用户自建目录(斜杠分隔路径),驱动产品目录树;NULL=未归类。
-- (knowledge_folders / knowledge_extract_templates 为【新表】,由 create_all 自动建,无需在此补列。)
CALL _add_column_if_missing('knowledge_notes', 'folder', 'VARCHAR(512) NULL DEFAULT NULL');
-- ---- 清理临时存储过程 --------------------------------------------------------
DROP PROCEDURE IF EXISTS _add_column_if_missing;
@@ -111,6 +115,20 @@ PREPARE stmt FROM @idx_ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
-- ---- knowledge_notes.folder 索引(幂等) --------------------------------------
SET @idx_exists = (
SELECT COUNT(*) FROM information_schema.statistics
WHERE table_schema = DATABASE()
AND table_name = 'knowledge_notes'
AND index_name = 'ix_knowledge_notes_folder'
);
SET @idx_ddl = IF(@idx_exists = 0,
'CREATE INDEX `ix_knowledge_notes_folder` ON `knowledge_notes` (`folder`(191))',
'SELECT ''SKIPPED ix_knowledge_notes_folder (already exists)'' AS result');
PREPARE stmt FROM @idx_ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
-- ---- 执行后自检:确认本批次涉及的列都已存在 ---------------------------------
SELECT table_name, column_name
FROM information_schema.columns
+9 -2
View File
@@ -310,7 +310,14 @@ summarization:
value: 60000
- type: messages
value: 80
# Keep a small *token* budget of recent context (same unit as the token
# trigger, well below it). This is what the model receives after compaction:
# [summary, ~16000 tokens of recent turns]. Small keep = small model context
# (fast, no overflow) AND infrequent re-compaction — the preserved window
# grows from ~16000 up to the 60000 trigger before folding into a new summary,
# which spans many turns. The full conversation still stays in state for the
# page; only what is sent to the model is compressed.
keep:
type: messages
value: 20
type: tokens
value: 16000
trim_tokens_to_summarize: 4000