后台挂起功能优化

This commit is contained in:
fengxiangdebaozi committed 2026-06-05 10:59:56 +08:00
1 parent cca3d8eb41
commit 9bbd04d6cf
14 files changed
+846 -136

No files matched your search

@@ -22,6 +22,12 @@ 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,
ensure_roundtable_functional_agents,
)
from deerflow.agents.roundtable_orchestrator.types import (
LeaderTurn,
ReportResult,
@@ -31,22 +37,14 @@ from deerflow.agents.roundtable_orchestrator.types import (
from deerflow.config.paths import get_paths
from deerflow.runtime.serialization import serialize_channel_values
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
ensure_roundtable_functional_agents,
)
logger = logging.getLogger(__name__)
_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
_REPORT_FILE = "方案总览.html"
# 各角色的工具裁剪(镜像 multi_agent 的 _leader_run / _special_run)。
_LEADER_EXCLUDED = ["web_search"]
_LEADER_SKILL_STOP = ["agent_orchestration"]
_SEAT_EXCLUDED = ["ask_clarification", "present_files", "view_image", "agent_orchestration"]
_REPORT_EXCLUDED = [*_SEAT_EXCLUDED, "bash", "str_replace"]
# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy
# (与 multi_agent 的 _leader_run / _special_run 同源),不再在本文件硬编码——历史上
# 这里漏禁了 web_search,导致后台席位 / 报告能联网搜索而前端不能,现由 policy 统一。
# ── 进程内底层操作 ───────────────────────────────────────────────────────────
@@ -101,6 +99,56 @@ def _message_text(message: dict[str, Any] | None) -> str:
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,
*,
@@ -216,27 +264,31 @@ class InProcessRoundtableGateway:
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,
thread_id=thread_ids[coordinator_name],
agent_name=coordinator_name,
message=msg,
model=model or self._model,
excluded_tools=_LEADER_EXCLUDED,
skill_stop_names=_LEADER_SKILL_STOP,
thinking_enabled=False,
reasoning_effort="low",
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)
final_text = _message_text(ai)
# 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)
return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls)
# 派活解析(复用 multi_agent 纯函数)
from app.gateway.routers.multi_agent import _extract_dispatch
@@ -249,38 +301,40 @@ class InProcessRoundtableGateway:
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)
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")
messages = await _run_agent_turn(
self._app,
thread_id=thread_ids[agent_id],
agent_name=agent_id,
message=task,
model=model or self._model,
excluded_tools=_SEAT_EXCLUDED,
skill_stop_names=[],
thinking_enabled=True,
reasoning_effort="medium",
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
return SeatTurn(final_text=_message_text(_latest_ai_message(messages)))
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:
policy = roundtable_run_policy("report")
messages = await _run_agent_turn(
self._app,
thread_id=report_thread_id,
agent_name=REPORT_AGENT_ID,
message=prompt,
model=model or self._model,
excluded_tools=_REPORT_EXCLUDED,
skill_stop_names=[],
thinking_enabled=True,
reasoning_effort="medium",
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
summary = _message_text(_latest_ai_message(messages))
summary = _latest_ai_text(messages)
html = _read_report_html(self._app, report_thread_id, self._user_id) or summary
return ReportResult(html=html, summary=summary, model=model or self._model)
@@ -17,7 +17,8 @@ from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Callable
from datetime import UTC, datetime
from typing import Any, Callable
from deerflow.agents.roundtable_orchestrator import (
JobProgressWriter,
@@ -36,6 +37,115 @@ class _JobUser:
self.id = user_id
# 头像配色:与前端 getAvatarTypeFor 对齐(内置圆桌席位固定色,其余按序轮换)。
_BUILTIN_AVATAR = {
"roundtable-intelligence": "purple",
"roundtable-environment": "emerald",
"roundtable-solution-design": "human",
"roundtable-risk-review": "amber",
"roundtable-execution-plan": "pink",
"roundtable-summary": "blue",
}
_FALLBACK_AVATAR = ["blue", "purple", "emerald", "amber", "pink"]
def _avatar_for(agent_id: str, index: int) -> str:
return _BUILTIN_AVATAR.get(agent_id) or _FALLBACK_AVATAR[index % len(_FALLBACK_AVATAR)]
# 工具步骤卡的回退文案。前端 getStepDisplay 对 tool 步骤是从 key+args 推图标/文案的,
# 存的 label 基本不展示——这里给个合理回退即可(与前端默认文案对齐)。
_STEP_LABEL = {
"agent_orchestration": "派活给子智能体",
"write_file": "写入文件",
"str_replace": "写入文件",
"read_file": "读取文件",
"ls": "列出文件夹",
"bash": "执行命令",
"web_search": "搜索相关信息",
"ask_clarification": "需要你的协助",
"present_files": "展示文件",
}
def _steps_from_tool_calls(tool_calls: list[dict], dialogue_id: object, ts: str) -> list[dict]:
"""把一轮 tool_calls 映射成前端 ``Step2Dialogue.steps``(工具步骤卡)。
- ``key`` = ``tool_calling:<toolName>``——前端 ``getStepDisplay`` 据此推图标 + 文案;
- ``id`` = ``<dialogueId>-<i>``——对齐前端 ``${bubbleId}-${stepIndex}``;
- ``args`` 原样带上(write_file 的 path/content、派活的 agent_name/task 等)。
"""
steps: list[dict] = []
for i, tc in enumerate(tool_calls or []):
name = tc.get("name") or ""
if not name:
continue
steps.append(
{
"id": f"{dialogue_id}-{i}",
"key": f"tool_calling:{name}",
"label": _STEP_LABEL.get(name, f"使用 “{name}” 工具"),
"ts": ts,
"args": tc.get("args") or {},
}
)
return steps
def _seat_stub_messages(dialogues: list[dict]) -> dict[str, list[dict]]:
"""从各对话的 write_file 调用重建 ``seatStubMessages``(threadId → [stub ai message])。
对齐前端 ``upsertSeatToolCall`` 的 stub 形状与 id 约定(消息 ``roundtable-stub-<tid>``、
工具调用 ``tc-<tid>-<i>``),使加载草稿后点开产物的虚拟 URL 能正确解析到内容。
同一 thread 跨轮写多次按 path 去重(后写覆盖),与前端一致。
"""
by_tid: dict[str, dict[str, dict]] = {}
for d in dialogues:
tid = d.get("threadId")
if not tid:
continue
for tc in d.get("toolCalls") or []:
if tc.get("name") != "write_file":
continue
path = str((tc.get("args") or {}).get("path") or "")
by_tid.setdefault(tid, {})[path] = tc # 后写覆盖
out: dict[str, list[dict]] = {}
for tid, by_path in by_tid.items():
tool_calls = [
{"id": f"tc-{tid}-{i}", "name": "write_file", "args": tc.get("args") or {}}
for i, tc in enumerate(by_path.values())
]
out[tid] = [{"id": f"roundtable-stub-{tid}", "type": "ai", "content": "", "tool_calls": tool_calls}]
return out
def _to_step2_dialogue(d: dict, idx_by_agent: dict[str, int]) -> dict:
"""把作业的中性对话记录映射成前端 Step2Dialogue 形状(含 avatarType / steps /
agentThreadId / time),使写进草稿 step2.runs 后,前端 hydrateStep2Run 能像手动跑
一样直接渲染——步骤卡、时间戳、点开产物全部可用。"""
dialogue_id = d.get("id")
time_str = d.get("time") or ""
base: dict = {
"id": dialogue_id,
"content": d.get("content") or "",
"time": time_str,
"streaming": False,
"steps": _steps_from_tool_calls(d.get("toolCalls") or [], dialogue_id, time_str),
}
thread_id = d.get("threadId")
if thread_id:
base["agentThreadId"] = thread_id
if d.get("role") == "leader":
return {**base, "sender": "总控协调", "avatarType": "blue", "confidence": "派活说明"}
aid = d.get("agentId") or ""
return {
**base,
"sender": d.get("name") or aid,
"avatarType": _avatar_for(aid, idx_by_agent.get(aid, 0)),
"confidence": "子智能体交付",
}
@dataclass
class StartParams:
job_id: str
@@ -52,6 +162,8 @@ class StartParams:
# 是否为「续跑」(resume 澄清)。首启=False → 真网关建全新线程(不复用前端线程,
# 避免与前端本地研讨的活跃 run 撞 409);续跑=True → 复用作业自己的线程。
is_resume: bool = False
# chain 模式来源链条(写进草稿 run 用,展示)。
chain: dict[str, Any] | None = None
# 网关工厂:给定一次启动参数,返回一个 RoundtableGateway 实例。
@@ -94,7 +206,19 @@ class RoundtableJobExecutor:
except Exception: # noqa: BLE001
logger.warning("persist thread_ids for job %s failed", params.job_id, exc_info=True)
progress = JobProgressWriter(self._store, job_id=params.job_id, agents=params.agents)
# 每轮模型返回后,把对话作为一个带 jobId 的 run 写进草稿 step2.runs[]
# —— 和手动跑同一存取路径(draft_store.update_draft),加载草稿即可显示。
async def _on_dialogue(dialogues: list[dict], consensus: int) -> None:
await self._write_run_to_draft(
params, dialogues=dialogues, status="running", consensus=consensus, thread_ids=thread_ids
)
progress = JobProgressWriter(
self._store,
job_id=params.job_id,
agents=params.agents,
on_dialogue=_on_dialogue if (self._draft_store and params.draft_id) else None,
)
result = await run_orchestration(
gateway=gateway,
progress=progress,
@@ -106,26 +230,21 @@ class RoundtableJobExecutor:
mode=params.mode,
intent_text=params.intent_text,
)
# 完成且有报告:
# 1) 把 step3 挂到 job 上 —— 前端「多运行」模型从单个 job 即可拿到完整结果
# (对话 dialogues + 报告 step3)组装一个 run;
# 2) 同时写回关联草稿的顶层 step3 + 推进 furthest_step=3,兼容既有
# 「查看报告 → 加载草稿 → Step3」路径。
# 终态:把最终对话 + step3 + jobStatus 落进同一个 run(含 furthest_step=3)。
await self._write_run_to_draft(
params,
dialogues=result.dialogues,
status=result.status,
consensus=100 if result.status == "done" else 95,
step3=result.step3,
thread_ids=thread_ids,
)
# step3 也挂 job(API / 弹窗用)。
if result.status == "done" and result.step3:
try:
await self._store.update_progress(params.job_id, step3=result.step3)
except Exception: # noqa: BLE001
logger.warning("persist step3 to job %s failed", params.job_id, exc_info=True)
if params.draft_id and self._draft_store is not None:
try:
await self._draft_store.update_draft(
params.draft_id,
params.user_id,
step3=result.step3,
furthest_step=3,
)
except Exception: # noqa: BLE001 — 写报告失败不影响作业本身已 done
logger.warning("persist step3 to draft %s failed", params.draft_id, exc_info=True)
except asyncio.CancelledError:
# 干净停止即可,**不在被取消的任务里写 DB**:取消态下二次 await 会再抛
# CancelledError、并可能打断进行中的 sqlite 连接。``cancelled`` 终态由
@@ -137,6 +256,72 @@ class RoundtableJobExecutor:
finally:
reset_current_user(token)
async def _write_run_to_draft(
self,
params: StartParams,
*,
dialogues: list[dict],
status: str,
consensus: int,
step3: dict | None = None,
thread_ids: dict[str, str] | None = None,
) -> None:
"""把本次后台作业作为一个带 jobId 的 run 写进草稿 step2.runs[](同手动跑存取)。
读现有 step2 → 按 run_id=``job-<jobId>`` upsert(保留其它手动 run)→ 设为 activeRun
→ update_draft。前端加载草稿即可像手动跑一样 hydrate 出对话。
"""
if self._draft_store is None or not params.draft_id:
return
idx_by_agent = {a.agent_id: i for i, a in enumerate(params.agents)}
s2_dialogues = [_to_step2_dialogue(d, idx_by_agent) for d in dialogues]
last_leader = next(
(d.get("content") for d in reversed(dialogues) if d.get("role") == "leader"), ""
)
run_id = f"job-{params.job_id}"
try:
draft = await self._draft_store.get_draft(params.draft_id, params.user_id)
except Exception: # noqa: BLE001
draft = None
step2 = dict((draft or {}).get("step2") or {})
runs = list(step2.get("runs") or [])
idx = next((i for i, r in enumerate(runs) if r.get("id") == run_id), -1)
created = runs[idx].get("createdAt") if idx >= 0 else datetime.now(UTC).isoformat()
prev_step3 = runs[idx].get("step3") if idx >= 0 else None
run = {
"id": run_id,
"title": "后台研讨",
"createdAt": created,
"source": params.mode,
"selectedAgents": [{"agent_id": a.agent_id, "name": a.name} for a in params.agents],
# 后台真建的 leader/seat 线程是真实持久的——存进去,前端加载后能像手动跑一样
# 按 thread 打开各席位产物(HTTP outputs 兜底)。hydrate 是纯展示、绝不 resume,
# 不会触发重跑(init 闸门 + isRoundTableRunning=false 双重拦截)。空则存 None。
"threadIds": (thread_ids or None),
"coordinatorName": params.coordinator_name,
"step2RoundtableDialogues": s2_dialogues,
"lastLeaderContent": last_leader or "",
"hasConsensus": status == "done",
"consensusPercentage": int(consensus),
"budgetLimit": 0,
# 从各席位 write_file 调用重建 stub,使加载草稿后点开产物的虚拟预览可用。
"seatStubMessages": _seat_stub_messages(dialogues),
"orchestrationMode": params.mode,
"chain": params.chain,
"step3": step3 if step3 is not None else prev_step3,
"jobId": params.job_id,
"jobStatus": status,
}
runs = [run if i == idx else r for i, r in enumerate(runs)] if idx >= 0 else [*runs, run]
step2["runs"] = runs
step2["activeRunId"] = run_id
try:
await self._draft_store.update_draft(
params.draft_id, params.user_id, step2=step2, furthest_step=3 if status == "done" else 2
)
except Exception: # noqa: BLE001 — 写草稿失败不影响作业本身
logger.warning("write run to draft %s failed", params.draft_id, exc_info=True)
def cancel(self, job_id: str) -> bool:
"""请求取消一个在跑的作业。返回是否命中在跑任务。"""
task = self._tasks.get(job_id)
@@ -0,0 +1,94 @@
"""圆桌各角色「单轮 run 参数」的**单一数据源**。
历史问题:前端手动路径(``multi_agent.py`` 的 ``_leader_run`` / ``_special_run``)
和后台进程内网关(``roundtable_inprocess_gateway.py``)各自硬编码了一份「每角色
工具裁剪 + thinking + reasoning」,结果**漂移**出差异(后台席位 / 报告少禁了
``web_search``,能联网搜索而前端不能)。
本模块把这套策略收敛成唯一定义,两条路径都从这里取,保证「后台挂起」与「前端
手动调用」传参**逐字一致**,且以后改一处即可,不再漂移。
角色(``role``):
- ``leader`` —— 总控派活:短决策,关 thinking + low reasoning(首字节快);
只禁 ``web_search``;并配合 ``SkillStopMiddleware`` 在 ``agent_orchestration``
调用后停图,便于网关从 AIMessage.tool_calls 解析派活列表。
- ``seat`` —— 子智能体交付:开 thinking + medium reasoning;禁派活 /
澄清 / 展示文件 / 看图 / 联网搜索(与前端 ``_special_run`` 一致)。
- ``report`` —— Step3 报告绘制:在 ``seat`` 基础上**额外禁 ``bash`` /
``str_replace``**:
* 禁 ``bash`` → 杜绝它写 Python 脚本再跑出 HTML(脚本产物落在 outputs,
但前端虚拟预览只抓 ``write_file`` 的 ``args.content``,抓不到脚本产物 →
预览停在半截 ``write_file``,正是「没写完就去生成 py 文件」的现象);
* 禁 ``str_replace`` → 强制返工用整文件 ``write_file`` 重写
(``str_replace`` 没有 ``content`` 字段,会让前端虚拟预览读到 undefined
而清空沙箱)。
"""
from __future__ import annotations
from dataclasses import dataclass
# ── 每角色基础工具裁剪(唯一定义处)─────────────────────────────────────────────
#
# 注意:``seat`` / ``report`` 都包含 ``web_search`` —— 圆桌研讨里联网搜索一律禁用
# (与前端 ``_special_run`` 的 excluded_tools 对齐)。leader 也禁 ``web_search``。
_LEADER_EXCLUDED: list[str] = ["web_search"]
_SEAT_EXCLUDED: list[str] = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"]
_REPORT_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "bash", "str_replace"]
# leader 内禀的 skill_stop:``agent_orchestration`` 调用后停图。调用方可在此基础上
# 再叠加自己的 skill_stop(见 ``merge_skill_stop_names``)。
_LEADER_SKILL_STOP: list[str] = ["agent_orchestration"]
@dataclass(frozen=True)
class RoundtableRunPolicy:
"""一个角色单轮 run 的归一化参数。
与 ``multi_agent._run_payload`` 的入参一一对应:``excluded_tools`` /
``skill_stop_names`` / ``thinking_enabled`` / ``reasoning_effort``。
"""
excluded_tools: list[str]
skill_stop_names: list[str]
thinking_enabled: bool
reasoning_effort: str
def roundtable_run_policy(role: str) -> RoundtableRunPolicy:
"""返回某角色(``leader`` / ``seat`` / ``report``)的单轮 run 参数。
返回值里的列表是**新副本**,调用方可安全地再 append/merge 而不污染本模块常量。
未知角色抛 ``ValueError``(早失败,避免悄悄退化到错误的工具面)。
"""
if role == "leader":
return RoundtableRunPolicy(
excluded_tools=list(_LEADER_EXCLUDED),
skill_stop_names=list(_LEADER_SKILL_STOP),
thinking_enabled=False,
reasoning_effort="low",
)
if role == "seat":
return RoundtableRunPolicy(
excluded_tools=list(_SEAT_EXCLUDED),
skill_stop_names=[],
thinking_enabled=True,
reasoning_effort="medium",
)
if role == "report":
return RoundtableRunPolicy(
excluded_tools=list(_REPORT_EXCLUDED),
skill_stop_names=[],
thinking_enabled=True,
reasoning_effort="medium",
)
raise ValueError(f"unknown roundtable role: {role!r} (expected leader/seat/report)")
def merge_skill_stop_names(policy_names: list[str], caller_names: list[str] | None) -> list[str]:
"""把角色内禀 skill_stop 与调用方追加的 skill_stop 合并去重(保序)。
取代历史上 ``list({"agent_orchestration", *skill_stop_names})`` 的 set 写法
(set 顺序不确定);这里用 dict.fromkeys 保序去重,结果确定、便于断言。
"""
return list(dict.fromkeys([*policy_names, *(caller_names or [])]))
@@ -57,18 +57,20 @@ import uuid
from typing import Any, AsyncIterator
import httpx
from fastapi import APIRouter, Body, HTTPException, Request
from fastapi.responses import StreamingResponse
from langgraph.checkpoint.base import empty_checkpoint
from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.config import get_app_config
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
from deerflow.utils.time import now_iso
from fastapi import APIRouter, Body, HTTPException, Request
from fastapi.responses import StreamingResponse
from langgraph.checkpoint.base import empty_checkpoint
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])
@@ -1008,19 +1010,20 @@ async def _leader_run(
_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})
# 角色参数取自单一数据源 roundtable_run_policy(leader:禁 web_search、停
# agent_orchestration 图、关 thinking + low reasoning——短决策、首字节快)。
# 调用方可再叠加自己的 skill_stop。后台进程内网关用同一份 policy,保证传参一致。
policy = roundtable_run_policy("leader")
effective_stop_names = merge_skill_stop_names(policy.skill_stop_names, 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"],
excluded_tools=policy.excluded_tools,
skill_stop_names=effective_stop_names,
# 总控派活是"下一步派谁、什么任务"的短决策,不需要 CoT。
# 关掉 thinking + low reasoning 让"调度中"几乎瞬间转入"正在派活",
# 子智能体仍保留 thinking(在 _special_run 里默认 True),保证交付质量。
thinking_enabled=False,
reasoning_effort="low",
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
t_payload_built = time.perf_counter()
logger.info(
@@ -1163,23 +1166,20 @@ async def _special_run(
"""
t0 = time.perf_counter()
thread_id = agent_threads[agent_name]
excluded_tools = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"]
if agent_name == REPORT_AGENT_ID:
# Step 3「结果绘制」的报告智能体只需**直接 write_file 写出完整 HTML**。
# 额外禁用 bash / str_replace:
# - 禁 bash → 杜绝它写 Python/脚本再跑出 HTML(脚本产物落在 outputs 里,
# 但前端虚拟预览只抓 write_file 的 args.content,抓不到脚本产物 → 预览停在
# 那个半截 write_file 上,正是用户看到的「没写完就去生成 py 文件」);
# - 禁 str_replace → 强制返工用整文件 write_file 重写(str_replace 没有 content
# 字段,会让前端虚拟预览读到 undefined 而清空沙箱)。
excluded_tools = [*excluded_tools, "bash", "str_replace"]
# 角色参数取自单一数据源 roundtable_run_policy(seat / report)。report 在 seat
# 基础上额外禁 bash / str_replace(理由见 roundtable_run_policy 模块注释)。
# 后台进程内网关用同一份 policy,保证「后台挂起」与「前端手动调用」传参一致。
role = "report" if agent_name == REPORT_AGENT_ID else "seat"
policy = roundtable_run_policy(role)
body = _run_payload(
agent_name=agent_name,
model_name=model_name,
thread_id=thread_id,
new_message=new_message,
excluded_tools=excluded_tools,
skill_stop_names=skill_stop_names,
excluded_tools=policy.excluded_tools,
skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names),
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
t_payload_built = time.perf_counter()
logger.info(
@@ -203,6 +203,7 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes
thread_ids=body.threadIds,
coordinator_name=body.coordinatorName,
draft_id=body.draftId,
chain=body.chain,
)
)
return _to_response(row)
@@ -287,6 +288,7 @@ async def resume_job(request: Request, job_id: str, body: ResumeJobRequest) -> R
coordinator_name=row.get("coordinator_name") or "roundtable-coordinator",
draft_id=row.get("draft_id"),
is_resume=True,
chain=row.get("chain"),
)
)
return _to_response(updated or row)
@@ -47,6 +47,7 @@ class OrchestrationResult:
pending_clarification: str | None = None
error: str | None = None
last_leader_content: str = ""
dialogues: list = field(default_factory=list)
@dataclass
@@ -85,12 +86,13 @@ async def run_orchestration(
mode: "recommend" | "chain"。
intent_text: 任务意图,用于报告 prompt(缺省退回 seed_message)。
"""
intent = (intent_text or seed_message or "").strip()
try:
await progress.begin()
if mode == "chain":
outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message)
outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent)
else:
outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message)
outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent)
if outcome.kind == "clarification":
question = outcome.question or "总控需要补充信息,请补充后继续。"
@@ -99,6 +101,7 @@ async def run_orchestration(
status="awaiting_input",
pending_clarification=question,
last_leader_content=outcome.last_leader_content,
dialogues=progress.dialogues(),
)
# consensus 或 max_cycles:均收口 → 报告绘制
@@ -118,14 +121,48 @@ async def run_orchestration(
consensus=True,
step3=step3,
last_leader_content=outcome.last_leader_content,
dialogues=progress.dialogues(),
)
except Exception as exc: # noqa: BLE001 — 顶层兜底:标记 error 并返回
logger.exception("roundtable orchestration failed")
await progress.error(str(exc))
return OrchestrationResult(status="error", error=str(exc))
return OrchestrationResult(status="error", error=str(exc), dialogues=progress.dialogues())
# ── recommend 模式 ───────────────────────────────────────────────────────────
# ── 上下文「广播」(inline 实现)────────────────────────────────────────────────
#
# 进程内网关没有 multi_agent 的 thread 广播,席位/总控会看不到「原始任务背景 + 各席位
# 前序交付」→ 席位反问用户、总控靠记忆瞎编。这里把这些信息**直接拼进 prompt**:
# - 每个席位的任务带上:任务背景 + 前序各席位交付 + 「直接产出、别提问」硬约束;
# - 总控的续轮 prompt 带上:各席位本轮交付,让它据实派活/收口。
_SEAT_NO_ASK = (
"注意:这是自动化圆桌研讨,没有用户在线应答——**请勿向用户提问或等待确认**,"
"直接基于已知信息做出合理假设并产出完整的结构化交付(Markdown)。"
)
def _deliveries_block(deliveries: dict[str, str], name_by_id: dict[str, str]) -> str:
parts: list[str] = []
for aid, text in deliveries.items():
t = (text or "").strip()
if not t:
continue
parts.append(f"### {name_by_id.get(aid, aid)}\n{t}")
return "\n\n".join(parts)
def _seat_message(intent_text: str, deliveries: dict[str, str], name_by_id: dict[str, str], task: str) -> str:
"""把任务背景 + 前序交付 + 防提问约束拼进席位任务(替代 thread 广播)。"""
parts: list[str] = []
if intent_text.strip():
parts.append(f"【本次研讨任务背景】\n{intent_text.strip()}")
block = _deliveries_block(deliveries, name_by_id)
if block:
parts.append("【前序各席位已完成的交付,请在此基础上承接、补充、不要重复或推翻】\n" + block)
parts.append(f"【你的本轮具体任务】\n{task}")
parts.append(_SEAT_NO_ASK)
return "\n\n---\n\n".join(parts)
async def _run_recommend(
@@ -136,6 +173,7 @@ async def _run_recommend(
coordinator_name: str,
model: str | None,
seed_message: str,
intent_text: str,
) -> _LoopOutcome:
deliveries: dict[str, str] = {}
last_leader = ""
@@ -152,24 +190,32 @@ async def _run_recommend(
)
if turn.final_text:
last_leader = turn.final_text
await progress.add_leader_dialogue(turn.final_text, cycle)
await progress.add_leader_dialogue(
turn.final_text, cycle, tool_calls=turn.tool_calls, thread_id=thread_ids.get(coordinator_name)
)
if turn.is_clarification:
return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries)
if turn.is_consensus:
return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
# **顺序**派活 —— 不要并行
# **顺序**派活 —— 不要并行;每个席位带上「背景 + 前序交付」上下文
await progress.dispatching(cycle)
for agent_id, task in turn.dispatched:
await progress.seat_active(agent_id)
seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=task, model=model)
seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model)
deliveries[agent_id] = seat.final_text
await progress.add_seat_dialogue(agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle)
await progress.add_seat_dialogue(
agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle,
tool_calls=seat.tool_calls, thread_id=thread_ids.get(agent_id),
)
await progress.seat_done(agent_id, cycle)
await progress.bump_consensus()
message = CONTINUE_PROMPT
# 续轮:把各席位本轮交付喂给总控,让它据实判断(总控 SOUL 依赖看到交付)
block = _deliveries_block(deliveries, name_by_id)
message = CONTINUE_PROMPT + (f"\n\n各席位最新交付如下:\n{block}" if block else "")
# 达到最大轮次:强制收口(对齐前端 toast「已达到最大研讨轮次」)
logger.info("roundtable recommend loop hit MAX_CYCLES=%d, forcing consensus", MAX_CYCLES)
@@ -187,13 +233,15 @@ async def _run_chain(
coordinator_name: str,
model: str | None,
seed_message: str,
intent_text: str,
) -> _LoopOutcome:
deliveries: dict[str, str] = {}
last_leader = ""
total = len(agents)
name_by_id = {a.agent_id: a.name for a in agents}
for i, seat in enumerate(agents):
# (a) 总控按链条顺序只给当前席位派活
# (a) 总控按链条顺序只给当前席位派活;带上前序各席位交付,让它据实派活
await progress.leader_thinking(i + 1)
leader_task = build_chain_leader_task(
seed_message=seed_message,
@@ -203,6 +251,9 @@ async def _run_chain(
seat_name=seat.name,
is_first=(i == 0),
)
block = _deliveries_block(deliveries, name_by_id)
if block:
leader_task += f"\n\n前序各席位已完成的交付如下,请据此为本席位分配承接性任务:\n{block}"
turn = await gateway.run_leader(
thread_ids=thread_ids,
coordinator_name=coordinator_name,
@@ -211,34 +262,44 @@ async def _run_chain(
)
if turn.final_text:
last_leader = turn.final_text
await progress.add_leader_dialogue(turn.final_text, i + 1)
await progress.add_leader_dialogue(
turn.final_text, i + 1, tool_calls=turn.tool_calls, thread_id=thread_ids.get(coordinator_name)
)
if turn.is_clarification:
return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries)
task = pick_chain_task(turn.dispatched, turn.final_text, seat.agent_id, seat.name)
# (b) 当前席位交付(顺序,不并行)
# (b) 当前席位交付(顺序,不并行);带上「背景 + 前序交付 + 别提问」上下文
await progress.seat_active(seat.agent_id)
seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=task, model=model)
seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
deliveries[seat.agent_id] = seat_turn.final_text
await progress.add_seat_dialogue(seat.agent_id, seat.name, seat_turn.final_text, i + 1)
await progress.add_seat_dialogue(
seat.agent_id, seat.name, seat_turn.final_text, i + 1,
tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id),
)
await progress.seat_done(seat.agent_id, i + 1)
if total > 0:
await progress.set_consensus_pct(round((i + 1) / total * 90))
# (c) 总控收口:综合全链交付,产出最终结论
# (c) 总控收口:综合全链交付,产出最终结论(带上全部席位交付)
if total > 0:
await progress.leader_thinking(total + 1)
block = _deliveries_block(deliveries, name_by_id)
closeout_msg = CHAIN_CLOSEOUT_PROMPT + (f"\n\n各席位完整交付如下:\n{block}" if block else "")
closeout = await gateway.run_leader(
thread_ids=thread_ids,
coordinator_name=coordinator_name,
message=CHAIN_CLOSEOUT_PROMPT,
message=closeout_msg,
model=model,
)
if closeout.final_text:
last_leader = closeout.final_text
await progress.add_leader_dialogue(closeout.final_text, total + 1)
await progress.add_leader_dialogue(
closeout.final_text, total + 1, tool_calls=closeout.tool_calls, thread_id=thread_ids.get(coordinator_name)
)
return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
@@ -279,27 +340,43 @@ def build_report_prompt(
席位交付按 ``agents`` 顺序列出(缺交付的席位跳过)。报告智能体据此用 write_file
写出自包含的 ``outputs/方案总览.html``(具体绘制规则在 roundtable-report 的 SOUL)。
"""
# 与前端 useStep3Report.buildReportPrompt 对齐 —— 关键是**明确叫报告 agent 用
# write_file 写到 /mnt/user-data/outputs/方案总览.html**,否则它只会在对话里回文字、
# 不落 HTML 文件(_read_report_html 就读不到)。
name_by_id = {a.agent_id: a.name for a in agents}
lines: list[str] = [
"【圆桌最终成果汇总 —— 请据此绘制方案总览 HTML 看板】",
"请基于以下圆桌多智能体研讨的成果,撰写一份图文并茂、以文字叙述为主线的 HTML「方案研讨汇报」,"
"用 write_file 工具写入 /mnt/user-data/outputs/方案总览.html。",
"目标:让用户能轻松读懂这群智能体讨论出了什么方案、为什么是它。请逐个席位提炼其核心观点与交付写成文字,"
"再综合成最终方案;图表用来点睛(风险矩阵、时间线、资源占比、贡献雷达等),详略得当——"
"缺明确数字时也要据研讨内容合理估算出图并标注「定性/估算」,不要通篇只有文字。",
"",
"## 任务意图",
"# 一、任务意图",
(intent_text or "").strip() or "(未提供)",
"",
"## 各席位交付",
"# 二、各席位最终交付",
]
name_by_id = {a.agent_id: a.name for a in agents}
has_delivery = False
for a in agents:
for i, a in enumerate(agents):
text = (deliveries.get(a.agent_id) or "").strip()
if not text:
continue
has_delivery = True
lines.append(f"### {name_by_id.get(a.agent_id, a.agent_id)}")
lines.append(f"## {i + 1}. {name_by_id.get(a.agent_id, a.agent_id)}")
lines.append(text)
lines.append("")
if not has_delivery:
lines.append("(各席位暂无可用交付,请基于任务意图与总控结论绘制。)")
lines.append("(各席位暂无文字交付,请基于任务意图与总控结论合理展开成方案。)")
lines.append("")
lines.append("## 总控最终结论")
lines.append("# 三、总控最终共识结论")
lines.append((consensus_text or "").strip() or "(未提供)")
lines.append("")
lines.append(
"请严格遵循你的职责设定:自包含离线(纯 inline SVG 图表、禁止任何外链 CDN)、研讨叙事型结构"
"(结论+KPI → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线)、"
"文字为主图表点睛、浅红主题(主色 #c25555);写完文件后只用一两句话总结,不要把 HTML 源码贴进对话。"
)
lines.append(
'硬性要求:① 全程使用简体中文;② HTML 必须是完整文档,<head> 最前面声明 <meta charset="utf-8">。'
)
return "\n".join(lines)
@@ -10,12 +10,21 @@
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Awaitable, Callable
from .types import SeatRef
logger = logging.getLogger(__name__)
# 北京时区(UTC+8,固定无 DST)——对话气泡时间戳用,对齐前端 nowTime()
# = new Date().toLocaleTimeString("zh-CN", {hour12:false}) 的 "HH:MM:SS" 形状。
_BEIJING_TZ = timezone(timedelta(hours=8))
def _now_hms() -> str:
return datetime.now(_BEIJING_TZ).strftime("%H:%M:%S")
# 节点状态机:pending → active → done。
_PENDING = "pending"
_ACTIVE = "active"
@@ -42,11 +51,15 @@ class JobProgressWriter:
job_id: str,
agents: list[SeatRef],
event_sink: EventSink = None,
on_dialogue: Any | None = None,
) -> None:
self._store = store
self._job_id = job_id
self._agents = list(agents)
self._event_sink = event_sink
# 每轮 leader/seat 模型返回后回调(dialogues, consensus)—— 执行器据此把对话
# 写进草稿(和手动跑同一存取路径)。
self._on_dialogue = on_dialogue
self._state: dict[str, str] = {a.agent_id: _PENDING for a in agents}
self._spoke_cycle: dict[str, int] = {}
self._consensus = 0
@@ -147,20 +160,75 @@ class JobProgressWriter:
# ── 对话记录 ─────────────────────────────────────────────────────────────
def _add(self, role: str, agent_id: str | None, name: str, content: str, cycle: int) -> None:
def _add(
self,
role: str,
agent_id: str | None,
name: str,
content: str,
cycle: int,
*,
tool_calls: list[dict] | None = None,
thread_id: str | None = None,
) -> None:
self._dlg_seq += 1
self._dialogues.append(
{"id": self._dlg_seq, "role": role, "agentId": agent_id, "name": name, "content": content, "cycle": cycle}
{
"id": self._dlg_seq,
"role": role,
"agentId": agent_id,
"name": name,
"content": content,
"cycle": cycle,
# 工具步骤卡数据源(每项 {name, args, id});映射成前端 Step2Dialogue.steps。
"toolCalls": list(tool_calls or []),
# 该气泡对应的 leader/seat 线程——映射成 agentThreadId,供点开产物 / 反查 stub。
"threadId": thread_id,
# 气泡时间戳,**创建时一次性盖章**(不随后续 flush 抖动),对齐前端 nowTime()。
"time": _now_hms(),
}
)
async def add_leader_dialogue(self, content: str, cycle: int) -> None:
"""总控一轮发言(派活说明 / 共识结论)。空内容跳过。"""
def dialogues(self) -> list[dict[str, Any]]:
return list(self._dialogues)
async def _flush_dialogue(self) -> None:
if self._on_dialogue is None:
return
try:
await self._on_dialogue(self.dialogues(), self._consensus)
except Exception: # noqa: BLE001 — 写草稿失败不打断编排
logger.warning("roundtable job dialogue → draft flush failed", exc_info=True)
async def add_leader_dialogue(
self,
content: str,
cycle: int,
*,
tool_calls: list[dict] | None = None,
thread_id: str | None = None,
) -> None:
"""总控一轮发言(派活说明 / 共识结论)。空内容跳过。
``tool_calls`` / ``thread_id`` 用于重建工具步骤卡(派活 chip)+ agentThreadId。
"""
if not (content or "").strip():
return
self._add("leader", None, "总控协调", content, cycle)
self._add("leader", None, "总控协调", content, cycle, tool_calls=tool_calls, thread_id=thread_id)
await self._persist()
await self._flush_dialogue()
async def add_seat_dialogue(self, agent_id: str, name: str, content: str, cycle: int) -> None:
"""某席位一轮交付。"""
self._add("seat", agent_id, name, content or "", cycle)
async def add_seat_dialogue(
self,
agent_id: str,
name: str,
content: str,
cycle: int,
*,
tool_calls: list[dict] | None = None,
thread_id: str | None = None,
) -> None:
"""某席位一轮交付。``tool_calls`` / ``thread_id`` 用于步骤卡 + 产物预览。"""
self._add("seat", agent_id, name, content or "", cycle, tool_calls=tool_calls, thread_id=thread_id)
await self._persist()
await self._flush_dialogue()
@@ -52,6 +52,9 @@ class LeaderTurn:
dispatched: list[tuple[str, str]] = field(default_factory=list)
# 命中澄清时的问题文本;非 None 即 clarification 分支。
clarification: str | None = None
# 本轮总控调用的工具(按出现顺序,每项 {name, args, id})——通常是 agent_orchestration
# 派活 / ask_clarification。用于在草稿 run 里重建「工具步骤卡」,与前端手动跑一致。
tool_calls: list[dict] = field(default_factory=list)
@property
def is_clarification(self) -> bool:
@@ -67,6 +70,9 @@ class SeatTurn:
"""某席位一轮交付的归一化结果(由 gateway.run_seat 产出)。"""
final_text: str = ""
# 本轮席位调用的工具(按出现顺序,每项 {name, args, id})——用于在草稿 run 里
# 重建「工具步骤卡」+ write_file stub(点开产物),与前端手动跑一致。
tool_calls: list[dict] = field(default_factory=list)
@dataclass