1. 从“单打独斗”到“团队协作”:AI Agent的范式转变

最近在折腾AI应用开发的朋友,可能都绕不开一个词: Agent 。从年初的AutoGPT爆火,到后来各种“自主智能体”框架层出不穷,大家似乎都在追求一个目标——让AI自己动起来,完成一个复杂的、多步骤的任务。我一开始也是这个思路的拥趸,花了不少时间研究如何让一个Agent“全自动”地规划、执行、反思,试图打造一个不知疲倦的“数字员工”。

但踩过几个坑之后,我发现事情没那么简单。让AI完全自主,就像让一个刚入职的新人独立负责一个跨部门大项目,结果往往是灾难性的:它可能会陷入死循环,在一个无关紧要的细节上钻牛角尖;可能会因为工具调用失败而“卡死”,不知道如何回退;更常见的是,它执行到一半,你发现方向偏了,却很难中途介入纠正,只能眼睁睁看着它跑偏,或者干脆重启整个流程。这种“黑盒式”的自动化,在演示时很酷,但在真实、严肃的业务场景中,可靠性和可控性都大打折扣。

于是,我的思路开始转变。与其追求不切实际的“完全自主”,不如回归一个更务实、更强大的模式: 人机协作 。我们搭建的AI Agent,不应该是一个取代人类的“终结者”,而应该是一个能力超强的“副驾驶”或“智能助手”。它的核心价值不在于“替代”,而在于“增强”——它能以惊人的速度处理信息、调用工具、生成草稿,而人类则负责提供高层意图、进行关键决策、纠正方向偏差。这种协作模式,才是当前技术条件下最具生产力和实用价值的路径。

而要实现有效的人机协作,一个无法回避的核心技术挑战就是 中断与恢复 。想象一下,你和助手正在合作写一份报告,你临时接到一个电话,离开了半小时。回来后,你希望助手能清晰地告诉你:“您离开前,我们刚完成了市场分析部分的第一小节,正在搜集竞争对手的数据,但遇到了XX网站访问限制的问题。这是目前的草稿,您看接下来是优先解决访问问题,还是跳过这部分先写结论?”——这才是理想的协作体验。我们的AI Agent也必须具备类似的能力:在长时间、多步骤的任务中,能够优雅地暂停(中断),完整地保存当前状态(上下文、历史、工具调用结果、临时变量),并在需要时准确地从断点继续(恢复)。这不仅是技术实现,更是设计理念的体现。

所以,今天我想分享的,就是如何从零开始,搭建一个以 人机协作 为设计核心、具备强大 中断恢复 能力的AI Agent系统。我们会从最根本的架构设计聊起,一步步拆解状态管理、对话编排、持久化存储等关键环节,并分享我在实现过程中趟过的那些坑和总结出的实用技巧。你会发现,一个“可中断、可恢复”的Agent,其代码结构和思维模型,与一个“一镜到底”的Agent有着本质的不同。

2. 架构基石:设计一个支持协作与状态管理的Agent核心

要支持人机协作和中断恢复,我们首先得抛弃那种“一次性流水线”的Agent设计。传统的简单Agent可能就是一个循环: 接收用户输入 -> LLM思考 -> 执行动作 -> 更新记忆 -> 循环 。这种设计里,状态是临时的,任务是不可中断的。

我们需要的是一个更具弹性的架构。我把这个核心架构称为“ 状态驱动的协作式Agent ”。它的核心组件包括:

  1. Agent 大脑 (Brain) :通常由大语言模型(LLM)担任,负责理解意图、规划步骤、决定何时调用工具或请求人工输入。
  2. 对话与任务管理器 (Orchestrator) :这是整个系统的调度中心。它管理一个或多个任务(Session)的生命周期,维护任务状态机(如:运行中、等待用户输入、已暂停、已完成、已失败),并负责协调“大脑”与下方各个模块的交互。
  3. 状态存储器 (State Store) :这是实现中断恢复的关键。它需要持久化保存任务的所有上下文信息。这不仅仅是聊天历史,而是一个结构化的“任务快照”。
  4. 工具集 (Toolkit) :Agent可以调用的外部能力,如搜索网络、查询数据库、执行代码、调用API等。
  5. 人机交互接口 (Human-in-the-loop Interface) :提供明确的机制,让Agent可以主动向用户提问、确认,也让用户可以随时中断、查看状态、修改指令或提供额外信息。

其中, 状态存储器 的设计是重中之重。我们需要保存哪些状态,才能让Agent在几天后恢复时,还能无缝衔接?至少需要以下几类:

  • 会话元数据 (Session Metadata) :会话ID、创建时间、最后活跃时间、当前状态(运行/暂停)、所属用户等。
  • 任务目标与参数 (Goal & Parameters) :用户最初的任务描述,以及任何解析出来的结构化参数。这是恢复后不会迷失方向的“北极星”。
  • 完整的对话历史 (Full Dialogue History) :包括用户消息、Agent的思考过程、工具调用请求、工具执行结果、以及系统提示词。注意,这里要保存的是原始记录,而不是经过总结的“记忆”。
  • 工具调用上下文 (Tool Call Context) :当前或最近一次工具调用的详细信息,包括函数名、参数、执行结果、成功或错误状态。
  • 工作内存/暂存器 (Working Memory) :任务执行过程中产生的中间结果、临时变量或结构化数据。例如,在撰写报告时,已经收集到的数据点列表;在编写代码时,已经定义好的函数接口。
  • 执行计划与进度 (Plan & Progress) :如果Agent进行了任务分解,那么当前的计划树、哪些子任务已完成、哪些正在进行、哪些尚未开始,这些信息都需要保存。

在技术选型上,对于原型或轻量级应用,你可以使用SQLite或本地JSON文件来存储状态。但对于需要高可靠性和并发访问的生产环境,我强烈推荐使用像 Redis (用于快速缓存和会话状态)配合 PostgreSQL MongoDB (用于持久化存储和复杂查询)的组合。Redis的键过期和数据结构(如Hash, List)非常适合管理活跃会话的实时状态,而关系型或文档型数据库则用于长期归档和复杂的状态查询。

一个常见的误区是只保存对话历史。当任务复杂、涉及多轮工具调用和中间状态时,仅凭对话历史很难精准恢复。比如,Agent调用了一个返回大量数据的搜索API,并在后续思考中引用了其中的第5条结果。如果只保存历史,恢复时你需要重新解析整个冗长的结果。而如果我们在工作内存中明确保存了“ search_results[4] = {title: ‘…’, url: ‘…’} ”,恢复就会高效且准确得多。

3. 实现中断:如何让Agent优雅地“按下暂停键”

有了支持状态保存的架构,接下来我们要设计中断机制。中断不是简单的 Ctrl+C 杀死进程,而是一个受控的状态转换过程。中断的触发可以由用户主动发起,也可以由Agent在特定条件下自动请求。

3.1 用户主动中断

这是最常见的场景。用户可能想中途修改需求、补充信息,或者只是暂时离开。我们需要在交互界面上提供一个清晰的“暂停”或“保存并退出”按钮。当这个信号发出时,Orchestrator(任务管理器)需要执行以下流程:

  1. 完成当前原子操作 :确保当前正在执行的最小单位任务完成。例如,如果正在调用一个写数据库的API,必须等待这个调用完成并收到响应,避免留下半截子事务。
  2. 捕获完整状态 :调用State Store,将当前会话的所有状态(如上一节所述)序列化并持久化。这里的关键是 一致性快照 。你需要确保保存的状态是某个瞬间的完整视图,而不是正在变化过程中的碎片。
  3. 更新会话状态 :将会话状态标记为“ PAUSED_BY_USER ”或类似的枚举值。这有助于区分是因为错误暂停还是主动暂停。
  4. 清理与确认 :释放可能占用的临时资源(如打开的文件句柄、网络连接池),并向用户返回一个明确的标识符(如 session_id ),并告知“任务已暂停,状态已保存。您可以使用会话ID abc123 随时恢复”。

3.2 Agent请求中断(等待人工输入)

这是人机协作的精髓。Agent在运行中,遇到无法自主决策、需要权限确认或信息补充时,应主动“中断”自己,向用户发起询问。例如:

“我已经分析了前三个季度的销售数据,发现Q3的增长率异常。我需要访问财务系统的原始交易日志进行深度归因。请问您是否授权我调用‘get_raw_transaction_logs’这个工具?【是/否】” “我正在为您起草项目计划书,已经列出了主要阶段。关于‘技术选型评估’这个阶段,您希望我侧重于对比哪些维度的指标?(例如:性能、成本、社区活跃度、学习曲线)”

实现这种中断,需要在Agent的“大脑”(LLM)的提示词(Prompt)工程中下功夫。我们要明确告诉LLM,在遇到某些情况时, 不要 自己猜测,而是必须停止执行,输出一个特定的结构化请求。

例如,在你的系统提示词中,可以加入这样的指令:

你是一个协作式AI助手。当遇到以下情况时,你必须暂停执行,并严格按格式输出:
1.  需要执行具有潜在风险或需要权限的操作(如删除文件、发送邮件、调用付费API)。
2.  任务描述中存在模糊、二义性或信息缺失,足以影响后续关键决策。
3.  你产生了多个可行的后续方案,需要用户选择方向。

输出格式必须是严格的JSON:
{
  “action”: “request_human_input”,
  “question”: “向用户提出的清晰问题”,
  “context”: “当前决策相关的背景信息”,
  “options”: [“可选选项A”, “可选选项B”] // 可选
}

然后,Orchestrator在解析到LLM输出这样的JSON后,就会将会话状态置为“ AWAITING_HUMAN_RESPONSE ”,并将这个请求通过接口推送给前端界面,等待用户回复。用户的回复会被作为新的输入,连同保存的完整状态,一起喂给LLM,让任务继续。

3.3 实现中的坑与技巧

  • 状态序列化的陷阱 :直接使用Python的 pickle 来序列化复杂的Agent对象(尤其是那些包含了LLM客户端、网络会话等不可序列化资源的对象)会失败。正确的做法是设计一个 纯数据的状态对象(Data Class) ,只包含可序列化的基本类型(str, int, dict, list等)、字典和列表。在保存时,将Agent运行时的内存状态“脱水”到这个数据对象中;在恢复时,再根据这个数据对象“水合”出一个新的Agent实例。这类似于Web开发中的 serialize deserialize
  • 工具调用的原子性 :确保工具调用是“幕等”的,或者至少在中断点附近是安全的。例如,一个“发送邮件”的工具,如果在“准备发送”状态被中断,恢复后不应该重复发送。这通常需要在工具设计层面考虑,比如生成一个唯一的操作ID,并在执行前检查状态。
  • 给中断点加“书签” :在保存状态时,除了全局状态,还可以显式地记录“中断原因”和“预期恢复点”。例如: “pause_reason”: “NEED_PERMISSION_FOR_TOOL_X”, “resume_step”: “call_tool_X_with_user_provided_params” 。这样恢复逻辑可以更精准。

4. 实现恢复:从任意断点“无缝”接管的艺术

恢复机制是中断机制的逆过程,但更复杂,因为它需要处理“时间流逝”带来的潜在问题。一个健壮的恢复流程,不仅仅是加载状态那么简单。

4.1 恢复流程的步骤

  1. 会话查找与验证 :用户提供 session_id ,系统首先在State Store中查找。检查会话是否存在、状态是否为可恢复的(如 PAUSED , AWAITING_INPUT ),并检查是否过期(可根据业务设置TTL)。
  2. 状态反序列化与水合 :从数据库或文件中加载序列化的状态数据,并将其“水合”成运行时的内存对象。重建Agent的“大脑”(LLM客户端可能需要重新初始化,但提示词和历史已包含在状态中)、工具集绑定等。
  3. 上下文重建与注入 :这是最关键的一步。你需要将加载的完整对话历史、工作内存等,重新构建成LLM能够理解的上下文。对于大多数LLM API,这意味着要将历史消息重新组装成合适的 messages 数组(包括 system , user , assistant 角色),并确保顺序和内容完整无损。同时,那些保存在工作内存中的中间结果,可能需要以某种方式“提醒”给LLM,例如在系统提示词中追加一段:“以下是任务暂停前已获取的信息: {{working_memory}} ”。
  4. 决定恢复策略
    • 继续执行 :如果中断是简单的暂停,那么直接让Agent从它上次停止的“思考环节”继续即可。LLM会根据完整的上下文,自然地接上后续步骤。
    • 处理待决请求 :如果中断是因为Agent在等待用户输入( AWAITING_HUMAN_RESPONSE ),那么恢复流程需要将用户新提供的答复,作为一条新的 user 消息,插入到历史上下文的末尾,然后触发Agent继续处理。
    • 重新评估 :在某些场景下,中断时间可能很长,外部环境已变(例如,恢复时发现一个之前可用的API现在不可用了)。更稳健的设计是,在恢复时让Agent先做一个简短的“状态自检”,输出如“我已恢复。在暂停前,我们正在执行XX,已完成YY,下一步计划是ZZ。外部环境是否有变化需要我知晓?”。这可以通过在恢复后的第一条系统提示词中增加指令来实现。
  5. 继续执行与状态更新 :恢复后的Agent开始运行,其产生的新的状态变化(新的对话、工具调用、工作内存更新)需要被实时地、增量地同步回State Store,确保即使再次中断,状态也是最新的。

