918 lines
38 KiB
Python
918 lines
38 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 re
|
||
import time
|
||
from typing import Any, AsyncIterator
|
||
|
||
import httpx
|
||
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
|
||
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 辅助函数（Gateway → Gateway，同进程）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _loopback_base(request: Request) -> str:
|
||
"""当前请求的 scheme+host，用于 httpx 打回本机 Gateway。"""
|
||
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:
|
||
"""``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
|
||
|
||
|
||
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],
|
||
) -> 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 解析派活列表。
|
||
"""
|
||
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,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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": "deepseek-chat" // 可选，写入 coordinator agent
|
||
}
|
||
|
||
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")
|
||
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
|
||
def _frame(payload: dict[str, Any]) -> str:
|
||
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
# trust_env=False: same-process loopback to 127.0.0.1 — honoring
|
||
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines)
|
||
# causes httpx to route loopback calls through an unreachable upstream
|
||
# proxy and wait ~40 seconds for timeout before falling back.
|
||
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
|
||
try:
|
||
# Per-seat prepare task — same body as before, but we no longer
|
||
# await them with asyncio.gather. Instead, asyncio.as_completed
|
||
# surfaces them in REAL completion order so we can emit a
|
||
# seat_ready frame the moment each finishes.
|
||
async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]:
|
||
info = await _get_agent(client, base, headers, sub_id)
|
||
display_name = str(info.get("name") or sub_id)
|
||
description = info.get("description", "") or ""
|
||
thread_id = await _create_thread(client, base, headers)
|
||
return sub_id.lower(), thread_id, display_name, description
|
||
|
||
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 we don't leak tasks
|
||
# past the AsyncClient close.
|
||
for t in tasks:
|
||
if not t.done():
|
||
t.cancel()
|
||
raise
|
||
|
||
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），"
|
||
"不要把括号里的中文显示名拼进 agent_name 里。\n"
|
||
"#可调度智能体：\n"
|
||
+ "\n".join(descriptions)
|
||
)
|
||
|
||
coord_thread = await _create_thread(client, base, headers)
|
||
agent_threads[main_agent_name] = coord_thread
|
||
ok, result = await _create_agent(
|
||
client,
|
||
base,
|
||
headers,
|
||
name=main_agent_name,
|
||
description="你是总控控制协调智能体,负责协调其他智能体干活",
|
||
model=model,
|
||
skills=["agent_orchestration"],
|
||
soul=soul,
|
||
agent_id=main_agent_name,
|
||
)
|
||
if not ok and isinstance(result, str) and "exists" not in result:
|
||
yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"})
|
||
return
|
||
|
||
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 慢、代理导致 loopback 超时（见 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 that carries dispatch tool calls.
|
||
|
||
The very last message can be a skill-stop ToolMessage injected by
|
||
``SkillStopMiddleware`` when it intercepts the dispatch, or a clarification
|
||
ToolMessage. Either way, the actionable tool_calls live on the AIMessage
|
||
*before* it, so we walk backwards until we hit one with ``tool_calls``.
|
||
Returns the latest AIMessage when none of them carry tool_calls so the
|
||
caller can still extract its text content.
|
||
"""
|
||
last_ai: dict[str, Any] = {}
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
|
||
if not last_ai:
|
||
last_ai = msg
|
||
if msg.get("tool_calls"):
|
||
return msg
|
||
return last_ai
|
||
|
||
|
||
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,
|
||
)
|
||
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,
|
||
) -> 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()
|
||
post_broadcast_task = asyncio.create_task(
|
||
_broadcast(
|
||
client,
|
||
base,
|
||
headers,
|
||
f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + content,
|
||
agent_threads,
|
||
[agent_name],
|
||
)
|
||
)
|
||
yield {
|
||
"status": f"子智能体{agent_name}完成{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": "deepseek-chat",
|
||
"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()
|
||
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",
|
||
)
|
||
_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]:
|
||
# trust_env=False: same-process loopback to 127.0.0.1 — honoring
|
||
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines)
|
||
# causes httpx to route loopback calls through an unreachable upstream
|
||
# proxy and wait ~40 seconds for timeout before falling back.
|
||
async with httpx.AsyncClient(timeout=None, trust_env=False) 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,
|
||
)
|
||
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",
|
||
},
|
||
)
|