接口修改

frontend-web/docs/multi-agent-api.md

@@ -0,0 +1,265 @@
+# 多智能体协调 API 接口文档
+
+后端路由：`offline-backend-20260512/backend/app/gateway/routers/multi_agent.py`
+路由前缀：`/api/multi-agent`
+
+## 概览
+
+提供一个"总控 + 多个子智能体"的协调编排接口。
+
+- **总控（leader）** 智能体：通过 `agent_orchestration` 技能给子智能体派活
+- **子智能体（special）**：执行被派到的具体任务，完成后把交付内容广播给其他线程
+
+调用方按下面顺序使用：
+
+1. 调 `POST /api/multi-agent/init` 一次性创建协调智能体 + 给每个子智能体建 `thread`，拿到 `thread_ids` 映射
+2. 客户端把 `thread_ids` 保管好，后续每次调 `POST /api/multi-agent/run/stream` 时回传
+3. 先调用 `agent_type="leader"` 拿派活清单，再针对清单里每个子智能体调用 `agent_type="special"`
+
+
+## 1. 初始化多智能体集群
+
+### `POST /api/multi-agent/init`
+
+创建协调智能体并为列表中的每个子智能体建立独立的 `thread`，同时把"职责说明"广播到所有子智能体线程作为初始上下文。
+
+#### 请求体
+
+```json
+{
+  "agent_names": ["taiwan-intelligence-gathering-agent", "us-intelligence-gathering-agent"],
+  "main_agent_name": "main-controller",
+  "model": "deepseek-chat"
+}
+```
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `agent_names` | `string[]` | 是 | 已有子智能体的 id 列表（kebab-case，区分大小写不敏感，自动转小写） |
+| `main_agent_name` | `string` | 否 | 要创建的协调智能体名，默认 `main-agent`。重名时会复用已存在的同名智能体 |
+| `model` | `string` | 否 | 协调智能体使用的模型名。不传则用后端默认模型 |
+
+#### 响应（200）
+
+```json
+{
+  "status": "success",
+  "main_agent_name": "main-controller",
+  "thread_ids": {
+    "main-controller": "thread-uuid-xxx",
+    "taiwan-intelligence-gathering-agent": "thread-uuid-yyy",
+    "us-intelligence-gathering-agent": "thread-uuid-zzz"
+  }
+}
+```
+
+`thread_ids` 是一个 `{agent_name: thread_id}` 字典。**前端必须把这个对象完整保管下来**，后续调用 stream 接口需要原样回传。
+
+#### 错误
+
+| 状态码 | 触发条件 |
+|---|---|
+| `400` | `agent_names` 为空 |
+| `401` | 未携带 `Authorization` 头 |
+| `500` | 协调智能体创建失败且不是"已存在"错误 |
+
+---
+
+## 2. 流式调度
+
+### `POST /api/multi-agent/run/stream`
+
+通过 SSE（Server-Sent Events）流式驱动一个 leader 或 special 智能体跑一轮对话。
+
+#### 请求体
+
+```json
+{
+  "agent_type": "leader",
+  "d_agent_thread_id": {
+    "main-controller": "thread-uuid-xxx",
+    "taiwan-intelligence-gathering-agent": "thread-uuid-yyy"
+  },
+  "agent_name": "main-controller",
+  "new_message": "请分析特朗普的支持率",
+  "skill_stop_names": [],
+  "model": "deepseek-chat"
+}
+```
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `agent_type` | `"leader"` \| `"special"` | 否 | 默认 `leader`。`leader` 调用协调智能体，`special` 调用子智能体 |
+| `d_agent_thread_id` | `object` | 是 | 由 `/init` 接口返回的 `thread_ids` 映射，原样回传 |
+| `agent_name` | `string` | 是 | 本轮要驱动的智能体名（必须存在于 `d_agent_thread_id` 中） |
+| `new_message` | `string` | 是 | 这轮发给智能体的用户消息 |
+| `skill_stop_names` | `string[]` | 否 | 触发后中断执行的技能名列表（默认空） |
+| `model` | `string` | 否 | 本轮 LLM 模型名，默认 `deepseek-chat` |
+
+#### 响应
+
+`Content-Type: text/event-stream` 的 SSE 长连接，按到达顺序产出多种数据：
+
+##### a. 透传上游事件
+
+原样转发底层 `POST /api/threads/{thread_id}/runs/stream` 的事件帧（LangGraph 协议），包含 `event:`、`data:` 等行。前端按 LangGraph SDK 既有方式渲染即可。
+
+##### b. 最后一帧：调度状态
+
+流结束前会额外产出一帧 `data:` 行，结构因 `agent_type` 而异：
+
+**`agent_type == "leader"`：**
+
+```
+data: {"status": [["us-intelligence-gathering-agent", "收集美国民众对特朗普的看法"], ["taiwan-intelligence-gathering-agent", "..."]]}
+```
+
+`status` 是一个 `[agent_name, task_text]` 二元组数组，代表协调智能体本轮派出去的任务清单。**前端拿到后应当对每个元素再调用一次本接口（`agent_type="special"`）来执行子智能体**。如果数组为空，说明 leader 觉得已有信息足够，不再派活，可以直接展示 leader 的最终文本回答。
+
+**`agent_type == "special"`：**
+
+```
+data: {"status": "子智能体xxx完成yyy工作"}
+```
+
+代表子智能体完成本轮工作并已把交付内容广播给其他线程。
+
+#### 副作用（重要）
+
+- **leader 调用**：自动把"总控智能体收到用户消息: …"广播到所有非自己的线程；派活后再把"总控智能体给 X 派活, 内容为: …"广播到所有非派活双方的线程
+- **special 调用**：完成后把"子智能体 X 完成 Y 工作, 交付内容为: …"广播到所有非自己的线程
+
+这意味着每个线程的对话历史里会逐步累积出全局上下文，无需前端再做合并。
+
+#### 错误
+
+| 状态码 | 触发条件 |
+|---|---|
+| `400` | `agent_name` 缺失或未在 `d_agent_thread_id` 中 |
+| `401` | 未携带 `Authorization` 头 |
+| `502` | 上游 `runs/stream` 返回非 200 / SSE 解析失败 |
+
+---
+
+## 3. 前端调用流程示例
+
+```ts
+// 1. 拿 token
+const loginResp = await fetch("/api/v1/auth/login/username", {
+  method: "POST",
+  headers: { "Content-Type": "application/json" },
+  body: JSON.stringify({ username: "yourname" }),
+});
+const { access_token } = await loginResp.json();
+const authHeaders = {
+  Authorization: `Bearer ${access_token}`,
+  "Content-Type": "application/json",
+};
+
+// 2. 初始化
+const initResp = await fetch("/api/multi-agent/init", {
+  method: "POST",
+  headers: authHeaders,
+  body: JSON.stringify({
+    agent_names: [
+      "taiwan-intelligence-gathering-agent",
+      "us-intelligence-gathering-agent",
+    ],
+    main_agent_name: "main-controller",
+    model: "deepseek-chat",
+  }),
+});
+const { thread_ids } = await initResp.json();
+
+// 3. 流式调用 leader
+async function streamLeader(message: string, model = "deepseek-chat") {
+  const resp = await fetch("/api/multi-agent/run/stream", {
+    method: "POST",
+    headers: authHeaders,
+    body: JSON.stringify({
+      agent_type: "leader",
+      d_agent_thread_id: thread_ids,
+      agent_name: "main-controller",
+      new_message: message,
+      skill_stop_names: [],
+      model,
+    }),
+  });
+  const reader = resp.body!.getReader();
+  const decoder = new TextDecoder();
+  let buf = "";
+  let dispatched: [string, string][] = [];
+
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    buf += decoder.decode(value, { stream: true });
+    const lines = buf.split("\n");
+    buf = lines.pop() ?? "";
+    for (const line of lines) {
+      if (!line.startsWith("data:")) continue;
+      try {
+        const payload = JSON.parse(line.slice(5).trim());
+        if (Array.isArray(payload.status)) {
+          dispatched = payload.status;
+        }
+        // 其余事件按 LangGraph SDK 协议自行渲染
+      } catch {
+        /* 非 JSON 帧，可忽略或交给现有渲染器 */
+      }
+    }
+  }
+  return dispatched;
+}
+
+// 4. 对派活清单逐个调用 special
+async function streamSpecial(agentName: string, task: string, model = "deepseek-chat") {
+  return fetch("/api/multi-agent/run/stream", {
+    method: "POST",
+    headers: authHeaders,
+    body: JSON.stringify({
+      agent_type: "special",
+      d_agent_thread_id: thread_ids,
+      agent_name: agentName,
+      new_message: task,
+      skill_stop_names: [],
+      model,
+    }),
+  });
+  // 同样按 SSE 解析
+}
+
+// 5. 编排：先 leader 派活 → 依次执行 special → 再让 leader 汇总
+const dispatched = await streamLeader("请分析特朗普的支持率");
+for (const [agentName, task] of dispatched) {
+  await streamSpecial(agentName, task);
+}
+const next = await streamLeader("根据上面收集到的信息出最终结论");
+// 当 leader 返回 status: [] 时，意味着汇报完毕，可结束循环
+```
+
+---
+
+## 4. 注意事项
+
+- **`model` 字段的取值**：必须是后端 `config.yaml` 中 `models[]` 已配置的模型名（如 `deepseek-chat`、`glm-4`、`gpt-4o` 等）。传错会导致上游 502
+- **`thread_ids` 必须每次回传**：后端不存储这个映射，是无状态接口
+- **协调智能体必须装有 `agent_orchestration` 技能**：`/init` 创建时会自动指定 `skills=["agent_orchestration"]`。后端 `skills/` 目录里必须存在该技能定义，否则 leader 无法派活
+- **同名协调智能体冲突**：`main_agent_name` 重名时后端不会报错（"already exists" 走兼容分支），会复用既有智能体——所以同一个用户多次 init 用同一个名字是安全的，但 SOUL 不会被覆盖
+- **SSE 断连**：本接口底层 `on_disconnect: "continue"`，断连后上游 run 会继续跑到结束（不会取消）
+- **并发广播**：广播消息使用 `asyncio.gather` 并发更新所有线程 state，性能上不会随子智能体数线性变慢
+
+---
+
+## 5. 相关接口（底层依赖）
+
+本接口内部转发以下 Gateway 原生接口，了解细节可查：
+
+| 接口 | 路由 | 用途 |
+|---|---|---|
+| 创建 thread | `POST /api/threads` | 为每个智能体建立独立会话 |
+| 读 thread state | `GET /api/threads/{id}/state` | 广播前取当前消息列表 |
+| 写 thread state | `POST /api/threads/{id}/state` | 把广播消息追加到 messages |
+| 取 agent 详情 | `GET /api/agents/{id}` | init 时读取子智能体的 name/description |
+| 建 agent | `POST /api/agents` | init 时创建协调智能体 |
+| 流式 run | `POST /api/threads/{id}/runs/stream` | leader/special 的实际执行 |

newpython/OUT-v1.py

@@ -0,0 +1,483 @@
+"""
+测试流式接口的完整脚本
+API流程：
+1. 登录获取token
+2. 创建thread（可选，也可以使用无状态接口）
+3. 调用流式接口
+
+两种方式：
+- 方式1: POST /api/runs/stream (无状态，自动创建thread)
+- 方式2: POST /api/threads/{thread_id}/runs/stream (需要先创建thread)
+"""
+import requests
+import json
+import asyncio
+import httpx  # 异步 HTTP 客户端
+# ============ 配置区域 ============
+BASE_URL = "http://127.0.0.1:8001"
+LOGIN_URL = f"{BASE_URL}/api/v1/auth/login/username"
+THREADS_URL = f"{BASE_URL}/api/threads"
+RUNS_STREAM_URL = f"{BASE_URL}/api/runs/stream"  # 无状态流式接口
+
+# ============ 登录获取token ============
+def get_token(username: str) -> str:
+    """登录获取token"""
+    response = requests.post(
+        LOGIN_URL,
+        json={"username": username},
+        headers={"Content-Type": "application/json"}
+    )
+    # print(f"登录状态码: {response.status_code}")
+    data = response.json()
+    # print(f"登录响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
+    
+    # 根据实际返回结构调整
+    token = data.get("token") or data.get("access_token") or data.get("data", {}).get("token")
+    return token
+# ============ 方式2: 先创建thread，再调用流式接口 ============
+def create_thread(token: str) -> str:
+    """创建thread，返回thread_id"""
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    
+    response = requests.post(
+        THREADS_URL,
+        json={},  # 空body，使用默认值
+        headers=headers
+    )
+    
+    print(f"创建thread状态码: {response.status_code}")
+    data = response.json()
+    print(f"创建thread响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
+    
+    return data.get("thread_id")
+def get_agent_name_description(agent_name="Taiwan-intelligence-gathering-agent"):
+    url = f"{BASE_URL}/api/agents/{agent_name}"
+    token=get_token("ji")
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    response = requests.get(
+            url,
+            headers=headers,
+        )
+    print(response.json())
+    return response.json()["name"].lower(),"agent_name:"+response.json()["name"].lower()+","+response.json()["description"]
+
+# ============ 创建智能体 ============
+def create_agent(
+    name: str,
+    description: str = "",
+    model: str | None = None,
+    tool_groups: list[str] | None = None,
+    skills: list[str] | None = None,
+    soul: str = "",
+):
+    """
+    创建一个新的智能体
+    
+    Args:
+        name: 智能体名称（必须是 hyphen-case，如 my-agent）
+        description: 智能体描述
+        model: 可选的模型覆盖（如 gpt-4, claude-3-opus 等）
+        tool_groups: 工具组白名单（如 ["web", "file"]）
+        skills: 技能白名单（如 ["skill1", "skill2"]）
+        soul: SOUL.md 内容（智能体人格/身份描述）
+    
+    Returns:
+        创建结果
+    """
+    TOKEN =get_token("ji")
+    url = f"{BASE_URL}/api/agents"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {TOKEN}"
+    }
+    
+    payload = {
+        "name": name,
+        "description": description,
+        "soul": soul
+    }
+    
+    # 添加可选字段
+    if model is not None:
+        payload["model"] = model
+    if tool_groups is not None:
+        payload["tool_groups"] = tool_groups
+    if skills is not None:
+        payload["skills"] = skills
+    
+    print(f"请求: POST {url}")
+    print(f"请求体: {json.dumps(payload, indent=2, ensure_ascii=False)}")
+    
+    response = requests.post(url, json=payload, headers=headers)
+    
+    print(f"状态码: {response.status_code}")
+    
+    if response.status_code == 201:
+        result = response.json()
+        print(f"✓ 智能体创建成功!")
+        print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
+        return result
+    else:
+        print(f"✗ 创建失败: {response.text}")
+        return f"✗ 创建失败: {response.text}"
+
+def create_main_agent(zhihui_name,agent_names):
+    d={}
+    token=get_token("ji")
+    l=[]
+    for i in agent_names:
+        # print(i)
+        name,description=get_agent_name_description(i)
+        thread_id_tmp=create_thread(token)
+        d[name]=thread_id_tmp
+        l.append(description)
+    name="你是总控控制协调智能体,负责协调其他智能体干活"
+    miaoshu=f"""#核心定位：
+    用途：跟用用户的任务目标,调用其他智能体干活,只负责协调汇总,当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果
+    #调度方法：
+    通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活
+    #可调度智能体：
+    {l}"""
+    thread_id_tmp=create_thread(token)
+    engname=zhihui_name
+    d[engname]=thread_id_tmp
+    tmp_skills=["agent_orchestration"]
+    tmpres=create_agent(
+            name=engname,
+            description=name,
+            model="glm-5",  # 指定模型
+            skills=tmp_skills,  # 关联技能
+            soul=miaoshu)
+    # print("创建智能体响应:",tmpres)
+    if "失败" in tmpres and "exists" not in tmpres:
+        return tmpres
+    print(f"创建协调智能体成功,可以协调列表:{l}")
+    return d
+import re
+def extract_name_task(text="python3 scripts/agent_orchestration.py \"Us-intelligence-gathering-agent\" \"收集和分析美国人民对特朗普的看法，包括支持率和反对率、不同群体的观点差异、主要支持理由和反对理由等\""):
+    try:
+        pattern = r'\.py\s+"([^"]*)"\s+"([^"]*)"'
+        match = re.search(pattern, text)
+        if match:
+            agent_name = match.group(1)
+            task_content = match.group(2)
+        return agent_name,task_content
+    except:
+        return "",""
+def update_agent_message(thread_id,new_content):
+    token=get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/state"
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    tmp = requests.get(
+            url,
+            headers=headers
+        ).json()
+    # print(tmp)
+    if "messages" in tmp["values"]:
+         tmp["values"]["messages"].append({"type": "human", "content": new_content})
+    else:
+         tmp["values"]["messages"]=[]
+         tmp["values"]["messages"].append({"type": "human", "content": new_content})
+    url = f"{BASE_URL}/api/threads/{thread_id}/state"
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    response = requests.post(
+            url,
+            json=tmp,
+            headers=headers,
+            stream=True
+        )
+    del tmp
+    print(f"线程{thread_id}更新状态码: {response.status_code}")
+def Broadcast_message(message,d_agent_thread_id,pingbi_agent_name):
+    for key in d_agent_thread_id:
+            if key not in pingbi_agent_name:
+                update_agent_message(d_agent_thread_id[key],message)
+                print("广播消息成功:",key,message)
+null=None
+true=True
+false=False
+async def leader_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names):
+    """使用 httpx 异步请求的 leader_agent"""
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, "总控智能体收到用户消息:" + new_message, d_agent_thread_id, [agent_name])
+    thread_id = d_agent_thread_id[agent_name]
+    l = []
+    token = get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/runs/stream"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    payload = {
+        "input": {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": new_message
+                        }
+                    ],
+                    "additional_kwargs": {}
+                }
+            ]
+        },
+        "config": {
+            "recursion_limit": 1000
+        },
+        "context": {
+            "agent_name": agent_name,
+            "model_name": "deepseek-chat",
+            "mode": "pro",
+            "reasoning_effort": "medium",
+            "thinking_enabled": true,
+            "is_plan_mode": false,
+            "subagent_enabled": false,
+            "thread_id": thread_id
+        },
+        "stream_mode": [
+            "messages-tuple",
+            "values",
+        ],
+        "stream_subgraphs": true,
+        "stream_resumable": true,
+        "assistant_id": "lead_agent",
+        "on_disconnect": "continue",
+        "excluded_tools": ["web_search"],
+        "skill_stop_names": skill_stop_names,
+    }
+    
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("POST", url, json=payload, headers=headers) as response:
+            print(f"状态码: {response.status_code}")
+            if response.status_code != 200:
+                body = await response.aread()
+                raise RuntimeError(
+                    f"upstream {url} returned HTTP {response.status_code}: "
+                    f"{body.decode('utf-8', 'replace')[:1000]}"
+                )
+            async for line in response.aiter_lines():
+                if line:
+                    yield line
+                    if line[:5] == "data:":
+                        print(line)
+                        l.append(line)
+    if len(l) < 2:
+        raise RuntimeError(
+            f"upstream stream produced no usable data lines (got {len(l)} line(s)). "
+            f"Check the server log on {url} for the real failure."
+        )
+    
+    null = None
+    tmpres = eval(l[-2][5:])
+    tmpres = tmpres["messages"][-1]
+    tmp_task = []
+    if tmpres.get("tool_calls", []) == []:
+        yield {"status": []}
+    else:
+        for tool_call in tmpres.get("tool_calls", []):
+            try:
+                agentname, task_text = extract_name_task(str(tool_call["args"]))
+                print("agentname", agentname)
+                print("task_text:", task_text)
+                # 使用 to_thread 异步运行同步的 Broadcast_message
+                await asyncio.to_thread(Broadcast_message, f"总控智能体给{agentname}派活,内容为:" + task_text, d_agent_thread_id, [agent_name, agentname])
+                tmp_task.append([agentname, task_text])
+            except:
+                pass
+        yield {"status": tmp_task}
+
+
+async def special_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names):
+    """使用 httpx 异步请求的 special_agent"""
+    thread_id = d_agent_thread_id[agent_name]
+    l = []
+    token = get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/runs/stream"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    payload = {
+        "input": {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": new_message
+                        }
+                    ],
+                    "additional_kwargs": {}
+                }
+            ]
+        },
+        "config": {
+            "recursion_limit": 1000
+        },
+        "context": {
+            "agent_name": agent_name,
+            "model_name": "deepseek-chat",
+            "mode": "pro",
+            "reasoning_effort": "medium",
+            "thinking_enabled": true,
+            "is_plan_mode": false,
+            "subagent_enabled": false,
+            "thread_id": thread_id
+        },
+        "stream_mode": [
+            "messages-tuple",
+            "values",
+        ],
+        "stream_subgraphs": true,
+        "stream_resumable": true,
+        "assistant_id": "lead_agent",
+        "on_disconnect": "continue",
+        "excluded_tools": ["web_search", "ask_clarification", "present_files", "view_image"],
+        "skill_stop_names": skill_stop_names,
+    }
+    
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("POST", url, json=payload, headers=headers) as response:
+            print(f"状态码: {response.status_code}")
+            if response.status_code != 200:
+                body = await response.aread()
+                raise RuntimeError(
+                    f"upstream {url} returned HTTP {response.status_code}: "
+                    f"{body.decode('utf-8', 'replace')[:1000]}"
+                )
+            async for line in response.aiter_lines():
+                if line:
+                    yield line
+                    if line[:5] == "data:":
+                        print(line)
+                        l.append(line)
+    if len(l) < 2:
+        raise RuntimeError(
+            f"upstream stream produced no usable data lines (got {len(l)} line(s)). "
+            f"Check the server log on {url} for the real failure."
+        )
+    
+    null = None
+    tmpres = eval(l[-2][5:])
+    tmpres = tmpres["messages"][-1]
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + tmpres["content"], d_agent_thread_id, [agent_name])
+    yield {"status": f"子智能体{agent_name}完成{new_message}工作"}
+
+async def agent_init(agent_names, main_agent_name):
+    """异步的 agent_init，同步函数调用使用 to_thread"""
+    agent_names = [i.lower() for i in agent_names]
+    # 使用 to_thread 异步运行同步的 create_main_agent
+    tmpd = await asyncio.to_thread(create_main_agent, main_agent_name, agent_names)
+    message = """
+你的职责:你只是若干个智能体的其中一员,你们的共同目标是完成一个庞大的任务,总控控制协调智能体会分配特定的活给你,你只需要用你自身的能力来完成这个特定的活
+注意:你只是一个子智能体,你不必把你的分析结果存放到文件里面去,你只关心于完成总控控制协调智能体分配的活即可
+"""
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, message, tmpd, [main_agent_name])
+    return main_agent_name, tmpd
+
+
+from fastapi import FastAPI, HTTPException,Body
+from fastapi.responses  import StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+import asyncio
+import uvicorn
+
+app = FastAPI()
+ # CORS 配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+import time
+@app.post("/generate_mul_agent")
+async def generate_stream(payload: dict = Body(...)):
+    """
+    根据 payload 中的 agent_type 字段判断调用 leader_agent 还是 special_agent
+    
+    payload 结构示例:
+    {
+        "agent_type": "leader",  # 或 "special"
+        "d_agent_thread_id": {...},
+        "agent_name": "xxx",
+        "new_message": "xxx",
+        "skill_stop_names": [...]
+    }
+    """
+    # 从 payload 中提取参数
+    agent_type = payload.get("agent_type", "leader")  # 默认为 leader
+    d_agent_thread_id = payload.get("d_agent_thread_id", {})
+    agent_name = payload.get("agent_name", "")
+    new_message = payload.get("new_message", "")
+    skill_stop_names = payload.get("skill_stop_names", [])
+    
+    # 流式响应生成器
+    async def stream_generator():
+        # 根据 agent_type 选择调用的函数
+        if agent_type == "leader":
+            async_generator = leader_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names)
+        else:  # special
+            async_generator = special_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names)
+        
+        # 异步迭代生成器
+        async for chunk in async_generator:
+            if isinstance(chunk, dict):
+                yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
+            else:
+                yield f"{chunk}"
+    
+    # 使用 SSE 流式响应
+    return StreamingResponse(
+        stream_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        }
+    )
+
+
+@app.post("/agent_init")
+async def init_agents(payload: dict = Body(...)):
+    """
+    初始化智能体接口
+    payload 结构示例:
+    {
+        "agent_names": ["agent1", "agent2"],
+        "main_agent_name": "main-agent"
+    }
+    """
+    agent_names = payload.get("agent_names", [])
+    main_agent_name = payload.get("main_agent_name", "main-agent")
+    
+    # 调用异步的 agent_init
+    name, thread_id_dict = await agent_init(agent_names, main_agent_name)
+    
+    return {
+        "status": "success",
+        "main_agent_name": name,
+        "thread_ids": thread_id_dict
+    }
+if __name__ == "__main__": 
+    uvicorn.run("testapi:app",  host="0.0.0.0", port=8095, workers=1,limit_concurrency=1000)  # 多进程提升并发能力 

