Content Creation Guardrails

In this assignment, you'll create a multi-agent AI application designed for automated social media content creation (like LinkedIn posts) with built-in safety guardrails. The system will ensure that generated content is appropriate and free from harmful elements like violence, as the process will be automated without manual review.

Input

Content Topic & Platform

Content Creation

Generates initial content

Safety Check

Content moderation

Output

Safe, Platform-Optimized Content

How It Works

Using CrewAI's Flow feature, we can control the sequence of operations and ensure that content passes through necessary safety checks before being published. This is crucial for automated social media posting where each piece of content needs to be verified for appropriateness.

Components

Crew

A specialized team of agents working together to create and moderate content. The crew includes a content creator for generating posts and a safety moderator for ensuring content appropriateness.

Pydantic Flow State

Manages the application's state throughout the content creation process. It tracks the content at each stage, from initial creation to final approval, ensuring data consistency and traceability.

Flow Implementation

Leverages CrewAI's Flow class to orchestrate the sequence of operations. The flow ensures that content is created, checked for safety, and approved in the correct order before being published.

Required Inputs

  • Tags for the content
  • Target platform

Useful Resources

Code Examples

Travel Advisor Flow

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

class TravelState(BaseModel):
    vacation_plan: str = ""
    is_plan_valid: bool = False
    generation_attempts_left: int = 2

class TravelAdvisorFlow(Flow[TravelState]):
    names: List[str] = Field(description="List of travelers")
    city: str = Field(description="Departure city")
    destination: str = Field(description="Dest city")

    def __init__(self, names: List[str], city: str, destination: str):
        super().__init__()
        self.names = names
        self.city = city
        self.destination = destination

    @start()
    @listen("regenerate")
    def generate_vacation_plan(self):
        print("Generating vacation plan")
        task = create_vacation_plan_task(self.names, self.city, self.destination)
        crew = Crew(agents=[planner_agent], tasks=[task])
        result = crew.kickoff()
        self.state.vacation_plan = result.raw
        print("Vacation plan generated!")

    @listen(generate_vacation_plan)
    def validate_vacation_plan(self):
        print("Start validation of the plan...")
        task = validate_vacation_plan_task(self.state.vacation_plan)
        crew = Crew(agents=[validator_agent], tasks=[task])
        result = crew.kickoff()
        self.state.is_plan_valid = "Valid" in result.raw
        print("Validation complete", "Valid" if self.state.is_plan_valid else "Invalid")

    @router(validate_vacation_plan)
    def route_vacation_plan(self):
        if self.state.is_plan_valid:
            return "valid"
        elif self.state.generation_attempts_left == 0:
            return "not_feasible"
        else:
            self.state.generation_attempts_left -= 1
            return "regenerate"

    @listen("valid")
    def finalize_vacation_plan(self):
        with open("vacation_plan.txt", "w") as file:
            file.write(self.state.vacation_plan)
        print("Vacation plan saved in file")

    @listen("not_feasible")
    def notify_user(self):
        print("Plan is not feasible, I'm sorry :(")