发布时间:2026-07-12
标签:AI Agent|LLM|State|状态管理|工程实现


系列导航

上一篇:AI Agent 工程实践(15):RAG 真正的瓶颈不是检索,而是知识治理
下一篇: AI Agent 工程实践(17):Agent 为什么需要可观测性(Observability)?

本文是 [AI Agent 工程实践] 系列的第 16 篇(第二季 · 工程实现)。


有一次 Agent 跑了 20 分钟——规划、执行、审查,到了最后一步,OOM 了。

重启后,它完全不记得刚才在干什么。像从昏迷中醒来的人:我是谁?我在哪?

20 分钟白跑。从头开始。

那一刻我意识到,Agent 缺了一个关键零件:State。 它没有自己的"运行状态"——每一步干完就忘,像金鱼,永远活在当下。

这不是 Memory 的问题(10 篇讲过),Memory 管的是"以前的事"。State 管的是"正在进行的事"——我是谁、我在哪一步、如果挂了怎么接上。Prompt 是无状态的函数调用,Agent 是有状态的长运行进程。 这一篇,把 State 补上。


本文你将学到

✓ State 和 Memory 的本质区别——一个管"过去",一个管"现在"
✓ 为什么没有 State 的 Agent 只能跑短任务,一长就崩
✓ Agent 的六状态生命周期:START → Planning → Running → Waiting → Retry → Finish
✓ State 的三种关键能力:断点恢复、异步等待、可观测性

适合阅读

✓ 被 Agent "跑一半崩溃、重来全忘"折磨过的人
✓ 在搭长任务 Agent、需要异步等待外部事件的开发者
✓ 刚意识到"Agent 不是一次 Prompt"的人


问题背景

99% 的 Agent 教程在教"怎么写一个 Prompt"或"怎么调一个 Function Call"。这些教程跑出来的效果很好——因为它们的任务 10 秒内就能完成。

但一跑真实场景,立刻暴露:

长任务中途崩溃 = 从头来。 Agent 没有"我在第四步"的概念。挂了就挂了,重启后没有断点,不知道已经完成了哪些步骤、哪些还没做。

需要等外部事件时只能傻等。 Agent 调用了一个审批 API,对方要 2 小时回复。没有 State 的 Agent 只能阻塞——要么超时失败,要么一直占着进程白等。

同一个 Agent 跑多个任务会互相踩。 用户 A 的任务和用户 B 的任务跑在同一个 Agent 里,中间状态混在一起——A 的任务数据泄漏给了 B。

出了问题不知道在哪一步。 Agent 给了一个错误答案,但你不知道是规划错了、执行错了、还是外部 API 返回了脏数据——因为没有状态日志,排查全靠猜。

一句话:没有 State 的 Agent,本质就是一个"增强版 curl"——你调它一次,它回你一次。 它不是一个进程,只是一个函数调用。


错误尝试

第一次:把 State 存进 Memory

最自然的想法:前面不是讲了 Memory(10 篇)吗?把当前进度写进去不就行了?

结果:Memory 是跨会话的持久数据,State 是当前任务的瞬时状态。把"正在审查第三份文件"存进 Memory,下次新对话 Agent 会以为"有个审查进行中"——Memory 污染了。Memory 和 State 不是一个东西:Memory 管"以前学到的",State 管"正在发生的"。

第二次:用全局变量存状态

简单粗暴——一个 current_step 全局变量,跑一步改一次。

结果:单任务能跑,多任务立刻崩。A 任务跑到第四步时 B 任务来了,current_step 被 B 覆盖成第二步——两个任务的状态串了。而且全局变量在进程里——进程重启就没了。全局变量是无隔离、无持久、无恢复的"三无状态"。

两次尝试指向同一个结论:State 需要三个核心能力——隔离(多任务不串)、持久(挂了能恢复)、可观测(知道在哪一步)。 这三个,Memory 和全局变量都给不了。


关键观察

我把 Agent 的"有 State"和"无 State"做了对比,差异判若两个系统:

