董宏凯andClaude Opus 4.8
8d1669adad
后台挂起作业每个 agent 轮次（leader/seat/report/summary/dashboard）开跑前 先取该用户的后台并发额度（默认 2，ROUNDTABLE_BACKGROUND_SLOTS_PER_USER 可调）， 跑完释放；同一用户多个后台作业共享额度，避免把单连接 SQLite checkpointer 和 离线模型端点打满拖垮前台实时会商。前台实时路径不经过此额度，不受限。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
604 lines
28 KiB
Python
604 lines
28 KiB
Python
"""圆桌真网关（Phase 3b）—— 进程内驱动真实 leader / seat / report agent run。
|
||
|
||
实现 ``RoundtableGateway`` 协议（定义在 harness 的 roundtable_orchestrator.types），
|
||
但**进程内**跑真模型：复用 ``services.start_run``（scheduled task 同款）驱动一轮 agent，
|
||
跑完从 checkpointer 读最终 messages，用 ``multi_agent`` 的纯解析函数解析派活 / 澄清。
|
||
不走 HTTP、不需要 Request（用 ``SimpleNamespace(app=app, state=...)`` 当虚拟 request，
|
||
``set_current_user`` 由执行器设好）。
|
||
|
||
本网关位于 app 层（依赖 app.state + multi_agent helpers），实现 harness 定义的协议——
|
||
不违反 harness→app 边界（harness 永不 import app；app import harness 合法）。
|
||
|
||
⚠️ 需在能起完整后端栈 + 真模型的环境验证（见 scripts/probe_roundtable_job.py）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
from langgraph.checkpoint.base import empty_checkpoint
|
||
|
||
from app.gateway.roundtable_run_policy import roundtable_run_policy
|
||
from app.gateway.routers._roundtable_seed import (
|
||
COORDINATOR_AGENT_ID,
|
||
REPORT_AGENT_ID,
|
||
SUMMARY_AGENT_ID,
|
||
ensure_roundtable_functional_agents,
|
||
)
|
||
from deerflow.agents.roundtable_orchestrator.types import (
|
||
LeaderTurn,
|
||
ReportResult,
|
||
SeatRef,
|
||
SeatTurn,
|
||
SummaryResult,
|
||
)
|
||
from deerflow.config.paths import get_paths
|
||
from deerflow.runtime.serialization import serialize_channel_values
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
|
||
_REPORT_FILE = "方案总览.html"
|
||
_SUMMARY_FILE = "方案总结报告.md"
|
||
|
||
# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy
|
||
# （与 multi_agent 的 _leader_run / _special_run 同源），不再在本文件硬编码——历史上
|
||
# 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。
|
||
|
||
|
||
# ── 后台作业「按用户限流」（2026-06-16）─────────────────────────────────────────
|
||
# 痛点：进 2+ 个后台挂起后，整个网关变慢——建立智能体连接、等待模型响应都慢。
|
||
# 根因：① checkpointer 用单连接 aiosqlite（AsyncSqliteSaver，内部一把 asyncio.Lock 串
|
||
# 行全进程的 checkpoint 读写）；② 后台作业无并发上限——每个作业进程内驱动真实 leader/
|
||
# seat/report run，DAG/recommend 的 stage 还用 asyncio.gather 把席位真并行铺开。多个后台
|
||
# 作业同时把几十个 agent run 压到同一把 SQLite 锁 + 同一个离线模型端点上，前台实时会商
|
||
# 的 checkpoint 读写 / 模型调用只能排队 → 「干啥都慢」。
|
||
#
|
||
# 解法：给**每个用户**一个并发额度（默认 2），后台每个 agent 轮次（leader/seat/report/
|
||
# summary/dashboard）真正开跑前先取一个额度，跑完释放。同一用户的所有后台作业共享这 2 个
|
||
# 额度——随便挂多少个作业都能立刻建好，但同一瞬间真正在打模型/写 checkpoint 的后台轮次
|
||
# 至多 2 个，不会把共享资源打满。**前台实时路径（multi_agent._leader_run/_special_run、
|
||
# 普通聊天）不经过本文件的 `_run_agent_turn`，永不取这个额度，故前台不受限。**
|
||
_BACKGROUND_SLOTS_PER_USER = max(1, int(os.getenv("ROUNDTABLE_BACKGROUND_SLOTS_PER_USER", "2") or "2"))
|
||
# 按 user_id 缓存 Semaphore。网关全程跑在同一个事件循环上，普通 dict 取/建即可（无跨线程）。
|
||
_user_background_sems: dict[str, asyncio.Semaphore] = {}
|
||
|
||
|
||
def _user_background_semaphore(user_id: str) -> asyncio.Semaphore:
|
||
"""取该用户的后台并发额度信号量（懒建，每用户一把）。"""
|
||
sem = _user_background_sems.get(user_id)
|
||
if sem is None:
|
||
sem = asyncio.Semaphore(_BACKGROUND_SLOTS_PER_USER)
|
||
_user_background_sems[user_id] = sem
|
||
return sem
|
||
|
||
|
||
# ── 进程内底层操作 ───────────────────────────────────────────────────────────
|
||
|
||
|
||
async def _create_thread(app) -> str:
|
||
"""同进程建一个带圆桌 metadata 的 thread + 空 checkpoint（镜像 _create_thread_direct）。"""
|
||
checkpointer = app.state.checkpointer
|
||
thread_store = app.state.thread_store
|
||
thread_id = str(uuid.uuid4())
|
||
await thread_store.create(thread_id, assistant_id=None, metadata=dict(_ROUNDTABLE_THREAD_METADATA))
|
||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||
await checkpointer.aput(
|
||
config,
|
||
empty_checkpoint(),
|
||
{"step": -1, "source": "input", "writes": None, "parents": {}, "created_at": datetime.now(UTC).isoformat()},
|
||
{},
|
||
)
|
||
return thread_id
|
||
|
||
|
||
async def _read_thread_messages(app, thread_id: str) -> list[dict[str, Any]]:
|
||
"""读 thread 最新 checkpoint 的 messages（已 serialize 成 dict 列表）。"""
|
||
checkpointer = app.state.checkpointer
|
||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||
ct = await checkpointer.aget_tuple(config)
|
||
if ct is None:
|
||
return []
|
||
checkpoint = getattr(ct, "checkpoint", {}) or {}
|
||
values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
|
||
return values.get("messages") or []
|
||
|
||
|
||
def _latest_ai_message(messages: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||
for m in reversed(messages):
|
||
if not isinstance(m, dict):
|
||
continue
|
||
if (m.get("type") or m.get("role")) in {"ai", "assistant"}:
|
||
return m
|
||
return None
|
||
|
||
|
||
def _message_text(message: dict[str, Any] | None) -> str:
|
||
if not message:
|
||
return ""
|
||
content = message.get("content")
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, list):
|
||
parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
|
||
return "\n".join(parts)
|
||
return ""
|
||
|
||
|
||
def _latest_ai_text(messages: list[dict[str, Any]]) -> str:
|
||
"""取**最后一条有正文**的 AI 消息文本。
|
||
|
||
席位常以「调一堆工具」的 AIMessage 收尾（正文为空、只有 tool_calls，甚至 dangling），
|
||
直接取最后一条 AI 消息会拿到空串 → 报告就拿不到席位产物。这里向前找到真正有文字的那条。
|
||
"""
|
||
for m in reversed(messages):
|
||
if not isinstance(m, dict):
|
||
continue
|
||
if (m.get("type") or m.get("role")) not in {"ai", "assistant"}:
|
||
continue
|
||
text = _message_text(m)
|
||
if text.strip():
|
||
return text
|
||
return ""
|
||
|
||
|
||
def _normalize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]:
|
||
"""把 checkpoint 里的 tool_calls 归一化成 {name, args, id}（丢掉无名项）。"""
|
||
out: list[dict[str, Any]] = []
|
||
for tc in tool_calls or []:
|
||
if not isinstance(tc, dict):
|
||
continue
|
||
name = tc.get("name") or ""
|
||
if not name:
|
||
continue
|
||
out.append({"name": name, "args": tc.get("args") or {}, "id": tc.get("id") or ""})
|
||
return out
|
||
|
||
|
||
def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""收集「最后一条 human 消息之后」各 AI 消息的 tool_calls（即**本轮**调用），保序。
|
||
|
||
席位 thread 跨轮累积历史，直接扫全量会把往轮的工具调用也算进来；以最后一条 human
|
||
（本轮 run 的输入）为界，只取其后的 AI 工具调用，对应前端「这一条气泡的步骤卡」。
|
||
"""
|
||
last_human = -1
|
||
for i, m in enumerate(messages):
|
||
if isinstance(m, dict) and (m.get("type") or m.get("role")) in {"human", "user"}:
|
||
last_human = i
|
||
out: list[dict[str, Any]] = []
|
||
for m in messages[last_human + 1 :]:
|
||
if not isinstance(m, dict):
|
||
continue
|
||
if (m.get("type") or m.get("role")) not in {"ai", "assistant"}:
|
||
continue
|
||
out.extend(_normalize_tool_calls(m.get("tool_calls")))
|
||
return out
|
||
|
||
|
||
async def _run_agent_turn(
|
||
app,
|
||
*,
|
||
user_id: str,
|
||
thread_id: str,
|
||
agent_name: str,
|
||
message: str,
|
||
model: str | None,
|
||
excluded_tools: list[str],
|
||
skill_stop_names: list[str],
|
||
thinking_enabled: bool,
|
||
reasoning_effort: str,
|
||
subagent_enabled: bool = False,
|
||
) -> list[dict[str, Any]]:
|
||
"""进程内驱动 agent_name 在 thread_id 上跑一轮（发 message），阻塞到完成，返回最终 messages。
|
||
|
||
镜像 multi_agent ``_run_payload`` 的 body 形状：agent 通过 ``context.agent_name`` 选，
|
||
assistant_id 固定 lead_agent；关闭 memory recall / subagent / plan mode（临时研讨任务）。
|
||
|
||
并发受**该用户的后台额度**约束（``_user_background_semaphore``）：真正开跑（start_run +
|
||
await run task）前先取额度，跑完释放。同一用户的所有后台作业共享这几个额度，避免 2+
|
||
后台挂起把单连接 SQLite checkpointer / 离线模型打满、拖垮前台实时会商。
|
||
"""
|
||
from app.gateway.routers.thread_runs import RunCreateRequest
|
||
from app.gateway.services import start_run
|
||
|
||
body = RunCreateRequest(
|
||
assistant_id="lead_agent",
|
||
input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
|
||
config={"recursion_limit": 1000},
|
||
context={
|
||
"agent_name": agent_name,
|
||
"model_name": model or "",
|
||
"mode": "pro",
|
||
"reasoning_effort": reasoning_effort,
|
||
"thinking_enabled": thinking_enabled,
|
||
"is_plan_mode": False,
|
||
# ultra 席位才置真 → 解锁 task 子代理工具（席位 policy 未禁 task）。leader 恒 False。
|
||
"subagent_enabled": subagent_enabled,
|
||
"thread_id": thread_id,
|
||
"memory_recall_disabled": True,
|
||
# 与 multi_agent._run_payload 对齐:关 builtin 记忆注入,避免单例总控/席位
|
||
# 被上轮会话 memory.json 带偏(臆造席位 / 直接干活)。详见该处注释。
|
||
"memory_injection_enabled": False,
|
||
},
|
||
excluded_tools=excluded_tools,
|
||
skill_stop_names=skill_stop_names,
|
||
on_disconnect="continue",
|
||
multitask_strategy="reject",
|
||
)
|
||
# 虚拟 request：带上空 ``state``（auth 中间件在真请求上会写 state.user / state.auth；
|
||
# 进程内没有这层，下游凡是读 request.state.X 的代码必须能拿到 None 而不是 AttributeError）。
|
||
request = SimpleNamespace(app=app, state=SimpleNamespace(user=None, auth=None))
|
||
# 取该用户的后台额度后再开跑：start_run + await task 是最重的部分（建 agent、调模型、
|
||
# 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。
|
||
async with _user_background_semaphore(user_id):
|
||
record = await start_run(body, thread_id, request) # type: ignore[arg-type]
|
||
if getattr(record, "task", None) is not None:
|
||
await record.task # 阻塞到这轮 agent 跑完，checkpoint 已落盘
|
||
return await _read_thread_messages(app, thread_id)
|
||
|
||
|
||
def _read_report_html(app, thread_id: str, user_id: str) -> str | None:
|
||
"""从 report thread 的 outputs 目录读出报告 HTML。
|
||
|
||
优先 ``方案总览.html``；找不到就扫目录里**任意 .html**（取体积最大的那个），
|
||
容忍报告 agent 用了别的文件名。
|
||
"""
|
||
try:
|
||
outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
|
||
preferred = outputs_dir / _REPORT_FILE
|
||
if preferred.exists():
|
||
return preferred.read_text(encoding="utf-8")
|
||
if outputs_dir.exists():
|
||
htmls = sorted(
|
||
outputs_dir.glob("*.html"),
|
||
key=lambda p: p.stat().st_size,
|
||
reverse=True,
|
||
)
|
||
if htmls:
|
||
logger.info("report html: 方案总览.html not found, using %s", htmls[0].name)
|
||
return htmls[0].read_text(encoding="utf-8")
|
||
logger.warning("report html: no .html found in %s", outputs_dir)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("read report html failed for thread %s", thread_id, exc_info=True)
|
||
return None
|
||
|
||
|
||
def _read_summary_md(app, thread_id: str, user_id: str) -> str | None:
|
||
"""从 summary thread 的 outputs 目录读出总结报告 Markdown。
|
||
|
||
优先 ``方案总结报告.md``；找不到就扫目录里**任意 .md**（取体积最大的那个），
|
||
容忍总结 agent 用了别的文件名。镜像 ``_read_report_html``。
|
||
"""
|
||
try:
|
||
outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
|
||
preferred = outputs_dir / _SUMMARY_FILE
|
||
if preferred.exists():
|
||
return preferred.read_text(encoding="utf-8")
|
||
if outputs_dir.exists():
|
||
mds = sorted(
|
||
outputs_dir.glob("*.md"),
|
||
key=lambda p: p.stat().st_size,
|
||
reverse=True,
|
||
)
|
||
if mds:
|
||
logger.info("summary md: 方案总结报告.md not found, using %s", mds[0].name)
|
||
return mds[0].read_text(encoding="utf-8")
|
||
logger.warning("summary md: no .md found in %s", outputs_dir)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("read summary md failed for thread %s", thread_id, exc_info=True)
|
||
return None
|
||
|
||
|
||
# Step3 产物自愈重试次数（首轮之外的纠偏轮数）。
|
||
# 实测痛点（2026-06-12，h:\xiugai\后台日志.txt）：deepseek 报告 agent 说「let me write
|
||
# it all at once」后发出超长 write_file，单次输出被 max_tokens 截断 → 工具调用从未执行、
|
||
# run 以 success 收尾但 outputs 下没有 .html → step3.html 回退成开场白文字。
|
||
# 自愈：检测产物缺失/截断后，同线程追加纠偏消息（要求 write_file 分段 + append 续写）重试。
|
||
#
|
||
# 取值依据（2026-06-15 修复「后台 HTML 看板只生成一半、概率渲染失败」）：模型单轮输出受
|
||
# ``config.yaml`` 的 ``max_tokens``（deepseek-chat = 4096）硬截。一份完整看板 HTML 几十 KB
|
||
# ≈ 上万 token，单轮写不完；旧值 2（最多 3 轮 ≈ 12k token）对长报告不够 → 落库半截 HTML →
|
||
# 渲染失败。调到 5（最多 6 轮 append ≈ 24k token，足够绝大多数看板）显著降低截断概率。
|
||
# 循环命中完整产物（``</html>`` / 非空 md / 合法 json）即 break，短报告不会白跑额外轮次。
|
||
_ARTIFACT_RETRY_LIMIT = 5
|
||
|
||
_REPORT_RETRY_MISSING = (
|
||
f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 还没有生成——你上一轮的 write_file 没有执行成功"
|
||
"（最常见原因：试图一次写入整份长 HTML，输出超长被截断，工具调用没有真正落盘）。\n"
|
||
"请立即重新写入，并**严格分段**：\n"
|
||
f"1. 先用 write_file 写入 /mnt/user-data/outputs/{_REPORT_FILE} 的开头部分（<!doctype html> 起，约 200-300 行）；\n"
|
||
"2. 再用 write_file 的 append=true 参数分多次续写后续片段，每段同样控制在 300 行以内；\n"
|
||
"3. 直到文档以 </html> 收尾。\n"
|
||
"不要在对话里粘贴 HTML 源码，写完用一两句话总结即可。"
|
||
)
|
||
|
||
_REPORT_RETRY_TRUNCATED = (
|
||
f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 已写入但**不完整**（没有 </html> 收尾，疑似上一轮输出被截断）。\n"
|
||
"请用 write_file 的 append=true 参数从中断处**继续补全剩余内容**（不要重写已有部分），"
|
||
"每段控制在 300 行以内，分多次续写直到文档以 </html> 收尾。不要在对话里粘贴源码。"
|
||
)
|
||
|
||
_SUMMARY_RETRY_MISSING = (
|
||
f"检测到 /mnt/user-data/outputs/{_SUMMARY_FILE} 还没有生成——你上一轮的 write_file 没有执行成功"
|
||
"（可能是单次输出过长被截断）。请立即用 write_file 重新写入完整报告；若内容很长，"
|
||
"先写前半部分、再用 write_file 的 append=true 参数分段续写。写完后把报告正文也输出到对话回复里。"
|
||
)
|
||
|
||
_DASHBOARD_FENCE = "```report-json"
|
||
_DASHBOARD_RETRY = (
|
||
"你刚才没有输出有效的 ```report-json 代码块（或其中的 JSON 无法解析）。"
|
||
"请重新在回复**最末尾**输出一个完整且**严格合法**的 ```report-json 代码块"
|
||
"（标准 JSON：无注释、无尾逗号、字符串用双引号），代码块之外只允许一两句说明；"
|
||
"不要调用任何工具、不要写文件、不要输出 HTML。"
|
||
)
|
||
|
||
|
||
def _extract_report_json(text: str) -> str | None:
|
||
"""从模型回复里提取最后一个 ```report-json 代码块的 JSON 字符串（须可解析）。"""
|
||
start = text.rfind(_DASHBOARD_FENCE)
|
||
if start < 0:
|
||
return None
|
||
after = text[start + len(_DASHBOARD_FENCE) :]
|
||
close = after.find("```")
|
||
body = (after[:close] if close >= 0 else after).strip()
|
||
if not body:
|
||
return None
|
||
try:
|
||
json.loads(body)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
return body
|
||
|
||
|
||
def _html_complete(html: str | None) -> bool:
|
||
"""报告 HTML 是否完整可用（非空 + 有 </html> 收尾，截断检测）。"""
|
||
return bool(html and html.strip()) and "</html>" in html.lower()
|
||
|
||
|
||
def _build_seat_intro(agents: list[SeatRef]) -> str:
|
||
"""镜像 init 注入 coordinator 的「可调度席位清单」。"""
|
||
lines = [f"- agent_name: {a.agent_id}（{a.name}）" for a in agents]
|
||
return (
|
||
"【圆桌系统初始化】本次研讨可调度的席位清单如下，请严格按 agent_name 通过 "
|
||
"agent_orchestration 技能派活，不要派给不在清单内的 agent：\n" + "\n".join(lines)
|
||
)
|
||
|
||
|
||
# ── 网关实现 ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class InProcessRoundtableGateway:
|
||
"""进程内真网关。executor 先调 ``init_threads`` 建好线程，再跑编排引擎。"""
|
||
|
||
def __init__(
|
||
self,
|
||
app,
|
||
*,
|
||
user_id: str,
|
||
agents: list[SeatRef],
|
||
model: str | None = None,
|
||
seat_thinking_enabled: bool | None = None,
|
||
seat_reasoning_effort: str | None = None,
|
||
seat_subagent_enabled: bool = False,
|
||
) -> None:
|
||
self._app = app
|
||
self._user_id = user_id
|
||
self._agents = list(agents)
|
||
self._model = model
|
||
# 席位执行模式（沿用 Step2 选择）：覆盖 seat policy 的 thinking/reasoning（None → policy 默认），
|
||
# subagent 仅 ultra 档为真。只作用 special 席位；leader 派活不受影响。
|
||
self._seat_thinking_enabled = seat_thinking_enabled
|
||
self._seat_reasoning_effort = seat_reasoning_effort
|
||
self._seat_subagent_enabled = seat_subagent_enabled
|
||
self._seat_intro = ""
|
||
self._first_leader = True
|
||
|
||
async def init_threads(self) -> tuple[dict[str, str], str]:
|
||
"""建 coordinator + 各席位 thread，返回 (thread_ids, coordinator_name)。
|
||
|
||
席位清单通过「首轮 leader 消息前缀」注入（coordinator thread 跨轮累积历史，
|
||
首轮带上即长期可见），无需单独写 checkpoint。
|
||
"""
|
||
ensure_roundtable_functional_agents()
|
||
thread_ids: dict[str, str] = {}
|
||
for a in self._agents:
|
||
thread_ids[a.agent_id] = await _create_thread(self._app)
|
||
thread_ids[COORDINATOR_AGENT_ID] = await _create_thread(self._app)
|
||
self._seat_intro = _build_seat_intro(self._agents)
|
||
return thread_ids, COORDINATOR_AGENT_ID
|
||
|
||
async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn:
|
||
msg = message
|
||
if self._first_leader and self._seat_intro:
|
||
msg = self._seat_intro + "\n\n" + message
|
||
self._first_leader = False
|
||
policy = roundtable_run_policy("leader")
|
||
messages = await _run_agent_turn(
|
||
self._app,
|
||
user_id=self._user_id,
|
||
thread_id=thread_ids[coordinator_name],
|
||
agent_name=coordinator_name,
|
||
message=msg,
|
||
model=model or self._model,
|
||
excluded_tools=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=policy.thinking_enabled,
|
||
reasoning_effort=policy.reasoning_effort,
|
||
)
|
||
ai = _latest_ai_message(messages)
|
||
# tool_calls 取最后一条 AI 消息（派活就在这条）；正文取最后一条有文字的 AI 消息。
|
||
final_text = _latest_ai_text(messages)
|
||
tool_calls = (ai.get("tool_calls") if ai else None) or []
|
||
# 本轮工具调用（派活 / 澄清），带进 LeaderTurn 供草稿 run 重建步骤卡。
|
||
turn_tool_calls = _normalize_tool_calls(tool_calls)
|
||
|
||
# 澄清优先
|
||
clar = next((tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"), None)
|
||
if clar is not None:
|
||
cargs = clar.get("args") or {}
|
||
question = str(cargs.get("question") or "").strip() or "总控需要补充信息，请补充后继续。"
|
||
return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls)
|
||
|
||
# 派活解析（复用 multi_agent 纯函数）
|
||
from app.gateway.routers.multi_agent import _extract_dispatch
|
||
|
||
dispatched: list[tuple[str, str]] = []
|
||
for tc in tool_calls:
|
||
try:
|
||
name, task = _extract_dispatch(tc)
|
||
if name and task:
|
||
dispatched.append((name, task))
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("dispatch parse failed for tool_call", exc_info=True)
|
||
return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls)
|
||
|
||
async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
|
||
policy = roundtable_run_policy("seat")
|
||
# 席位模式覆盖 policy 的 thinking/reasoning（None → policy 默认），与前端 _special_run 一致。
|
||
messages = await _run_agent_turn(
|
||
self._app,
|
||
user_id=self._user_id,
|
||
thread_id=thread_ids[agent_id],
|
||
agent_name=agent_id,
|
||
message=task,
|
||
model=model or self._model,
|
||
excluded_tools=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=(
|
||
policy.thinking_enabled
|
||
if self._seat_thinking_enabled is None
|
||
else self._seat_thinking_enabled
|
||
),
|
||
reasoning_effort=(
|
||
policy.reasoning_effort
|
||
if self._seat_reasoning_effort is None
|
||
else self._seat_reasoning_effort
|
||
),
|
||
subagent_enabled=self._seat_subagent_enabled,
|
||
)
|
||
return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages))
|
||
|
||
async def create_report_thread(self) -> str:
|
||
ensure_roundtable_functional_agents()
|
||
return await _create_thread(self._app)
|
||
|
||
async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult:
|
||
"""跑报告 agent，**带产物自愈**：首轮后 HTML 缺失/截断（无 </html>）→ 同线程追加
|
||
纠偏消息（分段 + append 续写）重试，至多 ``_ARTIFACT_RETRY_LIMIT`` 轮。"""
|
||
policy = roundtable_run_policy("report")
|
||
summary = ""
|
||
html: str | None = None
|
||
message = prompt
|
||
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
|
||
messages = await _run_agent_turn(
|
||
self._app,
|
||
user_id=self._user_id,
|
||
thread_id=report_thread_id,
|
||
agent_name=REPORT_AGENT_ID,
|
||
message=message,
|
||
model=model or self._model,
|
||
excluded_tools=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=policy.thinking_enabled,
|
||
reasoning_effort=policy.reasoning_effort,
|
||
)
|
||
summary = _latest_ai_text(messages) or summary
|
||
html = _read_report_html(self._app, report_thread_id, self._user_id)
|
||
if _html_complete(html):
|
||
break
|
||
if attempt >= _ARTIFACT_RETRY_LIMIT:
|
||
break
|
||
message = _REPORT_RETRY_TRUNCATED if (html and html.strip()) else _REPORT_RETRY_MISSING
|
||
logger.warning(
|
||
"report html %s after attempt %d for thread %s; retrying with corrective prompt",
|
||
"truncated" if (html and html.strip()) else "missing",
|
||
attempt + 1,
|
||
report_thread_id,
|
||
)
|
||
# 截断的 HTML 也比纯文字强（浏览器容忍未闭合标签），仅在完全没有文件时才回退正文。
|
||
final_html = html if (html and html.strip()) else summary
|
||
return ReportResult(html=final_html, summary=summary, model=model or self._model)
|
||
|
||
async def create_summary_thread(self) -> str:
|
||
ensure_roundtable_functional_agents()
|
||
return await _create_thread(self._app)
|
||
|
||
async def run_summary(self, *, summary_thread_id, prompt, model) -> SummaryResult:
|
||
"""进程内驱动 roundtable-summary 写「方案总结报告.md」+ 对话正文（与前端
|
||
useStep3Summary 同一智能体 / 同一 run policy / 同款产物），返回与
|
||
DraftStep3SummarySnapshot 对齐的结果。线程是真实持久线程——加载草稿后可续问。
|
||
|
||
带产物自愈：md 文件缺失时同线程追加纠偏消息重试（同 ``run_report``）。"""
|
||
policy = roundtable_run_policy("summary")
|
||
closing = ""
|
||
md: str | None = None
|
||
message = prompt
|
||
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
|
||
messages = await _run_agent_turn(
|
||
self._app,
|
||
user_id=self._user_id,
|
||
thread_id=summary_thread_id,
|
||
agent_name=SUMMARY_AGENT_ID,
|
||
message=message,
|
||
model=model or self._model,
|
||
excluded_tools=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=policy.thinking_enabled,
|
||
reasoning_effort=policy.reasoning_effort,
|
||
)
|
||
closing = _latest_ai_text(messages) or closing
|
||
md = _read_summary_md(self._app, summary_thread_id, self._user_id)
|
||
if md and md.strip():
|
||
break
|
||
if attempt >= _ARTIFACT_RETRY_LIMIT:
|
||
break
|
||
message = _SUMMARY_RETRY_MISSING
|
||
logger.warning(
|
||
"summary md missing after attempt %d for thread %s; retrying with corrective prompt",
|
||
attempt + 1,
|
||
summary_thread_id,
|
||
)
|
||
final_md = md if (md and md.strip()) else closing
|
||
return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id)
|
||
|
||
async def run_dashboard_data(self, *, prompt, model) -> str | None:
|
||
"""大屏多页（模式二）的结构化数据：进程内驱动 roundtable-report 输出
|
||
```report-json（prompt 已声明禁工具/禁文件），返回可解析的 JSON 字符串。
|
||
缺失 / 坏 JSON → 同线程纠偏重试（同 run_report/run_summary 的自愈语义）。"""
|
||
ensure_roundtable_functional_agents()
|
||
thread_id = await _create_thread(self._app)
|
||
policy = roundtable_run_policy("report")
|
||
message = prompt
|
||
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
|
||
messages = await _run_agent_turn(
|
||
self._app,
|
||
user_id=self._user_id,
|
||
thread_id=thread_id,
|
||
agent_name=REPORT_AGENT_ID,
|
||
message=message,
|
||
model=model or self._model,
|
||
excluded_tools=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=policy.thinking_enabled,
|
||
reasoning_effort=policy.reasoning_effort,
|
||
)
|
||
body = _extract_report_json(_latest_ai_text(messages))
|
||
if body:
|
||
return body
|
||
if attempt >= _ARTIFACT_RETRY_LIMIT:
|
||
break
|
||
message = _DASHBOARD_RETRY
|
||
logger.warning(
|
||
"dashboard report-json missing/invalid after attempt %d for thread %s; retrying",
|
||
attempt + 1,
|
||
thread_id,
|
||
)
|
||
return None
|