Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into xhy

添加 multi-llm 路由智能体（LangGraph），支持多 LLM 分发能力；在 Gateway 中注册 /api/multi-llm 路由，并豁免其 CSRF 和认证校验；同时新增 asyncmy 依赖和聊天流格式文档。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

NEW_CHAT_STREAM_FORMAT.md

@@ -0,0 +1,474 @@
+# New Chat 接口流式返回格式说明
+
+## 接口概览
+
+- **路径**：`POST /new_chat`
+- **传输协议**：Server-Sent Events (SSE)
+- **Content-Type**：`text/event-stream`
+- **协议基础**：Vercel AI SDK Data Stream Protocol
+
+---
+
+## 请求格式
+
+```json
+{
+  "chat_id": 123,
+  "search_space_id": 456,
+  "user_query": "用户问题",
+  "pic_query": [{"image_url": "/tmp/xxx.png"}],
+  "model_name": "gpt-4o",
+  "enable_citations": true,
+}
+```
+
+| 字段 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| `chat_id` | int | ✅ | 会话 ID，同时作为 LangGraph 的 thread_id |
+| `search_space_id` | int | ✅ | 搜索空间 ID |
+| `user_query` | string | ✅ | 用户问题 |
+| `pic_query` | list\|null | ❌ | 图片路径列表，格式 `[{"image_url": "..."}]` |
+| `model_name` | string\|null | ❌ | 指定模型名称（插件场景使用） |
+| `enable_citations` | bool | ❌ | 是否启用引用（默认 true） |
+| `messages` | list\|null | ❌ | 历史消息列表 |
+| `attachments` | list\|null | ❌ | 上传附件（已解析为 Markdown） |
+| `mentioned_document_ids` | list\|null | ❌ | 前端 @ 引用的文档 ID 列表 |
+
+---
+
+## 响应 Headers
+
+```
+Content-Type: text/event-stream
+Cache-Control: no-cache
+Connection: keep-alive
+X-Accel-Buffering: no
+```
+
+---
+
+## SSE 数据帧格式
+
+每条消息均为标准 SSE 格式：
+
+```
+data: {JSON 对象}\n\n
+```
+
+流结束时发送：
+
+```
+data: [DONE]\n\n
+```
+
+---
+
+## 事件类型详解
+
+### 1. 消息生命周期
+
+#### `start` — 消息开始
+
+```json
+{"type": "start", "messageId": "msg_abc123"}
+```
+
+| 字段 | 说明 |
+|------|------|
+| `messageId` | 本次消息的唯一 ID |
+
+---
+
+#### `finish` — 消息结束
+
+```json
+{"type": "finish"}
+```
+
+---
+
+#### `[DONE]` — 流终止标记
+
+```
+data: [DONE]
+```
+
+> 这是流中最后一帧，为纯字符串，不是 JSON。
+
+---
+
+### 2. 步骤控制（多轮工具调用）
+
+#### `start-step` — 步骤开始
+
+```json
+{"type": "start-step"}
+```
+
+#### `finish-step` — 步骤结束
+
+```json
+{"type": "finish-step"}
+```
+
+> 每次 LLM 调用算一个 step。工具调用期间会有多个 step 交替出现。
+
+---
+
+### 3. 文本块（start / delta / end 模式）
+
+#### `text-start` — 文本块开始
+
+```json
+{"type": "text-start", "id": "text_abc123"}
+```
+
+#### `text-delta` — 文本增量内容
+
+```json
+{"type": "text-delta", "id": "text_abc123", "delta": "Hello, "}
+```
+
+#### `text-end` — 文本块结束
+
+```json
+{"type": "text-end", "id": "text_abc123"}
+```
+
+---
+
+### 4. 推理块（Reasoning，适用于 DeepSeek R1 等思维链模型）
+
+#### `reasoning-start` — 推理块开始
+
+```json
+{"type": "reasoning-start", "id": "reasoning_abc123"}
+```
+
+#### `reasoning-delta` — 推理增量内容
+
+```json
+{"type": "reasoning-delta", "id": "reasoning_abc123", "delta": "让我先思考..."}
+```
+
+#### `reasoning-end` — 推理块结束
+
+```json
+{"type": "reasoning-end", "id": "reasoning_abc123"}
+```
+
+> 触发时机：
+> - 模型返回 `additional_kwargs.reasoning_content`（DeepSeek R1 原生）
+> - 模型输出中包含 `<think>...</think>` 标签（自动剥离）
+
+---
+
+### 5. 工具调用
+
+#### `tool-input-start` — 工具调用开始
+
+```json
+{"type": "tool-input-start", "toolCallId": "call_abc123", "toolName": "search_knowledge_base"}
+```
+
+#### `tool-input-delta` — 工具输入流式内容（增量）
+
+```json
+{"type": "tool-input-delta", "toolCallId": "call_abc123", "inputTextDelta": "San Fra"}
+```
+
+#### `tool-input-available` — 工具输入完整参数
+
+```json
+{
+  "type": "tool-input-available",
+  "toolCallId": "call_abc123",
+  "toolName": "search_knowledge_base",
+  "input": {"query": "机器学习入门"}
+}
+```
+
+#### `tool-output-available` — 工具执行结果
+
+```json
+{
+  "type": "tool-output-available",
+  "toolCallId": "call_abc123",
+  "output": {"status": "completed", "result_length": 1234}
+}
+```
+
+> 注意：`search_knowledge_base` 只返回摘要（不含全文），`write_todos`、`link_preview`、`display_image`、`scrape_webpage`、`generate_podcast` 会返回完整 output 供前端渲染组件。
+
+---
+
+### 6. 来源引用
+
+#### `source-url` — URL 来源
+
+```json
+{
+  "type": "source-url",
+  "sourceId": "https://example.com",
+  "url": "https://example.com",
+  "title": "示例页面"
+}
+```
+
+#### `source-document` — 文档来源
+
+```json
+{
+  "type": "source-document",
+  "sourceId": "doc_123",
+  "mediaType": "file",
+  "title": "报告.pdf",
+  "description": "第三季度报告"
+}
+```
+
+---
+
+### 7. 文件
+
+```json
+{"type": "file", "url": "https://example.com/image.png", "mediaType": "image/png"}
+```
+
+---
+
+### 8. 自定义数据（`data-*` 类型）
+
+#### `data-terminal-info` — 终端信息提示
+
+```json
+{
+  "type": "data-terminal-info",
+  "data": {
+    "text": "知识库搜索已完成",
+    "type": "success"
+  }
+}
+```
+
+`type` 枚举值：`info` | `error` | `success` | `warning`
+
+---
+
+#### `data-further-questions` — 推荐追问问题
+
+```json
+{
+  "type": "data-further-questions",
+  "data": {
+    "questions": ["问题1", "问题2", "问题3"]
+  }
+}
+```
+
+---
+
+#### `data-thinking-step` — 思考步骤（思维链 UI 展示）
+
+```json
+{
+  "type": "data-thinking-step",
+  "data": {
+    "id": "thinking-1",
+    "title": "正在搜索知识库",
+    "status": "in_progress",
+    "items": ["Query: 机器学习入门"]
+  }
+}
+```
+
+| 字段 | 说明 |
+|------|------|
+| `id` | 步骤唯一 ID（格式 `thinking-N`） |
+| `title` | 步骤标题 |
+| `status` | `pending` \| `in_progress` \| `completed` |
+| `items` | 步骤详情列表（可选） |
+
+> 同一个 `id` 的步骤事件会被重复发送以更新状态（先 `in_progress` 后 `completed`）。
+
+**常见步骤标题对照：**
+
+| 工具 / 场景 | 标题 |
+|------------|------|
+| 初始分析 | 正在理解您的请求 / 正在分析您的内容 |
+| `search_knowledge_base` | 查询个人知识库中 |
+| `search_web` | 网查搜索中 |
+| `link_preview` | 正在获取链接预览 |
+| `display_image` | 正在分析图片 |
+| `scrape_webpage` | 正在抓取网页 |
+| `write_todos`（首次） | 正在创建计划 |
+| `write_todos`（更新） | 正在更新进度 |
+| `generate_podcast` | 正在生成播客 |
+
+---
+
+### 9. 错误
+
+```json
+{
+  "type": "error",
+  "errorText": "API rate limit exceeded",
+  "errorType": "llm",
+  "errorCode": "rate_limit"
+}
+```
+
+| 字段 | 说明 |
+|------|------|
+| `errorText` | 错误描述文本 |
+| `errorType` | 错误类型（见下表） |
+| `errorCode` | 具体错误代码（可选） |
+
+**errorType 枚举：**
+
+| 值 | 含义 |
+|----|------|
+| `llm` | LLM / API 错误（模型不存在、限流、密钥错误等） |
+| `network` | 网络连通性错误 |
+| `timeout` | 请求超时 |
+| `authentication` | 鉴权 / 权限错误 |
+| `rate_limit` | 速率限制 |
+| `validation` | 输入验证错误 |
+| `internal` | 内部服务器错误 |
+| `unknown` | 未知错误 |
+
+---
+
+### 10. 中断（Human-in-the-loop）
+
+```json
+{
+  "type": "interrupt",
+  "messageId": "msg_abc123",
+  "interrupt": {
+    "action_requests": [...],
+    "review_configs": [...]
+  },
+  "payload": {"expertAgentId": 42}
+}
+```
+
+---
+
+## 完整流示例（普通问答）
+
+```
+data: {"type":"start","messageId":"msg_abc123"}
+
+data: {"type":"start-step"}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-1","title":"正在理解您的请求","status":"in_progress","items":["Processing: 什么是向量数据库？"]}}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-1","title":"正在理解您的请求","status":"completed","items":["Processing: 什么是向量数据库？"]}}
+
+data: {"type":"text-start","id":"text_xyz789"}
+
+data: {"type":"text-delta","id":"text_xyz789","delta":"向量数据库是"}
+
+data: {"type":"text-delta","id":"text_xyz789","delta":"一种专为存储和检索"}
+
+data: {"type":"text-delta","id":"text_xyz789","delta":"高维向量而设计的数据库..."}
+
+data: {"type":"text-end","id":"text_xyz789"}
+
+data: {"type":"finish-step"}
+
+data: {"type":"finish"}
+
+data: [DONE]
+```
+
+---
+
+## 完整流示例（含工具调用）
+
+```
+data: {"type":"start","messageId":"msg_abc123"}
+
+data: {"type":"start-step"}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-1","title":"正在理解您的请求","status":"in_progress","items":["Processing: 搜索最新 AI 论文"]}}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-1","title":"正在理解您的请求","status":"completed","items":["Processing: 搜索最新 AI 论文"]}}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-2","title":"查询个人知识库中","status":"in_progress","items":["Query: 最新 AI 论文"]}}
+
+data: {"type":"tool-input-start","toolCallId":"call_xxx","toolName":"search_knowledge_base"}
+
+data: {"type":"tool-input-available","toolCallId":"call_xxx","toolName":"search_knowledge_base","input":{"query":"最新 AI 论文"}}
+
+data: {"type":"data-thinking-step","data":{"id":"thinking-2","title":"Searching knowledge base","status":"completed","items":["Query: 最新 AI 论文","Found relevant information (2048 chars)"]}}
+
+data: {"type":"tool-output-available","toolCallId":"call_xxx","output":{"status":"completed","result_length":2048}}
+
+data: {"type":"data-terminal-info","data":{"text":"知识库搜索已完成","type":"success"}}
+
+data: {"type":"text-start","id":"text_xyz789"}
+
+data: {"type":"text-delta","id":"text_xyz789","delta":"根据知识库中的内容..."}
+
+data: {"type":"text-end","id":"text_xyz789"}
+
+data: {"type":"finish-step"}
+
+data: {"type":"finish"}
+
+data: [DONE]
+```
+
+---
+
+## 完整流示例（含推理过程）
+
+```
+data: {"type":"start","messageId":"msg_abc123"}
+
+data: {"type":"start-step"}
+
+data: {"type":"reasoning-start","id":"reasoning_r01"}
+
+data: {"type":"reasoning-delta","id":"reasoning_r01","delta":"让我先分析这个问题..."}
+
+data: {"type":"reasoning-delta","id":"reasoning_r01","delta":"需要考虑以下几个方面..."}
+
+data: {"type":"reasoning-end","id":"reasoning_r01"}
+
+data: {"type":"text-start","id":"text_xyz789"}
+
+data: {"type":"text-delta","id":"text_xyz789","delta":"综合分析后，结论是..."}
+
+data: {"type":"text-end","id":"text_xyz789"}
+
+data: {"type":"finish-step"}
+
+data: {"type":"finish"}
+
+data: [DONE]
+```
+
+---
+
+## 标准消息流顺序
+
+```
+start
+  └─ start-step
+       ├─ [data-thinking-step × N]   思考步骤（可多次）
+       ├─ [reasoning-start/delta/end] 推理过程（可选）
+       ├─ [tool-input-start]          工具调用开始（可多次）
+       ├─ [tool-input-available]      工具参数确认
+       ├─ [tool-output-available]     工具结果
+       ├─ [data-terminal-info]        终端提示（可多次）
+       ├─ [source-url / source-document] 来源引用（可选）
+       ├─ text-start
+       │    └─ text-delta × N         流式文本
+       └─ text-end
+     finish-step
+finish
+[DONE]
+```

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