能力 无 State(单次 Prompt) 有 State(进程级 Agent)
断点恢复 无——从头来 有——从上次中断位置继续
异步等待 阻塞 or 超时失败 进入 Waiting,外部事件唤醒
多任务并发 互相踩 隔离运行
出问题定位 靠猜 查 State 日志
最长可跑任务 分钟级 天级

Prompt 是无状态的函数调用,Agent 是有状态的长运行进程。

函数是 f(x) → y,进程是 start → run → wait → continue → finish。Agent 应该是后者——它是一个长运行的、可中断、可恢复的进程,不是一次性的函数调用。

定位真正原因

问题不在"Agent 不够聪明",而在Agent 被设计成了无状态函数——每次调用都是全新的,没有对"上一次调用的自己"的感知。

正确做法是给 Agent 赋予一个独立于 Memory 的任务级状态。Memory 管"长期知识"(跨会话),State 管"当前进度"(会话内)。两者各司其职,互不越界。

Memory = 学到了什么(知识)。State = 正在做什么(进程)。


最终方案:Agent 的六状态生命周期

你给的六阶段状态机——这是 Agent 作为"进程"的最小骨架:

六状态详解

状态 Agent 在干什么 转入条件 转出条件
START 刚启动,等待任务 系统启动 / 从 Retry 回来 收到任务 → Planning
Planning 拆解任务,生成执行计划 从 START 或 Retry 进入 计划生成 → Running
Running 正在执行步骤 从 Planning / Waiting / Retry 进入 完成→Finish / 等外部→Waiting / 出错→Retry
Waiting 等待外部事件(审批/API/人工输入) 需要外部输入时 事件就绪→Running / 超时→Retry
Retry 尝试从错误中恢复 执行出错 / 等待超时 恢复成功→Running / 不可恢复→START
Finish 任务完成 全部步骤通过 输出结果,回收 State

六个状态不是装饰——每一个都对应一种真实的生产场景。 没有 Waiting 就只能阻塞,没有 Retry 就只能失败,没有 START 和 Finish 就没有生命周期。

State 的核心数据结构

# agent_state.yaml —— 任务级状态的完整快照
task_id: "task_20260712_001"
status: running               # 当前状态
current_step: 3               # 执行到哪一步
plan:                         # Planner 产出的步骤清单
  - { id: 1, desc: "分析需求", status: done }
  - { id: 2, desc: "设计接口", status: done }
  - { id: 3, desc: "编写实现", status: running }
  - { id: 4, desc: "编写测试", status: pending }
  - { id: 5, desc: "代码审查", status: pending }
history:                      # 可观测——完整状态流转日志
  - { from: start, to: planning, at: "10:00" }
  - { from: planning, to: running, at: "10:02" }
  - { from: running, to: waiting, at: "10:15", reason: "等待审批" }
  - { from: waiting, to: running, at: "12:30", reason: "审批通过" }
retry_count: 1                # 重试次数
checkpoint:                   # 断点数据——从这里恢复
  step_id: 3
  snapshot: "编写实现的部分结果..."

State 里记录四样东西:在哪一步(status/current_step)、下一步是什么(plan)、怎么走到这里的(history)、挂了怎么恢复(checkpoint)。 四样齐全,Agent 才是一个"进程"。

断点恢复的实际效果

同一个长任务,有和无 State 的差别:

  • 无 State:跑了 20 分钟后 OOM → 重启 → 忘了在干什么 → 从头跑 → 20 分钟白费。
  • 有 State:跑了 20 分钟后 OOM → 重启 → 从 State 读到 current_step=3, checkpoint=... → 从第 3 步续跑 → 只损失了最后一步的时间。

断点恢复不是锦上添花——是长任务的生命线。


架构图 / 流程图

State 在 Agent 架构中的位置

关键点:State 和 Memory 是两个独立的数据域——State 在任务结束时回收(或归档为日志),Memory 跨会话持久。两者通过 Agent 引擎协调,不直接通信。


代码或配置示例

State Manager 核心逻辑

