Crewai 护栏：让您的 Ai 代理防弹的 7 个步骤！

大型语言模型 代理是非确定性的：为您的 AI 应用实施适当的保护措施

Photo by Muhammad Firdaus Abdullah on Unsplash

考虑到大型语言模型 的非确定性特征，输出结果很容易与我们的应用预期不完全符合。一个众所周知的例子是 Tay，这个微软 聊天机器人因发布冒犯性推文而广为人知。

每当我在开发大型语言模型 应用时，想要决定是否需要实施额外的安全策略时，我会关注以下几点：

• 内容安全：降低生成有害、偏见或不当内容的风险。
• 用户信任：通过透明和负责任的功能建立信心。
• 合规性：与法律框架和数据保护标准保持一致。
• 交互质量：通过确保清晰性、相关性和准确性来优化用户体验。
• 品牌保护：通过最小化风险来保护组织的声誉。
• 滥用预防：预见并阻止潜在的恶意或意外使用案例。

什么是保护措施？

在这个背景下，为代理实施保护措施意味着确保代理的第一次输出不是最终答案。

从技术上讲，您希望根据特定约束评估代理的输出，并在必要时强制代理再生成其答案，直到满足您的要求。

例如，想象一个应用程序，它总结了您过去一个月收到的所有电子邮件。您已指定必须对个人信息（例如发件人姓名）进行匿名处理。然而，由于大型语言模型的不可预测性，它们有时会“忘记”这一条件。在这种情况下，您可以依赖一个额外的步骤来验证您的严格条件是否已满足，因为这是一个关键要求。

CrewAI简介

CrewAI 是我在使用代理时的首选框架。它简单、拥有强大的社区支持、开源，并且完全专注于代理。它还提供了许多额外的功能，使其成为一个非常吸引人的选择。

在深入了解如何使用CrewAI实现保护措施之前，让我简要介绍一下我们将要使用的主要组件。

代理、任务和团队

在 CrewAI 中，任务 和 代理 之间有明确的分离。这种分离使您能够解耦原本作为单个大型提示的一部分的内容。通过保持这两个概念的独立，您可以清晰地定义代理的角色以及代理应该做什么。

fitness_tracker_agent = Agent(
    llm=llm,
    role="Fitness Tracker",
    backstory="An AI that sets goals, tracks progress, and recommends workouts.",
    goal="Set goals, track progress, and recommend workouts.",
    verbose=True,
)

set_and_track_goals = Task(
    description=f"Set the fitness goal ({fitness_goal}), track progress, and use weight history: {historical_weight_data}.",
    expected_output="A muscle gain plan with tracking.",
    agent=fitness_tracker_agent,
)

在这个例子中，我们有一个负责根据健身目标和历史体重数据创建肌肉增加计划的健身代理。正如您所看到的，fitness_tracker_agent 在我们创建任务并指定哪个代理将处理它之前，本身并不知道 要 做什么。

最后，我们将所有任务和代理汇集到一个 团队 中，这构成了我们应用程序的核心。

crew = Crew(
    agents=[fitness_tracker_agent, recommendation_agent],
    tasks=[set_and_track_goals, fetch_fitness_recommendations, provide_workout_plan],
    planning=True,
)

最后一步是使用将传递给任务的输入参数来执行我们的团队——在这种情况下，是健身目标和历史体重数据。

流程

我们刚刚讨论的内容足以构建一个简单的团队，但不足以创建一个具有保护措施的完全功能的代理应用程序。我们需要掌握的下一个概念是 CrewAI 流程。

流程的理念是轻松创建动态 AI 工作流，具有链式任务、无缝状态管理、事件驱动的响应能力，以及用于条件、循环和分支的灵活控制流。

以下装饰器有助于操控执行流程：

• **@start()**：标记流程的入口点，并在流程开始时启动任务。
• **@listen()**：在特定任务或事件完成时执行一个方法。
• **@router()**：根据条件或结果将流程引导到不同的路径。

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

class ExampleState(BaseModel):
    counter: int = 0
    message: str = ""

class StateExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        self.state.message = "Hello from first_method"
        self.state.counter += 1

    @listen(first_method)
    def second_method(self):
        self.state.message += " - updated by second_method"
        self.state.counter += 1
        return self.state.message

flow = StateExampleFlow()
final_output = flow.kickoff()
print(f"Final Output: {final_output}")
print("Final State:")
print(flow.state)

此外，流程允许访问一个共享对象，该对象存储和管理数据（ExampleState），从而实现任务之间的无缝通信，并在整个工作流中保持上下文。

使用CrewAI流程的保护措施

通过我们刚刚讨论的两个简单概念，我们准备好增强我们的AI代理。在这个例子中，我将演示如何创建一个多代理AI应用程序，该应用程序能够生成文本并在提供输出之前验证文本是否包含暴力内容。

文本中的暴力检查发生在应用程序内部。得益于流程的状态管理，我们可以控制文本再生成的频率，从而防止无限循环。这种方法为本质上是非确定性的事物引入了一定程度的确定性。

导入

主要使用的库是 CrewAI，但我们还将导入 Pydantic 来创建一个 BaseModel 类，因为 CrewAI 的文档要求这样做。

在这个例子中，我们假设已经创建了两个代理及其对应的任务，如前所述：一个用于生成文本，另一个用于检查文本是否包含暴力内容。