newpython/OUT.py

@@ -0,0 +1,483 @@
+"""
+测试流式接口的完整脚本
+API流程：
+1. 登录获取token
+2. 创建thread（可选，也可以使用无状态接口）
+3. 调用流式接口
+
+两种方式：
+- 方式1: POST /api/runs/stream (无状态，自动创建thread)
+- 方式2: POST /api/threads/{thread_id}/runs/stream (需要先创建thread)
+"""
+import requests
+import json
+import asyncio
+import httpx  # 异步 HTTP 客户端
+# ============ 配置区域 ============
+BASE_URL = "http://127.0.0.1:8001"
+LOGIN_URL = f"{BASE_URL}/api/v1/auth/login/username"
+THREADS_URL = f"{BASE_URL}/api/threads"
+RUNS_STREAM_URL = f"{BASE_URL}/api/runs/stream"  # 无状态流式接口
+
+# ============ 登录获取token ============
+def get_token(username: str) -> str:
+    """登录获取token"""
+    response = requests.post(
+        LOGIN_URL,
+        json={"username": username},
+        headers={"Content-Type": "application/json"}
+    )
+    # print(f"登录状态码: {response.status_code}")
+    data = response.json()
+    # print(f"登录响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
+    
+    # 根据实际返回结构调整
+    token = data.get("token") or data.get("access_token") or data.get("data", {}).get("token")
+    return token
+# ============ 方式2: 先创建thread，再调用流式接口 ============
+def create_thread(token: str) -> str:
+    """创建thread，返回thread_id"""
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    
+    response = requests.post(
+        THREADS_URL,
+        json={},  # 空body，使用默认值
+        headers=headers
+    )
+    
+    print(f"创建thread状态码: {response.status_code}")
+    data = response.json()
+    print(f"创建thread响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
+    
+    return data.get("thread_id")
+def get_agent_name_description(agent_name="Taiwan-intelligence-gathering-agent"):
+    url = f"{BASE_URL}/api/agents/{agent_name}"
+    token=get_token("ji")
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    response = requests.get(
+            url,
+            headers=headers,
+        )
+    print(response.json())
+    return response.json()["name"].lower(),"agent_name:"+response.json()["name"].lower()+","+response.json()["description"]
+
+# ============ 创建智能体 ============
+def create_agent(
+    name: str,
+    description: str = "",
+    model: str | None = None,
+    tool_groups: list[str] | None = None,
+    skills: list[str] | None = None,
+    soul: str = "",
+):
+    """
+    创建一个新的智能体
+    
+    Args:
+        name: 智能体名称（必须是 hyphen-case，如 my-agent）
+        description: 智能体描述
+        model: 可选的模型覆盖（如 gpt-4, claude-3-opus 等）
+        tool_groups: 工具组白名单（如 ["web", "file"]）
+        skills: 技能白名单（如 ["skill1", "skill2"]）
+        soul: SOUL.md 内容（智能体人格/身份描述）
+    
+    Returns:
+        创建结果
+    """
+    TOKEN =get_token("ji")
+    url = f"{BASE_URL}/api/agents"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {TOKEN}"
+    }
+    
+    payload = {
+        "name": name,
+        "description": description,
+        "soul": soul
+    }
+    
+    # 添加可选字段
+    if model is not None:
+        payload["model"] = model
+    if tool_groups is not None:
+        payload["tool_groups"] = tool_groups
+    if skills is not None:
+        payload["skills"] = skills
+    
+    print(f"请求: POST {url}")
+    print(f"请求体: {json.dumps(payload, indent=2, ensure_ascii=False)}")
+    
+    response = requests.post(url, json=payload, headers=headers)
+    
+    print(f"状态码: {response.status_code}")
+    
+    if response.status_code == 201:
+        result = response.json()
+        print(f"✓ 智能体创建成功!")
+        print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
+        return result
+    else:
+        print(f"✗ 创建失败: {response.text}")
+        return f"✗ 创建失败: {response.text}"
+
+def create_main_agent(zhihui_name,agent_names):
+    d={}
+    token=get_token("ji")
+    l=[]
+    for i in agent_names:
+        # print(i)
+        name,description=get_agent_name_description(i)
+        thread_id_tmp=create_thread(token)
+        d[name]=thread_id_tmp
+        l.append(description)
+    name="你是总控控制协调智能体,负责协调其他智能体干活"
+    miaoshu=f"""#核心定位：
+    用途：跟用用户的任务目标,调用其他智能体干活,只负责协调汇总,当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果
+    #调度方法：
+    通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活
+    #可调度智能体：
+    {l}"""
+    thread_id_tmp=create_thread(token)
+    engname=zhihui_name
+    d[engname]=thread_id_tmp
+    tmp_skills=["agent_orchestration"]
+    tmpres=create_agent(
+            name=engname,
+            description=name,
+            model="glm-5",  # 指定模型
+            skills=tmp_skills,  # 关联技能
+            soul=miaoshu)
+    # print("创建智能体响应:",tmpres)
+    if "失败" in tmpres and "exists" not in tmpres:
+        return tmpres
+    print(f"创建协调智能体成功,可以协调列表:{l}")
+    return d
+import re
+def extract_name_task(text="python3 scripts/agent_orchestration.py \"Us-intelligence-gathering-agent\" \"收集和分析美国人民对特朗普的看法，包括支持率和反对率、不同群体的观点差异、主要支持理由和反对理由等\""):
+    try:
+        pattern = r'\.py\s+"([^"]*)"\s+"([^"]*)"'
+        match = re.search(pattern, text)
+        if match:
+            agent_name = match.group(1)
+            task_content = match.group(2)
+        return agent_name,task_content
+    except:
+        return "",""
+def update_agent_message(thread_id,new_content):
+    token=get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/state"
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    tmp = requests.get(
+            url,
+            headers=headers
+        ).json()
+    # print(tmp)
+    if "messages" in tmp["values"]:
+         tmp["values"]["messages"].append({"type": "human", "content": new_content})
+    else:
+         tmp["values"]["messages"]=[]
+         tmp["values"]["messages"].append({"type": "human", "content": new_content})
+    url = f"{BASE_URL}/api/threads/{thread_id}/state"
+    headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}"
+        }
+    response = requests.post(
+            url,
+            json=tmp,
+            headers=headers,
+            stream=True
+        )
+    del tmp
+    print(f"线程{thread_id}更新状态码: {response.status_code}")
+def Broadcast_message(message,d_agent_thread_id,pingbi_agent_name):
+    for key in d_agent_thread_id:
+            if key not in pingbi_agent_name:
+                update_agent_message(d_agent_thread_id[key],message)
+                print("广播消息成功:",key,message)
+null=None
+true=True
+false=False
+async def leader_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names):
+    """使用 httpx 异步请求的 leader_agent"""
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, "总控智能体收到用户消息:" + new_message, d_agent_thread_id, [agent_name])
+    thread_id = d_agent_thread_id[agent_name]
+    l = []
+    token = get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/runs/stream"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    payload = {
+        "input": {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": new_message
+                        }
+                    ],
+                    "additional_kwargs": {}
+                }
+            ]
+        },
+        "config": {
+            "recursion_limit": 1000
+        },
+        "context": {
+            "agent_name": agent_name,
+            "model_name": "deepseek-chat",
+            "mode": "pro",
+            "reasoning_effort": "medium",
+            "thinking_enabled": true,
+            "is_plan_mode": false,
+            "subagent_enabled": false,
+            "thread_id": thread_id
+        },
+        "stream_mode": [
+            "messages-tuple",
+            "values",
+        ],
+        "stream_subgraphs": true,
+        "stream_resumable": true,
+        "assistant_id": "lead_agent",
+        "on_disconnect": "continue",
+        "excluded_tools": ["web_search"],
+        "skill_stop_names": skill_stop_names,
+    }
+    
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("POST", url, json=payload, headers=headers) as response:
+            print(f"状态码: {response.status_code}")
+            if response.status_code != 200:
+                body = await response.aread()
+                raise RuntimeError(
+                    f"upstream {url} returned HTTP {response.status_code}: "
+                    f"{body.decode('utf-8', 'replace')[:1000]}"
+                )
+            async for line in response.aiter_lines():
+                if line:
+                    yield line
+                    if line[:5] == "data:":
+                        print(line)
+                        l.append(line)
+    if len(l) < 2:
+        raise RuntimeError(
+            f"upstream stream produced no usable data lines (got {len(l)} line(s)). "
+            f"Check the server log on {url} for the real failure."
+        )
+    
+    null = None
+    tmpres = eval(l[-2][5:])
+    tmpres = tmpres["messages"][-1]
+    tmp_task = []
+    if tmpres.get("tool_calls", []) == []:
+        yield {"status": []}
+    else:
+        for tool_call in tmpres.get("tool_calls", []):
+            try:
+                agentname, task_text = extract_name_task(str(tool_call["args"]))
+                print("agentname", agentname)
+                print("task_text:", task_text)
+                # 使用 to_thread 异步运行同步的 Broadcast_message
+                await asyncio.to_thread(Broadcast_message, f"总控智能体给{agentname}派活,内容为:" + task_text, d_agent_thread_id, [agent_name, agentname])
+                tmp_task.append([agentname, task_text])
+            except:
+                pass
+        yield {"status": tmp_task}
+
+
+async def special_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names):
+    """使用 httpx 异步请求的 special_agent"""
+    thread_id = d_agent_thread_id[agent_name]
+    l = []
+    token = get_token("ji")
+    url = f"{BASE_URL}/api/threads/{thread_id}/runs/stream"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}"
+    }
+    payload = {
+        "input": {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": new_message
+                        }
+                    ],
+                    "additional_kwargs": {}
+                }
+            ]
+        },
+        "config": {
+            "recursion_limit": 1000
+        },
+        "context": {
+            "agent_name": agent_name,
+            "model_name": "deepseek-chat",
+            "mode": "pro",
+            "reasoning_effort": "medium",
+            "thinking_enabled": true,
+            "is_plan_mode": false,
+            "subagent_enabled": false,
+            "thread_id": thread_id
+        },
+        "stream_mode": [
+            "messages-tuple",
+            "values",
+        ],
+        "stream_subgraphs": true,
+        "stream_resumable": true,
+        "assistant_id": "lead_agent",
+        "on_disconnect": "continue",
+        "excluded_tools": ["web_search", "ask_clarification", "present_files", "view_image"],
+        "skill_stop_names": skill_stop_names,
+    }
+    
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("POST", url, json=payload, headers=headers) as response:
+            print(f"状态码: {response.status_code}")
+            if response.status_code != 200:
+                body = await response.aread()
+                raise RuntimeError(
+                    f"upstream {url} returned HTTP {response.status_code}: "
+                    f"{body.decode('utf-8', 'replace')[:1000]}"
+                )
+            async for line in response.aiter_lines():
+                if line:
+                    yield line
+                    if line[:5] == "data:":
+                        print(line)
+                        l.append(line)
+    if len(l) < 2:
+        raise RuntimeError(
+            f"upstream stream produced no usable data lines (got {len(l)} line(s)). "
+            f"Check the server log on {url} for the real failure."
+        )
+    
+    null = None
+    tmpres = eval(l[-2][5:])
+    tmpres = tmpres["messages"][-1]
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + tmpres["content"], d_agent_thread_id, [agent_name])
+    yield {"status": f"子智能体{agent_name}完成{new_message}工作"}
+
+async def agent_init(agent_names, main_agent_name):
+    """异步的 agent_init，同步函数调用使用 to_thread"""
+    agent_names = [i.lower() for i in agent_names]
+    # 使用 to_thread 异步运行同步的 create_main_agent
+    tmpd = await asyncio.to_thread(create_main_agent, main_agent_name, agent_names)
+    message = """
+你的职责:你只是若干个智能体的其中一员,你们的共同目标是完成一个庞大的任务,总控控制协调智能体会分配特定的活给你,你只需要用你自身的能力来完成这个特定的活
+注意:你只是一个子智能体,你不必把你的分析结果存放到文件里面去,你只关心于完成总控控制协调智能体分配的活即可
+"""
+    # 使用 to_thread 异步运行同步的 Broadcast_message
+    await asyncio.to_thread(Broadcast_message, message, tmpd, [main_agent_name])
+    return main_agent_name, tmpd
+
+
+from fastapi import FastAPI, HTTPException,Body
+from fastapi.responses  import StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+import asyncio
+import uvicorn
+
+app = FastAPI()
+ # CORS 配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+import time
+@app.post("/generate_mul_agent")
+async def generate_stream(payload: dict = Body(...)):
+    """
+    根据 payload 中的 agent_type 字段判断调用 leader_agent 还是 special_agent
+    
+    payload 结构示例:
+    {
+        "agent_type": "leader",  # 或 "special"
+        "d_agent_thread_id": {...},
+        "agent_name": "xxx",
+        "new_message": "xxx",
+        "skill_stop_names": [...]
+    }
+    """
+    # 从 payload 中提取参数
+    agent_type = payload.get("agent_type", "leader")  # 默认为 leader
+    d_agent_thread_id = payload.get("d_agent_thread_id", {})
+    agent_name = payload.get("agent_name", "")
+    new_message = payload.get("new_message", "")
+    skill_stop_names = payload.get("skill_stop_names", [])
+    
+    # 流式响应生成器
+    async def stream_generator():
+        # 根据 agent_type 选择调用的函数
+        if agent_type == "leader":
+            async_generator = leader_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names)
+        else:  # special
+            async_generator = special_agent(d_agent_thread_id, agent_name, new_message, skill_stop_names)
+        
+        # 异步迭代生成器
+        async for chunk in async_generator:
+            if isinstance(chunk, dict):
+                yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
+            else:
+                yield f"{chunk}"
+    
+    # 使用 SSE 流式响应
+    return StreamingResponse(
+        stream_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        }
+    )
+
+
+@app.post("/agent_init")
+async def init_agents(payload: dict = Body(...)):
+    """
+    初始化智能体接口
+    payload 结构示例:
+    {
+        "agent_names": ["agent1", "agent2"],
+        "main_agent_name": "main-agent"
+    }
+    """
+    agent_names = payload.get("agent_names", [])
+    main_agent_name = payload.get("main_agent_name", "main-agent")
+    
+    # 调用异步的 agent_init
+    name, thread_id_dict = await agent_init(agent_names, main_agent_name)
+    
+    return {
+        "status": "success",
+        "main_agent_name": name,
+        "thread_ids": thread_id_dict
+    }
+if __name__ == "__main__": 
+    uvicorn.run("testapi:app",  host="0.0.0.0", port=8095, workers=1,limit_concurrency=1000)  # 多进程提升并发能力 

