Amplifying Creativity: Building an AI-Powered Content Creation Assistant — Part 2
Welcome back to another installment of creating an AI-powered content creation assistant. In the previous part of this series, we learned about Jupyter Notebooks, LlamaIndex Workflows, and OpenAI’s models for text and image generation. We built an example workflow that created a blog post, a social media post and generated an image for both platforms. While this example works for a quick prototype, we will expand our solution to create a server that can serve users.
Source code: Github
What to expect:
- Creating a FastAPI server and testing it locally.
- Integrating OpenAI software development kit (SDK)
- Triggering a LlamaIndex workflow from a REST endpoint
Pre-Requisites
- You should be familiar with HTTP concepts
- Basic understanding of programming with Python
- Read the previous post to understand LlamaIndex workflows
- Have an API key from OpenAI and TavilySearch (only if you wish to code along)
FastAPI
FastAPI is a modern Python framework created by Tiangolo. It is purpose-built out of the box to handle asynchronous communication, enforce type safety using Pydantic, and be performant compared to other Python frameworks. With it, we will create a server to respond to clients' requests to create content. In the first iteration, we will create the same basic workflow from part 1 as the focus of this part is to transition from a local notebook to a server that can serve users.
Getting started
Create an empty directory to store the code and associated files. Here is the code structure used for this project:
.
├── README.md
├── app
│ ├── __init__.py
│ ├── config
│ │ ├── __init__.py
│ │ └── settings.py
│ ├── main.py
│ ├── prompts
│ │ ├── __init__.py
│ │ └── prompts.py
│ ├── utils
│ │ ├── __init__.py
│ │ └── utils.py
│ └── workflows
│ ├── __init__.py
│ └── basic_content_workflow.py
├── .env.example
├── .gitignore
└── requirements.txt
- README.md has a general description of the project.
- The app directory contains the source code.
- The config directory holds our settings.py which uses pydantic-settings to manage global configuration settings and secrets.
- main.py our application entry point.
- Prompts contains various prompt templates we supply to models based on step within workflow.
- Utils stores helper functions used throughout the code and will be referenced later in this series.
- Workflows will contain workflows and related files.
- .env.example shows how you should set your env files. Remove .example after updating this file or create a separate .env file.
- .gitignore allows us to ignore files we don’t want to include in version control.
- requirements.txt can be used to install the necessary dependencies for this project.
Virtual Environment and Dependencies
After creating a virtual environment, install the necessary dependencies for this project. I typically use Conda Mini but feel free to use what you are most comfortable with.
conda create -n "content_workflow_demo" python=3.12
conda activate content_workflow_demo
This command will create a virtual environment named “content_workflow_demo” using Python version 3.12. Now that we have a virtual environment to separate our dependencies from our global system let's install the ones needed for this project.
pip install fastapi "uvicorn[standard]" llama-index-core llama-index-llms-openai openai pydantic-settings
With the dependencies installed, let’s start up our FastAPI server. Within the app directory, create main.py and copy the following content.
main.py
from fastapi import FastAPI, status, Depends
from typing import Annotated
from pydantic import BaseModel, Field
from app.workflows.basic_content_workflow import ContentCreationWorkflow
from pydantic_settings import BaseSettings
from functools import lru_cache
from app.config.settings import Settings
class Topic(BaseModel):
topic: str
research: bool = False
@lru_cache
def get_settings():
return Settings()
app = FastAPI()
@app.post("/basic", status_code=status.HTTP_201_CREATED)
async def basic_content_workflow(data: Topic, settings: Annotated[Settings, Depends(get_settings)]) -> str:
workflow = ContentCreationWorkflow(settings=settings, timeout=None, verbose=False)
result = await workflow.run(query=data.topic, research=data.research)
return "workflow complete"
After importing the required modules, we then declare a data model as a class that inherits from BaseModel. With this data model attached to the parameter, FastAPI will:
- Read the body of the request as JSON
- Convert types if needed
- Validate the data
- Generate json schema for our model that can be used elsewhere
- Automated documentation using OpenAPI Schema
We use @lru_cache decorator to load our global settings once. Preventing the need to read our .env file each time a request comes in. The get_settings function returns our Settings object, which holds our OpenAI and Tavily API keys. You’ll see the settings code in the next section.
Then, an instance of the FastAPI class is instantiated, and a path operation decorator @app.post is declared to define a POST endpoint at the path /basic that will return a status code 201 when the workflow is created. We then define an async function that expects a topic as its parameter that will eventually return a string as the response.
settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
class Settings(BaseSettings):
TAVILY_SEARCH_API_KEY: str = Field(env="TAVILY_SEARCH_API_KEY")
OPENAI_API_KEY: str = Field(env="OPENAI_API_KEY")
class Config():
env_file = ".env"
Creating the workflow
from llama_index.core.workflow import Event, Workflow, step, StartEvent, Context, StopEvent
from llama_index.llms.openai import OpenAI as LlamaOpenAI
from openai import OpenAI
import os
import uuid
from app.utils.utils import TavilySearchInput, tavily_search, save_file, generate_image_with_retries
from app.prompts.prompts import *
from PIL import Image
import requests
from io import BytesIO
from app.config.settings import Settings
class ResearchEvent(Event):
query: str
class BlogEvent(Event):
query: str
research: str
class BlogWithoutResearchEvent(Event):
query: str
class SocialMediaEvent(Event):
blog: str
class SocialMediaCompleteEvent(Event):
result: str
class IllustratorEvent(Event):
blog: str
class IllustratorCompleteEvent(Event):
result: str
blog_template = BLOG_TEMPLATE
blog_and_research_template = BLOG_AND_RESEARCH_TEMPLATE
image_prompt_instructions = IMAGE_GENERATION_TEMPLATE
linked_in_template = LINKED_IN_TEMPLATE
class ContentCreationWorkflow(Workflow):
def __init__(self, settings, timeout=None, verbose=False):
super().__init__(timeout, verbose)
self.settings = settings
@step
async def start(self, ctx: Context, ev: StartEvent) -> ResearchEvent | BlogWithoutResearchEvent:
print("Starting content creation", ev.query)
id = str(uuid.uuid4())
if (ev.research) is False:
return BlogWithoutResearchEvent(query=ev.query, uuid=id)
return ResearchEvent(query=ev.query, uuid=id)
@step
async def step_research(self, ctx: Context, ev: ResearchEvent) -> BlogEvent:
print("Researching users query")
search_input = TavilySearchInput(
query=ev.query,
max_results=3,
search_depth="basic")
research = tavily_search(search_input, api_key=self.settings.TAVILY_SEARCH_API_KEY)
return BlogEvent(query=ev.query, research=research, uuid=ev.uuid)
@step
async def step_blog_without_research(self, ctx: Context, ev: BlogWithoutResearchEvent) -> SocialMediaEvent | IllustratorEvent:
print("Writing blog post without research")
print("uuid", ev.uuid)
llm = LlamaOpenAI(model="gpt-4o-mini", api_key=self.settings.OPENAI_API_KEY)
prompt = blog_template.format(query_str=ev.query)
result = await llm.acomplete(prompt, formatted=True)
save_file(result.text, ev.uuid)
print(result)
ctx.send_event(SocialMediaEvent(blog=result.text, uuid=ev.uuid))
ctx.send_event(IllustratorEvent(blog=result.text, uuid=ev.uuid))
@step
async def step_blog(self, ctx: Context, ev: BlogEvent) -> SocialMediaEvent | IllustratorEvent:
print("Writing blog post")
llm = LlamaOpenAI(model="gpt-4o-mini", api_key=self.settings.OPENAI_API_KEY)
prompt = blog_and_research_template.format(query_str=ev.query, research=ev.research)
result = await llm.acomplete(prompt, formatted=True)
save_file(result.text, ev.uuid)
ctx.send_event(SocialMediaEvent(blog=result.text, uuid=ev.uuid))
ctx.send_event(IllustratorEvent(blog=result.text, uuid=ev.uuid))
@step
async def step_social_media(self, ctx: Context, ev: SocialMediaEvent) -> SocialMediaCompleteEvent:
print("Writing social media post")
llm = LlamaOpenAI(model="gpt-4o-mini", api_key=self.settings.OPENAI_API_KEY)
prompt = linked_in_template.format(blog_content=ev.blog)
results = await llm.acomplete(prompt, formatted=True)
save_file(results.text, ev.uuid, type="LinkedIn")
return SocialMediaCompleteEvent(result="LinkedIn post written")
@step
async def step_illustrator(self, ctx: Context, ev:IllustratorEvent) -> IllustratorCompleteEvent:
print("Generating image")
llm = LlamaOpenAI(model="gpt-4o-mini", api_key=self.settings.OPENAI_API_KEY)
image_prompt_instruction_generator = image_prompt_instructions.format(blog_post=ev.blog)
image_prompt = await llm.acomplete(image_prompt_instruction_generator, formatted=True)
client = OpenAI(api_key=self.settings.OPENAI_API_KEY)
response = await generate_image_with_retries(client, image_prompt.text)
image_url = response.data[0].url
response = requests.get(image_url)
image = Image.open(BytesIO(response.content))
directory = f'./posts/{ev.uuid}'
os.makedirs(directory, exist_ok=True)
image.save(f'{directory}/generated_image.png')
return IllustratorCompleteEvent(result="Images drawn")
@step
async def step_collection(self, ctx: Context, ev: SocialMediaCompleteEvent | IllustratorCompleteEvent) -> StopEvent:
if (
ctx.collect_events(
ev,
[SocialMediaCompleteEvent, IllustratorCompleteEvent]
) is None
) : return None
return StopEvent(result="Done")
The only difference from part 1 is we added a settings parameter to the constructor of our ContentCreationWorkflow class so we can pass in our settings object. Similar to part 1, our workflow:
- ContentCreationWorkflow: This is the core workflow that orchestrates content generation:
- start: Determines the workflow path based on whether additional research is requested.
- step_research: Researches to gather additional context.
- step_blog_without_research & step_blog: Generates a blog post, with or without additional research.
- step_social_media: Formats the blog post for social media platforms like LinkedIn.
- step_illustrator: Generates an image based on the blog content.
- step_collection: Collects the outputs and finalizes the workflow.
See the GitHub repository for related utils and prompt instructions code snippets to copy so this workflow works.
What’s next?
We have moved from a local notebook to a server that can serve our use case to real users. However, we will add plenty of enhancements in part 3. For example, the workflow can take some time to run. The user is waiting for this interaction and is not receiving any feedback or able to provide any feedback. Also, as before, we save this to a local folder so the user won’t see their content.
In part 3, we will add web sockets, human-in-the-loop, Postgres, Minio, and Docker-compose to run a local solution that can provide real-time updates. Users can interact with the workflow, and their output is persisted to appropriate storage for later retrieval. All this is orchestrated in a containerized solution using docker-compose.
Summary
In the second part of the series on creating an AI-powered content creation assistant, we build a FastAPI server to handle real user requests. Transitioning from Jupyter Notebooks, we set up a server-side application to generate blog posts, social media content, and images using OpenAI's models.
The post covers setting up FastAPI, integrating the OpenAI SDK, and triggering LlamaIndex workflows through a REST endpoint. We outline the project structure, manage settings with Pydantic, and establish a virtual environment for dependencies.
The workflow generates content with and without research and creates social media posts and illustrative images. The FastAPI server is designed for asynchronous operations, ensuring efficient performance.
This foundation sets the stage for part three, which will introduce web sockets, user interaction enhancements, persistent storage, and Docker-compose orchestration for improved user experience and content management.
Check out my GitHub for the complete implementation. I look forward to any feedback and discussions around AI/ML. I am currently a Software Architect at Groups360. Feel free to connect with me on LinkedIn.
References
- FastAPI: Settings and Environments - https://fastapi.tiangolo.com/advanced/settings/?h=env#the-main-app-file
- FastAPI: Request Body - https://fastapi.tiangolo.com/tutorial/body/