问答模式智能路由智能体初始版

docs/superpowers/specs/2026-05-20-router-subagent-design.md

@@ -1,5 +1,13 @@
 # 问答模式路由智能体设计方案
 
+> **状态**：部分实现 ✓
+> - ✅ `subagent_mode` 注入机制已实现（`build_run_config`、`_CONTEXT_CONFIGURABLE_KEYS`）
+> - ✅ DeerFlow 运行时 `subagent_mode` 加载已实现（`make_lead_agent`）
+> - ✅ `router_quality_tool.py` 已创建
+> - ✅ 子 Agent 目录结构已创建（7个子 Agent 的 SOUL.md 和 config.yaml）
+> - ✅ 路由器 SOUL.md 已更新
+> - ⏳ 集成测试和质检提示词测试待完成
+
 ## 1. 背景与目标
 
 路由器智能体（`f0e75f44f7594053ba1567ed89258f35`）根据用户问题判断最合适的问答模式（fast/rag/bigMode/moreModes/taskChat/hxchat/deerFlow），选择后需要调用相应子 Agent 执行，路由器对执行结果进行**完整性质检**，通过则流式输出到前端，不通过则重新匹配模式并再次执行。
@@ -148,41 +156,40 @@ POST /api/threads/{thread_id}/runs/stream
 
 ## 6. 关键代码改动
 
-### 6.1 DeerFlow 运行时改动
+### 6.1 DeerFlow 运行时改动 ✓
 
 **文件**：`packages/harness/deerflow/agents/lead_agent/agent.py`
 
-改动点：
-- `make_lead_agent` 读取 `configurable.subagent_mode`
+已实现：
+- `make_lead_agent` 读取 `configurable.subagent_mode`（第362-394行）
 - 如果存在，加载 `.deer-flow/agents/{agent_id}/subagents/{subagent_mode}/SOUL.md` 和对应 `config.yaml`
 - 工具绑定使用子 Agent 的 tool_groups，而非父 Agent 的
 
-### 6.2 build_run_config 改动
+### 6.2 build_run_config 改动 ✓
 
 **文件**：`app/gateway/services.py`
 
-改动点：
-- `_CONTEXT_CONFIGURABLE_KEYS` 新增 `subagent_mode`
+已实现：
+- `_CONTEXT_CONFIGURABLE_KEYS` 已包含 `subagent_mode`（第149-171行）
 - `merge_run_context_overrides` 将 `subagent_mode` 透传到 configurable
 
-### 6.3 路由器 SOUL.md 改动
+### 6.3 路由器 SOUL.md 改动 ✓
 
-路由器 SOUL.md 新增功能描述：
+**文件**：`.deer-flow/agents/f0e75f44f7594053ba1567ed89258f35/SOUL.md`
+
+已更新：
 - 角色扩展：增加"在确定模式后，调用相应子 Agent 执行并对结果进行质检"的职责
-- 工具描述：新增一个质检工具（路由器自己调用 LLM 执行质检）
+- 工具描述：已声明 router_quality_check 工具
 
-### 6.4 新增路由器质检工具
+### 6.4 新增路由器质检工具 ✓
 
-**文件**：`packages/harness/deerflow/tools/builtins/router_quality_tool.py`（新建）
+**文件**：`packages/harness/deerflow/tools/builtins/router_quality_tool.py`
 
+已实现：
 ```python
-@tool
+@tool("router_quality_check", parse_docstring=True, return_direct=True)
 def router_quality_check(original_question: str, agent_output: str) -> str:
-    """质检工具 — 判断子 Agent 输出是否完整覆盖用户问题
-
-    返回 JSON：{"passed": bool, "reason": str, "missing_parts": list[str]}
-    """
-    # 调用 LLM 执行质检逻辑
+    """质检工具 — 判断子 Agent 输出是否完整覆盖用户问题"""
 ```
 
 ## 7. 数据流

offline-backend-20260512/backend/app/gateway/deps.py

@@ -274,6 +274,7 @@ def get_run_context(request: Request) -> RunContext:
         run_events_config=getattr(config, "run_events", None),
         thread_store=get_thread_store(request),
         app_config=config,
+        stream_bridge=get_stream_bridge(request),
     )
 
 

offline-backend-20260512/backend/config.yaml

@@ -139,6 +139,168 @@ checkpointer:
   type: sqlite
   connection_string: D:\cursor-work\offline-backend-20260512\backend\.deer-flow\data\deerflow.db
 
+subagents:
+  enabled: true
+  custom_agents:
+    fast:
+      description: 快速问答子Agent - 处理简单事实查询、one-hop问答
+      system_prompt: |
+        # 快速问答子Agent
+
+        ## 角色定义
+        你是一个快速问答专家。专门处理简单事实查询、one-hop 问答场景。
+
+        ## 能力范围
+        - 简单事实查询（天气、地理位置、基本概念等）
+        - 单次检索即可回答的问题
+        - 无需引用和深度检索的问答
+
+        ## 行为规范
+        1. 简洁明了，直接回答用户问题
+        2. 如果问题复杂或需要引用，及时返回说明
+      tools: [web_search, read_file]
+      model: inherit
+      max_turns: 10
+      timeout_seconds: 60
+
+    rag:
+      description: 知识库问答子Agent - 从知识库检索并返回引用
+      system_prompt: |
+        # 知识库问答子Agent
+
+        ## 角色定义
+        你是一个知识库问答专家。专门处理需要从知识库检索、需要返回文档引用的问答场景。
+
+        ## 能力范围
+        - 从知识库检索相关信息
+        - 返回文档引用和来源
+        - 处理结构化知识需求
+
+        ## 行为规范
+        1. 检索知识库获取相关信息
+        2. 返回答案时附带文档引用
+        3. 如未找到相关信息，明确告知用户
+      tools: [web_search, read_file]
+      model: inherit
+      max_turns: 20
+      timeout_seconds: 300
+
+    bigMode:
+      description: 大模型对话子Agent - 开放性问答、创意生成
+      system_prompt: |
+        # 大模型对话子Agent
+
+        ## 角色定义
+        你是一个大模型对话专家。专门处理开放性问答、创意生成、通用对话场景。
+
+        ## 能力范围
+        - 开放性问答
+        - 创意写作、翻译
+        - 通用对话
+
+        ## 行为规范
+        1. 根据用户需求生成创意内容
+        2. 无需外部检索，直接使用知识回答
+        3. 保持对话简洁明了
+      tools: []
+      model: inherit
+      max_turns: 15
+      timeout_seconds: 180
+
+    moreModes:
+      description: 多轮检索子Agent - 需要搜索引擎、多跳推理、复杂问题分解
+      system_prompt: |
+        # 多轮检索子Agent
+
+        ## 角色定义
+        你是一个多轮检索专家。专门处理需要多次搜索、多跳推理、复杂问题分解的场景。
+
+        ## 能力范围
+        - 连续搜索多个来源
+        - 多跳推理
+        - 复杂问题分解
+
+        ## 行为规范
+        1. 分析问题，分解为多个子问题
+        2. 逐个检索每个子问题
+        3. 汇总各子问题的答案形成完整回答
+        4. 必要时进行多轮搜索
+      tools: [web_search, read_file]
+      model: inherit
+      max_turns: 30
+      timeout_seconds: 600
+
+    taskChat:
+      description: 任务分析子Agent - 任务执行、工作流处理、步骤分解
+      system_prompt: |
+        # 任务分析子Agent
+
+        ## 角色定义
+        你是一个任务分析专家。专门处理任务执行、工作流处理、步骤分解、多工具协作的场景。
+
+        ## 能力范围
+        - 任务分解
+        - 工作流处理
+        - 步骤分解
+        - 多工具协作
+
+        ## 行为规范
+        1. 分析任务，分解为可执行的步骤
+        2. 使用适当工具逐步完成任务
+        3. 定期汇报进度
+        4. 最终输出任务完成结果
+      tools: [web_search, read_file, write_file, bash]
+      model: inherit
+      max_turns: 40
+      timeout_seconds: 900
+
+    hxchat:
+      description: 画像检索子Agent - 人物画像分析、社交媒体用户搜索
+      system_prompt: |
+        # 画像检索子Agent
+
+        ## 角色定义
+        你是一个画像检索专家。专门处理人物画像分析、社交媒体用户搜索、多平台数据汇总的场景。
+
+        ## 能力范围
+        - 人物画像分析
+        - 社交媒体用户搜索
+        - 多平台数据汇总
+
+        ## 行为规范
+        1. 搜索目标人物的公开信息
+        2. 分析和汇总来自不同平台的数据
+        3. 生成人物画像报告
+        4. 注意保护隐私信息
+      tools: [web_search]
+      model: inherit
+      max_turns: 20
+      timeout_seconds: 300
+
+    deerFlow:
+      description: 深度研究子Agent - 复杂研究任务、报告生成、长文本创作
+      system_prompt: |
+        # 深度研究子Agent
+
+        ## 角色定义
+        你是一个深度研究专家。专门处理复杂研究任务、长文本报告生成、多工具深度协作。
+
+        ## 能力范围
+        - 复杂研究任务和多角度分析
+        - 长文本报告生成
+        - 多工具深度协作
+
+        ## 行为规范
+        1. 制定研究计划，明确研究目标和角度
+        2. 进行多轮信息收集和整理
+        3. 深入分析，形成完整研究报告
+        4. 输出结构清晰、引用完备的长文本
+        5. 使用多个工具协作完成复杂任务
+      tools: [web_search, read_file, write_file, bash]
+      model: inherit
+      max_turns: 50
+      timeout_seconds: 1200
+
 # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。
 # 三层覆盖优先级(高->低):
 #   users/{user_id}/memory_config.json > .deer-flow/memory_user_overrides.json > 此处

offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -333,7 +333,6 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
     reasoning_effort = cfg.get("reasoning_effort", None)
     requested_model_name: str | None = cfg.get("model_name") or cfg.get("model")
     is_plan_mode = cfg.get("is_plan_mode", False)
-    subagent_enabled = cfg.get("subagent_enabled", False)
     max_concurrent_subagents = cfg.get("max_concurrent_subagents", 3)
     is_bootstrap = cfg.get("is_bootstrap", False)
     writing_mode = bool(cfg.get("writing_mode", False))
@@ -346,6 +345,8 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
     agent_display_name = agent_config.name if agent_config else agent_id
     # Custom agent model from agent config (if any), or None to let _resolve_model_name pick the default
     agent_model_name = agent_config.model if agent_config and agent_config.model else None
+    # Subagent delegation: runtime config overrides agent config, default to False
+    subagent_enabled = cfg.get("subagent_enabled", agent_config.subagent_enabled if agent_config else False)
 
     # Final model name resolution: request → agent config → global default, with fallback for unknown names
     model_name = _resolve_model_name(requested_model_name or agent_model_name, app_config=resolved_app_config)

offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py

@@ -48,6 +48,8 @@ class AgentConfig(BaseModel):
     # - [] (explicit empty list): no skills
     # - ["skill1", "skill2"]: load only the specified skills
     skills: list[str] | None = None