newpython/demo_call.py

@@ -0,0 +1,132 @@
+"""
+OUT.py 的调用链 demo —— 跑通 agent_init → leader 派活 → 各 special agent 干活 这一整套流程。
+
+前置条件:
+  1. DeerFlow 后端在 http://127.0.0.1:8001 在跑
+  2. OUT.py 已经 `python -m uvicorn OUT:app --host 0.0.0.0 --port 8095` 起好
+  3. 下面 SUB_AGENT_NAMES 里的 agent 已经在 DeerFlow 里创建好(/api/agents 能查到)
+
+直接 `python demo_call.py` 即可。
+"""
+import asyncio
+import json
+
+import httpx
+
+# ============ 改这里 ============
+OUT_BASE = "http://127.0.0.1:8095"
+
+# 已经在 DeerFlow 里建好的子 agent 名字 (会被 OUT.py 强制 lowercase)
+SUB_AGENT_NAMES = [
+    "us-intelligence-gathering-agent",
+    "taiwan-intelligence-gathering-agent",
+]
+
+# 总控 agent 名字 —— OUT.py 会现场创建它,所以填一个新名字即可 (hyphen-case)
+MAIN_AGENT_NAME = "demo-coordinator"
+
+# 用户提问 —— 总控会基于此派活
+USER_QUESTION = "对比中美两国 2024 年在 AI 监管政策上的差异"
+
+# leader 阶段命中即停的 skill 名字单 (OUT.py 用 agent_orchestration 作为派活信号)
+SKILL_STOP_NAMES = ["agent_orchestration"]
+# ================================
+
+
+async def call_agent_init() -> tuple[str, dict[str, str]]:
+    """步骤 1: 初始化 —— 拿回 thread_ids 字典 {agent_name → thread_id}."""
+    print(">>> [1/3] POST /agent_init")
+    async with httpx.AsyncClient(timeout=300.0) as client:
+        r = await client.post(
+            f"{OUT_BASE}/agent_init",
+            json={
+                "agent_names": SUB_AGENT_NAMES,
+                "main_agent_name": MAIN_AGENT_NAME,
+            },
+        )
+        r.raise_for_status()
+        body = r.json()
+    main_name = body["main_agent_name"]
+    thread_ids = body["thread_ids"]
+    print(f"    main_agent_name = {main_name}")
+    print(f"    thread_ids = {json.dumps(thread_ids, ensure_ascii=False, indent=2)}")
+    return main_name, thread_ids
+
+
+async def stream_generate(payload: dict) -> object | None:
+    """通用 SSE 消费:边收边打印,最后返回 OUT.py 自己 yield 的 {'status': ...} 里的 status 字段."""
+    final_status = None
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("POST", f"{OUT_BASE}/generate_mul_agent", json=payload) as resp:
+            print(f"    HTTP {resp.status_code}")
+            if resp.status_code != 200:
+                err = await resp.aread()
+                raise RuntimeError(f"OUT.py /generate_mul_agent failed: {err.decode('utf-8', 'replace')[:500]}")
+
+            async for line in resp.aiter_lines():
+                if not line:
+                    continue
+                # OUT.py 把 DeerFlow 的原始 SSE 流直接转发,自己最后会再 yield 一行
+                # `data: {"status": ...}\n\n`。两边都以 `data:` 开头,我们都尝试 JSON parse
+                # 一下,能解析成 {"status": ...} 的就是 OUT.py 的终态。
+                if line.startswith("data:"):
+                    body_text = line[5:].strip()
+                    try:
+                        obj = json.loads(body_text)
+                    except json.JSONDecodeError:
+                        continue
+                    if isinstance(obj, dict) and "status" in obj:
+                        final_status = obj["status"]
+                # 想看上游 DeerFlow 的完整流,把下面打开即可
+                # else:
+                #     print("   raw:", line[:200])
+    return final_status
+
+
+async def call_leader(main_agent_name: str, thread_ids: dict[str, str]) -> list[list[str]]:
+    """步骤 2: leader 阶段 —— 总控读完用户问题,产出派活清单."""
+    print(f"\n>>> [2/3] POST /generate_mul_agent (leader, agent={main_agent_name})")
+    status = await stream_generate(
+        {
+            "agent_type": "leader",
+            "d_agent_thread_id": thread_ids,
+            "agent_name": main_agent_name,
+            "new_message": USER_QUESTION,
+            "skill_stop_names": SKILL_STOP_NAMES,
+        }
+    )
+    dispatch = status or []
+    print(f"    派活清单 = {json.dumps(dispatch, ensure_ascii=False, indent=2)}")
+    return dispatch
+
+
+async def call_specials(thread_ids: dict[str, str], dispatch: list[list[str]]) -> None:
+    """步骤 3: 把派活清单里每一项交给对应子 agent 跑."""
+    if not dispatch:
+        print("\n>>> [3/3] 总控没派出任何任务,直接结束。")
+        return
+    for idx, item in enumerate(dispatch, 1):
+        agent_name, task_text = item[0], item[1]
+        print(f"\n>>> [3/3] ({idx}/{len(dispatch)}) POST /generate_mul_agent (special, agent={agent_name})")
+        print(f"    task = {task_text[:120]}{'...' if len(task_text) > 120 else ''}")
+        result = await stream_generate(
+            {
+                "agent_type": "special",
+                "d_agent_thread_id": thread_ids,
+                "agent_name": agent_name,
+                "new_message": task_text,
+                "skill_stop_names": [],
+            }
+        )
+        print(f"    完成: {result}")
+
+
+async def main() -> None:
+    main_agent_name, thread_ids = await call_agent_init()
+    dispatch = await call_leader(main_agent_name, thread_ids)
+    await call_specials(thread_ids, dispatch)
+    print("\n✅ demo 跑完")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

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

