Merge remote-tracking branch 'origin/master' into ji

This commit is contained in:
吉鑫升 committed 2026-06-11 16:39:30 +08:00
commit 5daa4b3e20
62 files changed
+4230 -1025

No files matched your search

@@ -19,24 +19,30 @@
并配合 ``SkillStopMiddleware`` 在 ``agent_orchestration`` 调用后停图,便于网关从
AIMessage.tool_calls 解析派活列表。
- ``seat`` —— 子智能体交付:开 thinking + medium reasoning;**可写文件**
(``write_file`` / ``str_replace`` 产出 md 等成品文档),但禁派活 / 澄清 /
展示文件 / 看图 / 联网搜索 / **``bash``**。禁 ``bash`` 的原因:圆桌席位是
「文档产出者」,弱模型爱写 ``.py`` 脚本再跑出产物(乱且抓不到预览),与
``report`` 角色禁 ``bash`` 同理;让它**直接 write_file 写成品文档**。
(``write_file`` / ``str_replace`` 产出 md 等成品文档)+ **可跑 ``bash``**
(2026-06 放开:业务需要跑知识库检索 / Python 辅助分析)。禁派活 / 澄清 /
展示文件 / 看图 / 联网搜索。``report`` 通过 ``[*_SEAT_EXCLUDED, ...]`` 继承本角色,
所以 Step3 报告也跟着拿到 bash;只有 ``leader`` 和 ``seat_parallel`` 仍各自显式禁。
历史警示(保留):弱模型用 ``bash`` 跑 ``python xxx.py > out.md`` 生成产物会
**绕过前端虚拟预览**(预览只拦截 ``write_file.args.content``)→ 沙箱里有文件但
Step2 右侧预览空白。席位 SOUL 必须明确「bash 只用于辅助查询拿数据,**最终交付
必须用 write_file 写 .md 落盘**,不要让脚本生成最终产物」。
- ``seat_parallel`` —— **DAG 快速(fast)模式的并行席位(方案 A,2026-06)**:**禁写文件**
(``write_file`` / ``str_replace``,产物走正文)+ 禁 ``bash`` + 禁「打断 / 越权」类,
(``write_file`` / ``str_replace``,产物走正文)+ **禁 ``bash``** + 禁「打断 / 越权」类,
**放开 ``web_search`` 与技能**(搜索不碰 sandbox;技能靠 ``read_file`` SKILL.md)——
否则并行席位几乎调不动工具、产生不了步骤(步骤条不显示)。
并行席位同时跑 bash 会在共享 sandbox 里抢路径 / 占进程数;且 fast 模式整体不
写文件,bash 跑出来的产物既不能预览也没意义。
「写文件产出 md」归 **沙箱(sandbox)模式**(走 ``seat`` policy,单列 + 沙箱跟随预览);
fast 模式快、双列、产物走正文 —— 用户明确要求 fast 不生成文件。详见
``docs/roundtable-dag-orchestration.md`` §4 / §12.1 / §12.8 / §12.10。
与 ``seat`` 的差异:``seat`` 禁 ``web_search``(旧「圆桌一律禁搜索」),
``seat_parallel`` **放开搜索**(本期诉求)。
- ``report`` —— Step3 报告绘制:在 ``seat``(已禁 ``bash``)基础上**额外禁
``str_replace``**:
* ``bash`` 已在 ``seat`` 禁 → 杜绝写 Python 脚本再跑出 HTML(脚本产物落在
outputs,但前端虚拟预览只抓 ``write_file`` 的 ``args.content``,抓不到脚本
产物 → 预览停在半截 ``write_file``,正是「没写完就去生成 py 文件」的现象);
``seat_parallel`` **放开搜索**(本期诉求);``seat`` 已放开 ``bash``,本角色仍禁。
- ``report`` —— Step3 报告绘制:在 ``seat`` 基础上**额外禁 ``str_replace``**:
* 继承 ``seat`` 的 bash 放开(2026-06)—— Step3 也可跑 bash 做辅助查询,同
``seat`` 警示:**结果必须 write_file 落盘**,否则前端虚拟预览会空白
(预览只抓 ``write_file.args.content``,抓不到脚本产物 → 预览停在半截
``write_file``,正是「没写完就去生成 py 文件」的现象)。
* 禁 ``str_replace`` → 强制返工用整文件 ``write_file`` 重写
(``str_replace`` 没有 ``content`` 字段,会让前端虚拟预览读到 undefined
而清空沙箱)。
@@ -72,16 +78,21 @@ _LEADER_EXCLUDED: list[str] = [
"hindsight_retain",
"read_peer_delivery", # 总控靠 _collect_delivery_status 直读各席位,不需要这个席位侧工具
]
# seat:席位**可以写文件**(write_file / str_replace)产出 md 等成品文档,但**禁 bash** ——
# 圆桌席位是「文档产出者」,不该跑 shell/python 脚本(弱模型爱写 .py 脚本再跑生成产物,既乱又
# 抓不到产物预览,与 report 角色禁 bash 同理)。禁 web_search(圆桌串行一律禁搜索,旧决策)。
# seat:席位**可以写文件**(write_file / str_replace)产出 md 等成品文档,**bash 放开**
# (2026-06 决策,业务需要 bash 跑知识库检索 / Python 辅助分析)。禁派活 / 澄清 /
# 展示文件 / 看图 / 联网搜索。``report`` 通过 ``[*_SEAT_EXCLUDED, ...]`` 继承本角色,
# 所以 Step3 报告跟着拿到 bash;只有 leader / seat_parallel 仍各自显式禁。
#
# **历史警示(保留)**:弱模型用 bash 跑 ``python xxx.py > out.md`` 生成最终产物会
# 绕过前端虚拟预览(只拦 write_file.args.content) → 沙箱里有文件但 Step2 右侧预览
# 空白。必须靠席位 SOUL 强约束「bash 只用于辅助查询拿数据,最终交付必须用 write_file
# 写 .md 落盘」。
_SEAT_EXCLUDED: list[str] = [
"web_search",
"ask_clarification",
"present_files",
"view_image",
"agent_orchestration",
"bash",
]
# 并行席位 = **快速(fast)模式**(双列):**禁写文件**(产物走正文)+ 禁跑命令 +「打断/越权」类。
# **禁 web_search** —— 离线内网部署无外网,席位检索一律走「技能(skill)」(靠 read_file 读
@@ -328,9 +328,15 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
# config.yaml 的 models[0],与 /api/ai-writing 等其它路由一致,内网部署只配
# 自己的本地模型即可工作。
model_name = str(payload.get("model") or "")
# 用户在 Step 1 上传的参考文件(已经过 /api/threads/{thread_id}/uploads)。
# 前端传 [{filename, size, path, status}, ...],这里原样透传给 _run_payload →
# additional_kwargs.files → UploadsMiddleware 自动注入清单 + 解锁 read_file 等工具。
raw_files = payload.get("files")
files = raw_files if isinstance(raw_files, list) else []
if not thread_id:
raise HTTPException(status_code=400, detail="thread_id is required")
if not message:
# 允许「仅文件、无文本」的回合:带了文件就不强制要求 message。
if not message and not files:
raise HTTPException(status_code=400, detail="message is required")
# 调用前自检:防御性二次检查——若管理员在 init 之后手动删了目录,这里也能自愈。
@@ -381,8 +387,17 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
_MAX_CLARIFICATION_ROUNDS,
)
# 带了上传文件时,解锁「读」类工具(read_file / ls / grep / view_image),
# 否则 UploadsMiddleware 注入了文件清单、模型却没工具读取。仍保留对
# write_file / str_replace / present_files / bash / agent_orchestration 的封锁,
# 防止意图阶段越界做方案/派活。
if files:
read_tools = {"read_file", "ls", "grep", "view_image"}
base_excluded_tools = [t for t in base_excluded_tools if t not in read_tools]
# 2. message 前缀——给模型一个明确的"本轮必须收敛"信号。
effective_message = message
# 仅文件、无文本时给一句兜底引导,让模型先读文件再澄清意图。
effective_message = message or "我上传了文件,请先阅读文件内容,再据此理解并澄清我的任务意图。"
if past_clarifications >= _MAX_CLARIFICATION_ROUNDS:
effective_message = (
f"[系统强制] 已澄清 {past_clarifications} 轮,触发硬上限。"
@@ -409,6 +424,7 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
# not reliable on its own, hard exclusion is.
excluded_tools=base_excluded_tools,
skill_stop_names=[],
files=files,
)
# captured:上游 SSE 全部 data 行,用于流结束后从 values 快照取 messages。
@@ -638,6 +638,7 @@ def _run_payload(
reasoning_effort: str = "medium",
multitask_strategy: str = "reject",
extra_context: dict[str, Any] | None = None,
files: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。
@@ -650,13 +651,19 @@ def _run_payload(
应传 ``thinking_enabled=False`` + ``reasoning_effort="low"``,省掉模型先吐
reasoning tokens 才开口的时间,显著降低前端"调度中"首字节延迟。
"""
# 用户上传的文件元数据透传给 UploadsMiddleware(与普通聊天同链路):
# before_agent 钩子读 additional_kwargs.files,生成 <uploaded_files> 清单注入消息,
# 之后席位/总控用内置 read_file / grep / view_image 工具按需读取。空时落空 dict,行为不变。
additional_kwargs: dict[str, Any] = {}
if files:
additional_kwargs["files"] = files
payload: dict[str, Any] = {
"input": {
"messages": [
{
"type": "human",
"content": [{"type": "text", "text": new_message}],
"additional_kwargs": {},
"additional_kwargs": additional_kwargs,
}
]
},
@@ -1258,6 +1265,7 @@ async def _leader_run(
seat_meta: dict[str, dict[str, Any]] | None = None,
synthesis_mode: bool = False,
suppress_pre_broadcast: bool = False,
files: list[dict[str, Any]] | None = None,
) -> AsyncIterator[str | dict[str, Any]]:
"""总控(leader)单轮:透传上游 SSE + 末尾 yield 一个 dict 状态帧(由路由层 ``json.dumps``)。
@@ -1346,18 +1354,26 @@ async def _leader_run(
# 调用方可再叠加自己的 skill_stop。后台进程内网关用同一份 policy,保证传参一致。
policy = roundtable_run_policy("leader")
effective_stop_names = merge_skill_stop_names(policy.skill_stop_names, skill_stop_names)
# 带了上传文件时,给总控解锁「读」类工具(leader policy 默认禁 read_file/ls/view_image),
# 否则文件清单注入了却无工具读取。写/派活类仍按 policy 封锁,不放开。
leader_excluded = policy.excluded_tools
if files:
_read_tools = {"read_file", "ls", "grep", "view_image"}
leader_excluded = [t for t in leader_excluded if t not in _read_tools]
body = _run_payload(
agent_name=agent_name,
model_name=model_name,
thread_id=thread_id,
# 注入了席位清单的输入(leader_input);seat_roster 为空时等同 new_message。
new_message=leader_input,
excluded_tools=policy.excluded_tools,
excluded_tools=leader_excluded,
skill_stop_names=effective_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
# 人工干预/恢复会重发 leader run 到同一总控 thread;抢占掉没结束的旧 run,避免 409。
multitask_strategy="rollback",
# 用户随干预上传的参考文件 → 总控可用 read_file 等工具读取后再派活。
files=files,
)
t_payload_built = time.perf_counter()
logger.info(
@@ -1532,6 +1548,7 @@ async def _special_run(
display_name: str = "",
no_file: bool = False,
peer_deliveries: list[dict[str, Any]] | None = None,
files: list[dict[str, Any]] | None = None,
) -> AsyncIterator[str | dict[str, Any]]:
"""子智能体(special)单轮:执行总控下发的 task 文本,交付 content。
@@ -1560,12 +1577,18 @@ async def _special_run(
else:
role = "seat"
policy = roundtable_run_policy(role)
# 带了上传文件时,给席位解锁「读」类工具(view_image 默认被禁;read_file 本就可用),
# 让席位能读取/识别用户上传的文件。写/派活类仍按 policy 封锁。
seat_excluded = policy.excluded_tools
if files:
_read_tools = {"read_file", "ls", "grep", "view_image"}
seat_excluded = [t for t in seat_excluded if t not in _read_tools]
body = _run_payload(
agent_name=agent_name,
model_name=model_name,
thread_id=thread_id,
new_message=new_message,
excluded_tools=policy.excluded_tools,
excluded_tools=seat_excluded,
skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names),
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
@@ -1575,6 +1598,8 @@ async def _special_run(
extra_context=(
{"roundtable_peer_deliveries": peer_deliveries} if peer_deliveries else None
),
# 用户随干预上传的参考文件 → 席位可用 read_file 等工具读取。
files=files,
)
t_payload_built = time.perf_counter()
logger.info(
@@ -1677,6 +1702,11 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
display_name: str = str(payload.get("display_name") or "").strip()
new_message: str = str(payload.get("new_message") or "")
skill_stop_names: list[str] = payload.get("skill_stop_names") or []
# 用户随干预消息上传的参考文件(已经过 /api/threads/{thread_id}/uploads)。
# 前端传 [{filename, size, path, status}, ...],透传给 _run_payload →
# additional_kwargs.files → UploadsMiddleware 注入清单 + 解锁 read_file 等工具。
_files_raw = payload.get("files")
files: list[dict[str, Any]] = _files_raw if isinstance(_files_raw, list) else []
# 综合轮标记(仅 leader 用):前端在「派活后让总控判断继续/综合」的轮次置真,网关据此
# 额外把各已交付席位的完整交付正文注入总控输入,供其写最终方案。缺省 False(派活决策轮
# 只给摘要,保持上下文精简)。旧前端不传 → False,行为安全退化。
@@ -1741,6 +1771,7 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
seat_meta=seat_meta,
synthesis_mode=synthesis_mode,
suppress_pre_broadcast=suppress_pre_broadcast,
files=files,
)
else:
gen = _special_run(
@@ -1748,6 +1779,7 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
display_name=display_name,
no_file=no_file,
peer_deliveries=peer_deliveries,
files=files,
)
async for chunk in gen:
if isinstance(chunk, dict):
@@ -2788,14 +2788,15 @@ class CuratorConfigResponse(BaseModel):
class CuratorConfigRequest(BaseModel):
"""Payload for updating skill_evolution overrides.
"""Payload for updating user-facing curator settings in skill_evolution.json.
All fields are required so the frontend always sends a complete config;
backend writes the entire payload to skill_evolution.json atomically.
All other settings (enabled, timeouts) are global-only and controlled via config.yaml.
"""
enabled: bool = Field(..., description="Whether the agent can create and modify skills under skills/custom.")
curator: CuratorSubConfig
interval_hours: int = Field(..., ge=1, description="Hours between curator runs.")
stale_after_days: int = Field(..., ge=1, description="Days of inactivity before a skill is marked stale.")
archive_after_days: int = Field(..., ge=1, description="Days of inactivity before a stale skill is archived.")
model_name: str | None = Field(default=None, description="Model for the curator LLM review; null = primary model.")
class CuratorLogRecord(BaseModel):
@@ -2811,11 +2812,7 @@ class CuratorLogsResponse(BaseModel):
def _build_curator_config_response(evo) -> CuratorConfigResponse:
"""Pure helper so both the GET route and the post-PUT refresh share one shape.
``evo`` is a ``SkillEvolutionConfig`` instance (already merged with per-user
overrides if any).
"""
"""Pure helper so both the GET route and the post-PUT refresh share one shape."""
return CuratorConfigResponse(
enabled=evo.enabled,
curator=CuratorSubConfig(
@@ -2831,8 +2828,7 @@ def _build_curator_config_response(evo) -> CuratorConfigResponse:
@router.get("/curator/config", response_model=CuratorConfigResponse, summary="Get Skill Evolution Config")
async def get_curator_config(request: Request) -> CuratorConfigResponse:
"""Return the current curator config for the authenticated user (yaml
defaults + that user's overrides)."""
"""Return the current curator config for the authenticated user (yaml defaults + per-user overrides)."""
try:
from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user
@@ -2845,21 +2841,22 @@ async def get_curator_config(request: Request) -> CuratorConfigResponse:
@router.put("/curator/config", response_model=CuratorConfigResponse, summary="Update Skill Evolution Config")
async def update_curator_config(body: CuratorConfigRequest, request: Request) -> CuratorConfigResponse:
"""Persist per-user overrides to ``users/{user_id}/skill_evolution.json``
and return the effective merged config.
"""Persist the three user-facing curator thresholds to the per-user ``skill_evolution.json``.
The payload is validated against the ``SkillEvolutionConfig`` shape; unknown
fields are rejected.
Other settings (enabled, model, timeouts) are global and controlled via config.yaml.
"""
try:
from deerflow.config.skill_evolution_config import SkillEvolutionConfig
from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user, save_user_overrides
user_id = _current_user_id(request)
payload = body.model_dump()
# Round-trip through SkillEvolutionConfig to ensure persisted shape stays valid.
SkillEvolutionConfig.model_validate(payload)
save_user_overrides(user_id, payload)
save_user_overrides(user_id, {
"curator": {
"interval_hours": body.interval_hours,
"stale_after_days": body.stale_after_days,
"archive_after_days": body.archive_after_days,
"model_name": body.model_name,
},
})
return _build_curator_config_response(load_skill_evolution_config_for_user(user_id))
except HTTPException:
raise
+111 -150
View File
@@ -10,90 +10,104 @@ gateway:
port: 8001
models:
# - name: deepseek-chat
# display_name: DeepSeek Chat
# use: langchain_openai:ChatOpenAI
# model: deepseek-chat
# api_key: sk-1520028668cb4cc388ed8315ff1e856e
# base_url: https://api.deepseek.com
# request_timeout: 600.0
# max_retries: 2
# max_tokens: 4096
# temperature: 1.0
# supports_vision: false
# supports_thinking: true
# - name: deepseek-v4-flash
# display_name: DeepSeek V4 Flash
# use: langchain_openai:ChatOpenAI
# model: deepseek-v4-flash
# api_key: sk-e1c22b7931dc446f8b67e57878edd105
# base_url: https://api.deepseek.com
# request_timeout: 600.0
# max_retries: 2
# max_tokens: 4096
# temperature: 1.0
# supports_vision: false
# supports_thinking: true
# - name: deepseek-v4-pro
# display_name: DeepSeek V4 Pro
# use: langchain_openai:ChatOpenAI
# model: deepseek-v4-pro
# api_key: sk-e1c22b7931dc446f8b67e57878edd105
# base_url: https://api.deepseek.com
# request_timeout: 600.0
# max_retries: 2
# max_tokens: 4096
# temperature: 1.0
# supports_vision: false
# supports_thinking: true
# - name: MiniMax-M2.7
# display_name: MiniMax-M2.7
# use: langchain_openai:ChatOpenAI
# model: MiniMax-M2.7
# api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
# base_url: https://api.minimaxi.com/v1
# max_tokens: 4096
# temperature: 1.0
# supports_vision: true
# supports_thinking: true
# - name: glm-5.1
# display_name: glm-5.1
# use: langchain_openai:ChatOpenAI
# model: glm-5.1
# api_key: 75121e4bed7640d5b71ac95cf8b4310a.RA3GyRBAR3kv7VrR
# base_url: https://open.bigmodel.cn/api/paas/v4/
# max_tokens: 4096
# temperature: 1.0
# supports_vision: true
# supports_thinking: true
# - name: deepseek-ai/DeepSeek-OCR
# display_name: deepseek-ai/DeepSeek-OCR
# use: langchain_openai:ChatOpenAI
# model: deepseek-ai/DeepSeek-OCR
# api_key: sk-beshfxfxfrllsczpqfrgjbugfbpzxgqgedbehceytpcvvtyb
# base_url: https://api.siliconflow.cn/v1/
# max_tokens: 4096
# temperature: 1.0
# supports_vision: true
# supports_thinking: true
# - name: zai-org/GLM-5-FP8
# display_name: zai-org/GLM-5-FP8
# use: langchain_openai:ChatOpenAI
# model: zai-org/GLM-5-FP8
# api_key: modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
# base_url: https://api.us-west-2.modal.direct/v1
# request_timeout: 600.0
# max_retries: 2
# max_tokens: 4096
# temperature: 1.0
# supports_vision: true
# supports_thinking: true
- name: deepseek-ai/DeepSeek-V4-Flash
display_name: deepseek-ai/DeepSeek-V4-Flash
- name: deepseek-chat
display_name: DeepSeek Chat
use: langchain_openai:ChatOpenAI
model: deepseek-ai/DeepSeek-V4-Flash
model: deepseek-chat
api_key: sk-3fee562f834a4c08bb0d471ef44cb445
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: false
supports_thinking: true
- name: zai-org/GLM-5-FP8
display_name: zai-org/GLM-5-FP8
use: langchain_openai:ChatOpenAI
model: zai-org/GLM-5-FP8
api_key: modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
base_url: https://api.us-west-2.modal.direct/v1
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: deepseek-v4-flash
display_name: DeepSeek V4 Flash
use: langchain_openai:ChatOpenAI
model: deepseek-v4-flash
api_key: sk-3fee562f834a4c08bb0d471ef44cb445
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: false
supports_thinking: true
- name: deepseek-v4-pro
display_name: DeepSeek V4 Pro
use: langchain_openai:ChatOpenAI
model: deepseek-v4-pro
api_key: sk-e1c22b7931dc446f8b67e57878edd105
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: false
supports_thinking: true
- name: MiniMax-M2.7
display_name: MiniMax-M2.7
use: langchain_openai:ChatOpenAI
model: MiniMax-M2.7
api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
base_url: https://api.minimaxi.com/v1
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: glm-5.1
display_name: glm-5.1
use: langchain_openai:ChatOpenAI
model: glm-5.1
api_key: 75121e4bed7640d5b71ac95cf8b4310a.RA3GyRBAR3kv7VrR
base_url: https://open.bigmodel.cn/api/paas/v4/
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: deepseek-ai/DeepSeek-OCR
display_name: deepseek-ai/DeepSeek-OCR
use: langchain_openai:ChatOpenAI
model: deepseek-ai/DeepSeek-OCR
api_key: sk-beshfxfxfrllsczpqfrgjbugfbpzxgqgedbehceytpcvvtyb
base_url: https://api.siliconflow.cn/v1/
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: zai-org/GLM-5-FP8
display_name: zai-org/GLM-5-FP8
use: langchain_openai:ChatOpenAI
model: zai-org/GLM-5-FP8
api_key: modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
base_url: https://api.us-west-2.modal.direct/v1
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: Qwen/Qwen3.6-27B
display_name: Qwen/Qwen3.6-27B
use: langchain_openai:ChatOpenAI
model: Qwen/Qwen3.6-27B
api_key: sk-ighylafootabtuthwueirmoycccyqtdgbifyhozduvolkdyt
base_url: https://api.siliconflow.cn/v1/
max_tokens: 4096
@@ -152,26 +166,6 @@ uploads:
auto_convert_documents: true
pdf_converter: auto
# Chrome 插件「当前网页发送到 Deer」入口。默认关闭,打开后才启用
# /api/browser-context/capture。它只保存当前页 Markdown 到线程 uploads,
# 不会改变普通聊天、上传或工具调用流程。
browser_context:
enabled: true
max_content_chars: 200000
max_title_chars: 300
max_url_chars: 2048
filename_prefix: webpage
route_base: /page/workspace
task_result_timeout_seconds: 60
task_poll_interval_seconds: 1
task_lease_seconds: 120
tool_result_max_chars: 60000
# 交互式页面操作(browser_act:点击/输入/滚动/等待/跳转)。关闭后只保留只读抓取。
actions_enabled: true
action_result_timeout_seconds: 45
max_observed_elements: 150
observation_text_max_chars: 12000
title:
enabled: true
max_words: 6
@@ -189,7 +183,6 @@ title:
tool_groups:
- name: web
- name: browser
- name: file:read
- name: file:write
- name: bash
@@ -205,14 +198,6 @@ tools:
use: deerflow.community.jina_ai.tools:web_fetch_tool
timeout: 15
- name: browser_fetch_page
group: browser
use: deerflow.tools.builtins.browser_context_tool:browser_fetch_page_tool
- name: browser_act
group: browser
use: deerflow.tools.builtins.browser_act_tool:browser_act_tool
- name: ls
group: file:read
use: deerflow.sandbox.tools:ls_tool
@@ -256,6 +241,18 @@ skills:
security_scan:
enabled: false
skill_evolution:
enabled: true
curator:
enabled: true # 每 24 小时自动清理 + 合并 skill
interval_hours: 24
review_worker:
enabled: true # 被动复盘开关
min_tool_calls: 5 # 触发阈值:本轮至少 5 次工具调用,与 prompt 中"≥5 步"硬条件一致
max_messages: 60 # 传给 LLM 的最大消息条数
timeout_seconds: 120 # 单次 review 超时(秒)
model_name: # 留空则使用主模型
agents_api:
enabled: true
@@ -286,7 +283,7 @@ auth_login:
database:
backend: sqlite
sqlite_dir: /app/backend/.deer-flow/data
sqlite_dir: .deer-flow/data
pool_size: 20 # 基础连接池(postgres/mysql 生效;sqlite 忽略)
max_overflow: 20 # 高峰期可临时超出 pool_size 的额外连接数,真实上限 = pool_size + max_overflow
pool_timeout: 30 # 池满时等待空闲连接的秒数,超时报错
@@ -294,7 +291,7 @@ database:
checkpointer:
type: sqlite
connection_string: /app/backend/.deer-flow/data/deerflow.db
connection_string: .deer-flow/data/deerflow.db
# 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。
# 三层覆盖优先级(高->低):
@@ -376,39 +373,3 @@ log_system:
enabled: true
url: "https://your-log-system.example.com"
label: "日志系统"
# 产品内知识库 —— obsidian-wiki 兼容 vault。知识元数据写入数据库,正文
# content_md 写入 Markdown vault(index.md/log.md/hot.md/.manifest.json/frontmatter/
# wikilinks/wiki-export 图谱产物)。知识库为全局共享,不做 user_id 隔离,
# 写入时记录 created_by / updated_by 作为审计。
knowledge:
enabled: true
provider: obsidian_wiki
vault_dir: data/knowledge/obsidian-vault
project_name: zncm
auto_ingest_enabled: false
auto_ingest_min_chars: 300
# 用大模型按 wiki-capture 规范提炼陈述式知识(而非堆砌对话原文)。
# 关闭则回退到规则抽取;extract_model_name 留空时用 models[0]。
llm_extract_enabled: true
extract_model_name: ""
extract_max_input_chars: 12000
max_thread_messages: 80
max_source_items: 20
rag_enabled: false
rag_limit: 5
# 阶段三:写入前脱敏(token/密钥/手机号/身份证/邮箱);重复检测相似度阈值。
sensitive_redaction_enabled: true
dedup_similarity_threshold: 0.82
export:
enabled: true
graph_html_enabled: true
graph_json_enabled: true
query:
use_hot_md: true
use_index_md: true
use_frontmatter_scan: true
embedding:
enabled: false
provider: local
model: ""
@@ -34,17 +34,20 @@
### 关键事件
**1987.11 开放赴中国大陆探亲**:
- 蒋经国政府宣布开放赴大陆探亲
- 表面为人道措施,实质开启人员与资金流动
- "探亲带带货"成为早期贸易雏形
**1990s 间接贸易兴起**:
- 贸易通过香港等第三地转口进行
- 台商以个人或第三地公司名义登陆
- 鞋业、成衣、玩具等劳力密集产业率先外移
- 1992 邓小平南巡后,台商投资第一波高峰
**1996.09 戒急用忍政策**:
- 李登辉提出"戒急用忍"
- 限制高科技与大型基础设施赴大陆投资
- 试图在政治风险上升时踩刹车
@@ -77,23 +80,27 @@
### 关键事件
**2001.01 金马小三通**:
- 金门、马祖与福建沿海通航通商
- 象征性大于实质经济意义
- 但打破了"完全不直接通商"的禁区
**2001.11 两岸加入 WTO**:
- 台湾以"台澎金马个别关税领域"名义加入
- 北京以"主权国家"身份加入
- 加入 WTO 后,间接贸易的合法性基础松动
- 台北面临"是否开放直航与直接投资"的压力
**2001-2004 积极开放有效管理**:
- 陈水扁政府调整"戒急用忍"为"积极开放有效管理"
- 放宽投资上限(8千万新台币限制放宽)
- 但 2006 后又收紧为"积极管理有效开放"
**2000s 中期 IT 产业大举西进**:
- 笔记本电脑、半导体封装测试等中下游产业外移
- 空间电脑、半导体封装测试等中下游产业外移
- 形成台湾接单、大陆生产、出口欧美的三角贸易结构
- 台湾对大陆出口依存度突破 20%
@@ -125,11 +132,13 @@
### 关键事件
**2008.07 周末包机与大陆观光客来台**:
- 两岸周末包机启动
- 大陆观光客来台(首波仅限特定省市)
- 大三通的前奏
**2008.12 大三通**:
- 海运直航、空运直航、直接通邮
- 两岸经贸进入"直接"时代
- 物流成本大幅下降,供应链整合加速
@@ -137,6 +146,7 @@
详见 `entities/people/ma_ying_jeou.md`。
**2010.06 ECFA 签订**:
- 两岸经济合作架构协议(ECFA)签署
- 包含"早收清单"(539 项台湾货品降税)
- 后续需谈判"货品贸易协议"与"服务贸易协议"
@@ -145,6 +155,7 @@
详见 `narratives/cross_strait/ecfa_legacy_debate.md`。
**2013.06 两岸服贸协议签署**:
- 服务贸易协议完成签署
- 开放金融、医疗、电信等服务领域
- 但需经立法院审查,尚未生效
@@ -177,6 +188,7 @@
### 关键事件
**2014.03-04 太阳花学运**:
- 反对服贸协议黑箱审查
- 占领立法院 24 天
- 服贸协议实质冻结
@@ -185,11 +197,13 @@
详见 `events/key_events/event_sunflower_2014.md`。
**2014.11 九合一选举**:
- 国民党大败
- 经济整合路线的选举惩罚
- 两岸政策从"积极整合"转向"保守防御"
**2015**:
- 两岸经贸谈判全面停滞
- 台商投资持续但增速放缓
- 青年创业成为新的交流形式(但规模有限)
@@ -224,26 +238,31 @@
### 关键事件
**2016.05 蔡英文上任**:
- 承认"九二会谈"但不承认"九二共识"
- 北京中断官方沟通机制
- 两岸经贸进入"冷对抗"状态
**2016-2020 新南向政策**:
- 鼓励台商投资东南亚与南亚
- 对大陆出口依存度短暂下降后回升
- 投资分散有限——2020 年对大陆投资仍占对外投资 40% 以上
**2018-2019 美中贸易战**:
- 台商回流与转单效应
- 台商将部分产能移回台湾或转至东南亚
- 但高阶制程仍在台湾,中低阶仍在中国大陆
**2020-2022 疫情与供应链重组**:
- 台湾对中国大陆出口依存度反而上升(因远程办公需求与大陆复工较快)
- 2021 年对大陆出口依存度达 42.3%(历史新高)
- 台积电等高阶半导体成为战略资产
**2022.08 裴洛西访台后大陆暂停部分农产品输入**:
- 柑橘类、冰鲜白带鱼、冻竹荚鱼等被暂停输入
- 目标为台湾中南部农渔业
- 经济制裁工具化的初步展现
@@ -276,30 +295,36 @@
### 关键事件
**2023.04 贸易壁垒调查**:
- 北京宣布对台湾 2455 项产品进行贸易壁垒调查
- 涵盖农产品、五矿化工、纺织品等
- 这是 ECFA 架构外的系统性施压
**2023.12 ECFA 部分中止**:
- 北京宣布中止 ECFA 早收清单中 12 项石化产品关税减让
- 丙烯、聚碳酸酯等受影响
- 明示与台湾选举结果的政治连动
**2024.01 台湾总统选举**:
- 赖清德当选
- 北京在选举后继续 ECFA 调整
**2024.05-06 ECFA 第二波中止**:
- 中止 134 项产品关税减让(机械、纺织、汽车零组件等)
- ECFA 早收清单 539 项中已有近 30% 被中止
- 同时暂停台湾芒果等农产品输入
**2024.09-10 进一步施压**:
- 北京宣布对台 34 项农产品停止零关税
- 涵盖水果、蔬菜、水产品
- 精准打击中南部农渔业
**2024-2025 持续的"边打边留"**:
- 北京未全面废除 ECFA——保留部分杠杆
- 高科技产业(半导体)未被直接制裁——避免自我伤害
- 传统产业与农业成为主要打击目标
@@ -328,14 +353,14 @@
### 变化一:政经分离的从可能到不可能
| 时期 | 台北的政经分离策略 | 北京的政经合一策略 | 结果 |
|---|---|---|---|
| 时期 | 台北的政经分离策略 | 北京的政经合一策略 | 结果 |
| --------- | ------------------ | -------------------- | ------------------ |
| 1987-1999 | 间接贸易、行政管制 | 默许民间交流以经促统 | 表面分离,实质连结 |
| 2000-2007 | 积极开放有效管理 | 吸引台资建立依赖 | 管制失效,依赖加深 |
| 2008-2013 | 先经后政 | 让利铺路 | 制度化整合 |
| 2014-2015 | 阻断整合 | 让利策略失效 | 分离再确认 |
| 2016-2022 | 经贸不扩展、新南向 | 有限度制裁 | 被动脱钩 |
| 2023- | 防御性脱钩 | 以经逼政、边打边留 | 强制政经合一 |
| 2000-2007 | 积极开放有效管理 | 吸引台资建立依赖 | 管制失效,依赖加深 |
| 2008-2013 | 先经后政 | 让利铺路 | 制度化整合 |
| 2014-2015 | 阻断整合 | 让利策略失效 | 分离再确认 |
| 2016-2022 | 经贸不扩展、新南向 | 有限度制裁 | 被动脱钩 |
| 2023- | 防御性脱钩 | 以经逼政、边打边留 | 强制政经合一 |
**分析含义**:政经分离在两岸关系中**结构性不可能**——因为北京始终将经济作为政治工具,而台北无法单方面维持分离。
@@ -370,26 +395,31 @@
分析两岸经贸关系,**持续观察以下指标**:
### 指标一:ECFA 存续状态
- 早收清单被中止的比例
- 北京是否全面废除 ECFA
- 台北是否主动退出 ECFA
### 指标二:贸易依存度变化
- 台湾对大陆(含香港)出口占总出口比例
- 关键产业(石化、机械、电子零组件)的出口结构变化
- 农产品出口市场的分散程度
### 指标三:投资流向
- 台商对大陆投资金额与件数
- 台商回台投资金额与产业别
- 新南向国家的台商投资趋势
### 指标四:制裁与反制裁的升级/降级
- 北京制裁的产业范围与力度
- 台北的反制措施(如对大陆产品的限制)
- 双方是否在 WTO 框架内解决争端
### 指标五:高科技供应链的连结与脱钩
- 半导体设备与技术的出口管制
- 台积电等关键企业在两岸的布局
- 美国科技管制对两岸经贸的间接影响
@@ -399,16 +429,19 @@
## 未来观察点(待追踪事件)
### 2025-2026 ECFA 的最终走向
- 北京是否进一步中止早收清单项目
- ECFA 是否名存实亡或正式终止
- 台湾传统产业的转型与淘汰
### 2028 前后的经贸政策辩论
- 各候选人的两岸经贸政见
- "重新谈判"vs"彻底脱钩"的路线竞争
- 北京是否以经济让利介入选举
### 非红供应链的成型程度
- 台湾、东南亚、印度的供应链替代能力
- 台商在大陆的"在地化"程度(是否可被替代)
- 关键原材料的替代来源
@@ -502,4 +535,4 @@ ECFA 只是关税减让框架,中止 ECFA 不等于停止贸易——**最惠
last_updated: 2026-05-08
**注**:本档案为活档案——需随事件持续更新,特别是 ECFA 调整与供应链重组的动态。
**注**:本档案为活档案——需随事件持续更新,特别是 ECFA 调整与供应链重组的动态。
@@ -109,20 +109,42 @@ async def _search_notebook(
"""从笔记 API 按关键词检索内容,返回 source_type='notebook' 的 Material 列表。"""
import httpx # 懒导入:仅在笔记模式下才需要
# 入参全 dump(token 只打长度,不打明文,防止泄漏到日志)
logger.info(
"[notebook-search] called keywords=%s search_space_id=%s source_ids=%s token_len=%d",
keywords,
search_space_id,
source_ids,
len(token) if token else 0,
)
base = (os.getenv("DEER_FLOW_NOTEBOOK_API_BASE") or "").rstrip("/")
if not base:
logger.warning("DEER_FLOW_NOTEBOOK_API_BASE not set, skipping notebook search")
# 改成 error 级别,warning 在很多部署里默认会被过滤掉看不到
logger.error(
"[notebook-search] DEER_FLOW_NOTEBOOK_API_BASE not set, skipping notebook search. "
"请在后端启动环境里设这个环境变量(指向笔记后端根地址,例如 http://localhost:8003)。"
)
return []
full_url = f"{base}{_NOTEBOOK_SEARCH_PATH}"
logger.info("[notebook-search] base=%s full_url=%s", base, full_url)
results: list[Material] = []
mat_idx = 0
try:
async with httpx.AsyncClient(timeout=20) as client:
for kw in keywords[:_NOTEBOOK_MAX_KEYWORDS]:
logger.info("[notebook-search] POST kw=%r body=%s", kw, {
"query": kw,
"source_ids": source_ids,
"search_space_id": search_space_id,
"top_k": _NOTEBOOK_TOP_K,
})
try:
# /mentioned-documents/search 接口无需鉴权,不传 Authorization header
resp = await client.post(
f"{base}{_NOTEBOOK_SEARCH_PATH}",
headers={"Authorization": f"Bearer {token}"},
full_url,
json={
"query": kw,
"source_ids": source_ids,
@@ -130,10 +152,29 @@ async def _search_notebook(
"top_k": _NOTEBOOK_TOP_K,
},
)
logger.info(
"[notebook-search] kw=%r http_status=%d body_len=%d",
kw,
resp.status_code,
len(resp.content or b""),
)
if resp.status_code >= 400:
# 把错误响应体打出来,方便定位 401 / 422 等
logger.error(
"[notebook-search] kw=%r HTTP %d response: %s",
kw,
resp.status_code,
(resp.text or "")[:1000],
)
resp.raise_for_status()
data = resp.json()
# 解析 documents 列表,每个 chunk 作为独立素材
documents: list[dict] = data.get("documents") if isinstance(data, dict) else []
logger.info(
"[notebook-search] kw=%r parsed documents=%d",
kw,
len(documents or []),
)
for doc in documents or []:
doc_title: str = doc.get("title") or "笔记文档"
doc_url: str = doc.get("url") or ""
@@ -154,10 +195,11 @@ async def _search_notebook(
url=doc_url,
))
except Exception as e:
logger.warning("notebook search kw=%r failed: %s", kw, e)
logger.warning("[notebook-search] kw=%r failed: %s", kw, e, exc_info=True)
except Exception as e:
logger.warning("notebook search client error: %s", e)
logger.warning("[notebook-search] client error: %s", e, exc_info=True)
logger.info("[notebook-search] done — returning %d materials", len(results))
return results
@@ -250,6 +292,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# Step 1.5: 笔记检索(如果传入了 notebook_search_space_id)
notebook_search_space_id: int | None = state.get("notebook_search_space_id")
notebook_materials: list[Material] = []
if notebook_search_space_id:
notebook_source_ids: list[int] = state.get("notebook_source_ids") or []
notebook_token: str = cfg.get("notebook_token") or ""
@@ -369,6 +412,13 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# 技能模式不再走下面的并发/合并逻辑,all_results_nested 留空、跳过 web 通用搜索
all_results_nested: list = []
# 把已收集到的素材挂在 raw_materials_pre,下方合并阶段直接挪过去
elif material_source == "notebook" and notebook_search_space_id:
# ── 纯笔记模式 ──
# 用户在表单选了「我的空间」作为素材来源 → 跳过 web/intranet 检索,
# 仅使用 Step 1.5 已拉取的 notebook_materials。
# 与 UI 文案"从空间中检索素材"对齐。
_emit(f"仅使用笔记素材,跳过 web/intranet 检索({len(notebook_materials)} 条)")
all_results_nested = []
else:
# ── 常规 web/intranet 模式 ──
provider = get_search_provider(ai_writing_config)
@@ -331,6 +331,13 @@ def _build_middlewares(
max_concurrent_subagents = cfg.get("max_concurrent_subagents", 3)
middlewares.append(SubagentLimitMiddleware(max_concurrent=max_concurrent_subagents))
# SkillReviewMiddleware — trigger background skill review after each turn.
# Gated by skill_evolution.enabled + skill_evolution.review_worker.enabled.
if resolved_app_config.skill_evolution.enabled and resolved_app_config.skill_evolution.review_worker.enabled:
from deerflow.agents.middlewares.skill_review_middleware import SkillReviewMiddleware
middlewares.append(SkillReviewMiddleware())
# LoopDetectionMiddleware — detect and break repetitive tool call loops
middlewares.append(LoopDetectionMiddleware())
@@ -1194,7 +1194,7 @@ When a search, knowledge, skill, or MCP tool returns a result list:
<notebook_mode>
## 角色
你是一个笔记本 AI 助手,**只能**基于笔记本文档来源回答用户问题,禁止使用任何其他知识。
你是一个空间 AI 助手,**只能**基于空间文档来源回答用户问题,禁止使用任何其他知识。
## 可用知识来源
{source_line}
@@ -1220,7 +1220,7 @@ When a search, knowledge, skill, or MCP tool returns a result list:
- 未调用 `search_notebook_sources` 就直接回答
- 引用不在检索结果中的 chunk_id
- 省略引用标记
- 使用笔记本来源之外的知识
- 使用空间来源之外的知识
</notebook_mode>"""
return prompt
@@ -0,0 +1,54 @@
"""SkillReviewMiddleware — triggers background skill review after each agent turn.
Hooks into ``after_agent``: once the lead agent graph has finished a full turn,
this middleware counts the tool calls made and, if the threshold is met, hands
the conversation messages to the SkillReviewWorker as a background asyncio task.
No blocking I/O or LLM calls happen in-band — the review runs fully in the
background and never delays the HTTP response.
"""
from __future__ import annotations
import logging
from typing import Any, override
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from langgraph.runtime import Runtime
from deerflow.config.app_config import get_app_config
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.skills.review_worker import maybe_start_review
logger = logging.getLogger(__name__)
class SkillReviewMiddlewareState(AgentState):
"""Minimal state schema — only needs ``messages``."""
pass
class SkillReviewMiddleware(AgentMiddleware[SkillReviewMiddlewareState]):
"""Dispatch a background skill review after every substantial agent turn.
'Substantial' is defined by ``skill_evolution.review_worker.min_tool_calls``
(default 5). Turns below the threshold are silently skipped to avoid
wasting LLM tokens on trivial exchanges.
"""
state_schema = SkillReviewMiddlewareState
@override
def after_agent(self, state: Any, runtime: Runtime) -> None:
"""Fire background review; never raises — failures are logged only."""
try:
app_config = get_app_config()
messages = state.get("messages", []) if hasattr(state, "get") else []
thread_id: str = runtime.context.get("thread_id", "") if runtime.context else ""
if not thread_id:
return
user_id = get_effective_user_id()
maybe_start_review(thread_id, messages, app_config, user_id)
except Exception:
logger.exception("SkillReviewMiddleware.after_agent: unexpected error (ignored)")
@@ -107,7 +107,7 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]):
content=last_message.content,
id=last_message.id,
name=last_message.name,
additional_kwargs={**last_message.additional_kwargs, "run_id": runtime.context.get("run_id"), "timestamp": datetime.now(UTC).isoformat()},
additional_kwargs={**last_message.additional_kwargs, "run_id": (runtime.context or {}).get("run_id"), "timestamp": datetime.now(UTC).isoformat()},
)
return {
@@ -1,30 +1,39 @@
from pydantic import BaseModel, Field
class CuratorConfig(BaseModel):
"""Configuration for the background skill curator."""
# 默认关闭:curator 仅在用户显式开启时才运行(配合 SkillEvolutionConfig.enabled
# 总开关,二者均为安全默认 False —— 未显式 opt-in 则整套技能进化不动作)。
enabled: bool = Field(default=False, description="Whether the curator runs automatically.")
interval_hours: int = Field(default=168, description="Hours between curator runs (default: 7 days).")
stale_after_days: int = Field(default=30, description="Days of inactivity before a skill is marked stale.")
archive_after_days: int = Field(default=90, description="Days of inactivity before a stale skill is archived.")
# LLM 评审整体超时:agent.invoke 内部可能多轮调 LLM,模型层的单次 request_timeout
# 兜不住整体;超过此秒数则中止本次评审,curator 正常收尾并释放运行锁。
review_timeout_seconds: int = Field(default=600, description="Overall timeout (seconds) for the LLM review pass.")
# curator LLM 评审使用的模型名(对应 config.yaml models[].name);为空则用主模型。
model_name: str | None = Field(default=None, description="Model name for the curator LLM review pass. Defaults to the primary chat model.")
class SkillEvolutionConfig(BaseModel):
"""Configuration for agent-managed skill evolution."""
enabled: bool = Field(
default=False,
description="Whether the agent can create and modify skills under skills/custom.",
)
curator: CuratorConfig = Field(
default_factory=CuratorConfig,
description="Configuration for the background skill curator.",
)
from pydantic import BaseModel, Field
class ReviewWorkerConfig(BaseModel):
"""Configuration for the post-turn background skill review worker."""
enabled: bool = Field(default=False, description="Whether the review worker is active.")
min_tool_calls: int = Field(default=5, description="Minimum tool calls in a turn to trigger review. Matches the prompt's '≥5 steps' hard condition so simple queries never reach the LLM reviewer.")
max_messages: int = Field(default=60, description="Max messages passed to the review LLM.")
model_name: str | None = Field(default=None, description="Model to use for review; defaults to primary model.")
timeout_seconds: int = Field(default=120, description="Overall timeout for a single review run.")
class CuratorConfig(BaseModel):
"""Configuration for the background skill curator."""
enabled: bool = Field(default=False, description="Whether the curator runs automatically.")
interval_hours: int = Field(default=168, description="Hours between curator runs (default: 7 days).")
stale_after_days: int = Field(default=30, description="Days of inactivity before a skill is marked stale.")
archive_after_days: int = Field(default=90, description="Days of inactivity before a stale skill is archived.")
review_timeout_seconds: int = Field(default=600, description="Overall timeout (seconds) for the LLM review pass.")
model_name: str | None = Field(default=None, description="Model name for the curator LLM review pass. Defaults to the primary chat model.")
class SkillEvolutionConfig(BaseModel):
"""Configuration for agent-managed skill evolution."""
enabled: bool = Field(
default=False,
description="Whether the agent can create and modify skills under skills/custom.",
)
curator: CuratorConfig = Field(
default_factory=CuratorConfig,
description="Configuration for the background skill curator.",
)
review_worker: ReviewWorkerConfig = Field(
default_factory=ReviewWorkerConfig,
description="Configuration for the post-turn background skill review worker.",
)
@@ -135,8 +135,18 @@ def resolve_user_path(user_id: str) -> Path:
return get_paths().user_dir(user_id) / _FILENAME
# Only these curator sub-fields may be overridden per user.
# All other fields (enabled, model_name, timeouts, etc.) are global-only.
_USER_ALLOWED_CURATOR_KEYS = {"interval_hours", "stale_after_days", "archive_after_days", "model_name"}
def load_user_overrides(user_id: str) -> dict[str, Any] | None:
"""Read the per-user overrides file for ``user_id`` (None if missing/invalid)."""
"""Read the per-user overrides file for ``user_id`` (None if missing/invalid).
Only the three curator threshold keys are honoured; any other keys
(e.g. legacy ``enabled`` or ``model_name``) are silently stripped so that
old files written by previous versions cannot override global-only settings.
"""
path = resolve_user_path(user_id)
if not path.exists():
return None
@@ -154,7 +164,14 @@ def load_user_overrides(user_id: str) -> dict[str, Any] | None:
if not isinstance(data, dict):
logger.warning("User skill_evolution overrides at %s must be a JSON object", path)
return None
return data
# Strip top-level keys that are not user-configurable (e.g. legacy "enabled").
curator_raw = data.get("curator")
if not isinstance(curator_raw, dict):
return None
filtered_curator = {k: v for k, v in curator_raw.items() if k in _USER_ALLOWED_CURATOR_KEYS}
if not filtered_curator:
return None
return {"curator": filtered_curator}
def save_user_overrides(user_id: str, payload: dict[str, Any]) -> Path:
@@ -0,0 +1,376 @@
"""Post-turn background skill review worker.
After a conversation turn completes, SkillReviewMiddleware calls
``maybe_start_review()``. This module runs a lightweight LLM React agent
(with skill_manage + skill_list tools) that reads the conversation history
and decides whether to create or patch skills.
Concurrency model: per-thread asyncio.Lock + a single _pending slot.
If a review is already running for the thread, the newer messages overwrite
the pending slot; once the current review finishes, it drains that slot.
"""
from __future__ import annotations
import asyncio
import logging
import textwrap
from typing import TYPE_CHECKING, Any
from langchain_core.messages import BaseMessage
if TYPE_CHECKING:
from deerflow.config.app_config import AppConfig
logger = logging.getLogger(__name__)
_locks: dict[str, asyncio.Lock] = {}
_pending: dict[str, list[BaseMessage]] = {}
def _get_lock(thread_id: str) -> asyncio.Lock:
if thread_id not in _locks:
_locks[thread_id] = asyncio.Lock()
return _locks[thread_id]
def extract_tool_call_count(messages: list[BaseMessage]) -> int:
"""Count total tool calls across all AI messages in the conversation."""
count = 0
for msg in messages:
tool_calls: Any = None
if hasattr(msg, "tool_calls"):
tool_calls = msg.tool_calls
elif isinstance(msg, dict):
tool_calls = msg.get("tool_calls")
if tool_calls:
count += len(tool_calls)
return count
def _format_messages_for_review(messages: list[BaseMessage], max_messages: int) -> str:
"""Serialize recent messages to a readable text block for the LLM reviewer."""
recent = messages[-max_messages:]
lines: list[str] = []
for msg in recent:
role = type(msg).__name__.replace("Message", "")
content = msg.content if isinstance(msg.content, str) else str(msg.content)
tool_calls = getattr(msg, "tool_calls", None)
if tool_calls:
calls_str = ", ".join(
f"{tc.name if hasattr(tc, 'name') else tc.get('name', '?')}()"
for tc in tool_calls
)
lines.append(f"[{role}] <tool_calls: {calls_str}>")
else:
preview = content[:400].replace("\n", " ")
lines.append(f"[{role}] {preview}")
return "\n".join(lines)
_REVIEW_SYSTEM_PROMPT = textwrap.dedent("""\
# Role: 技能评审代理(保守模式)
## Profile
- language: zh-CN
- description: 审阅已完成对话,仅在出现强证据的高价值可复用工作流时才沉淀为技能。**默认决策是 do-nothing** —— 宁可漏掉一个普通技能,也绝不创建垃圾技能。
- background: 历史问题:之前版本触发过于宽松,沉淀了大量低价值技能,污染了技能库。本版本针对性收紧。
- personality: 极度保守、苛刻挑剔、宁缺勿滥;面对模糊场景默认 do-nothing。
- expertise: 高价值工作流识别、严格去重、SKILL.md 规范、do-nothing 判定。
- target_audience: 技能库维护者、平台工程师。
## Skills
1. 工作流抽象与评审
- 价值评估: 区分一次性数据请求与具有沉淀价值的工作流。
- 去数据化归纳: 抽象出与具体数据无关的通用步骤、参数与注意事项。
- 决策映射: 在 do-nothing / patch / create 三者中做出可解释的唯一决定。
2. 文档与规范控制
- SKILL.md 编写: 简洁步骤 + 命令 + 坑点提示(非对话稿)。
- 命名规范: kebab-case(小写字母/数字/连字符,≤64 字符)。
- 触发描述: 以 "When" 开头的单句(≤1024 字符、无尖括号)。
- 结构校验: frontmatter 允许键、name 一致性、首尾分隔线。
3. 工具使用与错误处理
- skill_list 优先调用,识别功能重叠与现有覆盖范围。
- skill_manage 正确选 action=create / patch,并传递合规内容。
- 报错诊断与修正后才能重试,禁止内容不变重提。
4. 质量控制与安全
- 去重策略: 少而精,避免噪声与重复。
- 隐私合规: 严禁嵌入用户特定数据、文件路径、凭据。
- 一致性: name / description / content 三者语义对齐。
## Rules
### 0. 默认决策
**默认就是 do-nothing。** 仅当下面"必要条件"全部满足时才考虑创建;任一不满足 → do-nothing。垃圾技能比少一个技能更糟糕。
### 1. 创建必要条件(必须全部满足,AND 关系)
要创建一个新技能,以下条件**必须同时**满足:
1. **多步且有依赖**: 对话中至少 5 次工具调用,且步骤间存在依赖关系(不是并行查询,也不是同一工具的简单循环)。
2. **包含纠错经验**: 包含至少一次失败 / 报错 / 意外结果,agent 通过修正后达到目标。这种"踩坑—纠错"才是有沉淀价值的知识。
3. **跨场景可复用**: 抽离掉对话中的具体数据后,步骤本身在不同输入 / 不同数据集下仍然成立。
4. **不与现有技能重叠**: skill_list 中没有功能覆盖率 ≥70% 的现有技能(若有,应当 patch 而非新建)。
5. **能写出有料的 SKILL.md**: 抽象后能产出 ≥300 字、包含具体步骤 / 命令 / 坑点的工作流指南。写不出说明本来就没料。
### 2. 明确不要创建(即使工具调用很多也不要建)
- **单工具循环**: 连续多次 web_search 都只是搜索,没有形成工作流。
- **一次性数据查询**: 仅是"搜索 + 总结"模式,没有可复用结构。
- **LLM 常识可答**: 如"如何配置 Python 虚拟环境"这类公开文档 / 通识问题。
- **流程过度通用**: 如"先搜索再总结再回答"这种描述等于没说。
- **场景过窄**: 只在某一特定数据集 / 系统 / 项目下有效,换环境就失效。
- **内容字数不足**: 写不出 ≥300 字含真干货的 SKILL.md。
- **仅为"做点什么"而建**: 找不到清晰理由就 do-nothing,不要为了"完成 review"而强行创建。
### 3. 创建前自检(4 题,全部"是"才能 create)
提交 create 前在心中依次回答以下问题,任一为"否" → 立即放弃创建、改 do-nothing:
1. 如果我不创建这个技能,下一个遇到类似任务的用户会损失什么?(答得上来才证明有沉淀价值)
2. 这个技能是否能被一句话简短复述?若"能",说明信息量不足 → do-nothing。
3. SKILL.md 写完后是否包含至少 3 个具体步骤或命令?
4. 步骤中是否含有非显而易见的坑点提示(不是 LLM 凭常识就能猜到的)?
### 4. 单次 review 上限
**每次 review 最多创建 1 条新技能。** 即便发现多个候选,也只挑最有价值的那一个,其余 do-nothing。这是硬约束,不接受例外。
### 5. Patch 优先于 Create
若 skill_list 中已存在功能相近(即便不完全相同)的技能,优先选择 patch 现有技能补充改进点,而非新建。新技能名额只留给**全新类别**的工作流。
### 6. 格式与命名硬约束
- 命名:kebab-case,仅含小写字母 / 数字 / 连字符,≤64 字符(如 git-bisect-bug-hunt、api-load-test)。
- description:以 "When " 开头,单句,≤1024 字符,禁止尖括号 < >。
- content:合法 SKILL.md,包含 frontmatter 与正文。
- frontmatter 仅允许键:name、description、license、allowed-tools、metadata、compatibility、version、author。
- frontmatter 首行与末行必须为 "---" 独占一行。
- frontmatter 的 name 字段必须与 skill_manage 调用的 name 参数**完全一致**;不一致直接调用失败。
- 禁止添加 tags、category、type 等额外键。
- content 必须是简洁工作流指南,禁止对话转录。
- 严禁包含用户特定数据、文件路径、凭据。
### 7. 错误处理
- 调用 skill_manage 报错 → 阅读错误信息,定位根因,**修正内容后**再重试。
- 同名已存在 → 改用 action=patch(直接 create 同名会失败)。
- 禁止在未修改内容的情况下重复提交。
## Workflows
### 目标
仅在发现**高价值、可复用、有踩坑沉淀**的工作流时,创建或修补一条技能;其余一律 do-nothing。
### 步骤 1: 获取基线
调用 skill_list,记录现有技能名称与功能覆盖范围。这一步在每次 review 中必须执行。
### 步骤 2: 价值初筛(三关,任一不过即 do-nothing 结束)
通读对话并依次确认:
- 工具调用总数是否 ≥5?否则 do-nothing 结束。
- 是否包含失败 / 报错 / 修正过程?否则 do-nothing 结束。
- 步骤间是否存在依赖关系?否则 do-nothing 结束。
只有三关全过才进入步骤 3。
### 步骤 3: 候选抽象
将对话抽象为候选技能要素:
- 候选名称(kebab-case)
- 触发描述(When ... )
- 关键步骤、命令、坑点提示
- 剔除所有数据特定内容与隐私信息
### 步骤 4: 重叠检查
与 skill_list 比对:
- 完全无重叠 → 候选 action=create,进入步骤 5。
- 部分重叠(功能覆盖 ≥70%)→ 候选 action=patch,跳到步骤 6。
- 完全覆盖 → do-nothing 结束。
### 步骤 5: 创建前自检
若候选为 create,依次回答 Rules 第 3 节的 4 个自检问题。任一为"否" → 立即放弃创建,do-nothing 结束。全部为"是"才进入步骤 6。
### 步骤 6: 提交
- create:调用 skill_manage(action="create", name, content)。
- patch:调用 skill_manage(action="patch", name, content),仅针对差异 / 改进点。
- 一次 review 最多一次写操作。
### 步骤 7: 错误处理
若报错,按 Rules 第 7 节处理:阅读错误信息 → 修正根因 → 再试。禁止内容不变重提。
### 预期结果
- **多数情况下**: do-nothing。这是好的结果,不是失败。
- **少数情况下**: 一条高价值新技能,或对现有技能的一次有针对性 patch。
- **永远不要**: 为了"做点什么"而创建低价值技能。
## Initialization
作为技能评审代理(保守模式),你必须严格遵守 Rules,按 Workflows 执行。**默认决策是 do-nothing。** 只有强证据完全满足所有创建条件时,才考虑创建。创建一条垃圾技能比放弃一条普通技能更糟糕。
""")
async def _run_review(thread_id: str, messages: list[BaseMessage], app_config: "AppConfig", user_id: str = "default") -> None:
"""Core review logic: create_agent with full middleware chain, no 'silent creation' constraint.
Uses create_agent (the deerflow agent factory) directly so we can:
- Reuse the full middleware chain (ToolErrorHandlingMiddleware, etc.)
- Reuse the full tool set
- But pass app_config with skill_evolution.enabled=False to strip the
"Never create skills silently without user confirmation" constraint
from the system prompt.
Review instructions are injected via PromptPrefixMiddleware through
additional_kwargs["prompt_prefix"] on the HumanMessage.
"""
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
from deerflow.agents.lead_agent.prompt import apply_prompt_template
from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares
from deerflow.agents.thread_state import ThreadState
from deerflow.models import create_chat_model
from deerflow.tools.tools import get_available_tools
evolution_cfg = app_config.skill_evolution
timeout = evolution_cfg.review_worker.timeout_seconds
max_messages = evolution_cfg.review_worker.max_messages
model_name = evolution_cfg.review_worker.model_name
conversation_text = _format_messages_for_review(messages, max_messages)
# Disable skill_evolution so the system prompt does NOT include
# "Never create skills silently without user confirmation".
# Also prevents SkillReviewMiddleware from registering (no recursive reviews).
review_app_config = app_config.model_copy(deep=True)
review_app_config.skill_evolution.enabled = False
try:
model = create_chat_model(name=model_name, app_config=review_app_config)
except Exception:
logger.warning("skill_review_worker: could not load model %r, aborting review", model_name)
return
tools = get_available_tools(app_config=review_app_config, subagent_enabled=False)
middlewares = build_lead_runtime_middlewares(app_config=review_app_config)
middlewares.append(PromptPrefixMiddleware())
system_prompt = apply_prompt_template(
subagent_enabled=False,
max_concurrent_subagents=0,
app_config=review_app_config,
memory_injection_enabled=False,
)
agent = create_agent(
model=model,
tools=tools,
middleware=middlewares,
system_prompt=system_prompt,
state_schema=ThreadState,
)
# RunnableConfig supplies thread_id so ThreadDataMiddleware is satisfied.
import uuid
run_config = RunnableConfig(configurable={"thread_id": f"__skill-review__{uuid.uuid4()}"})
# Review instructions arrive as prompt_prefix: PromptPrefixMiddleware prepends
# them before every model call without persisting them to state.
first_message = HumanMessage(
content=(
"Review this conversation and apply any skill changes you deem appropriate.\n\n"
f"<conversation>\n{conversation_text}\n</conversation>"
),
additional_kwargs={"prompt_prefix": _REVIEW_SYSTEM_PROMPT},
)
def _invoke_with_user_context() -> dict:
from deerflow.runtime.user_context import set_current_user, reset_current_user
class _SimpleUser:
def __init__(self, uid: str) -> None:
self.id = uid
token = set_current_user(_SimpleUser(user_id))
try:
return agent.invoke({"messages": [first_message]}, config=run_config)
finally:
reset_current_user(token)
try:
result = await asyncio.wait_for(
asyncio.get_event_loop().run_in_executor(None, _invoke_with_user_context),
timeout=timeout,
)
# Log the final AI message so we can see what the review agent decided.
final_msgs = (result or {}).get("messages", [])
final_text = ""
for msg in reversed(final_msgs):
content = getattr(msg, "content", None)
if content and isinstance(content, str) and content.strip():
final_text = content.strip()[:1000]
break
logger.info(
"skill_review_worker: review completed for thread %s — agent said: %s",
thread_id,
final_text or "(no text response)",
)
except TimeoutError:
logger.warning("skill_review_worker: review timed out after %ds (thread %s)", timeout, thread_id)
except Exception:
logger.exception("skill_review_worker: review failed for thread %s", thread_id)
async def _run_review_with_pending(
thread_id: str,
messages: list[BaseMessage],
app_config: "AppConfig",
user_id: str = "default",
) -> None:
"""Run one review; if a newer snapshot arrived while we ran, drain it."""
lock = _get_lock(thread_id)
async with lock:
await _run_review(thread_id, messages, app_config, user_id)
pending = _pending.pop(thread_id, None)
if pending is not None:
asyncio.create_task(_run_review_with_pending(thread_id, pending, app_config, user_id))
def maybe_start_review(
thread_id: str,
messages: list[BaseMessage],
app_config: "AppConfig",
user_id: str = "default",
) -> bool:
"""Schedule a background review for ``thread_id``.
Returns True if a new task was started, False if queued or skipped.
"""
evolution_cfg = app_config.skill_evolution
if not evolution_cfg.enabled or not evolution_cfg.review_worker.enabled:
return False
tool_count = extract_tool_call_count(messages)
if tool_count < evolution_cfg.review_worker.min_tool_calls:
logger.debug(
"skill_review_worker: skipping review for thread %s (%d tool calls < min %d)",
thread_id,
tool_count,
evolution_cfg.review_worker.min_tool_calls,
)
return False
lock = _get_lock(thread_id)
if lock.locked():
_pending[thread_id] = messages
logger.debug("skill_review_worker: queued review for thread %s (review in progress)", thread_id)
return False
asyncio.create_task(_run_review_with_pending(thread_id, messages, app_config, user_id))
logger.info("skill_review_worker: started review for thread %s (%d tool calls) user=%s", thread_id, tool_count, user_id)
return True
@@ -1,7 +1,7 @@
"""笔记本 RAG 检索工具。
"""空间 RAG 检索工具。
从 LangGraph configurable 中读取 notebook_search_space_id / notebook_source_ids /
notebook_token,调用外部检索接口返回相关文本块,供 LLM 在笔记本模式下引用回答。
notebook_token,调用外部检索接口返回相关文本块,供 LLM 在空间模式下引用回答。
API base URL 通过环境变量 DEER_FLOW_NOTEBOOK_API_BASE 配置。
"""
@@ -134,27 +134,58 @@ def _default_agent_store() -> Any | None:
return None
async def _assert_agent_may_delete(store: SkillStore, name: str, user_id: str) -> dict[str, Any]:
"""Pre-flight ownership check before agent-delete touches the on-disk file.
async def _assert_skill_is_agent_authored(name: str) -> None:
"""Refuse to modify skills authored by a human (UI upload / adopt / legacy).
LocalSkillStorage doesn't know about users — it deletes anything that
matches by name. Without this guard, a malicious or hallucinating agent
could destroy another user's skill just by knowing the name. We refuse
unless the DB confirms ``user_id`` owns the skill outright.
Even when ownership lines up, we draw a hard line at content the *human*
curated through the Web UI: uploads, adopted orphan skills, and legacy
skills with no usage entry (which default to ``source='upload'``).
Those represent reviewed content the user has explicitly endorsed —
letting the agent rewrite them risks silent drift away from what was
approved. Only skills marked ``source='agent'`` (set by
``mark_agent_created`` in the ``create`` flow) remain editable so the
evolution loop can keep refining them.
Returns the DB record for use in the subsequent cascade step. Raises
``ValueError`` (mapped to a tool error visible to the LLM) on refusal.
Raises ``ValueError`` (mapped to a tool error visible to the LLM) on
refusal.
"""
from deerflow.skills.usage import get_entry
entry = await _to_thread(get_entry, name)
if entry.get("source") != "agent":
raise ValueError(
f"Refused: skill '{name}' was authored by a human via the Web UI "
f"and cannot be modified by an agent. Ask the owner to edit it "
f"through the Web UI, or have the user duplicate it to a new "
f"agent-owned skill first."
)
async def _assert_agent_may_modify(store: SkillStore, name: str, user_id: str) -> dict[str, Any]:
"""Pre-flight ownership check before an agent-driven write touches a custom skill.
LocalSkillStorage operates by name only — patch/edit/write_file/remove_file
and delete all resolve a name to a directory without knowing who owns it.
Without this guard a malicious or hallucinating agent could overwrite or
destroy another user's skill (including UI-uploaded skills) just by
knowing the name. We refuse unless the DB confirms ``user_id`` owns the
skill outright.
Returns the DB record so callers that need owner info (e.g. delete
cascade) can reuse it. Raises ``ValueError`` (mapped to a tool error
visible to the LLM) on refusal.
"""
record = await store.get_any(name)
if record is None:
# No DB row — could be a legacy/orphan skill. Refuse rather than
# delete blindly; admin can clean up via the Web UI.
# touch blindly; users should adopt it via /api/skills/reconcile-custom
# or an admin can clean it up via the Web UI.
raise ValueError(
f"Refused: skill '{name}' has no ownership record. Ask a human to delete it via the Web UI."
f"Refused: skill '{name}' has no ownership record. Ask a human to adopt or manage it via the Web UI."
)
owner = record.get("owner_user_id")
if owner is None:
raise ValueError(f"Refused: skill '{name}' is built-in and cannot be deleted by an agent.")
raise ValueError(f"Refused: skill '{name}' is built-in and cannot be modified by an agent.")
if owner != user_id:
raise ValueError(f"Refused: skill '{name}' is owned by another user.")
return record
@@ -200,6 +231,10 @@ async def _skill_manage_impl(
lock = _get_lock(name)
thread_id = _get_thread_id(runtime)
skill_storage = get_or_new_skill_storage()
# Resolved up-front so ownership checks in the write-action branches stay
# one-liners. Cheap when the DB is unconfigured (returns None).
user_id = get_effective_user_id()
skill_store = _default_skill_store()
async with lock:
if action == "create":
@@ -224,14 +259,16 @@ async def _skill_manage_impl(
# surfaces it (custom skills without a DB row are filtered out by
# ``GET /api/skills``). Resolve the user_id *here* on the main
# async stack — contextvars don't propagate into ``to_thread``.
store = _default_skill_store()
if store is not None:
await _ensure_skill_db_registered(store, name, get_effective_user_id())
if skill_store is not None:
await _ensure_skill_db_registered(skill_store, name, user_id)
await refresh_skills_system_prompt_cache_async()
return f"Created custom skill '{name}'."
if action == "edit":
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
if skill_store is not None:
await _assert_agent_may_modify(skill_store, name, user_id)
await _assert_skill_is_agent_authored(name)
if content is None:
raise ValueError("content is required for edit.")
await _to_thread(skill_storage.validate_skill_markdown_content, name, content)
@@ -250,6 +287,9 @@ async def _skill_manage_impl(
if action == "patch":
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
if skill_store is not None:
await _assert_agent_may_modify(skill_store, name, user_id)
await _assert_skill_is_agent_authored(name)
if find is None or replace is None:
raise ValueError("find and replace are required for patch.")
skill_file = skill_storage.get_custom_skill_file(name)
@@ -277,11 +317,10 @@ async def _skill_manage_impl(
# Ownership pre-check: LocalSkillStorage deletes purely by name,
# so without this an agent could blow away another user's skill
# just by knowing the name. Raise BEFORE touching the filesystem.
user_id = get_effective_user_id()
store = _default_skill_store()
record: dict[str, Any] | None = None
if store is not None:
record = await _assert_agent_may_delete(store, name, user_id)
if skill_store is not None:
record = await _assert_agent_may_modify(skill_store, name, user_id)
await _assert_skill_is_agent_authored(name)
await _to_thread(
skill_storage.delete_custom_skill,
@@ -306,9 +345,9 @@ async def _skill_manage_impl(
await _to_thread(forget, name)
except Exception:
logger.debug("Failed to forget usage entry for %r", name, exc_info=True)
if store is not None:
if skill_store is not None:
try:
await store.delete(name, user_id)
await skill_store.delete(name, user_id)
except Exception:
logger.debug("Failed to delete DB row for %r", name, exc_info=True)
agent_store = _default_agent_store()
@@ -330,6 +369,9 @@ async def _skill_manage_impl(
if action == "write_file":
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
if skill_store is not None:
await _assert_agent_may_modify(skill_store, name, user_id)
await _assert_skill_is_agent_authored(name)
if path is None or content is None:
raise ValueError("path and content are required for write_file.")
target = await _to_thread(skill_storage.ensure_safe_support_path, name, path)
@@ -347,6 +389,9 @@ async def _skill_manage_impl(
if action == "remove_file":
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
if skill_store is not None:
await _assert_agent_may_modify(skill_store, name, user_id)
await _assert_skill_is_agent_authored(name)
if path is None:
raise ValueError("path is required for remove_file.")
target = await _to_thread(skill_storage.ensure_safe_support_path, name, path)
@@ -0,0 +1,156 @@
"""Tests that the curator scheduler does not block application startup.
Verifies:
1. _curator_scheduler_loop is launched as a non-blocking asyncio task.
2. The scheduler sleeps for the startup delay before doing any work.
3. maybe_run_curator is NOT called during the startup delay window.
4. When the lock is unavailable the loop exits immediately without calling the curator.
NOTE: ``import app.gateway.app as m`` resolves to the FastAPI *instance* because
``app/gateway/__init__.py`` re-exports ``app = FastAPI(...)``. The actual module
object lives in ``sys.modules["app.gateway.app"]`` and is used here via a helper.
"""
from __future__ import annotations
import asyncio
import sys
from unittest.mock import MagicMock, patch
def _gateway_module():
"""Return the real app.gateway.app *module* (not the FastAPI instance)."""
import importlib
importlib.import_module("app.gateway.app") # side-effect: populates sys.modules
return sys.modules["app.gateway.app"]
_STARTUP_DELAY = 300 # matches _CURATOR_STARTUP_DELAY_SECONDS in app.gateway.app
_CHECK_INTERVAL = 3600 # matches _CURATOR_CHECK_INTERVAL_SECONDS
def test_scheduler_does_not_block_startup():
"""create_task must return immediately — startup must not await the loop body."""
mod = _gateway_module()
curator_called = False
async def fake_sleep(_delay):
pass
async def fake_curator():
nonlocal curator_called
curator_called = True
async def run():
with (
patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
patch.object(mod, "_release_singleton_lock"),
patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
patch("asyncio.sleep", fake_sleep),
):
task = asyncio.create_task(mod._curator_scheduler_loop())
# create_task must return immediately; the coroutine is still pending
assert not task.done(), "task should be running in the background, not finished"
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(run())
assert not curator_called, "curator must not fire before the startup delay completes"
def test_scheduler_sleeps_startup_delay_before_first_curator_call():
"""The very first asyncio.sleep must use the startup-delay value (300 s)."""
mod = _gateway_module()
sleep_calls: list[float] = []
curator_called = False
async def fake_sleep(delay):
sleep_calls.append(delay)
raise asyncio.CancelledError # stop after recording first sleep
async def fake_curator():
nonlocal curator_called
curator_called = True
async def run():
with (
patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
patch.object(mod, "_release_singleton_lock"),
patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
patch("asyncio.sleep", fake_sleep),
):
task = asyncio.create_task(mod._curator_scheduler_loop())
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(run())
assert sleep_calls, "asyncio.sleep was never called"
assert sleep_calls[0] == _STARTUP_DELAY, (
f"first sleep must be the startup delay ({_STARTUP_DELAY}s), got {sleep_calls[0]}"
)
assert not curator_called, "curator must not be called before startup delay elapses"
def test_scheduler_actually_calls_curator_after_startup_delay():
"""After the startup delay elapses, maybe_run_curator must be called."""
mod = _gateway_module()
curator_call_count = 0
async def fake_sleep(*_):
pass # instant — simulate time passing
async def fake_curator():
nonlocal curator_call_count
curator_call_count += 1
# After first call cancel the loop so the test terminates
raise asyncio.CancelledError
async def run():
with (
patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
patch.object(mod, "_release_singleton_lock"),
patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
patch("asyncio.sleep", fake_sleep),
):
task = asyncio.create_task(mod._curator_scheduler_loop())
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(run())
assert curator_call_count >= 1, (
f"maybe_run_curator should have been called at least once, got {curator_call_count}"
)
def test_scheduler_skips_when_lock_unavailable():
"""When another worker holds the lock the loop must exit without doing any work."""
mod = _gateway_module()
curator_called = False
async def fake_curator():
nonlocal curator_called
curator_called = True
async def run():
with (
patch.object(mod, "_acquire_singleton_lock", return_value=None),
patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
):
await mod._curator_scheduler_loop()
asyncio.run(run())
assert not curator_called, "curator must not run when lock is not acquired"
@@ -0,0 +1,108 @@
"""Tests for SkillReviewMiddleware."""
from unittest.mock import MagicMock, patch
from langchain_core.messages import AIMessage, HumanMessage
from deerflow.agents.middlewares.skill_review_middleware import SkillReviewMiddleware
def _make_state(messages):
state = MagicMock()
state.get.side_effect = lambda key, default=None: messages if key == "messages" else default
return state
def _make_runtime(thread_id="thread-abc"):
runtime = MagicMock()
runtime.context = {"thread_id": thread_id} if thread_id else {}
return runtime
def test_after_agent_calls_maybe_start_review():
"""When enabled and tool calls exist, maybe_start_review is called with correct thread_id."""
middleware = SkillReviewMiddleware()
ai_msg = MagicMock(spec=AIMessage)
tc = MagicMock()
tc.name = "bash"
ai_msg.tool_calls = [tc]
state = _make_state([HumanMessage(content="hi"), ai_msg])
runtime = _make_runtime("thread-xyz")
app_cfg = MagicMock()
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", return_value=app_cfg):
with patch("deerflow.agents.middlewares.skill_review_middleware.maybe_start_review") as mock_review:
middleware.after_agent(state, runtime)
mock_review.assert_called_once()
call_args = mock_review.call_args
assert call_args[0][0] == "thread-xyz" # thread_id
assert call_args[0][2] is app_cfg # app_config
def test_after_agent_no_thread_id_skips_review():
"""When thread_id is missing, maybe_start_review is NOT called."""
middleware = SkillReviewMiddleware()
state = _make_state([HumanMessage(content="hi")])
runtime = _make_runtime("")
app_cfg = MagicMock()
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", return_value=app_cfg):
with patch("deerflow.agents.middlewares.skill_review_middleware.maybe_start_review") as mock_review:
middleware.after_agent(state, runtime)
mock_review.assert_not_called()
def test_after_agent_exception_is_swallowed():
"""Exceptions in after_agent must not propagate."""
middleware = SkillReviewMiddleware()
state = _make_state([HumanMessage(content="hi")])
runtime = _make_runtime("thread-err")
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", side_effect=RuntimeError("boom")):
# Must not raise
middleware.after_agent(state, runtime)
def test_skill_review_middleware_in_chain_when_enabled():
"""SkillReviewMiddleware appears in middleware chain before LoopDetectionMiddleware when enabled."""
from unittest.mock import patch, MagicMock
from deerflow.agents.lead_agent.agent import _build_middlewares
from deerflow.agents.middlewares.skill_review_middleware import SkillReviewMiddleware
config = {
"configurable": {
"thread_id": "test",
"is_plan_mode": False,
"is_bootstrap": False,
"is_scheduled_run": False,
}
}
app_cfg = MagicMock()
app_cfg.skill_evolution.enabled = True
app_cfg.skill_evolution.review_worker.enabled = True
app_cfg.token_usage.enabled = False
app_cfg.tool_search.enabled = False
app_cfg.skill_stop.enabled = False
app_cfg.summarization.enabled = False
app_cfg.memory = MagicMock()
app_cfg.guardrails.enabled = False
app_cfg.title = MagicMock()
app_cfg.title.model_name = None
with patch("deerflow.agents.lead_agent.agent.get_app_config", return_value=app_cfg):
with patch("deerflow.agents.lead_agent.agent.build_lead_runtime_middlewares", return_value=[]):
with patch("deerflow.agents.lead_agent.agent._create_summarization_middleware", return_value=None):
with patch("deerflow.agents.lead_agent.agent._create_todo_list_middleware", return_value=None):
with patch("deerflow.models.vision.vision_enabled", return_value=False):
middlewares = _build_middlewares(config, model_name=None, app_config=app_cfg)
type_names = [type(m).__name__ for m in middlewares]
assert "SkillReviewMiddleware" in type_names, f"SkillReviewMiddleware not in chain: {type_names}"
review_idx = type_names.index("SkillReviewMiddleware")
loop_idx = type_names.index("LoopDetectionMiddleware")
assert review_idx < loop_idx, f"SkillReviewMiddleware ({review_idx}) must be before LoopDetectionMiddleware ({loop_idx})"
@@ -0,0 +1,102 @@
from deerflow.config.skill_evolution_config import SkillEvolutionConfig
def test_review_worker_config_defaults():
cfg = SkillEvolutionConfig()
assert cfg.review_worker.enabled is False
assert cfg.review_worker.min_tool_calls == 5
assert cfg.review_worker.timeout_seconds == 120
assert cfg.review_worker.model_name is None
assert cfg.review_worker.max_messages == 60
def test_review_worker_config_from_dict():
cfg = SkillEvolutionConfig.model_validate({
"enabled": True,
"review_worker": {"enabled": True, "min_tool_calls": 5, "model_name": "my-model"},
})
assert cfg.review_worker.enabled is True
assert cfg.review_worker.min_tool_calls == 5
assert cfg.review_worker.model_name == "my-model"
from unittest.mock import AsyncMock, MagicMock, patch
import asyncio
import pytest
from langchain_core.messages import HumanMessage
def _make_ai_message(tool_calls_data: list[dict]):
msg = MagicMock()
msg.__class__.__name__ = "AIMessage"
tcs = []
for tc in tool_calls_data:
m = MagicMock()
m.name = tc["name"]
m.args = tc.get("args", {})
tcs.append(m)
msg.tool_calls = tcs
return msg
def test_extract_tool_call_count_counts_all():
from deerflow.skills.review_worker import extract_tool_call_count
msgs = [
HumanMessage(content="do something"),
_make_ai_message([{"name": "bash"}, {"name": "read_file"}]),
_make_ai_message([{"name": "write_file"}]),
]
assert extract_tool_call_count(msgs) == 3
def test_extract_tool_call_count_empty():
from deerflow.skills.review_worker import extract_tool_call_count
assert extract_tool_call_count([]) == 0
def test_maybe_start_review_disabled_returns_false():
from deerflow.skills.review_worker import maybe_start_review
cfg = MagicMock()
cfg.skill_evolution.enabled = False
cfg.skill_evolution.review_worker.enabled = True
cfg.skill_evolution.review_worker.min_tool_calls = 2
msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])]
assert maybe_start_review("thread-1", msgs, cfg) is False
def test_maybe_start_review_too_few_tool_calls_returns_false():
from deerflow.skills.review_worker import maybe_start_review
cfg = MagicMock()
cfg.skill_evolution.enabled = True
cfg.skill_evolution.review_worker.enabled = True
cfg.skill_evolution.review_worker.min_tool_calls = 5
msgs = [_make_ai_message([{"name": "bash"}])]
assert maybe_start_review("thread-4", msgs, cfg) is False
def test_maybe_start_review_starts_task():
from deerflow.skills import review_worker as rw
# Reset module state so thread-3 has no lock
rw._locks.pop("thread-3", None)
rw._pending.pop("thread-3", None)
cfg = MagicMock()
cfg.skill_evolution.enabled = True
cfg.skill_evolution.review_worker.enabled = True
cfg.skill_evolution.review_worker.min_tool_calls = 2
cfg.skill_evolution.review_worker.timeout_seconds = 30
cfg.skill_evolution.review_worker.max_messages = 20
cfg.skill_evolution.review_worker.model_name = None
msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])]
with patch("deerflow.skills.review_worker.asyncio") as mock_asyncio:
fake_lock = MagicMock()
fake_lock.locked.return_value = False
mock_asyncio.Lock.return_value = fake_lock
mock_asyncio.create_task = MagicMock()
result = rw.maybe_start_review("thread-3", msgs, cfg)
assert result is True
mock_asyncio.create_task.assert_called_once()
@@ -1,10 +1,16 @@
{
"enabled": true,
"moderation_model_name": null,
"curator": {
"enabled": false,
"interval_hours": 168,
"enabled": true,
"interval_hours": 24,
"stale_after_days": 30,
"archive_after_days": 90
},
"review_worker": {
"enabled": true,
"min_tool_calls": 3,
"max_messages": 60,
"timeout_seconds": 120,
"model_name": null
}
}