XAgent 计划生成 Agent（PlanGenerateAgent）源码解析：从占位符解析到任务树构建

AI Agent大模型后端任务调度【免费下载链接】XAgentAn Autonomous LLM Agent for Complex Task Solving项目地址https://gitcode.com/gh_mirrors/xa/XAgent点击查看免费下载导读本文深入剖析 XAgent 开源项目中的计划生成模块PlanGenerateAgentXAgent/agent/plan_generate_agent/agent.py该 Agent 负责将用户查询拆解为必须达成的目标所对应的子任务树是整个自主任务求解流程的第一步。通过阅读本文你将掌握parse方法的参数语义与调用契约、占位符填充与 LLM 生成链路的底层实现以及它如何与调度器Dispatcher、计划执行器PlanExec配合完成初始计划的落地并可进一步复用或扩展出自己的计划生成 Agent。一、类定位BaseAgent 子类与能力注册PlanGenerateAgent是 BaseAgent 的直接子类职责单一而明确——生成计划。源码中类的声明与能力声明如下# XAgent/agent/plan_generate_agent/agent.py from typing import List from XAgent.agent.base_agent import BaseAgent from XAgent.utils import RequiredAbilities from XAgent.ai_functions import function_manager, objgenerator from XAgent.message_history import Message class PlanGenerateAgent(BaseAgent): abilities set([RequiredAbilities.plan_generation])其中abilities是一个集合声明了该 Agent 所需的核心能力。RequiredAbilities定义于 XAgent/utils.pyplan_generation是该能力枚举中的一员取值为 1。正是这个能力标识让调度器能够在多个 Agent 中按能力路由到本类。作为BaseAgent的子类PlanGenerateAgent自动继承了基类的三件套能力__init__(config, prompt_messages)接收全局配置与初始提示消息构造 Agent 实例fill_in_placeholders(placeholders)用占位符字典批量替换提示消息中的{{key}}模板generate(messages, arguments, functions, function_call, stop, ...)调用底层 LLM 完成生成并返回消息与 token 用量。值得注意的一点BaseAgent.parse是一个abc.abstractmethod抽象方法子类必须实现。PlanGenerateAgent对parse的实现正是整个计划生成流程的入口。二、parse 方法参数契约与执行流程parse是PlanGenerateAgent唯一对外暴露的核心方法其完整签名如下来自 agent.pydef parse( self, placeholders: dict {}, arguments: dict None, functionsNone, function_callNone, stopNone, additional_messages: List[Message] [], *args, **kwargs ):各参数语义如下表参数类型默认值作用placeholdersdict{}包含要填充到消息中的占位符的字典键为消息角色system/user值为{占位符名: 值}的映射argumentsdictNone要在函数调用中使用的参数通常为 JSON Schema会注入到函数参数定义中functions列表None计划生成过程中可供 LLM 调用的函数工具列表function_call对象None强制指定要调用的函数函数名或{name: ...}形式stop对象None停止条件指定后将在生成到达该条件时终止additional_messagesList[Message][]追加到初始提示消息之后的多条消息*args/**kwargs——可变长度参数列表与任意关键字参数透传给generate2.1 方法体执行流程方法体只有三行但承载了完整的模板填充 → 消息合并 → LLM 生成流水线agent.pyprompt_messages self.fill_in_placeholders(placeholders) messages prompt_messages additional_messages return self.generate( messagesmessages, argumentsarguments, functionsfunctions, function_callfunction_call, stopstop, *args, **kwargs )具体而言填充占位符调用基类fill_in_placeholders对self.prompt_messages做深拷贝deepcopy逐个消息、逐个占位符地将{{key}}替换为真实值base_agent.py。之所以深拷贝是为了保证多次调用parse不会污染原始提示模板合并附加消息将填充后的提示消息与additional_messages顺序拼接构成完整的对话上下文调用 generate把arguments、functions、function_call、stop等一并透传返回(message, tokens)二元组——其中message为 LLM 返回的消息对象OpenAI 模式下含arguments与可选function_calltokens为 token 用量统计。2.2 官方输出示例原文档给出的调用示例可直接作为自定义集成的模板# 创建 PlanGenerateAgent 对象 agent PlanGenerateAgent() # 解析占位符、参数、函数调用和附加消息生成计划 plan agent.parse( placeholders{placeholder1: value1, placeholder2: value2}, arguments{arg1: value1, arg2: value2}, functions[function1, function2], function_callfunction_call, stopstop_condition, additional_messages[message1, message2] ) # 打印计划结果 print(plan)注意事项实际使用时placeholders的键应是消息角色。基类实现按placeholders[role]取值base_agent.py因此应写成{system: {query: ...}, user: {...}}的形式同时可根据具体需求对parse传入不同的arguments与functions定制计划生成过程使其输出符合预期的子任务结构。三、提示工程支撑SYSTEM_PROMPT 与 USER_PROMPTPlanGenerateAgent的聪明程度很大程度上来自配套提示词模板 prompt.py。该文件定义了完整的系统提示SYSTEM_PROMPT与用户提示USER_PROMPT核心要点包括3.1 计划与子任务的树形结构系统提示明确规定计划是一棵子树形结构子任务按编号继承如任务 1 包含 1.1、1.2、1.3而 1.2 又包含 1.2.1……每个子任务包含三个 JSON 字段{ subtask name: string子任务名称, goal.goal: string子任务的主要目标以及达成目标将采取的行动, goal.criticism: string当前子任务及其目标可能存在的潜在问题, milestones: list[string]为确保子任务完成应达成的里程碑需详细具体 }这一结构在函数 Schema 中有对应的正式定义见 XAgent/ai_functions/pure_functions/task_manage_functions.yml其中subtask name、goal含goal与criticism、milestones均为required字段milestones被描述为后续子任务据此自动检查当前子任务是否完成的判据。3.2 SUBTASK_SPLIT 操作与任务处理约定系统提示要求 LLM 使用操作SUBTASK_SPLIT将查询拆分为多个子任务并明确以下约束任务处理 Agent 将以中序遍历方式处理所有子任务先处理任务 1成功则继续任务 2失败则把任务 1 拆成 1.1、1.2、1.3……再递归处理直到全部解决保持计划树简洁高效避免生成与现有子任务相似或重复的新子任务相似目标的子任务应合并为一个含子目标列表的子任务任务处理器由最先进的 LLM 驱动可直接回答许多问题因此计划应充分利用该能力并降低子树复杂度允许规划多个子任务但须最小化数量同时保证所有必须达成的目标都被包含子任务 ID 采用点号分隔字符串如1.2、2.3.3与 Plan 类 的get_subtask_id实现一一对应。USER_PROMPT则负责注入具体查询This is not the first time you are handling the task, so you should give a initial plan. Here is the query: {{query}} You will use operation SUBTASK_SPLIT to split the query into 2-4 subtasks and then commit.注意其中的{{query}}占位符正是parse方法中placeholders要填充的目标提示词同时限定初始拆分为2~4 个子任务呼应了系统提示中最小化子任务数量的原则。3.3 get_examples_for_dispatcher为调度器提供示例prompt.py 中还提供了get_examples_for_dispatcher函数返回三元组(example_input, example_system_prompt, example_user_prompt)即示例查询 系统提示 用户提示。其默认示例为Generate a plan for writing a Python-based calculator.被调度器用于动态润色refine特定 Agent 的提示词。四、调用链实战从 Dispatcher 到初始计划落地4.1 调度器按能力路由在 XAgentDispatcher 中get_examples依据能力类型选择对应 Agent 模块的示例函数if ability_type RequiredAbilities.plan_generation: from .plan_generate_agent import get_examples_for_dispatcher随后dispatch方法dispatcher.py完成两件事若enable为真调用DispatcherAgent.parse(target_task, example_input, example_system_prompt, example_user_prompt)对提示词进行润色若润色失败消息内容为空则回退到默认的SYSTEM_PROMPTUSER_PROMPT调用build_agent实例化PlanGenerateAgent(config, prompt_messages)构造失败时回退为config单参构造。4.2 计划执行器中的真实调用在 XAgent/workflow/plan_exec.py 的initial_plan_generation中展示了PlanGenerateAgent在完整流水线中的真实用法split_functions deepcopy(function_manager.get_function_schema(subtask_split_operation)) agent agent_dispatcher.dispatch( RequiredAbilities.plan_generation, target_taskfGenerate a plan to accomplish the task: {self.query.task}, ) new_message, _ agent.parse( placeholders{ system: { avaliable_tool_names: str([cont[name] for cont in self.avaliable_tools_description_list]), }, user: { query: self.plan.data.raw } }, argumentsdeepcopy(function_manager.get_function_schema(simple_thought)[parameters]), functions[split_functions], ) subtasks json5.loads(new_message[function_call][arguments]) for subtask_item in subtasks[subtasks]: subplan plan_function_output_parser(subtask_item) Plan.make_relation(self.plan, subplan)这段代码揭示了几个关键实现事实函数注册机制functions来自function_manager.get_function_schema(subtask_split_operation)该 Schema 定义于 task_manage_functions.yml其参数要求target_subtask_id点号分隔的字符串与subtasks长度为 1~3 的子任务数组元素为上述task_item结构占位符填充placeholders按system/user两个角色分别注入工具名列表与真实查询文本结果解析LLM 返回的function_call.arguments用json5解析兼容无引号键、尾逗号等宽松 JSON得到subtasks列表计划树构建通过 Plan.make_relation 将每个解析出的子计划挂接到根计划上形成父子关系子计划的数据结构对应 TaskSaveItem包含name、goal、milestones等字段。4.3 中序遍历与剩余子任务计划生成完成后Plan 类 提供get_inorder_travel中序遍历整棵树、pop_next_subtask取出下一个状态为TODO的子任务与get_remaining_subtask获取剩余全部子任务等方法供任务处理器逐个子任务推进——这正是系统提示中中序遍历处理约定的数据结构实现。五、generate 底层openai 与 xagent 两种请求模式parse最终把控制权交给基类 generate该方法按CONFIG.default_request_type分派openai 模式若未提供functions会自动包装一个名为reasoning的函数参数来自arguments的 JSON Schema并强制调用若仅有一个函数则把arguments[properties]合并进该函数的参数定义随后调用objgenerator.chatcompletion并把返回的function_call用json5解析后拆分为arguments与function_call两部分返回xagent 模式直接调用chatcompletion并将返回内容的 JSON 解析为最终消息其他请求类型会抛出NotImplementedError。同时当functions数量为 1 且未显式指定function_call时generate会自动设置function_call {name: functions[0][name]}base_agent.py确保 LLM 至少调用一次该函数——这是计划生成得以稳定输出结构化子任务的关键保障。六、继承复用PlanRefineAgentPlanGenerateAgent的 parse 流水线填充 → 合并 → 生成被后续的 PlanRefineAgent 直接继承复用。后者仅将abilities替换为RequiredAbilities.plan_refinement并在parse中额外支持additional_insert_index参数用于把附加消息插入到提示消息的任意位置而非仅追加在末尾messages prompt_messages[:additional_insert_index] additional_messages prompt_messages[additional_insert_index:]这种设计印证了PlanGenerateAgent的解析 生成骨架具备良好的可扩展性任何需要用 LLM 产出结构化结果的 Agent都可以继承它并定制abilities、提示词与函数 Schema。七、小结与阅读指引PlanGenerateAgent是 XAgent 任务求解链路的第一环它以abilities {plan_generation}声明能力通过parse完成占位符填充 → 附加消息合并 → LLM 生成的标准流程配合SUBTASK_SPLIT函数 Schema 与树形子任务 JSON 结构把用户查询转化为一棵可被中序遍历执行、失败可递归拆分、完成度可用里程碑校验的计划树。如需进一步深入建议按以下路径阅读仓库类实现XAgent/agent/plan_generate_agent/agent.py提示词与示例XAgent/agent/plan_generate_agent/prompt.py基类能力填充与生成XAgent/agent/base_agent.py调度与路由XAgent/agent/dispatcher.py初始计划生成调用链XAgent/workflow/plan_exec.py计划树数据结构XAgent/data_structure/plan.py子任务函数 SchemaXAgent/ai_functions/pure_functions/task_manage_functions.yml继承扩展示例XAgent/agent/plan_refine_agent/agent.py赞分享AI Agent大模型后端任务调度【免费下载链接】XAgentAn Autonomous LLM Agent for Complex Task Solving项目地址https://gitcode.com/gh_mirrors/xa/XAgent点击查看免费下载相关推荐XAgent 计划生成与细化机制深度解析PlanAgent、PlanRefineChain 与子任务树操作源码详解XAgent 计划生成与细化机制深度解析PlanAgent、PlanRefineChain 与子任务树操作源码详解 本篇指南围绕 XAgent 工作流中的计划AI Agent大模型后端任务调度XAgent 计划生成 Agent 提示词工程解析get_examples_for_dispatcher 与 SUBTASK_SPLIT 机制XAgent 计划生成 Agent 提示词工程解析get_examples_for_dispatcher 与 SUBTASK_SPLIT 机制 本指南以 XAAI Agent大模型后端任务调度XAgent 内循环搜索基类 BaseSearchMethod 解析任务树搜索的状态机、占位方法与 ReACT 子类实现XAgent 内循环搜索基类 BaseSearchMethod 解析任务树搜索的状态机、占位方法与 ReACT 子类实现 本篇围绕 XAgent/inner_AI Agent大模型后端任务调度上一篇Scientist终极PHP实验框架 - 安全重构代码的完整指南下一篇LiteGraph.js tree-shaking配置babel-plugin-import使用创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考