main.py

import os
from textwrap import dedent
from crewai import Agent, Task, Crew, Process
from typing import List
from pydantic import BaseModel, Field
import asyncio
from concurrent.futures import ThreadPoolExecutor
from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper

from langchain_core.tools import Tool

dalle3 = Tool(
    "Dall-E-Image-Generator",
    DallEAPIWrapper(model="dall-e-3").run,
    "A wrapper around OpenAI DALL-E API. Useful for when you need to generate images from a text description. Input "
    "should be an image description.",
)


class Step(BaseModel):
    text: str = Field(description="The text for the step")
    img_link: str = Field(description="The link for the generated illustration")


class Recipe(BaseModel):
    title: str = Field(description="the name of the recipe")
    ingredients: List[str] = Field(description="the list of ingredients")
    steps: List[Step] = Field(description="the list of step for the recipe and its illustration")


recipe_master = Agent(
    role="Recipe Master",
    goal="Create a recipe from a list of ingredients",
    backstory="You are an agent that have a deep knowledge and experience in food recipes",
    verbose=True,
    memory=True
)

copywriter = Agent(
    role="Copywriter",
    goal="Rewrite all the steps from the recipe so they sound like if they were said by a famous french chef cook",
    backstory=dedent("""
        You are a copywriter that lived in France for many years and visited several restaurants. Thus you
        saw and work with several french chef cooks, and you know how they describe the steps of their work, their
        mannerisms, and their accent. Therefore you are totally capable to impersonate one when rewriting the steps of 
        a recipe to sound more like them.   
    """),
    verbose=True,
    memory=True
)

illustrator = Agent(
    role="Illustrator",
    goal="Create a photo-realistic illustration for each the recipe's steps",
    backstory=dedent(""" You are a really capable artist that can draw very photo-realistic illustration 
        for each step of a recipe.
    """),
    tools=[dalle3],
    verbose=True,
    memory=True
)

recipe_task = Task(
    description=dedent("""Given the list of ingredients {ingredients_list} return a food recipe. 
    The recipe must contain all ingredients if possible. The recipes should be simple and have at most 3 steps.
    """),
    expected_output="A recipe with a list of ingredients and their quantities, and a list of steps to execute the recipe",
    agent=recipe_master,
)

rewrite_task = Task(
    description=dedent("""List the ingredients and rewrite the steps of the recipe so they sound like they were said by a french chef cook to 
    their apprentice in English, but with some French accent". Please  keep the name of the recipe and the ingredient list the same as the original, that is, 
    do not rewrite them. Also keep the URLs for the illustrations as provided, do not modify the URL.The final recipe should be formatted in markdown.
    """),
    expected_output="A recipe with a list of ingredients and their quantities, and a list of steps to execute the recipe along with their respective illustations. Everything should be formatted in mardown and images should be displayed inline",
    agent=copywriter,
)

illustrate_task = Task(
    description=dedent(
        """For each step the recipe create a photo-realistic illustration. Please keep the full URL for images as they were provided by the tool"""),
    context=[recipe_task],
    expected_output="The list of ingredients and a list containing each step of recipe along with the link to the its respective illustration. Please keep the entire url for the images, do not short them.",
    agent=illustrator
)

crew = Crew(
    agents=[recipe_master],
    tasks=[recipe_task, illustrate_task, rewrite_task],
    process=Process.sequential,
    full_output=True,
    verbose=True,
)


def run_crew(ingredients: str) -> str:
    return crew.kickoff(inputs={"ingredients_list": ingredients})['final_output']


executor = ThreadPoolExecutor()


async def async_run_crew(ingredients: str) -> str:
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, run_crew, ingredients)
    return result


if __name__ == "__main__":
    response = crew.kickoff(inputs={"ingredients_list": ["gnocchi", "ground beef"]})
    print(response)