智能体上下文管理：SKILL.state 显式执行状态驱动 Agent

最近智能体开发社区里有一个讨论度很高的话题：上下文越来越长，Agent 越来越贵。尤其是多轮对话、工具调用、人工介入混合出现的场景，历史消息很快就会把上下文窗口占满。Google 新论文 SKILL.state 提出的方向恰好切中这个痛点：与其把对话历史越堆越多，不如把执行状态显式拿出来，让模型只看当前该看的信息。

这篇文章不打算复述论文公式，而是从工程视角拆解 SKILL.state 背后的核心思路，并给出一个可运行的最小状态驱动智能体 Demo。如果你正在做智能体开发、Agent 框架设计，或者被长对话历史搞得很头痛，这篇内容应该能给你一些直接可用的思路。

1. 智能体上下文管理的核心矛盾

1.1 对话历史为什么会失控

不管是基于 LangChain、Dify、Coze，还是自己写 Agent 编排代码，最常见的做法是：把系统提示词、用户消息、助手回复、工具调用结果全部塞进同一个 messages 数组，每次请求都整体发给大模型。

这样做在 Demo 阶段没有任何问题。可一旦进入真实业务，对话历史就会快速膨胀。

我见过一个销售线索智能体，每个用户会话会经历：

1. 用户咨询产品功能；
2. 智能体调用 CRM 查询客户信息；
3. 智能体调用订单系统查询历史订单；
4. 智能体根据结果回答；
5. 用户追问其他产品；
6. 智能体再次调用多个接口。

十轮以内，messages 里已经有几十条记录。每次调用 API 时，这些历史都要重新编码成 token，导致三个直接问题：

• 成本上升：Token 是计费核心，历史越长，单次调用费用越高；
• 响应变慢：输入 token 越多，首字延迟就越明显；
• 关键信息被稀释：模型注意力可能被早期无效对话带走，导致最近的目标不明确。

更麻烦的是，很多开发者在每轮循环里直接messages.append(...)，压根没有做过上下文裁剪。等到上下文窗口溢出，才开始想办法压缩。

1.2 现有压缩方案为什么不够

目前常见的上下文管理方案主要有这么几类：

这些方案本质上还在优化“怎么把对话历史变小”。但 SKILL.state 给出的思路不同：把“历史”和“状态”拆开。

对话历史是流水账，而执行状态是程序当前运行到哪一步、还需要什么信息、已经拿到哪些结果。流水账可以丢弃，但状态必须保留。理解了这一点，再去看显式执行状态，就会通畅很多。

2. 理解 SKILL.state：从对话历史到显式执行状态

2.1 一个便于理解的生活化类比

把智能体想象成一个客服专员。传统的对话历史驱动方式，相当于每处理一个新问题，都让客服把之前几十通电话录音完整听一遍，然后自己总结客户目前到底在哪个环节。

这显然不合理。

而状态驱动的客服流程是这样的：每个工单会有一个状态卡片，上面写着“客户已提供订单号，等待确认退款方式”。客服接到新消息时，只需要看这张卡片，再结合客户最新一句话，就能决定下一步动作。

旧录音可以封存，卡片必须更新。所谓 SKILL.state，核心就是把这张“状态卡片”显式建模，而不是让它隐藏在对话历史里。

2.2 执行状态与对话历史到底有什么区别

我们可以把两者的差异列成一张对照表：

举个例子。一个工单处理任务：

• 对话历史会记录：用户说“网络断了”，助手问“什么优先级”，用户说“P1”，助手调用系统，系统返回“处理中”；
• 显式执行状态会记录：current_step=resolving，context={"ticket_type": "网络故障", "priority": "P1"}。

模型下一步不需要知道“网络断了”这句话，只要知道工单类型和优先级已经收集完整，当前进入解决方案落地阶段。

2.3 显式执行状态如何驱动智能体

一个状态驱动的智能体，通常由三部分组成：