class AgentState:
    def __init__(self, task_id: str):
        self.task_id = task_id
        self.status = "start"
        self.current_step = 0
        self.plan = []
        self.history = []
        self.checkpoint = None
        self.retry_count = 0

    def transition(self, to: str, reason: str = ""):
        """状态流转 + 记录历史日志"""
        self.history.append({
            "from": self.status, "to": to,
            "at": now(), "reason": reason,
        })
        self.status = to

    def save_checkpoint(self, data: dict):
        """保存断点——挂了就从这里恢复"""
        self.checkpoint = {
            "step_id": self.current_step,
            "data": data,
            "saved_at": now(),
        }

    @classmethod
    def restore(cls, task_id: str) -> "AgentState":
        """从持久化存储恢复——挂了的 Agent 复活"""
        state = db.load(f"state:{task_id}")
        if state and state.checkpoint:
            return state          # 有断点,从上次继续
        raise StateNotFound       # 无断点,从头开始

带 State 的 Agent 执行循环

async def agent_loop(task, state: AgentState):
    while state.status != "finish":
        match state.status:
            case "start":
                state.transition("planning")
            case "planning":
                state.plan = await plan(task)
                state.transition("running")
            case "running":
                try:
                    result = await execute_step(state.plan[state.current_step])
                    state.save_checkpoint(result)     # 每步都留断点
                    state.current_step += 1
                    if state.current_step == len(state.plan):
                        state.transition("finish")
                except RecoverableError:
                    state.transition("retry", "执行出错")
                except NeedExternalInput as e:
                    state.transition("waiting", e.reason)
            case "waiting":
                event = await wait_for_external(timeout=300)
                state.transition("running" if event else "retry")
            case "retry":
                state.retry_count += 1
                # 重试成功 → running;超过上限 → 回 start 重新规划
                state.transition("running" if state.retry_count < 3 else "start")

三段代码拼起来,Agent 就从一个"函数调用"变成了"带状态的进程":有断点恢复(checkpoint)、有异步等待(waiting)、有重试上限(retry_count<3)、有完整的状态日志(history)。差这 60 行代码,Agent 和 curl 的区别就体现在这里。


设计权衡

候选方案 优点 缺点 为什么不选
无 State(纯函数式 Agent) 最简单 无断点、无等待、无并发 只能跑 10 秒内的短任务
用 Memory 当 State 复用已有存储 长期知识被瞬时状态污染 Memory 和 State 是不同的数据域
全局变量 零延迟 无隔离、无持久、无恢复 三无状态,一挂全丢
独立 State + 六状态机 断点恢复、异步等待、状态可观测 需实现状态流转逻辑 选择理由:唯一让 Agent 能跑长任务的方案

不是所有 Agent 都需要 State。 单次问答 Agent("帮我查一下天气")用不上——调完就结束,没有"中间状态"给你存。State 的价值在长任务、多步骤、需要等外部事件时体现。和第 08/12 篇一样:规模不到,别上。


总结

✅ Agent 不是一次 Prompt——它是长运行的进程。没有 State 就没有断点、没有等待、没有并发。
✅ State 和 Memory 是两个独立的数据域:State 管"正在做什么"(进程),Memory 管"学到了什么"(知识)。
✅ 六状态生命周期:START → Planning → Running → Waiting → Retry → Finish,每个都对应真实生产场景。
✅ State 的核心结构:在哪(status/step)、怎么走(plan)、怎么来的(history)、怎么恢复(checkpoint)。
✅ State 只对长任务有价值——单次问答用不上。规模不到别上。


参考资料

  • 第 10 篇:Memory 架构设计 → Memory 和 State 的准确边界定义,State 不越界
  • 第 11 篇:Workflow 实现 → State 驱动的 Workflow 状态流转
  • Temporal / AWS Step Functions 文档 → 长运行进程的状态管理与断点恢复参考
  • Actor Model (Carl Hewitt) → State 隔离的并发模型,Agent 多任务并发的理论基础

系列导航

上一篇:AI Agent 工程实践(15):RAG 真正的瓶颈不是检索,而是知识治理
下一篇:AI Agent 工程实践(17):Agent 为什么需要可观测性(Observability)?
本文是 [AI Agent 工程实践] 系列的第 16 篇(第二季 · 工程实现)。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