448 lines
21 KiB
Python
448 lines
21 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, then **merges** their
|
||
LangGraph streams into ONE SSE tagged per ``agent_id``, so the client only consumes
|
||
one stream and never parses LangGraph internals.
|
||
|
||
Each fan-out agent runs as a pure **executing seat**: ``excluded_tools`` =
|
||
``_EXCLUDED_TOOLS`` (``write_file`` / ``str_replace`` / ``agent_orchestration`` /
|
||
``bash``) so a seat can neither write files (parallel writes would clash) nor
|
||
dispatch / run scripts (a weak model fed "总控 / 业务链条" framing would otherwise
|
||
mimic the controller and "派活"). Read-only tools + ``ask_clarification`` stay on,
|
||
so the step bar and clarification cards still work.
|
||
|
||
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
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Body, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.routers.multi_agent import (
|
||
_ROUNDTABLE_THREAD_METADATA,
|
||
_auth_headers,
|
||
_create_thread,
|
||
_flatten_content,
|
||
_loopback_base,
|
||
_make_loopback_client,
|
||
_parse_last_messages,
|
||
_run_payload,
|
||
_stream_upstream,
|
||
_validate_agent_id,
|
||
)
|
||
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
|
||
|
||
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
|
||
|
||
# 禁两类工具：①**写文件**（write_file / str_replace）—— 并行席位互相写文件会打架；
|
||
# ②**派活 / 脚本逃逸**（agent_orchestration / bash）—— 本面板里每个智能体只做自己那
|
||
# 份分析交付，绝不该扮演「总控」去调度别人。弱模型（deepseek 等）拿到带「总控 / 业务
|
||
# 链条」语气的任务后，会概率性地用 agent_orchestration 真去派活，或绕道 bash 跑脚本
|
||
# 调内部接口模仿总控调度（bug：单步单智能体链条里子智能体输出「总控调度指令…」并用
|
||
# glob / bash 翻找派活脚本）。物理禁掉这两类即可根治。其余只读工具（ls / read_file /
|
||
# 检索）与 ask_clarification（协作卡片）保留：既不互相打架，又能正常出步骤条 / 要澄清。
|
||
_EXCLUDED_TOOLS: list[str] = [
|
||
"write_file", # 并行不互相写文件打架
|
||
"str_replace",
|
||
"agent_orchestration", # 子智能体不派活 / 不调度（防模仿总控）
|
||
"bash", # 禁跑脚本（防绕道脚本调内部接口模仿派活）
|
||
]
|
||
|
||
|
||
# ── 专用「绕过常规登录」取 token ──────────────────────────────────────────────
|
||
# 内网部署对标准 /api/v1/auth/login/username 有额外操作限制（口令门 / 网关裁剪等），
|
||
# 嵌入方拿不到 token。这里给并行多智能体面板单开一个**公共**端点：传用户名直接签发 token
|
||
# （账号不存在则自动注册，与 username 登录同一套逻辑）。已在 auth_middleware / csrf_middleware
|
||
# 的白名单里放行（公共 + CSRF 豁免）。⚠️ 这是有意的鉴权旁路，仅供内网部署的本功能使用。
|
||
|
||
|
||
class _TokenRequest(BaseModel):
|
||
username: str = Field(min_length=1, max_length=128)
|
||
|
||
|
||
class _TokenResponse(BaseModel):
|
||
access_token: str
|
||
token_type: str = "bearer"
|
||
user_id: str
|
||
|
||
|
||
@router.post("/auth/token", response_model=_TokenResponse)
|
||
async def issue_token(body: _TokenRequest) -> _TokenResponse:
|
||
"""传用户名 → 直接签发 access_token（绕过常规登录，公共端点）。"""
|
||
# 懒加载，避免与 auth 路由的模块级循环依赖。
|
||
from app.gateway.auth import create_access_token
|
||
from app.gateway.routers.auth import _get_or_create_user_by_username
|
||
|
||
user, _created = await _get_or_create_user_by_username(body.username.strip())
|
||
token = create_access_token(str(user.id), token_version=user.token_version)
|
||
return _TokenResponse(access_token=token, token_type="bearer", user_id=str(user.id))
|
||
|
||
|
||
# ── 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 _ingest_tool_chunks(chunks: Any, acc: dict[int, dict[str, Any]]) -> None:
|
||
"""累积 ``tool_call_chunks`` 的增量（name + args 字符串），按 index 归并
|
||
（mirror multi-agent.ts ``ingestToolCallChunks``）。"""
|
||
if not isinstance(chunks, list):
|
||
return
|
||
for c in chunks:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
idx = c.get("index") if isinstance(c.get("index"), int) else 0
|
||
entry = acc.setdefault(idx, {"id": None, "name": "", "argsStr": ""})
|
||
cid = c.get("id")
|
||
if isinstance(cid, str) and cid:
|
||
if entry["id"] and entry["id"] != cid: # 同 index 换了新调用 → 重置
|
||
entry["name"] = ""
|
||
entry["argsStr"] = ""
|
||
entry["id"] = cid
|
||
name = c.get("name")
|
||
if isinstance(name, str) and name:
|
||
entry["name"] = name
|
||
a = c.get("args")
|
||
if isinstance(a, str):
|
||
entry["argsStr"] += a
|
||
|
||
|
||
def _synthetic_tool_calls(acc: dict[int, dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""把累积的 argsStr 尽力 parse 成完整 args，产出 ``{id,name,args}``（未成完整 JSON 则跳过）。"""
|
||
out: list[dict[str, Any]] = []
|
||
for entry in acc.values():
|
||
if not entry["name"]:
|
||
continue
|
||
s = str(entry["argsStr"]).strip()
|
||
if not s:
|
||
continue
|
||
try:
|
||
args = json.loads(s)
|
||
except Exception:
|
||
continue
|
||
if isinstance(args, dict):
|
||
out.append({"id": entry["id"], "name": entry["name"], "args": args})
|
||
return out
|
||
|
||
|
||
def _clarification_from_messages(messages: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||
"""从最终 messages 里找最后一个 ``ask_clarification`` 工具调用，抽出结构化澄清载荷
|
||
（字段对齐 multi_agent leader 的 clarification 帧）。"""
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
|
||
continue
|
||
for tc in msg.get("tool_calls") or []:
|
||
if isinstance(tc, dict) and (tc.get("name") or "").lower() == "ask_clarification":
|
||
cargs = tc.get("args") if isinstance(tc.get("args"), dict) else {}
|
||
return {
|
||
"question": str(cargs.get("question") or ""),
|
||
"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),
|
||
}
|
||
return None
|
||
|
||
|
||
async def _run_one_agent(
|
||
client: httpx.AsyncClient,
|
||
base: str,
|
||
headers: dict[str, str],
|
||
agent_id: str,
|
||
thread_id: str,
|
||
new_message: str,
|
||
model_name: str,
|
||
) -> AsyncIterator[str | dict[str, Any]]:
|
||
"""跑一个智能体一轮（禁写文件 + 禁派活 / 脚本，其余只读工具放开）。
|
||
|
||
与 ``multi_agent._special_run`` 同形（yield 原始上游 SSE 行 + 末尾一条终态 dict），
|
||
但 ``excluded_tools`` = ``_EXCLUDED_TOOLS``（write_file / str_replace /
|
||
agent_orchestration / bash），且不做 post-broadcast（独立面板各智能体互不广播）。
|
||
"""
|
||
body = _run_payload(
|
||
agent_name=agent_id,
|
||
model_name=model_name,
|
||
thread_id=thread_id,
|
||
new_message=new_message,
|
||
excluded_tools=list(_EXCLUDED_TOOLS),
|
||
skill_stop_names=[],
|
||
thinking_enabled=True,
|
||
reasoning_effort="medium",
|
||
multitask_strategy="rollback",
|
||
)
|
||
captured: list[str] = []
|
||
async for line, done in _stream_upstream(client, base, headers, thread_id, body):
|
||
if line is not None:
|
||
yield line
|
||
elif done is not None:
|
||
captured = done
|
||
messages = _parse_last_messages(captured)
|
||
if messages is None:
|
||
# 流被抢占/截断,没有终态 values 帧 —— 静默收尾(调用方据无终态帧兜底)。
|
||
return
|
||
last_ai: dict[str, Any] = {}
|
||
for msg in reversed(messages):
|
||
if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
|
||
last_ai = msg
|
||
break
|
||
content = _flatten_content(last_ai.get("content", ""))
|
||
# 收集**所有** AI 消息的 tool_calls（全量 args，按顺序）—— 用于在收口时补发所有步骤
|
||
# （含 ls / skill_list 这类无参/中间步骤），保证步骤条不丢步、且 args 完整。
|
||
final_tool_calls: list[dict[str, Any]] = []
|
||
for msg in messages:
|
||
if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
|
||
for tc in msg.get("tool_calls") or []:
|
||
if isinstance(tc, dict):
|
||
final_tool_calls.append(tc)
|
||
clarification = _clarification_from_messages(messages)
|
||
if clarification is not None:
|
||
# 智能体在向用户要澄清 —— 交给前端渲染「协作卡片」(对齐圆桌第一步)。
|
||
yield {
|
||
"type": "clarification",
|
||
"content": content,
|
||
"clarification": clarification,
|
||
"tool_calls": final_tool_calls,
|
||
}
|
||
else:
|
||
yield {
|
||
"status": f"子智能体 {agent_id} 完成",
|
||
"content": content,
|
||
"agent_name": agent_id,
|
||
"tool_calls": final_tool_calls,
|
||
}
|
||
|
||
|
||
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(
|
||
"请你从你的专业职责出发，针对以上任务给出你的分析与交付。"
|
||
"本轮为**并行**研讨：请把你的产出**直接写在回复正文里**（本轮禁止写文件）。"
|
||
)
|
||
# 角色边界护栏：本面板里每个智能体都是**执行席位**，不是总控 / 指挥。即便任务里带有
|
||
# 「总控 / 业务链条 / 调度 / 派活」字样，也只完成属于自己的那份交付，严禁模仿总控派活。
|
||
parts.append(
|
||
"【角色边界】你是本轮的**执行席位**，不是总控 / 指挥。即使任务里出现「总控 / 业务链条 / 调度 / "
|
||
"派活」等字样，你也只需完成**属于你自己**的那份分析与交付；严禁模仿总控口吻输出「总控调度指令 / "
|
||
"调度对象 / 分配给『…』的任务 / agent_name」之类调度内容，严禁罗列或转派其他席位的任务，严禁调用"
|
||
"任何派活 / 调度类工具，也不要用脚本（bash / python）去调用内部接口或翻找派活脚本。"
|
||
)
|
||
# 澄清护栏：要向用户确认时，必须用 ask_clarification 工具发起（前端据此渲染可点击的协作
|
||
# 卡片），不能只把问题写在正文里 —— 否则用户看不到卡片、无法回应。能合理假设就先假设并标注。
|
||
parts.append(
|
||
"【需要向用户确认时】如果确实缺少完成分析所必需的关键信息，请**调用 `ask_clarification` 工具**"
|
||
"发起确认（这样用户会看到一张可点击的协作卡片来回答）；**不要只把问题写在正文里**，否则用户无法回应。"
|
||
"能基于已有信息合理假设的，就先假设并在正文中标注你的假设；只有当不同假设会显著改变结论时，才发起确认。"
|
||
)
|
||
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 = ""
|
||
emitted_tools: set[str] = set() # 已发过 step 的 tool_call id（去重）
|
||
tool_acc: dict[int, dict[str, Any]] = {} # tool_call_chunks 累积
|
||
buffer: list[str] = []
|
||
done_sent = False
|
||
|
||
async def _emit_tool(tc: Any, allow_empty: bool = False) -> None:
|
||
"""一个 tool_call（含完整 args）→ 一条 step 事件；按 id 去重。
|
||
|
||
``allow_empty=False``（流式期间）：args 为空先不发 —— 流式帧里的 tool_calls 常先给
|
||
空 ``{}`` 占位、真实 args 走 tool_call_chunks 后到，过早发会把空 args 钉死。等累积出
|
||
完整 args（synthetic）或终态全量 tool_calls（``allow_empty=True``）再发。
|
||
"""
|
||
if not isinstance(tc, dict):
|
||
return
|
||
name = tc.get("name")
|
||
if not isinstance(name, str) or not name:
|
||
return
|
||
args = tc.get("args") if isinstance(tc.get("args"), dict) else None
|
||
if args is None:
|
||
return
|
||
if not args and not allow_empty: # 空 args 先不发，等 chunks / 终态补全
|
||
return
|
||
key_id = str(tc.get("id") or name)
|
||
if key_id in emitted_tools:
|
||
return
|
||
emitted_tools.add(key_id)
|
||
await queue.put(
|
||
{"type": "step", "agent_id": agent_id, "key": f"tool_calling:{name}", "label": name, "args": args}
|
||
)
|
||
|
||
try:
|
||
thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||
gen = _run_one_agent( # 只禁文件写入，其余工具放开（保留步骤条）
|
||
client, base, headers, agent_id, thread_id, new_message, model_name
|
||
)
|
||
async for chunk in gen:
|
||
if isinstance(chunk, dict):
|
||
# 终态：先用最终 tool_calls（全量 args）补发漏掉的 step，再发收口/澄清。
|
||
for tc in chunk.get("tool_calls") or []:
|
||
await _emit_tool(tc, allow_empty=True)
|
||
if chunk.get("type") == "clarification":
|
||
clar = chunk.get("clarification") or {}
|
||
await queue.put(
|
||
{
|
||
"type": "agent_clarification",
|
||
"agent_id": agent_id,
|
||
"content": str(chunk.get("content") or ""),
|
||
**clar,
|
||
}
|
||
)
|
||
else:
|
||
await queue.put(
|
||
{"type": "agent_done", "agent_id": agent_id, "content": str(chunk.get("content") or "")}
|
||
)
|
||
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})
|
||
# 完整 tool_calls（部分模型直接给）+ 累积 chunks 后的合成调用，都尝试发 step。
|
||
for tc in obj.get("tool_calls") or []:
|
||
await _emit_tool(tc)
|
||
_ingest_tool_chunks(obj.get("tool_call_chunks"), tool_acc)
|
||
for tc in _synthetic_tool_calls(tool_acc):
|
||
await _emit_tool(tc)
|
||
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",
|
||
},
|
||
)
|