1. 项目概述为什么我们需要Plan Mode如果你已经跟着这个系列从零开始搭建过基础的AI Agent可能会发现一个棘手的问题当任务稍微复杂一点比如“帮我查一下天气如果下雨就提醒我带伞然后顺便看看明天的会议安排”我们之前那种线性的、一步接一步的执行模式就开始捉襟见肘了。Agent要么会卡在某个步骤要么会忽略掉条件判断一股脑地把所有动作都执行了。这就像让一个新手司机不看导航、不顾红绿灯只凭感觉开车出事是迟早的。Plan Mode或者说规划模式就是为了解决这个“复杂任务的安全执行”问题而生的核心机制。简单来说Plan Mode让我们的AI Agent从一个只会执行简单指令的“实习生”升级为一个懂得规划、判断、回溯的“项目主管”。它不再是被动地响应单个指令而是主动地将一个复杂的、模糊的用户目标Goal拆解成一个清晰的、可执行的行动计划Plan并监督这个计划一步步安全地落地。这里面的关键词是“复杂”和“安全”。复杂意味着任务包含分支、循环、条件判断甚至异常处理安全则意味着执行过程可控、可预测、可中断不会因为一个步骤的失败导致整个系统崩溃或者做出危险操作。从最近大家搜索的热词也能看出无论是“状态机”、“三段式状态机”还是“安全执行要求”都指向了同一个核心如何为AI Agent赋予一个可靠的中枢神经系统。这个系统需要理解任务的结构状态机管理执行的生命周期规划、执行、检查并确保每一步都在预设的安全边界内安全执行。这也是为什么像“Harness”这样的基础设施层概念会被频繁提及——它就像宇航员的舱外航天服不代替宇航员Agent做决策但为其在复杂危险环境真实世界中的行动提供生命保障。所以这篇文章我们就来深入这个Agent的“驾驶舱”看看如何亲手打造一个属于你自己的Plan Mode。我们会从最经典的状态机模型入手用Python一步步实现一个具备任务分解、条件执行和错误处理能力的规划引擎。你会发现它不仅是Agent的核心其设计思想也能广泛应用在你日常开发的任何需要流程控制的场景中。2. 核心架构设计状态机与规划引擎要理解Plan Mode首先得理解它的理论基础有限状态机Finite-State Machine, FSM。别被这个名字吓到你可以把它想象成一个智能灯的开关面板。这个灯有“关闭”、“开启”、“闪烁”几种状态。你按“开”按钮它从“关闭”状态切换到“开启”状态再按“模式”按钮它可能从“开启”进入“闪烁”状态。任何时刻灯都处于且仅处于一种明确的状态而触发状态切换的就是你的按键指令事件。对于AI Agent的Plan Mode而言这个“状态机”管理的是任务执行的生命周期状态。一个经典且实用的模型是三段式状态机它通常包含以下状态规划状态PlanningAgent接收到一个复杂目标如“安排出差”。在此状态下Agent的核心工作是调用LLM进行任务分解和规划。输入是用户目标输出是一个结构化的计划通常可以表示为任务列表或流程图。例如计划可能是[预订机票 预订酒店 准备出差材料]。这个计划本身可能还带有依赖关系必须先订票才能知道时间订酒店和条件判断如果预算超支则选择更便宜的酒店。执行状态Executing计划生成后Agent进入执行状态。在此状态下Agent按照规划好的步骤依次调用相应的工具Skills或子Agent来完成任务。比如调用“机票预订工具”执行“预订机票”这个子任务。每执行完一步都会产生一个结果成功、失败、附带数据。检查状态Evaluating/Checking这是确保“安全执行”的关键环节。在一个子任务执行后系统不会盲目进入下一步而是进入检查状态。在此状态下Agent会评估上一步的执行结果是否成功返回的数据是否符合预期是否触发了某个条件如“预算超支”根据评估结果决定下一步是继续执行下一个任务、跳转到另一个分支、重试当前任务还是整体失败并进入错误处理流程。这三个状态循环往复构成了Plan Mode的主干。而驱动状态流转的“事件”就是子任务的完成信号、定时器超时或者用户的中断指令。基于这个模型我们可以设计一个规划引擎Planner Engine作为Plan Mode的大脑。它的职责是解析与生成计划将用户目标转化为可操作的计划结构。计划可以用JSON、YAML或自定义的DSL领域特定语言描述其中明确标出任务节点、依赖关系和条件分支。状态管理维护当前执行到了哪个节点整个计划处于什么状态进行中、已完成、已失败、已暂停。调度与决策根据当前状态和执行结果决定下一个要执行的任务是什么。这涉及到简单的顺序执行以及复杂的条件分支和循环判断。上下文管理在整个计划执行过程中维护一个共享的上下文Context。这个上下文就像一块公共白板记录了所有已执行任务的结果数据如机票订单号ABC123酒店入住日期2023-10-27供后续任务读取和判断。一个常见的计划结构描述可能长这样JSON格式{ “goal”: “安排一次北京至上海的出差” “plan”: [ { “id”: “task_1” “description”: “查询并预订北京到上海的机票” “skill”: “book_flight_ticket” “dependencies”: [] “on_success”: “task_2” “on_failure”: “handle_flight_failure” } { “id”: “task_2” “description”: “根据机票日期预订上海酒店” “skill”: “book_hotel” “dependencies”: [“task_1”] “condition”: “context.get(‘flight_booked’) True” “on_success”: “task_3” “on_failure”: “handle_hotel_failure” } { “id”: “task_3” “description”: “生成出差事项清单” “skill”: “generate_checklist” “dependencies”: [“task_1” “task_2”] } ] }这个引擎的实现可以选择纯代码编写也可以利用现有的状态机库如Python的transitions。但对于学习和深度定制而言我强烈建议从零开始实现一个轻量版这能让你透彻理解每一个流转细节。注意在设计规划引擎时一个重要的考量是“规划”的时机。有两种主要策略静态规划和动态重规划。静态规划在开始时生成完整计划之后严格按计划执行效率高但灵活性差。动态重规划则在每一步执行后都重新评估剩余计划甚至重新调用LLM生成后续步骤灵活性极高但耗时且可能产生不一致。对于大多数确定性较强的复杂任务我推荐采用“主计划静态生成局部条件动态调整”的混合策略。3. 核心组件实现状态机、上下文与安全装饰器理论说完了我们开始动手写代码。我们将用Python构建一个最小可行但功能完整的Plan Mode核心。这个核心主要包含三个部分状态机引擎、共享上下文管理器和一个确保安全执行的“防护网”——装饰器。3.1 实现一个轻量级状态机我们首先实现一个简单的状态机来管理PlanningExecutingEvaluating这三个核心状态。from enum import Enum from typing import Any Callable Optional class AgentState(Enum): Agent计划执行状态枚举 IDLE “idle” # 空闲等待目标 PLANNING “planning” # 正在规划 EXECUTING “executing” # 正在执行子任务 EVALUATING “evaluating” # 正在评估上一步结果 PAUSED “paused” # 已暂停 COMPLETED “completed” # 计划全部完成 FAILED “failed” # 计划执行失败 class PlanStateMachine: 一个简单的计划状态机 def __init__(self): self._state AgentState.IDLE self._state_handlers {} # 存储状态进入时的回调函数 self._transition_history [] # 记录状态流转历史用于调试 property def state(self) - AgentState: return self._state def register_handler(self state: AgentState handler: Callable): 注册当进入某个状态时要执行的处理函数 self._state_handlers[state] handler def transition_to(self new_state: AgentState **kwargs): 执行状态转换 old_state self._state # 这里可以添加状态转换的校验逻辑例如某些状态不能直接跳到另一些状态 if self._can_transition(old_state new_state): print(f“[State Machine] {old_state.value} - {new_state.value}”) self._transition_history.append((old_state new_state kwargs)) self._state new_state # 触发新状态的处理函数 handler self._state_handlers.get(new_state) if handler: handler(**kwargs) else: raise ValueError(f“Invalid transition from {old_state.value} to {new_state.value}”) def _can_transition(self old: AgentState new: AgentState) - bool: 定义允许的状态转换规则这是安全执行的第一道关卡 # 示例规则只能从EXECUTING进入EVALUATING不能从PLANNING直接跳COMPLETED allowed_transitions { AgentState.IDLE: [AgentState.PLANNING] AgentState.PLANNING: [AgentState.EXECUTING AgentState.FAILED] AgentState.EXECUTING: [AgentState.EVALUATING AgentState.FAILED AgentState.PAUSED] AgentState.EVALUATING: [AgentState.EXECUTING AgentState.COMPLETED AgentState.FAILED AgentState.PLANNING] # 评估后可能重新规划 AgentState.PAUSED: [AgentState.EXECUTING AgentState.FAILED] } return new in allowed_transitions.get(old [])这个状态机虽然简单但定义了Agent执行计划的“生命周期法则”。_can_transition方法里的规则就是安全护栏防止状态乱跳比如不可能还没执行EXECUTING就直接评估EVALUATING。3.2 设计共享执行上下文计划中的各个任务不是孤立的它们需要传递数据。比如任务1输出的“机票价格”任务2需要用它来判断选择什么档次的酒店。我们需要一个全局的Context来保存这些信息。class ExecutionContext: 计划执行上下文用于在任务间共享数据 def __init__(self initial_goal: str): self.goal initial_goal self._storage {} # 键值对存储 self._execution_log [] # 执行日志 def set(self key: str value: Any): 存储数据 self._storage[key] value self._log(f“Context SET: {key} {value}”) def get(self key: str default: Any None) - Any: 获取数据 return self._storage.get(key default) def append_log(self task_id: str message: str level: str “INFO”): 记录执行日志 log_entry { “task_id”: task_id “timestamp”: time.time() “level”: level “message”: message } self._execution_log.append(log_entry) def _log(self message: str): # 简单的内部日志 print(f“[Context] {message}”) def get_summary(self) - dict: 获取上下文摘要可用于报告或后续决策 return { “goal”: self.goal “variables”: dict(self._storage) “log_entries”: len(self._execution_log) }上下文对象是Plan Mode的“记忆中枢”。在评估状态Evaluating中Agent可以查询上下文中的数据来做条件判断比如if context.get(‘flight_cost’ 0) 1000: # 执行廉价酒店方案。3.3 打造安全执行装饰器这是实现“安全执行”最精妙也最实用的一环。我们要为每一个具体的技能函数Skill——也就是那些真正执行预订、查询、发送邮件等操作的函数——套上一个“安全套”。这个“安全套”就是装饰器它能在函数执行前后进行拦截实施安全检查、权限验证、输入输出过滤、异常捕获和重试。import functools import time from typing import Dict Any def safe_execute(max_retries: int 2 timeout: int 30 allowed_exceptions: tuple (Exception) fallback_result: Any None): 安全执行装饰器工厂函数。 为技能函数添加重试、超时、异常捕获和降级逻辑。 def decorator(func): functools.wraps(func) def wrapper(context: ExecutionContext *args **kwargs) - Dict[str Any]: task_id kwargs.get(‘task_id’ func.__name__) last_exception None for attempt in range(1 max_retries 1): try: context.append_log(task_id f“Attempt {attempt}/{max_retries} started.”) # 1. 超时控制 # 注意这里使用简单的信号量或线程超时生产环境建议用asyncio.wait_for start_time time.time() result func(context *args **kwargs) elapsed time.time() - start_time if elapsed timeout: raise TimeoutError(f“Function {func.__name__} exceeded timeout {timeout}s”) # 2. 结果基础验证可选可扩展 if not isinstance(result dict): raise ValueError(f“Skill function must return a dict got {type(result)}”) if “success” not in result: result[“success”] True # 默认标记成功 context.append_log(task_id f“Attempt {attempt} succeeded in {elapsed:.2f}s.”) return result except allowed_exceptions as e: last_exception e context.append_log(task_id f“Attempt {attempt} failed: {str(e)}” “ERROR”) if attempt max_retries: wait_time 2 ** attempt # 指数退避 context.append_log(task_id f“Retrying after {wait_time}s...”) time.sleep(wait_time) continue except Exception as e: # 不允许的异常直接抛出 context.append_log(task_id f“Unallowed exception occurred: {str(e)}” “CRITICAL”) raise # 所有重试都失败 context.append_log(task_id f“All {max_retries} attempts failed. Using fallback.” “WARNING”) if fallback_result is not None: return {“success”: False “error”: str(last_exception) “data”: fallback_result “fallback_used”: True} else: return {“success”: False “error”: str(last_exception) “data”: None} return wrapper return decorator # 使用示例定义一个预订机票的技能函数 safe_execute(max_retries3 timeout60 fallback_result{“flight_id”: None}) def skill_book_flight(context: ExecutionContext from_city: str to_city: str **kwargs): 模拟预订机票技能 # 这里是真实的业务逻辑例如调用航空公司API # 为了演示我们模拟一个可能失败的操作 import random if random.random() 0.3: # 30%概率模拟失败 raise ConnectionError(“Flight booking API temporarily unavailable.”) # 模拟成功预订 flight_id f“FL{random.randint(1000 9999)}” flight_cost random.randint(500 2000) # 将重要结果存入上下文供后续任务使用 context.set(‘flight_id’ flight_id) context.set(‘flight_cost’ flight_cost) context.set(‘flight_route’ f“{from_city} - {to_city}”) return { “success”: True “data”: { “flight_id”: flight_id “cost”: flight_cost } “message”: “Flight booked successfully.” }这个safe_execute装饰器是Plan Mode的“安全执行官”。它确保了单个技能的执行是受控的、可观测的和有韧性的。通过参数化配置你可以为不同危险等级的技能设置不同的安全策略比如发送邮件的重试次数多删除文件的操作不允许重试。实操心得装饰器是Python实现AOP面向切面编程的利器非常适合用来为非功能性的“横切关注点”如日志、安全、重试添加统一逻辑。在设计时记得把context作为第一个参数传入技能函数这样技能内部就能读写共享上下文了。另外fallback_result降级结果的设置是提高系统可用性的关键它能让计划在部分失败时依然可以继续执行一个简化流程而不是整体崩溃。4. 规划与执行引擎的整合现在我们把状态机、上下文和技能组装起来构建一个完整的PlannerEngine。import json import asyncio from dataclasses import dataclass from typing import List Dict Any Optional dataclass class PlanTask: 计划中的一个任务节点定义 id: str description: str skill_name: str # 对应要调用的技能函数名 parameters: Dict[str Any] # 调用技能时传入的参数 dependencies: List[str] # 依赖的其他任务ID condition: Optional[str] None # 执行条件是一个可求值的字符串表达式如“context.get(‘budget’) 1000” on_success: Optional[str] None # 成功后的下一个任务ID on_failure: Optional[str] None # 失败后的下一个任务ID错误处理任务 class PlannerEngine: 规划与执行引擎 def __init__(self skills_registry: Dict[str callable]): self.sm PlanStateMachine() self.context None self.plan: List[PlanTask] [] self.current_task_index 0 self.skills skills_registry # 技能注册表 {‘skill_name’: function} # 注册状态处理函数 self.sm.register_handler(AgentState.PLANNING self._handle_planning) self.sm.register_handler(AgentState.EXECUTING self._handle_executing) self.sm.register_handler(AgentState.EVALUATING self._handle_evaluating) def set_goal(self goal: str): 设置初始目标并初始化上下文 self.context ExecutionContext(goal) self.sm.transition_to(AgentState.PLANNING) def _handle_planning(self): 规划状态处理调用LLM或规则生成计划 print(f“[Planner] Planning for goal: {self.context.goal}”) # 这里应该集成LLM调用。为简化我们假设从一个预设的规划器或模板生成计划。 # 例如可以根据goal关键词匹配一个预定义的JSON计划模板。 # 模拟生成一个简单的计划 self.plan [ PlanTask( id“task_1” description“预订北京到上海的机票” skill_name“book_flight” parameters{“from_city”: “北京” “to_city”: “上海”} dependencies[] conditionNone on_success“task_2” on_failure“task_fail_1” ) PlanTask( id“task_2” description“预订上海酒店” skill_name“book_hotel” parameters{“city”: “上海” “days”: 3} dependencies[“task_1”] condition“context.get(‘flight_booked’) True” # 假设task_1会设置这个值 on_success“task_3” on_failure“task_fail_2” ) PlanTask(id“task_3” description“生成出差清单” skill_name“generate_checklist” parameters{} dependencies[“task_1” “task_2”]) ] self.context.append_log(“planner” “Plan generated with 3 tasks.”) # 规划完成进入执行状态 self.sm.transition_to(AgentState.EXECUTING) def _handle_executing(self): 执行状态处理执行当前任务 if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.COMPLETED) return current_task self.plan[self.current_task_index] # 1. 检查依赖是否全部满足 for dep_id in current_task.dependencies: # 需要检查依赖任务是否成功完成。这里简化处理假设依赖任务按顺序执行且成功。 # 实际中需要在上下文中记录每个任务的状态。 pass # 2. 检查执行条件如果有 if current_task.condition: try: # 警告使用eval有安全风险仅作演示。生产环境应使用安全的表达式求值库如 asteval。 condition_met eval(current_task.condition {“context”: self.context}) if not condition_met: print(f“[Planner] Task {current_task.id} condition not met. Skipping.”) self._move_to_next_task(skippedTrue) return except Exception as e: self.context.append_log(current_task.id f“Error evaluating condition: {e}” “ERROR”) # 条件求值失败视为不满足跳过或按失败处理 self._move_to_next_task(skippedTrue) return print(f“[Planner] Executing task: {current_task.id} - {current_task.description}”) # 3. 查找并执行技能 skill_func self.skills.get(current_task.skill_name) if not skill_func: error_msg f“Skill ‘{current_task.skill_name}’ not found.” self.context.append_log(current_task.id error_msg “ERROR”) self._task_failed(current_task error_msg) return # 执行技能这里假设是同步函数异步环境需用await try: # 将任务ID和上下文传入技能 result skill_func(self.context **current_task.parameters task_idcurrent_task.id) # 将执行结果暂存供评估状态使用 self._last_execution_result {“task”: current_task “result”: result} # 进入评估状态 self.sm.transition_to(AgentState.EVALUATING) except Exception as e: self._task_failed(current_task str(e)) def _handle_evaluating(self): 评估状态处理评估上一个任务的执行结果并决定下一步 task self._last_execution_result[“task”] result self._last_execution_result[“result”] print(f“[Planner] Evaluating task {task.id}. Success: {result.get(‘success’)}”) if result.get(‘success’): # 任务成功 self.context.append_log(task.id “Task completed successfully.”) # 根据任务配置的 on_success 跳转默认为下一个顺序任务 next_task_id task.on_success self._resolve_next_task(task.id next_task_id successTrue) else: # 任务失败 self.context.append_log(task.id f“Task failed: {result.get(‘error’)}” “ERROR”) # 根据任务配置的 on_failure 跳转 next_task_id task.on_failure self._resolve_next_task(task.id next_task_id successFalse) def _resolve_next_task(self current_task_id: str next_task_id: Optional[str] success: bool): 决定下一个要执行的任务ID并更新索引 if next_task_id: # 跳转到指定任务 for i t in enumerate(self.plan): if t.id next_task_id: self.current_task_index i self.sm.transition_to(AgentState.EXECUTING) return # 没找到指定任务按顺序执行下一个 print(f“[Planner] Next task ‘{next_task_id}’ not found. Proceeding sequentially.”) # 默认行为执行计划中的下一个任务 self.current_task_index 1 if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.EXECUTING) else: self.sm.transition_to(AgentState.COMPLETED) def _move_to_next_task(self skippedFalse): 移动到下一个顺序任务 self.current_task_index 1 if skipped: self.context.append_log(“planner” f“Task {self.plan[self.current_task_index-1].id} was skipped.”) if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.EXECUTING) else: self.sm.transition_to(AgentState.COMPLETED) def _task_failed(self task: PlanTask error: str): 处理任务失败 self.context.append_log(task.id f“Task execution failed: {error}” “CRITICAL”) # 这里可以触发更复杂的错误处理流程比如重试整个计划、通知用户等。 # 简单起见我们跳转到失败处理任务或直接标记计划失败。 if task.on_failure: self._resolve_next_task(task.id task.on_failure successFalse) else: # 没有配置失败处理整个计划失败 self.sm.transition_to(AgentState.FAILED) def run(self): 启动引擎简化同步版本 if self.sm.state ! AgentState.PLANNING: print(“Engine not in PLANNING state. Call set_goal() first.”) return # 这是一个简化的同步驱动循环实际中可能需要异步事件循环 while self.sm.state not in [AgentState.COMPLETED AgentState.FAILED AgentState.PAUSED]: # 状态处理函数已在状态转换时被触发这里只需驱动循环。 # 更复杂的实现需要处理异步IO和事件。 time.sleep(0.1) # 防止CPU空转 final_state self.sm.state print(f“[Planner] Plan finished with state: {final_state.value}”) print(f“[Planner] Context Summary: {json.dumps(self.context.get_summary() indent2 defaultstr)}”) return final_state这个PlannerEngine将之前分散的组件串联了起来。它从规划开始驱动状态机流转在EXECUTING状态调用被安全装饰器包裹的技能在EVALUATING状态根据结果决定下一步形成了一个完整的闭环。5. 实战演示与问题排查让我们用一个简单的例子把上面的代码跑起来并看看实际运行中会遇到哪些典型问题。5.1 一个完整的执行示例# 1. 注册技能 skills_registry { “book_flight”: skill_book_flight # 使用前面定义的安全装饰器技能 “book_hotel”: safe_execute(max_retries2)(lambda ctx **kwargs: {“success”: True “data”: {“hotel_id”: “HOTEL123”}}) # 一个简单的模拟酒店技能 “generate_checklist”: safe_execute()(lambda ctx **kwargs: {“success”: True “data”: {“checklist”: [“Ticket” “Hotel” “PPT”]}}) } # 2. 初始化引擎 engine PlannerEngine(skills_registry) # 3. 设置目标启动规划 engine.set_goal(“安排北京至上海出差”) # 4. 运行引擎 final_state engine.run() print(f“\n 计划执行结束 ) print(f“最终状态: {final_state.value}”)运行这段代码你会在控制台看到类似如下的输出清晰地展示了状态流转和任务执行过程[State Machine] idle - planning [Planner] Planning for goal: 安排北京至上海出差 [Context] Context SET: flight_id FL5678 [Context] Context SET: flight_cost 1200 [Context] Context SET: flight_route 北京 - 上海 [Planner] Evaluating task task_1. Success: True [State Machine] evaluating - executing [Planner] Executing task: task_2 - 预订上海酒店 ... [Planner] Plan finished with state: completed5.2 常见问题与排查技巧实录在实际开发和运行中你一定会遇到各种问题。下面是我在构建这类系统时踩过的一些坑和总结的技巧。问题1计划陷入死循环或状态卡住现象Agent一直停留在EXECUTING或EVALUATING状态没有进展。排查检查状态转换规则首先确认_can_transition函数中的规则是否允许当前状态转到下一个预期状态。比如EVALUATING后是否允许转回EXECUTING。检查任务依赖图打印出当前的计划任务列表和current_task_index。检查是否存在循环依赖A依赖BB又依赖A或者on_success/on_failure指向了一个不存在的任务ID。检查条件表达式如果任务有condition确保它在当前上下文中能被正确求值。使用print或日志输出condition字符串和求值结果。切记生产环境绝对不要用eval改用asteval等安全库。查看技能执行结果确认技能函数是否正常返回了包含success键的字典。如果技能抛出了未被装饰器捕获的异常或者返回格式不对引擎可能无法正确处理。问题2上下文数据丢失或污染现象任务B读取不到任务A设置的数据或者读到了错误的值。排查统一键名建立团队规范对存入上下文的变量键名进行统一管理避免拼写错误。例如使用常量定义FLIGHT_COST_KEY ‘flight_cost’。作用域隔离对于复杂的、多线程/异步执行的计划考虑为不同的执行分支子计划创建上下文的子副本或命名空间防止数据意外覆盖。日志追踪在context.set()和context.get()时增加详细日志记录操作的任务ID和时间戳。这样可以在日志中清晰地追溯数据的生命周期。问题3技能执行超时或资源泄漏现象某个技能如网络请求执行时间过长拖垮整个Agent或者数据库连接未关闭。解决强化装饰器在safe_execute装饰器中除了timeout参数还可以集成资源管理。例如使用with语句或try-finally确保资源如文件句柄、网络连接被正确释放。异步化对于IO密集型技能将其改造成异步函数并使用asyncio.wait_for来实现更精确的超时控制。这样在等待时不会阻塞整个事件循环。设置全局超时在PlannerEngine层面设置一个整个计划的最大执行时长超时后强制将状态转为FAILED或PAUSED。问题4LLM生成的计划质量不稳定现象规划阶段LLM输出的计划JSON格式错误、逻辑矛盾或步骤不可执行。解决结构化提示词给LLM的提示词Prompt要极度结构化明确要求输出格式并提供高质量示例Few-shot。例如“请严格按照以下JSON格式输出计划包含id description skill_name...”后置校验与修复在_handle_planning方法中添加一个计划校验环节。使用一个轻量级的校验函数或规则引擎检查计划的基本合法性如无环依赖、技能名存在等。对于简单错误可以尝试自动修复对于复杂错误可以触发重新规划或直接报错。分层规划不要指望LLM一次生成完美计划。可以采用“目标-子目标-具体任务”的分层规划方式。先让LLM生成高级别的子目标序列然后对每个子目标再调用LLM或规则引擎生成具体任务。避坑技巧在开发初期一定要实现一个详细的可视化调试工具。可以是一个简单的命令行打印将当前状态、上下文内容、任务队列以清晰的方式实时输出。这比查看分散的日志高效十倍。更进一步可以生成状态机的图表.dot文件用Graphviz渲染或任务流程图直观展示执行路径这对排查复杂分支逻辑的问题至关重要。6. 进阶思考与扩展方向实现了一个基础的Plan Mode之后你可以根据实际需求从以下几个方向进行深化和扩展打造更强大、更专业的Agent。1. 集成真正的LLM进行动态规划我们示例中的规划是静态的。真正的智能体现在动态规划上。你可以在_handle_planning方法中集成OpenAI、Claude或本地LLM的API。提示词Prompt需要精心设计包含用户目标、可用技能列表、当前上下文、以及之前已执行的任务和结果。让LLM决定下一步做什么甚至重新规划剩余步骤。这就是ReAct (Reasoning Acting)或Chain of Thought (CoT)模式在Agent中的具体应用。2. 实现更复杂的流程控制模式当前引擎主要支持顺序和条件跳转。你可以扩展PlanTask模型和引擎逻辑来实现并行执行对于没有依赖关系的任务可以并发执行以提高效率。需要引入异步编程和任务队列。循环Loop为任务节点添加loop_condition属性当条件满足时重复执行该任务或其子计划。子计划Subplan将一个任务节点指向另一个完整的计划Plan实现计划的嵌套和模块化复用。3. 完善持久化与可观测性生产级的Agent必须能够中断后恢复并且状态可追溯。持久化将ExecutionContext和PlannerEngine的状态当前任务索引、状态机状态等定期序列化如用Pickle或JSON存储到数据库或文件。当Agent重启时可以加载状态从中断点继续执行。可观测性除了基础的日志可以将关键的执行事件状态转换、任务开始/结束、上下文变更推送到监控系统如Prometheus、ELK。为每个计划生成唯一的execution_id便于全链路追踪。4. 构建技能市场与动态加载将技能Skill抽象成独立的、可插拔的模块。维护一个技能注册中心Agent在规划时可以从中心查询可用的技能及其描述、参数schema。技能可以实现动态加载如通过importlib这样无需重启Agent就能扩展其能力。5. 探索多Agent协作当一个任务过于复杂时可以将其拆解分配给多个具有不同专长的子Agent去执行。主Agent的Plan Mode负责顶层协调和任务分发子Agent内部也有自己的规划循环。这就需要定义Agent间的通信协议如通过消息队列或共享存储和协同机制。打造一个健壮的Plan Mode是AI Agent开发从玩具走向实用的关键一步。它不仅仅是几行状态机代码更是一套关于如何让智能体在复杂、不确定的环境中可靠工作的工程学思想。从简单的三段式状态机出发不断迭代和丰富其能力你的Agent才能真正胜任那些“复杂任务”并确保整个过程是“安全”的。 AI Agent规划模式：基于状态机与安全装饰器的复杂任务执行引擎
1. 项目概述为什么我们需要Plan Mode如果你已经跟着这个系列从零开始搭建过基础的AI Agent可能会发现一个棘手的问题当任务稍微复杂一点比如“帮我查一下天气如果下雨就提醒我带伞然后顺便看看明天的会议安排”我们之前那种线性的、一步接一步的执行模式就开始捉襟见肘了。Agent要么会卡在某个步骤要么会忽略掉条件判断一股脑地把所有动作都执行了。这就像让一个新手司机不看导航、不顾红绿灯只凭感觉开车出事是迟早的。Plan Mode或者说规划模式就是为了解决这个“复杂任务的安全执行”问题而生的核心机制。简单来说Plan Mode让我们的AI Agent从一个只会执行简单指令的“实习生”升级为一个懂得规划、判断、回溯的“项目主管”。它不再是被动地响应单个指令而是主动地将一个复杂的、模糊的用户目标Goal拆解成一个清晰的、可执行的行动计划Plan并监督这个计划一步步安全地落地。这里面的关键词是“复杂”和“安全”。复杂意味着任务包含分支、循环、条件判断甚至异常处理安全则意味着执行过程可控、可预测、可中断不会因为一个步骤的失败导致整个系统崩溃或者做出危险操作。从最近大家搜索的热词也能看出无论是“状态机”、“三段式状态机”还是“安全执行要求”都指向了同一个核心如何为AI Agent赋予一个可靠的中枢神经系统。这个系统需要理解任务的结构状态机管理执行的生命周期规划、执行、检查并确保每一步都在预设的安全边界内安全执行。这也是为什么像“Harness”这样的基础设施层概念会被频繁提及——它就像宇航员的舱外航天服不代替宇航员Agent做决策但为其在复杂危险环境真实世界中的行动提供生命保障。所以这篇文章我们就来深入这个Agent的“驾驶舱”看看如何亲手打造一个属于你自己的Plan Mode。我们会从最经典的状态机模型入手用Python一步步实现一个具备任务分解、条件执行和错误处理能力的规划引擎。你会发现它不仅是Agent的核心其设计思想也能广泛应用在你日常开发的任何需要流程控制的场景中。2. 核心架构设计状态机与规划引擎要理解Plan Mode首先得理解它的理论基础有限状态机Finite-State Machine, FSM。别被这个名字吓到你可以把它想象成一个智能灯的开关面板。这个灯有“关闭”、“开启”、“闪烁”几种状态。你按“开”按钮它从“关闭”状态切换到“开启”状态再按“模式”按钮它可能从“开启”进入“闪烁”状态。任何时刻灯都处于且仅处于一种明确的状态而触发状态切换的就是你的按键指令事件。对于AI Agent的Plan Mode而言这个“状态机”管理的是任务执行的生命周期状态。一个经典且实用的模型是三段式状态机它通常包含以下状态规划状态PlanningAgent接收到一个复杂目标如“安排出差”。在此状态下Agent的核心工作是调用LLM进行任务分解和规划。输入是用户目标输出是一个结构化的计划通常可以表示为任务列表或流程图。例如计划可能是[预订机票 预订酒店 准备出差材料]。这个计划本身可能还带有依赖关系必须先订票才能知道时间订酒店和条件判断如果预算超支则选择更便宜的酒店。执行状态Executing计划生成后Agent进入执行状态。在此状态下Agent按照规划好的步骤依次调用相应的工具Skills或子Agent来完成任务。比如调用“机票预订工具”执行“预订机票”这个子任务。每执行完一步都会产生一个结果成功、失败、附带数据。检查状态Evaluating/Checking这是确保“安全执行”的关键环节。在一个子任务执行后系统不会盲目进入下一步而是进入检查状态。在此状态下Agent会评估上一步的执行结果是否成功返回的数据是否符合预期是否触发了某个条件如“预算超支”根据评估结果决定下一步是继续执行下一个任务、跳转到另一个分支、重试当前任务还是整体失败并进入错误处理流程。这三个状态循环往复构成了Plan Mode的主干。而驱动状态流转的“事件”就是子任务的完成信号、定时器超时或者用户的中断指令。基于这个模型我们可以设计一个规划引擎Planner Engine作为Plan Mode的大脑。它的职责是解析与生成计划将用户目标转化为可操作的计划结构。计划可以用JSON、YAML或自定义的DSL领域特定语言描述其中明确标出任务节点、依赖关系和条件分支。状态管理维护当前执行到了哪个节点整个计划处于什么状态进行中、已完成、已失败、已暂停。调度与决策根据当前状态和执行结果决定下一个要执行的任务是什么。这涉及到简单的顺序执行以及复杂的条件分支和循环判断。上下文管理在整个计划执行过程中维护一个共享的上下文Context。这个上下文就像一块公共白板记录了所有已执行任务的结果数据如机票订单号ABC123酒店入住日期2023-10-27供后续任务读取和判断。一个常见的计划结构描述可能长这样JSON格式{ “goal”: “安排一次北京至上海的出差” “plan”: [ { “id”: “task_1” “description”: “查询并预订北京到上海的机票” “skill”: “book_flight_ticket” “dependencies”: [] “on_success”: “task_2” “on_failure”: “handle_flight_failure” } { “id”: “task_2” “description”: “根据机票日期预订上海酒店” “skill”: “book_hotel” “dependencies”: [“task_1”] “condition”: “context.get(‘flight_booked’) True” “on_success”: “task_3” “on_failure”: “handle_hotel_failure” } { “id”: “task_3” “description”: “生成出差事项清单” “skill”: “generate_checklist” “dependencies”: [“task_1” “task_2”] } ] }这个引擎的实现可以选择纯代码编写也可以利用现有的状态机库如Python的transitions。但对于学习和深度定制而言我强烈建议从零开始实现一个轻量版这能让你透彻理解每一个流转细节。注意在设计规划引擎时一个重要的考量是“规划”的时机。有两种主要策略静态规划和动态重规划。静态规划在开始时生成完整计划之后严格按计划执行效率高但灵活性差。动态重规划则在每一步执行后都重新评估剩余计划甚至重新调用LLM生成后续步骤灵活性极高但耗时且可能产生不一致。对于大多数确定性较强的复杂任务我推荐采用“主计划静态生成局部条件动态调整”的混合策略。3. 核心组件实现状态机、上下文与安全装饰器理论说完了我们开始动手写代码。我们将用Python构建一个最小可行但功能完整的Plan Mode核心。这个核心主要包含三个部分状态机引擎、共享上下文管理器和一个确保安全执行的“防护网”——装饰器。3.1 实现一个轻量级状态机我们首先实现一个简单的状态机来管理PlanningExecutingEvaluating这三个核心状态。from enum import Enum from typing import Any Callable Optional class AgentState(Enum): Agent计划执行状态枚举 IDLE “idle” # 空闲等待目标 PLANNING “planning” # 正在规划 EXECUTING “executing” # 正在执行子任务 EVALUATING “evaluating” # 正在评估上一步结果 PAUSED “paused” # 已暂停 COMPLETED “completed” # 计划全部完成 FAILED “failed” # 计划执行失败 class PlanStateMachine: 一个简单的计划状态机 def __init__(self): self._state AgentState.IDLE self._state_handlers {} # 存储状态进入时的回调函数 self._transition_history [] # 记录状态流转历史用于调试 property def state(self) - AgentState: return self._state def register_handler(self state: AgentState handler: Callable): 注册当进入某个状态时要执行的处理函数 self._state_handlers[state] handler def transition_to(self new_state: AgentState **kwargs): 执行状态转换 old_state self._state # 这里可以添加状态转换的校验逻辑例如某些状态不能直接跳到另一些状态 if self._can_transition(old_state new_state): print(f“[State Machine] {old_state.value} - {new_state.value}”) self._transition_history.append((old_state new_state kwargs)) self._state new_state # 触发新状态的处理函数 handler self._state_handlers.get(new_state) if handler: handler(**kwargs) else: raise ValueError(f“Invalid transition from {old_state.value} to {new_state.value}”) def _can_transition(self old: AgentState new: AgentState) - bool: 定义允许的状态转换规则这是安全执行的第一道关卡 # 示例规则只能从EXECUTING进入EVALUATING不能从PLANNING直接跳COMPLETED allowed_transitions { AgentState.IDLE: [AgentState.PLANNING] AgentState.PLANNING: [AgentState.EXECUTING AgentState.FAILED] AgentState.EXECUTING: [AgentState.EVALUATING AgentState.FAILED AgentState.PAUSED] AgentState.EVALUATING: [AgentState.EXECUTING AgentState.COMPLETED AgentState.FAILED AgentState.PLANNING] # 评估后可能重新规划 AgentState.PAUSED: [AgentState.EXECUTING AgentState.FAILED] } return new in allowed_transitions.get(old [])这个状态机虽然简单但定义了Agent执行计划的“生命周期法则”。_can_transition方法里的规则就是安全护栏防止状态乱跳比如不可能还没执行EXECUTING就直接评估EVALUATING。3.2 设计共享执行上下文计划中的各个任务不是孤立的它们需要传递数据。比如任务1输出的“机票价格”任务2需要用它来判断选择什么档次的酒店。我们需要一个全局的Context来保存这些信息。class ExecutionContext: 计划执行上下文用于在任务间共享数据 def __init__(self initial_goal: str): self.goal initial_goal self._storage {} # 键值对存储 self._execution_log [] # 执行日志 def set(self key: str value: Any): 存储数据 self._storage[key] value self._log(f“Context SET: {key} {value}”) def get(self key: str default: Any None) - Any: 获取数据 return self._storage.get(key default) def append_log(self task_id: str message: str level: str “INFO”): 记录执行日志 log_entry { “task_id”: task_id “timestamp”: time.time() “level”: level “message”: message } self._execution_log.append(log_entry) def _log(self message: str): # 简单的内部日志 print(f“[Context] {message}”) def get_summary(self) - dict: 获取上下文摘要可用于报告或后续决策 return { “goal”: self.goal “variables”: dict(self._storage) “log_entries”: len(self._execution_log) }上下文对象是Plan Mode的“记忆中枢”。在评估状态Evaluating中Agent可以查询上下文中的数据来做条件判断比如if context.get(‘flight_cost’ 0) 1000: # 执行廉价酒店方案。3.3 打造安全执行装饰器这是实现“安全执行”最精妙也最实用的一环。我们要为每一个具体的技能函数Skill——也就是那些真正执行预订、查询、发送邮件等操作的函数——套上一个“安全套”。这个“安全套”就是装饰器它能在函数执行前后进行拦截实施安全检查、权限验证、输入输出过滤、异常捕获和重试。import functools import time from typing import Dict Any def safe_execute(max_retries: int 2 timeout: int 30 allowed_exceptions: tuple (Exception) fallback_result: Any None): 安全执行装饰器工厂函数。 为技能函数添加重试、超时、异常捕获和降级逻辑。 def decorator(func): functools.wraps(func) def wrapper(context: ExecutionContext *args **kwargs) - Dict[str Any]: task_id kwargs.get(‘task_id’ func.__name__) last_exception None for attempt in range(1 max_retries 1): try: context.append_log(task_id f“Attempt {attempt}/{max_retries} started.”) # 1. 超时控制 # 注意这里使用简单的信号量或线程超时生产环境建议用asyncio.wait_for start_time time.time() result func(context *args **kwargs) elapsed time.time() - start_time if elapsed timeout: raise TimeoutError(f“Function {func.__name__} exceeded timeout {timeout}s”) # 2. 结果基础验证可选可扩展 if not isinstance(result dict): raise ValueError(f“Skill function must return a dict got {type(result)}”) if “success” not in result: result[“success”] True # 默认标记成功 context.append_log(task_id f“Attempt {attempt} succeeded in {elapsed:.2f}s.”) return result except allowed_exceptions as e: last_exception e context.append_log(task_id f“Attempt {attempt} failed: {str(e)}” “ERROR”) if attempt max_retries: wait_time 2 ** attempt # 指数退避 context.append_log(task_id f“Retrying after {wait_time}s...”) time.sleep(wait_time) continue except Exception as e: # 不允许的异常直接抛出 context.append_log(task_id f“Unallowed exception occurred: {str(e)}” “CRITICAL”) raise # 所有重试都失败 context.append_log(task_id f“All {max_retries} attempts failed. Using fallback.” “WARNING”) if fallback_result is not None: return {“success”: False “error”: str(last_exception) “data”: fallback_result “fallback_used”: True} else: return {“success”: False “error”: str(last_exception) “data”: None} return wrapper return decorator # 使用示例定义一个预订机票的技能函数 safe_execute(max_retries3 timeout60 fallback_result{“flight_id”: None}) def skill_book_flight(context: ExecutionContext from_city: str to_city: str **kwargs): 模拟预订机票技能 # 这里是真实的业务逻辑例如调用航空公司API # 为了演示我们模拟一个可能失败的操作 import random if random.random() 0.3: # 30%概率模拟失败 raise ConnectionError(“Flight booking API temporarily unavailable.”) # 模拟成功预订 flight_id f“FL{random.randint(1000 9999)}” flight_cost random.randint(500 2000) # 将重要结果存入上下文供后续任务使用 context.set(‘flight_id’ flight_id) context.set(‘flight_cost’ flight_cost) context.set(‘flight_route’ f“{from_city} - {to_city}”) return { “success”: True “data”: { “flight_id”: flight_id “cost”: flight_cost } “message”: “Flight booked successfully.” }这个safe_execute装饰器是Plan Mode的“安全执行官”。它确保了单个技能的执行是受控的、可观测的和有韧性的。通过参数化配置你可以为不同危险等级的技能设置不同的安全策略比如发送邮件的重试次数多删除文件的操作不允许重试。实操心得装饰器是Python实现AOP面向切面编程的利器非常适合用来为非功能性的“横切关注点”如日志、安全、重试添加统一逻辑。在设计时记得把context作为第一个参数传入技能函数这样技能内部就能读写共享上下文了。另外fallback_result降级结果的设置是提高系统可用性的关键它能让计划在部分失败时依然可以继续执行一个简化流程而不是整体崩溃。4. 规划与执行引擎的整合现在我们把状态机、上下文和技能组装起来构建一个完整的PlannerEngine。import json import asyncio from dataclasses import dataclass from typing import List Dict Any Optional dataclass class PlanTask: 计划中的一个任务节点定义 id: str description: str skill_name: str # 对应要调用的技能函数名 parameters: Dict[str Any] # 调用技能时传入的参数 dependencies: List[str] # 依赖的其他任务ID condition: Optional[str] None # 执行条件是一个可求值的字符串表达式如“context.get(‘budget’) 1000” on_success: Optional[str] None # 成功后的下一个任务ID on_failure: Optional[str] None # 失败后的下一个任务ID错误处理任务 class PlannerEngine: 规划与执行引擎 def __init__(self skills_registry: Dict[str callable]): self.sm PlanStateMachine() self.context None self.plan: List[PlanTask] [] self.current_task_index 0 self.skills skills_registry # 技能注册表 {‘skill_name’: function} # 注册状态处理函数 self.sm.register_handler(AgentState.PLANNING self._handle_planning) self.sm.register_handler(AgentState.EXECUTING self._handle_executing) self.sm.register_handler(AgentState.EVALUATING self._handle_evaluating) def set_goal(self goal: str): 设置初始目标并初始化上下文 self.context ExecutionContext(goal) self.sm.transition_to(AgentState.PLANNING) def _handle_planning(self): 规划状态处理调用LLM或规则生成计划 print(f“[Planner] Planning for goal: {self.context.goal}”) # 这里应该集成LLM调用。为简化我们假设从一个预设的规划器或模板生成计划。 # 例如可以根据goal关键词匹配一个预定义的JSON计划模板。 # 模拟生成一个简单的计划 self.plan [ PlanTask( id“task_1” description“预订北京到上海的机票” skill_name“book_flight” parameters{“from_city”: “北京” “to_city”: “上海”} dependencies[] conditionNone on_success“task_2” on_failure“task_fail_1” ) PlanTask( id“task_2” description“预订上海酒店” skill_name“book_hotel” parameters{“city”: “上海” “days”: 3} dependencies[“task_1”] condition“context.get(‘flight_booked’) True” # 假设task_1会设置这个值 on_success“task_3” on_failure“task_fail_2” ) PlanTask(id“task_3” description“生成出差清单” skill_name“generate_checklist” parameters{} dependencies[“task_1” “task_2”]) ] self.context.append_log(“planner” “Plan generated with 3 tasks.”) # 规划完成进入执行状态 self.sm.transition_to(AgentState.EXECUTING) def _handle_executing(self): 执行状态处理执行当前任务 if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.COMPLETED) return current_task self.plan[self.current_task_index] # 1. 检查依赖是否全部满足 for dep_id in current_task.dependencies: # 需要检查依赖任务是否成功完成。这里简化处理假设依赖任务按顺序执行且成功。 # 实际中需要在上下文中记录每个任务的状态。 pass # 2. 检查执行条件如果有 if current_task.condition: try: # 警告使用eval有安全风险仅作演示。生产环境应使用安全的表达式求值库如 asteval。 condition_met eval(current_task.condition {“context”: self.context}) if not condition_met: print(f“[Planner] Task {current_task.id} condition not met. Skipping.”) self._move_to_next_task(skippedTrue) return except Exception as e: self.context.append_log(current_task.id f“Error evaluating condition: {e}” “ERROR”) # 条件求值失败视为不满足跳过或按失败处理 self._move_to_next_task(skippedTrue) return print(f“[Planner] Executing task: {current_task.id} - {current_task.description}”) # 3. 查找并执行技能 skill_func self.skills.get(current_task.skill_name) if not skill_func: error_msg f“Skill ‘{current_task.skill_name}’ not found.” self.context.append_log(current_task.id error_msg “ERROR”) self._task_failed(current_task error_msg) return # 执行技能这里假设是同步函数异步环境需用await try: # 将任务ID和上下文传入技能 result skill_func(self.context **current_task.parameters task_idcurrent_task.id) # 将执行结果暂存供评估状态使用 self._last_execution_result {“task”: current_task “result”: result} # 进入评估状态 self.sm.transition_to(AgentState.EVALUATING) except Exception as e: self._task_failed(current_task str(e)) def _handle_evaluating(self): 评估状态处理评估上一个任务的执行结果并决定下一步 task self._last_execution_result[“task”] result self._last_execution_result[“result”] print(f“[Planner] Evaluating task {task.id}. Success: {result.get(‘success’)}”) if result.get(‘success’): # 任务成功 self.context.append_log(task.id “Task completed successfully.”) # 根据任务配置的 on_success 跳转默认为下一个顺序任务 next_task_id task.on_success self._resolve_next_task(task.id next_task_id successTrue) else: # 任务失败 self.context.append_log(task.id f“Task failed: {result.get(‘error’)}” “ERROR”) # 根据任务配置的 on_failure 跳转 next_task_id task.on_failure self._resolve_next_task(task.id next_task_id successFalse) def _resolve_next_task(self current_task_id: str next_task_id: Optional[str] success: bool): 决定下一个要执行的任务ID并更新索引 if next_task_id: # 跳转到指定任务 for i t in enumerate(self.plan): if t.id next_task_id: self.current_task_index i self.sm.transition_to(AgentState.EXECUTING) return # 没找到指定任务按顺序执行下一个 print(f“[Planner] Next task ‘{next_task_id}’ not found. Proceeding sequentially.”) # 默认行为执行计划中的下一个任务 self.current_task_index 1 if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.EXECUTING) else: self.sm.transition_to(AgentState.COMPLETED) def _move_to_next_task(self skippedFalse): 移动到下一个顺序任务 self.current_task_index 1 if skipped: self.context.append_log(“planner” f“Task {self.plan[self.current_task_index-1].id} was skipped.”) if self.current_task_index len(self.plan): self.sm.transition_to(AgentState.EXECUTING) else: self.sm.transition_to(AgentState.COMPLETED) def _task_failed(self task: PlanTask error: str): 处理任务失败 self.context.append_log(task.id f“Task execution failed: {error}” “CRITICAL”) # 这里可以触发更复杂的错误处理流程比如重试整个计划、通知用户等。 # 简单起见我们跳转到失败处理任务或直接标记计划失败。 if task.on_failure: self._resolve_next_task(task.id task.on_failure successFalse) else: # 没有配置失败处理整个计划失败 self.sm.transition_to(AgentState.FAILED) def run(self): 启动引擎简化同步版本 if self.sm.state ! AgentState.PLANNING: print(“Engine not in PLANNING state. Call set_goal() first.”) return # 这是一个简化的同步驱动循环实际中可能需要异步事件循环 while self.sm.state not in [AgentState.COMPLETED AgentState.FAILED AgentState.PAUSED]: # 状态处理函数已在状态转换时被触发这里只需驱动循环。 # 更复杂的实现需要处理异步IO和事件。 time.sleep(0.1) # 防止CPU空转 final_state self.sm.state print(f“[Planner] Plan finished with state: {final_state.value}”) print(f“[Planner] Context Summary: {json.dumps(self.context.get_summary() indent2 defaultstr)}”) return final_state这个PlannerEngine将之前分散的组件串联了起来。它从规划开始驱动状态机流转在EXECUTING状态调用被安全装饰器包裹的技能在EVALUATING状态根据结果决定下一步形成了一个完整的闭环。5. 实战演示与问题排查让我们用一个简单的例子把上面的代码跑起来并看看实际运行中会遇到哪些典型问题。5.1 一个完整的执行示例# 1. 注册技能 skills_registry { “book_flight”: skill_book_flight # 使用前面定义的安全装饰器技能 “book_hotel”: safe_execute(max_retries2)(lambda ctx **kwargs: {“success”: True “data”: {“hotel_id”: “HOTEL123”}}) # 一个简单的模拟酒店技能 “generate_checklist”: safe_execute()(lambda ctx **kwargs: {“success”: True “data”: {“checklist”: [“Ticket” “Hotel” “PPT”]}}) } # 2. 初始化引擎 engine PlannerEngine(skills_registry) # 3. 设置目标启动规划 engine.set_goal(“安排北京至上海出差”) # 4. 运行引擎 final_state engine.run() print(f“\n 计划执行结束 ) print(f“最终状态: {final_state.value}”)运行这段代码你会在控制台看到类似如下的输出清晰地展示了状态流转和任务执行过程[State Machine] idle - planning [Planner] Planning for goal: 安排北京至上海出差 [Context] Context SET: flight_id FL5678 [Context] Context SET: flight_cost 1200 [Context] Context SET: flight_route 北京 - 上海 [Planner] Evaluating task task_1. Success: True [State Machine] evaluating - executing [Planner] Executing task: task_2 - 预订上海酒店 ... [Planner] Plan finished with state: completed5.2 常见问题与排查技巧实录在实际开发和运行中你一定会遇到各种问题。下面是我在构建这类系统时踩过的一些坑和总结的技巧。问题1计划陷入死循环或状态卡住现象Agent一直停留在EXECUTING或EVALUATING状态没有进展。排查检查状态转换规则首先确认_can_transition函数中的规则是否允许当前状态转到下一个预期状态。比如EVALUATING后是否允许转回EXECUTING。检查任务依赖图打印出当前的计划任务列表和current_task_index。检查是否存在循环依赖A依赖BB又依赖A或者on_success/on_failure指向了一个不存在的任务ID。检查条件表达式如果任务有condition确保它在当前上下文中能被正确求值。使用print或日志输出condition字符串和求值结果。切记生产环境绝对不要用eval改用asteval等安全库。查看技能执行结果确认技能函数是否正常返回了包含success键的字典。如果技能抛出了未被装饰器捕获的异常或者返回格式不对引擎可能无法正确处理。问题2上下文数据丢失或污染现象任务B读取不到任务A设置的数据或者读到了错误的值。排查统一键名建立团队规范对存入上下文的变量键名进行统一管理避免拼写错误。例如使用常量定义FLIGHT_COST_KEY ‘flight_cost’。作用域隔离对于复杂的、多线程/异步执行的计划考虑为不同的执行分支子计划创建上下文的子副本或命名空间防止数据意外覆盖。日志追踪在context.set()和context.get()时增加详细日志记录操作的任务ID和时间戳。这样可以在日志中清晰地追溯数据的生命周期。问题3技能执行超时或资源泄漏现象某个技能如网络请求执行时间过长拖垮整个Agent或者数据库连接未关闭。解决强化装饰器在safe_execute装饰器中除了timeout参数还可以集成资源管理。例如使用with语句或try-finally确保资源如文件句柄、网络连接被正确释放。异步化对于IO密集型技能将其改造成异步函数并使用asyncio.wait_for来实现更精确的超时控制。这样在等待时不会阻塞整个事件循环。设置全局超时在PlannerEngine层面设置一个整个计划的最大执行时长超时后强制将状态转为FAILED或PAUSED。问题4LLM生成的计划质量不稳定现象规划阶段LLM输出的计划JSON格式错误、逻辑矛盾或步骤不可执行。解决结构化提示词给LLM的提示词Prompt要极度结构化明确要求输出格式并提供高质量示例Few-shot。例如“请严格按照以下JSON格式输出计划包含id description skill_name...”后置校验与修复在_handle_planning方法中添加一个计划校验环节。使用一个轻量级的校验函数或规则引擎检查计划的基本合法性如无环依赖、技能名存在等。对于简单错误可以尝试自动修复对于复杂错误可以触发重新规划或直接报错。分层规划不要指望LLM一次生成完美计划。可以采用“目标-子目标-具体任务”的分层规划方式。先让LLM生成高级别的子目标序列然后对每个子目标再调用LLM或规则引擎生成具体任务。避坑技巧在开发初期一定要实现一个详细的可视化调试工具。可以是一个简单的命令行打印将当前状态、上下文内容、任务队列以清晰的方式实时输出。这比查看分散的日志高效十倍。更进一步可以生成状态机的图表.dot文件用Graphviz渲染或任务流程图直观展示执行路径这对排查复杂分支逻辑的问题至关重要。6. 进阶思考与扩展方向实现了一个基础的Plan Mode之后你可以根据实际需求从以下几个方向进行深化和扩展打造更强大、更专业的Agent。1. 集成真正的LLM进行动态规划我们示例中的规划是静态的。真正的智能体现在动态规划上。你可以在_handle_planning方法中集成OpenAI、Claude或本地LLM的API。提示词Prompt需要精心设计包含用户目标、可用技能列表、当前上下文、以及之前已执行的任务和结果。让LLM决定下一步做什么甚至重新规划剩余步骤。这就是ReAct (Reasoning Acting)或Chain of Thought (CoT)模式在Agent中的具体应用。2. 实现更复杂的流程控制模式当前引擎主要支持顺序和条件跳转。你可以扩展PlanTask模型和引擎逻辑来实现并行执行对于没有依赖关系的任务可以并发执行以提高效率。需要引入异步编程和任务队列。循环Loop为任务节点添加loop_condition属性当条件满足时重复执行该任务或其子计划。子计划Subplan将一个任务节点指向另一个完整的计划Plan实现计划的嵌套和模块化复用。3. 完善持久化与可观测性生产级的Agent必须能够中断后恢复并且状态可追溯。持久化将ExecutionContext和PlannerEngine的状态当前任务索引、状态机状态等定期序列化如用Pickle或JSON存储到数据库或文件。当Agent重启时可以加载状态从中断点继续执行。可观测性除了基础的日志可以将关键的执行事件状态转换、任务开始/结束、上下文变更推送到监控系统如Prometheus、ELK。为每个计划生成唯一的execution_id便于全链路追踪。4. 构建技能市场与动态加载将技能Skill抽象成独立的、可插拔的模块。维护一个技能注册中心Agent在规划时可以从中心查询可用的技能及其描述、参数schema。技能可以实现动态加载如通过importlib这样无需重启Agent就能扩展其能力。5. 探索多Agent协作当一个任务过于复杂时可以将其拆解分配给多个具有不同专长的子Agent去执行。主Agent的Plan Mode负责顶层协调和任务分发子Agent内部也有自己的规划循环。这就需要定义Agent间的通信协议如通过消息队列或共享存储和协同机制。打造一个健壮的Plan Mode是AI Agent开发从玩具走向实用的关键一步。它不仅仅是几行状态机代码更是一套关于如何让智能体在复杂、不确定的环境中可靠工作的工程学思想。从简单的三段式状态机出发不断迭代和丰富其能力你的Agent才能真正胜任那些“复杂任务”并确保整个过程是“安全”的。