248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""Server-side parallel fan-out for the standalone parallel multi-agent panel.
|
||
|
||
The Vue embeddable panel sends a single request — ``{task_name, brief, agent_ids}``
|
||
— and this endpoint **fans out** to every agent in parallel (file writes disabled),
|
||
then **merges** their LangGraph streams into ONE SSE tagged per ``agent_id``, so the
|
||
client only consumes one stream and never parses LangGraph internals.
|
||
|
||
It reuses the roundtable single-agent machinery from ``multi_agent``:
|
||
``_special_run(..., no_file=True)`` runs one seat under the ``seat_parallel`` run
|
||
policy (which excludes ``write_file`` / ``str_replace`` / ``bash`` / ``web_search``
|
||
/ ``agent_orchestration`` …), so parallel agents can't write files or dispatch.
|
||
|
||
Unified event protocol (one JSON object per ``data:`` frame)::
|
||
|
||
{"type": "agent_start", "agent_id": "..."}
|
||
{"type": "delta", "agent_id": "...", "text": "..."} # AI text increment
|
||
{"type": "step", "agent_id": "...", "key": "...", "label": "...", "args": {...}}
|
||
{"type": "agent_done", "agent_id": "...", "content": "..."}
|
||
{"type": "agent_error", "agent_id": "...", "message": "..."}
|
||
{"type": "all_done"}
|
||
|
||
The frame-normalisation (LangGraph ``messages-tuple`` → delta / tool step) mirrors
|
||
the frontend parser in ``frontend-web/src/roundtable-planning/api/multi-agent.ts``
|
||
(``extractAiTextDelta`` + ``tool_call_chunks`` accumulation).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from typing import Any, AsyncIterator
|
||
|
||
from fastapi import APIRouter, Body, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
|
||
from app.gateway.routers.multi_agent import (
|
||
_ROUNDTABLE_THREAD_METADATA,
|
||
_auth_headers,
|
||
_create_thread,
|
||
_loopback_base,
|
||
_make_loopback_client,
|
||
_special_run,
|
||
_validate_agent_id,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/parallel-agents", tags=["parallel-agents"])
|
||
|
||
# Soft cap on parallel degree — protects remote model concurrency + sandbox.
|
||
MAX_PARALLEL_AGENTS = 8
|
||
|
||
|
||
# ── frame normalisation (mirrors multi-agent.ts) ─────────────────────────────
|
||
|
||
|
||
def _extract_ai_text(chunk: dict[str, Any]) -> str:
|
||
"""AI text delta from a ``messages-tuple`` chunk obj; "" when not AI text."""
|
||
type_ = str(chunk.get("type") or "").lower()
|
||
if type_ not in ("ai", "aimessagechunk", "aimessage"):
|
||
return ""
|
||
content = chunk.get("content")
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, list):
|
||
out: list[str] = []
|
||
for item in content:
|
||
if isinstance(item, str):
|
||
out.append(item)
|
||
elif isinstance(item, dict):
|
||
t = item.get("text")
|
||
if isinstance(t, str):
|
||
out.append(t)
|
||
return "".join(out)
|
||
return ""
|
||
|
||
|
||
def _extract_tool_steps(chunk: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Tool-call steps from a chunk obj — both complete ``tool_calls`` and
|
||
incremental ``tool_call_chunks`` (we only need the name to surface a step)."""
|
||
steps: list[dict[str, Any]] = []
|
||
tool_calls = chunk.get("tool_calls")
|
||
if isinstance(tool_calls, list):
|
||
for tc in tool_calls:
|
||
if not isinstance(tc, dict):
|
||
continue
|
||
name = tc.get("name")
|
||
if not isinstance(name, str) or not name:
|
||
continue
|
||
ident = tc.get("id") or name
|
||
args = tc.get("args") if isinstance(tc.get("args"), dict) else None
|
||
steps.append({"key": f"tool:{ident}", "label": f"调用工具 {name}", "args": args})
|
||
tcc = chunk.get("tool_call_chunks")
|
||
if isinstance(tcc, list):
|
||
for c in tcc:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
name = c.get("name")
|
||
if not isinstance(name, str) or not name:
|
||
continue
|
||
idx = c.get("index") if isinstance(c.get("index"), int) else 0
|
||
ident = c.get("id") or f"{name}:{idx}"
|
||
steps.append({"key": f"tool:{ident}", "label": f"调用工具 {name}", "args": None})
|
||
return steps
|
||
|
||
|
||
def _build_task_message(task_name: str, brief: str) -> str:
|
||
parts: list[str] = []
|
||
if task_name.strip():
|
||
parts.append(f"任务名称：{task_name.strip()}")
|
||
if brief.strip():
|
||
parts.append(f"任务简介：\n{brief.strip()}")
|
||
parts.append(
|
||
"请你从你的专业职责出发，针对以上任务给出你的分析与交付。"
|
||
"本轮为**并行**研讨：请把你的产出**直接写在回复正文里**，"
|
||
"不要写文件、不要执行脚本/命令。"
|
||
)
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
# ── endpoint ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.post("/run/stream")
|
||
async def run_parallel_agents_stream(request: Request, payload: dict = Body(...)) -> StreamingResponse:
|
||
"""并行扇出 N 个智能体，合并成一条带 ``agent_id`` 标记的 SSE。文件写入固定禁用。"""
|
||
task_name: str = str(payload.get("task_name") or "").strip()
|
||
brief: str = str(payload.get("brief") or "").strip()
|
||
raw_ids = payload.get("agent_ids") or []
|
||
model_name: str = str(payload.get("model") or "")
|
||
|
||
if not isinstance(raw_ids, list) or not raw_ids:
|
||
raise HTTPException(status_code=400, detail="agent_ids 不能为空")
|
||
agent_ids: list[str] = []
|
||
for aid in raw_ids:
|
||
sid = str(aid or "").strip().lower()
|
||
_validate_agent_id(sid, "agent_ids")
|
||
if sid not in agent_ids:
|
||
agent_ids.append(sid)
|
||
if len(agent_ids) > MAX_PARALLEL_AGENTS:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"并行智能体数量上限 {MAX_PARALLEL_AGENTS}（当前 {len(agent_ids)}）",
|
||
)
|
||
if not task_name and not brief:
|
||
raise HTTPException(status_code=400, detail="任务名称与简介至少填写其一")
|
||
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
new_message = _build_task_message(task_name, brief)
|
||
|
||
async def _consume_agent(
|
||
client: Any,
|
||
agent_id: str,
|
||
queue: asyncio.Queue[dict[str, Any]],
|
||
) -> None:
|
||
await queue.put({"type": "agent_start", "agent_id": agent_id})
|
||
last_event = ""
|
||
seen_steps: set[str] = set()
|
||
buffer: list[str] = []
|
||
done_sent = False
|
||
try:
|
||
thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||
gen = _special_run(
|
||
client,
|
||
base,
|
||
headers,
|
||
{agent_id: thread_id},
|
||
agent_id,
|
||
new_message,
|
||
[],
|
||
model_name,
|
||
display_name="",
|
||
no_file=True, # seat_parallel policy → 禁 write_file/bash/str_replace…
|
||
)
|
||
async for chunk in gen:
|
||
if isinstance(chunk, dict):
|
||
# 终态帧 {status, content, agent_name}
|
||
content = str(chunk.get("content") or "")
|
||
await queue.put({"type": "agent_done", "agent_id": agent_id, "content": content})
|
||
done_sent = True
|
||
continue
|
||
line = chunk.strip() if isinstance(chunk, str) else ""
|
||
if line.startswith("event:"):
|
||
last_event = line[6:].strip()
|
||
continue
|
||
if not line.startswith("data:"):
|
||
continue
|
||
raw = line[5:].strip()
|
||
if not raw:
|
||
continue
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except Exception:
|
||
continue
|
||
if last_event == "messages" and isinstance(parsed, list) and parsed:
|
||
obj = parsed[0]
|
||
if isinstance(obj, dict):
|
||
text = _extract_ai_text(obj)
|
||
if text:
|
||
buffer.append(text)
|
||
await queue.put({"type": "delta", "agent_id": agent_id, "text": text})
|
||
for step in _extract_tool_steps(obj):
|
||
key = step["key"]
|
||
if key in seen_steps:
|
||
continue
|
||
seen_steps.add(key)
|
||
await queue.put({"type": "step", "agent_id": agent_id, **step})
|
||
if not done_sent:
|
||
# 流被抢占/截断,没收到终态帧 —— 用已累积正文兜底收口本席位。
|
||
await queue.put(
|
||
{"type": "agent_done", "agent_id": agent_id, "content": "".join(buffer)}
|
||
)
|
||
except Exception as exc: # noqa: BLE001 — 单个 agent 失败不拖垮其它
|
||
logger.warning("[parallel-agents] %s failed: %r", agent_id, exc)
|
||
await queue.put({"type": "agent_error", "agent_id": agent_id, "message": str(exc)})
|
||
finally:
|
||
await queue.put({"type": "__agent_end__", "agent_id": agent_id})
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
|
||
async with _make_loopback_client(request) as client:
|
||
tasks = [asyncio.create_task(_consume_agent(client, aid, queue)) for aid in agent_ids]
|
||
ended = 0
|
||
try:
|
||
while ended < len(agent_ids):
|
||
event = await queue.get()
|
||
if event.get("type") == "__agent_end__":
|
||
ended += 1
|
||
continue
|
||
yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
|
||
yield f"data: {json.dumps({'type': 'all_done'}, ensure_ascii=False)}\n\n"
|
||
finally:
|
||
for t in tasks:
|
||
if not t.done():
|
||
t.cancel()
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
return StreamingResponse(
|
||
stream_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
},
|
||
)
|