4.2 恢复时的挑战与解决方案

  • 上下文长度限制 :这是使用LLM时最现实的挑战。一个运行了上百轮对话、包含大量工具输出结果的复杂任务,其完整历史很可能超出模型的上下文窗口。简单的截断(只保留最近N条)会导致遗忘早期关键信息。
    • 解决方案1:分层记忆系统 。维护一个“核心记忆”或“摘要记忆”。在每次保存状态前,或者当历史达到一定长度时,让LLM对之前的对话和结果进行一次 增量式摘要 ,提炼出对完成最终目标至关重要的“事实”和“决策”。将这份摘要作为系统提示词的一部分,而原始详细历史则可以存档或丢弃。恢复时,加载的是“摘要”+“近期详细历史”。
    • 解决方案2:向量化检索 。将所有历史对话和工具结果块(chunk)都存入向量数据库。恢复时,根据当前任务目标和最近几步,去向量库中检索最相关的历史片段,动态地构建上下文。这更灵活,但实现复杂度更高。
  • 工具状态过期 :中断期间,外部工具或数据源可能已发生变化。例如,一个查询股票价格的工具,恢复后价格早已更新。
    • 解决方案 :对于对实时性敏感的工具调用,在恢复后的首次相关操作中,设计一个“验证”或“刷新”机制。或者,在Agent的提示词中强调,对于时间敏感的信息,应以恢复后的重新查询为准。
  • “失忆”与逻辑连贯性 :即使恢复了全部文本历史,LLM有时也可能在逻辑衔接上出现轻微“断层”,尤其是当任务规划非常复杂时。
    • 解决方案 :在恢复后的第一条助理消息中,可以“强迫”Agent先输出一个简短的回顾。在提示词中设计:“首先,请简要复述我们暂停前正在进行的任务、已完成的关键步骤以及下一步计划。确认无误后,我们再继续。” 这相当于让LLM自己做一个上下文加载正确性的校验。

5. 实战演练:构建一个支持协作的网页爬取与分析Agent

光讲理论有点抽象,我们用一个具体的例子来串起所有概念:构建一个 网页爬取与分析Agent 。它的任务是:用户给定一个主题(比如“大语言模型推理优化最新进展”),它能自动搜索、爬取相关文章,提取核心内容,并生成一份摘要报告。在这个过程中,用户可能需要介入,比如确认要爬取的网站列表、在遇到反爬机制时决定策略、或者对摘要的侧重点提出要求。

5.1 系统组件设计

  • Agent Brain :使用GPT-4或Claude 3等高性能LLM,负责规划步骤(如“先搜索关键词 -> 筛选前10个结果 -> 逐个爬取 -> 解析内容 -> 总结”),决定何时调用工具。
  • Orchestrator :用FastAPI或类似框架构建一个Web服务,管理会话。每个 /start_task 请求创建一个新会话。
  • State Store
    • Redis :存储活跃会话的实时状态( session_id -> 状态哈希 ),包括当前步骤、临时数据、锁等。设置30分钟过期时间,用于会话保持。
    • PostgreSQL :存储完整的、结构化的会话状态快照。表结构设计如下:
      CREATE TABLE agent_sessions (
          id VARCHAR(64) PRIMARY KEY,
          user_id VARCHAR(64),
          goal TEXT,
          status VARCHAR(32), -- ‘RUNNING’, ‘PAUSED’, ‘AWAITING_INPUT’, ‘COMPLETED’, ‘FAILED’
          current_step VARCHAR(255),
          serialized_state JSONB, -- 存储完整的对话历史、工作内存等
          created_at TIMESTAMP,
          updated_at TIMESTAMP
      );
      
  • Toolkit
    • web_search(query: str) -> List[SearchResult] : 调用SerpAPI或SearXNG进行搜索。
    • fetch_webpage(url: str) -> str : 使用 requests playwright 爬取网页内容,处理基础反爬(如User-Agent)。
    • parse_content_with_llm(html: str) -> dict : 调用LLM从HTML中提取标题、正文、发布时间等结构化信息。
    • generate_report(summaries: List[dict]) -> str : 调用LLM根据多篇文章摘要生成综合报告。
  • Human Interface :一个简单的Web前端,展示任务状态、实时日志,并提供“暂停”、“继续”、“提供输入”的按钮和输入框。

5.2 协作与中断点设计

我们在任务流程中预设几个关键的协作中断点:

  1. 搜索关键词确认 :Agent规划的第一步是生成搜索关键词。它可能会输出:“我将使用‘大语言模型 推理优化 2024 论文’进行搜索。您是否同意?或者有更具体的关键词?” 此时,Orchestrator会解析到这个 request_human_input ,暂停任务,将问题推给前端。
  2. 目标网站筛选 :搜索返回了50个结果。Agent可以分析这些结果的域名,然后询问:“搜索结果来自arxiv.org, huggingface.co, medium.com等。您希望我优先爬取哪些域名的内容?还是全部尝试?” 这避免了盲目爬取可能被禁止的网站。
  3. 爬取遇阻处理 :在爬取某个网站时, fetch_webpage 工具返回了 403 Forbidden 错误。Agent不应该无限重试或直接失败,而是可以上报:“尝试爬取 example.com/research 时遇到403错误,可能触发了反爬虫机制。现有策略:a) 跳过此网站;b) 添加随机延迟后重试;c) 切换到无头浏览器模式(较慢)。请选择。”
  4. 报告大纲确认 :在生成最终报告前,Agent可以先列出一个报告大纲(如:一、引言;二、主流优化技术对比;三、代表性论文解读;四、未来趋势)。用户可以修改这个大纲,Agent再基于此填充内容。

5.3 状态保存与恢复示例

假设任务在“爬取遇阻处理”这个点被中断,用户选择了策略 c) 。State Store中保存的 serialized_state JSON可能包含:

{
  “session_id”: “task_123”,
  “goal”: “调研大语言模型推理优化最新进展并生成报告”,
  “status”: “AWAITING_INPUT”,
  “current_step”: “handle_fetch_error”,
  “dialogue_history”: [
    {“role”: “user”, “content”: “帮我调研一下大语言模型推理优化的最新进展,写份摘要。”},
    {“role”: “assistant”, “content”: “我将开始这个任务。首先,我需要搜索相关的最新资料。我计划使用‘大语言模型 推理优化 2024 论文’作为初始关键词进行搜索,您看可以吗?”},
    {“role”: “user”, “content”: “可以,再加上‘efficient inference’这个英文关键词。”},
    {“role”: “assistant”, “content”: “好的,已更新关键词。正在执行搜索...”},
    // ... 更多历史
    {“role”: “assistant”, “content”: “在爬取 ‘example.com/research’ 时遇到403错误。现有策略:a) 跳过;b) 延迟重试;c) 用无头浏览器。请选择。”}
  ],
  “working_memory”: {
    “search_keywords”: [“大语言模型 推理优化 2024 论文”, “efficient inference”],
    “search_results”: [...],
    “crawled_successfully”: [...],
    “current_failed_url”: “example.com/research”,
    “error_detail”: “403 Forbidden”
  },
  “pending_human_request”: {
    “question”: “在爬取 ‘example.com/research’ 时遇到403错误。现有策略:a) 跳过;b) 延迟重试;c) 用无头浏览器。请选择。”,
    “options”: [“a”, “b”, “c”]
  }
}

当用户通过前端恢复此会话,并输入选择“c”后,Orchestrator会:

  1. 加载此状态。
  2. 将用户的输入“c”作为一条新的 user 消息(如“选择c:用无头浏览器模式”)追加到 dialogue_history
  3. 清除 pending_human_request
  4. 将状态置为 RUNNING
  5. 将重组后的完整历史(包括新消息)发送给Agent Brain。
  6. Agent Brain看到最新的用户消息是“选择c”,结合上下文( current_failed_url ),它就会明白:现在应该用更高级的无头浏览器工具去重试那个URL。任务得以继续。

6. 避坑指南:从设计到部署的实战经验

在真正构建这样一个系统的过程中,我遇到了不少预料之外的问题。这里分享几个关键的“坑”和应对策略。

6.1 状态爆炸与存储优化

随着任务进行,对话历史、工具返回的原始数据(尤其是爬取的HTML)会非常庞大,很快撑爆数据库或超出LLM上下文。

  • 对策
    • 非核心数据外链 :对于工具返回的巨型原始数据(如完整的HTML、长的API响应),不要直接存在主状态JSON里。将它们存储到对象存储(如S3/MinIO)或文件系统,在主状态中只保存一个引用链接或文件路径。
    • 增量式摘要 :如前所述,定期对历史进行摘要。可以每10轮对话或当历史token数达到阈值时,触发一次摘要,将摘要存入工作内存,并清空或归档旧的历史细节。
    • 压缩与清理 :在保存状态前,对可压缩的文本数据进行压缩(如gzip)。定期清理已完成或失败已久的会话数据。

6.2 工具调用的副作用与幕等性

很多工具调用是有副作用的(发邮件、改数据库)。如果中断恢复后不小心重复执行,会造成严重问题。

  • 对策
    • 为操作生成唯一ID :每次执行有副作用的工具时,生成一个全局唯一的操作ID(如UUID),并将 (session_id, tool_name, operation_id) 作为组合键,记录操作状态( pending , success , failed )到数据库。
    • 执行前检查 :在工具执行函数内部,首先检查这个操作ID是否已经成功执行过。如果是,则直接返回之前存储的结果,而不是真正执行。
    • 设计幕等API :如果可能,尽量让你调用的外部API本身是幕等的(比如用 PUT 代替 POST )。

6.3 长时任务的超时与心跳

一个复杂的Agent任务可能运行几十分钟甚至几个小时。网络可能不稳定,Orchestrator服务可能重启。

  • 对策
    • 实现心跳机制 :Agent执行器(或Orchestrator)定期(如每30秒)更新Redis中会话的 last_heartbeat 时间戳。
    • 设置看门狗 :一个独立的守护进程定期扫描所有 RUNNING 状态的会话,如果某个会话的 last_heartbeat 超过阈值(如5分钟),则认为该会话已僵死,将其状态置为 FAILED ,并可能触发告警或重试机制。
    • 任务可重入 :设计任务逻辑时,尽量让每个大的步骤自身是可重试的。例如,“爬取并分析A网站”这个子任务失败了,恢复后可以从头重试这个子任务,而不影响已经成功的“爬取并分析B网站”的结果。

6.4 用户交互的超时与放弃

用户可能发起一个请求后离开,或者长时间不回复Agent的提问。

  • 对策
    • 设置等待超时 :对于 AWAITING_INPUT 状态,设置一个超时时间(如24小时)。超时后,可以自动将会话状态置为 PAUSED ,并可能通过通知(如邮件)提醒用户。
    • 提供默认选项 :在设计Agent的提问时,可以提供一个合理的“默认选项”。例如,“如果30分钟内未收到您的回复,我将默认选择方案A并继续。” 这需要谨慎评估业务逻辑,确保默认选择是安全的。

构建一个支持人机协作与中断恢复的AI Agent,远比构建一个自动化的脚本复杂。它要求我们将软件工程的经典思想——状态管理、容错设计、用户交互——与LLM的能力深度结合。这个过程虽然充满挑战,但当你看到自己打造的Agent能够像一个真正的合作伙伴一样,与你并肩处理复杂任务,随时可以停下讨论,又能随时无缝继续时,那种成就感是无与伦比的。这或许就是当下AI应用开发最具魅力的方向之一。

更多推荐