跳到内容

🧩 组件

组件是 🤖 Agent 的构建块。它们是继承自 AgentComponent 或实现一个或多个 ⚙️ 协议 的类,这些类赋予 Agent 额外的能力或处理能力。

组件可用于实现各种功能,例如为提示提供消息、执行代码或与外部服务交互。它们可以被启用或禁用、排序,并且可以相互依赖。

通过 self 在 Agent 的 __init__ 中分配的组件在 Agent 实例化时会自动检测。例如,在 __init__ 内部:self.my_component = MyComponent()。您可以使用任何有效的 Python 变量名,对于组件能否被检测到的关键是其类型(AgentComponent 或任何继承自它的协议)。

访问 内置组件 查看开箱即用的组件。

from forge.agent import BaseAgent
from forge.agent.components import AgentComponent

class HelloComponent(AgentComponent):
    pass

class SomeComponent(AgentComponent):
    def __init__(self, hello_component: HelloComponent):
        self.hello_component = hello_component

class MyAgent(BaseAgent):
    def __init__(self):
        # These components will be automatically discovered and used
        self.hello_component = HelloComponent()
        # We pass HelloComponent to SomeComponent
        self.some_component = SomeComponent(self.hello_component)

组件配置

每个组件都可以使用标准的 pydantic BaseModel 定义自己的配置。为确保配置从文件正确加载,组件必须继承自 ConfigurableComponent[BM],其中 BM 是它使用的配置模型。ConfigurableComponent 提供一个 config 属性,用于保存配置实例。可以直接设置 config 属性,或将配置实例传递给组件的构造函数。可以传递额外的配置(即不属于 Agent 的组件),这些配置将被静默忽略。即使稍后添加组件,额外的配置也不会被应用。要查看内置组件的配置,请访问 内置组件

from pydantic import BaseModel
from forge.agent.components import ConfigurableComponent

class MyConfig(BaseModel):
    some_value: str

class MyComponent(AgentComponent, ConfigurableComponent[MyConfig]):
    def __init__(self, config: MyConfig):
        super().__init__(config)
        # This has the same effect as above:
        # self.config = config

    def get_some_value(self) -> str:
        # Access the configuration like a regular model
        return self.config.some_value

敏感信息

虽然可以直接在代码中将敏感数据传递给配置,但建议对 API 密钥等敏感数据使用 UserConfigurable(from_env="ENV_VAR_NAME", exclude=True) 字段。数据将从环境变量中加载,但请记住,代码中传递的值优先。所有字段,即使是被排除的字段(exclude=True),在配置从文件加载时也会被加载。排除允许你在序列化期间跳过它们,未排除的 SecretStr 将被字面序列化为 "**********" 字符串。

from pydantic import BaseModel, SecretStr
from forge.models.config import UserConfigurable

class SensitiveConfig(BaseModel):
    api_key: SecretStr = UserConfigurable(from_env="API_KEY", exclude=True)

配置序列化

BaseAgent 提供两个方法: 1. dump_component_configs:将所有组件的配置序列化为 json 字符串。 1. load_component_configs:将 json 字符串反序列化为配置并应用它。

JSON 配置

你可以在启动 Agent 时指定一个 JSON 文件(例如 config.json)用于配置。此文件包含 AutoGPT 使用的各个 组件 的设置。要指定文件,请使用 --component-config-file CLI 选项,例如使用 config.json

./autogpt.sh run --component-config-file config.json

注意

如果你使用 Docker 运行 AutoGPT,你需要将配置文件挂载或复制到容器中。更多信息请参见 Docker 指南

示例 JSON 配置

你可以复制想要更改的配置,例如到 classic/original_autogpt/config.json 并根据你的需要进行修改。大多数配置都有默认值,最好只设置你想要修改的值。你可以在 内置组件 中查看可用的配置字段和默认值。你也可以在 .json 文件中设置敏感变量,但建议使用环境变量。