from typing import List
from pydantic import BaseModel, Field
from crewai import Flow, start, listen, router

除了 Flow 类，我们还需要使用三个装饰器来构建我们的应用程序。

状态

状态 是我们在流程执行过程中用于持久化数据的方式。我们需要持久化的第一条信息是生成的文本。这是我们在所有迭代中存储文本的地方，并根据需要进行更新。

class ViolenceCheckState(BaseModel):
    generated_text: str = ""
    contains_violence: bool = False
    generation_attempts_left: int = 2

为了控制流程，我们还将使用一个标志来指示文本是否包含暴力，以及一个计数器来跟踪剩余的生成尝试次数。这些属性将在流程执行过程中被访问和更新。

流程类

我们正在实现的类由旨在控制应用程序流程的方法组成。让我们一步一步地构建它，从 __init__ 函数开始。

class ViolenceCheckFlow(Flow[ViolenceCheckState]):
    topic: str = Field(description="文本生成的主题")

    def __init__(self, topic: str):
        super().__init__()
        self.topic = topic

到目前为止，一切正常。我们只是初始化了我们类的唯一属性。这里的新内容是 ViolenceCheckState Pydantic 模型，我们将其传递给超类 Flow。该模型将表示我们的状态。

    @start()
    def generate_text(self):
        print(f"根据输入主题生成文本: {self.topic}")
        task = create_text_generation_task(self.topic)  
        crew = Crew(agents=[text_generator_agent], tasks=[task])
        result = crew.kickoff()
        self.state.generated_text = result.raw
        print("文本生成完成！")

@listen(generate_text)
def validate_text_for_violence(self):
    print("正在验证文本是否包含暴力...")
    task = create_violence_check_task(self.state.generated_text)
    crew = Crew(agents=[violence_checker_agent], tasks=[task])
    result = crew.kickoff()
    self.state.contains_violence = "Violence" in result.raw
    print("验证完成:", "检测到暴力" if self.state.contains_violence else "未检测到暴力")

第一个方法必须使用 @start 装饰器来定义我们流程的起点。接下来是第二个方法，它监听第一个方法并在其后执行。

这两个方法执行以下任务：

• generate_text: 该方法创建一个 Crew，根据给定主题生成文本。生成的结果通过 self.state.generated_text 保存到状态中。
• validate_text_for_violence: 该方法使用不同的 Crew 检查文本是否包含暴力内容。如果检测到暴力，我们将 self.state.contains_violence 设置为 True。

添加路由复杂性

这里变得有趣了。现在我们使用 @router 装饰器根据应用程序是否需要再生成文本来指导流程。这个决定基于 self.state.contains_violence 的值。

@router(validate_text_for_violence)
def route_text_validation(self):
    if not self.state.contains_violence:
        return "safe"
    elif self.state.generation_attempts_left == 0:
        return "not_feasible"
    else:
        return "regenerate"

route_text_validation 方法在 validate_text_for_violence 之后立即执行，如装饰器所指定。它检查状态中的 contains_violence 属性，并：

• 如果未检测到暴力，则返回 "safe"。
• 如果检测到暴力且仍有尝试次数，则触发文本的再生成。

处理信号

我们课程的最后部分包括三个方法，这些方法监听从 route_text_validation 发送的特定信号："safe"、"regenerate" 和 "not_feasible"。

@listen("safe")
def save_safe_text(self):
    with open("safe_text.txt", "w") as file:
        file.write(self.state.generated_text)
    print("Safe text saved to file")

@listen("regenerate")
def regenerate_text(self):
    self.state.generation_attempts_left -= 1
    self.generate_text()

@listen("not_feasible")
def notify_user(self):
    print("Generated text contains violence and further attempts are not feasible.")

以下是如何处理每种情况：

• “safe”：如果未检测到暴力，则生成的文本将保存到文件中。
• “regenerate”：如果触发文本再生成，则 generation_attempts_left 计数器减少 1，并再次调用 generate_text 方法，重新开始该过程。
• “not_feasible”：当达到最大再生成尝试次数（在这种情况下为 2）时发生，应用程序无法生成无暴力的文本。

结论

这种方法对我来说简直是救命稻草。该实现专注于组织循环和处理迭代，而不是重新发明应用程序流程背后的逻辑。在这个例子中，我专注于检查文本是否包含暴力，并在发现时确定适当的操作。得益于框架提供的这个干净的内置功能，我不必浪费时间管理流程逻辑本身。

更多用例

在这种方法的众多用例中，我想强调查询注入。这比你想象的要普遍，尤其是在“提示工程”等角色日益重要的情况下。使用CrewAI，你可以创建一个流程来评估初始查询，以确定攻击者是否试图利用聊天机器人，例如。

挑战

这是框架的方式：它给你很多，但当出现问题时，它可能会连本带利地收回所有。我曾经启动简单的应用程序，却发现自己在 GitHub 上打开拉取请求以修复经过数小时工作后的错误。

话虽如此，情况并没有听起来那么糟糕。CrewAI 仍然是我在多代理应用程序中的首选。然而，与任何新库一样，依赖它意味着接受风险并准备好应对这些风险。

如果你想了解更多关于这个库的信息，请查看我上次的演讲: