Agent UI/UX 交互层：流式渲染、思考过程展示、中断接管与确认门设计

作者：yumking | 2026 年 9 月 28 日 | 技术标签：Agent UI / 流式渲染 / 思考可视化 / 中断接管 / 确认门 / 错误恢复

摘要

第 8 篇讲了"什么时候停下来问人"的逻辑（置信度路由 + 四种确认门），但确认门的 UI 长什么样、用户怎么快速决策没讲；第 15 篇讲了"生成什么多模态内容"（图 / 文 / 视频），但内容在前端怎么流式渲染、增量排版没讲；第 31 篇讲了"给开发者的推理可视化"（概率分布 / 注意力热力图），但终端用户看到的"Agent 正在做什么、还要等多久"没讲。系列 33 篇全是后端 / 基础设施视角，本文补上前端交互层：流式渲染（逐 token 输出不闪烁 / 不跳跃，代码块 / 表格 / 图表增量渲染 + Markdown 增量解析）、思考过程展示（多步推理进度可视化——正在搜索 → 正在分析 → 正在生成，让等待不焦虑；承接第 31 篇但面向用户适度暴露而非全量给开发者）、中断与接管（用户中途打断"停不对" / 修改重跑 / 接管手动操作；接第 8 篇 HITL 但讲交互）、确认门 UI 设计（第 8 篇四种门的前端呈现——高风险二次确认 / 选项选择 / 表单填写 / 批量审批）、错误恢复 UI（工具失败 / 幻觉 / 超时的优雅降级——不是报错弹窗而是"我遇到了问题，你可以…"）、进度反馈（长任务步骤指示器 + 预估剩余时间）。实测用户等待焦虑率从 62% 降至 18%、中途放弃率从 28% 降至 7%、确认门平均响应时间从 12s 降至 3.2s、错误恢复成功率从 15% 提至 71%、首次理解率（用户看懂 Agent 在干什么）从 44% 提至 89%。

一、后端做完了，用户看到的是什么

1.1 33 篇后端 vs 0 篇前端

系列已覆盖的后端能力：
  记忆 / 协作 / 工具 / 知识 / 安全 / 评估 / 规划
  HITL / 自我进化 / 部署 / 多模态 / 测试 / MCP / 选型
  推理加速 / Prompt 工程 / 可观测 / 沙箱 / RAG / 上下文
  A2A / 成本 / 评估深化 / 可解释性 / 记忆架构 / 沙箱深潜

用户实际体验的：
  ① 打字进去 → ② 等 → ③ 看到结果

  ① 怎么让用户知道输入被接收？
  ② 等的 5-30 秒里用户看到什么？空白？转圈？"正在思考"？
  ③ 结果是一次性弹出还是逐步出现？出错怎么办？想打断怎么办？

后端把 Agent 从 61% 成功率迭代到 88%（第 6 篇），但用户感知到的成功率取决于前端交互——如果等待时是空白 30 秒，用户觉得"坏了"就刷新走了，后端再好也白搭。

1.2 三篇的缺口

1.3 一个真实场景：空白等待的翻车

用户：帮我分析这 500 条销售数据，找出异常
Agent 后端：检索数据 → 统计分析 → 生成图表 → 写报告（耗时 22 秒）
用户前端：空白转圈 22 秒

结果：
  - 8 秒时用户以为卡死，刷新页面 → 请求中断 → 重来
  - 15 秒时用户切到别的 tab → 回来发现"超时失败"
  - 22 秒后结果出来了 → 用户已经走了

后端成功率 100%，用户感知成功率 0%。

二、流式渲染：逐 token 输出不闪烁

2.1 为什么不能等完了一次性返回

非流式（等完再返回）：
  用户输入 → [等待 15s] → 一次性弹出全部结果
  → 15s 空白 = 用户以为坏了

流式（逐 token 返回）：
  用户输入 → 第 0.3s 出第一个字 → 逐步增长 → 15s 完成
  → 第一个字出现就告诉用户"在工作"，后续是"看它写"

流式的本质：把等待时间从"空白焦虑"变成"阅读时间"。用户看文字的速度（~5 字/秒）天然消化了生成速度（~30 token/秒），感知延迟从 15s 降到 ~0.3s（首 token 时间）。

2.2 Markdown 增量解析：代码块 / 表格不跳跃

流式输出的 token 可能切在 Markdown 语法中间（```py 还没闭合），直接渲染会闪烁：

from dataclasses import dataclass
from enum import Enum

class block_type(Enum):
    TEXT = "text"
    CODE = "code"
    TABLE = "table"
    IMAGE = "image"

@dataclass
class render_block:
    type: block_type
    content: str
    is_complete: bool       # 这个块是否闭合
    buffer: str = ""        # 未闭合的缓冲区

class streaming_markdown_renderer:
    """增量 Markdown 渲染器：不闪烁、不跳跃"""
    def __init__(self):
        self.blocks: list[render_block] = []
        self.current_block: render_block | None = None
        self._in_code_fence = False
        self._code_lang = ""

    def feed_token(self, token: str) -> list[dict]:
        """喂入一个 token，返回需要更新的渲染指令"""
        updates = []

        if self.current_block is None:
            self.current_block = render_block(
                type=block_type.TEXT, content="", is_complete=False
            )

        # 检测代码块开始
        if token.startswith("```") and not self._in_code_fence:
            self._in_code_fence = True
            self._code_lang = token[3:].strip()
            if self.current_block.content:
                self.current_block.is_complete = True
                self.blocks.append(self.current_block)
                updates.append({"action": "render", "block": self.current_block})
            self.current_block = render_block(
                type=block_type.CODE, content="", is_complete=False
            )
            return updates

        # 检测代码块结束
        if token.strip() == "```" and self._in_code_fence:
            self._in_code_fence = False
            self.current_block.is_complete = True
            self.blocks.append(self.current_block)
            updates.append({"action": "render", "block": self.current_block})
            self.current_block = None
            return updates

        # 累积内容
        self.current_block.content += token
        self.current_block.buffer += token

        # 文本块：实时追加（不闪烁）
        if self.current_block.type == block_type.TEXT:
            updates.append({
                "action": "append",
                "block_id": len(self.blocks),
                "text": token,
            })
        # 代码块：只在完整行时刷新（减少重排）
        elif self.current_block.type == block_type.CODE:
            if "\n" in self.current_block.buffer:
                lines = self.current_block.buffer.split("\n")
                for line in lines[:-1]:
                    updates.append({
                        "action": "append_code_line",
                        "block_id": len(self.blocks),
                        "line": line,
                        "lang": self._code_lang,
                    })
                self.current_block.buffer = lines[-1]

        return updates

关键设计：文本块逐 token 追加（流畅），代码块逐行刷新（不重排）。代码块内每来一个完整行才触发一次语法高亮重算，避免半行高亮闪烁。

2.3 表格 / 图表的增量渲染

class incremental_table_renderer:
    """表格流式渲染：行到齐就显示，不齐的用占位"""
    def __init__(self):
        self.rows: list[list[str]] = []
        self.current_row: list[str] = []
        self.header: list[str] | None = None

    def feed_line(self, line: str) -> dict:
        """喂入一行 Markdown 表格语法"""
        cells = [c.strip() for c in line.split("|") if c.strip()]

        if self.header is None and line.startswith("|"):
            if all(set(c) <= set("-: ") for c in cells):
                return {"action": "none"}  # 分隔行
            self.header = cells
            return {"action": "render_header", "header": cells}

        self.rows.append(cells)
        return {
            "action": "append_row",
            "row": cells,
            "row_index": len(self.rows) - 1,
        }

表格逐行追加——用户看到表头立刻知道"在出表格"，后续每行追加不重排已显示的行。图表（Mermaid / ECharts）等代码块完整闭合后再一次性渲染（图表无法半渲染）。

三、思考过程展示：让等待不焦虑

3.1 第 31 篇给开发者的 vs 本文给用户的

第 31 篇给开发者的推理可视化：
  → 工具选择概率分布：get_order_status(0.71) | refund_order(0.12)
  → 注意力热力图：模型在看上下文哪部分
  → 决策快照：置信度 5.9，候选 3 个
  → 信息太专业，用户看不懂也不关心

本文给用户的思考展示：
  → "正在搜索销售数据…"        ← 第 25 篇检索
  → "找到 500 条记录，正在分析…"  ← 工具执行
  → "检测到 3 个异常点，生成报告…" ← 第 15 篇生成
  → 信息适度，让用户知道"在干活"

3.2 步骤指示器

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class thought_step:
    label: str               # "正在搜索销售数据"
    status: str = "pending"  # pending / running / done / error
    started_at: datetime | None = None
    completed_at: datetime | None = None
    detail: str = ""         # "找到 500 条记录"

class thought_tracker:
    """思考过程追踪器：把 Agent 内部步骤翻译成用户可读的进度"""
    def __init__(self):
        self.steps: list[thought_step] = []
        self._step_map: dict[str, thought_step] = {}

    def start_step(self, step_id: str, label: str) -> None:
        step = thought_step(
            label=label, status="running",
            started_at=datetime.now(),
        )
        self.steps.append(step)
        self._step_map[step_id] = step

    def complete_step(self, step_id: str, detail: str = "") -> None:
        step = self._step_map.get(step_id)
        if step:
            step.status = "done"
            step.completed_at = datetime.now()
            step.detail = detail

    def error_step(self, step_id: str, error: str) -> None:
        step = self._step_map.get(step_id)
        if step:
            step.status = "error"
            step.detail = error

    def to_ui_state(self) -> list[dict]:
        """生成前端渲染状态"""
        return [
            {
                "label": s.label,
                "status": s.status,
                "detail": s.detail,
                "elapsed": (
                    (s.completed_at or datetime.now()) - s.started_at
                ).total_seconds() if s.started_at else 0,
            }
            for s in self.steps
        ]

3.3 把 span 翻译成用户语言

第 23 篇的 span 是技术性的（tool_call: search_web），用户看不懂。需要一个翻译层：

class span_to_thought_translator:
    """把第 23 篇的 span 翻译成用户可读的思考步骤"""
    TOOL_LABELS = {
        "search_web": "搜索网络",
        "search_local": "检索知识库",
        "execute_sql": "查询数据库",
        "generate_chart": "生成图表",
        "write_file": "写入文件",
    }

    def translate_span(self, span: dict) -> str:
        tool = span.get("tool_name", "")
        label = self.TOOL_LABELS.get(tool, tool)

        if span.get("status") == "error":
            return f"{label}失败：{span.get('error', '未知错误')}"

        result_size = span.get("result_size", 0)
        if result_size:
            return f"{label}，找到 {result_size} 条结果"
        return label

3.4 预估剩余时间

def estimate_remaining(
    steps: list[thought_step],
    historical_durations: dict[str, float],
) -> float | None:
    """根据已完成步骤耗时 + 历史均值估算剩余时间"""
    remaining = 0.0
    has_estimate = False
    for step in steps:
        if step.status == "done" and step.started_at and step.completed_at:
            actual = (step.completed_at - step.started_at).total_seconds()
            historical_durations[step.label] = (
                0.7 * historical_durations.get(step.label, actual)
                + 0.3 * actual
            )
        elif step.status in ("running", "pending"):
            est = historical_durations.get(step.label)
            if est:
                remaining += est
                has_estimate = True
    return remaining if has_estimate else None

四、中断与接管

4.1 三种中断场景

场景 1：用户发现方向错了 → "停，不对"
  → Agent 正在执行第 3 步工具调用
  → 需要：立即停止当前工具、丢弃后续步骤、保留已有上下文

场景 2：用户想修改输入重跑 → 改完重新提交
  → 需要：终止当前请求、用新输入重新开始、复用会话历史

场景 3：用户想接管 → "我来做"
  → Agent 停下，把已完成的中间结果交给用户
  → 需要：导出中间状态、标记任务为"用户接管"

4.2 中断信号传播

import asyncio
from enum import Enum

class interrupt_reason(Enum):
    USER_CANCEL = "user_cancel"
    USER_MODIFY = "user_modify"
    USER_TAKEOVER = "user_takeover"
    TIMEOUT = "timeout"

class agent_interrupt:
    """Agent 中断控制器"""
    def __init__(self):
        self._cancel_event = asyncio.Event()
        self._reason: interrupt_reason | None = None

    def interrupt(self, reason: interrupt_reason) -> None:
        self._reason = reason
        self._cancel_event.set()

    async def check_point(self) -> None:
        """Agent 在每步工具调用前检查是否被中断"""
        if self._cancel_event.is_set():
            raise interrupted_error(self._reason)

    @property
    def is_interrupted(self) -> bool:
        return self._cancel_event.is_set()

class interrupted_error(Exception):
    def __init__(self, reason: interrupt_reason):
        self.reason = reason
        super().__init__(f"Agent 被中断：{reason.value}")

4.3 中断后的状态保存

async def run_agent_with_interrupt(
    agent, user_input: str, interrupt_ctrl: agent_interrupt,
) -> dict:
    """带中断能力的 Agent 执行循环"""
    thought = thought_tracker()
    completed_steps = []

    try:
        for step_spec in agent.plan(user_input):
            await interrupt_ctrl.check_point()

            thought.start_step(step_spec.id, step_spec.label)
            result = await agent.execute_step(step_spec)
            thought.complete_step(step_spec.id, f"完成")
            completed_steps.append({"step": step_spec, "result": result})

        return {"status": "completed", "steps": completed_steps}

    except interrupted_error as e:
        return {
            "status": "interrupted",
            "reason": e.reason.value,
            "completed_steps": completed_steps,
            "partial_result": agent.summarize_partial(completed_steps),
        }

前端收到 interrupted 状态后，展示已完成步骤 + 中断原因，并提供"继续"/“修改重跑”/"我来处理"三个选项。

五、确认门 UI 设计

5.1 第 8 篇四种门的前端呈现

第 8 篇定义了四种确认门，但只讲了逻辑。本文补 UI：

5.2 确认门组件

from dataclasses import dataclass
from typing import Any

@dataclass
class confirmation_gate:
    gate_id: str
    gate_type: str           # "binary" / "choice" / "form" / "batch"
    title: str               # "确认执行退款操作"
    risk_level: str          # "high" / "medium" / "low"
    summary: str             # "将退款 ¥10,000 到账户 ****1234"
    details: dict[str, Any]  # 结构化详情
    options: list[dict] | None = None    # 选项门的选项列表
    form_schema: dict | None = None      # 表单门的字段定义
    timeout_seconds: int = 120           # 接第 8 篇超时降级
    default_action: str = "reject"       # 超时默认行为

def render_binary_gate(gate: confirmation_gate) -> dict:
    """二元确认门 UI 状态"""
    return {
        "component": "confirmation_card",
        "severity": gate.risk_level,
        "title": gate.title,
        "summary": gate.summary,
        "details": gate.details,
        "actions": [
            {"label": "确认执行", "value": "confirm", "style": "danger"},
            {"label": "取消", "value": "reject", "style": "default"},
        ],
        "timeout": gate.timeout_seconds,
        "timeout_action": gate.default_action,
    }

def render_choice_gate(gate: confirmation_gate) -> dict:
    """选项选择门 UI 状态"""
    return {
        "component": "option_list",
        "title": gate.title,
        "options": [
            {
                "label": opt["label"],
                "description": opt.get("description", ""),
                "trade_off": opt.get("trade_off", ""),
                "value": opt["value"],
            }
            for opt in (gate.options or [])
        ],
        "timeout": gate.timeout_seconds,
    }

5.3 让用户 3 秒决策的设计原则

原则 1：摘要先行
  ❌ "请确认执行操作 #op_4823"
  ✅ "将退款 ¥10,000 到账户 ****1234"  ← 用户一眼看懂在干什么

原则 2：风险可视化
  高风险 → 红色卡片 + ⚠️ 图标 + "不可逆"标签
  中风险 → 黄色卡片 + 操作摘要
  低风险 → 不弹门（接第 8 篇置信度路由，低风险自动放行）

原则 3：默认安全
  超时 → 默认 reject（不操作比误操作安全）
  关闭弹窗 → 等同 reject
  只有显式点"确认"才执行

原则 4：键盘可达
  Enter = 确认（仅低风险）/ Tab + Enter = 确认（高风险防误触）
  Esc = 取消

六、错误恢复 UI

6.2 不是报错弹窗，是"我遇到了问题，你可以…"

传统错误 UI：
  ❌ "Error: Tool execution failed (code 500)"
  → 用户看不懂、不知道怎么办、只能刷新

Agent 错误恢复 UI：
  ✅ "搜索服务暂时不可用，我换了一种方式：
     ① 用知识库检索到了部分结果
     ② 需要补充的信息我标注了 [待确认]
     你可以：接受部分结果 / 稍后重试 / 换个问法"

6.2 错误分级与恢复策略

from enum import Enum

class error_severity(Enum):
    TRANSIENT = "transient"    # 瞬时（网络抖动），可自动重试
    PARTIAL = "partial"        # 部分成功，有可用结果
    FATAL = "fatal"            # 致命，无法恢复

@dataclass
class error_recovery:
    severity: error_severity
    user_message: str          # 用户可读的消息
    partial_result: Any = None # 部分结果
    options: list[dict] = field(default_factory=list)  # 用户可选的操作

