最近智能体开发社区里有一个讨论度很高的话题上下文越来越长Agent 越来越贵。尤其是多轮对话、工具调用、人工介入混合出现的场景历史消息很快就会把上下文窗口占满。Google 新论文 SKILL.state 提出的方向恰好切中这个痛点与其把对话历史越堆越多不如把执行状态显式拿出来让模型只看当前该看的信息。这篇文章不打算复述论文公式而是从工程视角拆解 SKILL.state 背后的核心思路并给出一个可运行的最小状态驱动智能体 Demo。如果你正在做智能体开发、Agent 框架设计或者被长对话历史搞得很头痛这篇内容应该能给你一些直接可用的思路。1. 智能体上下文管理的核心矛盾1.1 对话历史为什么会失控不管是基于 LangChain、Dify、Coze还是自己写 Agent 编排代码最常见的做法是把系统提示词、用户消息、助手回复、工具调用结果全部塞进同一个 messages 数组每次请求都整体发给大模型。这样做在 Demo 阶段没有任何问题。可一旦进入真实业务对话历史就会快速膨胀。我见过一个销售线索智能体每个用户会话会经历用户咨询产品功能智能体调用 CRM 查询客户信息智能体调用订单系统查询历史订单智能体根据结果回答用户追问其他产品智能体再次调用多个接口。十轮以内messages 里已经有几十条记录。每次调用 API 时这些历史都要重新编码成 token导致三个直接问题成本上升Token 是计费核心历史越长单次调用费用越高响应变慢输入 token 越多首字延迟就越明显关键信息被稀释模型注意力可能被早期无效对话带走导致最近的目标不明确。更麻烦的是很多开发者在每轮循环里直接messages.append(...)压根没有做过上下文裁剪。等到上下文窗口溢出才开始想办法压缩。1.2 现有压缩方案为什么不够目前常见的上下文管理方案主要有这么几类方案基本思路主要问题滑动窗口只保留最近 N 条消息会丢失早期约束和用户真实需求摘要压缩把旧历史用 LLM 总结成一段文字有损关键事实可能被错误概括向量检索从历史中检索相似片段放回上下文检索质量不稳定执行类信息难以检索结构化历史只保留工具调用结果或关键字段仍属于“记录过程”不是“记录状态”这些方案本质上还在优化“怎么把对话历史变小”。但 SKILL.state 给出的思路不同把“历史”和“状态”拆开。对话历史是流水账而执行状态是程序当前运行到哪一步、还需要什么信息、已经拿到哪些结果。流水账可以丢弃但状态必须保留。理解了这一点再去看显式执行状态就会通畅很多。2. 理解 SKILL.state从对话历史到显式执行状态2.1 一个便于理解的生活化类比把智能体想象成一个客服专员。传统的对话历史驱动方式相当于每处理一个新问题都让客服把之前几十通电话录音完整听一遍然后自己总结客户目前到底在哪个环节。这显然不合理。而状态驱动的客服流程是这样的每个工单会有一个状态卡片上面写着“客户已提供订单号等待确认退款方式”。客服接到新消息时只需要看这张卡片再结合客户最新一句话就能决定下一步动作。旧录音可以封存卡片必须更新。所谓 SKILL.state核心就是把这张“状态卡片”显式建模而不是让它隐藏在对话历史里。2.2 执行状态与对话历史到底有什么区别我们可以把两者的差异列成一张对照表维度对话历史显式执行状态本质发生了什么现在进行到哪一步增长方式随轮数线性增长固定结构只更新字段信息类型用户原话、模型回复、工具输出当前步骤、已完成步骤、临时变量、分支条件丢失影响可能丢失部分上下文任务无法继续执行是否适合直接给模型越多越容易分散注意力小且精确节省 attention举个例子。一个工单处理任务对话历史会记录用户说“网络断了”助手问“什么优先级”用户说“P1”助手调用系统系统返回“处理中”显式执行状态会记录current_stepresolvingcontext{ticket_type: 网络故障, priority: P1}。模型下一步不需要知道“网络断了”这句话只要知道工单类型和优先级已经收集完整当前进入解决方案落地阶段。2.3 显式执行状态如何驱动智能体一个状态驱动的智能体通常由三部分组成状态结构描述任务当前进度、关键变量、错误信息状态转移规则定义哪些状态可以跳到哪些状态状态视图提示词把状态序列化成一段简短文本交给 LLM 决策。每一轮循环中LLM 的任务不是“回忆整个对话”而是读当前状态视图结合用户最新输入输出一个结构化动作例如{next_step: gathering_info, status: waiting_user}执行动作后更新状态。这样模型每次看到的输入长度几乎恒定不会随着对话轮数增长而膨胀。2.4 为什么能降低上下文消耗我们做一个粗略的 token 估算。假设每一轮对话平均 120 个 token10 轮之后完整历史方案单次请求约 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任务完成。合法转移关系如下当前步骤允许转移到的步骤initgathering_info、decidinggathering_infogathering_info、decidingdecidingresolving、doneresolvingdonedone无这里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.py4.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_factorylist) context: Dict[str, Any] field(default_factorydict) 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, storageNone): self._states storage if storage is not None else {} def create(self, task_id: str) - AgentState: state AgentState(task_idtask_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 state4.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_asciiFalse)}, 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_asciiFalse)) manager.transition( task_id, Step(decision[next_step]), statusAgentStatus(decision[status]), context_updatesdecision.get(update_context, {}), ) print(\n[执行后状态快照]) print(json.dumps(state.snapshot(), ensure_asciiFalse, indent2)) print(\n 最终状态 ) print(json.dumps(state.snapshot(), ensure_asciiFalse, indent2))最后是入口文件。# main.py from agent import run_agent if __name__ __main__: run_agent( task_idticket-001, messages[ 网络连接不稳定, 优先级是 P1, 开始处理, ], )4.5 运行结果说明运行python main.py后程序会经历以下几个阶段初始状态为init状态视图为空LLM 决定询问用户工单类型用户提供“网络连接不稳定”状态视图变为gathering_infocontext 记录ticket_type用户提供“优先级是 P1”context 记录priority系统内部自动进入deciding生成方案系统进入resolving模拟执行解决动作系统最终进入done状态变为completed。每一轮都会打印状态快照。可以看到状态视图中不会出现全部历史对话只有当前步骤和关键字段。这也正是 SKILL.state 想要解决的问题模型只需要关注“现在在哪一步”而不是“过去说了什么”。5. 把存量对话历史型智能体迁移到状态驱动如果你已经有一套基于完整对话历史的智能体不建议推倒重来。可以按下面几步渐进迁移。5.1 第一步区分状态与过程打开现有代码把 messages 里出现的所有信息分成两类状态类用户身份、订单号、审批人、当前阶段、已确认字段、工具返回结果过程类无意义的寒暄、中间推理过程、已经消费掉的旧回答。状态类信息必须显式保留过程类信息可以丢弃或压缩。5.2 第二步设计显式状态 智能体上下文管理：SKILL.state 显式执行状态驱动 Agent
最近智能体开发社区里有一个讨论度很高的话题上下文越来越长Agent 越来越贵。尤其是多轮对话、工具调用、人工介入混合出现的场景历史消息很快就会把上下文窗口占满。Google 新论文 SKILL.state 提出的方向恰好切中这个痛点与其把对话历史越堆越多不如把执行状态显式拿出来让模型只看当前该看的信息。这篇文章不打算复述论文公式而是从工程视角拆解 SKILL.state 背后的核心思路并给出一个可运行的最小状态驱动智能体 Demo。如果你正在做智能体开发、Agent 框架设计或者被长对话历史搞得很头痛这篇内容应该能给你一些直接可用的思路。1. 智能体上下文管理的核心矛盾1.1 对话历史为什么会失控不管是基于 LangChain、Dify、Coze还是自己写 Agent 编排代码最常见的做法是把系统提示词、用户消息、助手回复、工具调用结果全部塞进同一个 messages 数组每次请求都整体发给大模型。这样做在 Demo 阶段没有任何问题。可一旦进入真实业务对话历史就会快速膨胀。我见过一个销售线索智能体每个用户会话会经历用户咨询产品功能智能体调用 CRM 查询客户信息智能体调用订单系统查询历史订单智能体根据结果回答用户追问其他产品智能体再次调用多个接口。十轮以内messages 里已经有几十条记录。每次调用 API 时这些历史都要重新编码成 token导致三个直接问题成本上升Token 是计费核心历史越长单次调用费用越高响应变慢输入 token 越多首字延迟就越明显关键信息被稀释模型注意力可能被早期无效对话带走导致最近的目标不明确。更麻烦的是很多开发者在每轮循环里直接messages.append(...)压根没有做过上下文裁剪。等到上下文窗口溢出才开始想办法压缩。1.2 现有压缩方案为什么不够目前常见的上下文管理方案主要有这么几类方案基本思路主要问题滑动窗口只保留最近 N 条消息会丢失早期约束和用户真实需求摘要压缩把旧历史用 LLM 总结成一段文字有损关键事实可能被错误概括向量检索从历史中检索相似片段放回上下文检索质量不稳定执行类信息难以检索结构化历史只保留工具调用结果或关键字段仍属于“记录过程”不是“记录状态”这些方案本质上还在优化“怎么把对话历史变小”。但 SKILL.state 给出的思路不同把“历史”和“状态”拆开。对话历史是流水账而执行状态是程序当前运行到哪一步、还需要什么信息、已经拿到哪些结果。流水账可以丢弃但状态必须保留。理解了这一点再去看显式执行状态就会通畅很多。2. 理解 SKILL.state从对话历史到显式执行状态2.1 一个便于理解的生活化类比把智能体想象成一个客服专员。传统的对话历史驱动方式相当于每处理一个新问题都让客服把之前几十通电话录音完整听一遍然后自己总结客户目前到底在哪个环节。这显然不合理。而状态驱动的客服流程是这样的每个工单会有一个状态卡片上面写着“客户已提供订单号等待确认退款方式”。客服接到新消息时只需要看这张卡片再结合客户最新一句话就能决定下一步动作。旧录音可以封存卡片必须更新。所谓 SKILL.state核心就是把这张“状态卡片”显式建模而不是让它隐藏在对话历史里。2.2 执行状态与对话历史到底有什么区别我们可以把两者的差异列成一张对照表维度对话历史显式执行状态本质发生了什么现在进行到哪一步增长方式随轮数线性增长固定结构只更新字段信息类型用户原话、模型回复、工具输出当前步骤、已完成步骤、临时变量、分支条件丢失影响可能丢失部分上下文任务无法继续执行是否适合直接给模型越多越容易分散注意力小且精确节省 attention举个例子。一个工单处理任务对话历史会记录用户说“网络断了”助手问“什么优先级”用户说“P1”助手调用系统系统返回“处理中”显式执行状态会记录current_stepresolvingcontext{ticket_type: 网络故障, priority: P1}。模型下一步不需要知道“网络断了”这句话只要知道工单类型和优先级已经收集完整当前进入解决方案落地阶段。2.3 显式执行状态如何驱动智能体一个状态驱动的智能体通常由三部分组成状态结构描述任务当前进度、关键变量、错误信息状态转移规则定义哪些状态可以跳到哪些状态状态视图提示词把状态序列化成一段简短文本交给 LLM 决策。每一轮循环中LLM 的任务不是“回忆整个对话”而是读当前状态视图结合用户最新输入输出一个结构化动作例如{next_step: gathering_info, status: waiting_user}执行动作后更新状态。这样模型每次看到的输入长度几乎恒定不会随着对话轮数增长而膨胀。2.4 为什么能降低上下文消耗我们做一个粗略的 token 估算。假设每一轮对话平均 120 个 token10 轮之后完整历史方案单次请求约 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任务完成。合法转移关系如下当前步骤允许转移到的步骤initgathering_info、decidinggathering_infogathering_info、decidingdecidingresolving、doneresolvingdonedone无这里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.py4.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_factorylist) context: Dict[str, Any] field(default_factorydict) 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, storageNone): self._states storage if storage is not None else {} def create(self, task_id: str) - AgentState: state AgentState(task_idtask_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 state4.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_asciiFalse)}, 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_asciiFalse)) manager.transition( task_id, Step(decision[next_step]), statusAgentStatus(decision[status]), context_updatesdecision.get(update_context, {}), ) print(\n[执行后状态快照]) print(json.dumps(state.snapshot(), ensure_asciiFalse, indent2)) print(\n 最终状态 ) print(json.dumps(state.snapshot(), ensure_asciiFalse, indent2))最后是入口文件。# main.py from agent import run_agent if __name__ __main__: run_agent( task_idticket-001, messages[ 网络连接不稳定, 优先级是 P1, 开始处理, ], )4.5 运行结果说明运行python main.py后程序会经历以下几个阶段初始状态为init状态视图为空LLM 决定询问用户工单类型用户提供“网络连接不稳定”状态视图变为gathering_infocontext 记录ticket_type用户提供“优先级是 P1”context 记录priority系统内部自动进入deciding生成方案系统进入resolving模拟执行解决动作系统最终进入done状态变为completed。每一轮都会打印状态快照。可以看到状态视图中不会出现全部历史对话只有当前步骤和关键字段。这也正是 SKILL.state 想要解决的问题模型只需要关注“现在在哪一步”而不是“过去说了什么”。5. 把存量对话历史型智能体迁移到状态驱动如果你已经有一套基于完整对话历史的智能体不建议推倒重来。可以按下面几步渐进迁移。5.1 第一步区分状态与过程打开现有代码把 messages 里出现的所有信息分成两类状态类用户身份、订单号、审批人、当前阶段、已确认字段、工具返回结果过程类无意义的寒暄、中间推理过程、已经消费掉的旧回答。状态类信息必须显式保留过程类信息可以丢弃或压缩。5.2 第二步设计显式状态