825 lines
38 KiB
Python
825 lines
38 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 stay on. ``ask_clarification``
|
||
is available only to strategy intent, the sole stage with a reply surface.
|
||
|
||
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
|
||
import time
|
||
from collections.abc import AsyncIterator
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Body, HTTPException, Query, Request, Response
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.roundtable_model_fallback import is_llm_error_message, reason_text
|
||
from app.gateway.routers import agents as agents_router
|
||
from app.gateway.routers._strategy_selection_seed import (
|
||
STRATEGY_AGENT_ID_PREFIX,
|
||
STRATEGY_INTENT_AGENT_ID,
|
||
)
|
||
from app.gateway.routers.multi_agent import (
|
||
_ROUNDTABLE_THREAD_METADATA,
|
||
_auth_headers,
|
||
_create_thread,
|
||
_loopback_base,
|
||
_make_loopback_client,
|
||
_parse_last_messages,
|
||
_run_payload,
|
||
_stream_upstream,
|
||
_validate_agent_id,
|
||
)
|
||
from deerflow.agents.roundtable_orchestrator.delivery import visible_seat_text
|
||
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
|
||
from deerflow.utils.stream_text import InlineThinkTagFilter
|
||
|
||
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
|
||
|
||
# Strategy roles are expected to return a substantive task/strategy/evaluation,
|
||
# not merely punctuation or a broken provider placeholder. Counting Unicode
|
||
# alphanumerics keeps the check language-neutral while rejecting outputs such as
|
||
# a long run of "!!!", which otherwise look non-empty to the UI.
|
||
MIN_MEANINGFUL_CONTENT_CHARS = 20
|
||
|
||
# Long-running read-only skills can be silent while they wait on an upstream
|
||
# service. Emit user-visible progress often enough to keep proxies and the UI
|
||
# alive, and fail one seat instead of leaving the whole strategy page pending
|
||
# forever when the upstream stream has truly stalled.
|
||
PROGRESS_HEARTBEAT_SECONDS = 5.0
|
||
AGENT_STREAM_IDLE_TIMEOUT_SECONDS = 180.0
|
||
|
||
_TOOL_PROGRESS_LABELS: dict[str, str] = {
|
||
"search_skills": "正在搜索匹配技能",
|
||
"skill_list": "正在查找可用技能",
|
||
"skill_view": "正在读取技能说明",
|
||
"llmwiki_search": "正在检索知识库",
|
||
"web_search": "正在检索外部信息",
|
||
"web_fetch": "正在读取外部资料",
|
||
"knowledge_search": "正在检索知识资料",
|
||
"read_file": "正在读取资料",
|
||
"read_multiple_files": "正在读取多份资料",
|
||
"ls": "正在查看资料目录",
|
||
"glob": "正在查找相关资料",
|
||
"grep": "正在筛选相关内容",
|
||
}
|
||
|
||
|
||
def _tool_progress_message(name: str, args: dict[str, Any]) -> str:
|
||
base = _TOOL_PROGRESS_LABELS.get(name, f"正在执行 {name}" if name else "正在执行策略分析步骤")
|
||
detail = next(
|
||
(
|
||
str(value).strip()
|
||
for value in (
|
||
args.get("query"),
|
||
args.get("skill_name"),
|
||
args.get("skill"),
|
||
args.get("name"),
|
||
args.get("path"),
|
||
)
|
||
if value is not None and str(value).strip()
|
||
),
|
||
"",
|
||
)
|
||
if not detail:
|
||
return base
|
||
compact = " ".join(detail.split())
|
||
if len(compact) > 64:
|
||
compact = f"{compact[:64]}…"
|
||
return f"{base}：{compact}"
|
||
|
||
|
||
def _has_substantive_content(content: str) -> bool:
|
||
visible = visible_seat_text(content)
|
||
return sum(character.isalnum() for character in visible) >= MIN_MEANINGFUL_CONTENT_CHARS
|
||
|
||
|
||
# 禁两类工具：①**写文件**（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", # 禁跑脚本（防绕道脚本调内部接口模仿派活）
|
||
]
|
||
|
||
|
||
def _strategy_stage_allows_clarification(agent_id: str) -> bool:
|
||
"""Return whether a strategy-selection stage has a user reply surface.
|
||
|
||
The workbench renders and resumes clarification cards only for the intent
|
||
stage. Angle generation, judging and final synthesis are one-shot stages;
|
||
letting those agents call ``ask_clarification`` produces an interrupt that
|
||
the UI cannot answer and therefore turns a recoverable run into a failure.
|
||
"""
|
||
|
||
return agent_id.strip().casefold() == STRATEGY_INTENT_AGENT_ID.casefold()
|
||
|
||
|
||
def _excluded_tools_for_agent(agent_id: str) -> list[str]:
|
||
excluded = list(_EXCLUDED_TOOLS)
|
||
normalized = agent_id.strip().casefold()
|
||
if normalized.startswith(STRATEGY_AGENT_ID_PREFIX.casefold()) and not _strategy_stage_allows_clarification(agent_id):
|
||
excluded.append("ask_clarification")
|
||
return excluded
|
||
|
||
|
||
# ── 并行面板兼容登录：复用单点开关、密码、限速与封禁校验 ────────────────────
|
||
|
||
|
||
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
|
||
|
||
|
||
def _force_situation_detail_compat_token(username: str) -> _TokenResponse:
|
||
"""Return a non-JWT value accepted only by open force-situation detail APIs.
|
||
|
||
Existing task-detail skills call this legacy endpoint before every detail
|
||
GET/PUT and require a non-empty value even though those endpoints are now
|
||
credential-free. Keeping the compatibility behavior generic avoids
|
||
coupling backend routing to a particular skill name or service account.
|
||
"""
|
||
|
||
normalized = username.casefold()
|
||
return _TokenResponse(
|
||
# Deliberately not a signed JWT: AuthMiddleware rejects it everywhere
|
||
# except the explicitly credential-free force-situation detail routes.
|
||
access_token=f"force-situation-detail-compat-{normalized}",
|
||
token_type="bearer",
|
||
user_id=f"external:{normalized}",
|
||
)
|
||
|
||
|
||
@router.post("/auth/token", response_model=_TokenResponse)
|
||
async def issue_token(body: _TokenRequest, request: Request, response: Response, password: str | None = Query(default=None)) -> _TokenResponse:
|
||
"""Issue a normal login token, or detail-route compatibility value on policy rejection."""
|
||
username = body.username.strip()
|
||
from app.gateway.routers.auth import login_by_username
|
||
|
||
try:
|
||
result = await login_by_username(
|
||
request,
|
||
response,
|
||
password=password,
|
||
)
|
||
except HTTPException as exc:
|
||
# The caller needs this endpoint only as a preflight before the open
|
||
# force-situation detail routes. Do not make individual skill names or
|
||
# current password/SSO policy a prerequisite for that workflow.
|
||
if exc.status_code in {401, 403}:
|
||
return _force_situation_detail_compat_token(username)
|
||
raise
|
||
return _TokenResponse(access_token=result.access_token, token_type="bearer", user_id=result.user_id)
|
||
|
||
|
||
# ── frame normalisation (mirrors multi-agent.ts) ─────────────────────────────
|
||
|
||
|
||
def _extract_ai_stream_parts(chunk: dict[str, Any]) -> tuple[str, str]:
|
||
"""Return ``(answer, reasoning)`` deltas from one AI message chunk.
|
||
|
||
OpenAI-compatible providers do not agree on where displayable reasoning
|
||
lives. Most put it in ``additional_kwargs.reasoning_content`` (or the
|
||
legacy ``reasoning`` key), while a few return typed content blocks. Keep
|
||
that channel separate from the answer so the strategy workbench can show
|
||
progress without mixing internal trace text into the generated strategy.
|
||
Literal ``<think>`` blocks in the normal text channel are handled by the
|
||
per-agent :class:`InlineThinkTagFilter` in ``_consume_agent`` because tags
|
||
may be split across arbitrary SSE chunks.
|
||
"""
|
||
type_ = str(chunk.get("type") or "").lower()
|
||
if type_ not in ("ai", "aimessagechunk", "aimessage"):
|
||
return "", ""
|
||
content = chunk.get("content")
|
||
answer_parts: list[str] = []
|
||
reasoning_parts: list[str] = []
|
||
if isinstance(content, str):
|
||
answer_parts.append(content)
|
||
elif isinstance(content, list):
|
||
for item in content:
|
||
if isinstance(item, str):
|
||
answer_parts.append(item)
|
||
elif isinstance(item, dict):
|
||
block_type = str(item.get("type") or "").lower()
|
||
value = item.get("thinking")
|
||
if not isinstance(value, str):
|
||
value = item.get("text")
|
||
if not isinstance(value, str):
|
||
value = item.get("content")
|
||
if not isinstance(value, str):
|
||
continue
|
||
if block_type in {
|
||
"thinking",
|
||
"thinking_delta",
|
||
"reasoning",
|
||
"reasoning_delta",
|
||
}:
|
||
reasoning_parts.append(value)
|
||
else:
|
||
answer_parts.append(value)
|
||
|
||
additional_kwargs = chunk.get("additional_kwargs")
|
||
direct_reasoning = chunk.get("reasoning_content") or chunk.get("reasoning")
|
||
nested_reasoning: object = None
|
||
if isinstance(additional_kwargs, dict):
|
||
nested_reasoning = additional_kwargs.get("reasoning_content") or additional_kwargs.get("reasoning")
|
||
raw_reasoning = nested_reasoning or direct_reasoning
|
||
if isinstance(raw_reasoning, str):
|
||
reasoning_parts.append(raw_reasoning)
|
||
|
||
return "".join(answer_parts), "".join(reasoning_parts)
|
||
|
||
|
||
def _extract_ai_text(chunk: dict[str, Any]) -> str:
|
||
"""Compatibility helper returning only the ordinary AI text delta."""
|
||
return _extract_ai_stream_parts(chunk)[0]
|
||
|
||
|
||
def _split_final_inline_reasoning(content: str) -> tuple[str, str]:
|
||
"""Split complete inline thinking, including template-injected close tags.
|
||
|
||
Some reasoning chat templates insert ``<think>`` before generation, so the
|
||
provider returns only ``reasoning</think>answer``. The streaming filter
|
||
cannot classify that prefix before the closing marker arrives, but the
|
||
terminal values frame can. Returning the recovered trace here also lets
|
||
the client correct any transient streaming classification.
|
||
"""
|
||
if not content:
|
||
return "", ""
|
||
|
||
working = content
|
||
prefix_reasoning = ""
|
||
lowered = working.lower()
|
||
first_open = lowered.find("<think>")
|
||
first_close = lowered.find("</think>")
|
||
if first_close >= 0 and (first_open < 0 or first_close < first_open):
|
||
prefix_reasoning = working[:first_close]
|
||
working = working[first_close + len("</think>") :]
|
||
|
||
text_filter = InlineThinkTagFilter()
|
||
visible, reasoning = text_filter.push_parts(working)
|
||
visible_tail, reasoning_tail = text_filter.finish_parts()
|
||
final_reasoning = "".join(part for part in (prefix_reasoning, reasoning, reasoning_tail) if part)
|
||
# ``visible_seat_text`` remains the canonical complete-output cleanup for
|
||
# malformed/multiple tags; the filter-derived visible value is only a
|
||
# fallback for an otherwise empty canonical result.
|
||
final_visible = visible_seat_text(content) or f"{visible}{visible_tail}".strip()
|
||
return final_visible, final_reasoning.strip()
|
||
|
||
|
||
def _should_probe_implicit_reasoning(
|
||
request: Request,
|
||
agent_id: str,
|
||
requested_model: str,
|
||
) -> bool:
|
||
"""Enable close-only probing only for a thinking-capable effective model.
|
||
|
||
Ordinary models must stream untagged answer chunks immediately. The
|
||
stricter prelude buffer is reserved for providers that advertise thinking
|
||
support and may use templates where ``<think>`` is injected outside the
|
||
returned content while ``</think>`` remains in-band.
|
||
"""
|
||
|
||
config = getattr(request.app.state, "config", None)
|
||
models = list(getattr(config, "models", None) or [])
|
||
effective_name = requested_model.strip()
|
||
if not effective_name:
|
||
try:
|
||
effective_name = str(agents_router._load_config_extras(agent_id).get("model") or "").strip()
|
||
except Exception:
|
||
effective_name = ""
|
||
if not effective_name and models:
|
||
effective_name = str(getattr(models[0], "name", "") or "")
|
||
selected = next(
|
||
(entry for entry in models if str(getattr(entry, "name", "") or "") == effective_name),
|
||
None,
|
||
)
|
||
return bool(selected and getattr(selected, "supports_thinking", False))
|
||
|
||
|
||
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=_excluded_tools_for_agent(agent_id),
|
||
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
|
||
failure_reason = is_llm_error_message(last_ai)
|
||
if failure_reason:
|
||
raise RuntimeError(f"模型调用失败：{reason_text(failure_reason)}")
|
||
answer_channel, structured_reasoning = _extract_ai_stream_parts(last_ai)
|
||
content, inline_reasoning = _split_final_inline_reasoning(answer_channel)
|
||
reasoning = "\n\n".join(part.strip() for part in (structured_reasoning, inline_reasoning) if part.strip())
|
||
# 收集**所有** 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,
|
||
"reasoning": reasoning,
|
||
"clarification": clarification,
|
||
"tool_calls": final_tool_calls,
|
||
}
|
||
else:
|
||
if not _has_substantive_content(content):
|
||
raise RuntimeError("智能体未生成有效正文，请重试或更换全局模型")
|
||
yield {
|
||
"status": f"子智能体 {agent_id} 完成",
|
||
"content": content,
|
||
"reasoning": reasoning,
|
||
"agent_name": agent_id,
|
||
"tool_calls": final_tool_calls,
|
||
}
|
||
|
||
|
||
def _build_task_message(task_name: str, brief: str, *, agent_id: 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 会产生前端无法回答的中断并被误报为生成失败。
|
||
if agent_id and agent_id.casefold().startswith(STRATEGY_AGENT_ID_PREFIX.casefold()) and not _strategy_stage_allows_clarification(agent_id):
|
||
parts.append(
|
||
"【本阶段禁止反问】当前是无需用户中途介入的一次性交付阶段。不得调用 `ask_clarification`，"
|
||
"不得在正文里向用户提问、要求补充材料或要求点击重试。信息不足时请采用保守的合理假设，"
|
||
"明确降低相应结论的置信度，并把缺口列入“数据限制与待核验”，然后直接完成本阶段要求的全部交付；"
|
||
"不要重复输出同一正文或多个版本。"
|
||
)
|
||
else:
|
||
# 澄清护栏：要向用户确认时，必须用 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)}）",
|
||
)
|
||
role_store = getattr(request.app.state, "strategy_agent_role_store", None)
|
||
agent_store = getattr(request.app.state, "agent_store", None)
|
||
strategy_ids = [agent_id for agent_id in agent_ids if agent_id.casefold().startswith(STRATEGY_AGENT_ID_PREFIX.casefold())]
|
||
if strategy_ids and role_store is not None and agent_store is not None:
|
||
# Every worker owns an independent memory store. Synchronize from
|
||
# durable folders/sidecars immediately before execution so a request
|
||
# routed to another worker cannot see random 404s or revived angles.
|
||
from app.gateway.routers.strategy_agents import _ensure_registry
|
||
|
||
await _ensure_registry(agent_store, role_store)
|
||
for agent_id in agent_ids:
|
||
if not agent_id.casefold().startswith(STRATEGY_AGENT_ID_PREFIX.casefold()):
|
||
continue
|
||
if role_store is None or agent_store is None:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail="策略智能体分类或配置服务不可用",
|
||
)
|
||
role = await role_store.get_role(agent_id)
|
||
if role is None:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"未注册的策略智能体：{agent_id}",
|
||
)
|
||
if not bool(role.get("enabled")):
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=f"策略智能体已停用：{agent_id}",
|
||
)
|
||
record = await agent_store.get_any(agent_id)
|
||
if record is None or str(record.get("id") or "") != agent_id or record.get("user_id") is not None:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=f"策略智能体必须由系统全局持有：{agent_id}",
|
||
)
|
||
if not task_name and not brief:
|
||
raise HTTPException(status_code=400, detail="任务名称与简介至少填写其一")
|
||
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
|
||
async def _consume_agent(
|
||
client: Any,
|
||
agent_id: str,
|
||
queue: asyncio.Queue[dict[str, Any]],
|
||
) -> None:
|
||
new_message = _build_task_message(task_name, brief, agent_id=agent_id)
|
||
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] = []
|
||
raw_answer_buffer: list[str] = []
|
||
probe_implicit_close = await asyncio.to_thread(
|
||
_should_probe_implicit_reasoning,
|
||
request,
|
||
agent_id,
|
||
model_name,
|
||
)
|
||
inline_thinking = InlineThinkTagFilter(allow_implicit_close=probe_implicit_close)
|
||
reasoning_emitted = False
|
||
done_sent = False
|
||
started_at = time.monotonic()
|
||
progress_message = "智能体正在分析任务"
|
||
|
||
async def _emit_progress_heartbeats() -> None:
|
||
while not done_sent:
|
||
await asyncio.sleep(PROGRESS_HEARTBEAT_SECONDS)
|
||
# A terminal event may have been queued while this task was
|
||
# sleeping. Never let a late heartbeat move a completed card
|
||
# back into the running state on the frontend.
|
||
if done_sent:
|
||
return
|
||
await queue.put(
|
||
{
|
||
"type": "agent_progress",
|
||
"agent_id": agent_id,
|
||
"message": progress_message,
|
||
"elapsed_seconds": max(0, round(time.monotonic() - started_at)),
|
||
}
|
||
)
|
||
|
||
async def _emit_text_parts(text: str, provider_reasoning: str = "") -> None:
|
||
nonlocal progress_message, reasoning_emitted
|
||
if text:
|
||
raw_answer_buffer.append(text)
|
||
visible_prelude = inline_thinking.confirm_visible_channel() if provider_reasoning else ""
|
||
visible, inline_reasoning = inline_thinking.push_parts(text)
|
||
reasoning = f"{provider_reasoning}{inline_reasoning}"
|
||
if reasoning:
|
||
progress_message = "正在分析并组织策略"
|
||
reasoning_emitted = True
|
||
await queue.put(
|
||
{
|
||
"type": "reasoning_delta",
|
||
"agent_id": agent_id,
|
||
"text": reasoning,
|
||
}
|
||
)
|
||
confirmed_visible = f"{visible_prelude}{visible}"
|
||
if confirmed_visible:
|
||
progress_message = "正在生成策略正文"
|
||
buffer.append(confirmed_visible)
|
||
await queue.put({"type": "delta", "agent_id": agent_id, "text": confirmed_visible})
|
||
|
||
async def _flush_text_filter() -> None:
|
||
nonlocal reasoning_emitted
|
||
visible, reasoning = inline_thinking.finish_parts()
|
||
if reasoning:
|
||
reasoning_emitted = True
|
||
await queue.put(
|
||
{
|
||
"type": "reasoning_delta",
|
||
"agent_id": agent_id,
|
||
"text": reasoning,
|
||
}
|
||
)
|
||
if visible:
|
||
buffer.append(visible)
|
||
await queue.put({"type": "delta", "agent_id": agent_id, "text": visible})
|
||
|
||
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``）再发。
|
||
"""
|
||
nonlocal progress_message
|
||
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)
|
||
progress_message = _tool_progress_message(name, args)
|
||
await queue.put({"type": "step", "agent_id": agent_id, "key": f"tool_calling:{name}", "label": name, "args": args})
|
||
|
||
heartbeat_task = asyncio.create_task(_emit_progress_heartbeats())
|
||
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
|
||
).__aiter__()
|
||
while True:
|
||
try:
|
||
chunk = await asyncio.wait_for(
|
||
anext(gen),
|
||
timeout=AGENT_STREAM_IDLE_TIMEOUT_SECONDS,
|
||
)
|
||
except StopAsyncIteration:
|
||
break
|
||
except TimeoutError as exc:
|
||
await gen.aclose()
|
||
raise RuntimeError(f"智能体连续 {int(AGENT_STREAM_IDLE_TIMEOUT_SECONDS)} 秒没有返回进度，已停止本次执行，请重试") from exc
|
||
if isinstance(chunk, dict):
|
||
# 终态：先用最终 tool_calls（全量 args）补发漏掉的 step，再发收口/澄清。
|
||
await _flush_text_filter()
|
||
for tc in chunk.get("tool_calls") or []:
|
||
await _emit_tool(tc, allow_empty=True)
|
||
final_content = visible_seat_text(str(chunk.get("content") or ""))
|
||
final_reasoning = str(chunk.get("reasoning") or "")
|
||
if final_reasoning and not reasoning_emitted:
|
||
reasoning_emitted = True
|
||
await queue.put(
|
||
{
|
||
"type": "reasoning_delta",
|
||
"agent_id": agent_id,
|
||
"text": final_reasoning,
|
||
}
|
||
)
|
||
if chunk.get("type") == "clarification":
|
||
done_sent = True
|
||
clar = chunk.get("clarification") or {}
|
||
await queue.put(
|
||
{
|
||
"type": "agent_clarification",
|
||
"agent_id": agent_id,
|
||
"content": final_content,
|
||
**clar,
|
||
}
|
||
)
|
||
else:
|
||
done_sent = True
|
||
await queue.put({"type": "agent_done", "agent_id": agent_id, "content": final_content})
|
||
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, reasoning = _extract_ai_stream_parts(obj)
|
||
if text or reasoning:
|
||
await _emit_text_parts(text, reasoning)
|
||
# 完整 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 _flush_text_filter()
|
||
buffered_content, recovered_reasoning = _split_final_inline_reasoning("".join(raw_answer_buffer))
|
||
if not buffered_content:
|
||
buffered_content = "".join(buffer)
|
||
if recovered_reasoning and not reasoning_emitted:
|
||
reasoning_emitted = True
|
||
await queue.put(
|
||
{
|
||
"type": "reasoning_delta",
|
||
"agent_id": agent_id,
|
||
"text": recovered_reasoning,
|
||
}
|
||
)
|
||
failure_reason = is_llm_error_message({"type": "ai", "content": buffered_content})
|
||
if failure_reason:
|
||
raise RuntimeError(f"模型调用失败：{reason_text(failure_reason)}")
|
||
if not _has_substantive_content(buffered_content):
|
||
raise RuntimeError("智能体未生成有效正文，请重试或更换全局模型")
|
||
done_sent = True
|
||
await queue.put(
|
||
{
|
||
"type": "agent_done",
|
||
"agent_id": agent_id,
|
||
"content": buffered_content,
|
||
}
|
||
)
|
||
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:
|
||
heartbeat_task.cancel()
|
||
await asyncio.gather(heartbeat_task, return_exceptions=True)
|
||
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",
|
||
},
|
||
)
|