"""中断展示与恢复共用一个载荷和标识,批准范围等于卡片展示范围。""" import hashlib import json from dataclasses import dataclass from typing import Any from obs import get_logger logger = get_logger(__name__) @dataclass(frozen=True) class ResumeDecision: interrupt_token: str = "" approved: bool | None = None status: str | None = None answers: list[str] | None = None error: str | None = None class ResumeError(ValueError): def __init__(self, code: str, message: str): self.code = code super().__init__(message) def pack(item): value = getattr(item, "value", item) payload = dict(value) if isinstance(value, dict) else {"message": str(value)} if getattr(item, "id", None): payload["_interrupt_id"] = item.id return payload def token_for(payload): return hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode()).hexdigest() def interrupt_event(writer, payload): payload = payload if isinstance(payload, dict) else {"message": str(payload)} token = token_for(payload) if payload.get("kind") == "ask_user": return writer.event("interrupt", interrupt_token=token, ask_user={"questions": payload["questions"]}) actions = payload.get("action_requests") or [] if actions: first = actions[0] return writer.interrupt_tool_approval( tool=first.get("name", ""), args=first.get("args"), reason=first.get("description", ""), side_effect="批准后执行卡片中的全部操作", actions=actions, pending_count=len(actions), interrupt_token=token, ) return writer.event("interrupt", message=payload.get("message", "智能体暂停,等待输入"), interrupt_token=token) def resume_value(request, state): items = [item for task in getattr(state, "tasks", ()) for item in getattr(task, "interrupts", ())] if not items: raise ResumeError(**{"code": "no_pending_interrupt", "message": "没有待确认请求"}) selected = next((item for item in items if token_for(pack(item)) == request.interrupt_token), None) if selected is None: raise ResumeError(**{"code": "stale_interrupt", "message": "确认请求已变化,请刷新后重试"}) payload = pack(selected) if payload.get("kind") == "ask_user": if request.status == "cancelled": value = {"cancelled": True} else: if not request.answers or len(request.answers) != len(payload["questions"]) or any(not a.strip() for a in request.answers): raise ResumeError(**{"code": "invalid_answers", "message": "请回答全部问题"}) value = {"answers": request.answers} else: if request.approved is None and request.status not in {"cancelled", "error"}: raise ResumeError(**{"code": "invalid_decision", "message": "请选择批准或拒绝"}) value = official_resume(request, len(payload.get("action_requests") or [])) interrupt_id = getattr(selected, "id", None) return {interrupt_id: value} if interrupt_id else value def official_resume(payload: ResumeDecision, pending_count: int) -> dict[str, Any]: """把领域层的 ResumeDecision 翻译成官方 HITL 的 ``{"decisions": [...]}``。 官方要求 decisions 与 action_requests **数量严格一一对应**,数量不符会 抛错,所以这里按 ``pending_count`` 复制同一决策。 翻译表: - ``approved=True`` → ``approve`` - ``approved=False`` → ``reject``(带拒绝理由) - ``status=cancelled`` → ``reject``(用户取消等价于不执行) - ``status=error`` → ``reject``(前端恢复出错,不执行更安全) - 有 ``answers`` → ``respond``(把人的回答作为工具结果) """ count = max(pending_count, 1) if count > 1 and payload.approved is True and not payload.interrupt_token: raise ResumeError(**{"code": "approval_refresh_required", "message": "请刷新并确认完整操作列表"}) if payload.approved is True: decision: dict[str, Any] = {"type": "approve"} elif payload.approved is False: decision = {"type": "reject", "message": "用户拒绝了该操作,请勿重试,改为说明情况"} elif payload.status == "cancelled": decision = {"type": "reject", "message": "用户取消了该操作"} elif payload.status == "error": decision = { "type": "reject", "message": f"前端恢复出错,该操作未执行:{payload.error or '未提供详情'}", } elif payload.answers: decision = {"type": "respond", "message": "\n".join(payload.answers)} else: decision = {"type": "reject", "message": "未收到明确答复,该操作未执行"} return {"decisions": [dict(decision) for _ in range(count)]} async def pending_interrupts(agent: Any, config: dict[str, Any]) -> list[Any]: """读取当前挂起的中断载荷列表。 供两处复用:``GET /chat/interrupt``(刷新页面后补渲染卡片)与 ``POST /chat/resume``(在构造 decisions 前先知道有几个待审批动作—— 官方 HITL 要求 decisions 与 action_requests **数量严格一一对应**, 数量对不上会直接抛错)。 """ get_state = getattr(agent, "aget_state", None) if get_state is None: return [] try: state = await get_state(config) except Exception as exc: logger.warning("读取图状态失败:%s: %s", type(exc).__name__, exc) raise found: list[Any] = [] for task in getattr(state, "tasks", None) or (): for item in getattr(task, "interrupts", None) or (): found.append(pack(item)) return found