1. 状态结构：描述任务当前进度、关键变量、错误信息；
2. 状态转移规则：定义哪些状态可以跳到哪些状态；
3. 状态视图提示词：把状态序列化成一段简短文本，交给 LLM 决策。

每一轮循环中，LLM 的任务不是“回忆整个对话”，而是：

• 读当前状态视图；
• 结合用户最新输入；
• 输出一个结构化动作，例如{"next_step": "gathering_info", "status": "waiting_user"}；
• 执行动作后更新状态。

这样，模型每次看到的输入长度几乎恒定，不会随着对话轮数增长而膨胀。

2.4 为什么能降低上下文消耗

我们做一个粗略的 token 估算。假设每一轮对话平均 120 个 token，10 轮之后：

• 完整历史方案：单次请求约 1200 个 token；
• 状态视图方案：状态视图可能只有 200 到 300 个 token，加上最近一轮输入，仍然远低于完整历史。

更重要的是，历史方案中模型每次都要重新处理所有早期消息，状态方案只需要处理当前快照。即便引入“最近 N 条原始消息”用于辅助理解，整体长度也远远低于全量历史。

这也是为什么 SKILL.state 提出的方向，值得所有 Agent 开发者关注。

3. 设计一个带显式执行状态的智能体执行框架

3.1 状态模型包含哪些字段

设计显式执行状态时，不能简单存一整个 JSON，要根据任务场景拆分字段。一个通用的 AgentState 可以参考下面这些字段：

• task_id：任务唯一 ID，用于关联状态；
• status：任务当前生命周期状态，例如运行中、等待用户、已完成；
• current_step：当前执行步骤，例如收集信息、决策、执行工具；
• completed_steps：已完成的步骤列表，方便排障和审计；
• context：关键上下文变量，例如用户填写的字段、工具返回结果；
• last_error：最近一次错误信息；
• retry_count：当前步骤重试次数，用于避免死循环；
• updated_count：状态更新次数，便于观察状态变化频率。

这里的核心思想是：凡是下一步决策需要的信息，都尽量放进显式字段；凡是过去已经消费掉的对话内容，不进状态。

3.2 状态定义与状态转移表

状态不能随便跳，否则会出现“信息没收集完就进入解决阶段”之类的问题。

以工单处理为例，我们可以定义如下步骤：

• init：初始状态；
• gathering_info：收集用户信息；
• deciding：生成处理方案；
• resolving：执行解决动作；
• done：任务完成。

合法转移关系如下：

这里gathering_info允许自循环，是因为用户可能分多次补全信息。每次补充后仍停留在收集信息阶段，直到关键字段齐全。

3.3 提示词中如何放置状态视图

状态驱动的另一个关键点是提示词组织。不需要把所有历史塞进 messages，而是把状态视图放在 system 或 user 前缀中，再附加最近一条用户消息。

示例结构：

系统：你是工单处理助手。请根据当前执行状态和用户输入，输出下一步动作。 当前执行状态： - 当前步骤: gathering_info - 已完成步骤: init - 关键上下文: {"ticket_type": "网络故障"} 用户输入： 优先级是 P1 请输出 JSON 动作。

这样模型每次需要处理的内容非常短，决策路径也更清晰。

3.4 状态存储与一致性

状态一定是可持久化的，不能只放在内存里。

• 单机场景：可以用字典存储；
• 多实例场景：建议用 Redis 存储序列化后的状态；
• 强一致场景：需要把状态更新做成幂等操作，避免并发状态下覆盖；
• 审计场景：可以同步记录状态变更日志，保留完整历史便于排查。

状态更新时，建议带版本号或更新时间，防止多个并发请求同时修改同一个状态对象。

4. 完整代码示例：一个可运行的 SKILL.state 最小实现

下面我用 Python 写一个最小但完整的状态驱动智能体 Demo。它不依赖外部大模型 API，用规则函数模拟 LLM 决策，重点展示状态管理、状态视图、状态转移的完整流程。

4.1 项目结构

skill_state_demo/ ├── state.py ├── state_manager.py ├── agent.py └── main.py

