AutoGPT Forge 组件开发指南协议、Pydantic 配置与命令系统的完整实践【免费下载链接】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 仓库中 Forgeclassic/forge的组件开发文档与配套源码系统讲解如何从零创建 Agent 组件如何继承AgentComponent与各类protocol来获得消息提供、命令扩展能力如何用 Pydantic 模型为组件接入可序列化配置以及如何通过command装饰器把普通 Python 方法变成 Agent 可调用的工具。读完本文后你可以独立实现一个带配置、带命令的完整组件并理解它被自动收集、排序与执行的底层机制。一、组件体系总览AgentComponent与协议组件Component用于实现各类功能向 prompt 提供消息、执行代码、与外部服务交互等。按照文档定义组件是继承自AgentComponent的类或者实现了至少一个protocol的类由于每个protocol都继承自AgentComponent只要你的类继承任一protocol它就自动成为组件。class MyComponent(AgentComponent): pass这已经是一个合法的组件只是还什么都没做。要让它发挥作用需要实现一个或多个protocol。从源码看AgentComponent是一个抽象基类位于 agent/components.py它本身还内置了两个与文档示例直接相关的机制启用开关类属性_enabled布尔值或可调用对象与enabled属性用于在运行时跳过某些组件_disabled_reason记录禁用原因执行顺序声明run_after(*components)方法允许组件声明“我需要在某些组件之后运行”存入_run_after列表供 Agent 做拓扑排序后文详述。Forge 内置的协议全部定义在 agent/protocols.py包括协议抽象方法作用MessageProviderget_messages() - Iterator[ChatMessage]向 prompt 注入消息CommandProviderget_commands() - Iterator[Command]向 Agent 提供可执行命令DirectiveProviderget_constraints()/get_resources()/get_best_practices()提供指令性文本默认返回空迭代器AfterParseafter_parse(result)解析出行动提案后的钩子ExecutionFailureexecution_failure(error)命令执行失败时的钩子AfterExecuteafter_execute(result)命令执行完成后的钩子二、最小可运行组件MessageProvider先创建一个最简单的组件——把 Hello World! 加入 Agent 的 prompt。这需要实现MessageProvider协议# 因为 MessageProvider 已经继承了 AgentComponent # 所以不再需要显式继承 AgentComponent class HelloComponent(MessageProvider): def get_messages(self) - Iterator[ChatMessage]: yield ChatMessage.user(Hello World!)然后把组件挂到已有 Agent 上或者新建一个 Agent 类加入其中class MyAgent(Agent): self.hello_component HelloComponent()get_messages会在 Agent 每次构建新 prompt 时被调用yield出来的消息会被依次加入 prompt。这个挂在实例属性上、由 Agent 统一调用的模式不是文档的口头约定而是有源码支撑的BaseAgent的元类AgentMeta在实例创建后会自动扫描所有属性凡是AgentComponent实例都会被收集进agent.components见 agent/base.py 的AgentMeta与_collect_components。因此把组件写成实例属性是它被 Agent 识别的关键。三、向组件传递数据、组件之间传递数据组件就是普通 Python 类可以在__init__中传入任意数据包括其他组件。例如传入一个 config 对象在需要时从中读取 API keyclass DataComponent(MessageProvider): def __init__(self, config: Config): self.config config def get_messages(self) - Iterator[ChatMessage]: if self.config.openai_credentials.api_key: yield ChatMessage.system(API key found!) else: yield ChatMessage.system(API key not found!)注意原文档此处说明针对组件自身的专门配置处理component-specific configuration当时尚未实现但在当前仓库源码中ConfigurableComponent已提供了完整的组件级配置机制即下一节的内容。四、配置组件ConfigurableComponent与 Pydantic 模型组件可以通过 Pydantic 模型进行配置。要让组件可配置它必须继承ConfigurableComponent[BM]其中BM是继承自 PydanticBaseModel的配置类。你可以把配置实例传给ConfigurableComponent的__init__也可以直接设置其config属性。使用配置后配置就可以从文件加载并且便于序列化/反序列化方便任意 Agent 复用。# 组件配置示例 class UserGreeterConfiguration(BaseModel): user_name: str class UserGreeterComponent(MessageProvider, ConfigurableComponent[UserGreeterConfiguration]): def __init__(self): # 创建配置实例 # 也可以把配置传给组件构造函数 # 例如 def __init__(self, config: UserGreeterConfiguration): config UserGreeterConfiguration(user_nameWorld) # 把配置实例传给父类 UserGreeterComponent.__init__(self, config) # 下面这行与上面的效果相同 # self.config UserGreeterConfiguration(user_nameWorld) def get_messages(self) - Iterator[ChatMessage]: # 像使用普通模型一样使用配置 yield ChatMessage.system(fHello, {self.config.user_name}!)结合源码这个机制还有几个文档未展开但实践中很重要的细节见 agent/components.py子类必须声明config_class类属性。ConfigurableComponent.__init_subclass__会检查每个子类是否定义了config_class否则抛出NotImplementedError。配置类的 getter 依赖它来构造默认实例。可参考内置组件 MathUtilsComponentclass MathUtilsComponent( DirectiveProvider, CommandProvider, ConfigurableComponent[MathUtilsConfiguration] ): config_class MathUtilsConfiguration def __init__(self, config: Optional[MathUtilsConfiguration] None): ConfigurableComponent.__init__(self, config)配置可以自动合并环境变量。config的 setter 在首次赋值时会调用_update_user_config_from_env做深度合并deep_update这意味着组件配置中的敏感字段如 API key可以直接从环境变量注入而不必硬编码——这正是文档提到的敏感信息处理的落点。整 Agent 级别可整体序列化。BaseAgent提供dump_component_configs/load_component_configs把所有ConfigurableComponent的配置按配置类名打包为 JSON 字符串或从序列化字符串反向恢复方便把 Agent 的完整状态持久化后重建。五、提供命令CommandProvider与command装饰器要扩展 Agent 的能力需要通过CommandProvider协议提供命令。例如让 Agent 会相乘两个数class MultiplicatorComponent(CommandProvider): def get_commands(self) - Iterator[Command]: # yield 出命令Agent 才能使用它 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)command装饰器把方法包装成Command对象其实现见 command/decorator.py行为规则如下names可选的命令名列表不传时使用函数名如multiply。内置组件常借此提供多个别名例如ask_yes_no同时注册为[ask_yes_no, confirm]description可选的简短描述不传时取 docstring 中第一个空行之前的内容正则压缩空白。没有description且没有 docstring 会直接抛ValueError——所以给命令方法写 docstring 是必须的parametersdict[str, JSONSchema]为每个函数参数声明 JSON Schema类型、描述、是否必填。所有命令参数都必须提供 schemaLLM 依据它生成结构化调用参数。命令的返回值通常返回字符串内置组件普遍以 JSON 字符串返回结果便于 LLM 解析异常则抛CommandExecutionError例如 MathUtilsComponent.calculate 会把语法错误、除零、溢出统一包装为该异常。六、Prompt 的组装当组件提供了所有必要数据后Agent 还需要把最终 prompt 组装出来发给 LLM。文档指出当前由PromptStrategy注意它不是protocol负责构建最终 prompt若要改变 prompt 的构建方式需要新建一个PromptStrategy类并在 Agent 中调用相应方法。从源码结构看Forge 的BaseAgentagent/base.py把 prompt 构建相关的参数如send_token_limit、default_cycle_instruction都收敛到BaseAgentConfiguration中而向 LLM 发什么这一步则通过run_pipeline统一驱动各协议方法例如所有MessageProvider.get_messages按序收集。如果你自定义 Agent仍可按文档思路参考默认策略OneShotAgentPromptStrategy的实现以及在 Agent 中搜索self.prompt_strategy观察其调用位置即可理解 prompt 的拼装点。七、完整示例五步实现UserInteractionComponent文档给出了一个略加简化的用户交互组件实现——它让 Agent 能在终端向用户提问。按步骤拆解1. 创建继承自CommandProvider的组件类class MyUserInteractionComponent(CommandProvider): Provides commands to interact with the user. pass2. 实现向用户提问并返回答案的命令方法def ask_user(self, question: str) - str: If you need more details or information regarding the given goals, you can ask the user for input. print(f\nQ: {question}) resp input(A:) return fThe users answer: {resp}3. 用command装饰该方法并为参数声明 schemacommand( parameters{ question: JSONSchema( typeJSONSchema.Type.STRING, descriptionThe question or prompt to the user, requiredTrue, ) }, ) def ask_user(self, question: str) - str: If you need more details or information regarding the given goals, you can ask the user for input. print(f\nQ: {question}) resp input(A:) return fThe users answer: {resp}4. 实现get_commands把命令 yield 出去def get_commands(self) - Iterator[Command]: yield self.ask_user5. 处理非交互模式。由于 Agent 不一定运行在终端或交互模式下当无法向用户提问时应通过self._enabled False禁用该组件def __init__(self, interactive_mode: bool): self._enabled interactive_mode汇总起来最终组件如下# 1. class MyUserInteractionComponent(CommandProvider): Provides commands to interact with the user. def __init__(self, interactive_mode: bool): # 5. self._enabled interactive_mode # 4. def get_commands(self) - Iterator[Command]: # Yield 命令供 Agent 使用 # 组件被禁用时不会生效 yield self.ask_user # 3. command( # 必须为所有命令参数提供 schema parameters{ question: JSONSchema( typeJSONSchema.Type.STRING, descriptionThe question or prompt to the user, requiredTrue, ) }, ) # 2. 命令名即方法名描述即 docstring def ask_user(self, question: str) - str: If you need more details or information regarding the given goals, you can ask the user for input. print(f\nQ: {question}) resp input(A:) return fThe users answer: {resp}对照内置实现 components/user_interaction/user_interaction.py 可以看到生产版本的完整形态它同样继承CommandProvider但用click.prompt替代裸input()对终端输入更稳健且一次 yield 出三个命令——ask_user、ask_yes_no支持默认值[Y/n]提示、ask_choice支持单选/多选返回值均为 JSON 字符串让 LLM 能结构化地读取用户答案。八、把组件接入 Agent替换与禁用默认组件如果想用自己的用户交互组件替换默认组件需要先移除默认的那个当 Agent 继承自Agent时默认组件会随之继承再加入自己的。最简方式是直接覆写user_interaction属性class MyAgent(Agent): def __init__( self, settings: AgentSettings, llm_provider: MultiProvider, file_storage: FileStorage, app_config: Config, ): # 调用父类构造函数带上默认组件 super().__init__(settings, llm_provider, file_storage, app_config) # 覆写默认用户交互组件 self.user_interaction MyUserInteractionComponent()或者把默认组件置为None来禁用它再新增自己的组件class MyAgent(Agent): def __init__( self, settings: AgentSettings, llm_provider: MultiProvider, file_storage: FileStorage, app_config: Config, ): super().__init__(settings, llm_provider, file_storage, app_config) # 禁用默认用户交互组件 self.user_interaction None # 添加自己的组件 self.my_user_interaction MyUserInteractionComponent(app_config)这里禁用如何生效有明确的源码依据BaseAgent.run_pipeline在遍历组件调用协议方法前会检查component.enabled被禁用的组件会直接跳过并记入 traceagent/base.py。同时注意AgentComponent._enabled既可以是布尔值也可以是可调用对象返回 bool这意味着组件可以在每次被访问时动态决定自己是否可用例如依据运行环境或配置热更新。组件的执行顺序同样由框架管理如果 Agent 未显式指定顺序_collect_components会对收集到的组件做拓扑排序_topological_sort依据的就是第五步之外的run_after()声明。执行管线还内置了容错单个组件抛ComponentEndpointError会原地重试默认 3 次整条管线抛EndpointPipelineError则回滚参数并从头重跑。九、延伸阅读内置组件与相关文档学习组件开发的最好材料是仓库中的内置实现内置组件目录classic/forge/forge/components/包含user_interaction、math_utils、web、todo、skills等十几个可参考的组件包其中 MathUtilsComponent 是多协议 可配置组合的典范同时是DirectiveProviderCommandProviderConfigurableComponent命令系统classic/forge/forge/command/配套文档Components 总览含组件配置与顺序机制、Commands 指南、Agents 指南如何扩展内置 Agent、构建自己的 Agent、Protocols 参考。小结Forge 的组件模型可以概括为——用继承AgentComponent或其子类protocol声明我能做什么用实例属性挂载到 Agent 完成注册用 PydanticConfigurableComponent完成配置与持久化用commandJSONSchema完成能力暴露而收集、排序、禁用跳过与失败重试则全部由BaseAgent的管线机制兜底。【免费下载链接】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），仅供参考