wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(roundtable): 轻量 thread 接口降低广播与席位状态核对开销

Browse Source 
新增 messages/append 与 last-ai-message 接口，圆桌广播和总控派活轮改走服务端轻量读写，避免 loopback 传输整份 state。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
董宏凯andCursor committed 2026-06-07 21:20:58 +08:00
1 parent 924c8f2fac
commit 1ee2e6ca0a
2 files changed
+191 -50
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
+58 -49
View File
@@ -469,22 +469,17 @@ async def _append_thread_message(
失败只打 warning，不阻断主流程——广播是 best-effort，席位真正干活仍靠
前端对该席位 thread 发起的 ``/run/stream``。
走**轻量追加**接口(``POST /messages/append``):只传一条消息,服务端读 checkpoint+追加+写回,
省掉原先 GET /state(下载全量)+POST /state(上传全量)把几万字状态经 loopback 传两遍的开销。
"""
state_url = f"{base}/api/threads/{thread_id}/state"
get_resp = await client.get(state_url, headers=headers)
if get_resp.status_code != 200:
logger.warning("get state %s failed: %s %s", thread_id, get_resp.status_code, get_resp.text[:200])
return
state = get_resp.json()
values = state.setdefault("values", {})
messages = values.setdefault("messages", [])
if not isinstance(messages, list):
messages = []
values["messages"] = messages
messages.append({"type": "human", "content": content})
post_resp = await client.post(state_url, json=state, headers=headers)
if post_resp.status_code >= 400:
logger.warning("update state %s failed: %s %s", thread_id, post_resp.status_code, post_resp.text[:200])
resp = await client.post(
f"{base}/api/threads/{thread_id}/messages/append",
json={"content": content, "type": "human"},
headers=headers,
)
if resp.status_code >= 400:
logger.warning("append message %s failed: %s %s", thread_id, resp.status_code, resp.text[:200])
def _spawn_detached_broadcast(
@@ -1129,34 +1124,38 @@ async def _collect_delivery_status(
base: str,
headers: dict[str, str],
seat_threads: dict[str, str],
) -> dict[str, tuple[bool, str]]:
*,
max_chars: int = 0,
) -> dict[str, tuple[bool, str, int]]:
"""读各席位 thread 的真实状态,客观判定「是否已交付 + 交付摘要」。
返回 ``{seat_id: (delivered, full_text)}``。``full_text`` 是该席位**完整交付正文**
(调用方自行决定截多少:摘要块截短、综合轮全文)。判定依据:席位 thread 的对话历史里
**是否存在该席位自己产出的、非空的 AI 消息**(它的交付正文)。只有 broadcast 进来的
human 消息(派活通知 / 别的席位的交付摘要)不算它交付。
返回 ``{seat_id: (delivered, text, full_len)}``。``text`` 是该席位**最后一条 AI 交付**(按
``max_chars`` 截断:派活轮取 600 字摘要、综合轮 ``max_chars<=0`` 取全文);``full_len`` 是交付
全文长度(供调用方判断 text 是否被截断、要不要加省略号)。判定依据:席位 thread 是否存在非空
AI 交付,只有 broadcast 进来的 human 消息不算。
这是「确定性双保险」的数据源:总控弱模型会臆造「三方已交付」,这里由网关**客观核对**
每个席位 thread 的实际产出,再把结论喂回总控输入(见 ``_build_seat_status_block``),
总控以系统核对为准,不靠自己读历史猜。任一席位读失败按「未交付」保守处理,绝不阻断。
走**轻量读**接口(``GET /last-ai-message?max_chars=N``):服务端只回最后一条 AI 消息(可截断),
不把整份 state 序列化经 loopback 传回 —— 派活轮每席位只取 600 字,大幅压低原先 ~11s 的
``payload_built``。任一席位读失败按「未交付」保守处理,绝不阻断。
这是「确定性双保险」的数据源:总控弱模型会臆造「三方已交付」,这里由网关**客观核对**每个席位
thread 的实际产出,再把结论喂回总控输入,总控以系统核对为准,不靠自己读历史猜。
"""
async def _one(sid: str, tid: str) -> tuple[str, tuple[bool, str]]:
async def _one(sid: str, tid: str) -> tuple[str, tuple[bool, str, int]]:
try:
resp = await client.get(f"{base}/api/threads/{tid}/state", headers=headers)
resp = await client.get(
f"{base}/api/threads/{tid}/last-ai-message",
params={"max_chars": max_chars},
headers=headers,
)
if resp.status_code != 200:
return sid, (False, "")
messages = ((resp.json().get("values") or {}).get("messages")) or []
for msg in reversed(messages):
if not isinstance(msg, dict):
continue
if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
text = _flatten_content(msg.get("content", "")).strip()
if text:
return sid, (True, text)
return sid, (False, "")
return sid, (False, "", 0)
data = resp.json()
content = str(data.get("content") or "")
full_len = int(data.get("length") or len(content))
return sid, (bool(data.get("delivered")), content, full_len)
except Exception:
return sid, (False, "")
return sid, (False, "", 0)
results = await asyncio.gather(*(_one(sid, tid) for sid, tid in seat_threads.items()))
return dict(results)
@@ -1169,7 +1168,7 @@ _STATUS_EXCERPT_CHARS = 600
def _build_seat_status_block(
seats: list[tuple[str, dict[str, Any]]],
delivery: dict[str, tuple[bool, str]],
delivery: dict[str, tuple[bool, str, int]],
*,
undelivered_count: int = 0,
) -> str:
@@ -1186,11 +1185,11 @@ def _build_seat_status_block(
lines: list[str] = []
for sid, meta in seats:
name = str((meta or {}).get("name") or sid)
delivered, full_text = delivery.get(sid, (False, ""))
delivered, text, full_len = delivery.get(sid, (False, "", 0))
if delivered:
excerpt = full_text[:_STATUS_EXCERPT_CHARS]
ellipsis = "…" if len(full_text) > _STATUS_EXCERPT_CHARS else ""
tail = f" —— 交付摘要:{excerpt}{ellipsis}" if excerpt else ""
# text 已是服务端按 max_chars 截好的摘要;full_len > len(text) 说明被截断,加省略号。
ellipsis = "…" if full_len > len(text) else ""
tail = f" —— 交付摘要:{text}{ellipsis}" if text else ""
lines.append(f"- {sid}（{name}）：✅ 已交付{tail}")
else:
lines.append(f"- {sid}（{name}）：⬜ 未交付")
@@ -1217,7 +1216,7 @@ def _build_seat_status_block(
def _build_full_deliverables_block(
seats: list[tuple[str, dict[str, Any]]],
delivery: dict[str, tuple[bool, str]],
delivery: dict[str, tuple[bool, str, int]],
) -> str:
"""综合轮专用:把各已交付席位的**完整交付正文**拼成一块,供总控写最终方案。
@@ -1226,7 +1225,7 @@ def _build_full_deliverables_block(
"""
sections: list[str] = []
for sid, meta in seats:
delivered, full_text = delivery.get(sid, (False, ""))
delivered, full_text, _ = delivery.get(sid, (False, "", 0))
if not delivered or not full_text:
continue
name = str((meta or {}).get("name") or sid)
@@ -1281,22 +1280,32 @@ async def _leader_run(
if seat_meta:
seat_ids = [name for name in agent_threads if name != agent_name]
if seat_ids:
# 第 1 阶段:**轻量**读各席位最后一条 AI 交付,只取 ≤_STATUS_EXCERPT_CHARS 字 ——
# 派活轮只需 ✅/⬜ + 摘要,绝不下载几万字全文,把 payload_built 从 ~11s 压下来。
delivery: dict[str, tuple[bool, str]] = {}
try:
delivery = await _collect_delivery_status(
client, base, headers, {sid: agent_threads[sid] for sid in seat_ids}
client, base, headers, {sid: agent_threads[sid] for sid in seat_ids},
max_chars=_STATUS_EXCERPT_CHARS,
)
except Exception:
logger.warning("[leader-roster] 收集席位交付状态失败,退化为仅席位清单", exc_info=True)
undelivered = [sid for sid in seat_ids if not delivery.get(sid, (False, ""))[0]]
seats_meta_list = [(sid, seat_meta.get(sid, {})) for sid in seat_ids]
seat_status_block = _build_seat_status_block(seats_meta_list, delivery, undelivered_count=len(undelivered))
# 只有「综合轮」**且所有席位都已交付」才注入「完整交付 + 输出最终方案」的综合块。
# 还有 ⬜ 未交付席位时不注入 —— 否则那个"综合输出最终方案"的指令会怂恿弱模型在席位
# 没集齐时就提前收口(实测:3/5 交付就臆称"三方已发言"草草收口)。此时只给摘要,让总控
# 倾向于继续把 ⬜ 席位派活集齐。
# 第 2 阶段(仅「综合轮 且 所有席位都已交付」):再**全文**取各已交付席位的交付,拼综合块。
# 还有 ⬜ 未交付席位时不注入 —— 否则"综合输出最终方案"的指令会怂恿弱模型在席位没集齐时
# 就提前收口(实测:3/5 交付就臆称"三方已发言"草草收口)。全文读只在这一轮才付出。
if synthesis_mode and not undelivered:
full_deliverables_block = _build_full_deliverables_block(seats_meta_list, delivery)
try:
full_delivery = await _collect_delivery_status(
client, base, headers,
{sid: agent_threads[sid] for sid in seat_ids if delivery.get(sid, (False, ""))[0]},
max_chars=0,
)
full_deliverables_block = _build_full_deliverables_block(seats_meta_list, full_delivery)
except Exception:
logger.warning("[leader-roster] 综合轮取全文交付失败,退化为仅摘要", exc_info=True)
logger.info(
"[leader-roster] %s 注入席位交付状态:%s synthesis_mode=%s undelivered=%d full_deliverables=%s",
agent_name,
offline-backend-20260512/backend/app/gateway/routers/threads.py
+133 -1
View File
@@ -20,7 +20,7 @@ import uuid
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from langgraph.checkpoint.base import empty_checkpoint
from langgraph.checkpoint.base import empty_checkpoint, uuid6
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
@@ -1009,6 +1009,138 @@ async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, re
)
# ---------------------------------------------------------------------------
# Lightweight message append / last-AI read
#
# 圆桌广播与「席位交付状态」核对原本都走 GET /state + POST /state,要把动辄几万字的整份
# thread 状态经 loopback 传一两遍 —— 实测让总控每轮 payload_built 飙到 ~11s、广播也很慢。
# 下面两个轻量接口把重活留在服务端,loopback 只传一条小消息 / 截断后的摘要:
# • POST /{id}/messages/append —— 只传 1 条消息,服务端读 checkpoint+追加+写回;
# • GET /{id}/last-ai-message —— 只回最后一条 AI 交付(可 max_chars 截断),核对交付状态用。
# 注意:Postgres checkpoint 的 aput/aget 本身仍是 O(state) 的(快照模型),这里省掉的是
# 「整份 state 的 JSON 序列化 + loopback 往返」,这是可避免的那部分开销。
# ---------------------------------------------------------------------------
class AppendMessageRequest(BaseModel):
"""Request body for the lightweight single-message append."""
content: str = Field(..., description="Message text to append")
type: str = Field(default="human", description="Message role: human/ai/system")
def _flatten_msg_content(content: Any) -> str:
"""把 message.content(str / 结构化 block 列表)拍平成纯文本。"""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict):
parts.append(str(item.get("text", "")))
return "".join(parts)
return str(content) if content else ""
def _extract_last_ai_text(messages: list[Any]) -> str:
"""返回最后一条 AI 消息的纯文本;没有则空串。兼容 message 对象与序列化 dict 两种存储形态。"""
for msg in reversed(messages or []):
if isinstance(msg, dict):
mtype = str(msg.get("type") or "").lower()
content = msg.get("content", "")
else:
mtype = str(getattr(msg, "type", "") or "").lower()
content = getattr(msg, "content", "")
if mtype in ("ai", "aimessage", "aimessagechunk"):
text = _flatten_msg_content(content)
if text.strip():
return text
return ""
async def _append_message_to_checkpoint(checkpointer: Any, thread_id: str, content: str, type_: str) -> int | None:
"""把一条消息追加进 thread 最新 checkpoint 的 ``messages`` 通道,返回新消息总数。
``None`` 表示线程不存在(无 checkpoint)。纯粹围绕 checkpointer 的 aget/aput,便于单测。
语义与 ``update_thread_state`` 一致:仅替换 ``messages`` 通道为追加后的列表,其它通道原样保留。
"""
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
ckpt_tuple = await checkpointer.aget_tuple(config)
if ckpt_tuple is None:
return None
checkpoint: dict[str, Any] = dict(getattr(ckpt_tuple, "checkpoint", {}) or {})
metadata: dict[str, Any] = dict(getattr(ckpt_tuple, "metadata", {}) or {})
channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {}))
messages = list(channel_values.get("messages") or [])
messages.append({"type": type_, "content": content})
channel_values["messages"] = messages
checkpoint["channel_values"] = channel_values
# 生成**新的** checkpoint 快照 id + ts —— 每次追加都落成一个新快照(uuid6 时间可排序,新 id
# 天然成为最新),而不是复用旧 id 原地覆盖(后端语义不一致,会让连续追加互相丢失)。
checkpoint["id"] = str(uuid6(clock_seq=-2))
checkpoint["ts"] = now_iso()
# **关键**:checkpointer 用「版本化 blob」存 channel —— 必须 bump ``messages`` 的版本并把它
# 放进 aput 的 new_versions,aput 才会真正写入该 channel 的新内容、aget 才能读回。否则(像原
# GET/POST /state 那样只改 channel_values、new_versions 传 {})对没有现成 messages 版本的线程
# 不生效。
versions = dict(checkpoint.get("channel_versions", {}))
new_ver = checkpointer.get_next_version(versions.get("messages"), None)
versions["messages"] = new_ver
checkpoint["channel_versions"] = versions
metadata["updated_at"] = now_iso()
await checkpointer.aput(config, checkpoint, metadata, {"messages": new_ver})
return len(messages)
@router.post("/{thread_id}/messages/append", response_model=None)
@require_permission("threads", "write", owner_check=True, require_existing=True)
async def append_thread_message(thread_id: str, body: AppendMessageRequest, request: Request) -> dict[str, Any]:
"""轻量追加一条消息到 thread checkpoint —— 调用方只传一条消息,不必 GET+POST 整份 state。
语义与 ``update_thread_state`` 的 messages 写入一致:直接追加到 ``messages`` 通道,下次图
运行时 ``add_messages`` 做规范化/按 id 去重。失败抛 5xx 由调用方按 best-effort 处理。
"""
checkpointer = get_checkpointer(request)
try:
count = await _append_message_to_checkpoint(checkpointer, thread_id, body.content, body.type)
except Exception:
logger.exception("Failed to append message on thread %s", sanitize_log_param(thread_id))
raise HTTPException(status_code=500, detail="Failed to append thread message")
if count is None:
raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
return {"ok": True, "message_count": count}
@router.get("/{thread_id}/last-ai-message", response_model=None)
@require_permission("threads", "read", owner_check=True)
async def get_thread_last_ai_message(thread_id: str, request: Request, max_chars: int = 0) -> dict[str, Any]:
"""轻量读取 thread 最后一条 AI 消息(可截断),供圆桌核对席位交付状态。
返回 ``{delivered, content, length}``:``delivered`` = 是否存在非空 AI 交付;``max_chars>0``
时 ``content`` 截断到该长度(派活轮取摘要),``<=0`` 取全文(综合轮)。``length`` 是全文长度。
线程不存在时返回 ``delivered=False``(而非 404),方便调用方把"还没建好/没交付"统一当未交付。
"""
checkpointer = get_checkpointer(request)
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
try:
ckpt_tuple = await checkpointer.aget_tuple(config)
except Exception:
logger.exception("Failed to read last-ai-message on thread %s", sanitize_log_param(thread_id))
raise HTTPException(status_code=500, detail="Failed to get thread state")
if ckpt_tuple is None:
return {"delivered": False, "content": "", "length": 0}
channel_values = (getattr(ckpt_tuple, "checkpoint", {}) or {}).get("channel_values", {}) or {}
messages = channel_values.get("messages") or []
text = _extract_last_ai_text(messages if isinstance(messages, list) else [])
full_len = len(text)
if max_chars and max_chars > 0:
text = text[:max_chars]
return {"delivered": full_len > 0, "content": text, "length": full_len}
@router.delete("/{thread_id}/messages/{message_id}")
@require_permission("threads", "write", owner_check=True, require_existing=True)
async def delete_thread_message(thread_id: str, message_id: str, request: Request) -> dict[str, Any]: