引言从Demo到生产可靠性是最大的鸿沟过去两年LLM智能体从概念验证快速走向实际应用。然而行业数据揭示了令人警醒的现实大量AI智能体永远停留在原型阶段它们在演示环境下表现得智能而流畅一旦投入生产就暴露出不可预测的故障、安全漏洞和失控的成本。一个典型的生产级智能体失败场景是这样的用户提交了一个复杂的多步骤请求Agent在ReAct循环中逐步推理、执行工具调用。但某个中间步骤返回了意外结果Agent陷入无限重试循环每次迭代都在消耗Token和API费用。更糟糕的是它可能在某个时刻被恶意提示注入绕过安全控制执行了未经授权的操作。这些问题的根源往往不是模型能力不足而是架构设计缺乏对可靠性的系统性思考。本文聚焦Plan-then-Execute计划-执行模式通过将战略规划与战术执行分离构建具备可预测性、安全性和可恢复性的LLM智能体服务。以下所有代码示例均基于生产实践提炼可直接用于项目参考。一、为什么选择Plan-then-Execute模式1.1 ReAct模式的局限性ReActReason-Act是最常见的Agent设计模式。Agent在一个循环中运行思考 → 行动 → 观察 → 再思考。这种模式简单灵活但存在一个致命弱点——“短视思维”。# ReAct模式的典型实现——每个步骤独立决策 while not done: thought llm.think(context) # 思考下一步 action llm.select_action(thought) # 决定动作 observation execute(action) # 执行工具 context.append(observation) # 追加观察结果 # 问题Agent看不到全局容易路径依赖在复杂多步骤任务中Agent可能因为某个中间观察结果而偏离主线陷入低效路径甚至死循环。更关键的是安全风险如果某个步骤触发了间接提示注入攻击恶意内容可能渗透后续所有步骤的上下文。1.2 Plan-then-Execute的核心优势Plan-then-Execute模式通过显式分离规划与执行从根本上解决了ReAct模式存在的短视决策和安全隐患问题。该模式将整个Agent工作流划分为三个核心层次第一层Planner战略规划层Planner是整个Agent的大脑承担高层次的战略决策职责。它的输入是用户提交的原始目标输出则是一份结构化、可执行的计划通常以JSON格式或DAG有向无环图形式呈现。这份计划包含完整的步骤序列、每一步所需调用的工具、步骤间的依赖关系以及预期输出格式说明。Planner的一个关键设计原则是仅在此阶段调用昂贵的大模型如GPT-4且调用次数严格控制在1次仅在必要时进行少量重规划。这种设计大幅降低了Token消耗使成本变得可预测。更重要的是Planner的输入仅限于用户最初提交的目标不包含任何执行阶段的观察结果或中间数据从而确保恶意内容无法渗透到规划阶段污染后续的控制流。第二层Executor战术执行层Executor是Agent的手脚负责将Planner生成的计划忠实地转化为具体行动。它按照计划中定义的顺序和依赖关系逐步执行每一步可以调用外部工具、访问API接口或启动子Agent来完成特定任务。与Planner不同Executor可以使用更轻量级的小模型甚至纯确定性代码来实现因为它的职责是执行既定计划而非制定策略。这种分层设计不仅降低了推理成本还大幅提升了执行效率。Executor每次执行前都会验证当前步骤是否仍与原始计划一致这种控制流固化机制有效防止了运行时注入攻击对Agent行为路径的篡改。第三层Verifier验证与审批层可选但强烈推荐Verifier在整个架构中扮演质检员和安全阀的角色虽然它是可选组件但在生产环境中几乎是不可或缺的。Verifier的职责分为两个阶段执行前验证在Planner生成计划之后、Executor启动执行之前Verifier会主动检查计划的合理性、安全性和可行性。例如它能够识别出是否包含高风险操作如删除数据、修改系统配置是否存在逻辑矛盾或依赖循环以及所需工具是否真正可用。对于识别出的中低风险问题Verifier会记录告警但不阻塞流程对于高风险或明显不合理的计划Verifier会标记为需要人工介入。执行后验证在Executor完成某些关键步骤后Verifier可以对执行结果进行校验判断输出是否符合预期格式和内容要求防止因模型幻觉或工具异常而产生错误结果。此外Verifier还承担着Human-in-the-Loop人工介入的协调职责。当遇到高风险操作、计划验证不通过或执行结果异常时Verifier会暂停整个Agent的执行流程生成待审批请求并交由人工决策只有在获得明确批准后才允许继续执行。这种机制确保了AI系统在关键决策点上始终处于人类的监督之下。通过这三层的协同配合Plan-then-Execute模式成功将智能体的行为从不可预测的自由发挥转变为可观察、可控制、可恢复的工程化流程为LLM智能体的生产级部署奠定了坚实的架构基础。二、生产级实现从零构建可靠Agent2.1 核心组件实现以下代码实现一个生产级的Plan-then-Execute Agent包含计划生成、执行器、验证器和人工介入机制。2.1.1 状态定义使用类型安全的状态管理基于LangGraph的状态图模式fromtypingimportTypedDict,Annotated,Sequence,List,Optionalfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpointimportMemorySaverfromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessageimportoperatorclassAgentState(TypedDict):Agent的全局状态 - 单一可信源messages:Annotated[Sequence[BaseMessage],operator.add]user_goal:strplan:Optional[List[dict]]# 结构化计划: [{step_id: 1, task: ..., tool: ..., depends_on: []}]current_step:intexecution_results:Annotated[List[dict],operator.add]step_status:dict# 每个步骤的执行状态: {step_id: pending|running|done|failed}human_approval_needed:boolerror_count:intmax_retries:int32.1.2 Planner生成结构化计划Planner负责将用户目标转换为可执行的步骤序列。关键设计使用结构化输出JSON Schema强制计划格式。frompydanticimportBaseModel,FieldfromtypingimportList,Optionalfromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportPydanticOutputParserclassPlanStep(BaseModel):计划步骤的结构化定义step_id:intField(description步骤编号)task_description:strField(description该步骤要完成的任务描述)required_tool:strField(description执行该步骤所需的工具名称)depends_on:List[int]Field(default[],description依赖的前置步骤ID列表)expected_output:strField(description该步骤期望的输出格式说明)classPlanResult(BaseModel):完整的计划结果steps:List[PlanStep]Field(description步骤列表)overall_goal:strField(description对整体目标的描述)estimated_steps:intField(description总步骤数)defcreate_planner(llm:ChatOpenAI)-callable:创建Planner节点parserPydanticOutputParser(pydantic_objectPlanResult)planner_prompt你是一个任务规划专家。请将以下用户目标分解为具体的、可执行的步骤序列。 用户目标: {goal} **重要约束**: 1. 每个步骤必须可独立执行 2. 必须指明步骤间的依赖关系 3. 步骤总数不超过10步 4. 只规划你知道如何执行的步骤 **可用工具**: {available_tools} {format_instructions} defplanner_node(state:AgentState)-dict:messagesstate.get(messages,[])user_goalstate.get(user_goal,)# 只基于用户原始输入规划不依赖中间执行结果promptplanner_prompt.format(goaluser_goal,available_tools, .join(list_available_tools()),format_instructionsparser.get_format_instructions())responsellm.invoke(prompt)planparser.parse(response.content)return{plan:[step.dict()forstepinplan.steps],current_step:0,step_status:{step.step_id:pendingforstepinplan.steps}}returnplanner_node关键设计决策Planner的输入只包含用户原始目标不包含任何执行阶段的上下文。这确保了即使执行过程中遇到恶意输入Planner阶段生成的计划也不会被污染。2.1.3 Executor安全执行计划Executor接收固化的计划按顺序或依赖关系执行每个步骤。每个步骤执行前需要验证该步骤是否仍在计划中防止控制流被篡改。fromtypingimportDict,Anyimportasynciofromdatetimeimportdatetimeimportlogging loggerlogging.getLogger(__name__)classToolRegistry:工具注册中心 - 所有工具调用经过统一入口_tools:Dict[str,callable]{}classmethoddefregister(cls,name:str,func:callable):cls._tools[name]funcclassmethoddefexecute(cls,tool_name:str,**kwargs)-Any:iftool_namenotincls._tools:raiseValueError(f未知工具:{tool_name})# 执行前记录审计日志logger.info(f执行工具:{tool_name}, 参数:{kwargs})returncls._tools[tool_name](**kwargs)defcreate_executor(llm:Optional[ChatOpenAI]None)-callable:创建Executor节点 - 执行计划中的每一步defexecute_step(step:dict,state:AgentState)-dict:执行单个步骤带重试和超时控制step_idstep[step_id]tool_namestep[required_tool]taskstep[task_description]# 检查前置依赖fordep_idinstep.get(depends_on,[]):dep_statusstate[step_status].get(dep_id)ifdep_status!done:raiseValueError(f依赖步骤{dep_id}尚未完成)# 执行工具调用带超时和重试max_retriesstate.get(max_retries,3)forattemptinrange(max_retries):try:resultasyncio.wait_for(ToolRegistry.execute(tool_name,tasktask),timeout30)return{execution_results:[{step_id:step_id,result:result,status:success,timestamp:datetime.now().isoformat()}],step_status:{step_id:done}}exceptasyncio.TimeoutError:logger.warning(f步骤{step_id}超时, 重试{attempt1}/{max_retries})exceptExceptionase:logger.error(f步骤{step_id}执行失败:{e})ifattemptmax_retries-1:return{execution_results:[{step_id:step_id,error:str(e),status:failed}],step_status:{step_id:failed},error_count:state.get(error_count,0)1}return{}defexecutor_node(state:AgentState)-dict:planstate.get(plan,[])current_idxstate.get(current_step,0)ifcurrent_idxlen(plan):return{current_step:current_idx}# 所有步骤完成# 获取当前步骤stepplan[current_idx]# 【安全关键】检查当前步骤是否仍在原始计划中# 防止执行阶段被注入篡改控制流ifstep[step_id]!state[plan][current_idx][step_id]:raiseSecurityError(检测到控制流篡改)# 执行步骤resultexecute_step(step,state)result[current_step]current_idx1returnresultreturnexecutor_node安全要点工具调用统一入口所有工具调用经过ToolRegistry便于审计、限流和安全策略注入。步骤编号验证执行前验证当前步骤编号与原始计划一致防止注入攻击篡改控制流。超时与重试每个步骤独立超时控制防止单个卡住阻塞整个流程。2.2 可靠性增强验证器与人工介入仅靠PlannerExecutor还不够。生产环境中我们需要计划验证和**人工介入HITL**机制。2.2.1 Verifier计划合理性校验defcreate_verifier(llm:ChatOpenAI)-callable:计划验证器 - 在执行前检查计划是否安全、合理verifier_prompt请验证以下计划是否安全、合理 计划: {plan} 用户目标: {goal} 验证维度: 1. 安全合规: 是否有高风险操作是否包含删除、修改敏感数据的步骤 2. 逻辑合理性: 步骤依赖关系是否正确是否遗漏关键步骤 3. 可行性: 所有工具是否可用步骤描述是否足够清晰 请以JSON格式输出验证结果: { is_valid: true/false, issues: [问题列表], risk_level: low|medium|high, suggestions: [改进建议] } defverifier_node(state:AgentState)-dict:planstate.get(plan,[])goalstate.get(user_goal,)# 如果计划为空无需验证ifnotplan:return{human_approval_needed:False}responsellm.invoke(verifier_prompt.format(planplan,goalgoal))try:validationjson.loads(response.content)except:# 解析失败默认通过但标记为需要人工复核return{human_approval_needed:True}# 高风险或无效计划 → 需要人工介入ifnotvalidation.get(is_valid)orvalidation.get(risk_level)high:return{human_approval_needed:True,validation_issues:validation.get(issues,[])}# 中风险 → 记录但不阻塞ifvalidation.get(risk_level)medium:logger.warning(f计划存在中风险问题:{validation.get(issues)})return{human_approval_needed:False}returnverifier_node2.2.2 人工介入Human-in-the-Loop对于高风险操作Agent应只提议不执行等待人工审批。fromtypingimportLiteraldefcreate_human_approval_node()-callable:人工介入节点 - 暂停执行等待审批defhuman_node(state:AgentState)-dict:返回待审批状态由外部系统处理pending_stepstate[plan][state[current_step]]# 生成审批请求approval_request{step_id:pending_step[step_id],task:pending_step[task_description],tool:pending_step[required_tool],risk_level:high,human_review_required:True}# 状态持久化等待外部回调# 外部系统通过 update_state() 注入审批结果return{pending_approval:approval_request,step_status:{pending_step[step_id]:pending_approval}}returnhuman_nodedefapprove_step(step_id:int,approved:bool,signature:str)-dict:外部审批回调函数ifnotapproved:return{step_status:{step_id:rejected},execution_results:[{step_id:step_id,status:rejected_by_human,human_signature:signature}]}return{step_status:{step_id:approved},human_approval_needed:False}2.3 组装完整Agent使用LangGraph将各节点组装成有状态图fromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpointimportMemorySaver# 生产环境建议用PostgresSaverdefbuild_agent(llm:ChatOpenAI)-StateGraph:构建完整的Plan-then-Execute Agent# 创建节点plannercreate_planner(llm)verifiercreate_verifier(llm)executorcreate_executor(llm)human_approvalcreate_human_approval_node()# 构建状态图workflowStateGraph(AgentState)# 添加节点workflow.add_node(planner,planner)workflow.add_node(verifier,verifier)workflow.add_node(executor,executor)workflow.add_node(human_approval,human_approval)# 设置入口workflow.set_entry_point(planner)# 定义边workflow.add_edge(planner,verifier)# Verifier后的条件路由defafter_verifier(state:AgentState)-Literal[human_approval,executor,__end__]:ifstate.get(human_approval_needed):returnhuman_approvalifnotstate.get(plan):return__end__returnexecutorworkflow.add_conditional_edges(verifier,after_verifier)# Human Approval后的路由defafter_approval(state:AgentState)-Literal[executor,__end__]:pendingstate.get(pending_approval)ifpending:step_statusstate[step_status].get(pending[step_id])ifstep_statusapproved:returnexecutor# 拒绝或其他状态 → 终止return__end__workflow.add_conditional_edges(human_approval,after_approval)# Executor后的路由 - 循环或结束defafter_executor(state:AgentState)-Literal[executor,__end__]:currentstate.get(current_step,0)totallen(state.get(plan,[]))# 检查是否有步骤失败ifstate.get(error_count,0)3:return__end__# 重试耗尽# 检查是否有步骤需要人工介入forstep_id,statusinstate[step_status].items():ifstatuspending_approval:returnhuman_approvalifcurrenttotal:return__end__returnexecutor# 继续下一步workflow.add_conditional_edges(executor,after_executor)returnworkflow# 使用示例defrun_agent(user_goal:str):llmChatOpenAI(modelgpt-4-turbo,temperature0.1)# 生产环境使用PostgresSaver持久化状态checkpoint_saverMemorySaver()appbuild_agent(llm).compile(checkpointercheckpoint_saver)# 配置线程ID支持恢复config{configurable:{thread_id:user_session_123}}initial_state{messages:[HumanMessage(contentuser_goal)],user_goal:user_goal}# 流式执行可处理人工介入暂停foreventinapp.stream(initial_state,config):print(event)# 检查最终状态final_stateapp.get_state(config)returnfinal_state三、可靠性设计模式总结3.1 五种核心稳健性模式生产级Agent系统通常组合使用以下设计模式模式适用场景关键实现Plan-then-Execute复杂多步骤任务规划与执行分离计划前置固化有界循环防止无限重试步骤数、Token数、时间三重预算限制双LLM模式防止提示注入特权LLM隔离处理阻断恶意数据传递审查器-评判器内容质量保证独立模型验证减少幻觉和偏差检查点恢复长时间任务状态持久化支持断点续传和人工介入3.2 生产环境部署清单从演示到生产确保以下核心能力就位状态持久化用PostgresSaver替代MemorySaver支持任务恢复预算限制每轮最多10步、8000 Token、120秒可观测性OpenTelemetry追踪、Prometheus指标、Grafana仪表盘策略引擎Open Policy Agent (OPA) 控制工具调用权限金丝雀发布先影子模式→内部小范围→1%流量→逐步放量CI/CD黄金测试集验证每次变更准确率下降即构建失败结语构建高可靠性的LLM智能体核心不在于让模型更聪明而在于用工程化手段驯服其非确定性。Plan-then-Execute模式通过分离规划与执行、固化控制流、引入验证和人工介入将Agent从不可预测的提示链转变为可观察、可控制、可恢复的生产级服务。正如一位从业者所言“优秀智能体的衡量标准不是智力而是它对安全边界的服从。”当你的Agent能够优雅地处理失败、主动寻求帮助、并让每次行为都可审计时它才真正准备好面对真实世界的复杂与不确定。