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大多数回答在这段时间内完成。用 SSEServer-Sent Events流式推送 token 到前端——用户看到文字一个一个出来感知延迟大幅降低。流式输出的心理效应用户看到第一个字出现时焦虑感立即下降——因为他知道系统正在生成答案只是需要时间写完。这比等 30 秒后突然看到一整段文字的体验好得多。流式输出的工程细节每个 SSE event 包含一个 delta增量文本而非完整文本。前端累加 delta 并实时渲染。如果连接中断前端用最后收到的 delta 位置做断点续传——重新建立 SSE 连接后从断点位置继续接收。Layer 3异步降级30s如果推理超过 30 秒还没完切换到后台模式告知用户这需要一些时间完成后通知你在后台继续推理完成结果推送到消息中心 / WebSocket用户不必一直等异步降级的关键不要丢掉已经生成的部分内容。如果 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), timeoutself._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), timeoutself._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 无信息。