This commit is contained in:
fengxiangdebaozi committed 2026-05-23 17:23:22 +08:00
commit 3987ee7ee3
48 files changed
+1858 -334

No files matched your search

@@ -181,9 +181,10 @@ async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse
# V1 模式:每个桶读 memory.json,内容为 {facts, user, history}
buckets: list[AdminMemoryBucket] = []
try:
buckets.append(AdminMemoryBucket(name="global", display_name="用户记忆", data=_read_v1_memory_bucket(user_id, None)))
buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data=_read_v1_memory_bucket(user_id, None)))
except Exception:
buckets.append(AdminMemoryBucket(name="global", display_name="用户记忆", data={}))
logger.exception("Failed to read V1 global memory for user %s", user_id)
buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data={}))
for agent in owned_agents:
agent_id: str = agent["id"]
@@ -198,7 +199,7 @@ async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse
# V2 模式:读 USER.md / MEMORY.md,内容为 {entries: [...]}
buckets = [
AdminMemoryBucket(name="global", display_name="用户画像", data=_read_memory_bucket(user_id, None))
AdminMemoryBucket(name="default", display_name="默认助手", data=_read_memory_bucket(user_id, None))
]
for agent in owned_agents:
agent_id = agent["id"]
@@ -39,7 +39,7 @@ router = APIRouter(prefix="/api", tags=["memory"])
# 表示"全局桶 (无 agent)"的 agent 取值。
_GLOBAL_AGENT_VALUES: frozenset[str] = frozenset({"", "global", "__global__"})
_GLOBAL_AGENT_VALUES: frozenset[str] = frozenset({"", "global", "__global__", "default"})
def _resolve_agent_name(agent: str | None) -> str | None:
@@ -161,6 +161,13 @@ class SkillUploadValidateResponse(BaseModel):
warnings: list[str] = Field(default_factory=list, description="Non-fatal warnings to surface to the user")
class SkillReconcileResponse(BaseModel):
"""Response model for reconciling custom skills found on disk."""
adopted: list[str] = Field(default_factory=list, description="Skills found on disk that had no DB row; now owned by the caller")
already_registered: list[str] = Field(default_factory=list, description="Skills found on disk that already had a DB row (unchanged)")
class CustomSkillContentResponse(SkillResponse):
content: str = Field(..., description="Raw SKILL.md content")
@@ -483,6 +490,54 @@ async def install_skill(
raise HTTPException(status_code=500, detail=f"Failed to install skill: {str(e)}")
@router.post(
"/skills/reconcile-custom",
response_model=SkillReconcileResponse,
summary="Adopt Orphan Custom Skills",
description=(
"Scan the on-disk ``skills/custom/`` directory and register any skill that has a "
"valid SKILL.md but no DB ownership row. Adopted skills are recorded as owned by "
"the calling user with ``published=false``. Skips skills that already have a row "
"(regardless of owner). Useful after manually unzipping skills onto the filesystem."
),
)
async def reconcile_custom_skills(
request: Request,
config: AppConfig = Depends(get_config),
skill_store: SkillStore = Depends(get_skill_store),
) -> SkillReconcileResponse:
try:
user_id = _current_user_id(request)
storage = get_or_new_skill_storage(app_config=config)
custom_skills = [s for s in storage.load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
adopted: list[str] = []
already: list[str] = []
for skill in custom_skills:
existing = await skill_store.get_any(skill.name)
if existing is not None:
already.append(skill.name)
continue
await skill_store.ensure_legacy(
{"name": skill.name, "owner_user_id": user_id, "published": False}
)
try:
from deerflow.skills.usage import mark_uploaded
mark_uploaded(skill.name)
except Exception:
pass
adopted.append(skill.name)
if adopted:
await refresh_skills_system_prompt_cache_async()
return SkillReconcileResponse(adopted=adopted, already_registered=already)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to reconcile custom skills: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to reconcile custom skills: {e}")
@router.get("/skills/custom", response_model=SkillsListResponse, summary="List Custom Skills")
async def list_custom_skills(
request: Request,
+3 -1
View File
@@ -125,6 +125,8 @@ sandbox:
skills:
path: D:\project\bj\deerflow-server\skills
security_scan:
enabled: false
agents_api:
enabled: true
@@ -147,7 +149,7 @@ checkpointer:
# users/{user_id}/memory_config.json > .deer-flow/memory_user_overrides.json > 此处
memory:
enabled: true
version: v2 # 记忆版本: 'v1'(原版 DeerFlow, memory.json + LLM 自动提取)
version: v1 # 记忆版本: 'v1'(原版 DeerFlow, memory.json + LLM 自动提取)
# 'v2'(USER.md/MEMORY.md + 可选 Hindsight)
provider: hindsight # 'builtin' | 'hindsight'(外部 provider 名)
# builtin 始终启用,这里只决定 external 选哪个
@@ -464,11 +464,22 @@ You are {agent_name}, an internal super agent.
- ❌ DO NOT skip clarification for "efficiency" - accuracy matters more than speed
- ❌ DO NOT make assumptions when information is missing - ALWAYS ask
- ❌ DO NOT proceed with guesses - STOP and call ask_clarification first
- ❌ DO NOT cram several unrelated questions into one `question` string — split them into separate tool calls
- ✅ Analyze the request in thinking → Identify unclear aspects → Ask BEFORE any action
- ✅ If you identify the need for clarification in your thinking, you MUST call the tool IMMEDIATELY
- ✅ After calling ask_clarification, execution will be interrupted automatically
- ✅ Wait for user response - do NOT continue with assumptions
**Batching multiple independent questions:**
- When the request leaves **several independent aspects** unclear (e.g. project type AND tech stack
AND deployment target), call `ask_clarification` multiple times in the same turn — one call per
aspect, each with its own focused `question` and `options`.
- The frontend groups all clarification cards from one turn into a single submission, so the user
answers them together in one reply. You will then receive ONE combined user message addressing
every question in order — read it carefully and match answers back to the questions you asked.
- If the next question only makes sense after the previous answer (sequential dependency), ask
only the first one this turn and wait for the reply before asking the next.
**How to Use:**
```python
ask_clarification(
@@ -288,7 +288,7 @@ def build_memory_manager(
manager.initialize_all(
user_id=user_id,
agent_id=agent_id or "default",
agent_id=agent_id,
thread_id=thread_id,
base_dir=base_dir,
)
@@ -171,21 +171,21 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
def _build_retry_message(self, attempt: int, wait_ms: int, reason: str) -> str:
seconds = max(1, round(wait_ms / 1000))
reason_text = "provider is busy" if reason == "busy" else "provider request failed temporarily"
return f"LLM request retry {attempt}/{self.retry_max_attempts}: {reason_text}. Retrying in {seconds}s."
reason_text = "服务繁忙" if reason == "busy" else "网络/连接异常"
return f"大模型因{reason_text}调用失败,正在第 {attempt}/{self.retry_max_attempts} 次重试,{seconds} 秒后继续…"
def _build_circuit_breaker_message(self) -> str:
return "The configured LLM provider is currently unavailable due to continuous failures. Circuit breaker is engaged to protect the system. Please wait a moment before trying again."
return "大模型因连续多次调用失败已被熔断保护,请稍候再试。"
def _build_user_message(self, exc: BaseException, reason: str) -> str:
detail = _extract_error_detail(exc)
if reason == "quota":
return "The configured LLM provider rejected the request because the account is out of quota, billing is unavailable, or usage is restricted. Please fix the provider account and try again."
return "大模型因账户额度不足、账单异常或使用受限而拒绝请求,请检查模型账户后再试。"
if reason == "auth":
return "The configured LLM provider rejected the request because authentication or access is invalid. Please check the provider credentials and try again."
return "大模型因鉴权失败或无访问权限而拒绝请求,请检查模型的 API Key 与权限配置后再试。"
if reason in {"busy", "transient"}:
return "The configured LLM provider is temporarily unavailable after multiple retries. Please wait a moment and continue the conversation."
return f"LLM request failed: {detail}"
return "大模型因服务繁忙或网络异常,多次重试后仍未恢复,请稍候再继续。"
return f"大模型因「{detail}」原因调用失败。"
def _emit_retry_event(self, attempt: int, wait_ms: int, reason: str) -> None:
try:
@@ -117,15 +117,18 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
except RuntimeError:
return None
def _resolve_agent_id(self, runtime: Runtime | None) -> str:
"""解析 agent_id;优先 middleware 构造时的 agent_name,再 runtime context。"""
def _resolve_agent_id(self, runtime: Runtime | None) -> str | None:
"""解析 agent_id;优先 middleware 构造时的 agent_name,再 runtime context。
无 agent 上下文时返回 None,使 V1 记忆存入全局文件(users/{id}/memory.json)。
"""
if self._agent_name:
return self._agent_name
if runtime is not None and getattr(runtime, "context", None):
agent_id = runtime.context.get("agent_id")
if agent_id:
return str(agent_id)
return "default"
return None
# ---- wrap_model_call:Hindsight 动态召回注入 -------------------------
@@ -105,8 +105,16 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
return prompt, user_msg
def _strip_think_tags(self, text: str) -> str:
"""Remove <think>...</think> blocks emitted by reasoning models (e.g. minimax, DeepSeek-R1)."""
return re.sub(r"<think>[\s\S]*?</think>", "", text, flags=re.IGNORECASE).strip()
"""Remove <think>...</think> blocks emitted by reasoning models (e.g. minimax, GLM, DeepSeek-R1).
Also handles malformed cases: an unclosed `<think>` (response truncated mid-reasoning)
or a stray `</think>` (open tag eaten by upstream parsing) — both can otherwise leak
raw reasoning into the title.
"""
text = re.sub(r"<think>[\s\S]*?</think>", "", text, flags=re.IGNORECASE)
text = re.sub(r"<think>[\s\S]*$", "", text, flags=re.IGNORECASE)
text = re.sub(r"^[\s\S]*?</think>", "", text, flags=re.IGNORECASE)
return text.strip()
def _parse_title(self, content: object) -> str:
"""Normalize model output into a clean title string."""
@@ -146,6 +154,36 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
_, user_msg = self._build_title_prompt(state)
return {"title": self._fallback_title(user_msg)}
def _force_disable_thinking(self, model: object) -> None:
"""Force-disable thinking on the title model via `extra_body` overrides.
Title generation never needs reasoning. The model factory only emits
disable params when the model config declares `when_thinking_disabled`
or a matching `when_thinking_enabled` shape — many user-added reasoning
models (e.g. GLM-4.5/5) only set `supports_thinking: true` and would
otherwise keep thinking on, which is slow and leaks <think> tags into
the title prompt response. Injecting both common OpenAI-compatible
disable shapes covers Zhipu GLM (`thinking.type`) and Qwen/vLLM
(`chat_template_kwargs.enable_thinking`); other providers ignore
unknown body fields.
"""
if not hasattr(model, "extra_body"):
return
current = getattr(model, "extra_body", None) or {}
merged_chat_template_kwargs = {
**(current.get("chat_template_kwargs") or {}),
"enable_thinking": False,
}
merged = {
**current,
"thinking": {**(current.get("thinking") or {}), "type": "disabled"},
"chat_template_kwargs": merged_chat_template_kwargs,
}
try:
model.extra_body = merged
except Exception:
logger.debug("Could not override extra_body to disable thinking on title model", exc_info=True)
async def _agenerate_title_result(self, state: TitleMiddlewareState) -> dict | None:
"""Generate a title asynchronously and fall back locally on failure."""
if not self._should_generate_title(state):
@@ -162,6 +200,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
model = create_chat_model(name=config.model_name, **model_kwargs)
else:
model = create_chat_model(**model_kwargs)
self._force_disable_thinking(model)
response = await model.ainvoke(prompt, config=self._get_runnable_config())
title = self._parse_title(response.content)
if title:
@@ -6,6 +6,18 @@ from pydantic import BaseModel, Field
from deerflow.config.runtime_paths import project_root, resolve_path
class SecurityScanConfig(BaseModel):
"""Toggle for the LLM-based skill security scanner."""
enabled: bool = Field(
default=True,
description=(
"When false, skip the LLM security scan applied to skill installs, "
"custom-skill edits, and rollbacks. The scan is then treated as 'allow'."
),
)
class SkillsConfig(BaseModel):
"""Configuration for skills system"""
@@ -21,6 +33,10 @@ class SkillsConfig(BaseModel):
default="/mnt/skills",
description="Path where skills are mounted in the sandbox container",
)
security_scan: SecurityScanConfig = Field(
default_factory=SecurityScanConfig,
description="LLM security-scan toggle for skill content.",
)
def get_skills_path(self) -> Path:
"""
@@ -713,7 +713,7 @@ class ScheduledTaskService:
retry_count: int = 0,
) -> None:
"""Mark a run failed and reflect the outcome on the parent task."""
result = _build_result(task, scheduled_for=scheduled_for, content=f"Scheduled task failed: {error}")
result = _build_result(task, scheduled_for=scheduled_for, content=f"定时任务执行失败:{error}")
await self.store.update_run(
run_id,
{"status": "failed", "finished_at": datetime.now(UTC), "error": error, "result": result, "retry_count": retry_count},
@@ -39,6 +39,12 @@ def _extract_json_object(raw: str) -> dict | None:
async def scan_skill_content(content: str, *, executable: bool = False, location: str = SKILL_MD_FILE, app_config: AppConfig | None = None) -> ScanResult:
"""Screen skill content before it is written to disk."""
config = app_config or get_app_config()
# Operator-level kill switch: when ``skills.security_scan.enabled`` is false,
# bypass the LLM round-trip entirely and let the content through.
if not config.skills.security_scan.enabled:
return ScanResult("allow", "Security scan disabled by config.")
rubric = (
"You are a security reviewer for AI agent skills. "
"Classify the content as allow, warn, or block. "
@@ -49,7 +55,6 @@ async def scan_skill_content(content: str, *, executable: bool = False, location
prompt = f"Location: {location}\nExecutable: {str(executable).lower()}\n\nReview this content:\n-----\n{content}\n-----"
try:
config = app_config or get_app_config()
# 安全审核复用 curator 的评审模型配置(skill_evolution.curator.model_name):
# 二者用同一个模型。未配置时 model_name 为 None,create_chat_model 回退主模型。
model_name = config.skill_evolution.curator.model_name
@@ -36,18 +36,31 @@ def ask_clarification_tool(
- You're about to perform a potentially dangerous operation
- You have a recommendation but need user approval
Batching multiple independent questions:
- **You CAN and SHOULD call ask_clarification multiple times in the same turn** when the user's
request leaves several **independent** aspects unclear (e.g. project type AND tech stack AND
deployment target). Emit one tool call per aspect — do NOT cram several unrelated questions
into one `question` string.
- The frontend automatically groups all clarification cards in one turn into a single submission,
so the user answers them together in one reply. You will then receive one combined user message
addressing every question in order.
- When the questions DEPEND on each other (the second only makes sense after the first is answered),
ask only the first one and wait for the reply before asking the next.
Best practices:
- Ask ONE clarification at a time for clarity
- Be specific and clear in your question
- Don't make assumptions when clarification is needed
- For risky operations, ALWAYS ask for confirmation
- After calling this tool, execution will be interrupted automatically
- One question per ask_clarification call — keep each `question` focused on a single decision.
- Be specific and clear in your question.
- Provide `options` (with concise labels) whenever there is a discrete set of reasonable choices —
this lets the user multi-select in the UI instead of typing free text.
- Don't make assumptions when clarification is needed.
- For risky operations, ALWAYS ask for confirmation.
- After calling this tool (or several in one turn), execution will be interrupted automatically.
Args:
question: The clarification question to ask the user. Be specific and clear.
question: The clarification question to ask the user. Focused on a single decision; be specific and clear.
clarification_type: The type of clarification needed (missing_info, ambiguous_requirement, approach_choice, risk_confirmation, suggestion).
context: Optional context explaining why clarification is needed. Helps the user understand the situation.
options: Optional list of choices (for approach_choice or suggestion types). Present clear options for the user to choose from.
options: Optional list of choices (for approach_choice or suggestion types). Present clear options for the user to choose from; the UI supports multi-select.
"""
# This is a placeholder implementation
# The actual logic is handled by ClarificationMiddleware which intercepts this tool call
@@ -0,0 +1,138 @@
"""Tests for the ``POST /api/skills/reconcile-custom`` adoption flow.
Covers the scenario where a user drops a skill directory onto disk by hand
(bypassing the upload API) and then calls the reconcile route so it shows up
in the UI.
"""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import pytest
from app.gateway.routers import skills as skills_router
from deerflow.persistence.skills import MemorySkillStore
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
def _write_skill_on_disk(root: Path, name: str, description: str = "from disk") -> None:
"""Drop a valid SKILL.md into ``<root>/custom/<name>/`` like a manual unzip would."""
skill_dir = root / "custom" / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: {description}\n---\n# {name}\n",
encoding="utf-8",
)
def _fake_request(user_id: str | None = "user-a") -> SimpleNamespace:
user = SimpleNamespace(id=user_id, system_role="user") if user_id else None
return SimpleNamespace(state=SimpleNamespace(user=user))
@pytest.fixture
def skills_root(tmp_path: Path) -> Path:
root = tmp_path / "skills"
(root / "public").mkdir(parents=True)
(root / "custom").mkdir(parents=True)
return root
@pytest.fixture
def patched_storage(
skills_root: Path, monkeypatch: pytest.MonkeyPatch
) -> LocalSkillStorage:
"""Force the router to use a LocalSkillStorage rooted at ``skills_root``."""
storage = LocalSkillStorage(host_path=str(skills_root))
monkeypatch.setattr(
skills_router, "get_or_new_skill_storage", lambda **_: storage
)
async def _noop_refresh() -> None:
return None
monkeypatch.setattr(
skills_router, "refresh_skills_system_prompt_cache_async", _noop_refresh
)
return storage
@pytest.mark.asyncio
async def test_reconcile_adopts_orphan_custom_skill(
skills_root: Path, patched_storage: LocalSkillStorage
):
_write_skill_on_disk(skills_root, "manually-dropped")
store = MemorySkillStore()
result = await skills_router.reconcile_custom_skills(
request=_fake_request("user-a"), config=None, skill_store=store
)
assert result.adopted == ["manually-dropped"]
assert result.already_registered == []
record = await store.get_any("manually-dropped")
assert record is not None
assert record["owner_user_id"] == "user-a"
assert record["published"] is False
@pytest.mark.asyncio
async def test_reconcile_skips_already_registered(
skills_root: Path, patched_storage: LocalSkillStorage
):
_write_skill_on_disk(skills_root, "already-known")
store = MemorySkillStore()
# Pre-existing ownership row — should be left alone.
await store.create(
{"name": "already-known", "owner_user_id": "user-b", "published": True}
)
result = await skills_router.reconcile_custom_skills(
request=_fake_request("user-a"), config=None, skill_store=store
)
assert result.adopted == []
assert result.already_registered == ["already-known"]
# Ownership must not be overwritten.
record = await store.get_any("already-known")
assert record is not None
assert record["owner_user_id"] == "user-b"
assert record["published"] is True
@pytest.mark.asyncio
async def test_reconcile_mixed_adopts_only_orphans(
skills_root: Path, patched_storage: LocalSkillStorage
):
_write_skill_on_disk(skills_root, "orphan-one")
_write_skill_on_disk(skills_root, "orphan-two")
_write_skill_on_disk(skills_root, "registered")
store = MemorySkillStore()
await store.create(
{"name": "registered", "owner_user_id": "user-a", "published": False}
)
result = await skills_router.reconcile_custom_skills(
request=_fake_request("user-a"), config=None, skill_store=store
)
assert sorted(result.adopted) == ["orphan-one", "orphan-two"]
assert result.already_registered == ["registered"]
@pytest.mark.asyncio
async def test_reconcile_with_empty_custom_dir_is_noop(
skills_root: Path, patched_storage: LocalSkillStorage
):
store = MemorySkillStore()
result = await skills_router.reconcile_custom_skills(
request=_fake_request("user-a"), config=None, skill_store=store
)
assert result.adopted == []
assert result.already_registered == []
+3
View File
@@ -109,6 +109,8 @@ title:
Assistant: {assistant_msg}
skills:
path: /app/skills
security_scan:
enabled: false
agents_api:
enabled: true
database:
@@ -138,6 +140,7 @@ ai_writing:
# users/{user_id}/memory_config.json > .deer-flow/memory_user_overrides.json > 此处
memory:
enabled: true
version: v1
provider: hindsight # 'builtin' | 'hindsight'(外部 provider 名)
# builtin 始终启用,这里只决定 external 选哪个
builtin: