解析 ‘The Psychology of Waiting’：在 Agent 执行长耗时任务时，如何利用图中节点的“进度心跳”缓解用户焦虑

漫长等待中的心理学：利用Agent节点进度心跳缓解用户焦虑

尊敬的各位开发者，各位产品经理，大家好。今天，我们将深入探讨一个在现代软件，尤其是基于AI Agent的系统中日益凸显的问题：用户在执行长耗时任务时的焦虑感。我们将从“等待的心理学”出发，剖析这种焦虑的根源，并提出一套系统性的解决方案——利用Agent内部节点的“进度心跳”，将不确定性转化为透明度，从而极大地提升用户体验。

一、等待的心理学：为何漫长的等待令人不安？

在我们的日常生活中，等待无处不在。从排队购物到等待快递，从系统加载到AI模型推理，等待是人机交互中不可避免的一部分。然而，并不是所有的等待都能被平等对待。心理学研究表明，等待的体验质量受到多种因素的影响，其中有几个核心痛点尤其值得我们关注：

1. 不确定性 (Uncertainty)： 当用户不知道任务是否正在进行、进展到何种程度、还需要多久才能完成时，不确定性会急剧增加焦虑。这种“黑箱”效应是等待中最令人沮丧的因素。
2. 失控感 (Loss of Control)： 用户在等待时往往感到自己处于被动地位，无法影响任务的进程，也无法获得反馈。这种失控感会加剧无助和焦虑。
3. 时间感知 (Perceived Time)： 无聊的等待感觉比有意义的等待更漫长。缺乏信息更新、缺乏进展指示的等待，会让用户觉得时间被拉长，效率低下。
4. 认知负荷 (Cognitive Load)： 用户可能会不断猜测任务状态，这本身就是一种认知负担，消耗注意力和耐心。
5. 目标阻碍 (Goal Blockage)： 长时间的等待阻碍了用户实现其目标，导致挫败感和不满意。

在传统的软件应用中，我们常常使用一个简单的加载动画或一个通用的进度条来表示任务正在进行。这在短时间任务中或许勉强足够，但在AI Agent执行的复杂、长耗时任务中，这种粗粒度的反馈机制是远远不够的。Agent任务往往涉及多步骤的推理、复杂的外部API调用、大规模的数据处理或模型生成，其耗时可能从几秒到几分钟，甚至更久。用户面对一个长时间旋转的加载图标，很快就会陷入上述的焦虑循环。

我们的目标，就是将这种因不确定性产生的焦虑，转化为一种有预期、有反馈、甚至带有参与感的体验。这不仅是为了技术上的实现，更是为了满足用户深层次的心理需求。

二、Agent长耗时任务的解构：从黑箱到透明的步骤

要有效缓解用户焦虑，我们首先需要理解Agent长耗时任务的内部结构。一个典型的AI Agent任务，尤其是那些涉及复杂决策或多轮交互的，通常不是一个原子操作，而是由一系列离散的、逻辑独立的“节点”或“步骤”组成的。这些节点可以是：

• 用户意图解析 (User Intent Parsing): 理解用户的查询或指令。
• 知识检索 (Knowledge Retrieval): 从数据库、文档或网络中查找相关信息。
• 工具调用 (Tool Invocation): 调用外部API、执行代码或与第三方服务交互。
• 信息综合与推理 (Information Synthesis & Reasoning): 结合检索到的信息进行逻辑推理，得出结论。
• 内容生成 (Content Generation): 利用大模型生成文本、图片或其他形式的响应。
• 安全检查 (Safety & Compliance Check): 对生成内容进行审核，确保符合规范。
• 用户反馈处理 (User Feedback Processing): 接收并处理用户对Agent响应的反馈。

这些节点构成了一个任务执行图（Task Execution Graph），可以是线性的序列，也可以是复杂的有向无环图（DAG）。从系统内部看，任务的进展是清晰可见的；但从用户界面看，它往往被抽象成一个单一的“加载中”状态。这种信息不对称正是焦虑的温床。