4.2 状态定义与状态管理器

先定义枚举和状态数据类。

# state.py from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, Optional class AgentStatus(str, Enum): CREATED = "created" RUNNING = "running" WAITING_USER = "waiting_user" COMPLETED = "completed" FAILED = "failed" class Step(str, Enum): INIT = "init" GATHERING_INFO = "gathering_info" DECIDING = "deciding" RESOLVING = "resolving" DONE = "done" @dataclass class AgentState: task_id: str status: AgentStatus = AgentStatus.CREATED current_step: Step = Step.INIT completed_steps: list = field(default_factory=list) context: Dict[str, Any] = field(default_factory=dict) last_error: Optional[str] = None retry_count: int = 0 updated_count: int = 0 def snapshot(self) -> Dict[str, Any]: return { "task_id": self.task_id, "status": self.status.value, "current_step": self.current_step.value, "completed_steps": list(self.completed_steps), "context": dict(self.context), "last_error": self.last_error, "retry_count": self.retry_count, "updated_count": self.updated_count, }

然后是状态管理器，负责创建状态和执行合法转移。

# state_manager.py from state import AgentState, AgentStatus, Step STATE_MACHINE = { Step.INIT: {Step.GATHERING_INFO, Step.DECIDING}, Step.GATHERING_INFO: {Step.GATHERING_INFO, Step.DECIDING}, Step.DECIDING: {Step.RESOLVING, Step.DONE}, Step.RESOLVING: {Step.DONE}, Step.DONE: set(), } class StateManager: def __init__(self, storage=None): self._states = storage if storage is not None else {} def create(self, task_id: str) -> AgentState: state = AgentState(task_id=task_id) self._states[task_id] = state return state def get(self, task_id: str) -> AgentState: return self._states[task_id] def transition(self, task_id: str, new_step: Step, status: AgentStatus = None, context_updates: dict = None) -> AgentState: state = self._states[task_id] if isinstance(new_step, str): new_step = Step(new_step) if isinstance(status, str): status = AgentStatus(status) allowed = STATE_MACHINE[state.current_step] if new_step not in allowed: raise ValueError( f"非法的状态迁移: {state.current_step} -> {new_step}" ) if state.current_step.value not in state.completed_steps: state.completed_steps.append(state.current_step.value) state.current_step = new_step if status is not None: state.status = status if context_updates: state.context.update(context_updates) state.updated_count += 1 return state

4.3 构造状态视图与模拟 LLM 决策

下面这段代码是核心。它会把状态对象序列化成一段简短的“状态视图”，提供给 LLM。同时提供一个模拟的call_llm，真实项目中应替换为对真实模型接口的调用。

# agent.py import json from state import AgentState, AgentStatus, Step from state_manager import StateManager def build_state_view(state: AgentState) -> str: lines = [ f"任务ID: {state.task_id}", f"当前步骤: {state.current_step.value}", f"状态: {state.status.value}", f"已完成步骤: {', '.join(state.completed_steps) if state.completed_steps else '无'}", f"关键上下文: {json.dumps(state.context, ensure_ascii=False)}", f"重试次数: {state.retry_count}", ] return "\n".join(lines) def build_history_prompt(messages) -> str: # 模拟“把所有对话历史都塞进 prompt”的老方案，用于对比 lines = [] for role, content in messages: lines.append(f"{role}: {content}") return "\n".join(lines) def estimate_tokens(text: str) -> int: # 粗粒度估算：中英文混合场景约 0.75 个字符折算 1 个 token return max(1, int(len(text) / 0.75)) def call_llm(state_view: str, user_input: str) -> dict: # 这里用规则函数模拟 LLM 的决策输出。 # 正式项目中应替换为真实 LLM 调用，并要求模型输出 JSON 动作。 if "ticket_type" not in state_view: if not user_input: return { "next_step": "gathering_info", "status": "waiting_user", "question": "请提供工单类型", "update_context": {}, } return { "next_step": "gathering_info", "status": "waiting_user", "question": "已收到工单类型，请补充优先级", "update_context": {"ticket_type": user_input}, } if "priority" not in state_view: if not user_input: return { "next_step": "gathering_info", "status": "waiting_user", "question": "请提供优先级", "update_context": {}, } return { "next_step": "gathering_info", "status": "waiting_user", "question": "好的，正在生成处理方案", "update_context": {"priority": user_input}, } if "当前步骤: deciding" in state_view: return { "next_step": "resolving", "status": "running", "update_context": {"solution": "根据 P1 优先级生成解决方案"}, } if "当前步骤: resolving" in state_view: return { "next_step": "done", "status": "completed", "update_context": {"final_result": "工单已关闭，用户已收到通知"}, } return { "next_step": "deciding", "status": "running", "update_context": {"analysis": "完成信息收集，进入决策"}, }

