Manufacturing AI Assistant with Python

assistant.py

import pandas as pd
from openai import AssistantEventHandler, AzureOpenAI
from typing_extensions import override
import os
from IPython.display import Image
import datetime
import json
from PIL import Image

#Azure Open AI resource settings
api_key = "<use-your-own-api-key>"
azure_endpoint = "https://<use-your-own-aoai-resource>.openai.azure.com/"
deployment_name = "<the-name-of-your-deployment>" # For example: gpt4o-manufacturing

client = AzureOpenAI(
    api_key=api_key,
    api_version="2024-05-01-preview",
    azure_endpoint = azure_endpoint
    )

assistant_instructions="""
You are a helpful AI assistant for the Contoso Motors company in the Manufacturing industry.
You can help with queries related to production, quality, maintenance, and more.
For answering questions, only use the data provided in the uploaded examples.
When a calculation needs to be done, use the code_interpreter tool to do math.
"""

customInstructions = """
The schema of the provided JSON files is the same for every entity, so to determine all of the columns for a graph or table you can get them from the first entity.
Always use Panda's dataframes to display the data in a table format
When possible, create a graph to visualize the data.

Do not ask for confirmation to do the analysis, just go ahead with your intuition, and we will ask back if we need more insights.
Do NOT make up data, only use the data provided in the uploaded examples.

ALWAYS use the code_interpreter tool to load the files and do the analysis.

Only use the attached data to answer the questions.
To get to the answers, use all of the chunks for the uploaded files. Filter on the requested entities first and then do the calculation for all of the matching data.

Uploaded files are always in the JSON format. So no need to check if it's Excel or CSV.

When providing answers in a list or table, also create a graph to visualize this data, do add different colors.
"""

class DateTimeEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()
        return super(DateTimeEncoder, self).default(obj)


# Get the current date and time
current_datetime = datetime.datetime.now()

# Format the datetime as YYYYMMddHHmmss
datetime_prefix = current_datetime.strftime("%Y%m%d%H%M%S")

# Load the CSV file into a DataFrame
manufacturing_data = pd.read_csv('manufacturing_data.csv')

# Convert the DataFrame to a JSON string
manufacturing_json_data = manufacturing_data.to_json(orient='records')

# Save the JSON data to a file for uploading later on
with open('manufacturing_data.json', 'w') as manufacturing_json_file:
    json.dump(json.loads(manufacturing_json_data), manufacturing_json_file, indent=4)

# Create an assistant
assistant = client.beta.assistants.create(
    name="Manufacturing Assistant",
    instructions=assistant_instructions,
    tools=[{"type": "code_interpreter"}],
    model=deployment_name
)

# Create a thread with files
message_file = client.files.create(
    file=open('manufacturing_data.json', "rb"), purpose="assistants"
)

# First, we create a EventHandler class to define
# how we want to handle the events in the response stream. 
class EventHandler(AssistantEventHandler):    
  @override
  def on_error(error):
    print(error)

  @override
  def on_text_created(self, text) -> None:
    print(f"\nassistant > ", end="", flush=True)
      
  @override
  def on_text_delta(self, delta, snapshot):
    print(delta.value, end="", flush=True)
      
  def on_tool_call_created(self, tool_call):
    print(f"\nassistant > {tool_call.type}\n", flush=True)
  
  def on_tool_call_delta(self, delta, snapshot):
    if delta.type == 'code_interpreter':
      if delta.code_interpreter.input:
        print(delta.code_interpreter.input, end="", flush=True)
      if delta.code_interpreter.outputs:
        print(f"code_interpreter > \n", flush=True)
        for output in delta.code_interpreter.outputs:
          if output.type == "image":
            file_id = output.image.file_id  # Get the file_id property
            print(f"\nDownloading image: {file_id}")
            image_data = client.files.content(file_id)  # Download image data using file_id
            image_data_bytes = image_data.read()  # Read image data
            output_dir = "assistant-output"
            if not os.path.exists(output_dir):
              os.makedirs(output_dir)  # Create folder if it doesn't exist
            with open(f"./{output_dir}/image_{file_id}.png", "wb") as file:
              file.write(image_data_bytes)  # Save the image to file
              print(f"Image saved as ./{output_dir}/image_{file_id}.png")
              # Display the image
              # display(Image(filename=f"./{output_dir}/image_{file_id}.png"))
              # Open and display the image
              image = Image.open(f"./{output_dir}/image_{file_id}.png")
              image.show()
          if output.type == "logs":
            print(f"\n{output.logs}", flush=True)

# Create a thread and attach the file to the message
thread = client.beta.threads.create()

# Then, we use the `stream` SDK helper 
# with the `EventHandler` class to create the Run 
# and stream the response.
numberOfRuns = 0
while True:
    followup = input("\nWhat is your follow-up question? ")
    if followup.lower() == "exit":
        break

    # Add user message to the thread
    if numberOfRuns == 0:
        my_thread_message = client.beta.threads.messages.create(
            thread_id = thread.id,
            role = "user",
            content = followup,
            attachments= [
                { "file_id": message_file.id, "tools": [{"type": "code_interpreter"}] }
            ]
        )
        numberOfRuns += 1
    else:
        my_thread_message = client.beta.threads.messages.create(
            thread_id = thread.id,
            role = "user",
            content = followup,
        )
        numberOfRuns += 1
        
    print(f"\nuser > {followup}\n", flush=True)
    with client.beta.threads.runs.stream(
        thread_id=thread.id,
        assistant_id=assistant.id,
        instructions=customInstructions,
        event_handler=EventHandler(),
    ) as stream:
        stream.until_done()

manufacturing_data.csv

requirements.txt

azure-kusto-data==4.5.1
ipykernel==6.29.4
openpyxl==3.1.4
pandas==2.2.2
semantic-kernel==1.2.0
loguru==0.7.2
typing==3.7.4.3
black==24.4.2
isort==5.13.2
pillow==11.0.0
pytest==8.2.2
pytest-cov==5.0.0
pytest-asyncio==0.23.7

What is this?

These files provide a sample to create an AOAI Assistant AI to be used for analysis of the provided Manufacturing data. It can be run by using Python.

How to run?

1. Create a new virtual environment.
   python -m venv myenv
2. Activate the environment
   myenv\Scripts\activate
3. Install all dependencies
   pip install -r requirements.txt
4. Run the script
   python assistant.py

Running the script should load the CSV file, store it as a JSON file, and upload it to the created AOAI Assistant.
The output of the assistant is shown in the terminal and you can also ask questions on the provided data.

Any improvements?

You can add multiple files to an Assistant. So after each question, you might want to provide additional data to analyze. Because this is a sample, used for educational purposes, it is not implemented but can be added.

Important

Take into account, this is sample code, so don't use in Production scenarios!