@@ -1,10 +1,10 @@
-import asyncio
-import logging
-import os
-import queue
-import sys
-import threading
-from collections.abc import AsyncGenerator
+import asyncio
+import logging
+import os
+import queue
+import sys
+import threading
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 from dotenv import load_dotenv
@@ -36,6 +36,7 @@ from app.gateway.routers import (
     mcp,
     memory,
     models,
+    multi_agent,
     notifications,
     public_scheduled_tasks,
     recommended_questions,
@@ -247,7 +248,7 @@ async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
 
 async def _run_db_migrations() -> None:
     """Run Alembic migrations at startup so schema changes are applied automatically."""
-    from pathlib import Path
+    from pathlib import Path
 
     from alembic import command
     from alembic.config import Config
@@ -260,51 +261,51 @@ async def _run_db_migrations() -> None:
         logger.info("Persistence backend=memory — skipping Alembic migrations")
         return
 
-    if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
-        logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
-        return
-
-    db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
-    migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
-    result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)
-
-    def _run_sync() -> None:
-        try:
-            cfg = Config()
-            cfg.set_main_option("script_location", migrations_dir)
-            cfg.set_main_option("sqlalchemy.url", db_url)
-            command.upgrade(cfg, "head")
-        except BaseException as exc:
-            result_queue.put(exc)
-        else:
-            result_queue.put(None)
-
-    try:
-        try:
-            timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
-        except ValueError:
-            timeout_s = 60.0
-
-        worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
-        worker.start()
-        try:
-            deadline = asyncio.get_running_loop().time() + timeout_s
-            while result_queue.empty():
-                if asyncio.get_running_loop().time() >= deadline:
-                    raise asyncio.TimeoutError
-                await asyncio.sleep(0.2)
-            result = result_queue.get_nowait()
-        except asyncio.TimeoutError:
-            logger.error(
-                "Database migration did not finish within %.0fs; continuing startup on create_all schema. "
-                "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
-                "to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
-                timeout_s,
-            )
-            return
-        if result is not None:
-            raise result
-        logger.info("Database migrations applied (head)")
+    if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
+        logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
+        return
+
+    db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
+    migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
+    result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)
+
+    def _run_sync() -> None:
+        try:
+            cfg = Config()
+            cfg.set_main_option("script_location", migrations_dir)
+            cfg.set_main_option("sqlalchemy.url", db_url)
+            command.upgrade(cfg, "head")
+        except BaseException as exc:
+            result_queue.put(exc)
+        else:
+            result_queue.put(None)
+
+    try:
+        try:
+            timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
+        except ValueError:
+            timeout_s = 60.0
+
+        worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
+        worker.start()
+        try:
+            deadline = asyncio.get_running_loop().time() + timeout_s
+            while result_queue.empty():
+                if asyncio.get_running_loop().time() >= deadline:
+                    raise asyncio.TimeoutError
+                await asyncio.sleep(0.2)
+            result = result_queue.get_nowait()
+        except asyncio.TimeoutError:
+            logger.error(
+                "Database migration did not finish within %.0fs; continuing startup on create_all schema. "
+                "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
+                "to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
+                timeout_s,
+            )
+            return
+        if result is not None:
+            raise result
+        logger.info("Database migrations applied (head)")
     except Exception:
         logger.exception("Database migration failed — schema may be out of date")
 