@@ -0,0 +1,229 @@
+# 问答模式路由智能体设计方案
+
+> **状态**：部分实现 ✓
+> - ✅ `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 执行，路由器对执行结果进行**完整性质检**，通过则流式输出到前端，不通过则重新匹配模式并再次执行。
+
+## 2. 整体流程
+
+```
+用户问题
+    │
+    ▼
+┌──────────────────────────────┐
+│ 路由器智能体                  │
+│ (已有 SOUL.md)                │
+│ ① 判断最佳模式                │
+└─────────────┬────────────────┘
+              │ 选择模式
+              ▼
+┌──────────────────────────────┐
+│ 子 Agent 执行                 │
+│ (完整执行完毕，结果缓存)        │
+└─────────────┬────────────────┘
+              │ 执行完毕
+              ▼
+┌──────────────────────────────┐
+│ 路由器质检                    │
+│ LLM 判断结果是否完整覆盖用户问题 │
+└─────────────┬────────────────┘
+              │
+         ③ 质检结果
+         ┌────┴────┐
+         │通过      │不通过
+         ▼         ▼
+     流式输出   重新匹配模式
+     到前端     (最多重试2次)
+```
+
+- **最大重试次数**：每个问题最多 2 次重试（总共最多 3 次执行），超过后强制输出当前结果并标注"结果可能不完整"
+
+## 3. 目录结构
+
+```
+.deer-flow/agents/f0e75f44f7594053ba1567ed89258f35/
+├── SOUL.md                     ← 路由器智能体（已有）
+├── config.yaml                 ← 路由器配置
+└── subagents/
+    ├── fast/
+    │   ├── SOUL.md             ← 快速问答子 Agent
+    │   └── config.yaml         ← tool_groups 声明
+    ├── rag/
+    │   ├── SOUL.md
+    │   └── config.yaml
+    ├── bigMode/
+    │   ├── SOUL.md
+    │   └── config.yaml
+    ├── moreModes/
+    │   ├── SOUL.md
+    │   └── config.yaml
+    ├── taskChat/
+    │   ├── SOUL.md
+    │   └── config.yaml
+    ├── hxchat/
+    │   ├── SOUL.md
+    │   └── config.yaml
+    └── deerFlow/
+        ├── SOUL.md
+        └── config.yaml
+```
+
+**父子关系**：子 Agent 目录放在路由器下二级 `subagents/`，通过目录层级体现父子结构。
+
+## 4. 子 Agent 配置
+
+### 4.1 config.yaml 格式
+
+```yaml
+id: rag
+name: 知识库问答子Agent
+description: 从知识库检索并返回引用
+tool_groups:
+  - knowledge          # 知识库检索工具组
+  - file:read          # 只读文件工具
+skills: []
+model: deepseek-chat
+```
+
+### 4.2 tool_groups 说明
+
+- tool_groups 在系统级 `config.yaml` 中统一定义（`tool_groups` 列表）
+- 子 Agent 的 config.yaml 通过 `tool_groups` 字段声明自己可以使用哪些工具组
+- 工具组示例：web（网页搜索）、knowledge（知识库检索）、file:read、file:write、bash 等
+
+## 5. 调用方式
+
+### 5.1 复用 Gateway API（方案 A — 推荐）
+
+路由器通过 DeerFlow 现有 `/api/threads/{thread_id}/runs/stream` 接口调用子 Agent：
+
+```
+POST /api/threads/{thread_id}/runs/stream
+{
+  "assistant_id": "f0e75f44f7594053ba1567ed89258f35",
+  "context": {
+    "subagent_mode": "rag"          ← 路由器注入子模式
+  },
+  "input": {
+    "messages": [...]
+  },
+  "stream_mode": ["values"]
+}
+```
+
+**执行步骤**：
+1. 路由器启动一个 run，assistant_id 指向路由器本身
+2. 路由器通过 `context.subagent_mode` 告知 DeerFlow 运行时本次应加载哪个子 Agent
+3. DeerFlow 运行时加载 `subagents/{subagent_mode}/SOUL.md` 和对应的 tool_groups
+4. 子 Agent 完整执行，结果缓存在 StreamBridge 中
+5. 执行完毕后，路由器读取完整结果
+
+**context.subagent_mode 注入机制**：
+- `build_run_config` 中新增 `subagent_mode` 到 `_CONTEXT_CONFIGURABLE_KEYS`
+- `make_lead_agent` 读取 `configurable.subagent_mode`，如果存在则加载对应子 Agent 的 SOUL.md 和工具配置
+- 如果 `subagent_mode` 不存在，则正常运行路由器自己的 SOUL.md
+
+### 5.2 质检提示词
+
+路由器在子 Agent 执行完毕后，用 LLM 进行完整性质检：
+
+```
+你是一个质检员。上游子 Agent 已根据问题执行完毕。
+请判断以下结果是否完整覆盖了用户问题的所有子问题。
+
+用户原始问题：{original_question}
+
+子 Agent 执行结果：
+{agent_output}
+
+请只返回 JSON 格式：
+{
+  "passed": true/false,
+  "reason": "判断理由（10字内）",
+  "missing_parts": ["缺失的子问题1", "缺失的子问题2"]  // 如果 passed=false
+}
+```
+
+**质检维度**：完整性（结果是否覆盖了用户问题的所有子问题）
+
+## 6. 关键代码改动
+
+### 6.1 DeerFlow 运行时改动 ✓
+
+**文件**：`packages/harness/deerflow/agents/lead_agent/agent.py`
+
+已实现：
+- `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 改动 ✓
+
+**文件**：`app/gateway/services.py`
+
+已实现：
+- `_CONTEXT_CONFIGURABLE_KEYS` 已包含 `subagent_mode`（第149-171行）
+- `merge_run_context_overrides` 将 `subagent_mode` 透传到 configurable
+
+### 6.3 路由器 SOUL.md 改动 ✓
+
+**文件**：`.deer-flow/agents/f0e75f44f7594053ba1567ed89258f35/SOUL.md`
+
+已更新：
+- 角色扩展：增加"在确定模式后，调用相应子 Agent 执行并对结果进行质检"的职责
+- 工具描述：已声明 router_quality_check 工具
+
+### 6.4 新增路由器质检工具 ✓
+
+**文件**：`packages/harness/deerflow/tools/builtins/router_quality_tool.py`
+
+已实现：
+```python
+@tool("router_quality_check", parse_docstring=True, return_direct=True)
+def router_quality_check(original_question: str, agent_output: str) -> str:
+    """质检工具 — 判断子 Agent 输出是否完整覆盖用户问题"""
+```
+
+## 7. 数据流
+
+```
+用户问题
+    │
+    ▼
+路由器 SOUL.md 判断模式
+    │
+    ▼
+启动子 Agent Run（context.subagent_mode = rag）
+    │
+    ▼
+子 Agent SOUL.md + tool_groups 执行
+    │
+    ▼
+StreamBridge 缓存完整结果
+    │
+    ▼
+路由器调用 router_quality_check 质检
+    │
+    ├── passed=true → 流式输出到前端
+    └── passed=false → 重新匹配模式（最多2次重试）
+```
+
+## 8. 测试计划
+
+- **单元测试**：`tests/test_router_subagent.py` — 测试模式判断、质检逻辑、重试机制
+- **集成测试**：手动调用 `/api/threads/{thread_id}/runs/stream`，验证完整流程
+- **质检提示词测试**：针对不同问题类型测试质检结果的准确性
+
+## 9. 依赖项
+
+- 路由器复用现有 Gateway API，无需新建接口
+- 质检使用 LLM（deepseek-chat 或配置中指定的模型）
+- tool_groups 机制依赖系统级 config.yaml 中的工具组定义

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

