AI Agent大模型后端任务调度【免费下载链接】XAgentAn Autonomous LLM Agent for Complex Task Solving项目地址https://gitcode.com/gh_mirrors/xa/XAgent点击查看免费下载导读本文以 XAgent 开源仓库中 Markdown_Docs/XAgent/agent/tool_agent/agent.md 文档为骨架结合 XAgent/agent/tool_agent/agent.py 等源码实现系统讲解 ToolAgent 的完整职责它负责在复杂任务求解中围绕工具树及其函数执行搜索与调用其核心parse()方法承担了提示词组装、模型调用、工具调用参数校验与自动修复、以及消息到工具节点ToolNode的转换工作。读完本文你将掌握 ToolAgent 在 XAgent 多代理体系中的定位、parse()的完整执行链路、openai 请求类型下的函数过滤与描述改写规则、基于 jsonschema 的 tool_call 参数验证与动态修复原理以及如何将模型返回的消息还原为可执行的数据结构。ToolAgent 的定位继承自 BaseAgent 的工具树搜索代理ToolAgent是 XAgent 中负责“实际执行”的代理。从源码看它直接继承自BaseAgentXAgent/agent/base_agent.py类定义如下class ToolAgent(BaseAgent): abilities set([RequiredAbilities.tool_tree_search])其核心特性如下能力标识abilities属性是一个集合用于存储当前 ToolAgent 的能力默认被设置为RequiredAbilities.tool_tree_search该枚举定义于 XAgent/utils.py取值为0。这表示 ToolAgent 的本质能力是“工具树搜索”——在工具树结构中检索并调用合适的工具来完成子任务。与基类的关系BaseAgent是抽象基类其默认能力集合覆盖了plan_generation、plan_refinement、task_evaluator、tool_tree_search、reflection、summarization六项见 XAgent/agent/base_agent.py而 ToolAgent 收窄为单一能力。基类同时定义了fill_in_placeholders()占位符填充和generate()模型调用两个关键方法ToolAgent 的parse()正是围绕它们展开。注册与调度ToolAgent 在 XAgent 组件集中被注册为四个可用代理之一。在 XAgent/core.py 中available_agents列表包含了PlanGenerateAgent、PlanRefineAgent、ToolAgent、ReflectAgent说明它在“计划生成 → 计划精炼 → 工具执行 → 反思”的循环中扮演执行者角色。ToolAgent 定义了两个公开方法也是文档与源码共同聚焦的核心方法职责parse()基于输入参数调用generate()生成消息列表与 token 列表按特定条件修改后返回message_to_tool_node()将模型返回的消息字典转换为ToolNode对象供工作流后续执行parse 方法模型调用的完整执行链路parse()是 ToolAgent 最核心的方法它在源码中带有retry(stopstop_after_attempt(CONFIG.max_retry_times), reraiseTrue)装饰器XAgent/agent/tool_agent/agent.py意味着解析失败时最多重试max_retry_times次该值来自全局配置见 XAgent/config.py失败则向上抛出异常。参数清单参数类型说明placeholdersdict可选存储占位符及其映射关系的字典默认{}argumentsdict可选存储参数详细信息的字典openai 类型下会被临时置空functionslist允许插入函数字段的函数列表用于 openai 类型function_calldict表示当前正在处理的函数调用的字典stopdict循环的终止条件additional_messageslist可选要附加到现有消息列表的附加消息列表默认[]additional_insert_indexint可选插入附加消息的索引位置默认-1*args / **kwargs-父类generate()的可变长度参数与任意关键字参数返回值为一个元组(message, tokens)message是解析后的消息字典tokens是 token 使用情况列表。第一步占位符填充与消息组装prompt_messages self.fill_in_placeholders(placeholders) messages prompt_messages[:additional_insert_index] additional_messages prompt_messages[additional_insert_index:] messages [message.raw() for message in messages]fill_in_placeholders()是基类方法XAgent/agent/base_agent.py它对self.prompt_messages做深拷贝并按placeholders中role → {key: value}的映射把消息内容中的{{key}}替换为真实值。ToolAgent 的提示词中即存在{{all_plan}}、{{subtask_id}}、{{workspace_files}}、{{max_length}}、{{step_num}}、{{human_help_prompt}}等占位符见 XAgent/agent/tool_agent/prompt.py。随后将附加消息按additional_insert_index插入到填充后的消息序列中默认插到末尾。最后把所有Message对象转换为原始字典.raw()。第二步openai 请求类型的特殊处理当self.config.default_request_type openai时parse()会执行一系列面向 OpenAI Function Calling 的预处理arguments None functions list(filter(lambda x: x[name] not in [subtask_submit,subtask_handle],functions)) if CONFIG.enable_ask_human_for_help: functions [function_manager.get_function_schema(ask_human_for_help)] messages[0][content] \n--- Avaliable Tools ---\n....format(json.dumps(functions,indent2))置空 arguments临时禁用显式参数传递避免与函数调用机制冲突。函数过滤从可用函数列表中剔除subtask_submit与subtask_handle——这两个函数属于“纯函数”定义于 XAgent/ai_functions/pure_functions/task_handle_functions.yml不应作为可调用工具暴露给模型而是作为外层包装。按需注入求助函数若配置启用了enable_ask_human_for_help则把ask_human_for_help的函数 schema 追加进列表该函数是模型与人类交互的唯一入口其 schema 同样位于上述 yml 文件中。注入可用工具清单把过滤后的函数 schema 以 JSON 形式拼接到第一条系统消息末尾明确告诉模型它只能在subtask_handle.tool_call函数字段中使用这些工具且tool_input字段必须是 JSON 格式。随后调用change_tool_call_description(messages[0])改写首条消息中的描述文案详见下文最后把 functions 列表重置为仅包含两个包装函数functions [function_manager.get_function_schema(subtask_submit), function_manager.get_function_schema(subtask_handle)]这样模型最终只能输出subtask_handle携带具体工具调用或subtask_submit结束子任务真实工具调用被嵌套在subtask_handle.tool_call中。第三步调用父类 generate() 生成消息message,tokens self.generate( messagesmessages, argumentsarguments, functionsfunctions, function_callfunction_call, stopstop, *args,**kwargs )基类generate()XAgent/agent/base_agent.py按CONFIG.default_request_type分发openai 类型通过objgenerator.chatcompletion()调用 OpenAI 接口解析函数调用参数使用 json5 解析并将返回内容拆分为message[arguments]与message[function_call]。xagent 类型直接调用支持自研模型协议的接口把返回的 content 用 json5 解析为消息字典。其他类型抛出NotImplementedError。第四步tool_call 参数验证与自动修复这是parse()中最具技术含量的环节。当请求类型为 openai 且函数调用参数中包含tool_call字段时即模型嵌套声明了具体工具调用tool_schema function_manager.get_function_schema(function_call_args[tool_call][tool_name]) assert tool_schema is not None, fFunction {function_call_args[tool_call][tool_name]} not found! Poential Schema Validation Error! tool_call_args function_call_args[tool_call][tool_input] if tool_input in function_call_args[tool_call] else 通过FunctionManager.get_function_schema()XAgent/ai_functions/function_manager.py按工具名查找函数 schema。FunctionManager在初始化时从functions与pure_functions两个目录加载全部 yaml/yml 配置文件构建function_cfgs字典。若查不到 schema触发AssertionError消息为Function xxx not found! Poential Schema Validation Error!——这正是文档中提到的“如果在可能的函数列表中找不到指定的函数模式则抛出 AssertionError”的实现位置。提取tool_input作为待验证参数。随后执行validate()闭包详见下文“validate 函数”一节核心是jsonschema.validate(instancetool_call_args, schematool_schema[parameters])。若验证抛出异常则进入修复流程messages[0] change_tool_call_description(messages[0], reverseTrue) tool_call_args objgenerator.dynamic_json_fixes( broken_jsontool_call_args, function_schematool_schema, messagesmessages, error_messagestr(e))[choices][0][message][function_call][arguments] validate()修复策略很有意思先把首条消息的描述文案反向还原reverseTrue即把改写后的描述恢复成原始表述再调用objgenerator.dynamic_json_fixes()定义于 XAgent/ai_functions/request/obj_generator.py——这是一个“动态 JSON 修复器”它把损坏的 JSON、函数 schema、当前消息上下文以及验证错误信息一并交给模型让模型重新生成符合 schema 的参数然后再次验证。若第二次验证仍失败异常会向上抛出由retry装饰器兜底重试。第五步消息结构重组验证通过后parse()把嵌套的tool_call提升为顶层function_callfunction_call_args[tool_call][tool_input] tool_call_args message[function_call] function_call_args.pop(tool_call) message[function_call][name] message[function_call].pop(tool_name) message[function_call][arguments] message[function_call].pop(tool_input) message[arguments] function_call_args即将tool_name重命名为name、tool_input重命名为arguments而plan/thought/reasoning/criticism等推理字段作为arguments保留。最终返回统一格式的消息字典供下游message_to_tool_node()消费。文档中给出的返回示例示意如下{ message: { content: 这是一个示例消息, role: system, type: text }, tokens: [这是, 一个, 示例, 令牌] }change_tool_call_description工具调用描述的定向改写该函数定义于 XAgent/agent/tool_agent/agent.py用于修改工具调用的描述文案。它内部维护了一个“描述对”列表des_pairs每个元素是一组(原始描述, 改写描述)des_pairs [ (Use tools to handle the subtask, Use subtask_handle to make a normal tool call to handle the subtask), (5.1 Please remember to generate the function call field after the criticism field.\n 5.2 Please check all content is in json format carefully., 5.1. Please remember to generate the tool_call field after the criticism field.\n 5.2. Please remember to generate comma if the tool_call field is after the criticism field.\n 5.3. Please check whether the **tool_call** field is in the function call json carefully.), (After decide the action, use subtask_handle functions to apply action., After decide the action, call functions to apply action.), ]函数通过message[content].replace(...)遍历替换当reverseFalse时把原始描述替换为改写后的描述强调模型必须使用subtask_handle包裹工具调用、注意 tool_call 字段的 JSON 格式与逗号当reverseTrue时反向替换把改写描述还原为原始表述。这个双向替换的设计在验证修复流程中非常关键正向改写让模型在生成阶段聚焦“subtask_handle tool_call”的正确格式反向还原则在参数校验失败、需要让模型重新生成 JSON 时把提示词恢复为更宽松的原始版本避免过度约束干扰修复。文档中的示例输入{content: Use tools to handle the subtask}且reverseFalse返回{content: Use subtask_handle to make a normal tool call to handle the subtask}。validate基于 jsonschema 的工具参数校验validate()是parse()内部定义的闭包函数通过nonlocal声明捕获外层变量tool_schema与tool_call_argsdef validate(): nonlocal tool_schema, tool_call_args if isinstance(tool_call_args, str): tool_call_args {} if tool_call_args else json5.loads(tool_call_args) jsonschema.validate(instancetool_call_args, schematool_schema[parameters])其行为分两步类型归一化如果tool_call_args是字符串模型可能直接输出 JSON 字符串先做转换——空字符串转为空字典{}非空字符串用json5.loads()解析为字典使用 json5 是为了容忍 JSON 中的注释、尾逗号等非标准写法。模式验证调用jsonschema.validate(instancetool_call_args, schematool_schema[parameters])其中tool_schema[parameters]来自对应工具的 yaml 配置例如 XAgent/ai_functions/functions/parse_web_text.yml 等文件中function.parameters字段。若参数不符合模式缺必填字段、类型错误、枚举越界等将抛出异常由parse()中的 try/except 捕获并进入修复流程。jsonschema 校验的存在确保了模型生成的工具参数在进入真实执行环境之前就是结构合法的避免了脏数据污染工具树与工作空间。message_to_tool_node消息到 ToolNode 的转换parse()产出的消息字典最终要通过message_to_tool_node()转换为ToolNode对象才能接入 XAgent 的工具树数据结构。方法签名与核心实现def message_to_tool_node(self, message) - ToolNode: new_node ToolNode() if content in message.keys(): new_node.data[content] message[content] if arguments in message.keys(): new_node.data[thoughts][properties] message[arguments] if function_call in message.keys(): new_node.data[command][properties][name] message[function_call][name] new_node.data[command][properties][args] message[function_call][arguments] else: logger.typewriter_log(message_to_tool_node warning: no function_call in message, Fore.RED) return new_node映射规则清晰对应文档中的描述消息字段ToolNode 落点contentnew_node.data[content]argumentsnew_node.data[thoughts][properties]即 plan/thought/reasoning/criticism 等推理信息function_call.namenew_node.data[command][properties][name]function_call.argumentsnew_node.data[command][properties][args]ToolNode类定义于 XAgent/data_structure/node.py其data结构预先定义了content、thoughts.properties、command.properties、tool_output、tool_status_code等字段并维护father、children、expand_num、history、workspace_hash_id等树结构属性。message_to_tool_node()实际上完成了“模型输出 → 树节点”的适配content是节点的可见输出thoughts.properties保存推理轨迹command.properties保存待执行的工具命令。若消息中缺少function_call字段则记录红色警告日志使用 colorama 的Fore.RED提示消息不可执行。文档给出的转换示例# 输入消息 { content: The content is useless, function_call: {name: xxx, arguments: xxx}, arguments: {xxx: xxx, xxx: xxx} } # 转换后的 ToolNode.data { content: The content is useless, thoughts: {properties: {xxx: xxx, xxx: xxx}}, command: {properties: {name: xxx, args: xxx}} }配套提示词SYSTEM_PROMPT、USER_PROMPT 与调度器示例ToolAgent 的运行高度依赖 XAgent/agent/tool_agent/prompt.py 中定义的两套提示词SYSTEM_PROMPT定义 ToolAgent 的角色“实验性前沿超级自主代理”、工作流先读任务与计划 → 决定动作 → 调用函数 → 写任务报告 →subtask_submit提交、可用资源网络、FileSystemEnv、Python notebook、ShellEnv、ask_for_human_help以及性能准则持续自我批评、每步有成本要高效、函数调用 JSON 格式检查等。值得注意的是第 5 条关于“在 criticism 字段后生成 function call 字段、仔细检查 JSON 格式”的规则正是change_tool_call_description中第二组描述对的来源——提示词与代码逻辑在此闭环。USER_PROMPT携带当前子任务编号{{subtask_id}}、文件系统结构{{workspace_files}}、工具调用预算{{max_length}}与当前步数{{step_num}}并强调“达到所有里程碑或确认工具无法解决时必须使用subtask_submit提交同时给计划精炼代理Plan Refine Agent提供细化建议”。get_examples_for_dispatcher()返回三元组(example_input, example_system_prompt, example_user_prompt)供调度器dispatcher生成提示时作为示例。example_input是一个“24 点游戏寻找可行示例”的子任务 JSON 示例包含name、goal、handler、tool_budget、prior_plan_criticsim、milestones、expected_tools、exceute_status等字段直观展示了子任务对象的结构example_system_prompt与example_user_prompt直接复用上述两套提示词。在整体工作流中的位置与配置依赖从 XAgent/core.py 可以看到ToolAgent 与 PlanGenerateAgent、PlanRefineAgent、ReflectAgent 并列注册为available_agents。在 XAgent 的执行循环中任务被拆分为树状计划后每个子任务交由 ToolAgent 在工具树中搜索并调用工具执行执行结果通过subtask_submit提交其suggestions_for_latter_subtasks_plan字段schema 见 XAgent/ai_functions/pure_functions/task_handle_functions.yml反馈给 Plan Refine Agent 迭代精炼后续计划。ToolAgent 的行为受以下全局配置控制配置解析见 XAgent/config.py配置项影响default_request_type决定走openaiFunction Calling 流程还是xagent自研模型协议分支openai 分支才会触发函数过滤、描述改写与 tool_call 验证max_retry_times控制parse()的重试次数也影响请求层obj_generator的重试上限XAgent/ai_functions/request/openai.py 中为max_retry_times 3enable_ask_human_for_help决定是否向模型开放ask_human_for_help函数对应人类协助能力小结ToolAgent 是 XAgent 多代理体系中“动手执行”的关键一环它通过parse()完成占位符填充、openai 函数过滤与描述改写、模型调用、tool_call 的 jsonschema 校验与动态修复再通过message_to_tool_node()将模型输出落为可执行的 ToolNode 树节点最后配合subtask_submit/subtask_handle等纯函数与后续的计划精炼、反思代理形成完整闭环。理解其参数验证与消息重组逻辑是二次开发 XAgent 工具链、接入自定义工具或适配新模型协议的重要基础。赞分享AI Agent大模型后端任务调度【免费下载链接】XAgentAn Autonomous LLM Agent for Complex Task Solving项目地址https://gitcode.com/gh_mirrors/xa/XAgent点击查看免费下载相关推荐炉石传说HsMod插件5个实用功能彻底改变你的游戏体验炉石传说HsMod插件5个实用功能彻底改变你的游戏体验 还在为炉石传说漫长的动画等待而烦恼想要更高效的卡包开启体验HsMod插件正是为你量身打造的解决方案游戏开发10分钟上手XuperChain从环境搭建到部署第一条联盟链的完整指南10分钟上手XuperChain从环境搭建到部署第一条联盟链的完整指南 XuperChain是一款高性能、高灵活性的区块链架构专为企业级联盟链场景设计。本指MassTransit消息调度机制深度解析MassTransit消息调度机制深度解析 什么是消息调度 在分布式系统中消息调度是指按照预定时间或周期发送消息的能力。MassTransit作为.NET生态后端消息队列微服务消息路由上一篇Floccus终极指南跨浏览器书签同步的私有云解决方案下一篇PANIX与MITRE ATTCK矩阵Linux持久化技术映射与防御策略终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考