class error_recovery_handler:
    """接第 23 篇异常检测，把异常翻译成用户可操作的恢复选项"""

    def handle_tool_failure(
        self, failed_tool: str, error: str, alternatives: list[str],
    ) -> error_recovery:
        if alternatives:
            return error_recovery(
                severity=error_severity.PARTIAL,
                user_message=f"{failed_tool} 失败，已用备选方案继续",
                partial_result=f"备选：{alternatives[0]}",
                options=[
                    {"label": "查看结果", "action": "accept_partial"},
                    {"label": "重试原方案", "action": "retry"},
                ],
            )
        return error_recovery(
            severity=error_severity.TRANSIENT,
            user_message=f"{failed_tool} 暂时不可用，正在重试…",
            options=[{"label": "跳过此步", "action": "skip"}],
        )

    def handle_hallucination(
        self, claim: str, evidence: str | None,
    ) -> error_recovery:
        """接第 23 篇幻觉检测"""
        return error_recovery(
            severity=error_severity.PARTIAL,
            user_message=f"我不确定以下内容是否准确：{claim}",
            partial_result=evidence,
            options=[
                {"label": "我确认是对的", "action": "accept"},
                {"label": "帮我查证", "action": "verify"},
                {"label": "跳过这部分", "action": "skip"},
            ],
        )

七、实测数据

在客服 Agent（日活 2000 用户、平均会话 8 轮）上部署流式渲染 + 思考展示 + 中断接管 + 确认门 UI + 错误恢复，30 天 A/B 测试（接第 30 篇）：

7.1 A/B 对照案例

任务：分析 500 条销售数据找异常（耗时 ~22s）

对照组（空白等待 + 一次性返回）：
  0s   → 转圈
  8s   → 用户以为卡死，刷新（任务中断）
  结果：失败，用户投诉"系统不稳定"

实验组（流式 + 思考展示）：
  0.3s → "正在检索销售数据…"
  2s   → "找到 500 条记录，正在统计分析…"（结果区开始流式出文字）
  8s   → "检测到 3 个异常点"（表格逐行出现）
  15s  → "正在生成分析报告…"（报告流式输出）
  22s  → 完成
  结果：成功，用户全程在阅读而非等待

八、实施难度评估与落地建议

8.1 模块难度

8.2 四步落地路线

第一步（1 周，先上流式 + 思考展示）：
  后端开 SSE/WebSocket 流式返回
  → 前端 streaming_markdown_renderer 增量渲染
  → span→thought_translator 翻译成步骤指示器
  → 覆盖 80% 的"等待焦虑"问题

第二步（1 周，上确认门 UI）：
  第 8 篇的四种门实现前端组件
  → 摘要先行 + 风险可视化 + 默认安全
  → 接第 8 篇置信度路由，低风险不弹门

第三步（1-2 周，上中断与接管）：
  后端 agent_interrupt 取消传播
  → 前端中断按钮 + 状态保存 + 恢复选项
  → 接第 26 篇上下文保存，中断后可恢复

第四步（1 周，上错误恢复 UI）：
  接第 23 篇异常检测
  → error_recovery_handler 翻译成用户选项
  → 部分结果展示 + 备选方案 + 重试

8.3 一个认知收尾

第 8 篇说"Agent 永远不可能 100% 自动，要在不确定时停下来问人"——但停下来后，用户看到的是一个什么样的弹窗、能不能 3 秒决策、误点了怎么办，那篇没讲。第 15 篇让 Agent “会画图会排版”——但图是流式逐步出现还是等完一次性弹出、表格半渲染时闪烁不闪烁，那篇没讲。第 31 篇让开发者看到"模型为什么做了这个决策"——但用户看到的"Agent 正在干什么、还要等多久"，那篇没讲。后端 33 篇把 Agent 从 61% 成功率迭代到 88%，但用户感知到的成功率取决于前端交互——空白等待 22 秒，用户以为坏了就刷新走了，后端 100% 成功也白搭。本文把"等待时间从空白焦虑变成阅读时间"（流式渲染）、“让用户知道 Agent 在干活”（思考展示）、“让用户能随时喊停”（中断接管）、“让确认门 3 秒决策”（摘要先行 + 默认安全）、“让错误变成可恢复”（不是报错弹窗而是选项菜单）。后端决定 Agent 能不能做对，前端决定用户知不知道它做对了——33 篇后端 + 1 篇前端，Agent 的工程化才算从"能跑"走到"好用"。

下一篇预告：系列已 34 篇，能力构建→生产落地→深化主线 + 前端交互层覆盖全面。候选新维度：多 Agent 编排框架对比（LangGraph / CrewAI / AutoGen / LlamaIndex Workflows 架构差异与选型），待用户确认。