diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 98526746..6aab16ca 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -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, diff --git a/offline-backend-20260512/backend/app/gateway/routers/threads.py b/offline-backend-20260512/backend/app/gateway/routers/threads.py index f1089335..5af5b5f9 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/threads.py +++ b/offline-backend-20260512/backend/app/gateway/routers/threads.py @@ -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]: