819 lines
32 KiB
Python
819 lines
32 KiB
Python
"""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
|
||
import time
|
||
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,
|
||
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:
|
||
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+"([^"]*)"')
|
||
|
||
# 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]:
|
||
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(...)) -> StreamingResponse:
|
||
"""Streaming /init.
|
||
|
||
Emits text/event-stream frames as each seat/coordinator becomes ready, so
|
||
the frontend can visualize the parallel bring-up rather than blocking on
|
||
one bulk JSON. Frames:
|
||
|
||
data: {"event":"seat_ready","agent_name":"...","thread_id":"...","display_name":"...","description":"..."}
|
||
data: {"event":"coordinator_ready","main_agent_name":"...","thread_id":"..."}
|
||
data: {"event":"init_done","main_agent_name":"...","thread_ids":{...}}
|
||
data: {"event":"error","detail":"..."} # on failure
|
||
|
||
Validation errors (missing agent_names, bad id pattern) still raise
|
||
HTTPException synchronously so the connection fails fast with a real 4xx
|
||
instead of a 200 SSE that just carries an error frame.
|
||
"""
|
||
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]:
|
||
async with httpx.AsyncClient(timeout=None) 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 = (
|
||
"#核心定位：\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",
|
||
},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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] = []
|
||
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]]:
|
||
# Fine-grained timing for the leader path. The user reported the
|
||
# coordinator "stalls before output" — we log each major await so we can
|
||
# see whether the wait is in pre-broadcast scheduling, upstream stream
|
||
# setup, first-token latency, or post-stream parsing.
|
||
t0 = time.perf_counter()
|
||
|
||
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 []
|
||
|
||
# Clarification branch: the coordinator called ask_clarification. The
|
||
# ClarificationMiddleware intercepts with Command(goto=END) and stashes
|
||
# the structured question in the tool_call args; sometimes a richer
|
||
# follow-up AIMessage is also produced after the ToolMessage. Emit a
|
||
# dedicated frame so the frontend can render it as a question instead of
|
||
# treating an empty dispatch as "consensus reached".
|
||
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)),
|
||
"agent_name": agent_name,
|
||
}
|
||
return
|
||
|
||
if not tool_calls:
|
||
yield {"status": [], "content": leader_text, "agent_name": agent_name}
|
||
return
|
||
|
||
# Run all dispatch broadcasts concurrently — they target disjoint thread
|
||
# 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]]:
|
||
# Same instrumentation as _leader_run so all 6 seats are diagnosable.
|
||
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,
|
||
)
|
||
|
||
|
||
@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",
|
||
)
|
||
_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]:
|
||
async with httpx.AsyncClient(timeout=None) 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",
|
||
},
|
||
)
|