@@ -38,6 +38,7 @@ from app.gateway.routers import (
     mcp,
     memory,
     models,
+    multi_llm,
     multi_agent,
     notifications,
     public_embed,
@@ -872,6 +873,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # AI Writing multi-agent API is mounted at /api/ai-writing
     app.include_router(ai_writing.router)
 
+    # Multi-LLM routing agent API is mounted at /api/multi-llm
+    app.include_router(multi_llm.router)
+
     # Scheduled task API is mounted at /api/scheduled-tasks
     app.include_router(scheduled_tasks.router)
 

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

@@ -33,6 +33,7 @@ _PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
     "/redoc",
     "/openapi.json",
     "/api/public/",
+    "/api/multi-llm/",
 )
 
 # Exact auth paths that are public (login/register/status check).

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

@@ -49,6 +49,9 @@ def should_check_csrf(request: Request) -> bool:
     # PUT/POST returns 403 even though auth was bypassed earlier in the chain.
     if path.startswith("/api/public/"):
         return False
+    # Exempt multi-llm streaming endpoint (API-to-API calls, no browser session)
+    if path.startswith("/api/multi-llm"):
+        return False
     return True
 
 

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

@@ -320,6 +320,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/app/gateway/routers/__init__.py

@@ -1,3 +1,3 @@
-from . import ai_writing, artifacts, assistants_compat, mcp, models, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
+from . import ai_writing, artifacts, assistants_compat, mcp, models, multi_llm, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
 