{
    "CodeExecutorConfiguration": {
        "execute_local_commands": false,
        "shell_command_control": "allowlist",
        "shell_allowlist": ["cat", "echo"],
        "shell_denylist": [],
        "docker_container_name": "agent_sandbox"
    },
    "FileManagerConfiguration": {
        "storage_path": "agents/AutoGPT/",
        "workspace_path": "agents/AutoGPT/workspace"
    },
    "GitOperationsConfiguration": {
        "github_username": null
    },
    "ActionHistoryConfiguration": {
        "llm_name": "gpt-3.5-turbo",
        "max_tokens": 1024,
        "spacy_language_model": "en_core_web_sm"
    },
    "ImageGeneratorConfiguration": {
        "image_provider": "dalle",
        "huggingface_image_model": "CompVis/stable-diffusion-v1-4",
        "sd_webui_url": "https://:7860"
    },
    "WebSearchConfiguration": {
        "duckduckgo_max_attempts": 3
    },
    "WebSeleniumConfiguration": {
        "llm_name": "gpt-3.5-turbo",
        "web_browser": "chrome",
        "headless": true,
        "user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.97 Safari/537.36",
        "browse_spacy_language_model": "en_core_web_sm"
    }
}

组件排序

组件的执行顺序很重要,因为有些组件可能依赖于前面组件的结果。默认情况下,组件按字母顺序排列。

单个组件排序

你可以通过将其他组件(或它们的类型)传递给 run_after 方法来对单个组件进行排序。这样可以确保该组件将在指定的组件之后执行。run_after 方法返回组件本身,因此你可以在将组件赋值给变量时调用它

class MyAgent(Agent):
    def __init__(self):
        self.hello_component = HelloComponent()
        self.calculator_component = CalculatorComponent().run_after(self.hello_component)
        # This is equivalent to passing a type:
        # self.calculator_component = CalculatorComponent().run_after(HelloComponent)

警告

排序组件时,请确保不要形成循环依赖!

所有组件排序

你也可以通过在 Agent 的 __init__ 方法中设置 self.components 列表来排序所有组件。这样可以确保没有循环依赖,并且任何 run_after 调用都会被忽略。

警告

确保包含所有组件 - 通过设置 self.components 列表,你将覆盖自动发现组件的默认行为。由于这通常不是预期行为,如果某些组件被跳过,Agent 会在终端中通知你。

class MyAgent(Agent):
    def __init__(self):
        self.hello_component = HelloComponent()
        self.calculator_component = CalculatorComponent()
        # Explicitly set components list
        self.components = [self.hello_component, self.calculator_component]

禁用组件

你可以通过设置组件的 _enabled 属性来控制哪些组件被启用。组件默认是启用的。你可以提供一个 bool 值或一个 Callable[[], bool],每次组件即将执行时都会检查它。这样你就可以根据某些条件动态地启用或禁用组件。你还可以通过设置 _disabled_reason 提供禁用组件的原因。该原因将在调试信息中显示。

class DisabledComponent(MessageProvider):
    def __init__(self):
        # Disable this component
        self._enabled = False
        self._disabled_reason = "This component is disabled because of reasons."

        # Or disable based on some condition, either statically...:
        self._enabled = self.some_property is not None
        # ... or dynamically:
        self._enabled = lambda: self.some_property is not None

    # This method will never be called
    def get_messages(self) -> Iterator[ChatMessage]:
        yield ChatMessage.user("This message won't be seen!")

    def some_condition(self) -> bool:
        return False

如果你根本不需要某个组件,可以直接将其从 Agent 的 __init__ 方法中移除。如果你想移除继承自父类的组件,可以将相关属性设置为 None

警告

移除其他组件所需的组件时要小心。这可能会导致错误和意外行为。

class MyAgent(Agent):
    def __init__(self):
        super().__init__(...)
        # Disable WatchdogComponent that is in the parent class
        self.watchdog = None

异常

提供了自定义错误,可用于在出现问题时控制执行流程。所有这些错误都可以在协议方法中抛出,并将由 Agent 捕获。
默认情况下,如果异常仍未解决,Agent 将重试三次,然后重新抛出该异常。所有传递的参数都会自动处理,并在需要时恢复值。所有错误都接受一个可选的 str 消息。以下是按范围递增排序的错误

  1. ComponentEndpointError:单个端点方法执行失败。Agent 将在该组件上重试执行此端点。
  2. EndpointPipelineError:管道执行失败。Agent 将为所有组件重试执行该端点。
  3. ComponentSystemError:多个管道失败。

示例

from forge.agent.components import ComponentEndpointError
from forge.agent.protocols import MessageProvider

# Example of raising an error
class MyComponent(MessageProvider):
    def get_messages(self) -> Iterator[ChatMessage]:
        # This will cause the component to always fail 
        # and retry 3 times before re-raising the exception
        raise ComponentEndpointError("Endpoint error!")