+    # subagent_enabled allows the agent to delegate tasks to subagents via the task tool
+    subagent_enabled: bool = False
 
 
 def load_agent_config(agent_id: str | None) -> AgentConfig | None:

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -46,6 +46,7 @@ def _build_runtime_context(
     run_id: str,
     caller_context: Any | None,
     app_config: AppConfig | None = None,
+    stream_bridge: Any = None,
 ) -> dict[str, Any]:
     """Build the dict that becomes ``ToolRuntime.context`` for the run.
 
@@ -65,6 +66,8 @@ def _build_runtime_context(
             runtime_ctx.setdefault(key, value)
     if app_config is not None:
         runtime_ctx["app_config"] = app_config
+    if stream_bridge is not None:
+        runtime_ctx["stream_bridge"] = stream_bridge
     return runtime_ctx
 
 
@@ -83,6 +86,7 @@ class RunContext:
     run_events_config: Any | None = field(default=None)
     thread_store: Any | None = field(default=None)
     app_config: AppConfig | None = field(default=None)
+    stream_bridge: Any | None = field(default=None)
 
 
 def _install_runtime_context(config: dict, runtime_context: dict[str, Any]) -> None:
@@ -214,7 +218,7 @@ async def run_agent(
         # access thread-level data. langgraph-cli does this automatically; we must do it
         # manually here because we drive the graph through ``agent.astream(config=...)``
         # without passing the official ``context=`` parameter.
-        runtime_ctx = _build_runtime_context(thread_id, run_id, config.get("context"), ctx.app_config)
+        runtime_ctx = _build_runtime_context(thread_id, run_id, config.get("context"), ctx.app_config, ctx.stream_bridge)
         _install_runtime_context(config, runtime_ctx)
         runtime = Runtime(context=cast(Any, runtime_ctx), store=store)
         config.setdefault("configurable", {})["__pregel_runtime"] = runtime

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/__init__.py

@@ -1,6 +1,8 @@
 from .clarification_tool import ask_clarification_tool
+from .fast_chat_tool import fast_chat_tool, fast_llm_answer
 from .memory_tool import memory_tool
 from .present_file_tool import present_file_tool
+from .router_quality_tool import router_quality_check
 from .scheduled_task_tool import scheduled_task_tool
 from .setup_agent_tool import setup_agent
 from .task_tool import task_tool
@@ -16,4 +18,5 @@ __all__ = [
     "task_tool",
     "fast_chat_tool",
     "fast_llm_answer",
+    "router_quality_check",
 ]

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/fast_chat_tool.py

@@ -92,18 +92,37 @@ async def fast_chat_tool(
 
         req_url = f"{backend_url}/api/v1/new_chat"
 
+        # 从 context 获取 newChatToken
+        new_chat_token = _get_runtime_value(runtime, "newChatToken")
+
+        logger.info(
+            "fast_chat_tool request: thread_id=%s, prompt=%s, search_space_id=%s, model_name=%s, has_token=%s",
+            thread_id,
+            prompt[:100] if prompt else None,
+            effective_search_space_id,
+            effective_model_name,
+            bool(new_chat_token),
+        )
+
         def _make_request():
             data = json.dumps(req_body).encode("utf-8")
+            headers = {"Content-Type": "application/json"}
+
+            # 添加认证 token 到请求头
+            if new_chat_token:
+                headers["Authorization"] = f"Bearer {new_chat_token}"
+
             req = urllib.request.Request(
                 req_url,
                 data=data,
-                headers={"Content-Type": "application/json"},
+                headers=headers,
                 method="POST",
             )
             with urllib.request.urlopen(req, timeout=30) as response:
                 return response.read().decode("utf-8")
 
         result = await asyncio.to_thread(_make_request)
+        logger.info("fast_chat_tool response: thread_id=%s, result=%s", thread_id, result[:500] if result else None)
         return result
 
     except urllib.error.HTTPError as exc:

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/task_tool.py

@@ -37,6 +37,14 @@ def _get_runtime_app_config(runtime: Any) -> "AppConfig | None":
     return None
 
 
+def _get_runtime_stream_bridge(runtime: Any) -> Any:
+    """Get StreamBridge from runtime context for publishing events to SSE."""
+    context = getattr(runtime, "context", None)
+    if isinstance(context, dict):
+        return context.get("stream_bridge")
+    return None
+
+
 def _merge_skill_allowlists(parent: list[str] | None, child: list[str] | None) -> list[str] | None:
     """Return the effective subagent skill allowlist under the parent policy."""
     if parent is None:
@@ -123,6 +131,8 @@ async def task_tool(
     parent_model = None
     trace_id = None
     metadata: dict = {}
+    stream_bridge = None
+    run_id = None
 
     if runtime is not None:
         sandbox_state = runtime.state.get("sandbox")
@@ -138,6 +148,10 @@ async def task_tool(
         # Get or generate trace_id for distributed tracing
         trace_id = metadata.get("trace_id") or str(uuid.uuid4())[:8]
 
+        # Get StreamBridge from runtime context for SSE event publishing
+        stream_bridge = _get_runtime_stream_bridge(runtime)
+        run_id = runtime.context.get("run_id") if runtime.context else None
+
     parent_available_skills = metadata.get("available_skills")
     if parent_available_skills is not None:
         overrides["skills"] = _merge_skill_allowlists(list(parent_available_skills), config.skills)
@@ -188,14 +202,22 @@ async def task_tool(
     poll_count = 0
     last_status = None
     last_message_count = 0  # Track how many AI messages we've already sent
-    # Polling timeout: execution timeout + 60s buffer, checked every 5s
-    max_poll_count = (config.timeout_seconds + 60) // 5
+    # Polling timeout: execution timeout + 60s buffer, checked every 0.5s for near real-time streaming
+    max_poll_count = (config.timeout_seconds + 60) // 0.5
 
     logger.info(f"[trace={trace_id}] Started background task {task_id} (subagent={subagent_type}, timeout={config.timeout_seconds}s, polling_limit={max_poll_count} polls)")
 
-    writer = get_stream_writer()
-    # Send Task Started message'
-    writer({"type": "task_started", "task_id": task_id, "description": description})
+    # Use StreamBridge for SSE events if available, otherwise fall back to StreamWriter
+    async def publish_event(event_type: str, data: Any) -> None:
+        """Publish event to SSE stream via bridge or writer."""
+        if stream_bridge is not None and run_id is not None:
+            await stream_bridge.publish(run_id, event_type, data)
+        else:
+            writer = get_stream_writer()
+            writer({"type": event_type, **data})
+
+    # Send Task Started message
+    await publish_event("task_started", {"task_id": task_id, "description": description})
 
     try:
         while True:
@@ -203,7 +225,7 @@ async def task_tool(
 
             if result is None:
                 logger.error(f"[trace={trace_id}] Task {task_id} not found in background tasks")
-                writer({"type": "task_failed", "task_id": task_id, "error": "Task disappeared from background tasks"})
+                await publish_event("task_failed", {"task_id": task_id, "error": "Task disappeared from background tasks"})
                 cleanup_background_task(task_id)
                 return f"Error: Task {task_id} disappeared from background tasks"
 
@@ -219,46 +241,43 @@ async def task_tool(
                 # Send task_running event for each new message
                 for i in range(last_message_count, current_message_count):
                     message = ai_messages[i]
-                    writer(
-                        {
-                            "type": "task_running",
-                            "task_id": task_id,
-                            "message": message,
-                            "message_index": i + 1,  # 1-based index for display
-                            "total_messages": current_message_count,
-                        }
-                    )
+                    await publish_event("task_running", {
+                        "task_id": task_id,
+                        "message": message,
+                        "message_index": i + 1,  # 1-based index for display
+                        "total_messages": current_message_count,
+                    })
                     logger.info(f"[trace={trace_id}] Task {task_id} sent message #{i + 1}/{current_message_count}")
                 last_message_count = current_message_count
 
             # Check if task completed, failed, or timed out
             if result.status == SubagentStatus.COMPLETED:
-                writer({"type": "task_completed", "task_id": task_id, "result": result.result})
+                await publish_event("task_completed", {"task_id": task_id, "result": result.result})
                 logger.info(f"[trace={trace_id}] Task {task_id} completed after {poll_count} polls")
                 cleanup_background_task(task_id)
                 return f"Task Succeeded. Result: {result.result}"
             elif result.status == SubagentStatus.FAILED:
-                writer({"type": "task_failed", "task_id": task_id, "error": result.error})
+                await publish_event("task_failed", {"task_id": task_id, "error": result.error})
                 logger.error(f"[trace={trace_id}] Task {task_id} failed: {result.error}")
                 cleanup_background_task(task_id)
                 return f"Task failed. Error: {result.error}"
             elif result.status == SubagentStatus.CANCELLED:
-                writer({"type": "task_cancelled", "task_id": task_id, "error": result.error})
+                await publish_event("task_cancelled", {"task_id": task_id, "error": result.error})
                 logger.info(f"[trace={trace_id}] Task {task_id} cancelled: {result.error}")
                 cleanup_background_task(task_id)
                 return "Task cancelled by user."
             elif result.status == SubagentStatus.TIMED_OUT:
-                writer({"type": "task_timed_out", "task_id": task_id, "error": result.error})
+                await publish_event("task_timed_out", {"task_id": task_id, "error": result.error})
                 logger.warning(f"[trace={trace_id}] Task {task_id} timed out: {result.error}")
                 cleanup_background_task(task_id)
                 return f"Task timed out. Error: {result.error}"
 
-            # Still running, wait before next poll
-            await asyncio.sleep(5)
+            # Still running, wait before next poll (reduced from 5s to 0.5s for near real-time streaming)
+            await asyncio.sleep(0.5)
             poll_count += 1
 
             # Polling timeout as a safety net (in case thread pool timeout doesn't work)
-            # Set to execution timeout + 60s buffer, in 5s poll intervals
+            # Set to execution timeout + 60s buffer, in 0.5s poll intervals for near real-time streaming
             # This catches edge cases where the background task gets stuck
             # Note: We don't call cleanup_background_task here because the task may
             # still be running in the background. The cleanup will happen when the
@@ -292,7 +311,7 @@ async def task_tool(
                     logger.warning(f"[trace={trace_id}] Deferred cleanup for task {task_id} timed out after {cleanup_poll_count} polls")
                     return
 
-                await asyncio.sleep(5)
+                await asyncio.sleep(0.5)
                 cleanup_poll_count += 1
 
         def log_cleanup_failure(cleanup_task: asyncio.Task[None]) -> None:

offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py

@@ -6,7 +6,7 @@ from deerflow.config import get_app_config
 from deerflow.config.app_config import AppConfig
 from deerflow.reflection import resolve_variable
 from deerflow.sandbox.security import is_host_bash_allowed
-from deerflow.tools.builtins import ask_clarification_tool, memory_tool, present_file_tool, scheduled_task_tool, task_tool, view_image_tool
+from deerflow.tools.builtins import ask_clarification_tool, fast_chat_tool, fast_llm_answer, memory_tool, present_file_tool, router_quality_check, scheduled_task_tool, task_tool, view_image_tool
 from deerflow.tools.builtins.tool_search import reset_deferred_registry
 
 logger = logging.getLogger(__name__)
@@ -16,6 +16,7 @@ BUILTIN_TOOLS = [
     ask_clarification_tool,
     fast_chat_tool,
     fast_llm_answer,
+    router_quality_check,
 ]
 
 SUBAGENT_TOOLS = [

offline-backend-20260512/backend/uv.lock

@@ -786,7 +786,6 @@ dependencies = [
     { name = "dingtalk-stream" },
     { name = "email-validator" },
     { name = "fastapi" },
-    { name = "greenlet" },
     { name = "httpx" },
     { name = "langgraph-sdk" },
     { name = "lark-oapi" },
@@ -826,7 +825,6 @@ requires-dist = [
     { name = "dingtalk-stream", specifier = ">=0.24.3" },
     { name = "email-validator", specifier = ">=2.0.0" },
     { name = "fastapi", specifier = ">=0.115.0" },
-    { name = "greenlet", specifier = ">=3.4.0" },
     { name = "httpx", specifier = ">=0.28.0" },
     { name = "langgraph-sdk", specifier = ">=0.1.51" },
     { name = "lark-oapi", specifier = ">=1.4.0" },