AutoGPT Forge 协议（Protocols）完全指南：组件如何向 Agent 注入指令、命令与执行钩子

AutoGPT Forge 协议Protocols完全指南组件如何向 Agent 注入指令、命令与执行钩子【免费下载链接】AutoGPTAutoGPT is the vision of accessible AI for everyone, to use and to build on. Our mission is to provide the tools, so that you can focus on what matters.项目地址: https://gitcode.com/GitHub_Trending/au/AutoGPT本篇技术指南讲解 AutoGPT Forge 子项目中的协议Protocols机制——即组件Components向 Agent 扩展能力的标准接口。读完本文你将理解 6 个内置协议DirectiveProvider、CommandProvider、MessageProvider、AfterParse、ExecutionFailure、AfterExecute各自的职责边界与接口签名掌握用command装饰器注册可调用的命令并能从源码层面看懂 Agent 是如何发现组件、按序执行协议管线以及处理失败重试的。1. 什么是协议组件与 Agent 之间的接口契约协议是 组件 所实现的接口用于把相关功能归组每个协议都需要 Agent 在执行过程的某个节点显式处理。Forge 提供了一套完整的内置协议且这些协议都已在内置 Agent 中被处理——因此只要从基础 Agent 类继承所有内置协议开箱即用参见 协议文档。文档中明确协议按照默认执行顺序排列并分为两大类分类协议含义顺序无关Order-independentDirectiveProvider、CommandProvider只贡献数据/能力不依赖其他组件的执行结果可任意顺序添加顺序依赖Order-dependentMessageProvider、AfterParse、ExecutionFailure、AfterExecute组件可能依赖前面组件的结果执行顺序有讲究在源码中这些协议全部定义在 protocols.py它们共同继承自AgentComponent基类class DirectiveProvider(AgentComponent): def get_constraints(self) - Iterator[str]: return iter([]) def get_resources(self) - Iterator[str]: return iter([]) def get_best_practices(self) - Iterator[str]: return iter([]) class CommandProvider(AgentComponent): abstractmethod def get_commands(self) - Iterator[Command]: ... class MessageProvider(AgentComponent): abstractmethod def get_messages(self) - Iterator[ChatMessage]: ... class AfterParse(AgentComponent, Generic[AnyProposal]): abstractmethod def after_parse(self, result: AnyProposal) - None | Awaitable[None]: ... class ExecutionFailure(AgentComponent): abstractmethod def execution_failure(self, error: Exception) - None | Awaitable[None]: ... class AfterExecute(AgentComponent): abstractmethod def after_execute(self, result: ActionResult) - None | Awaitable[None]: ...从源码结构看有两个值得注意的实现细节AfterParse是泛型协议Generic[AnyProposal]其after_parse接收的参数类型随 Agent 的提案模型而变比文档中写死的ThoughtProcessOutput更通用顺序依赖的三个协议都声明返回None | Awaitable[None]即协议方法既可以是同步函数也可以是协程——Agent 的执行管线会自动await可等待结果见 base.py 中run_pipeline的inspect.isawaitable(result)判断。所有组件的公共基类是 AgentComponent它提供了两个关键属性class AgentComponent(ABC): _run_after: list[type[AgentComponent]] [] # 声明我必须在这些组件之后运行 _enabled: Callable[[], bool] | bool True # 静态布尔或动态函数决定是否启用 _disabled_reason: str 这意味着任何协议实现都可以声明自己的执行顺序偏好run_after与启用开关enabled而不必硬编码执行次序。2. 顺序无关协议一DirectiveProvider——给 LLM 注入知情提示DirectiveProvider为 Agent 产出约束constraints、资源resources与最佳实践best practices。它不直接影响其他协议纯粹是信息性的这些文本会在构建 prompt 时一并传递给 LLM。三个方法均有默认的空实现见 protocols.py所以按需重写即可class DirectiveProvider(AgentComponent): def get_constraints(self) - Iterator[str]: return iter([]) def get_resources(self) - Iterator[str]: return iter([]) def get_best_practices(self) - Iterator[str]: return iter([])示例一个提供互联网资源信息的 web 搜索组件。注意原文档特别提醒——仅提供资源信息并不会让 Agent 真正具备上网能力必须另配一个能提供相应Command的组件才能真正执行搜索class WebSearchComponent(DirectiveProvider): def get_resources(self) - Iterator[str]: yield Internet access for searches and information gathering. # 如果不需要 get_constraints 和 get_best_practices可以直接跳过不实现这个声明能力与提供执行分离的设计正是顺序无关协议的价值所在DirectiveProvider负责让 LLM知道有什么资源可用CommandProvider负责让 LLM能够真正调用。3. 顺序无关协议二CommandProvider——注册 Agent 可调用的命令CommandProvider为 Agent 提供可执行的命令Commandclass CommandProvider(AgentComponent): def get_commands(self) - Iterator[Command]: ...提供命令最简便的方式是在组件方法上使用command装饰器然后在get_commands中 yield 该方法。每条命令需要名称默认使用方法名描述默认取 docstring 中Args:或Returns:之前的第一部分参数 schema用JSONSchema定义可通过装饰器参数传入。示例可执行乘法的计算器组件。Agent 会在当前任务相关时自主调用该命令并看到返回结果from forge.agent import CommandProvider, Component from forge.command import command from forge.models.json_schema import JSONSchema class CalculatorComponent(CommandProvider): get_commands(self) - Iterator[Command]: yield self.multiply command(parameters{ a: JSONSchema( typeJSONSchema.Type.INTEGER, descriptionThe first number, requiredTrue, ), b: JSONSchema( typeJSONSchema.Type.INTEGER, descriptionThe second number, requiredTrue, )}) def multiply(self, a: int, b: int) - str: Multiplies two numbers. Args: a: First number b: Second number Returns: Result of multiplication return str(a * b)效果Agent 会看到一个名为multiply、接收两个参数的命令其描述为Multiplies two numbers.调用后 LLM 直接获得a * b的计算结果。装饰器与命令模型的完整实现在 classic/forge/forge/command/command.py定义Command模型decorator.py实现commandparameter.py负责参数解析。更多命令细节见 Commands 文档。4. 顺序依赖协议三MessageProvider——向 prompt 追加消息MessageProvider产出的消息会直接加入 Agent 的 prompt。有两种角色可用ChatMessage.user(...)被解释为用户发送的消息ChatMessage.system(...)权重更高、更受重视的系统消息。class MessageProvider(AgentComponent): def get_messages(self) - Iterator[ChatMessage]: ...示例向 Agent 的 prompt 中注入一条消息class HelloComponent(MessageProvider): def get_messages(self) - Iterator[ChatMessage]: yield ChatMessage.user(Hello World!)由于多个组件可能同时实现该协议消息的注入顺序取决于组件顺序——这正是它被归为顺序依赖协议的原因。5. 顺序依赖协议四执行阶段钩子以下三个协议分别挂载在解析 → 执行链路的三个关键节点上是典型的观察点logging、拦截点与状态同步点。5.1AfterParseLLM 响应解析之后在响应被解析完成、即将进入执行阶段时回调class AfterParse(AgentComponent): def after_parse(self, response: ThoughtProcessOutput) - None: ...示例记录解析后的响应日志class LoggerComponent(AfterParse): def after_parse(self, response: ThoughtProcessOutput) - None: logger.info(fResponse: {response})5.2ExecutionFailure命令执行失败时当某条命令执行抛错时回调参数是捕获到的异常class ExecutionFailure(AgentComponent): def execution_failure(self, error: Exception) - None: ...示例class LoggerComponent(ExecutionFailure): def execution_failure(self, error: Exception) - None: logger.error(fCommand execution failed: {error})5.3AfterExecute命令成功执行之后命令由 Agent 成功执行完毕后回调参数是ActionResultclass AfterExecute(AgentComponent): def after_execute(self, result: ActionResult) - None: ...示例class LoggerComponent(AfterExecute): def after_execute(self, result: ActionResult) - None: logger.info(fResult: {result})6. 源码纵深Agent 如何发现组件并运行协议管线协议本身只是接口声明真正让整套机制运转起来的是 BaseAgentclassic/forge/forge/agent/base.py中的组件收集与管线执行逻辑。以下三块源码能帮你把文档概念落到实现层面。6.1 组件自动发现AgentMeta元类Agent 不需要手工注册组件。AgentMeta元类在实例化 Agent 后自动调用_collect_componentsclass AgentMeta(ABCMeta): def __call__(cls, *args, **kwargs): instance super().__call__(*args, **kwargs) instance._collect_components() # 实例创建后自动收集组件 return instance_collect_components会扫描 Agent 实例的全部属性把所有AgentComponent实例挑出来如果组件列表尚未显式设置还会调用_topological_sort按各组件声明的_run_after依赖做拓扑排序——这就是第 4 节顺序依赖在实现上的落地文档说的顺序很重要最终由run_after声明 拓扑排序来保证。若某组件挂在 Agent 上却漏加进components列表源码会打出一条logger.warning可作为排查提示。6.2 协议管线执行run_pipeline的重试策略每个协议的遍历执行都统一走run_pipeline。它通过protocol_method.__qualname__反查 protocols 模块中的协议类逐个检查组件是否为该协议的实例isinstance、是否启用component.enabled然后调用同名方法。核心容错逻辑是两级重试组件级单个组件抛出ComponentEndpointError定义于 components.py时只重试该组件默认retry_limit3次管线级抛出更严重的EndpointPipelineError如该端点整条管线结果不可用时回滚管线参数到原始副本并从第一个组件重新执行整条管线其他异常直接上抛。此外run_pipeline会维护self._trace执行轨迹⬇️ 协议名、✅ 成功组件、❌ 失败组件及其原因便于调试协议管线卡在哪一环。参数在重试前通过_selective_copy做浅/深拷贝保护避免重试污染原始数据。6.3 组合起来一个自定义协议组件把文档中的示例组装成完整组件并附带顺序与启用控制源码中run_after与enabled是组件级能力适用于任何协议实现from forge.agent import CommandProvider from forge.agent.components import AgentComponent from forge.command import command from forge.models.json_schema import JSONSchema class CalculatorComponent(CommandProvider, AgentComponent): # 仅在配置允许时启用声明在 MessageProvider 之后运行 run_after () # 需要时改为 run_after(MyMessageProvider) get_commands(self) - Iterator[Command]: yield self.multiply command(parameters{ a: JSONSchema(typeJSONSchema.Type.INTEGER, descriptionThe first number, requiredTrue), b: JSONSchema(typeJSONSchema.Type.INTEGER, descriptionThe second number, requiredTrue)}) def multiply(self, a: int, b: int) - str: Multiplies two numbers. Args: a: First number b: Second number Returns: Result of multiplication return str(a * b)继承内置 Agent 后该命令自动进入get_commands管线LLM 即可在决策时看到multiply命令并按需调用执行失败会触发ExecutionFailure管线成功则触发AfterExecute管线——文档中继承自基础 Agent 即所有内置协议生效的承诺正是由上述run_pipeline 元类发现机制实现的。7. 实践要点速查需求推荐协议注意告诉 LLM有什么资源/限制DirectiveProvider纯信息性不授予实际能力需配合Command才可执行暴露可调用工具CommandProvidercommand名称默认取方法名描述取 docstring 首段参数必须声明JSONSchema向 prompt 注入上下文MessageProvider区分user与system消息权重多组件时顺序敏感记录/审查 LLM 提案AfterParse参数类型随 Agent 的提案模型泛型变化捕获执行异常ExecutionFailure参数是捕获到的Exception消费执行结果AfterExecute参数是ActionResult常用于状态回写与日志适用范围与前提本文所有协议与示例均基于当前仓库classic/forge/子项目Forge Agent 框架文档与源码位于 classic/forge/forge/agent/ 与 classic/forge/forge/command/组件需继承内置 Agent 才能享受内置协议处理顺序依赖协议的组件顺序可借助run_after声明并由拓扑排序保证。相关延伸阅读Forge 组件总览、Commands 文档。【免费下载链接】AutoGPTAutoGPT is the vision of accessible AI for everyone, to use and to build on. Our mission is to provide the tools, so that you can focus on what matters.项目地址: https://gitcode.com/GitHub_Trending/au/AutoGPT创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考