fix: align deerflow runner with deerflow 2.0 (#7500)

zouyonghe · web-flow · commit 533a0bde6aec · 2026-04-13T12:47:27.000+09:00
* fix: align deerflow runner with deerflow 2.0

* fix: address deerflow review feedback
diff --git a/astrbot/builtin_stars/builtin_commands/commands/conversation.py b/astrbot/builtin_stars/builtin_commands/commands/conversation.py
@@ -1,9 +1,12 @@
 from astrbot.api import sp, star
 from astrbot.api.event import AstrMessageEvent, MessageEventResult
+from astrbot.core import logger
 from astrbot.core.agent.runners.deerflow.constants import (
+    DEERFLOW_AGENT_RUNNER_PROVIDER_ID_KEY,
     DEERFLOW_PROVIDER_TYPE,
     DEERFLOW_THREAD_ID_KEY,
 )
+from astrbot.core.agent.runners.deerflow.deerflow_api_client import DeerFlowAPIClient
 from astrbot.core.utils.active_event_registry import active_event_registry
 
 from .utils.rst_scene import RstScene
@@ -17,6 +20,85 @@
 THIRD_PARTY_AGENT_RUNNER_STR = ", ".join(THIRD_PARTY_AGENT_RUNNER_KEY.keys())
 
 
+async def _cleanup_deerflow_thread_if_present(
+    context: star.Context,
+    umo: str,
+) -> None:
+    try:
+        thread_id = await sp.get_async(
+            scope="umo",
+            scope_id=umo,
+            key=DEERFLOW_THREAD_ID_KEY,
+            default="",
+        )
+        if not thread_id:
+            return
+
+        cfg = context.get_config(umo=umo)
+        provider_id = cfg["provider_settings"].get(
+            DEERFLOW_AGENT_RUNNER_PROVIDER_ID_KEY,
+            "",
+        )
+        if not provider_id:
+            return
+
+        merged_provider_config = context.provider_manager.get_provider_config_by_id(
+            provider_id,
+            merged=True,
+        )
+        if not merged_provider_config:
+            logger.warning(
+                "Failed to resolve DeerFlow provider config for remote thread cleanup: provider_id=%s",
+                provider_id,
+            )
+            return
+
+        client = DeerFlowAPIClient(
+            api_base=merged_provider_config.get(
+                "deerflow_api_base",
+                "http://127.0.0.1:2026",
+            ),
+            api_key=merged_provider_config.get("deerflow_api_key", ""),
+            auth_header=merged_provider_config.get("deerflow_auth_header", ""),
+            proxy=merged_provider_config.get("proxy", ""),
+        )
+        try:
+            await client.delete_thread(thread_id)
+        finally:
+            try:
+                await client.close()
+            except Exception as e:
+                logger.warning(
+                    "Failed to close DeerFlow API client after thread cleanup: %s",
+                    e,
+                )
+    except Exception as e:
+        logger.warning(
+            "Failed to clean up DeerFlow thread for session %s: %s",
+            umo,
+            e,
+        )
+
+
+async def _clear_third_party_agent_runner_state(
+    context: star.Context,
+    umo: str,
+    agent_runner_type: str,
+) -> None:
+    session_key = THIRD_PARTY_AGENT_RUNNER_KEY.get(agent_runner_type)
+    if not session_key:
+        return
+
+    if agent_runner_type == DEERFLOW_PROVIDER_TYPE:
+        await _cleanup_deerflow_thread_if_present(context, umo)
+
+    await sp.remove_async(
+        scope="umo",
+        scope_id=umo,
+        key=session_key,
+    )
+
+
 class ConversationCommands:
     def __init__(self, context: star.Context) -> None:
         self.context = context
@@ -65,10 +147,10 @@ async def reset(self, message: AstrMessageEvent) -> None:
         agent_runner_type = cfg["provider_settings"]["agent_runner_type"]
         if agent_runner_type in THIRD_PARTY_AGENT_RUNNER_KEY:
             active_event_registry.stop_all(umo, exclude=message)
-            await sp.remove_async(
-                scope="umo",
-                scope_id=umo,
-                key=THIRD_PARTY_AGENT_RUNNER_KEY[agent_runner_type],
+            await _clear_third_party_agent_runner_state(
+                self.context,
+                umo,
+                agent_runner_type,
             )
             message.set_result(
                 MessageEventResult().message("✅ Conversation reset successfully.")
@@ -139,10 +221,10 @@ async def new_conv(self, message: AstrMessageEvent) -> None:
         agent_runner_type = cfg["provider_settings"]["agent_runner_type"]
         if agent_runner_type in THIRD_PARTY_AGENT_RUNNER_KEY:
             active_event_registry.stop_all(message.unified_msg_origin, exclude=message)
-            await sp.remove_async(
-                scope="umo",
-                scope_id=message.unified_msg_origin,
-                key=THIRD_PARTY_AGENT_RUNNER_KEY[agent_runner_type],
+            await _clear_third_party_agent_runner_state(
+                self.context,
+                message.unified_msg_origin,
+                agent_runner_type,
             )
             message.set_result(
                 MessageEventResult().message("✅ New conversation created.")
diff --git a/astrbot/core/agent/runners/deerflow/deerflow_agent_runner.py b/astrbot/core/agent/runners/deerflow/deerflow_agent_runner.py
@@ -410,18 +410,20 @@ def _build_messages(
         )
         return messages
 
-    def _build_runtime_context(self, thread_id: str) -> dict[str, T.Any]:
-        runtime_context: dict[str, T.Any] = {
+    def _build_runtime_configurable(self, thread_id: str) -> dict[str, T.Any]:
+        runtime_configurable: dict[str, T.Any] = {
             "thread_id": thread_id,
             "thinking_enabled": self.thinking_enabled,
             "is_plan_mode": self.plan_mode,
             "subagent_enabled": self.subagent_enabled,
         }
         if self.subagent_enabled:
-            runtime_context["max_concurrent_subagents"] = self.max_concurrent_subagents
+            runtime_configurable["max_concurrent_subagents"] = (
+                self.max_concurrent_subagents
+            )
         if self.model_name:
-            runtime_context["model_name"] = self.model_name
-        return runtime_context
+            runtime_configurable["model_name"] = self.model_name
+        return runtime_configurable
 
     def _build_payload(
         self,
@@ -430,16 +432,19 @@ def _build_payload(
         image_urls: list[str],
         system_prompt: str | None,
     ) -> dict[str, T.Any]:
+        runtime_configurable = self._build_runtime_configurable(thread_id)
         return {
             "assistant_id": self.assistant_id,
             "input": {
                 "messages": self._build_messages(prompt, image_urls, system_prompt),
             },
             "stream_mode": ["values", "messages-tuple", "custom"],
-            # LangGraph 0.6+ prefers context instead of configurable.
-            "context": self._build_runtime_context(thread_id),
+            # DeerFlow 2.0 consumes runtime overrides from config.configurable.
+            # Keep the legacy context mirror for older compat paths.
+            "context": dict(runtime_configurable),
             "config": {
                 "recursion_limit": self.recursion_limit,
+                "configurable": runtime_configurable,
             },
         }
 
diff --git a/astrbot/core/agent/runners/deerflow/deerflow_api_client.py b/astrbot/core/agent/runners/deerflow/deerflow_api_client.py
@@ -10,6 +10,33 @@
 SSE_MAX_BUFFER_CHARS = 1_048_576
 
 
+class DeerFlowAPIError(Exception):
+    def __init__(
+        self,
+        *,
+        operation: str,
+        status: int,
+        body: str,
+        url: str,
+        thread_id: str | None = None,
+    ) -> None:
+        self.operation = operation
+        self.status = status
+        self.body = body
+        self.url = url
+        self.thread_id = thread_id
+
+        message = (
+            f"DeerFlow {operation} failed: status={status}, url={url}, body={body}"
+        )
+        if thread_id is not None:
+            message = (
+                f"DeerFlow {operation} failed: thread_id={thread_id}, "
+                f"status={status}, url={url}, body={body}"
+            )
+        super().__init__(message)
+
+
 def _normalize_sse_newlines(text: str) -> str:
     """Normalize CRLF/CR to LF so SSE block splitting works reliably."""
     return text.replace("\r\n", "\n").replace("\r", "\n")
@@ -152,11 +179,33 @@ async def create_thread(self, timeout: float = 20) -> dict[str, Any]:
         ) as resp:
             if resp.status not in (200, 201):
                 text = await resp.text()
-                raise Exception(
-                    f"DeerFlow create thread failed: {resp.status}. {text}",
+                raise DeerFlowAPIError(
+                    operation="create thread",
+                    status=resp.status,
+                    body=text,
+                    url=url,
                 )
             return await resp.json()
 
+    async def delete_thread(self, thread_id: str, timeout: float = 20) -> None:
+        session = self._get_session()
+        url = f"{self.api_base}/api/threads/{thread_id}"
+        async with session.delete(
+            url,
+            headers=self.headers,
+            timeout=timeout,
+            proxy=self.proxy,
+        ) as resp:
+            if resp.status not in (200, 202, 204, 404):
+                text = await resp.text()
+                raise DeerFlowAPIError(
+                    operation="delete thread",
+                    status=resp.status,
+                    body=text,
+                    url=url,
+                    thread_id=thread_id,
+                )
+
     async def stream_run(
         self,
         thread_id: str,
@@ -200,8 +249,12 @@ async def stream_run(
         ) as resp:
             if resp.status != 200:
                 text = await resp.text()
-                raise Exception(
-                    f"DeerFlow runs/stream request failed: {resp.status}. {text}",
+                raise DeerFlowAPIError(
+                    operation="runs/stream request",
+                    status=resp.status,
+                    body=text,
+                    url=url,
+                    thread_id=thread_id,
                 )
             async for event in _stream_sse(resp):
                 yield event
diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py
@@ -2671,12 +2671,12 @@ class ChatProviderTemplate(TypedDict):
                     "deerflow_assistant_id": {
                         "description": "Assistant ID",
                         "type": "string",
-                        "hint": "LangGraph assistant_id，默认为 lead_agent。",
+                        "hint": "DeerFlow 2.0 LangGraph assistant_id，默认为 lead_agent。",
                     },
                     "deerflow_model_name": {
                         "description": "模型名称覆盖",
                         "type": "string",
-                        "hint": "可选。覆盖 DeerFlow 默认模型（对应 runtime context 的 model_name）。",
+                        "hint": "可选。覆盖 DeerFlow 默认模型（对应运行时 configurable 的 model_name）。",
                     },
                     "deerflow_thinking_enabled": {
                         "description": "启用思考模式",
@@ -2685,17 +2685,17 @@ class ChatProviderTemplate(TypedDict):
                     "deerflow_plan_mode": {
                         "description": "启用计划模式",
                         "type": "bool",
-                        "hint": "对应 DeerFlow 的 is_plan_mode。",
+                        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 is_plan_mode。",
                     },
                     "deerflow_subagent_enabled": {
                         "description": "启用子智能体",
                         "type": "bool",
-                        "hint": "对应 DeerFlow 的 subagent_enabled。",
+                        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 subagent_enabled。",
                     },
                     "deerflow_max_concurrent_subagents": {
                         "description": "子智能体最大并发数",
                         "type": "int",
-                        "hint": "对应 DeerFlow 的 max_concurrent_subagents。仅在启用子智能体时生效，默认 3。",
+                        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 max_concurrent_subagents。仅在启用子智能体时生效，默认 3。",
                     },
                     "deerflow_recursion_limit": {
                         "description": "递归深度上限",
diff --git a/astrbot/core/provider/manager.py b/astrbot/core/provider/manager.py
@@ -505,6 +505,26 @@ def get_merged_provider_config(self, provider_config: dict) -> dict:
                 pc = merged_config
         return pc
 
+    def get_provider_config_by_id(
+        self,
+        provider_id: str,
+        *,
+        merged: bool = False,
+    ) -> dict | None:
+        """Get a provider config by id.
+
+        Args:
+            provider_id: Provider id to resolve.
+            merged: Whether to merge provider_source config into the provider config.
+        """
+        for provider_config in self.providers_config:
+            if provider_config.get("id") != provider_id:
+                continue
+            if merged:
+                return self.get_merged_provider_config(provider_config)
+            return copy.deepcopy(provider_config)
+        return None
+
     def _resolve_env_key_list(self, provider_config: dict) -> dict:
         keys = provider_config.get("key", [])
         if not isinstance(keys, list):
diff --git a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
@@ -1604,26 +1604,26 @@
       },
       "deerflow_assistant_id": {
         "description": "Assistant ID",
-        "hint": "LangGraph assistant_id，默认为 lead_agent。"
+        "hint": "DeerFlow 2.0 LangGraph assistant_id，默认为 lead_agent。"
       },
       "deerflow_model_name": {
         "description": "模型名称覆盖",
-        "hint": "可选。覆盖 DeerFlow 默认模型（对应 runtime context 的 model_name）。"
+        "hint": "可选。覆盖 DeerFlow 默认模型（对应运行时 configurable 的 model_name）。"
       },
       "deerflow_thinking_enabled": {
         "description": "启用思考模式"
       },
       "deerflow_plan_mode": {
         "description": "启用计划模式",
-        "hint": "对应 DeerFlow 的 is_plan_mode。"
+        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 is_plan_mode。"
       },
       "deerflow_subagent_enabled": {
         "description": "启用子智能体",
-        "hint": "对应 DeerFlow 的 subagent_enabled。"
+        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 subagent_enabled。"
       },
       "deerflow_max_concurrent_subagents": {
         "description": "子智能体最大并发数",
-        "hint": "对应 DeerFlow 的 max_concurrent_subagents。仅在启用子智能体时生效，默认 3。"
+        "hint": "对应 DeerFlow 2.0 运行时 configurable 的 max_concurrent_subagents。仅在启用子智能体时生效，默认 3。"
       },
       "deerflow_recursion_limit": {
         "description": "递归深度上限",
diff --git a/docs/zh/providers/agent-runners/deerflow.md b/docs/zh/providers/agent-runners/deerflow.md
@@ -2,6 +2,8 @@
 
 在 v4.19.2 及之后，AstrBot 支持接入 [DeerFlow](https://github.com/bytedance/deer-flow) Agent Runner。
 
+当前适配面向 DeerFlow **2.0 `main` 分支**。DeerFlow 官方已将原始 Deep Research 框架迁移到 `main-1.x` 分支持续维护，因此如果你使用的是 2.0，请以 `main` 分支文档和后端 API 为准。
+
 ## 预备工作：部署 DeerFlow
 
 如果你还没有部署 DeerFlow，请先参考 DeerFlow 官方文档完成安装和启动：
@@ -25,19 +27,20 @@
 - `API Base URL`：DeerFlow API 网关地址，默认为 `http://127.0.0.1:2026`
 - `DeerFlow API Key`：可选。若你的 DeerFlow 网关使用 Bearer 鉴权，可在此填写
 - `Authorization Header`：可选。自定义 Authorization 请求头，优先级高于 `DeerFlow API Key`
-- `Assistant ID`：对应 LangGraph 的 `assistant_id`，默认为 `lead_agent`
+- `Assistant ID`：对应 DeerFlow 2.0 LangGraph 的 `assistant_id`，默认为 `lead_agent`
 - `模型名称覆盖`：可选。覆盖 DeerFlow 默认模型
 - `启用思考模式`：是否启用 DeerFlow 的思考模式
-- `启用计划模式`：对应 DeerFlow 的 `is_plan_mode`
-- `启用子智能体`：对应 DeerFlow 的 `subagent_enabled`
-- `子智能体最大并发数`：对应 `max_concurrent_subagents`，仅在启用子智能体时生效，默认 `3`
+- `启用计划模式`：对应 DeerFlow 2.0 运行时 `config.configurable.is_plan_mode`
+- `启用子智能体`：对应 DeerFlow 2.0 运行时 `config.configurable.subagent_enabled`
+- `子智能体最大并发数`：对应 DeerFlow 2.0 运行时 `config.configurable.max_concurrent_subagents`，仅在启用子智能体时生效，默认 `3`
 - `递归深度上限`：对应 LangGraph 的 `recursion_limit`，默认 `1000`
 
 填写完成后点击「保存」。
 
 > [!TIP]
 > - 如果 DeerFlow 侧已经配置了默认模型，可以将 `模型名称覆盖` 留空。
 > - 只有在 DeerFlow 侧已经启用了相应能力时，才建议开启 `计划模式` 或 `子智能体` 相关选项。
+> - AstrBot 会同时发送 DeerFlow 2.0 推荐的 `config.configurable` 运行时参数，并保留兼容字段，便于对接上游近期版本。
 
 ## 选择 Agent 执行器
 
@@ -51,3 +54,4 @@
 - `API Base URL` 是否能从 AstrBot 所在环境访问
 - 鉴权配置是否填写正确
 - `Assistant ID` 是否与 DeerFlow 中实际可用的 assistant 一致
+- 如果通过 `/reset`、`/new`、`/del` 重置 DeerFlow 会话，AstrBot 会尝试同步清理 DeerFlow 远端 thread；若 DeerFlow 网关不可达，则只会清理 AstrBot 本地会话标识
diff --git a/tests/test_conversation_commands.py b/tests/test_conversation_commands.py
diff --git a/tests/test_deerflow_agent_runner.py b/tests/test_deerflow_agent_runner.py
diff --git a/tests/test_deerflow_api_client.py b/tests/test_deerflow_api_client.py
