From edb3238f3e7ecaa00479fff4be8c6cbd8bd3ebae Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Fri, 22 May 2026 20:40:36 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend-web/docs/multi-agent-api.md | 265 ++++++++++ newpython/OUT-v1.py | 483 ++++++++++++++++++ newpython/OUT.py | 483 ++++++++++++++++++ newpython/__pycache__/OUT.cpython-310.pyc | Bin 0 -> 12141 bytes .../__pycache__/demo_call.cpython-310.pyc | Bin 0 -> 4123 bytes newpython/demo_call.py | 132 +++++ .../backend/app/gateway/app.py | 110 ++-- .../app/gateway/routers/multi_agent.py | 440 ++++++++++++++++ .../app/gateway/routers/thread_runs.py | 3 + .../backend/app/gateway/services.py | 13 +- .../deerflow/agents/lead_agent/agent.py | 34 ++ .../middlewares/skill_stop_middleware.py | 144 ++++++ .../harness/deerflow/config/app_config.py | 3 + .../deerflow/config/skill_stop_config.py | 31 ++ .../packages/harness/deerflow/tools/tools.py | 19 + 服务启停说明.md | 7 + 16 files changed, 2113 insertions(+), 54 deletions(-) create mode 100644 frontend-web/docs/multi-agent-api.md create mode 100644 newpython/OUT-v1.py create mode 100644 newpython/OUT.py create mode 100644 newpython/__pycache__/OUT.cpython-310.pyc create mode 100644 newpython/__pycache__/demo_call.cpython-310.pyc create mode 100644 newpython/demo_call.py create mode 100644 offline-backend-20260512/backend/app/gateway/routers/multi_agent.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/skill_stop_middleware.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/config/skill_stop_config.py diff --git a/frontend-web/docs/multi-agent-api.md b/frontend-web/docs/multi-agent-api.md new file mode 100644 index 00000000..a493801e --- /dev/null +++ b/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 的实际执行 | diff --git a/newpython/OUT-v1.py b/newpython/OUT-v1.py new file mode 100644 index 00000000..f11ae88e --- /dev/null +++ b/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) # 多进程提升并发能力 \ No newline at end of file diff --git a/newpython/OUT.py b/newpython/OUT.py new file mode 100644 index 00000000..f11ae88e --- /dev/null +++ b/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) # 多进程提升并发能力 \ No newline at end of file diff --git a/newpython/__pycache__/OUT.cpython-310.pyc b/newpython/__pycache__/OUT.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..0f8e4dc72fbd9f1e4171879140074fda107645c8 GIT binary patch literal 12141 zcmb_iTXY-8dEQ+t79ap#Mcri^rfgZTd68ttaTwK!EMJqDP9-@`Hg-@Lvm`-+06n|V zg$7enluYVENw#IlmMEKaB-=_O`KIVb$$DrX+SBxFv8RiY%wzQ?Lr%^q0~Rn%4};;)u{pVgt=B7E!dRUVPpVx!WiST2eDlCu0#^d4f% z*z!GIy^5`1EB8oxwHRTIUc)xA+t}^syO!O-!l+;=ncloAxU3ttnRwYxRW(^$ZWMpDc3UU&9}c>zuQYi`fuG*{~-IesE-);vWL1AK4mm?-N(MeTG@BmM)nu%Vet%QE9Psf z6nP}`5JVct{r14jhi~M5JLp_HoI7;JIsB{X6MLNV2Xk*+nAvk8H};#rrXN3r>gf*- ze17djV109V`mIUl`WrKcE;)yfS?O*g71+=mc1BJ*lUJ=yZp7H<*G8PfAI=W!`~2F$ zsVnawKQ(z~>iUi8qaWmsk5BKvm>U>$ANBdQeS!O%!!sWZQC-;>0)eT^r>EZ^%^kaf zanXaD!ft+jYxu{{Y~2=aiS;I0cqV1Gm=+$|6JA^F8;_kC{zHboP$S?`)Qoq^)mpr)BlFwzRC@@a^Vx_*>ul(7JW&`&Nq4 zZC~FK%UGQ)$#h2|)sis{o{IGteIZeqe{f&T!WXgqg_gF)4J*S_amcCoQIYfFyBS~e z2S_ZbOSYhfSt$Y4WFDmIsHb1f%Hy61sb9$|<0|t`$h{3BrxkO)gfx+_>-S_mSQ7qB zEjqkKzFa}+TxfGTWf>`}Y1^(|W3DFF+nY?pV^$)aYUwi5sk4f$(y#67#1b&LX?w-u z&_l6!+%V0kxUy9ivtkji?dQfz8N;+pTkTDomR<3qXP$VfE&A-1A8{X^W_vmeE24@g zQ(Ln)ox=Q)t>P6(&Pu$72Bkgo?6&6KU3LYUyH8k(Vw37rMUgf7`_iwiQB|4Op-|j> z%Y4!k#e6=G44H|nbW%Enxg1maWp{oPQr4YU#quz3R_Riy#gHsZm{l+4*@JS;@^)!i z5A!iUs~GhZTX@EOEHEKshJ%^g@rd%JM2q#tQRn2_#RbSoymP)sj2Jg~_^iT7`6*c#5%FQ{`~A9b!CZ4J+oFdv7sv_y(QoOZ}anHg?G zV`e;&uq%qgi1>LBO?e239o+Whmd7?dx|IZEdwY2THwH~BW@XH1JPmORp?O)Rr`P0F zxC62yIdMNF)szU?kzj=6IIT3hG7iTR*aj##P*V=$7*M2C@z8%-TCtJA=+I+|iAu;z)F=9A`7h}c5Lf_O<`ARs7; z1%VIA{SY@(vx*SB7lQX#N|z@qL!8>E(yv--RwY3Tk&7JjEtkf%35ogLn!@!`iB&9> z@@0YL(o(6N>}{|uulZttW{a(KC4{K?A`FR>yw1vEzHC+4iJ)j0rF`q0+-L`xVa9o) zmsBp|v;B{3+WMGK7_qR0tZ3%>N=jULB-1+#4s8pRy0SePo?IxK1!7r2+EltFTK3?F z(7Cw;NtNuE6-CHN6;k>tq&^_?2&!y}H_#aFqP({U%OhzAt#mpW?cnK5uW4&$cOsdD z>YM3IGEPgm@S(+4)W$hnSP&G1qggUUrkqmxeL_rHQB3|UG&AdE+L))mBC9Z8M&@^9 zl`J5}Mu~J0*Y{*SxURspDz00yDy{qI|T!k!87u8j=Dpp<6V4>H(Y(>`3 zYG7LfXjz*LbQSNUJgbAtO=knb>ITtb5iJ$`){8!a3;WClw^Hq5-0Lrlpmc==vn2CR zFd`~}b*kQcb&=_uLPb7_o7s@+u*A>ROe*m*{>y4s*J>4sE z8y4J+)z})|`pmOGYVKiS=as$A`73y4aSo@hUd@ePYni$FH&c_>-5Dr6T+GB4;jPPL z;EHW}^7Y)?W1p@sw1DH=}VIzYi&Iq@)Q+h5-4@hnI36@6N-R z3~9bwg5}T}=8JLtI42zyUgD2bU*;KR{{h#T6-a(R_IgR8?pSjpN1Xld+?gS>UGNwg zvM;|MJDy*K&gqY5E}Ur%M=D&8dN*oIHizFvm%qGoLKP;B}SEa0wYm^!#0Q*IrMa5M}=5%A6++*QGb6vDXw(4cXZ^538rht2Pu4df%hF2$h8E0|71*J_qS#XE4~W*V zNKEaJ`-5;cfvg6c5ri$$M!o$Z=F5h*OZ*8qo6vTdKamZ=dU#od+n0+&U^4=3Vu@wQ z)lFR=hnnXGCUb}0$GH540`U6mz)@VyTzUt2g6|`jp!rjilg|E$>67Pk@u)o z+Aw8Zu zk#aH0Zj91=5Qz$9rf*27bqIN=zE}IguTiNcz$6(1VGzEk@u_kZ+VG9I^^gCKq<0q( zcIy7Hi?Ym?uMq)ajc~)W<~L%`hHyjf=*8KSuR!e1$Sb+Gk4%4d$T@kP+{Z_+Iv-w{ z-ghN8c0Bjid0Lud2gE`==p5XeJMbHKolhScrh5;MQsYCze;62?x;*L}IXL~D^bHOXrQ-|{9eja67kpyHvk9`?8G~5uKN=!C9sGo7e`~A%!Lt!_X1#bi$g@$kU;Cxk^06v0+Ul$((4|TqfQ|~q`7-2P0h6+9 zL7gxne^#^=GJhKP$2~|29||^$9L+jh2%NBpSeC2;@4eWfn8mYP%1T2DVg!(>sF8*| zzzB+vlG2u;sL;7`Blp_5+{F=cfdZPfLVM_Wt`SykU;Z8JT9$vQqK(ySdl3)6m?(?n zY)`kb>lUSXgac$D3@A>>KjHxd>cB?`7Z9;cTpd?;w8cAJ5k%*#jy z0guLAI!ZwdjMwatC)B49x;+VFLD$Oxn*g1>pi2lQX%@f`;%-X4P%R(R0Cn8(&JWcB zk;1?aBd`mNDIIb`VwGba{K^!dC19*r4Xb5!me0}uhVRau>-E8L+VzJV9~Rt4p5Zzj>JEg>H(p@ z(_h=S-m142XKPtA^(Vk|UH>A?e@nJbC`o{=%PtZ*cZIFb)?sWLPIrl$p zS`Fr_F1rY2n2qP8Bl1ft3$sFb)|9O&jV5w`hQN-CMHMW?RHwz{^KKZeG0p>ke%p`CpkE%=j^kmm#E_f#?Oaq~Yw zfnFP9EJ6HFEE(zW$wW`Wnw^0qnyX}n(Q6t;cT>DGX4#%zp0>3f zgC#OOXZ@}t4TzxKMQ}+{m%w$Zz9#3 zv2+d9?TL=OU)gEtpc;weFFe$}z_UAethdwDYx30;A@rnxc~;haR-(s9XDnN#-3Adzb3)Rye~)CY z0xXuh>o8Xxc1e1Jg`eEE?Z@Fcr8PX~%~46a1<P_;)Cok0wXGgXzFyle14XBhH5H`6S~mZ?LIg+bcCEm^w?Jq6&- z75vu8>P_`#J+?(OT=`KB7yre*e_DGpRIQN1mo?O@-g<$V>u^^Mwo~c_c3z{@<1P(l zd;|lX2M7pm6#>D60uan|`t#Ua;q(bl`vjhLoj&1dY!TBA3Y>l*@AQLtrw@cHoIadC ztFf>bV-f1Eh10KNF!d+_-wO9Hu_f=4+g~c&zS`xb&EUM-U-qPwcl#P!k$3wm^KPH& z$nE>sDmd~YFH3oL8{Gb0IPVJK_R$hs(Q zxI6Fm&lKJMGPYW{{k!1y&se^q+XpxhZhxO}`^$=MAAKN(aNR!g=B09H4!8ebxBOjs zORGw_{k!4zr|>>&+-tbjd&HZq&Aa_&aQoG8`w=+#)ofXr+s8Htfr0*Nt7g2`s$=(M ziK}m<_=q)~@&XvZrOyL{V&tr4Udwf-ym@!L3{lGZ3`K|E=#I@^Bc6aC6gEq+Uc%4v zEx7TU-7DWpO}0_;EG0jo66|Y5@dHmxQD`#86Ycq3 zP`#4q3JNQTmp1jl_5=cyL=VQt{|dLXafXx|lu(c<+zD@|Tn8nclq4wWqNJOWBqcqR z5UI;kl%y%?rQ{__IFbyhoOA9-k-c+yAZ+4~9C0@ZFJb?Eta0k<>8Z&#OU9ZLsNg2` zKqCU+W_STDcMfR4oWD7h}4vf zFdw4Q|K9lWJ=BgYFW*baE0hdVLS}cNMXj8-Hr9c8Fb|@cGT>MKKMX1{N&gpv+N`XS z`92J29{E6V6-nW5@ihqw4k9dM${s{bj}pQL89|Yipl1m6dtiVR1~NjFi_o$dZJ`d4 zgq&5x&Z9R&KFx)*1Y}*rmPjQm+V?qf5GdCfzMgyIBbVaLYhBTK&m)dt(#2jXsgOzG zA1Dmz1x*|`kS3&8m6w3+g*+lMUxRv(1WNNd-iqu(F$FdzfzjgWp5BKCNJi9mz0`Mm zX?)EalW71I^I>EPP1`cRMB@NxVj{u@4}!sN#Rd3eu-!8U&Vae5U9tiyMrC|36jj%iA>_3@1b;f<{%#F7&=JGBAxlcdH?Kxa} zv;4@)lw7pM?XVEk~=m$HFy|)C z)+ALyIw%8GN<9>Hnb$A#QQQQQ(Dyj#2^>VSD+y#jwli*sZE#zC1TZWj+qGK-97m+G zC4W`~g;krM*|PQN1Y=19v65lS_eNB`61X?LBZ`w0IFw_;j*Cl>%5)e4{=_V_=rOBY7Enh9HU^AnPbTkungu|-MNfPfJJ~mr4Ei8w)p2pSWJ1~J<~?)(XP+X< z>4=W954wal*!V#ngb?yRilFzY!_u%ktPFdG)pk&etYVL!&Mjo+A&J-#BK}y_n1{Bk ziQ>y_Bn=7UlN-O18$BP+Z={8x23RF|13=VW*g1DRcmB7|$m!g%bFTW&i*v&>H{OJD zgy$Jl(X&Vj{fZ5iJ=3SJPJMR3-GY8u)YA|eToKdL8d}2*F4)BuYqdheFIeLFVblxx0!hs3F zZffq3%bXTrlal8m?^XPe4I{ZPLdGQbJ*bSpjswzFTGEIzB8_-P)FHKBO8G3g{5o5J zhLV7W!qQuwA%!p`LhHis*nxzJ#gFhfj_8FY;+9L>(%voXb+riKqWN`>NXr}V%%r-{dPUsC zk5V@xjH1QuI&VR<*5VSuv>Tll@BWnZx(-$3erx0^rN$Ewr+KtL_eF}j%i_ke{SPyy*$biPx=O z@CK*oZRwbjG7zc`A$p>}dXG{qHlq+jAwK(3`;wnR6FLZh+d)yl70N2|;uhT9?3o(Q*iXtoz#GjWGfI5Vbu_$q^)aZN9u2fsi<`X>!sZ zeQBY*r|68E<7sa4*QnN)*`B~vO7~zs^(s)D(Na55MIe0KEuiD{AX>~tfzU#XMr{wy z0hdX(BwqhU(3F{^@i0UynG~t0l0WDk0#OxrE3nQ}R~@KRY6}F4I)~g%P%U3+*!&5o z%AW&X_<6r_IiIY$qz8n`HLk2=84y7S+;wI@#f9c<#=}%zY`F^G|ok?d%-<9rclQNY~ zEA8&P-_P^B?-knGWDc$?$;9ZjjU4wi%dEX3ka-jy>z|MiPT>eoP@Y$K|BV#nzXe4g zf+7-;NW(~xNJt5huo5PbVP1)ls2q-tBGm& z;pNM3H6}j5+OTfgs$m*7{_6EV;JmI8o#NWX>$QuEeu;PLHEb32oT23dW9(h|f>kcx zczI|cj`p_ zPZzv1pW+eQE~a~Wdi#Ew+75Sbdgu1-y%=g>TF|I3g`4yDE`L#9Jc(Z{j@u&z6L*i{ z(#uA!Kuu9) zkE#8?9LNl+nf_-6vbgcaaoD5Q()`ib1Xtz^_1=*gZ}BR0w6#Yo8a@7&S$KK!TJ6KL z-jUCfvRt3JSUYicWtME8)tr{k!wI~Iy{W%=CoizgX#n=POnKAytp1?~%0Dbw-G*uF z`Md##lGD3~HG4#-hB@5r4_Us;){Fz1`Tva)Z1C<}p+JdYYIcE^(lDrZ;Y#ho^wPrl z`t=h_3m4XJm*>5|zN%~pAX1>Y z5#6$>W*Y_54N=`LQ8ULD)Vd`2h{E%Cc&w|C9O7-D*(89|=0_1hM4005=XP^Me3JX% zt_tUHqk=8Ya73C#wq%D$=nNv^DHK>#c!!@0S2>=8ccjWy(DTqz#2rPKcjP*AA zvvwYUaoGJ^)&h}87grJB9fh|D^N<~3hMPtHY&a6LBcoAABymR6sg)W>aKs@XQnI78 z2NqB}f41@2$GA6*y(6>UC&#?w@3iPsJ9fjH`6ISRC@@Afh=slL?|O$9@GHS~ncA4{ z^Jm4Pk}7Z6wQCptU2o5h7x1%tvV%B)lFz?uqMjt)LD`DFh(+ed#Q;w~%NFpV+ z+PCF)jA@3cu9lqRU2)VZm{f$0TzO?&tK5m`FzzqwPuz!3W(~zcp^=evvnzQW1+lLpK7nR3Gcl4!9k+G0p9G zrhj*44fm7;)eK8lhYD1+Y|SoNG{mMZkSUka0KaFM@3?4YPT2 zv=d%8<7@fmb}-^K_l4=&kHN(H5hQ3Lf;xGTkMIZ~Ui>!xZ6qx7TmK!ABYZo*1 zl;8uO)f*K6JqbC&sMG=^`zGOi0_}H%xlols8Y}^};CUQz4m=N$tIR{i4ujtbTD{7H zZM4euapebg`1a3p70HpT%>DsKnv+Sq%G0tV0l<0IYAyy83YEile6Gz2RryNzdCrtY z&W?|^kLO#|**|cAvpbxyotQ+lOxm6B%N+C)Chb5Ye*eIO9FUM-AtB6ig@o{J9bZer z|Fv}ZEawxJ*?zRvmem01^033;mB3x$Y<2+GE*)9EcqP4j{Yq{65=-BF+nf4)TkXsQ zc#`EC(~N;lKe7DAXWr#M<8edJ6Xx{rEB6GLx+jfOsZ^z<^FbjaYi<6^RE7#Lq9QUj zmN)c(?CKzJx;j>BB8W0kD@Peoo*W!}7B~LE@^>`{{hkJRNHg*}N#n9qvWL2Nx?!pp z^IA?f{so{ZOGLA^bc6NkhJv{y*U>m+X*3#-A&h zOd|#;El@W^biXpD8MaQACTL%n##Fap{jbf8r4B3{_@vH2f%;SE$pcC1! z3=`Ub-*H7^@B<2&m3=F)J--tvW@v@kJv>k59-CB=;uM^@o_ufq0 z5;u1)i<@Z@n))i|t3ZxPR_kKg$D|>kDSQMz$N|z2G%EzYhzJhva5l09BA$^Zd7zsF zbOYKJyN1DQfVT+Q;;7^Z1PDl?YyAVjR4zRQ9XEYLU%C}DHQu`fkXrE%6N|MIpDbTk z^yZHPoNGreH{P9ZeDvP7rK_KKZ(I!mtlGk1Z|ZcayhHXKL*J@nXl>Oh67kj`Hrp12 zDv#XclS(KkIwvV|;0NsSq7H##Q~g$?-ECGgC9mo#u-ji_?*{60t{j{1L-1BS@ zVEM44>sT{!_0Z@Kpumj~!zyZaZsc~v#C#18R`(mQw6;KkCK90AOuK_y{EvJq65RqQ zq1&OF_CWF-Ffbsj1SIp2IRqhv6>SA2EMm5ygb0)9aqfjDevc{$LIwau7YC`(#UWG? zD-zgC4gvs$Y~(;V17Bf>g)NcUa5J2Ug3cl$2kHz~04gHy6Y0YwX7VHsiXA{tAZ-rk zh>kSOk@hC^5I`>kV{|y7mOt3v&gPa(lztAAZUSvb8rO~%Jm2=QvB!8{;kNjbizi>){Lp7MRkW(@OvP;LHUXknG!%esg$RN0!f575*&$VvoI zW|vdKO>=r56qcA6GD{#%?Z~M#2G}qmbfdai0>=;X>l>F#X>NPMGH)pA^@xPaAW>n2Dk5kE)f;Kp)9J- z!LxEPPr@EwdUnAFlb%WD%ef~xz!vyVKpOaEIlhaIclN(l5Qpl;Q64qFhXA}Y^`@YI|PaR-L*%WCvOHz=yl3gn1;ebH# z=}%axA5n$CDd;&n5Dah`Q0p_+pMGlsOny|=6==y^Gx<)|iSI=sKN~BMQeNN1@Mzr) liO8|gErLV=i7cQKL!uB8?i4#W-hJoY>*P(5aDo?a{TC2H+@=5k literal 0 HcmV?d00001 diff --git a/newpython/demo_call.py b/newpython/demo_call.py new file mode 100644 index 00000000..c4df49e6 --- /dev/null +++ b/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()) diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 3d32f61b..e1e10155 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/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. diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py new file mode 100644 index 00000000..e0dabe0e --- /dev/null +++ b/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", + }, + ) diff --git a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py b/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py index 0bb99f8b..38129fa5 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py +++ b/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): diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index b5a9b452..c48b8651 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py index 7f408e8b..d27dd73f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/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), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/skill_stop_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/skill_stop_middleware.py new file mode 100644 index 00000000..f0950fd0 --- /dev/null +++ b/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 "" ""``. + + 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 "" ""``, + 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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py index fce0f936..2d56d443 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py +++ b/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") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_stop_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_stop_config.py new file mode 100644 index 00000000..a3b29262 --- /dev/null +++ b/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.", + ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py index bd5673c3..9ad2c249 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py +++ b/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 diff --git a/服务启停说明.md b/服务启停说明.md index 348fa7e6..c7291d35 100644 --- a/服务启停说明.md +++ b/服务启停说明.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": ["查询天气的智能体","查询黄历的智能体"] } \ No newline at end of file