Agent 超时与异步模式：不要让用户等一个不确定的回复

一、Agent 推理了 2 分钟，然后回了句"抱歉，超时了"

Agent 最差的用户体验不是回答错了，是等了很久然后什么都没得到。LLM 推理时间天生不确定——简单的问答 1 秒，复杂推理可能 30 秒，如果涉及多个 tool call 的串联，轻松破分钟。

更具体的场景：用户问"帮我分析这个季度的财务数据"，Agent 需要：调用数据库查询（5 秒）→ LLM 分析数据（10 秒）→ 调用图表生成工具（3 秒）→ LLM 生成最终报告（15 秒）。总计约 33 秒。如果数据库查询超时（如查询慢），整个过程可能变成 60 秒甚至更久。用户盯着"思考中…"看 60 秒是不可接受的。

实际数据：某客服 Agent 的 95% 响应在 5 秒内完成，但 5% 的复杂请求（涉及多步推理 + 工具调用）需要 30-60 秒。这 5% 的请求如果超时后返回空结果，用户会认为系统出了问题——因为他们不知道系统在后台做了很多工作。

让用户在聊天界面盯着"思考中…"看 60 秒是不可接受的。解决这个问题的工程手段是超时分片 + 异步推模式。

核心思路：把 Agent 的回答分成不同的时间预算层级。每种层级有不同的超时和不同的用户反馈。2 秒内必须给即时反馈（告诉用户收到了），30 秒内尝试给出完整答案（流式输出），超时后转为后台处理（完成后推送通知）。

sequenceDiagram
    participant U as 用户
    participant A as Agent 控制器
    participant LLM as LLM 服务
    
    U->>A: 提问
    
    rect rgb(200, 240, 200)
        Note over A: Tier 1: 即时反馈 (0-2s)
        A->>U: "正在分析你的问题..."
    end
    
    rect rgb(200, 220, 255)
        Note over A: Tier 2: 快速回答 (2-10s)
        A->>LLM: 生成快速回复
        LLM-->>A: 简短回答
        alt 可在 10s 内完成
            A->>U: 流式输出回答
        end
    end
    
    rect rgb(255, 240, 200)
        Note over A: Tier 3: 深度推理 (10-60s)
        A->>LLM: 多步推理 + tool calls
        A->>U: [进度] "正在搜索相关资料..."
        A->>U: [进度] "找到 3 条结果，正在分析..."
        A->>U: 流式输出完整回答
    end
    
    rect rgb(255, 200, 200)
        Note over A: Tier 4: 超时降级 (>60s)
        A->>U: "这个问题需要更多时间处理。
        我会在后台继续分析，完成后通知你。"
    end

二、超时与异步的三层设计

Layer 1：即时反馈（0-2s）

用户发出请求后 2 秒内必须有反馈。即使是"正在分析…"的占位也好——关键是让用户知道系统收到请求了。没有即时反馈的后果：用户认为请求丢失了，可能会重复提交——同一个问题被发 3 次，Agent 处理 3 次，浪费 3 倍资源。

即时反馈的内容应该尽量具体：不是笼统的"正在处理"，而是根据问题类型给出具体提示——"正在查询数据库…"（如果判定问题需要数据查询）、"正在搜索相关资料…"（如果判定问题需要知识检索）。这种具体性让用户感知系统在"做正确的事"，而非"卡住了"。

Layer 2：流式输出（2-30s）

大多数回答在这段时间内完成。用 SSE（Server-Sent Events）流式推送 token 到前端——用户看到文字一个一个出来，感知延迟大幅降低。流式输出的心理效应：用户看到第一个字出现时，焦虑感立即下降——因为他知道系统正在生成答案，只是需要时间写完。这比等 30 秒后突然看到一整段文字的体验好得多。

流式输出的工程细节：每个 SSE event 包含一个 delta（增量文本）而非完整文本。前端累加 delta 并实时渲染。如果连接中断，前端用最后收到的 delta 位置做断点续传——重新建立 SSE 连接后从断点位置继续接收。

Layer 3：异步降级（>30s）

如果推理超过 30 秒还没完，切换到后台模式：

1. 告知用户"这需要一些时间，完成后通知你"
2. 在后台继续推理
3. 完成结果推送到消息中心 / WebSocket
4. 用户不必一直等

异步降级的关键：不要丢掉已经生成的部分内容。如果 Agent 在前 30 秒已经生成了部分答案（比如 500 字的分析），超时降级时应该把这些内容先推送给用户——"已完成初步分析（见下方），完整报告将在完成后推送"。这比"全部等后台处理"好——用户至少有部分信息可以参考。

三、生产级超时控制实现

import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional, AsyncIterator


class ResponseTier(Enum):
    IMMEDIATE = "immediate"      # < 2s
    STREAMING = "streaming"      # < 30s  
    ASYNC_BACKGROUND = "async"   # > 30s


@dataclass
class TierConfig:
    immediate_timeout: float = 2.0
    streaming_timeout: float = 30.0
    async_timeout: float = 300.0  # 后台超时 5 分钟


class AgentTimeoutController:
    """Agent 超时控制器。
    
    核心设计：
    - 分级超时——不同复杂度的问题给不同的处理时间
    - 超时不是失败——超时后进入异步模式继续处理
    - 状态可查询——用户随时查询后台任务进展
    """

    def __init__(self, config: Optional[TierConfig] = None):
        self._config = config or TierConfig()
        self._background_tasks: dict[str, asyncio.Task] = {}
        self._task_results: dict[str, str] = {}

    async def process_with_timeout(
        self,
        task_id: str,
        user_prompt: str,
        llm_stream: AsyncIterator[str],
    ) -> AsyncIterator[dict]:
        """带超时控制的分层处理。
        
        Yields:
            {"type": "immediate"|"streaming"|"progress"|"async"|"complete", "content": str}
        """
        try:
            # Tier 1: 即时反馈
            yield {"type": "immediate", "content": "正在分析你的问题..."}
            await asyncio.sleep(0.1)

            # Tier 2: 流式输出（30s 超时）
            collected = ""
            try:
                async for token in asyncio.wait_for(
                    self._collect_stream(llm_stream),
                    timeout=self._config.streaming_timeout,
                ):
                    collected += token
                    yield {"type": "streaming", "content": token}

                # 完整收集，返回
                yield {"type": "complete", "content": collected}
                return

            except asyncio.TimeoutError:
                # 30s 到了但还没完——进入异步模式
                pass

            # Tier 3: 异步后台处理
            yield {
                "type": "async",
                "content": (
                    f"这个问题需要更多时间处理（已分析 {len(collected)} 字符）。"
                    f"我将在后台继续，完成后通知你。\n"
                    f"[任务 ID: {task_id}]"
                ),
            }

            # 启动后台任务继续处理
            background_task = asyncio.create_task(
                self._background_process(task_id, llm_stream, collected)
            )
            self._background_tasks[task_id] = background_task

        except Exception as e:
            yield {
                "type": "error",
                "content": f"处理请求时发生错误，请重试。",
            }

    async def get_task_status(self, task_id: str) -> dict:
        """查询后台任务状态。"""
        if task_id in self._task_results:
            return {
                "status": "completed",
                "result": self._task_results.pop(task_id),
            }
        if task_id in self._background_tasks:
            task = self._background_tasks[task_id]
            if task.done():
                if task.exception():
                    return {"status": "failed", "error": str(task.exception())}
                return {"status": "processing"}
            return {"status": "processing"}
        return {"status": "not_found"}

    async def _collect_stream(self, stream: AsyncIterator[str]) -> AsyncIterator[str]:
        """收集流式输出——透传但保留一份副本。"""
        async for chunk in stream:
            yield chunk

    async def _background_process(
        self, task_id: str, stream: AsyncIterator[str], prefix: str
    ) -> None:
        """后台继续收集剩余的流式输出。"""
        full_response = prefix
        try:
            async for chunk in asyncio.wait_for(
                self._collect_stream(stream),
                timeout=self._config.async_timeout,
            ):
                full_response += chunk
            self._task_results[task_id] = full_response
        except asyncio.TimeoutError:
            self._task_results[task_id] = (
                full_response + "\n\n[回答超时，部分内容未能完整生成]"
            )
        except Exception as e:
            self._task_results[task_id] = (
                full_response + f"\n\n[后台处理出错: {str(e)}]"
            )
        finally:
            self._background_tasks.pop(task_id, None)

    async def cancel_task(self, task_id: str) -> bool:
        """取消后台任务。"""
        task = self._background_tasks.pop(task_id, None)
        if task and not task.done():
            task.cancel()
            return True
        return False

四、异步模式的代价

用户体验断裂

同步等待"10 秒出一个完整答案"的体验 > 异步"先告诉我等着，5 分钟后来看结果"。异步只适用于复杂任务——简单任务异步化反而降低体验。判断标准：如果任务在 30 秒内能完成，不应该进入异步模式——强制异步会让用户觉得系统"不可靠"。

异步模式的心理成本：用户需要在 5 分钟后回来查看结果，但 5 分钟后他可能已经忘了这个问题。如果推送通知不够醒目（如只发了一条 Slack 消息），用户可能永远不看结果。解决方案：通知内容应该包含结果摘要而非只说"任务完成了"——让用户一眼就知道结果是否值得详细查看。

状态一致性

后台任务可能运行一半就失败了。需要有完整的错误处理、状态持久化、以及向用户传达失败时的优雅反馈。状态持久化的要求：后台任务的结果需要持久化到数据库而非内存——因为 Agent 实例可能在后台任务完成前就重启了。Redis 存储 + TTL 是合适的方案（参考断点续传的检查点设计）。

不能异步的场景

• 单步问答（不需要——30 秒内能完成）
• 实时对话（异步破坏对话连贯性——用户等 5 分钟后看到答案，但对话语境已经变了）
• 支付/交易（需要同步确认——异步化等于"不知道支付是否成功")

Trade-off 讨论

• 超时阈值 vs 用户容忍度：30 秒阈值适合大多数场景，但技术型用户可能容忍 60 秒，非技术用户可能只容忍 10 秒。阈值应该可配而非固定。
• 异步通知频率 vs 用户打扰度：后台任务每完成一步就通知一次（如"已搜索到 3 条结果"），给用户进度感但增加打扰。只在最终完成时通知一次，打扰少但用户不知道进度。推荐：关键里程碑通知（如"数据已查询完成"），而非每步都通知。
• 后台任务 TTL vs 存储成本：后台任务结果存储 24h 后自动过期。如果用户 24h 后才来看结果，已经丢失了。延长 TTL 到 7 天可以覆盖更多场景但增加存储。推荐：24h TTL + 结果摘要写入对话历史（永久保存）。

五、总结

Agent 的超时控制不是"设一个 timeout 值就完事"，而是按时间预算分层的回答策略。2 秒内给即时反馈、30 秒内流式输出、超时后转后台——用户不用干等，Agent 也有充分时间完成复杂推理。关键是让超时变成计划内的降级路径，而不是意外的失败终点。每层超时都有对应的用户反馈——即时反馈消除焦虑、流式输出降低感知延迟、异步通知保证结果可达。这个分层设计的核心收益：用户等待 30 秒的体验 > 用户等待 30 秒然后收到"超时"的体验——差距是"有信息 vs 无信息"。