-__all__ = ["ai_writing", "artifacts", "assistants_compat", "mcp", "models", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]
+__all__ = ["ai_writing", "artifacts", "assistants_compat", "mcp", "models", "multi_llm", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]

offline-backend-20260512/backend/app/gateway/routers/multi_llm.py

@@ -0,0 +1,89 @@
+"""Multi-LLM routing agent — SSE streaming endpoint.
+
+POST /api/multi-llm/{thread_id}/stream
+  Body: {"message": "..."}
+  Returns text/event-stream with frames (all as "data: <json>\\n\\n"):
+    {"type": "routing",  "system": "B"}                       ← router decision
+    {"type": "thinking", "content": "...", "system": "B"}     ← reasoning (if available)
+    {"type": "text",     "content": "...", "system": "B"}     ← final text chunks
+    {"type": "done"}                                           ← stream end
+    {"type": "error",    "message": "..."}                    ← on failure
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import StreamingResponse
+from langchain_core.messages import HumanMessage
+from langgraph.types import Checkpointer
+from pydantic import BaseModel
+
+from app.gateway.deps import get_checkpointer
+from deerflow.agents.multi_llm.graph import build_multi_llm_graph
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/multi-llm", tags=["multi-llm"])
+
+# Lazy singleton: built once on first request, reusing the app-scoped checkpointer.
+_graph = None
+_graph_lock = asyncio.Lock()
+
+
+async def _get_graph(checkpointer: Checkpointer):
+    global _graph
+    if _graph is None:
+        async with _graph_lock:
+            if _graph is None:
+                _graph = build_multi_llm_graph(checkpointer=checkpointer)
+    return _graph
+
+
+class ChatRequest(BaseModel):
+    message: str
+
+
+@router.post("/{thread_id}/stream", summary="Multi-LLM routing agent (SSE stream)")
+async def multi_llm_stream(
+    thread_id: str,
+    body: ChatRequest,
+    request: Request,
+    checkpointer: Checkpointer = Depends(get_checkpointer),
+) -> StreamingResponse:
+    graph = await _get_graph(checkpointer)
+    lg_config = {"configurable": {"thread_id": thread_id}}
+    input_state = {"messages": [HumanMessage(content=body.message)]}
+
+    async def generate():
+        try:
+            async for event in graph.astream_events(input_state, config=lg_config, version="v2"):
+                if await request.is_disconnected():
+                    break
+
+                ev = event["event"]
+
+                if ev == "on_custom_event" and event.get("name") == "llm_chunk":
+                    d = event["data"]
+                    yield f"data: {json.dumps({'type': d['type'], 'content': d['content'], 'system': d['system']}, ensure_ascii=False)}\n\n"
+
+                elif ev == "on_chain_end" and event.get("name") == "router":
+                    output = event.get("data", {}).get("output", {})
+                    system = output.get("selected_system", "") if isinstance(output, dict) else ""
+                    if system:
+                        yield f"data: {json.dumps({'type': 'routing', 'system': system})}\n\n"
+
+            yield f"data: {json.dumps({'type': 'done'})}\n\n"
+
+        except Exception as e:
+            logger.exception("multi_llm stream failed: thread=%s", thread_id)
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={"X-Accel-Buffering": "no", "Cache-Control": "no-cache"},
+    )

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

@@ -383,10 +383,13 @@ async def _sync_legacy_agent_records(store) -> None:
     """
     from deerflow.config.agents_config import list_custom_agents
 
+    print(f"[_sync_legacy_agent_records] store type: {type(store).__name__}, id: {id(store)}", file=sys.stderr)
+    logger.info("_sync_legacy_agent_records: starting sync")
     for agent_cfg in list_custom_agents():
         agent_id = agent_cfg.id or agent_cfg.name
         if not agent_id:
             continue
+        logger.info(f"  Syncing agent: {agent_id}")
         await store.ensure_builtin(
             {
                 "id": agent_id,
@@ -395,6 +398,7 @@ async def _sync_legacy_agent_records(store) -> None:
                 "published": True,
             }
         )
+    logger.info("_sync_legacy_agent_records: done")
 
 
 # Throttle per-request legacy-sync. Matches ``routers/agents._LEGACY_SYNC_TTL_SECONDS``
@@ -466,12 +470,18 @@ async def _enforce_agent_access(body: Any, request: Request) -> None:
         return
 
     from app.gateway.deps import get_agent_store
-    from deerflow.config.agents_config import validate_agent_id
+    from deerflow.config.agents_config import validate_agent_id, list_custom_agents
     from deerflow.runtime.user_context import get_effective_user_id
 
     store = get_agent_store(request)
     await _maybe_sync_legacy_agent_records(store)
     user_id = get_effective_user_id()
+    logger.info(f"_enforce_agent_access: user_id={user_id}, agent_ids={agent_ids}")
+
+    # Debug: list all agents from list_custom_agents
+    all_agents = list_custom_agents()
+    logger.info(f"  list_custom_agents() returned {len(all_agents)} agents: {[a.id for a in all_agents]}")
+
     for agent_id in agent_ids:
         try:
             normalized = validate_agent_id(agent_id)
@@ -479,7 +489,13 @@ async def _enforce_agent_access(body: Any, request: Request) -> None:
             raise HTTPException(status_code=422, detail=str(exc)) from exc
         if normalized is None:
             continue
-        if await store.get_visible(normalized, user_id) is None:
+        record = await store.get_visible(normalized, user_id)
+        logger.info(f"  get_visible({normalized}, {user_id}) = {record}")
+        # Also check get_any
+        any_record = await store.get_any(normalized)
+        import sys
+        print(f"[_enforce_agent_access] get_any({normalized}) = {any_record}", file=sys.stderr)
+        if record is None:
             raise HTTPException(status_code=404, detail=f"Agent {normalized} not found")
 
 

offline-backend-20260512/backend/config.yaml

@@ -218,6 +218,168 @@ checkpointer:
   type: sqlite
   connection_string: .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

@@ -386,7 +386,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)
     # 写作配置 Q&A：与 is_bootstrap 同构的另一种「最小化 bootstrap」——挂 setup_writing 工具、
@@ -410,6 +409,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)
@@ -422,6 +423,41 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
         logger.warning(f"Thinking mode is enabled but model '{model_name}' does not support it; fallback to non-thinking mode.")
         thinking_enabled = False
 