4.4 主流程运行与验证

主流程会负责：状态机运行、等待用户输入、调用 LLM 决策、执行状态转移、打印状态快照。

# agent.py（续） def run_agent(task_id: str, messages): manager = StateManager() state = manager.create(task_id) msg_index = 0 while state.status != AgentStatus.COMPLETED: if state.status == AgentStatus.WAITING_USER: if msg_index >= len(messages): print("等待用户补充信息，但消息已经用完了，退出。") break user_input = messages[msg_index] msg_index += 1 else: user_input = "" print("-" * 70) print("用户输入:", user_input if user_input else "（内部自动执行）") state_view = build_state_view(state) print("\n[状态视图，发送给 LLM 的输入]") print(state_view) history_messages = [ ("system", "你是一个工单处理助手"), ("user", "网络连接不稳定"), ("assistant", "请问是什么优先级？"), ("user", "优先级是 P1"), ("assistant", "正在处理…"), ] history_prompt = build_history_prompt(history_messages) print( f"\n[对比] 状态视图约 {estimate_tokens(state_view)} tokens；" f"完整历史约 {estimate_tokens(history_prompt)} tokens" ) decision = call_llm(state_view, user_input) print("\n[LLM 决策]", json.dumps(decision, ensure_ascii=False)) manager.transition( task_id, Step(decision["next_step"]), status=AgentStatus(decision["status"]), context_updates=decision.get("update_context", {}), ) print("\n[执行后状态快照]") print(json.dumps(state.snapshot(), ensure_ascii=False, indent=2)) print("\n=== 最终状态 ===") print(json.dumps(state.snapshot(), ensure_ascii=False, indent=2))

最后是入口文件。

# main.py from agent import run_agent if __name__ == "__main__": run_agent( task_id="ticket-001", messages=[ "网络连接不稳定", "优先级是 P1", "开始处理", ], )

4.5 运行结果说明

运行python main.py后，程序会经历以下几个阶段：

1. 初始状态为init，状态视图为空，LLM 决定询问用户工单类型；
2. 用户提供“网络连接不稳定”，状态视图变为gathering_info，context 记录ticket_type；
3. 用户提供“优先级是 P1”，context 记录priority；
4. 系统内部自动进入deciding，生成方案；
5. 系统进入resolving，模拟执行解决动作；
6. 系统最终进入done，状态变为completed。

每一轮都会打印状态快照。可以看到，状态视图中不会出现全部历史对话，只有当前步骤和关键字段。这也正是 SKILL.state 想要解决的问题：模型只需要关注“现在在哪一步”，而不是“过去说了什么”。

5. 把存量对话历史型智能体迁移到状态驱动

如果你已经有一套基于完整对话历史的智能体，不建议推倒重来。可以按下面几步渐进迁移。

5.1 第一步：区分状态与过程

打开现有代码，把 messages 里出现的所有信息分成两类：

• 状态类：用户身份、订单号、审批人、当前阶段、已确认字段、工具返回结果；
• 过程类：无意义的寒暄、中间推理过程、已经消费掉的旧回答。

状态类信息必须显式保留，过程类信息可以丢弃或压缩。

5.2 第二步：设计显式状态