我们的核心思想，就是打破这个黑箱，将Agent内部每个节点的执行状态、进度和关键信息，实时地反馈给用户。这就是我们所说的“进度心跳”。

三、进度心跳：节点级粒度的实时反馈机制

“进度心跳”（Progress Heartbeat）是指Agent任务中每个独立节点在执行过程中周期性或事件驱动地发出的、包含其当前状态和进展信息的更新消息。它就像医生在手术过程中持续监测病人的生命体征，而不是等到手术结束才告知结果。

进度心跳的核心价值：

1. 提升透明度： 用户清楚地知道Agent正在“思考”什么、正在“做”什么。例如，不是“Agent正在处理”，而是“Agent正在解析您的查询”或“Agent正在从知识库检索信息”。
2. 提供掌控感： 即使用户无法直接干预，但知晓任务进展本身就能带来一种心理上的掌控感，减少无助。
3. 优化时间感知： 将一个漫长的等待分解成一系列小的、可识别的阶段，每个阶段都有明确的开始和结束。这使得整体等待时间在心理上显得更短，更容易接受。
4. 即时错误反馈： 如果某个节点执行失败，用户可以立即得到通知，而不是等到整个任务超时或失败，从而更快地进行问题排查或重新尝试。
5. 增强用户信任： 持续、透明的沟通能够建立用户对系统可靠性和智能性的信任。

进度心跳应包含的关键信息：

一个设计良好的进度心跳应该提供足够的信息，既不过于冗余，又能清晰地传达状态。以下是一个建议的数据结构：

字段名称	数据类型	描述	示例值
task_id	String	整体Agent任务的唯一标识符。用于将心跳与特定任务关联。	uuid-v4-string
node_id	String	发出心跳的Agent节点的唯一标识符。	query_parsing_node_123
node_name	String	人类可读的节点名称。用于在UI上显示给用户。	Query Parsing, Data Retrieval, Response Generation
status	String	节点的当前状态。常用的状态包括：PENDING, RUNNING, COMPLETED, FAILED, SKIPPED, WARNING。	RUNNING
message	String	针对当前状态的详细、人类可读的描述信息。	Starting semantic analysis of your query., Searching 1000 documents in vector database., Generated draft response, performing safety check.
progress_percentage	Integer	(可选) 如果节点内部有可量化的进度，0-100的百分比。	50
timestamp	Float	心跳发出的时间戳 (Unix timestamp)。	1678886400.123
severity	String	(可选) 信息的严重程度，用于UI的颜色编码或特殊处理。如 INFO, WARNING, ERROR。	INFO
metadata	JSON Object	(可选) 任何额外相关的上下文信息，例如API响应码、处理的数据量、检索到的文档数量等。这些信息可能对高级用户或调试有用。	{"api_name": "OpenAI GPT", "model_version": "gpt-4", "token_count": 500}, {"documents_found": 12, "source": "internal_wiki"}

四、设计可观测节点：架构考量

为了实现进度心跳，我们的Agent架构需要进行相应的调整，使其内部的节点具备“可观测性”。

4.1 节点抽象与接口

首先，我们需要定义一个通用的 AgentTaskNode 抽象基类或接口，强制所有节点实现一套标准的方法来报告进度。

# Python 示例：AgentNode 抽象基类
from abc import ABC, abstractmethod
import time
import uuid
import json
import asyncio

# 假设有一个WebSocketManager用于广播消息
class WebSocketManager:
    """A simplified WebSocket Manager for broadcasting messages."""
    def __init__(self):
        self.active_connections = {} # {task_id: [websocket_connection, ...]}

    async def connect(self, websocket, task_id):
        if task_id not in self.active_connections:
            self.active_connections[task_id] = []
        self.active_connections[task_id].append(websocket)
        print(f"[WebSocket] Client connected to task {task_id}")

    async def disconnect(self, websocket, task_id):
        if task_id in self.active_connections:
            self.active_connections[task_id].remove(websocket)
            if not self.active_connections[task_id]:
                del self.active_connections[task_id]
        print(f"[WebSocket] Client disconnected from task {task_id}")

    async def send_to_task_subscribers(self, task_id: str, message: str):
        if task_id in self.active_connections:
            for connection in self.active_connections[task_id]:
                try:
                    await connection.send_text(message)
                except Exception as e:
                    print(f"[WebSocket Error] Failed to send to connection for task {task_id}: {e}")

