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-independent）	`DirectiveProvider`、`CommandProvider`	只贡献数据/能力，不依赖其他组件的执行结果，可任意顺序添加
顺序依赖（Order-dependent）	`MessageProvider`、`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 提供可执行的命令（Command）：

class 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( type=JSONSchema.Type.INTEGER, description="The first number", required=True, ), "b": JSONSchema( type=JSONSchema.Type.INTEGER, description="The second number", required=True, )}) 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实现@command，parameter.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.1`AfterParse`：LLM 响应解析之后

在响应被解析完成、即将进入执行阶段时回调：

class AfterParse(AgentComponent): def after_parse(self, response: ThoughtProcessOutput) -> None: ...

示例：记录解析后的响应日志：

class LoggerComponent(AfterParse): def after_parse(self, response: ThoughtProcessOutput) -> None: logger.info(f"Response: {response}")

5.2`ExecutionFailure`：命令执行失败时

当某条命令执行抛错时回调，参数是捕获到的异常：

class ExecutionFailure(AgentComponent): def execution_failure(self, error: Exception) -> None: ...

示例：

class LoggerComponent(ExecutionFailure): def execution_failure(self, error: Exception) -> None: logger.error(f"Command execution failed: {error}")

5.3`AfterExecute`：命令成功执行之后

命令由 Agent 成功执行完毕后回调，参数是ActionResult：

class AfterExecute(AgentComponent): def after_execute(self, result: ActionResult) -> None: ...

示例：

class LoggerComponent(AfterExecute): def after_execute(self, result: ActionResult) -> None: logger.info(f"Result: {result}")

6. 源码纵深：Agent 如何发现组件并运行协议管线

协议本身只是接口声明，真正让整套机制运转起来的是 BaseAgent（classic/forge/forge/agent/base.py）中的组件收集与管线执行逻辑。以下三块源码能帮你把文档概念落到实现层面。

6.1 组件自动发现：`AgentMeta`元类

Agent 不需要手工注册组件。AgentMeta元类在实例化 Agent 后自动调用_collect_components：

class 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_limit=3次）；
管线级：抛出更严重的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(type=JSONSchema.Type.INTEGER, description="The first number", required=True), "b": JSONSchema(type=JSONSchema.Type.INTEGER, description="The second number", required=True)}) 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`才可执行
暴露可调用工具	`CommandProvider`+`@command`	名称默认取方法名，描述取 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 文档。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

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