@@ -860,6 +861,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Stateless Runs API (stream/wait without a pre-existing thread)
     app.include_router(runs.router)
 
+    # Multi-agent orchestration (coordinator + sub-agent broadcast/dispatch)
+    app.include_router(multi_agent.router)
+
     @app.get("/health", tags=["health"])
     async def health_check() -> dict:
         """Health check endpoint.

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

@@ -0,0 +1,440 @@
+"""Multi-agent orchestration endpoints.
+
+Wraps the legacy ``newpython/OUT.py`` demo into the Gateway. Provides:
+
+* ``POST /api/multi-agent/init`` — create a coordinator agent and per-sub-agent
+  threads. Returns the ``{agent_name: thread_id}`` map the client passes back
+  on subsequent calls.
+* ``POST /api/multi-agent/run/stream`` — drive a *leader* or *special* agent
+  via SSE. The leader dispatches work to sub-agents through the
+  ``agent_orchestration`` skill; sub-agents execute and broadcast their result.
+
+Internal calls loop back to the same Gateway over HTTP, reusing the caller's
+``Authorization`` header — so the demo's hard-coded ``get_token("ji")`` is gone.
+The model used by both the coordinator agent and every streamed run is taken
+from the request's ``model`` field.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import re
+from typing import Any, AsyncIterator
+
+import httpx
+from fastapi import APIRouter, Body, HTTPException, Request
+from fastapi.responses import StreamingResponse
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])
+
+
+# ---------------------------------------------------------------------------
+# Loopback helpers
+# ---------------------------------------------------------------------------
+
+
+def _loopback_base(request: Request) -> str:
+    return f"{request.url.scheme}://{request.url.netloc}"
+
+
+def _auth_headers(request: Request) -> dict[str, str]:
+    token = request.headers.get("authorization")
+    if not token:
+        raise HTTPException(status_code=401, detail="Missing Authorization header")
+    return {"Authorization": token, "Content-Type": "application/json"}
+
+
+async def _create_thread(client: httpx.AsyncClient, base: str, headers: dict[str, str]) -> str:
+    resp = await client.post(f"{base}/api/threads", json={}, headers=headers)
+    resp.raise_for_status()
+    return resp.json()["thread_id"]
+
+
+async def _get_agent(client: httpx.AsyncClient, base: str, headers: dict[str, str], agent_id: str) -> dict[str, Any]:
+    resp = await client.get(f"{base}/api/agents/{agent_id}", headers=headers)
+    resp.raise_for_status()
+    return resp.json()
+
+
+async def _create_agent(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    *,
+    name: str,
+    description: str,
+    model: str | None,
+    skills: list[str] | None,
+    soul: str,
+) -> tuple[bool, dict[str, Any] | str]:
+    payload: dict[str, Any] = {"name": name, "description": description, "soul": soul}
+    if model is not None:
+        payload["model"] = model
+    if skills is not None:
+        payload["skills"] = skills
+    resp = await client.post(f"{base}/api/agents", json=payload, headers=headers)
+    if resp.status_code == 201:
+        return True, resp.json()
+    return False, resp.text
+
+
+async def _append_thread_message(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    thread_id: str,
+    content: str,
+) -> None:
+    state_url = f"{base}/api/threads/{thread_id}/state"
+    get_resp = await client.get(state_url, headers=headers)
+    if get_resp.status_code != 200:
+        logger.warning("get state %s failed: %s %s", thread_id, get_resp.status_code, get_resp.text[:200])
+        return
+    state = get_resp.json()
+    values = state.setdefault("values", {})
+    messages = values.setdefault("messages", [])
+    if not isinstance(messages, list):
+        messages = []
+        values["messages"] = messages
+    messages.append({"type": "human", "content": content})
+    post_resp = await client.post(state_url, json=state, headers=headers)
+    if post_resp.status_code >= 400:
+        logger.warning("update state %s failed: %s %s", thread_id, post_resp.status_code, post_resp.text[:200])
+
+
+async def _broadcast(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    message: str,
+    agent_threads: dict[str, str],
+    skip: list[str],
+) -> None:
+    targets = [tid for name, tid in agent_threads.items() if name not in skip]
+    if not targets:
+        return
+    await asyncio.gather(
+        *(_append_thread_message(client, base, headers, tid, message) for tid in targets),
+        return_exceptions=True,
+    )
+
+
+_TOOL_CALL_PATTERN = re.compile(r'\.py\s+"([^"]*)"\s+"([^"]*)"')
+
+
+def _extract_name_task(text: str) -> tuple[str, str]:
+    match = _TOOL_CALL_PATTERN.search(text)
+    if match:
+        return match.group(1), match.group(2)
+    return "", ""
+
+
+def _flatten_content(content: Any) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for item in content:
+            if isinstance(item, dict):
+                parts.append(str(item.get("text", "")))
+            else:
+                parts.append(str(item))
+        return "".join(parts)
+    return str(content)
+
+
+def _run_payload(
+    *,
+    agent_name: str,
+    model_name: str,
+    thread_id: str,
+    new_message: str,
+    excluded_tools: list[str],
+    skill_stop_names: list[str],
+) -> dict[str, Any]:
+    return {
+        "input": {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": [{"type": "text", "text": new_message}],
+                    "additional_kwargs": {},
+                }
+            ]
+        },
+        "config": {"recursion_limit": 1000},
+        "context": {
+            "agent_name": agent_name,
+            "model_name": model_name,
+            "mode": "pro",
+            "reasoning_effort": "medium",
+            "thinking_enabled": True,
+            "is_plan_mode": False,
+            "subagent_enabled": False,
+            "thread_id": thread_id,
+        },
+        "stream_mode": ["messages-tuple", "values"],
+        "stream_subgraphs": True,
+        "stream_resumable": True,
+        "assistant_id": "lead_agent",
+        "on_disconnect": "continue",
+        "excluded_tools": excluded_tools,
+        "skill_stop_names": skill_stop_names,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Init endpoint
+# ---------------------------------------------------------------------------
+
+
+@router.post("/init")
+async def init_multi_agent(request: Request, payload: dict = Body(...)) -> dict[str, Any]:
+    agent_names_in = payload.get("agent_names") or []
+    agent_names: list[str] = [str(name).lower() for name in agent_names_in]
+    main_agent_name: str = str(payload.get("main_agent_name") or "main-agent").lower()
+    model: str | None = payload.get("model")
+
+    if not agent_names:
+        raise HTTPException(status_code=400, detail="agent_names is required")
+
+    base = _loopback_base(request)
+    headers = _auth_headers(request)
+
+    async with httpx.AsyncClient(timeout=None) as client:
+        agent_threads: dict[str, str] = {}
+        descriptions: list[str] = []
+        for sub_id in agent_names:
+            info = await _get_agent(client, base, headers, sub_id)
+            sub_name = str(info["name"]).lower()
+            agent_threads[sub_name] = await _create_thread(client, base, headers)
+            descriptions.append(f"agent_name:{sub_name},{info.get('description', '')}")
+
+        soul = (
+            "#核心定位：\n"
+            "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
+            "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
+            "#调度方法：\n"
+            "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活\n"
+            "#可调度智能体：\n"
+            f"{descriptions}"
+        )
+
+        agent_threads[main_agent_name] = await _create_thread(client, base, headers)
+        ok, result = await _create_agent(
+            client,
+            base,
+            headers,
+            name=main_agent_name,
+            description="你是总控控制协调智能体,负责协调其他智能体干活",
+            model=model,
+            skills=["agent_orchestration"],
+            soul=soul,
+        )
+        if not ok and isinstance(result, str) and "exists" not in result:
+            raise HTTPException(status_code=500, detail=f"create coordinator agent failed: {result}")
+
+        intro = (
+            "你的职责:你只是若干个智能体的其中一员,你们的共同目标是完成一个庞大的任务,"
+            "总控控制协调智能体会分配特定的活给你,你只需要用你自身的能力来完成这个特定的活\n"
+            "注意:你只是一个子智能体,你不必把你的分析结果存放到文件里面去,"
+            "你只关心于完成总控控制协调智能体分配的活即可"
+        )
+        await _broadcast(client, base, headers, intro, agent_threads, [main_agent_name])
+
+    return {
+        "status": "success",
+        "main_agent_name": main_agent_name,
+        "thread_ids": agent_threads,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Stream endpoint
+# ---------------------------------------------------------------------------
+
+
+async def _stream_upstream(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    thread_id: str,
+    body: dict[str, Any],
+) -> AsyncIterator[tuple[str | None, list[str] | None]]:
+    """Yield (raw_line, None) for each SSE line, then ``(None, captured_data_lines)`` once."""
+
+    url = f"{base}/api/threads/{thread_id}/runs/stream"
+    captured: list[str] = []
+    async with client.stream("POST", url, json=body, headers=headers) as resp:
+        if resp.status_code != 200:
+            body_text = (await resp.aread()).decode("utf-8", "replace")[:1000]
+            raise HTTPException(
+                status_code=502,
+                detail=f"upstream {url} returned HTTP {resp.status_code}: {body_text}",
+            )
+        async for line in resp.aiter_lines():
+            if not line:
+                continue
+            yield line, None
+            if line.startswith("data:"):
+                captured.append(line)
+    yield None, captured
+
+
+def _parse_last_message(captured: list[str]) -> dict[str, Any]:
+    if len(captured) < 2:
+        raise HTTPException(
+            status_code=502,
+            detail=f"upstream produced no usable data lines (got {len(captured)})",
+        )
+    raw = captured[-2][5:].strip()
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise HTTPException(status_code=502, detail=f"failed to parse upstream frame: {exc}") from exc
+    messages = parsed.get("messages") or []
+    if not messages:
+        return {}
+    return messages[-1] or {}
+
+
+async def _leader_run(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    agent_threads: dict[str, str],
+    agent_name: str,
+    new_message: str,
+    skill_stop_names: list[str],
+    model_name: str,
+) -> AsyncIterator[str | dict[str, Any]]:
+    await _broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name])
+    thread_id = agent_threads[agent_name]
+    body = _run_payload(
+        agent_name=agent_name,
+        model_name=model_name,
+        thread_id=thread_id,
+        new_message=new_message,
+        excluded_tools=["web_search"],
+        skill_stop_names=skill_stop_names,
+    )
+
+    captured: list[str] = []
+    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
+        if line is not None:
+            yield line
+        elif done is not None:
+            captured = done
+
+    last_msg = _parse_last_message(captured)
+    tool_calls = last_msg.get("tool_calls") or []
+    if not tool_calls:
+        yield {"status": []}
+        return
+
+    dispatched: list[list[str]] = []
+    for tool_call in tool_calls:
+        try:
+            sub_name, task_text = _extract_name_task(str(tool_call.get("args", "")))
+            if not sub_name:
+                continue
+            await _broadcast(
+                client,
+                base,
+                headers,
+                f"总控智能体给{sub_name}派活,内容为:" + task_text,
+                agent_threads,
+                [agent_name, sub_name],
+            )
+            dispatched.append([sub_name, task_text])
+        except Exception as exc:
+            logger.warning("dispatch failed: %s", exc)
+    yield {"status": dispatched}
+
+
+async def _special_run(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    agent_threads: dict[str, str],
+    agent_name: str,
+    new_message: str,
+    skill_stop_names: list[str],
+    model_name: str,
+) -> AsyncIterator[str | dict[str, Any]]:
+    thread_id = agent_threads[agent_name]
+    body = _run_payload(
+        agent_name=agent_name,
+        model_name=model_name,
+        thread_id=thread_id,
+        new_message=new_message,
+        excluded_tools=["web_search", "ask_clarification", "present_files", "view_image"],
+        skill_stop_names=skill_stop_names,
+    )
+
+    captured: list[str] = []
+    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
+        if line is not None:
+            yield line
+        elif done is not None:
+            captured = done
+
+    last_msg = _parse_last_message(captured)
+    content = _flatten_content(last_msg.get("content", ""))
+    await _broadcast(
+        client,
+        base,
+        headers,
+        f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + content,
+        agent_threads,
+        [agent_name],
+    )
+    yield {"status": f"子智能体{agent_name}完成{new_message}工作"}
+
+
+@router.post("/run/stream")
+async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) -> StreamingResponse:
+    agent_type: str = str(payload.get("agent_type") or "leader").lower()
+    agent_threads: dict[str, str] = payload.get("d_agent_thread_id") or {}
+    agent_name: str = str(payload.get("agent_name") or "").lower()
+    new_message: str = str(payload.get("new_message") or "")
+    skill_stop_names: list[str] = payload.get("skill_stop_names") or []
+    model_name: str = str(payload.get("model") or "deepseek-chat")
+
+    if not agent_name or agent_name not in agent_threads:
+        raise HTTPException(
+            status_code=400,
+            detail=f"agent_name '{agent_name}' is missing from d_agent_thread_id",
+        )
+
+    base = _loopback_base(request)
+    headers = _auth_headers(request)
+
+    async def stream_generator() -> AsyncIterator[str]:
+        async with httpx.AsyncClient(timeout=None) as client:
+            if agent_type == "leader":
+                gen = _leader_run(
+                    client, base, headers, agent_threads, agent_name, new_message, skill_stop_names, model_name,
+                )
+            else:
+                gen = _special_run(
+                    client, base, headers, agent_threads, agent_name, new_message, skill_stop_names, model_name,
+                )
+            async for chunk in gen:
+                if isinstance(chunk, dict):
+                    yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
+                else:
+                    yield chunk if chunk.endswith("\n") else chunk + "\n"
+
+    return StreamingResponse(
+        stream_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )

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

