1362 lines
59 KiB
Python
1362 lines
59 KiB
Python
"""圆桌规划 · Step 2「多智能体研讨」Gateway 路由。
|
||
|
||
前端页面：``RoundtablePlanningPage`` Step 2
|
||
前端模块：``frontend-web/src/roundtable-planning/api/multi-agent.ts``
|
||
|
||
本模块把原 ``newpython/OUT.py`` 演示编排进 FastAPI Gateway，通过 **HTTP loopback**
|
||
（``127.0.0.1`` + 调用方 ``Authorization``）复用既有 ``/api/threads``、
|
||
``/api/agents``、``/api/threads/{id}/runs/stream``，不再硬编码 token。
|
||
|
||
对外接口
|
||
--------
|
||
1. ``POST /api/multi-agent/init``（SSE）
|
||
- 为每个研讨席位（``agent_names``）并行创建 thread，按完成顺序推送
|
||
``seat_ready``；
|
||
- 动态创建**当次会话**的总控协调智能体（``main_agent_name``，如
|
||
``roundtable-coordinator-xxx``），写入席位列表 SOUL + ``agent_orchestration`` skill；
|
||
- 结束帧 ``init_done`` 携带 ``thread_ids: { agent_name: thread_id, ... }``，
|
||
含协调智能体自身 thread。
|
||
|
||
2. ``POST /api/multi-agent/run/stream``（SSE）
|
||
- ``agent_type=leader``：总控一轮——可能派活、可能澄清、可能共识（空 dispatch）；
|
||
- ``agent_type=special``：单席位子智能体执行一轮交付。
|
||
|
||
跨席位「广播」语义
|
||
------------------
|
||
通过 ``_append_thread_message`` 向其它 thread 的 checkpoint ``messages`` 追加
|
||
**只读上下文**的 human 消息（非触发 run），使各席位 LangGraph 状态里能看到
|
||
总控派活/用户干预/子 agent 完成摘要。前端编排循环仍靠显式多次 ``/run/stream``
|
||
驱动各 agent 真正推理。
|
||
|
||
Leader 终态帧（前端 ``isFinalStatusFrame``）
|
||
-------------------------------------------
|
||
- ``status: [[agent_name, task], ...]`` — 派活列表，前端并行 special run；
|
||
- ``status: []`` — 无 tool_calls，视为共识，可进 Step 3；
|
||
- ``status: "clarification"`` — 总控 ``ask_clarification``，前端暂停编排展示澄清条；
|
||
- ``status: "子智能体 X 完成..."`` — special 完成（字符串状态）；
|
||
- ``status: "error"`` — 失败。
|
||
|
||
共享导出
|
||
--------
|
||
``intent.py`` / ``recommend.py`` 从此模块 import 底层流式与消息解析函数，
|
||
勿在 intent/recommend 中重复实现 loopback 逻辑。
|
||
|
||
相关文档：``frontend-web/docs/multi-agent-backend-dev.md``
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
import traceback
|
||
import uuid
|
||
from typing import Any, AsyncIterator
|
||
|
||
import httpx
|
||
from deerflow.config import get_app_config
|
||
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
from deerflow.utils.time import now_iso
|
||
from fastapi import APIRouter, Body, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from langgraph.checkpoint.base import empty_checkpoint
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 共享:异常诊断器(intent/recommend/multi-agent 共用)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _diagnose_exception(exc: BaseException, *, context: str = "") -> dict[str, Any]:
|
||
"""把后端异常转成结构化诊断 dict,直接给前端展示用。
|
||
|
||
设计目标:让运维 / 前端开发不用翻 backend 日志就能看到根因和处置建议。
|
||
返回字段:
|
||
- code: 短稳定分类(loopback_connect_failed / upstream_connect_failed / ...)
|
||
- message: 一行人类可读
|
||
- exception_type: 表层异常类名
|
||
- exception_repr: repr(exc)
|
||
- root_type / root_repr: 沿 __cause__ 链找到的最深异常(若与表层不同)
|
||
- hint: 针对常见失败模式的可执行建议(中文)
|
||
- trace_tail: traceback 最后 ~12 行(避免响应体过大)
|
||
- context: 调用方标签(如 "intent_init")
|
||
"""
|
||
exc_type = type(exc).__name__
|
||
exc_repr = repr(exc)
|
||
|
||
# 沿 __cause__ 链找最深异常:httpx 经常把 ConnectError 包成 RemoteProtocolError 等。
|
||
root: BaseException = exc
|
||
seen: set[int] = {id(root)}
|
||
while getattr(root, "__cause__", None) is not None and id(root.__cause__) not in seen:
|
||
root = root.__cause__ # type: ignore[assignment]
|
||
seen.add(id(root))
|
||
|
||
code = "unexpected"
|
||
hint = "查看 trace_tail 最后一帧定位代码位置;若信息不足请到 backend 日志找完整 traceback。"
|
||
|
||
if isinstance(root, httpx.ConnectError):
|
||
target = str(root).strip().lower()
|
||
# loopback 走真 TCP 到 127.0.0.1:<gateway port>;失败 = Gateway 进程没起 / 端口被占。
|
||
# outbound 失败(其它 host)= LLM 提供商不可达。
|
||
if "127.0.0.1" in target or "localhost" in target or "loopback" in target:
|
||
code = "loopback_connect_failed"
|
||
hint = (
|
||
"本地 Gateway loopback 连接失败:确认后端进程正在监听配置的端口。"
|
||
"端口配置优先级(高→低):env DEER_FLOW_GATEWAY_PORT > config.yaml 的 "
|
||
"gateway.port > 默认 8001。若部署机改了端口,在 config.yaml 加 "
|
||
"`gateway:\\n port: <端口>` 即可。"
|
||
)
|
||
else:
|
||
code = "upstream_connect_failed"
|
||
hint = (
|
||
"外网/上游服务连接失败:大概率是 LLM 提供商不可达"
|
||
"(如 api.deepseek.com 在内网被防火墙拦截)。"
|
||
"把 config.yaml 里所有 base_url 改成内网可达的 LLM endpoint,"
|
||
"或加白名单/正向代理(HTTPS_PROXY)。"
|
||
)
|
||
elif isinstance(root, httpx.TimeoutException):
|
||
code = "upstream_timeout"
|
||
hint = (
|
||
"上游请求超时:网络抖动 / 模型推理过慢。"
|
||
"先 curl <base_url>/v1/models 确认可达,再看模型是否过载或 max_tokens 过大。"
|
||
)
|
||
elif isinstance(root, httpx.HTTPStatusError):
|
||
code = "upstream_http_error"
|
||
status = getattr(getattr(root, "response", None), "status_code", "?")
|
||
hint = (
|
||
f"上游接口返回 HTTP {status}:可能是 API key 失效、模型名不对、"
|
||
"或目标服务自身报错。查看 backend 日志里 response.text 拿详细错误体。"
|
||
)
|
||
elif isinstance(root, FileNotFoundError):
|
||
code = "file_not_found"
|
||
hint = (
|
||
f"找不到路径:{root}。常见原因:内置 agent 目录(.deer-flow/agents/<id>/)缺失,"
|
||
"或 SOUL.md 没拷过来;详见后端开发文档 §11.7。"
|
||
)
|
||
elif isinstance(root, KeyError):
|
||
code = "key_error"
|
||
hint = (
|
||
f"缺少 key:{root}。若 key 是 agent_name/agent_id 类:大概率内置 agent 目录缺失;"
|
||
"若是配置 key:检查 config.yaml schema。"
|
||
)
|
||
elif isinstance(root, ValueError) and "model" in str(root).lower():
|
||
code = "no_models_configured"
|
||
hint = "config.yaml 没有任何 models[],或所有条目都解析失败。检查 yaml 语法与 $ENV 引用。"
|
||
|
||
trace_lines = traceback.format_exception(type(exc), exc, exc.__traceback__)
|
||
trace_tail = "".join(trace_lines).splitlines()[-12:]
|
||
|
||
return {
|
||
"code": code,
|
||
"message": f"{exc_type}: {exc}" if str(exc) else exc_type,
|
||
"exception_type": exc_type,
|
||
"exception_repr": exc_repr,
|
||
"root_type": type(root).__name__ if root is not exc else None,
|
||
"root_repr": repr(root) if root is not exc else None,
|
||
"hint": hint,
|
||
"trace_tail": trace_tail,
|
||
"context": context or None,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Loopback 辅助函数（Gateway → Gateway，同进程）
|
||
#
|
||
# 圆桌三路由（intent / recommend / multi_agent）需要从一个 FastAPI handler 里
|
||
# 调用本机 Gateway 的另一个 endpoint（`/api/threads`、`/api/agents`、
|
||
# `/api/threads/{id}/runs/stream`）。这里用真 TCP loopback 到本机的 127.0.0.1
|
||
# + Gateway 监听端口。
|
||
#
|
||
# ⚠️ 历史踩坑(三个) ⚠️
|
||
# 1. 直接走系统代理:开发机若设了 HTTP_PROXY / HTTPS_PROXY,httpx 默认会把
|
||
# 127.0.0.1 也走代理 → 上游代理不可达,40s 超时(典型日志 headers_in=42.160s)。
|
||
# → 修法:client 用 `trust_env=False`,强制忽略代理 env vars。
|
||
#
|
||
# 2. 用 request.url.netloc 拼 base URL:Nginx 反代部署时,客户端打到 nginx 的
|
||
# public host(如 example.com),nginx 把请求转给 gateway 时 Host header
|
||
# 通常是 `gateway:8001` 或 public host;后者 httpx 会真的去 DNS 解析,
|
||
# 在 gateway 容器内可能不可达 → 连接失败。
|
||
# → 修法:**写死 127.0.0.1**,完全无视 request.url.netloc。
|
||
#
|
||
# 3. 一度尝试用 httpx.ASGITransport(app=request.app) 进程内直调,看似一劳永逸,
|
||
# 但 httpx 0.28 的 ASGITransport 对 streaming response **整段 buffer**,
|
||
# 一直收齐所有 body chunks 才一次性把响应给消费方 → SSE 完全失效,前端表现
|
||
# 为"调度阶段很长 + 流式结果一次性出现"。已弃用。
|
||
# → 详见:git log 6e3a07b(那次改 ASGI 引入)及随后回退。
|
||
#
|
||
# 端口策略(优先级,从高到低):
|
||
# 1. 环境变量 DEER_FLOW_GATEWAY_PORT
|
||
# ── 部署侧紧急覆盖通道,适合 docker-compose / k8s 注入或临时调试
|
||
# 2. config.yaml 的 `gateway.port`
|
||
# ── 部署侧**推荐**的配置方式,改完不需要重启二进制(get_app_config 有 mtime
|
||
# 热加载,但仍建议跑一次 reload 以确保所有 worker 拿到新值)
|
||
# 3. request.url.port
|
||
# ── 浏览器直连 gateway:8001 时拿得到;走 Nginx 80/443 反代时会落到下一档
|
||
# 4. 默认 8001(与 `make dev` / `make gateway` / `scripts/start-all.sh` 写死的一致)
|
||
#
|
||
# 端口在哪里改(给运维 / 部署侧的提示):
|
||
# - yaml 配置:打开 config.yaml,加 / 改:
|
||
# gateway:
|
||
# port: 9999
|
||
# - 环境变量:
|
||
# export DEER_FLOW_GATEWAY_PORT=9999
|
||
#
|
||
# 修改这两个函数时务必跑一次 SSE 流式 smoke 验证(见 multi-agent-backend-dev.md
|
||
# §13 测试建议),确认 chunks 是逐个到达而不是一次喷出。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# 写死的最终 fallback。**不要**为了改端口而修改这个常量——优先用 config.yaml /
|
||
# env 这两层(它们在部署侧不需要改源码即可生效)。这个值只在 yaml 缺失 + env
|
||
# 没设 + 请求里也拿不到合法 port 时才会用到。
|
||
_DEFAULT_GATEWAY_PORT = 8001
|
||
|
||
|
||
def _resolve_gateway_port(request: Request) -> int:
|
||
"""按四档 fallback 解析 Gateway 监听端口。详见模块顶部「端口策略」。"""
|
||
# 第 1 档:环境变量。允许非法值(garbage),只警告并继续下一档,不抛异常。
|
||
env_port = os.environ.get("DEER_FLOW_GATEWAY_PORT", "").strip()
|
||
if env_port:
|
||
try:
|
||
port = int(env_port)
|
||
if 1 <= port <= 65535:
|
||
return port
|
||
raise ValueError("out of range")
|
||
except ValueError:
|
||
logger.warning(
|
||
"DEER_FLOW_GATEWAY_PORT=%r 不是合法端口,fallback 到 config.yaml / request / %d",
|
||
env_port,
|
||
_DEFAULT_GATEWAY_PORT,
|
||
)
|
||
|
||
# 第 2 档:config.yaml 的 gateway.port。GatewayConfig.port 的 pydantic 校验
|
||
# 已经保证它在 1-65535 范围内,无需再次校验。读 config 失败时降级,不阻断业务。
|
||
try:
|
||
configured = get_app_config().gateway.port
|
||
if configured:
|
||
return configured
|
||
except Exception as exc:
|
||
# 极端场景:config.yaml 不存在或解析失败。这种情况下 gateway 本身也起不来,
|
||
# loopback 端口"猜错"也已经无所谓。仅记 debug 日志方便排查。
|
||
logger.debug("read gateway.port from app config failed: %s", exc)
|
||
|
||
# 第 3 档:从当前请求拿 port。浏览器直连 gateway 时是真实监听端口;
|
||
# 走 Nginx 反代时通常是 80/443(public port),那就跳过,落到最终默认。
|
||
inbound_port = request.url.port
|
||
if inbound_port is not None and inbound_port not in (80, 443):
|
||
return inbound_port
|
||
|
||
# 第 4 档:写死默认。
|
||
return _DEFAULT_GATEWAY_PORT
|
||
|
||
|
||
def _loopback_base(request: Request) -> str:
|
||
"""返回 `http://127.0.0.1:<port>`,供 loopback 调用拼 URL。
|
||
|
||
强制 127.0.0.1 是为了避免 Nginx 反代场景下读到经过 Host 改写的 netloc。
|
||
端口按 `_resolve_gateway_port` 的四档 fallback 解析。
|
||
"""
|
||
return f"http://127.0.0.1:{_resolve_gateway_port(request)}"
|
||
|
||
|
||
def _make_loopback_client(request: Request) -> httpx.AsyncClient:
|
||
"""构造同进程 loopback 用的 httpx.AsyncClient(真 TCP)。
|
||
|
||
所有圆桌路由(intent / recommend / multi_agent)的 loopback 调用必须走这里,
|
||
保持配置一致。
|
||
- `trust_env=False`:**关键**,忽略 HTTP_PROXY / HTTPS_PROXY,防代理拦截 127.0.0.1
|
||
- `timeout=None`:SSE 长连接需要,LangGraph runs/stream 可能跑数十秒
|
||
- `request` 参数当前未被使用,保留是为了:(a) 未来若需要从 request 取部署
|
||
上下文(如多租户端口隔离)可以无侵入扩展;(b) 与调用方现有签名兼容。
|
||
"""
|
||
return httpx.AsyncClient(
|
||
timeout=None,
|
||
trust_env=False,
|
||
)
|
||
|
||
|
||
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:
|
||
"""``POST /api/threads`` → 新 thread_id（intent/recommend/init 席位均用）。"""
|
||
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,
|
||
agent_id: str | None = None,
|
||
) -> tuple[bool, dict[str, Any] | str]:
|
||
payload: dict[str, Any] = {"name": name, "description": description, "soul": soul}
|
||
if agent_id is not None:
|
||
payload["id"] = agent_id
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 同进程直调 helpers(仅供 init_multi_agent 用,不要扩散到其它地方)
|
||
#
|
||
# init 阶段串了 N+1 个 thread create + N 个 GET /api/agents + 1 个 POST /api/agents,
|
||
# 用 HTTP loopback 串走每次都得 TCP + 鉴权中间件 + Pydantic 反序列化一遍。每个
|
||
# 调用 10-30ms,加起来对前端就是「点进 Step 2 后席位灯一颗一颗亮」的卡顿来源。
|
||
#
|
||
# 这里把这三个调用改成直接吃 app.state 里的 store/checkpointer —— 跑在同一个事
|
||
# 件循环里,省掉 TCP + 序列化 + 中间件链。鉴权 / 校验 / 文件副作用按需在直调
|
||
# helper 里显式做掉,不要"绕过"任何一项。
|
||
#
|
||
# 设计边界:这里只覆盖 init 阶段。SSE 长连接(/runs/stream / state)留在 HTTP
|
||
# loopback,因为它们本就是别的进程模型(LangGraph 子树),抽出来风险大收益小。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _resolve_user_id(request: Request) -> str:
|
||
"""与 routers/agents._current_user_id 完全等价,直调路径里用。"""
|
||
user = getattr(request.state, "user", None)
|
||
if user is not None:
|
||
return str(user.id)
|
||
return get_effective_user_id()
|
||
|
||
|
||
async def _create_thread_direct(request: Request) -> str:
|
||
"""同进程版的 ``POST /api/threads``。
|
||
|
||
完全镜像 ``routers/threads.create_thread`` 的写盘动作(thread_meta +
|
||
空 checkpoint),但跳过 TCP + Pydantic 反序列化。仅供 init 阶段批量建席位
|
||
线程用,失败时抛 HTTPException 由 SSE 帧上报。
|
||
"""
|
||
from app.gateway.deps import get_checkpointer, get_thread_store
|
||
|
||
checkpointer = get_checkpointer(request)
|
||
thread_store = get_thread_store(request)
|
||
thread_id = str(uuid.uuid4())
|
||
now = now_iso()
|
||
|
||
try:
|
||
await thread_store.create(thread_id, assistant_id=None, metadata={})
|
||
except Exception:
|
||
logger.exception("[init-direct] thread_meta create failed for %s", thread_id)
|
||
raise HTTPException(status_code=500, detail="Failed to create thread")
|
||
|
||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||
ckpt_metadata = {
|
||
"step": -1,
|
||
"source": "input",
|
||
"writes": None,
|
||
"parents": {},
|
||
"created_at": now,
|
||
}
|
||
try:
|
||
await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {})
|
||
except Exception:
|
||
logger.exception("[init-direct] checkpoint create failed for %s", thread_id)
|
||
raise HTTPException(status_code=500, detail="Failed to create thread")
|
||
return thread_id
|
||
|
||
|
||
async def _get_agents_batch_direct(
|
||
request: Request,
|
||
agent_ids: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
"""同进程版的 N 个并行 ``GET /api/agents/{id}`` 合并为 1 次 DB 查询。
|
||
|
||
返回 ``{agent_id: {id, name, description}}``。**严格模式**:任一席位对当前
|
||
用户不可见就抛 404 —— 与原 HTTP loopback 的 ``_get_agent`` 行为一致(原路径
|
||
单查 404 会 raise,在外层被 except 捕获并 yield error 帧),不要悄悄回退。
|
||
"""
|
||
from app.gateway.deps import get_agent_store
|
||
|
||
store = get_agent_store(request)
|
||
user_id = _resolve_user_id(request)
|
||
records = await store.list_by_ids(agent_ids, user_id)
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for r in records:
|
||
rid = str(r.get("id") or "").lower()
|
||
if not rid:
|
||
continue
|
||
out[rid] = {
|
||
"id": rid,
|
||
"name": str(r.get("name") or rid),
|
||
"description": str(r.get("description") or ""),
|
||
}
|
||
missing = [sid for sid in agent_ids if sid.lower() not in out]
|
||
if missing:
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail=f"Agent(s) not visible to current user: {', '.join(missing)}",
|
||
)
|
||
return out
|
||
|
||
|
||
async def _create_coordinator_direct(
|
||
request: Request,
|
||
*,
|
||
agent_id: str,
|
||
name: str,
|
||
description: str,
|
||
soul: str,
|
||
model: str | None,
|
||
skills: list[str] | None,
|
||
) -> tuple[bool, dict[str, Any] | str]:
|
||
"""同进程版的协调 agent 创建。
|
||
|
||
完整复刻 ``routers/agents.create_agent_endpoint`` 的关键副作用:
|
||
- ``agents_api.enabled`` master switch
|
||
- 显示名校验(空白则 422 → 这里转 (False, msg))
|
||
- 重复 id 检查(409 → (False, msg))
|
||
- 写 ``config.yaml`` + ``SOUL.md``(用 ``asyncio.to_thread`` 防同步 IO 阻塞)
|
||
- ``agent_store.create`` 入表
|
||
|
||
返回 ``(True, record)`` 或 ``(False, error_text)``;调用方决定要不要发 SSE error 帧。
|
||
"""
|
||
from app.gateway.deps import get_agent_store
|
||
from app.gateway.routers.agents import (
|
||
_require_agents_api_enabled,
|
||
_write_agent_config,
|
||
_write_agent_soul,
|
||
)
|
||
from deerflow.config.paths import get_paths
|
||
|
||
_require_agents_api_enabled()
|
||
normalized_name = (name or "").strip()
|
||
if not normalized_name:
|
||
return False, "Agent name cannot be empty"
|
||
|
||
store = get_agent_store(request)
|
||
if await store.get_any(agent_id) is not None:
|
||
return False, f"Agent id '{agent_id}' already exists"
|
||
|
||
agent_dir = get_paths().agent_dir(agent_id)
|
||
if agent_dir.exists():
|
||
return False, f"Agent id '{agent_id}' already exists"
|
||
|
||
try:
|
||
# config.yaml / SOUL.md 是同步 IO —— 全部丢线程池,不阻塞 init SSE 的发帧节奏。
|
||
await asyncio.to_thread(
|
||
_write_agent_config,
|
||
agent_id,
|
||
name=normalized_name,
|
||
description=description or "",
|
||
skills=skills,
|
||
model=model,
|
||
tool_groups=None,
|
||
)
|
||
await asyncio.to_thread(_write_agent_soul, agent_id, soul)
|
||
record = await store.create(
|
||
{
|
||
"id": agent_id,
|
||
"name": normalized_name,
|
||
"description": description or "",
|
||
"user_id": _resolve_user_id(request),
|
||
"published": False,
|
||
}
|
||
)
|
||
return True, record
|
||
except Exception as exc:
|
||
# 与 routers/agents 失败回滚一致 —— 文件已写出但 DB 失败时清掉 dir,避免下次
|
||
# `agent_dir.exists()` 误判为"已存在"导致后续 init 永远 409。
|
||
try:
|
||
if agent_dir.exists():
|
||
import shutil
|
||
|
||
await asyncio.to_thread(shutil.rmtree, agent_dir)
|
||
except Exception:
|
||
logger.exception("[init-direct] cleanup after coordinator create failure")
|
||
logger.exception("[init-direct] coordinator create failed")
|
||
return False, str(exc)
|
||
|
||
|
||
async def _append_thread_message(
|
||
client: httpx.AsyncClient,
|
||
base: str,
|
||
headers: dict[str, str],
|
||
thread_id: str,
|
||
content: str,
|
||
) -> None:
|
||
"""向指定 thread 的 checkpoint 追加一条 human 消息（广播上下文，不触发 LLM run）。
|
||
|
||
失败只打 warning，不阻断主流程——广播是 best-effort，席位真正干活仍靠
|
||
前端对该席位 thread 发起的 ``/run/stream``。
|
||
"""
|
||
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:
|
||
"""并行向 ``agent_threads`` 中除 ``skip`` 外的所有 thread 追加同一条 human 消息。"""
|
||
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+"([^"]*)"')
|
||
|
||
# Mirrors the upstream agent-id validator in the LangGraph runtime. Reject early
|
||
# so callers get a clear 400 here instead of a stream that aborts with HTTP 422
|
||
# from the loopback /api/threads/.../runs/stream endpoint.
|
||
_AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
|
||
|
||
|
||
def _validate_agent_id(value: str, field: str) -> str:
|
||
if not _AGENT_ID_PATTERN.match(value):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=(
|
||
f"{field} '{value}' is not a valid agent id. "
|
||
f"Must match ^[A-Za-z0-9_-]+$ (use the hyphen-case id, not the display name)."
|
||
),
|
||
)
|
||
return value
|
||
|
||
|
||
def _extract_dispatch(tool_call: dict[str, Any]) -> tuple[str, str]:
|
||
"""Return ``(agent_name, task)`` parsed from a leader tool call.
|
||
|
||
Two formats are accepted:
|
||
|
||
1. **Structured** — ``{"name": "agent_orchestration", "args": {"agent_name": ..., "task": ...}}``.
|
||
This is what most LLMs emit when told to "use the agent_orchestration skill",
|
||
and the form the roundtable coordinator's SOUL prompt elicits.
|
||
|
||
2. **Legacy bash argv** — ``{"name": "bash", "args": {"command": '... .py "X" "Y" ...'}}``.
|
||
Kept as a fallback so older skills that dispatch through bash still work.
|
||
"""
|
||
name = (tool_call.get("name") or "").lower()
|
||
args = tool_call.get("args")
|
||
|
||
# Format 1: structured agent_orchestration call.
|
||
if name == "agent_orchestration" and isinstance(args, dict):
|
||
agent_name = str(args.get("agent_name") or "").strip()
|
||
task = str(args.get("task") or "").strip()
|
||
if agent_name and task:
|
||
return _strip_display_suffix(agent_name), task
|
||
|
||
# Format 2: legacy bash + .py "X" "Y".
|
||
match = _TOOL_CALL_PATTERN.search(str(args))
|
||
if match:
|
||
return _strip_display_suffix(match.group(1)), match.group(2)
|
||
|
||
return "", ""
|
||
|
||
|
||
def _strip_display_suffix(agent_name: str) -> str:
|
||
"""Tolerate models that paste the human-readable suffix into agent_name.
|
||
|
||
The coordinator SOUL lists each seat as ``agent_name: roundtable-x（中文）：...``
|
||
so weaker models sometimes echo the entire ``roundtable-x（中文）`` label
|
||
back as the dispatch target. Strip everything from the first non-id
|
||
character onward so we recover the bare hyphen-case id; if what's left
|
||
still violates the id pattern we hand it back untouched and let the
|
||
upstream validator reject it.
|
||
"""
|
||
head = re.split(r"[^A-Za-z0-9_-]", agent_name, maxsplit=1)[0]
|
||
return head or agent_name
|
||
|
||
|
||
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],
|
||
thinking_enabled: bool = True,
|
||
reasoning_effort: str = "medium",
|
||
) -> dict[str, Any]:
|
||
"""构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。
|
||
|
||
- ``context.agent_name``：本次 run 使用的 agent（内置 id 或 init 创建的 coordinator id）。
|
||
- ``excluded_tools``：按路由裁剪工具面（intent 仅澄清、special 禁派活等）。
|
||
- ``skill_stop_names``：leader 侧配合 SkillStopMiddleware 在 ``agent_orchestration``
|
||
调用后截断图，以便网关从 AIMessage.tool_calls 解析派活列表。
|
||
- ``thinking_enabled`` / ``reasoning_effort``：默认开 CoT + medium，适合 Step 1 澄清
|
||
与子智能体交付这类需要推理的场景。**推荐弹窗与总控派活**是短决策任务，调用方
|
||
应传 ``thinking_enabled=False`` + ``reasoning_effort="low"``，省掉模型先吐
|
||
reasoning tokens 才开口的时间，显著降低前端"调度中"首字节延迟。
|
||
"""
|
||
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": reasoning_effort,
|
||
"thinking_enabled": thinking_enabled,
|
||
"is_plan_mode": False,
|
||
"subagent_enabled": False,
|
||
"thread_id": thread_id,
|
||
# 圆桌四个 router(intent / recommend / leader / special)都是临时任务,
|
||
# 用户个人记忆与本次研讨无关。MemoryMiddleware 的 Hindsight recall 是
|
||
# 同步阻塞 HTTP 调用(单次 timeout 15s),如果记忆服务慢/不通会让前端
|
||
# "调度中"卡很久;这里默认跳过,builtin USER.md/MEMORY.md 仍通过系统
|
||
# 提示注入不受影响。
|
||
"memory_recall_disabled": True,
|
||
},
|
||
"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,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /api/multi-agent/init — Step 2 席位 + 总控初始化（SSE）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.post("/init")
|
||
async def init_multi_agent(request: Request, payload: dict = Body(...)) -> StreamingResponse:
|
||
"""流式初始化圆桌 Step 2 所需的 thread 映射与协调智能体。
|
||
|
||
请求体::
|
||
|
||
{
|
||
"agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id
|
||
"main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id，前端生成
|
||
"model": "<可选>" // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0]
|
||
}
|
||
|
||
SSE 事件（每条 ``data: {...}\\n\\n``）::
|
||
|
||
seat_ready — 某一席位 thread 已创建（``asyncio.as_completed`` 顺序，先完先发）
|
||
coordinator_ready — 总控 agent + thread 已就绪
|
||
init_done — ``thread_ids`` 全量映射，前端存 ``threadIds`` 并开始编排
|
||
error — 失败详情（同步校验失败则直接 HTTP 4xx，不进 SSE）
|
||
|
||
协调智能体 SOUL 在运行时根据各席位 ``GET /api/agents/{id}`` 的 name/description 拼接，
|
||
强调 ``agent_name`` 必须用英文 id（``roundtable-intelligence``），禁止带中文括号后缀。
|
||
"""
|
||
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")
|
||
|
||
for name in agent_names:
|
||
_validate_agent_id(name, "agent_names entry")
|
||
_validate_agent_id(main_agent_name, "main_agent_name")
|
||
|
||
def _frame(payload: dict[str, Any]) -> str:
|
||
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
# init 阶段 fast path:全部走同进程直调(_create_thread_direct /
|
||
# _get_agents_batch_direct / _create_coordinator_direct),不再走 HTTP
|
||
# loopback —— TCP + 鉴权中间件 + Pydantic 反序列化每次 10-30ms,N+1 个席位
|
||
# 累计就是"点进 Step 2 后席位一颗颗亮"的卡顿来源。
|
||
try:
|
||
t_init_start = time.perf_counter()
|
||
# 1) 一次 DB 查询拿到所有席位的 name/description,替代之前的 N 个并行
|
||
# GET /api/agents/{id}。SOUL 拼接需要描述,这里一次性拿齐。
|
||
agent_meta = await _get_agents_batch_direct(request, agent_names)
|
||
t_meta_done = time.perf_counter()
|
||
|
||
# 2) 每个席位 thread 并行创建(直调 thread_store + checkpointer),
|
||
# 用 asyncio.as_completed 保持"先完先发 seat_ready"的体验。
|
||
async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]:
|
||
thread_id = await _create_thread_direct(request)
|
||
meta = agent_meta.get(sub_id.lower(), {})
|
||
return (
|
||
sub_id.lower(),
|
||
thread_id,
|
||
str(meta.get("name") or sub_id),
|
||
str(meta.get("description") or ""),
|
||
)
|
||
|
||
tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names]
|
||
prepared: list[tuple[str, str, str, str]] = []
|
||
try:
|
||
for fut in asyncio.as_completed(tasks):
|
||
sid, thread_id, display_name, description = await fut
|
||
prepared.append((sid, thread_id, display_name, description))
|
||
yield _frame({
|
||
"event": "seat_ready",
|
||
"agent_name": sid,
|
||
"thread_id": thread_id,
|
||
"display_name": display_name,
|
||
"description": description,
|
||
})
|
||
except Exception:
|
||
# Cancel any still-pending sibling so partially-prepared seats
|
||
# don't keep running past the stream close.
|
||
for t in tasks:
|
||
if not t.done():
|
||
t.cancel()
|
||
raise
|
||
t_seats_done = time.perf_counter()
|
||
|
||
agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
|
||
descriptions: list[str] = [
|
||
f"- agent_name: {sid}（{display_name}）：{description}"
|
||
for sid, _, display_name, description in prepared
|
||
]
|
||
|
||
# 动态 SOUL：把本次 init 的席位列表写进总控 prompt，供 agent_orchestration 派活。
|
||
soul = (
|
||
"#核心定位：\n"
|
||
"用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
|
||
"当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
|
||
"#调度方法：\n"
|
||
"通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。"
|
||
"注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence,"
|
||
"或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID）,"
|
||
"不要把括号里的中文显示名拼进 agent_name 里。\n"
|
||
"#关键约束(必读)：\n"
|
||
"1. **不要对同一子智能体重复派同样的活**：每次派活前回顾对话历史,"
|
||
"如果某个子智能体已经针对你正要派的任务完成过交付("
|
||
"看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息),"
|
||
"就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n"
|
||
"2. **看到交付要会用**：子智能体的交付内容会作为 human 消息追加到你的对话历史里,"
|
||
"你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),"
|
||
"让讨论收敛。\n"
|
||
"3. **避免无意义循环**：如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n"
|
||
"#可调度智能体：\n"
|
||
+ "\n".join(descriptions)
|
||
)
|
||
|
||
# 3) coordinator thread + agent 并行创建。thread 与 agent 之间无依赖,
|
||
# asyncio.gather 比串行省一个直调往返的开销。
|
||
coord_thread_task = asyncio.create_task(_create_thread_direct(request))
|
||
coord_agent_task = asyncio.create_task(
|
||
_create_coordinator_direct(
|
||
request,
|
||
agent_id=main_agent_name,
|
||
name=main_agent_name,
|
||
description="你是总控控制协调智能体,负责协调其他智能体干活",
|
||
soul=soul,
|
||
model=model,
|
||
skills=["agent_orchestration"],
|
||
)
|
||
)
|
||
coord_thread = await coord_thread_task
|
||
agent_threads[main_agent_name] = coord_thread
|
||
ok, result = await coord_agent_task
|
||
if not ok and isinstance(result, str) and "exists" not in result:
|
||
yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"})
|
||
return
|
||
t_coord_done = time.perf_counter()
|
||
logger.info(
|
||
"[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d",
|
||
t_meta_done - t_init_start,
|
||
t_seats_done - t_meta_done,
|
||
t_coord_done - t_seats_done,
|
||
t_coord_done - t_init_start,
|
||
len(prepared),
|
||
)
|
||
|
||
yield _frame({
|
||
"event": "coordinator_ready",
|
||
"main_agent_name": main_agent_name,
|
||
"thread_id": coord_thread,
|
||
})
|
||
yield _frame({
|
||
"event": "init_done",
|
||
"main_agent_name": main_agent_name,
|
||
"thread_ids": agent_threads,
|
||
})
|
||
except HTTPException as exc:
|
||
logger.warning("multi-agent init HTTPException: %s %s", exc.status_code, exc.detail)
|
||
yield _frame({"event": "error", "detail": f"[{exc.status_code}] {exc.detail}"})
|
||
except Exception as exc:
|
||
logger.exception("multi-agent init unexpected failure")
|
||
yield _frame({"event": "error", "detail": f"unexpected: {exc!r}"})
|
||
|
||
return StreamingResponse(
|
||
stream_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 上游 LangGraph SSE 透传（intent / recommend / leader / special 共用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
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]]:
|
||
"""消费 ``/api/threads/{thread_id}/runs/stream``，逐行 yield 原始 SSE。
|
||
|
||
产出元组：
|
||
- ``(line, None)`` — 可原样转发给浏览器（含 ``event:`` / ``data:`` 行）；
|
||
- ``(None, captured)`` — 流结束，``captured`` 为所有 ``data:`` 行文本列表，
|
||
供 ``_parse_last_messages`` 从倒数 values 快照提取 ``messages``。
|
||
|
||
日志 ``[upstream-timing]``：排查首 token 慢；headers_in 长尾若 >> 网络往返(loopback
|
||
本机往返通常 <10ms),怀疑 HTTP_PROXY 拦了回环——检查 client 是否走 _make_loopback_client
|
||
(trust_env=False)。
|
||
"""
|
||
|
||
url = f"{base}/api/threads/{thread_id}/runs/stream"
|
||
captured: list[str] = []
|
||
t_send = time.perf_counter()
|
||
async with client.stream("POST", url, json=body, headers=headers) as resp:
|
||
t_headers = time.perf_counter()
|
||
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}",
|
||
)
|
||
agent_for_log = (body.get("context") or {}).get("agent_name") or thread_id
|
||
logger.info(
|
||
"[upstream-timing] %s headers_in=%.3fs",
|
||
agent_for_log,
|
||
t_headers - t_send,
|
||
)
|
||
first_line = True
|
||
async for line in resp.aiter_lines():
|
||
if not line:
|
||
continue
|
||
if first_line:
|
||
first_line = False
|
||
logger.info(
|
||
"[upstream-timing] %s first_line=%.3fs (LangGraph runtime startup + LLM first byte)",
|
||
agent_for_log,
|
||
time.perf_counter() - t_headers,
|
||
)
|
||
yield line, None
|
||
if line.startswith("data:"):
|
||
captured.append(line)
|
||
yield None, captured
|
||
|
||
|
||
def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]]:
|
||
"""Return the full ``messages`` list from the most recent ``values`` snapshot.
|
||
|
||
The upstream sends the final ``values`` frame second-to-last (the very last
|
||
line is the ``end`` event with ``null``). We tolerate trailing non-JSON
|
||
frames (e.g. ``data: null``) by scanning backwards for the first frame that
|
||
successfully parses as a state object with a ``messages`` key.
|
||
"""
|
||
if not captured:
|
||
raise HTTPException(status_code=502, detail="upstream produced no data lines")
|
||
|
||
for raw_line in reversed(captured):
|
||
raw = raw_line[5:].strip() if raw_line.startswith("data:") else raw_line.strip()
|
||
if not raw or raw == "null":
|
||
continue
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if isinstance(parsed, dict) and "messages" in parsed:
|
||
messages = parsed.get("messages") or []
|
||
if isinstance(messages, list):
|
||
return messages
|
||
|
||
raise HTTPException(status_code=502, detail="no parseable values frame in upstream stream")
|
||
|
||
|
||
def _find_trailing_ai_text(messages: list[dict[str, Any]], *, exclude_text: str = "") -> str:
|
||
"""Return the text content of the latest AIMessage with no tool_calls.
|
||
|
||
Used by the clarification branch: ClarificationMiddleware ends the run
|
||
after emitting the structured ToolMessage, but the model often produces a
|
||
richer human-readable follow-up AIMessage right before the interrupt.
|
||
Prefer that text over the brief preamble that accompanied the
|
||
ask_clarification tool_call. When the only trailing AIMessage matches the
|
||
preamble (``exclude_text``), return empty so the caller can fall back to
|
||
the question itself.
|
||
"""
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
|
||
continue
|
||
if msg.get("tool_calls"):
|
||
continue
|
||
text = _flatten_content(msg.get("content", "")).strip()
|
||
if not text:
|
||
continue
|
||
if exclude_text and text == exclude_text.strip():
|
||
continue
|
||
return text
|
||
return ""
|
||
|
||
|
||
def _find_dispatch_message(messages: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""Locate the latest AIMessage from THIS turn.
|
||
|
||
Trailing ToolMessages (skill_stop / clarification ack injected by
|
||
``SkillStopMiddleware`` or ``ClarificationMiddleware``) are skipped because
|
||
the actionable tool_calls live on the AIMessage that emitted them. We stop
|
||
at the **first AIMessage encountered** — its ``tool_calls`` (or lack
|
||
thereof) is the authoritative signal for this turn.
|
||
|
||
历史踩坑:之前实现是"一直往前翻直到撞到带 tool_calls 的 AI",看似优雅,
|
||
实际上让 leader 永远收敛不了:本轮 LLM 不论怎么共识,只要历史上派过活,
|
||
上一轮的派活 AI 就会被翻出来再用一次,前端会把同样的 task 又发给子智能体,
|
||
陷入死循环。现在停在最新一条 AI 上,把判定交给调用方:`tool_calls=[]` →
|
||
共识分支,有 tool_calls → 派活分支。
|
||
"""
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
msg_type = (msg.get("type") or "").lower()
|
||
if msg_type in ("tool", "toolmessage"):
|
||
continue # 跳过末尾 skill_stop / clarification ack
|
||
if msg_type in ("ai", "aimessage", "aimessagechunk"):
|
||
return msg
|
||
break # 撞到 Human/其他类型,本轮没 AI(异常路径,留空让上游报错)
|
||
return {}
|
||
|
||
|
||
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]]:
|
||
"""总控（leader）单轮：透传上游 SSE + 末尾 yield 一个 dict 状态帧（由路由层 ``json.dumps``）。
|
||
|
||
时序要点：
|
||
1. **预广播**（与 LLM 并行）：把用户/综合消息写入其它席位 thread，skip 总控自身；
|
||
2. 跑总控 thread 的 stream，yield 每一行原始 SSE；
|
||
3. 解析 messages → 澄清 / 空 dispatch（共识）/ 派活列表；
|
||
4. 若有派活，**派活广播**后再 yield ``{status: dispatched, ...}``。
|
||
|
||
``skill_stop_names`` 会与 ``agent_orchestration`` 合并，确保 middleware 在工具调用后停图。
|
||
"""
|
||
t0 = time.perf_counter()
|
||
|
||
# 与上游 LLM 并行，减少「总控已开始想但席位 thread 里还没有用户消息」的窗口。
|
||
pre_broadcast_task = asyncio.create_task(
|
||
_broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name])
|
||
)
|
||
thread_id = agent_threads[agent_name]
|
||
effective_stop_names = list({"agent_orchestration", *skill_stop_names})
|
||
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=effective_stop_names,
|
||
# 总控派活是"下一步派谁、什么任务"的短决策，不需要 CoT。
|
||
# 关掉 thinking + low reasoning 让"调度中"几乎瞬间转入"正在派活"，
|
||
# 子智能体仍保留 thinking（在 _special_run 里默认 True），保证交付质量。
|
||
thinking_enabled=False,
|
||
reasoning_effort="low",
|
||
)
|
||
t_payload_built = time.perf_counter()
|
||
logger.info(
|
||
"[leader-timing] %s payload_built=%.3fs (broadcast running in background)",
|
||
agent_name,
|
||
t_payload_built - t0,
|
||
)
|
||
|
||
captured: list[str] = []
|
||
first_line_ts: float | None = None
|
||
line_count = 0
|
||
async for line, done in _stream_upstream(client, base, headers, thread_id, body):
|
||
if line is not None:
|
||
if first_line_ts is None:
|
||
first_line_ts = time.perf_counter()
|
||
logger.info(
|
||
"[leader-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
|
||
agent_name,
|
||
first_line_ts - t_payload_built,
|
||
first_line_ts - t0,
|
||
)
|
||
line_count += 1
|
||
yield line
|
||
elif done is not None:
|
||
captured = done
|
||
t_stream_done = time.perf_counter()
|
||
logger.info(
|
||
"[leader-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
|
||
agent_name,
|
||
t_stream_done - (first_line_ts or t_payload_built),
|
||
line_count,
|
||
t_stream_done - t0,
|
||
)
|
||
|
||
messages = _parse_last_messages(captured)
|
||
# Pre-broadcast must finish before we yield the final status frame on any
|
||
# branch — the seats that get dispatched next rely on having seen the
|
||
# original user message. The upstream LLM stream usually takes several
|
||
# seconds, so the broadcast has typically already completed and this
|
||
# await is effectively free.
|
||
t_before_broadcast_await = time.perf_counter()
|
||
try:
|
||
await pre_broadcast_task
|
||
except Exception as exc:
|
||
logger.warning("pre-broadcast failed: %s", exc)
|
||
t_after_broadcast_await = time.perf_counter()
|
||
logger.info(
|
||
"[leader-timing] %s pre_broadcast_await=%.3fs (>0 means broadcast was still in flight after LLM finished)",
|
||
agent_name,
|
||
t_after_broadcast_await - t_before_broadcast_await,
|
||
)
|
||
|
||
dispatch_msg = _find_dispatch_message(messages)
|
||
leader_text = _flatten_content(dispatch_msg.get("content", ""))
|
||
tool_calls = dispatch_msg.get("tool_calls") or []
|
||
|
||
# 澄清分支：与 intent.py 相同字段；status 为字符串 "clarification"（非数组），
|
||
# 前端据此暂停 runOrchestration 循环，展示 Step 2 澄清条。
|
||
clarification_call = next(
|
||
(tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"),
|
||
None,
|
||
)
|
||
if clarification_call is not None:
|
||
cargs = clarification_call.get("args") or {}
|
||
question = str(cargs.get("question") or "").strip()
|
||
follow_up = _find_trailing_ai_text(messages, exclude_text=leader_text)
|
||
yield {
|
||
"status": "clarification",
|
||
"content": follow_up or question or leader_text,
|
||
"question": question,
|
||
"clarification_type": str(cargs.get("clarification_type") or ""),
|
||
"clarification_context": str(cargs.get("context") or ""),
|
||
"options": cargs.get("options") or [],
|
||
"allow_custom": bool(cargs.get("allow_custom", True)),
|
||
"allow_multiple": resolve_allow_multiple(cargs),
|
||
"agent_name": agent_name,
|
||
}
|
||
return
|
||
|
||
if not tool_calls:
|
||
# 空 status 数组 = 总控不再派活，前端视为共识（hasConsensus）。
|
||
yield {"status": [], "content": leader_text, "agent_name": agent_name}
|
||
return
|
||
|
||
# 派活广播：告知各席位「总控给 X 派了什么活」，与即将发起的 special run 配合。
|
||
# sets and we don't care about ordering, so serializing them just
|
||
# multiplied latency before the frontend could start a sub-agent run.
|
||
dispatched: list[list[str]] = []
|
||
dispatch_broadcasts: list[asyncio.Task[None]] = []
|
||
for tool_call in tool_calls:
|
||
try:
|
||
sub_name, task_text = _extract_dispatch(tool_call)
|
||
if not sub_name:
|
||
continue
|
||
dispatch_broadcasts.append(
|
||
asyncio.create_task(
|
||
_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)
|
||
if dispatch_broadcasts:
|
||
t_before_dispatch = time.perf_counter()
|
||
await asyncio.gather(*dispatch_broadcasts, return_exceptions=True)
|
||
logger.info(
|
||
"[leader-timing] %s dispatch_broadcasts_await=%.3fs count=%d (gates how soon the next sub agent can start)",
|
||
agent_name,
|
||
time.perf_counter() - t_before_dispatch,
|
||
len(dispatch_broadcasts),
|
||
)
|
||
yield {"status": dispatched, "content": leader_text, "agent_name": agent_name}
|
||
|
||
|
||
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,
|
||
*,
|
||
display_name: str = "",
|
||
) -> AsyncIterator[str | dict[str, Any]]:
|
||
"""子智能体（special）单轮：执行总控下发的 task 文本，交付 content。
|
||
|
||
- ``excluded_tools`` 禁用派活/澄清等，避免席位抢总控活；
|
||
- 终态 ``status`` 为描述性字符串（非数组），``content`` 为交付正文；
|
||
- **后广播**在 yield 终态帧之后 await，保证其它席位 state 里有完成摘要，
|
||
同时尽量让前端先收到结果关闭 loading。
|
||
"""
|
||
t0 = time.perf_counter()
|
||
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", "agent_orchestration"],
|
||
skill_stop_names=skill_stop_names,
|
||
)
|
||
t_payload_built = time.perf_counter()
|
||
logger.info(
|
||
"[special-timing] %s payload_built=%.3fs",
|
||
agent_name,
|
||
t_payload_built - t0,
|
||
)
|
||
|
||
captured: list[str] = []
|
||
first_line_ts: float | None = None
|
||
line_count = 0
|
||
async for line, done in _stream_upstream(client, base, headers, thread_id, body):
|
||
if line is not None:
|
||
if first_line_ts is None:
|
||
first_line_ts = time.perf_counter()
|
||
logger.info(
|
||
"[special-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
|
||
agent_name,
|
||
first_line_ts - t_payload_built,
|
||
first_line_ts - t0,
|
||
)
|
||
line_count += 1
|
||
yield line
|
||
elif done is not None:
|
||
captured = done
|
||
t_stream_done = time.perf_counter()
|
||
logger.info(
|
||
"[special-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
|
||
agent_name,
|
||
t_stream_done - (first_line_ts or t_payload_built),
|
||
line_count,
|
||
t_stream_done - t0,
|
||
)
|
||
|
||
messages = _parse_last_messages(captured)
|
||
# Sub-agent's final deliverable is the latest AIMessage's text content.
|
||
last_msg: dict[str, Any] = {}
|
||
for msg in reversed(messages):
|
||
if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
|
||
last_msg = msg
|
||
break
|
||
content = _flatten_content(last_msg.get("content", ""))
|
||
# Kick the "子智能体完成" broadcast off in the background and yield the
|
||
# status frame first — the next leader cycle that depends on this
|
||
# broadcast having landed only starts after a fresh /run/stream request
|
||
# from the frontend, which is many seconds away (full leader LLM call).
|
||
# Awaiting after the yield ensures the task finishes before the AsyncClient
|
||
# closes; the SSE connection stays open a bit longer but the frontend has
|
||
# already consumed the result.
|
||
t_before_post = time.perf_counter()
|
||
# 优先用 display_name(中文友好,例如"情报收集")提示 leader,id 放括号备查;
|
||
# display_name 缺失时退化为仅 agent_name —— 与旧行为一致。
|
||
speaker_label = f"{display_name}({agent_name})" if display_name else agent_name
|
||
post_broadcast_task = asyncio.create_task(
|
||
_broadcast(
|
||
client,
|
||
base,
|
||
headers,
|
||
f"子智能体{speaker_label}完成{new_message}工作,交付内容为:" + content,
|
||
agent_threads,
|
||
[agent_name],
|
||
)
|
||
)
|
||
yield {
|
||
"status": f"子智能体{speaker_label}完成{new_message}工作",
|
||
"content": content,
|
||
"agent_name": agent_name,
|
||
}
|
||
try:
|
||
await post_broadcast_task
|
||
except Exception as exc:
|
||
logger.warning("post-broadcast failed: %s", exc)
|
||
logger.info(
|
||
"[special-timing] %s post_broadcast_total=%.3fs (broadcast time after yield — keeps SSE open this long)",
|
||
agent_name,
|
||
time.perf_counter() - t_before_post,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /api/multi-agent/run/stream — leader / special 单轮 SSE
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.post("/run/stream")
|
||
async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) -> StreamingResponse:
|
||
"""驱动总控或子智能体的一轮 LangGraph run。
|
||
|
||
请求体::
|
||
|
||
{
|
||
"agent_type": "leader" | "special",
|
||
"d_agent_thread_id": { "<agent_id>": "<thread_id>", ... }, // init_done.thread_ids
|
||
"agent_name": "<本次运行的 agent_id>",
|
||
"new_message": "<用户消息或派活 task 文本>",
|
||
"model": "<可选>", // 不传由 lead_agent 回退到 config.yaml models[0]
|
||
"skill_stop_names": [] // 可选,leader 一般留空,由网关注入 agent_orchestration
|
||
}
|
||
|
||
响应：上游 SSE 行透传 + 末尾一条 ``data:`` JSON（leader 为 dict，special 同上）。
|
||
客户端应使用 AbortSignal 支持「挂起协商」时取消 in-flight 请求。
|
||
"""
|
||
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()
|
||
# 人类可读名,仅 special 用 —— 后端把它拼进 post-broadcast 文本,这样 leader
|
||
# 下一轮看到的是"子智能体[中文名](id)完成..."而非"子智能体[UUID]完成...";
|
||
# LLM 对 UUID 不友好,只用 id 会导致总控认不出已完成的子智能体,无限循环派活。
|
||
# 缺失时 fallback 到 agent_name,旧前端不传也不破坏。
|
||
display_name: str = str(payload.get("display_name") or "").strip()
|
||
new_message: str = str(payload.get("new_message") or "")
|
||
skill_stop_names: list[str] = payload.get("skill_stop_names") or []
|
||
# 不写死任何默认模型;空串会让 lead_agent._resolve_model_name() 回退到
|
||
# config.yaml 的 models[0],与 /api/ai-writing 等其它路由一致。
|
||
model_name: str = str(payload.get("model") or "")
|
||
|
||
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",
|
||
)
|
||
_validate_agent_id(agent_name, "agent_name")
|
||
for key in agent_threads:
|
||
_validate_agent_id(key, "d_agent_thread_id key")
|
||
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
# TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client：
|
||
# trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
|
||
async with _make_loopback_client(request) as client:
|
||
try:
|
||
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,
|
||
display_name=display_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"
|
||
except HTTPException as exc:
|
||
logger.warning("multi-agent stream HTTPException: %s %s", exc.status_code, exc.detail)
|
||
err_frame = {
|
||
"status": "error",
|
||
"content": f"[{exc.status_code}] {exc.detail}",
|
||
"agent_name": agent_name,
|
||
}
|
||
yield f"data: {json.dumps(err_frame, ensure_ascii=False)}\n\n"
|
||
except Exception as exc:
|
||
logger.exception("multi-agent stream unexpected failure")
|
||
err_frame = {
|
||
"status": "error",
|
||
"content": f"unexpected: {exc!r}",
|
||
"agent_name": agent_name,
|
||
}
|
||
yield f"data: {json.dumps(err_frame, ensure_ascii=False)}\n\n"
|
||
|
||
return StreamingResponse(
|
||
stream_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
},
|
||
)
|