# 进度管理器，负责接收节点心跳并广播
class ProgressManager:
    """Manages and broadcasts progress updates for a task."""
    def __init__(self, websocket_manager: WebSocketManager):
        self.websocket_manager = websocket_manager
        # 存储每个任务下所有节点的最新状态，以便前端连接时可以获取历史状态
        self.task_node_states = {} # {task_id: {node_id: latest_node_update_dict}}

    async def broadcast_progress(self, task_id: str, node_update: dict):
        if task_id not in self.task_node_states:
            self.task_node_states[task_id] = {}

        # 更新该节点在任务中的最新状态
        self.task_node_states[task_id][node_update['node_id']] = node_update

        # 将更新发送给所有订阅该任务的客户端
        await self.websocket_manager.send_to_task_subscribers(
            task_id, 
            json.dumps({"type": "progress_update", "data": node_update})
        )
        print(f"[Progress] Broadcasted update for task {task_id}: {node_update['node_name']} -> {node_update['status']}")

    async def get_task_current_state(self, task_id: str) -> dict:
        """Retrieves the current aggregated state for a given task_id."""
        return self.task_node_states.get(task_id, {})

class AgentNode(ABC):
    def __init__(self, node_id: str, name: str, progress_manager: ProgressManager):
        self.node_id = node_id
        self.name = name
        self.progress_manager = progress_manager

    async def _emit_progress(self, task_id: str, status: str, message: str, progress_percentage: int = None, severity: str = "INFO", metadata: dict = None):
        """Helper method for nodes to emit progress heartbeats."""
        update = {
            "task_id": task_id,
            "node_id": self.node_id,
            "node_name": self.name,
            "status": status,
            "message": message,
            "timestamp": time.time(),
            "severity": severity,
        }
        if progress_percentage is not None:
            update["progress_percentage"] = progress_percentage
        if metadata is not None:
            update["metadata"] = metadata
        await self.progress_manager.broadcast_progress(task_id, update)

    @abstractmethod
    async def execute(self, task_id: str, input_data: dict) -> dict:
        """Abstract method for node execution logic."""
        pass

在这个设计中：

• WebSocketManager 负责管理与前端的WebSocket连接。
• ProgressManager 是核心，它接收来自各个 AgentNode 的进度更新，并负责通过 WebSocketManager 将这些更新广播给所有订阅了特定 task_id 的前端客户端。它还会维护一个 task_node_states 字典，存储每个任务下所有节点的最新状态，这对于新连接的客户端获取任务的当前整体进度至关重要。
• AgentNode 是所有Agent任务节点的基类。它包含一个 _emit_progress 辅助方法，允许派生类轻松地发出进度心跳，而无需关心底层的通信细节。

4.2 通信渠道选择

将进度心跳从后端发送到前端，我们需要选择合适的实时通信技术：

1. WebSockets：

   • 优点： 全双工通信，服务器和客户端可以随时互相发送消息。低延迟，实时性极强。是实现双向交互和持续更新的理想选择。
   • 缺点： 相比HTTP连接，协议略复杂，需要服务器和客户端都实现WebSocket协议。
   • 适用场景： 大多数现代Agent系统，需要高实时性、多节点并发更新，且前端可能需要发送控制指令（如取消任务）的场景。

2. Server-Sent Events (SSE)：

   • 优点： 基于HTTP协议，单向通信（服务器到客户端）。比WebSockets简单，易于实现。会自动处理断线重连。
   • 缺点： 只能单向传输，客户端无法主动向服务器发送消息。
   • 适用场景： 只需要服务器向客户端推送事件流，客户端无需发送复杂指令的场景。对于纯粹的进度更新来说，SSE是一个非常好的选择。

3. 短轮询 (Short Polling)：

   • 优点： 最简单，基于传统HTTP请求，无需特殊协议。
   • 缺点： 客户端周期性发送请求，服务器响应。会产生大量不必要的请求，效率低，实时性差，服务器压力大。
   • 适用场景： 仅适用于对实时性要求不高，或任务更新频率极低的场景，通常不推荐用于长耗时Agent任务的实时进度更新。

推荐： 对于Agent长耗时任务，WebSockets 是最佳选择，它提供了强大的实时双向通信能力，能够完美支撑进度心跳和可能的交互需求（如取消任务）。SSE也是一个可行的次优方案，如果只需要单向进度推送。

五、后端实现：将进度心跳融入Agent工作流

现在我们来看如何在Agent的具体执行逻辑中集成进度心跳。

5.1 具体节点实现

每个Agent节点在执行其核心业务逻辑时，应在关键步骤调用 _emit_progress 方法发送心跳。

# Python 示例：具体 Agent 节点实现

class QueryParsingNode(AgentNode):
    def __init__(self, progress_manager: ProgressManager):
        super().__init__(str(uuid.uuid4()), "Query Parsing", progress_manager)

    async def execute(self, task_id: str, input_data: dict) -> dict:
        await self._emit_progress(task_id, "RUNNING", "Starting query parsing...", 0)
        await asyncio.sleep(1) # 模拟耗时操作：初步解析

        user_query = input_data.get('user_query', 'No query')
        parsed_query = f"Parsed: {user_query.strip().lower()}"

        await self._emit_progress(task_id, "RUNNING", "Syntactic and semantic analysis complete.", 50, metadata={"original_query": user_query})
        await asyncio.sleep(1.5) # 模拟耗时操作：深度分析

        await self._emit_progress(task_id, "COMPLETED", "Query successfully parsed.", 100, metadata={"parsed_query": parsed_query})
        return {"parsed_query": parsed_query}

class DataRetrievalNode(AgentNode):
    def __init__(self, progress_manager: ProgressManager):
        super().__init__(str(uuid.uuid4()), "Data Retrieval", progress_manager)

    async def execute(self, task_id: str, input_data: dict) -> dict:
        parsed_query = input_data.get('parsed_query', 'unknown')
        await self._emit_progress(task_id, "RUNNING", f"Initiating data search for: '{parsed_query}'", 0)
        await asyncio.sleep(2) # 模拟耗时操作：连接数据库/API

        # 模拟数据检索的迭代过程
        total_records = 5000
        retrieved_count = 0
        for i in range(1, 4):
            retrieved_count += 1000
            await self._emit_progress(task_id, "RUNNING", f"Scanning database ({retrieved_count}/{total_records} records processed)...", int(retrieved_count / total_records * 100))
            await asyncio.sleep(0.5)

        retrieved_data = f"Data related to '{parsed_query}' from internal KB."
        await self._emit_progress(task_id, "COMPLETED", "Relevant data retrieved.", 100, metadata={"data_count": 5, "sources": ["internal_kb", "web_search"]})
        return {"retrieved_data": retrieved_data}

class ResponseGenerationNode(AgentNode):
    def __init__(self, progress_manager: ProgressManager):
        super().__init__(str(uuid.uuid4()), "Response Generation", progress_manager)

    async def execute(self, task_id: str, input_data: dict) -> dict:
        retrieved_data = input_data.get('retrieved_data', '')
        await self._emit_progress(task_id, "RUNNING", "Synthesizing response using retrieved data...", 0)
        await asyncio.sleep(1.5) # 模拟耗时操作：准备上下文

        await self._emit_progress(task_id, "RUNNING", "Invoking large language model for generation...", 30)
        await asyncio.sleep(3) # 模拟耗时操作：LLM调用

        generated_response = f"Here is the detailed response based on the data: {retrieved_data}. Please let me know if you need further details."

        await self._emit_progress(task_id, "RUNNING", "Performing final review and safety checks...", 80)
        await asyncio.sleep(1) # 模拟耗时操作：安全检查

        await self._emit_progress(task_id, "COMPLETED", "Response ready and delivered.", 100)
        return {"final_response": generated_response}

5.2 Agent任务编排与心跳集成

Agent的编排器（Orchestrator）负责定义任务的工作流，并按顺序或并行执行各个节点。编排器也应该在任务的整体层面发出心跳，并处理任何节点级别的错误。

# Python 示例：Agent Orchestrator (使用 FastAPI)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import asyncio
import uuid
import time

app = FastAPI()

# 初始化 WebSocketManager 和 ProgressManager
websocket_manager = WebSocketManager()
progress_manager = ProgressManager(websocket_manager)

# 定义构成Agent工作流的节点实例
query_parsing_node = QueryParsingNode(progress_manager)
data_retrieval_node = DataRetrievalNode(progress_manager)
response_generation_node = ResponseGenerationNode(progress_manager)

async def run_agent_workflow(task_id: str, user_query: str):
    """Executes the full agent workflow and emits overall task status."""
    try:
        # 整体任务开始心跳
        await progress_manager.broadcast_progress(
            task_id,
            {"task_id": task_id, "node_id": "overall_task", "node_name": "Overall Task", "status": "RUNNING", "message": "Agent workflow initiated.", "timestamp": time.time(), "progress_percentage": 0}
        )

        # 步骤 1: 查询解析
        parsed_result = await query_parsing_node.execute(task_id, {"user_query": user_query})

        # 步骤 2: 数据检索
        retrieval_result = await data_retrieval_node.execute(task_id, parsed_result)

        # 步骤 3: 响应生成
        final_result = await response_generation_node.execute(task_id, retrieval_result)

        # 整体任务完成心跳
        await progress_manager.broadcast_progress(
            task_id,
            {"task_id": task_id, "node_id": "overall_task", "node_name": "Overall Task", "status": "COMPLETED", "message": "Agent workflow finished successfully.", "timestamp": time.time(), "progress_percentage": 100, "metadata": {"final_output": final_result["final_response"]}}
        )
        return final_result
    except Exception as e:
        # 整体任务失败心跳
        await progress_manager.broadcast_progress(
            task_id,
            {"task_id": task_id, "node_id": "overall_task", "node_name": "Overall Task", "status": "FAILED", "message": f"Agent workflow failed: {e}", "timestamp": time.time(), "severity": "ERROR", "progress_percentage": 100}
        )
        print(f"[ERROR] Agent task {task_id} failed: {e}")
        raise

@app.websocket("/ws/task/{task_id}")
async def websocket_endpoint(websocket: WebSocket, task_id: str):
    await websocket_manager.connect(websocket, task_id)
    try:
        # 新连接的客户端，发送该任务当前所有节点的最新状态
        current_state = await progress_manager.get_task_current_state(task_id)
        for node_id, node_update in current_state.items():
            await websocket.send_text(json.dumps({"type": "progress_update", "data": node_update}))

        # 保持连接，等待新的心跳或客户端消息
        while True:
            # 可以接收客户端消息，例如取消任务指令
            data = await websocket.receive_text()
            print(f"Received from client {task_id}: {data}")
            # ... process client messages ...
    except WebSocketDisconnect:
        await websocket_manager.disconnect(websocket, task_id)
    except Exception as e:
        print(f"WebSocket error for task {task_id}: {e}")
        await websocket_manager.disconnect(websocket, task_id)

@app.post("/start_agent_task/")
async def start_agent_task(user_query: str):
    """API endpoint to start a new agent task."""
    task_id = str(uuid.uuid4())
    # 在后台异步运行Agent工作流，不阻塞HTTP响应
    asyncio.create_task(run_agent_workflow(task_id, user_query)) 
    return {"message": "Agent task started", "task_id": task_id}

@app.get("/")
async def get_root():
    """Simple HTML page to demonstrate frontend."""
    return HTMLResponse("""
        <!DOCTYPE html>
        <html>
        <head>
            <title>Agent Progress Monitor</title>
            <link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
            <style>
                body { padding: 20px; }
                .card-body { padding: 10px; }
                .progress-container { margin-top: 20px; }
                .overall-progress-bar { height: 25px; line-height: 25px; text-align: center; color: white; margin-bottom: 10px; }
            </style>
        </head>
        <body>
            <div class="container">
                <h1 class="mb-4">Agent Task Progress Monitor</h1>
                <div class="input-group mb-3">
                    <input type="text" id="user-query-input" class="form-control" placeholder="Enter your query..." value="What is the capital of France?">
                    <button class="btn btn-primary" onclick="startAgentTask()">Start Agent Task</button>
                </div>

                <div class="overall-progress-container progress-container">
                    <h3>Overall Task Progress:</h3>
                    <div id="overall-progress-display" class="overall-progress-bar bg-secondary">Waiting to start...</div>
                    <div class="progress" style="height: 10px;">
                        <div id="overall-progress-bar-inner" class="progress-bar bg-info" role="progressbar" style="width: 0%" aria-valuenow="0" aria-valuemin="0" aria-valuemax="100"></div>
                    </div>
                </div>

                <div class="node-status-container progress-container mt-4">
                    <h3>Node Progress Details:</h3>
                    <div id="status-updates">
                        <p class="text-muted">No updates yet. Start a task to see progress.</p>
                    </div>
                </div>
            </div>
            <script src="/static/app.js"></script> <!-- Link to our frontend JS -->
        </body>
        </html>
    """)

# 简单的静态文件服务，用于提供 app.js
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")

# 为了运行这个 FastAPI 应用，你需要创建 'static' 目录并在其中放入 'app.js'
# 命令行: uvicorn your_module_name:app --reload

在 run_agent_workflow 中，我们不仅在各个节点内部发送心跳，还在整个工作流的开始和结束时，以及遇到错误时，发送一个特殊的 overall_task 节点的心跳。这为用户提供了任务的宏观视角。

六、前端呈现：将心跳转化为可见的安心

接收到后端发送的进度心跳后，前端的工作是将这些原始数据转化为用户易于理解和感知的可视化信息。

// JavaScript 示例：Frontend (static/app.js)
const userQueryInput = document.getElementById('user-query-input');
const overallProgressDisplay = document.getElementById('overall-progress-display');
const overallProgressBarInner = document.getElementById('overall-progress-bar-inner');
const statusContainer = document.getElementById('status-updates');

let currentTaskId = null;
let ws = null;
const nodeStatusMap = {}; // Stores latest status for each node

function connectWebSocket(taskId) {
    if (ws && ws.readyState === WebSocket.OPEN) {
        ws.close(); // Close existing connection if any
    }
    ws = new WebSocket(`ws://localhost:8000/ws/task/${taskId}`);

    ws.onopen = (event) => {
        console.log("WebSocket connected for task:", taskId);
        overallProgressDisplay.textContent = "Connected. Waiting for updates...";
        // On reconnection, request full current state
        ws.send(JSON.stringify({ action: "request_full_state" })); 
    };

    ws.onmessage = (event) => {
        const update = JSON.parse(event.data);
        if (update.type === "progress_update") {
            const data = update.data;
            console.log("Received progress update:", data);

            // Update individual node status
            nodeStatusMap[data.node_id] = data;
            renderNodeStatuses();

            // Update overall progress if applicable
            if (data.node_id === "overall_task") {
                if (data.progress_percentage !== undefined) {
                    overallProgressBarInner.style.width = `${data.progress_percentage}%`;
                    overallProgressBarInner.setAttribute('aria-valuenow', data.progress_percentage);
                }
                overallProgressDisplay.textContent = `${data.node_name}: ${data.message} (${data.status})`;
                overallProgressDisplay.className = `overall-progress-bar bg-${getStatusColorClass(data.status)}`;
            }
        }
    };

    ws.onclose = (event) => {
        console.log("WebSocket disconnected:", event.code, event.reason);
        if (event.wasClean) {
            overallProgressDisplay.textContent = `Connection closed. Code: ${event.code}`;
        } else {
            overallProgressDisplay.textContent = `Connection died. Code: ${event.code}. Reason: ${event.reason}`;
            overallProgressDisplay.className = `overall-progress-bar bg-danger`;
        }
        ws = null; // Clear WebSocket instance
    };

    ws.onerror = (error) => {
        console.error("WebSocket error:", error);
        overallProgressDisplay.textContent = `WebSocket Error: ${error.message}`;
        overallProgressDisplay.className = `overall-progress-bar bg-danger`;
    };
}

function renderNodeStatuses() {
    statusContainer.innerHTML = ''; // Clear previous statuses
    const sortedNodes = Object.values(nodeStatusMap).sort((a, b) => a.timestamp - b.timestamp);

    sortedNodes.forEach(node => {
        if (node.node_id === "overall_task") return; // Skip overall task as it's handled separately

        const statusColorClass = getStatusColorClass(node.status);
        const progressBarStyle = `width: ${node.progress_percentage !== undefined ? node.progress_percentage : 0}%`;

        statusContainer.innerHTML += `
            <div class="card mb-2">
                <div class="card-body">
                    <h5 class="card-title text-${statusColorClass}">${node.node_name}</h5>
                    <p class="card-text">Status: <strong>${node.status}</strong></p>
                    <p class="card-text">${node.message}</p>
                    ${node.progress_percentage !== undefined ? `
                        <div class="progress mt-1" style="height: 5px;">
                            <div class="progress-bar bg-${statusColorClass}" role="progressbar" style="${progressBarStyle}" aria-valuenow="${node.progress_percentage}" aria-valuemin="0" aria-valuemax="100"></div>
                        </div>` : ''}
                    <small class="text-muted">${new Date(node.timestamp * 1000).toLocaleTimeString()}</small>
                    ${node.metadata ? `<pre class="text-muted mt-2 small">Metadata: ${JSON.stringify(node.metadata, null, 2)}</pre>` : ''}
                </div>
            </div>
        `;
    });
}

function getStatusColorClass(status) {
    switch (status) {
        case 'COMPLETED': return 'success';
        case 'RUNNING': return 'info';
        case 'FAILED': return 'danger';
        case 'WARNING': return 'warning';
        case 'PENDING': return 'muted';
        default: return 'secondary';
    }
}

async function startAgentTask() {
    const userQuery = userQueryInput.value;
    if (!userQuery) {
        alert("Please enter a query.");
        return;
    }

    // Reset UI for new task
    nodeStatusMap = {};
    renderNodeStatuses();
    overallProgressDisplay.textContent = "Starting new task...";
    overallProgressDisplay.className = "overall-progress-bar bg-secondary";
    overallProgressBarInner.style.width = "0%";
    overallProgressBarInner.setAttribute('aria-valuenow', 0);

    try {
        const response = await fetch('/start_agent_task/', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ user_query: userQuery })
        });
        const data = await response.json();
        if (data.task_id) {
            currentTaskId = data.task_id;
            console.log("Task started with ID:", currentTaskId);
            connectWebSocket(currentTaskId); // Connect to WebSocket for the new task
        } else {
            console.error("Failed to start task:", data);
            overallProgressDisplay.textContent = `Error starting task: ${data.message || 'Unknown error'}`;
            overallProgressDisplay.className = `overall-progress-bar bg-danger`;
        }
    } catch (error) {
        console.error("Network error starting task:", error);
        overallProgressDisplay.textContent = `Network error: ${error.message}`;
        overallProgressDisplay.className = `overall-progress-bar bg-danger`;
    }
}

前端可视化策略：

1. 整体进度条与状态显示：
   • 一个顶部的、粗粒度的进度条和状态信息，显示整个Agent任务的宏观进展（由 overall_task 节点的心跳驱动）。
   • 例如：“Agent workflow initiated…” -> “Agent workflow finished successfully.”
2. 节点列表与详细状态：
   • 下方是一个按时间顺序排列的节点状态列表。每个卡片代表一个Agent节点。
   • 每个卡片显示节点名称、当前状态、详细消息、以及可能的百分比进度条。
   • 颜色编码： RUNNING (蓝色/信息色), COMPLETED (绿色/成功色), FAILED (红色/危险色), WARNING (黄色/警告色), PENDING (灰色/静默色)。
   • 时间戳： 显示每条心跳的接收时间，提供时间轴感。
   • 元数据： 如果有关键的元数据，可以考虑以折叠或可展开的方式展示，供高级用户查看。
3. 动态更新：
   • 利用JavaScript监听WebSocket消息，实时更新UI。当新心跳到达时，更新或添加相应的节点卡片。
   • 确保UI更新是高效的，避免重绘整个列表，只更新变化的部分。
4. “下一步”提示 (Optional)：
   • 如果任务流是预定义的，可以在某个节点完成后，提示用户“下一个步骤是：[下一个节点名称]”，进一步减少不确定性。

七、高级考量与最佳实践

7.1 错误处理与容错

进度心跳不仅用于成功路径，更是错误报告的关键。当某个节点失败时，它应该发出 FAILED 状态的心跳，并附带详细的错误消息和堆栈信息（如果合适，可放入 metadata 中）。前端可以捕获这些错误，以红色突出显示，并提供重试或联系支持的选项。

7.2 任务取消与用户交互

既然是双向通信，用户也应该有机会向Agent发送指令。例如，一个“取消任务”按钮。当用户点击时，前端通过WebSocket发送取消指令给后端，后端接收后，Agent编排器可以尝试优雅地停止正在运行的节点。

7.3 负载均衡与可伸缩性

在分布式Agent系统中，多个Agent实例可能并行处理任务。确保所有实例的进度心跳都能汇聚到同一个 ProgressManager 并正确广播。这可能需要一个消息队列（如Kafka, RabbitMQ）作为中间件，将各个Agent实例的进度事件发送到 ProgressManager 服务。

7.4 安全性

WebSocket连接应进行身份验证和授权。确保只有授权的用户才能订阅其任务的进度更新，防止信息泄露或恶意订阅。可以利用JWT等令牌进行WebSocket连接的认证。

7.5 性能优化

• 心跳频率： 避免过于频繁的心跳，尤其是在短时间节点中，以减少网络和服务器负载。根据节点的预期耗时和信息重要性调整频率。
• 数据量： 心跳消息应尽可能精简，只包含必要信息。大的 metadata 可以按需加载或只包含关键摘要。
• 前端渲染： 优化前端渲染逻辑，避免在每次更新时都重绘整个列表，使用虚拟列表或只更新受影响的DOM元素。

7.6 估计剩余时间 (ETA)

虽然难以精确，但对于某些可预测的节点，可以根据历史数据或任务规模估算剩余时间（Estimated Time of Arrival, ETA）。例如，如果一个节点需要处理10000条记录，当前已处理5000条，可以估算大约还需一半时间。将ETA信息包含在心跳中，能够进一步缓解用户的焦虑。

八、成功的衡量：超越技术指标的用户体验

我们投入精力实现进度心跳，最终是为了提升用户体验。成功的衡量不应仅仅局限于系统是否正常运行，更应关注用户的主观感受和行为数据：

• 用户满意度： 通过问卷调查、A/B测试等方式，比较有无进度心跳功能的用户满意度。
• 任务放弃率： 观察长耗时任务的用户放弃率是否降低。
• 等待时长感知： 虽然任务实际时长不变，但用户是否感觉等待时间变短。
• 支持请求量： 关于“我的任务怎么了？”或“为什么这么慢？”的支持请求是否减少。
• 用户留存与活跃度： 更好的体验会带来更高的用户留存和更积极的互动。

这些指标将帮助我们量化进度心跳带来的业务价值，并指导未来的优化方向。

九、透明交互的艺术

今天，我们探讨了如何通过Agent内部节点的“进度心跳”，将等待的心理学原理应用于实际系统设计。这不是简单的技术实现，而是一门关于透明度、信任和用户赋能的艺术。通过将一个复杂的黑箱任务解构为一系列可观测、可理解的步骤，我们不仅提供了技术上的实时反馈，更重要的是，我们建立了一个与用户沟通的桥梁。

这种粒度化、实时的进度反馈，能够显著降低用户在长耗时任务中的不确定性和失控感，优化他们对时间流逝的感知，最终将焦虑转化为期待，将等待转化为一种更积极、更具参与感的体验。在Agent和AI系统日益普及的今天，精细化、人性化的用户体验将是产品成功的关键差异点。