+    # Check if we're in subagent mode (router agent delegating to a subagent)
+    subagent_mode = cfg.get("subagent_mode")
+    if subagent_mode and agent_id:
+        # Load subagent config and SOUL.md from subagents/{subagent_mode}/
+        from deerflow.config.paths import get_paths
+
+        paths = get_paths()
+        subagent_dir = paths.agent_dir(agent_id) / "subagents" / subagent_mode
+        subagent_soul_path = subagent_dir / "SOUL.md"
+        subagent_config_path = subagent_dir / "config.yaml"
+
+        if subagent_soul_path.exists():
+            subagent_soul = subagent_soul_path.read_text(encoding="utf-8").strip()
+        else:
+            subagent_soul = None
+
+        subagent_tool_groups = None
+        if subagent_config_path.exists():
+            try:
+                import yaml
+
+                with open(subagent_config_path, encoding="utf-8") as f:
+                    subagent_data = yaml.safe_load(f) or {}
+                subagent_tool_groups = subagent_data.get("tool_groups")
+            except yaml.YAMLError:
+                logger.warning(f"Failed to parse subagent config: {subagent_config_path}")
+
+        logger.info(
+            "Subagent mode(%s) for router(%s) -> subagent_soul: %s, tool_groups: %s",
+            subagent_mode,
+            agent_id,
+            "found" if subagent_soul else "not found",
+            subagent_tool_groups,
+        )
+
     logger.info(
         "Create Agent(%s) -> thinking_enabled: %s, reasoning_effort: %s, model_name: %s, is_plan_mode: %s, subagent_enabled: %s, max_concurrent_subagents: %s, writing_mode: %s",
         agent_id or "default",

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

@@ -955,6 +955,7 @@ def apply_prompt_template(
     available_skills: set[str] | None = None,
     is_bootstrap: bool = False,
     app_config: AppConfig | None = None,
+    soul_override: str | None = None,
     memory_injection_enabled: bool = True,
     rag_mode_enabled: bool = False,
     is_notebook_mode: bool = False,
@@ -1052,7 +1053,7 @@ When a search, knowledge, skill, or MCP tool returns a result list:
     # Format the prompt with dynamic skills and memory
     prompt = SYSTEM_PROMPT_TEMPLATE.format(
         agent_name=agent_name or storage_agent_id or "CM 助手",
-        soul=get_agent_soul(storage_agent_id),
+        soul=f"<soul>\n{soul_override}\n</soul>\n" if soul_override else get_agent_soul(storage_agent_id),
         skills_section=skills_section,
         bootstrap_section=bootstrap_section,
         deferred_tools_section=deferred_tools_section,

offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/graph.py

@@ -0,0 +1,51 @@
+"""Multi-LLM routing graph.
+
+Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
+
+Graph layout:
+  START → router → (conditional) → fast / bigmodel / hxchat / more_modes / rag / task_chat → END
+"""
+
+from __future__ import annotations
+
+from langgraph.graph import END, START, StateGraph
+
+from deerflow.agents.multi_llm.nodes.router import route_by_system, router_node
+from deerflow.agents.multi_llm.nodes.systems import (
+    bigmodel_node,
+    fast_node,
+    hxchat_node,
+    more_modes_node,
+    rag_node,
+    task_chat_node,
+)
+from deerflow.agents.multi_llm.state import MultiLLMState
+
+# LangGraph node name → node function
+_SYSTEM_NODES = {
+    "fast": fast_node,
+    "bigmodel": bigmodel_node,
+    "hxchat": hxchat_node,
+    "more_modes": more_modes_node,
+    "rag": rag_node,
+    "task_chat": task_chat_node,
+}
+
+
+def build_multi_llm_graph(checkpointer=None):
+    g = StateGraph(MultiLLMState)
+
+    g.add_node("router", router_node)
+    for name, fn in _SYSTEM_NODES.items():
+        g.add_node(name, fn)
+
+    g.add_edge(START, "router")
+    g.add_conditional_edges(
+        "router",
+        route_by_system,
+        {name: name for name in _SYSTEM_NODES},
+    )
+    for name in _SYSTEM_NODES:
+        g.add_edge(name, END)
+
+    return g.compile(checkpointer=checkpointer)

offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/nodes/router.py

@@ -0,0 +1,77 @@
+"""Router node: uses an LLM to decide which system to invoke."""
+
+from __future__ import annotations
+
+import logging
+from typing import Literal
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import RunnableConfig
+
+from deerflow.agents.multi_llm.state import MultiLLMState
+from deerflow.config.app_config import get_app_config
+from deerflow.models import create_chat_model
+
+logger = logging.getLogger(__name__)
+
+SystemName = Literal["fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat"]
+
+_VALID_SYSTEMS: tuple[str, ...] = ("fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat")
+
+_ROUTER_SYSTEM_PROMPT = f"""你是一个路由助手，负责将用户请求分配给最合适的AI系统。
+
+可用系统（共 {len(_VALID_SYSTEMS)} 个）：
+- fast       : 快速问答，适合通用问题、知识库检索
+- bigmodel   : 大模型，适合复杂推理、长文本分析
+- hxchat     : HX 对话，适合...（TODO: 补充说明）
+- moreModes  : 多模态，适合...（TODO: 补充说明）
+- rag        : RAG 检索增强，适合需要查找文档的问题
+- taskChat   : 任务对话，适合需要执行多步骤任务的场景
+
+请根据用户请求，仅输出一个系统名称（区分大小写），不要输出任何其他内容。
+有效值：fast / bigmodel / hxchat / moreModes / rag / taskChat"""
+
+
+async def router_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    messages = state.get("messages", [])
+    if not messages:
+        return {"selected_system": "fast"}
+
+    last_human = next(
+        (
+            m
+            for m in reversed(messages)
+            if getattr(m, "type", None) == "human" or (isinstance(m, dict) and m.get("role") == "user")
+        ),
+        None,
+    )
+    user_content = ""
+    if last_human is not None:
+        user_content = last_human.content if hasattr(last_human, "content") else last_human.get("content", "")
+
+    try:
+        app_config = get_app_config()
+        llm = create_chat_model(app_config=app_config)
+        prompt = [
+            SystemMessage(content=_ROUTER_SYSTEM_PROMPT),
+            HumanMessage(content=user_content or "请帮我处理这个请求"),
+        ]
+        response = await llm.ainvoke(prompt, config={"run_name": "multi_llm_router"})
+        raw = (response.content or "").strip()
+        selected = raw if raw in _VALID_SYSTEMS else "fast"
+    except Exception:
+        logger.exception("multi_llm router failed, defaulting to fast")
+        selected = "fast"
+
+    return {"selected_system": selected}
+
+
+def route_by_system(state: MultiLLMState) -> str:
+    system = state.get("selected_system", "fast")
+    if system not in _VALID_SYSTEMS:
+        system = "fast"
+    # LangGraph node names use underscores
+    return system.lower() if system in ("fast", "bigmodel", "hxchat", "rag") else {
+        "moreModes": "more_modes",
+        "taskChat": "task_chat",
+    }[system]

offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/nodes/systems.py

@@ -0,0 +1,223 @@
+"""System nodes: call each external LLM system and stream chunks via custom events.
+
+Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
+
+System URLs are read from config.yaml under multi_llm_systems.<name>_url,
+falling back to environment variables FAST_URL / BIGMODEL_URL / etc.
+
+fast uses a custom request format (Vercel AI SDK / POST /new_chat):
+  {"chat_id": <thread_id>, "search_space_id": <id>, "user_query": "..."}
+  search_space_id from config multi_llm_systems.fast_search_space_id
+  or env FAST_SEARCH_SPACE_ID (default 0).
+
+All other systems use generic {"messages": [...]} format (placeholder until
+their real formats are confirmed).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+import httpx
+from langchain_core.callbacks.manager import adispatch_custom_event
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.runnables import RunnableConfig
+
+from deerflow.agents.multi_llm.parsers import PARSERS
+from deerflow.agents.multi_llm.state import MultiLLMState
+from deerflow.config.app_config import get_app_config
+from deerflow.models import create_chat_model
+
+logger = logging.getLogger(__name__)
+
+SYSTEM_NAMES = ["fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat"]
+
+# Maps system name → env var suffix for URL lookup
+_ENV_VAR_MAP = {
+    "fast": "FAST_URL",
+    "bigmodel": "BIGMODEL_URL",
+    "hxchat": "HXCHAT_URL",
+    "moreModes": "MOREMODES_URL",
+    "rag": "RAG_URL",
+    "taskChat": "TASKCHAT_URL",
+}
+
+
+def _get_multi_llm_config() -> dict:
+    try:
+        cfg = get_app_config()
+        return getattr(cfg, "multi_llm_systems", None) or {}
+    except Exception:
+        return {}
+
+
+def _get_url(system: str) -> str:
+    c = _get_multi_llm_config()
+    config_key = f"{system}_url"
+    env_key = _ENV_VAR_MAP.get(system, f"{system.upper()}_URL")
+    return c.get(config_key) or os.environ.get(env_key, "")
+
+
+def _extract_last_user_content(messages: list) -> str:
+    for m in reversed(messages):
+        if hasattr(m, "type") and m.type == "human":
+            return m.content if isinstance(m.content, str) else str(m.content)
+        if isinstance(m, dict) and m.get("role") == "user":
+            return m.get("content", "")
+    return ""
+
+
+def _build_generic_payload(messages: list) -> dict:
+    result = []
+    for m in messages:
+        if hasattr(m, "type"):
+            role = "user" if m.type == "human" else "assistant"
+            content = m.content if isinstance(m.content, str) else str(m.content)
+        elif isinstance(m, dict):
+            role = m.get("role", "user")
+            content = m.get("content", "")
+        else:
+            continue
+        result.append({"role": role, "content": content})
+    return {"messages": result}
+
+
+_FAST_TEST_URL = "http://47.88.25.99:8000/api/v1/new_chat"
+_FAST_TEST_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI5MmFhNmRkMC03NDA5LTQ0NDEtOGQzNC1iNTY4YmZjMzg5ODIiLCJhdWQiOlsiZmFzdGFwaS11c2VyczphdXRoIl0sImV4cCI6MTc3OTQzNjY0NH0.Da_KDlQaEc-cH95AyUzcx82pwZmU3t-9fGmJC9tRF7E"
+_FAST_TEST_MODEL = "Qwen3.5-plus"
+_FAST_TEST_SEARCH_SPACE_ID = 197
+_FAST_TEST_CHAT_ID = 600
+
+
+def _build_fast_payload(messages: list, thread_id: str = "") -> dict:  # noqa: ARG001
+    """Build the /new_chat request payload for the fast system."""
+    return {
+        "model_name": _FAST_TEST_MODEL,
+        "chat_id": _FAST_TEST_CHAT_ID,
+        "search_space_id": _FAST_TEST_SEARCH_SPACE_ID,
+        "user_query": _extract_last_user_content(messages),
+    }
+
+
+async def _call_system(system: str, messages: list, thread_id: str = "") -> str:
+    if system == "fast":
+        url = _FAST_TEST_URL
+        headers = {"Authorization": f"Bearer {_FAST_TEST_TOKEN}"}
+    else:
+        url = _get_url(system)
+        headers = {}
+        if not url:
+            raise ValueError(
+                f"No URL configured for system '{system}'. "
+                f"Set multi_llm_systems.{system}_url in config.yaml or {_ENV_VAR_MAP.get(system, system.upper() + '_URL')} env var."
+            )
+
+    parser = PARSERS[system]
+    payload = _build_fast_payload(messages, thread_id) if system == "fast" else _build_generic_payload(messages)
+    text_buf = ""
+
+    async with httpx.AsyncClient(timeout=120) as client:
+        async with client.stream("POST", url, json=payload, headers=headers) as resp:
+            resp.raise_for_status()
+            async for chunk in parser.parse(resp):
+                if chunk.content:
+                    if chunk.type == "thinking":
+                        await adispatch_custom_event("llm_chunk", {"type": "thinking", "content": chunk.content, "system": system})
+                    else:
+                        text_buf += chunk.content
+                        await adispatch_custom_event("llm_chunk", {"type": "text", "content": chunk.content, "system": system})
+
+    return text_buf
+
+
+def _extract_chunk_parts(chunk: object) -> tuple[str, str]:
+    """Return (thinking_delta, text_delta) from a LangChain streaming chunk."""
+    thinking = ""
+    text = ""
+    content = getattr(chunk, "content", "")
+    if isinstance(content, str):
+        text = content
+    elif isinstance(content, list):
+        for block in content:
+            if isinstance(block, str):
+                text += block
+            elif isinstance(block, dict):
+                btype = block.get("type", "")
+                if btype in ("thinking", "thinking_delta"):
+                    thinking += block.get("thinking", "")
+                elif btype in ("text", "text_delta"):
+                    text += block.get("text", "")
+                elif btype == "reasoning":
+                    thinking += block.get("content", "")
+    if not thinking:
+        kwargs = getattr(chunk, "additional_kwargs", {}) or {}
+        r = kwargs.get("reasoning_content") or ""
+        if isinstance(r, str):
+            thinking = r
+    return thinking, text
+
+
+async def _fallback_to_local_llm(messages: list, config: RunnableConfig) -> str:
+    """Fall back to the local configured LLM when an external system fails."""
+    app_config = get_app_config()
+    llm = create_chat_model(app_config=app_config)
+
+    lc_messages = []
+    for m in messages:
+        if hasattr(m, "type"):
+            content = m.content if isinstance(m.content, str) else str(m.content)
+            lc_messages.append(HumanMessage(content=content) if m.type == "human" else AIMessage(content=content))
+        elif isinstance(m, dict):
+            content = m.get("content", "")
+            lc_messages.append(HumanMessage(content=content) if m.get("role") == "user" else AIMessage(content=content))
+
+    text_buf = ""
+    async for chunk in llm.astream(lc_messages):
+        thinking, text = _extract_chunk_parts(chunk)
+        if thinking:
+            await adispatch_custom_event("llm_chunk", {"type": "thinking", "content": thinking, "system": "local"})
+        if text:
+            text_buf += text
+            await adispatch_custom_event("llm_chunk", {"type": "text", "content": text, "system": "local"})
+    return text_buf
+
+
+async def _system_node(system: str, state: MultiLLMState, config: RunnableConfig) -> dict:
+    messages = state.get("messages", [])
+    thread_id = (config or {}).get("configurable", {}).get("thread_id", "")
+    try:
+        result = await _call_system(system, messages, thread_id=thread_id)
+    except Exception as e:
+        logger.warning("multi_llm system '%s' failed (%s), falling back to local LLM", system, e)
+        await adispatch_custom_event("llm_chunk", {"type": "fallback", "system": "local", "reason": str(e)})
+        try:
+            result = await _fallback_to_local_llm(messages, config)
+        except Exception as e2:
+            logger.exception("local LLM fallback also failed")
+            result = f"[系统异常，请稍后重试: {e2}]"
+    return {"messages": [AIMessage(content=result)]}
+
+
+async def fast_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("fast", state, config)
+
+
+async def bigmodel_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("bigmodel", state, config)
+
+
+async def hxchat_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("hxchat", state, config)
+
+
+async def more_modes_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("moreModes", state, config)
+
+
+async def rag_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("rag", state, config)
+
+
+async def task_chat_node(state: MultiLLMState, config: RunnableConfig) -> dict:
+    return await _system_node("taskChat", state, config)

offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/parsers.py

@@ -0,0 +1,88 @@
+"""Stream parsers for 6 external LLM systems.
+
+Each system returns a different streaming format. All parsers normalize the
+raw HTTP byte stream into a sequence of StreamChunk(type, content) where
+type is either "thinking" (reasoning process) or "text" (final output).
+
+Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
+Currently implemented: fast (Vercel AI SDK Data Stream Protocol / POST /new_chat)
+Others: TODO — using plain-text placeholder until formats are confirmed.
+"""
+
+from __future__ import annotations
+
+import abc
+import json
+from dataclasses import dataclass
+from typing import AsyncIterator, Literal
+
+
+@dataclass
+class StreamChunk:
+    type: Literal["thinking", "text"]
+    content: str
+
+
+class BaseSystemParser(abc.ABC):
+    """Parse a raw httpx response stream into normalized StreamChunks."""
+
+    @abc.abstractmethod
+    async def parse(self, response) -> AsyncIterator[StreamChunk]:
+        ...
+
+
+class FastParser(BaseSystemParser):
+    """fast: Vercel AI SDK Data Stream Protocol (POST /new_chat).
+
+    Relevant event types:
+      text-delta      → StreamChunk(type="text",    content=payload["delta"])
+      reasoning-delta → StreamChunk(type="thinking", content=payload["delta"])
+
+    All other event types (start, finish, start-step, finish-step,
+    tool-input-*, tool-output-*, data-*, source-*, error, etc.) are silently
+    skipped — they are UI-control signals, not content.
+    """
+
+    async def parse(self, response) -> AsyncIterator[StreamChunk]:
+        async for line in response.aiter_lines():
+            if not line.startswith("data:"):
+                continue
+            raw = line[5:].strip()
+            if not raw or raw == "[DONE]":
+                continue
+            try:
+                payload = json.loads(raw)
+            except json.JSONDecodeError:
+                continue
+            ev_type = payload.get("type", "")
+            if ev_type == "text-delta":
+                delta = payload.get("delta", "")
+                if delta:
+                    yield StreamChunk(type="text", content=delta)
+            elif ev_type == "reasoning-delta":
+                delta = payload.get("delta", "")
+                if delta:
+                    yield StreamChunk(type="thinking", content=delta)
+
+
+class PlainTextParser(BaseSystemParser):
+    """Fallback: plain text stream, no thinking content.
+
+    Used as placeholder for bigmodel / hxchat / moreModes / rag / taskChat
+    until their real streaming formats are confirmed.
+    """
+
+    async def parse(self, response) -> AsyncIterator[StreamChunk]:
+        async for chunk in response.aiter_text():
+            if chunk:
+                yield StreamChunk(type="text", content=chunk)
+
+
+PARSERS: dict[str, BaseSystemParser] = {
+    "fast": FastParser(),
+    "bigmodel": PlainTextParser(),   # TODO: replace with real parser
+    "hxchat": PlainTextParser(),     # TODO: replace with real parser
+    "moreModes": PlainTextParser(),  # TODO: replace with real parser
+    "rag": PlainTextParser(),        # TODO: replace with real parser
+    "taskChat": PlainTextParser(),   # TODO: replace with real parser
+}

offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/state.py

@@ -0,0 +1,9 @@
+from typing import Annotated
+
+from langgraph.graph.message import add_messages
+from typing_extensions import NotRequired, TypedDict
+
+
+class MultiLLMState(TypedDict):
+    messages: Annotated[list, add_messages]
+    selected_system: NotRequired[str]

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

@@ -69,6 +69,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:
@@ -222,7 +226,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,7 +1,9 @@
 from .agent_orchestration_tool import agent_orchestration_tool
 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 .setup_writing_tool import setup_writing
@@ -18,4 +20,7 @@ __all__ = [
     "scheduled_task_tool",
     "view_image_tool",
     "task_tool",
+    "fast_chat_tool",
+    "fast_llm_answer",
+    "router_quality_check",
 ]

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

@@ -0,0 +1,159 @@
+"""Fast chat tool - call new_chat API for fast Q&A mode."""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from langchain.tools import InjectedToolCallId, ToolRuntime, tool
+from langchain_core.messages import HumanMessage
+from langchain_core.runnables import RunnableConfig
+from langgraph.typing import ContextT
+
+from deerflow.agents.thread_state import ThreadState
+from deerflow.config.app_config import get_app_config
+from deerflow.models import create_chat_model
+
+logger = logging.getLogger(__name__)
+
+
+def _get_runtime_value(runtime: ToolRuntime[ContextT, ThreadState], key: str, default: Any = None) -> Any:
+    """Get value from runtime context or config."""
+    if runtime.context and isinstance(runtime.context, dict):
+        value = runtime.context.get(key)
+        if value is not None:
+            return value
+
+    runtime_config = getattr(runtime, "config", None) or {}
+    configurable = runtime_config.get("configurable", {})
+    if isinstance(configurable, dict):
+        value = configurable.get(key)
+        if value is not None:
+            return value
+
+    return default
+
+
+@tool("fast_chat", parse_docstring=True, return_direct=True)
+async def fast_chat_tool(
+    runtime: ToolRuntime[ContextT, ThreadState],
+    prompt: str,
+    tool_call_id: Annotated[str, InjectedToolCallId],
+    search_space_id: int | None = None,
+    mentioned_document_ids: list[int] | None = None,
+    pic_query: list[dict[str, Any]] | None = None,
+    model_name: str | None = None,
+) -> str:
+    """Fast chat tool - 快速问答模式调用后端聊天接口.
+
+    用于 fast 子 Agent 直接调用后端 new_chat 接口进行快速问答.
+
+    Args:
+        runtime: 运行时上下文，自动注入.
+        prompt: 用户问题内容.
+        tool_call_id: 工具调用 ID，自动注入.
+        search_space_id: 搜索空间 ID (可选，从 context.subagent_search_space_id 获取).
+        mentioned_document_ids: 引用的文档 ID 列表 (可选).
+        pic_query: 图片查询列表 (可选).
+        model_name: 使用的模型名称 (默认使用 context.subagent_model_name 或 OpenAI3).
+
+    Returns:
+        API 响应结果的 JSON 字符串.
+    """
+    import asyncio
+    import urllib.error
+    import urllib.request
+
+    try:
+        # 从 context 读取参数，支持外部传入或运行时获取
+        thread_id = _get_runtime_value(runtime, "thread_id")
+        if thread_id is None:
+            return json.dumps({"error": "thread_id not found in context", "passed": False})
+
+        # 参数优先级：显式传入 > context 注入
+        effective_search_space_id = search_space_id if search_space_id is not None else _get_runtime_value(runtime, "subagent_search_space_id")
+        effective_model_name = model_name or _get_runtime_value(runtime, "subagent_model_name", "OpenAI3")
+        effective_mentioned_ids = mentioned_document_ids or _get_runtime_value(runtime, "subagent_mentioned_document_ids") or []
+        effective_pic_query = pic_query or _get_runtime_value(runtime, "subagent_pic_query") or []
+
+        app_config = get_app_config()
+        backend_url = getattr(app_config, "backend_url", "http://localhost:8001")
+
+        messages = [{"role": "user", "content": prompt}]
+
+        req_body = {
+            "chat_id": thread_id,
+            "user_query": prompt,
+            "search_space_id": effective_search_space_id,
+            "messages": messages,
+            "mentioned_document_ids": effective_mentioned_ids,
+            "pic_query": effective_pic_query,
+            "model_name": effective_model_name,
+        }
+
+        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=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:
+        logger.warning(f"Fast chat API HTTP error: {exc.code} {exc.reason}")
+        return json.dumps({"error": f"HTTP {exc.code}: {exc.reason}", "passed": False})
+    except Exception as exc:
+        logger.warning(f"Fast chat API failed: {exc}")
+        return json.dumps({"error": str(exc), "passed": False})
+
+
+@tool("fast_llm_answer", parse_docstring=True, return_direct=True)
+def fast_llm_answer(prompt: str) -> str:
+    """Fast LLM answer tool - 使用 LLM 直接回答简单问题.
+
+    当 fast 模式不需要调用外部 API 时，直接使用 LLM 回答简单事实查询.
+
+    Args:
+        prompt: 用户问题.
+
+    Returns:
+        LLM 回答内容.
+    """
+    try:
+        app_config = get_app_config()
+        model_name = app_config.models[0].name if app_config.models else None
+        model = create_chat_model(name=model_name, thinking_enabled=False, app_config=app_config)
+
+        response = model.invoke([HumanMessage(content=prompt)], config=RunnableConfig())
+        content = response.content if hasattr(response, "content") else str(response)
+        return content
+
+    except Exception as exc:
+        logger.warning(f"Fast LLM answer failed: {exc}")
+        return f"抱歉，处理您的问题时出现错误: {str(exc)}"

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

@@ -0,0 +1,75 @@
+"""Router quality check tool — validates sub-agent output completeness."""
+
+import json
+import logging
+
+from langchain_core.messages import HumanMessage
+from langchain_core.runnables import RunnableConfig
+from langchain_core.tools import tool
+
+from deerflow.config.app_config import get_app_config
+from deerflow.models import create_chat_model
+
+logger = logging.getLogger(__name__)
+
+QUALITY_CHECK_PROMPT = """你是一个质检员。上游子 Agent 已根据问题执行完毕。
+请判断以下结果是否完整覆盖了用户问题的所有子问题。
+
+用户原始问题：{original_question}
+
+子 Agent 执行结果：
+{agent_output}
+
+请只返回 JSON 格式：
+{{
+  "passed": true/false,
+  "reason": "判断理由（10字内）",
+  "missing_parts": ["缺失的子问题1", "缺失的子问题2"]  // 如果 passed=false
+}}"""
+
+
+@tool("router_quality_check", parse_docstring=True, return_direct=True)
+def router_quality_check(original_question: str, agent_output: str) -> str:
+    """质检工具 — 判断子 Agent 输出是否完整覆盖用户问题。
+
+    路由器在子 Agent 执行完毕后调用此工具进行完整性质检。
+    质检通过则流式输出到前端，不通过则重新匹配模式并再次执行。
+
+    Args:
+        original_question: 用户原始问题。
+        agent_output: 子 Agent 的执行结果。
+
+    Returns:
+        JSON 格式的质检结果：{{"passed": bool, "reason": str, "missing_parts": list[str]}}
+    """
+    try:
+        app_config = get_app_config()
+        model_name = app_config.models[0].name if app_config.models else None
+        model = create_chat_model(name=model_name, thinking_enabled=False, app_config=app_config)
+
+        prompt = QUALITY_CHECK_PROMPT.format(
+            original_question=original_question,
+            agent_output=agent_output,
+        )
+
+        response = model.invoke([HumanMessage(content=prompt)], config=RunnableConfig())
+        content = response.content if hasattr(response, "content") else str(response)
+
+        # Try to parse JSON from the response
+        # The model might wrap it in ```json or just return raw JSON
+        json_str = content.strip()
+        if json_str.startswith("```json"):
+            json_str = json_str[7:]
+        if json_str.startswith("```"):
+            json_str = json_str[3:]
+        if json_str.endswith("```"):
+            json_str = json_str[:-3]
+        json_str = json_str.strip()
+
+        result = json.loads(json_str)
+        return json.dumps(result, ensure_ascii=False)
+
+    except Exception as exc:
+        logger.warning(f"Router quality check failed: {exc}")
+        # Fallback: pass with unknown reason on error
+        return json.dumps({"passed": True, "reason": "质检异常默认通过", "missing_parts": []})

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
+    # Polling timeout: execution timeout + 60s buffer, checked every 0.5s for near real-time streaming
-    max_poll_count = (config.timeout_seconds + 60) // 5
+    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()
+    # Use StreamBridge for SSE events if available, otherwise fall back to StreamWriter
-    # Send Task Started message'
+    async def publish_event(event_type: str, data: Any) -> None:
-    writer({"type": "task_started", "task_id": task_id, "description": description})
+        """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(
+                    await publish_event("task_running", {
-                        {
+                        "task_id": task_id,
-                            "type": "task_running",
+                        "message": message,
-                            "task_id": task_id,
+                        "message_index": i + 1,  # 1-based index for display
-                            "message": message,
+                        "total_messages": current_message_count,
-                            "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
+            # Still running, wait before next poll (reduced from 5s to 0.5s for near real-time streaming)
-            await asyncio.sleep(5)
+            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,6 +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, 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 import agent_orchestration_tool, ask_clarification_tool, memory_tool, present_file_tool, scheduled_task_tool, task_tool, view_image_tool
 from deerflow.tools.builtins.tool_search import reset_deferred_registry
 
@@ -14,6 +15,9 @@ logger = logging.getLogger(__name__)
 BUILTIN_TOOLS = [
     present_file_tool,
     ask_clarification_tool,
+    fast_chat_tool,
+    fast_llm_answer,
+    router_quality_check,
     agent_orchestration_tool,
 ]
 

offline-backend-20260512/backend/pyproject.toml

@@ -22,7 +22,6 @@ dependencies = [
     "pyjwt>=2.9.0",
     "email-validator>=2.0.0",
     "croniter>=6.2.2",
-    "hindsight-client>=0.4.22",
     "asyncmy>=0.2.11",
 ]