⚙️ 协议

协议是由组件实现的*接口*，用于将相关功能分组。每个协议都需要在执行的某个阶段由代理显式处理。我们提供了一个全面的内置协议列表，这些协议已经在内置的Agent中处理，因此当你继承自基础代理时，所有内置协议都将正常工作！

协议按默认执行顺序列出。

无序协议

仅实现无序协议的组件可以按任意顺序添加，包括在有序协议之间。

DirectiveProvider

为代理提供约束、资源和最佳实践。这对其他协议没有直接影响；纯粹是信息性的，将在构建提示时传递给llm。

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([])

示例 一个网络搜索组件可以提供资源信息。请记住，这实际上并不允许代理访问互联网。要实现这一点，需要提供相关的Command。

class WebSearchComponent(DirectiveProvider):
    def get_resources(self) -> Iterator[str]:
        yield "用于搜索和信息收集的互联网访问。"
    # 如果不需要，可以跳过 "get_constraints" 和 "get_best_practices"

CommandProvider

提供代理可以执行的命令。

class CommandProvider(AgentComponent):
    def get_commands(self) -> Iterator[Command]:
        ...

提供命令的最简单方法是使用command装饰器在组件方法上，然后生成该方法。每个命令都需要一个名称、描述和一个使用JSONSchema的参数模式。默认情况下，方法名称用作命令名称，文档字符串的第一部分（在Args:或Returns:之前）用于描述，模式可以在装饰器中提供。

示例 一个可以执行乘法的计算器组件。如果与当前任务相关，代理能够调用此命令并查看返回的结果。

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="第一个数字",
                required=True,
            ),
            "b": JSONSchema(
                type=JSONSchema.Type.INTEGER,
                description="第二个数字",
                required=True,
            )})
    def multiply(self, a: int, b: int) -> str:
        """
        将两个数字相乘。

        参数:
            a: 第一个数字
            b: 第二个数字

        返回:
            乘法结果
        """
        return str(a * b)

代理将能够调用这个名为multiply的命令，并带有两个参数，并接收结果。命令描述将是：将两个数字相乘。

要了解更多关于命令的信息，请参阅🛠️ 命令。

有序协议

实现有序协议的组件的顺序很重要。 某些组件可能依赖于其前面组件的结果。

MessageProvider

生成将添加到代理提示中的消息。你可以使用ChatMessage.user()：这将被解释为用户发送的消息，或ChatMessage.system()：这将更为重要。

class MessageProvider(AgentComponent):
    def get_messages(self) -> Iterator[ChatMessage]:
        ...

示例 提供消息到代理提示的组件。

class HelloComponent(MessageProvider):
    def get_messages(self) -> Iterator[ChatMessage]:
        yield ChatMessage.user("Hello World!")

AfterParse

在解析响应后调用的协议。

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}")

ExecutionFailure

在命令执行失败时调用的协议。

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

示例 在命令失败时记录错误的组件。

class LoggerComponent(ExecutionFailure):
    def execution_failure(self, error: Exception) -> None:
        logger.error(f"命令执行失败: {error}")

AfterExecute

协议在命令被代理成功执行后调用。

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}")