346 lines
15 KiB
Python
346 lines
15 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 logging
|
||
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,
|
||
ensure_roundtable_functional_agents,
|
||
)
|
||
from deerflow.agents.roundtable_orchestrator.types import (
|
||
LeaderTurn,
|
||
ReportResult,
|
||
SeatRef,
|
||
SeatTurn,
|
||
)
|
||
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"
|
||
|
||
# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy
|
||
# （与 multi_agent 的 _leader_run / _special_run 同源），不再在本文件硬编码——历史上
|
||
# 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。
|
||
|
||
|
||
# ── 进程内底层操作 ───────────────────────────────────────────────────────────
|
||
|
||
|
||
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,
|
||
*,
|
||
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,
|
||
) -> 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（临时研讨任务）。
|
||
"""
|
||
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,
|
||
"subagent_enabled": False,
|
||
"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))
|
||
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 _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) -> None:
|
||
self._app = app
|
||
self._user_id = user_id
|
||
self._agents = list(agents)
|
||
self._model = model
|
||
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,
|
||
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")
|
||
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=policy.excluded_tools,
|
||
skill_stop_names=policy.skill_stop_names,
|
||
thinking_enabled=policy.thinking_enabled,
|
||
reasoning_effort=policy.reasoning_effort,
|
||
)
|
||
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=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)
|
||
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)
|