@@ -54,6 +54,9 @@ class RunCreateRequest(BaseModel):
     after_seconds: float | None = Field(default=None, description="Delayed execution")
     if_not_exists: Literal["reject", "create"] = Field(default="create", description="Thread creation policy")
     feedback_keys: list[str] | None = Field(default=None, description="LangSmith feedback keys")
+    excluded_tools: list[str] | None = Field(default=None, description="Tool names to exclude from the main agent (e.g. ['task', 'ask_clarification'])")
+    skill_stop_names: list[str] | None = Field(default=None, description="Skill names to stop execution when invoked (e.g. ['newsletter-generation'])")
+    skill_stop_message: str | None = Field(default=None, description="Custom message to display when skill stop is triggered")
 
 
 class RunResponse(BaseModel):

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

@@ -162,6 +162,9 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
         "max_concurrent_subagents",
         "agent_name",
         "is_bootstrap",
+        "excluded_tools",
+        "skill_stop_names",
+        "skill_stop_message",
     }
 )
 
@@ -434,7 +437,15 @@ async def start_run(
     # The ``context`` field is a custom extension for the langgraph-compat layer
     # that carries agent configuration (model_name, thinking_enabled, etc.).
     # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
-    merge_run_context_overrides(config, getattr(body, "context", None))
+    body_context = dict(getattr(body, "context", None) or {})
+    # Promote whitelisted top-level body fields into the context map so
+    # callers can pass them either nested under ``context`` or as flat
+    # top-level fields (``excluded_tools``, ``skill_stop_names``, …).
+    for top_level_key in ("excluded_tools", "skill_stop_names", "skill_stop_message"):
+        value = getattr(body, top_level_key, None)
+        if value is not None and top_level_key not in body_context:
+            body_context[top_level_key] = value
+    merge_run_context_overrides(config, body_context or None)
 
     stream_modes = normalize_stream_modes(body.stream_mode)
 

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

@@ -312,6 +312,37 @@ def _build_middlewares(
 
     middlewares.append(SetupAgentApprovalMiddleware())
 
+    # SkillStopMiddleware — intercepts skill invocations and halts execution
+    # when a configured skill is about to run. Request-level options
+    # (skill_stop_names / skill_stop_message) take precedence; static
+    # config in ``app_config.skill_stop`` is the fallback.
+    skill_stop_names = cfg.get("skill_stop_names")
+    skill_stop_message = cfg.get("skill_stop_message")
+    if skill_stop_names:
+        from deerflow.agents.middlewares.skill_stop_middleware import SkillStopMiddleware
+
+        middlewares.append(
+            SkillStopMiddleware(
+                skill_names=skill_stop_names,
+                skills_root=resolved_app_config.skills.container_path or "/mnt/skills",
+                stop_message=skill_stop_message,
+                stop_on_match=True,
+            )
+        )
+        logger.info("SkillStopMiddleware enabled (from request): %s", skill_stop_names)
+    elif resolved_app_config.skill_stop.enabled and resolved_app_config.skill_stop.skill_names:
+        from deerflow.agents.middlewares.skill_stop_middleware import SkillStopMiddleware
+
+        middlewares.append(
+            SkillStopMiddleware(
+                skill_names=resolved_app_config.skill_stop.skill_names,
+                skills_root=resolved_app_config.skills.container_path or "/mnt/skills",
+                stop_message=resolved_app_config.skill_stop.stop_message,
+                stop_on_match=resolved_app_config.skill_stop.stop_on_match,
+            )
+        )
+        logger.info("SkillStopMiddleware enabled (from config): %s", resolved_app_config.skill_stop.skill_names)
+
     # ClarificationMiddleware should always be last
     middlewares.append(ClarificationMiddleware())
     return middlewares
@@ -345,6 +376,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
     if not isinstance(writing_artifact_path, str) or not writing_artifact_path.strip():
         writing_artifact_path = None
     agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name"))
+    requested_excluded_tools = cfg.get("excluded_tools") or None
 
     try:
         agent_config = load_agent_config(agent_id) if not is_bootstrap else None
@@ -415,6 +447,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
                 model_name=model_name,
                 subagent_enabled=subagent_enabled,
                 is_scheduled_run=cfg.get("is_scheduled_run", False),
+                excluded_tools=requested_excluded_tools,
                 app_config=resolved_app_config,
             )
             + [setup_agent],
@@ -440,6 +473,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
             groups=agent_config.tool_groups if agent_config else None,
             subagent_enabled=subagent_enabled,
             is_scheduled_run=cfg.get("is_scheduled_run", False),
+            excluded_tools=requested_excluded_tools,
             app_config=resolved_app_config,
         ),
         middleware=_build_middlewares(config, model_name=model_name, agent_name=agent_id, app_config=resolved_app_config),

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/skill_stop_middleware.py

@@ -0,0 +1,144 @@
+"""Middleware for intercepting skill calls and stopping execution when specific skills are invoked."""
+
+import logging
+import re
+from collections.abc import Callable
+from typing import Any, override
+
+from langchain.agents import AgentState
+from langchain.agents.middleware import AgentMiddleware
+from langchain_core.messages import ToolMessage
+from langgraph.prebuilt.tool_node import ToolCallRequest
+from langgraph.types import Command, interrupt
+
+logger = logging.getLogger(__name__)
+
+
+def extract_name_task(text: str) -> tuple[str, str]:
+    """Extract `agent_name` and `task_content` from a bash invocation that triggers
+    a sub-agent via ``python3 scripts/agent_orchestration.py "<agent>" "<task>"``.
+
+    Returns empty strings when no match is found so the caller can treat the
+    absence of a match as "not a target skill call".
+    """
+    try:
+        pattern = r'\.py\s+"([^"]*)"\s+"([^"]*)"'
+        match = re.search(pattern, text)
+        if match:
+            return match.group(1), match.group(2)
+        return "", ""
+    except Exception:
+        return "", ""
+
+
+class SkillStopMiddlewareState(AgentState):
+    """Compatible with the `ThreadState` schema."""
+
+    pass
+
+
+class SkillStopMiddleware(AgentMiddleware[SkillStopMiddlewareState]):
+    """Intercepts skill file reads and stops execution when specific skills are invoked.
+
+    When the model reads a file from a skill directory that matches one of the
+    configured skill names, this middleware:
+    1. Intercepts the tool call before execution
+    2. Returns a Command that stops execution (goes to END)
+    3. Optionally returns a message indicating which skill was triggered
+
+    This is useful for implementing "skill triggers" that pause the agent
+    when certain skills are about to be used.
+    """
+
+    state_schema = SkillStopMiddlewareState
+
+    def __init__(
+        self,
+        *,
+        skill_names: list[str],
+        skills_root: str = "/mnt/skills",
+        stop_message: str | None = None,
+        stop_on_match: bool = True,
+    ):
+        """Initialize the skill stop middleware.
+
+        Args:
+            skill_names: List of skill names to intercept (e.g., ["newsletter-generation", "podcast-generation"]).
+            skills_root: Root directory where skills are located (default: "/mnt/skills").
+            stop_message: Optional message to return when stopping. If None, a default message is used.
+            stop_on_match: If True, stop when skill matches. If False, continue execution (for logging only).
+        """
+        super().__init__()
+        self._skill_names = set(skill_names)
+        self._skills_root = skills_root.rstrip("/")
+        self._stop_message = stop_message
+        self._stop_on_match = stop_on_match
+
+    def _is_target_skill_call(self, tool_call: dict[str, Any]) -> str | None:
+        """Return the matched skill name if this tool call invokes one of the
+        configured target skills, otherwise ``None``.
+
+        Matching today is bash-based: the lead agent reaches into a skill by
+        running ``python3 scripts/agent_orchestration.py "<skill-name>" "<task>"``,
+        so we only look at ``bash`` calls and extract the skill name from the
+        argv-style quoted form.
+        """
+        name = tool_call.get("name") or ""
+        skill_name = ""
+        if name == "bash" and "args" in tool_call:
+            skill_name, _ = extract_name_task(str(tool_call["args"]))
+        if not skill_name:
+            return None
+        lowered_targets = {s.lower() for s in self._skill_names}
+        if skill_name in self._skill_names or skill_name.lower() in lowered_targets:
+            logger.info("SkillStopMiddleware: matched skill_name=%s", skill_name)
+            return skill_name
+        return None
+
+    def _handle_skill_stop(self, request: ToolCallRequest, skill_name: str) -> Command:
+        """Build the stop response — an interrupt that includes a ToolMessage
+        describing which skill was triggered."""
+        logger.info("Skill stop triggered: skill_name=%s", skill_name)
+
+        tool_call_id = request.tool_call.get("id", "")
+        message_content = self._stop_message or f"⏹️ Skill '{skill_name}' triggered. Execution stopped."
+
+        tool_message = ToolMessage(
+            id=f"skill_stop:{tool_call_id}",
+            content=message_content,
+            tool_call_id=tool_call_id,
+            name="skill_stop",
+        )
+
+        return interrupt(
+            {
+                "messages": [tool_message],
+                "skill_name": skill_name,
+            }
+        )
+
+    @override
+    def wrap_tool_call(
+        self,
+        request: ToolCallRequest,
+        handler: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        matched_skill = self._is_target_skill_call(request.tool_call)
+        if matched_skill:
+            if self._stop_on_match:
+                return self._handle_skill_stop(request, matched_skill)
+            logger.info("Skill detected (not stopping): skill_name=%s", matched_skill)
+        return handler(request)
+
+    @override
+    async def awrap_tool_call(
+        self,
+        request: ToolCallRequest,
+        handler: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        matched_skill = self._is_target_skill_call(request.tool_call)
+        if matched_skill:
+            if self._stop_on_match:
+                return self._handle_skill_stop(request, matched_skill)
+            logger.info("Skill detected (not stopping): skill_name=%s", matched_skill)
+        return await handler(request)

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

@@ -22,6 +22,7 @@ from deerflow.config.sandbox_config import SandboxConfig
 from deerflow.config.skill_evolution_config import SkillEvolutionConfig
 from deerflow.config.skill_evolution_runtime import get_overrides_mtime as _get_skill_evolution_overrides_mtime
 from deerflow.config.skill_evolution_runtime import load_overrides as _load_skill_evolution_overrides
+from deerflow.config.skill_stop_config import SkillStopConfig
 from deerflow.config.skills_config import SkillsConfig
 from deerflow.config.stream_bridge_config import StreamBridgeConfig, load_stream_bridge_config_from_dict
 from deerflow.config.subagents_config import SubagentsAppConfig, load_subagents_config_from_dict
@@ -117,6 +118,8 @@ class AppConfig(BaseModel):
     subagents: SubagentsAppConfig = Field(default_factory=SubagentsAppConfig, description="Subagent runtime configuration")
     guardrails: GuardrailsConfig = Field(default_factory=GuardrailsConfig, description="Guardrail middleware configuration")
     circuit_breaker: CircuitBreakerConfig = Field(default_factory=CircuitBreakerConfig, description="LLM circuit breaker configuration")
+    excluded_tools: list[str] = Field(default_factory=list, description="Tool names to exclude from the main agent (e.g. ['task', 'ask_clarification'])")
+    skill_stop: SkillStopConfig = Field(default_factory=SkillStopConfig, description="Skill stop middleware configuration")
     model_config = ConfigDict(extra="allow")
     database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="Unified database backend configuration")
     run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")

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

@@ -0,0 +1,31 @@
+"""Configuration for skill stop middleware."""
+
+from pydantic import BaseModel, Field
+
+
+class SkillStopConfig(BaseModel):
+    """Configuration for the skill stop middleware.
+
+    When enabled, the middleware will intercept skill invocations and stop
+    execution when a matching skill is about to be used.
+    """
+
+    enabled: bool = Field(
+        default=False,
+        description="Enable skill stop middleware. When True, execution will stop when specified skills are invoked.",
+    )
+
+    skill_names: list[str] = Field(
+        default_factory=list,
+        description="List of skill names to intercept. When the model reads a file from these skill directories, execution will stop.",
+    )
+
+    stop_message: str | None = Field(
+        default=None,
+        description="Custom message to display when stopping. If None, a default message is used.",
+    )
+
+    stop_on_match: bool = Field(
+        default=True,
+        description="If True, stop execution when skill matches. If False, only log the match without stopping.",
+    )

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

@@ -39,6 +39,7 @@ def get_available_tools(
     model_name: str | None = None,
     subagent_enabled: bool = False,
     is_scheduled_run: bool = False,
+    excluded_tools: list[str] | None = None,
     *,
     app_config: AppConfig | None = None,
 ) -> list[BaseTool]:
@@ -52,6 +53,9 @@ def get_available_tools(
         include_mcp: Whether to include tools from MCP servers (default: True).
         model_name: Optional model name to determine if vision tools should be included.
         subagent_enabled: Whether to include subagent tools (task, task_status).
+        excluded_tools: Optional list of tool names to exclude from the result
+            (applied AFTER all other resolution). Takes precedence over
+            ``app_config.excluded_tools``.
 
     Returns:
         List of available tools.
@@ -209,4 +213,19 @@ def get_available_tools(
                 "Duplicate tool name %r detected and skipped — check your config.yaml and MCP server registrations (issue #1803).",
                 t.name,
             )
+
+    # Drop tools the caller (or config) wants excluded. Request-level
+    # ``excluded_tools`` wins; the config-level list is the static fallback.
+    effective_excluded = excluded_tools if excluded_tools else getattr(config, "excluded_tools", None) or []
+    if effective_excluded:
+        excluded_set = set(effective_excluded)
+        filtered_tools = [t for t in unique_tools if t.name not in excluded_set]
+        if len(filtered_tools) < len(unique_tools):
+            logger.info(
+                "Excluded %d tool(s) by request/config: %s",
+                len(unique_tools) - len(filtered_tools),
+                effective_excluded,
+            )
+        unique_tools = filtered_tools
+
     return unique_tools

服务启停说明.md

@@ -111,3 +111,10 @@ uv sync
 ```
 
 再启动。
+
+
+{ "agent_type": "leader", "d_agent_thread_id": {
+        "查询天气的智能体": "76c95f3e-ee02-4e9b-b767-6df129a5aea5",
+        "查询黄历的智能体": "9a5536a1-5e1e-4345-b9b7-9807c31154e4",
+        "main-agent": "276eac46-1acb-47e3-ad42-8f01af51e869"
+    }, "agent_name": "main-agent", "new_message": "帮我看看今天出门合适吗", "skill_stop_names": ["查询天气的智能体","查询黄历的智能体"] }