app.py

# App.py File.
import base64
import logging
import numpy as np
import os
import langchain
import base64
import gradio as gr
import shutil
import json
import re
from pathlib import Path
from openai import OpenAI
import soundfile as sf
from pydub import AudioSegment
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.chains import TransformChain
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain import globals
from langchain_core.runnables import chain
from langchain_core.output_parsers import JsonOutputParser
from langchain.memory import ConversationSummaryBufferMemory, ConversationBufferMemory

os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

client = OpenAI()

# Set up logging
logging.basicConfig(level=logging.INFO)
def transform_text_to_speech(text: str, user):
  # Generate speech from transcription
  speech_file_path_mp3 = Path.cwd() / f"{user}-speech.mp3"
  speech_file_path_wav = Path.cwd() / f"{user}-speech.wav"
  response = client.audio.speech.create(
                model="tts-1",
                voice="onyx",
                input=text
            )

  with open(speech_file_path_mp3, "wb") as f:
      f.write(response.content)

  # Convert mp3 to wav
  audio = AudioSegment.from_mp3(speech_file_path_mp3)
  audio.export(speech_file_path_wav, format="wav")

  # Read the audio file and encode it to base64
  with open(speech_file_path_wav, "rb") as audio_file:
      audio_data = audio_file.read()
      audio_base64 = base64.b64encode(audio_data).decode('utf-8')

  # Create an HTML audio player with autoplay
  audio_html = f"""
  <audio controls autoplay>
      <source src="data:audio/wav;base64,{audio_base64}" type="audio/wav">
      Your browser does not support the audio element.
  </audio>
  """
  return audio_html


def transform_speech_to_text(audio, user):
  file_path = f"{user}-saved_audio.wav"
  sample_rate, audio_data = audio
  sf.write(file_path, audio_data, sample_rate)
  # Transcribe audio
  with open(file_path, "rb") as audio_file:
      transcription = client.audio.transcriptions.create(
          model="whisper-1",
          file=audio_file
      )
  return transcription.text



def load_image(inputs: dict) -> dict:
    """Load image from file and encode it as base64."""
    image_path = inputs["image_path"]

    def encode_image(image_path):
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    image_base64 = encode_image(image_path)
    return {"image": image_base64}

from langchain.chains import TransformChain
load_image_chain = TransformChain(
    input_variables=["image_path"],
    output_variables=["image"],
    transform=load_image
)

class GenerateQuestion(BaseModel):
 """Information about an image."""
 question: str = Field(description= "React to the user input and ask a follow back question, using conversation and photo provded as a guide.")

class GenerateDescription(BaseModel):
 """Information about an image."""
 description: str = Field(description= "A description of the people and context in the photo in 2 lines and a question to start converstion around the photograph")

question_parser = JsonOutputParser(pydantic_object=GenerateQuestion)
description_parser = JsonOutputParser(pydantic_object=GenerateDescription)

# Set verbose
# globals.set_debug(True)

@chain
def image_model(inputs: dict) -> str | list[str] | dict:
 """Invoke model with image and prompt."""
 model = ChatOpenAI(temperature=0.5, model="gpt-4o", max_tokens=1024)
 msg = model.invoke(
             [HumanMessage(
             content=[
             {"type": "text", "text": inputs["prompt"]},
             {"type": "text", "text": inputs["parser"].get_format_instructions()},
             {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{inputs['image']}"}},
             ])]
             )
 return msg.content

CONVERSATION_STARTER_PROMPT = """
  Given the image uploaded by a old person named {name}.
  You are Studs Terkel, and your role is to be a curious friend who is genuinely interested in the story behind the photograph that the older person has provided.
  Provide the following information,
  - A description of the image in 2 lines and a question that gives context to the photograph.
  """


CONVERSATION_EXPANDING_PROMPT = """
  Given the image uploaded by a old person named {name}.
  Here is the conversation history around the Image between {name} and Studs Terkel:
  {history}
  You are Studs Terkel, and your role is to be a curious friend who is genuinely interested in the story behind the image uploaded by the {name}.
  Your task is to use your external knowledge and conversation to respond to {name} recent reply and ask a question that encourages the person to expand on their answer about the photograph. Ask for more details or their feelings about the situation depicted in the photograph. Use the conversation history provided above and ask only one question at a time.
  Use your knowledge to respond with the information.
  Studs Terkel:
  """


CONVERSATION_ENDING_PROMPT = """
  Given the image uploaded by a old person named {name}.
  Here is the conversation history around the Image between {name} and Stud's Terkel:
  {history}
  You are Studs Terkel, and your role is to be a curious friend who is genuinely interested in the story behind the image uploaded by the {name}.
  Your task is to use your external knowledge , conversation history, image uploaded to respond to {name} recent reply and ask if they would like to tell more about the story depicted in the photograph, discuss anything that the photograph reminds them of, or if they are ready to move on to another photograph or stop reminiscing. Use the conversation history provided above and ask only one question at a time.
  Studs Terkel:
  """

def get_prompt(image_path: str, iter: int, memory: str, firstname: str) -> dict:

   if iter == 1:
    parser = description_parser
    prompt = CONVERSATION_STARTER_PROMPT.format(name=firstname)
   elif  iter >= 2 and iter <= 5:
    parser = question_parser
    prompt=   CONVERSATION_EXPANDING_PROMPT.format(name= firstname, history=memory)
   else:
    parser = question_parser
    prompt=  CONVERSATION_ENDING_PROMPT.format(name= firstname, history=memory)

   vision_chain = load_image_chain | image_model | question_parser
   return vision_chain.invoke({'image_path': f'{image_path}', 'prompt': prompt, 'parser':parser})



def retrieve_memory(input_filepath, name):
  with open(input_filepath, 'r') as f:
    conversation = f.read()
  lines = conversation.strip().split('\n')
  last_reply = None

  # Loop through the lines from the end
  for line in reversed(lines):
      if re.match(r'(Studs Terkel|' + re.escape(name) + '):', line):
        last_reply = line
        break

  # Determine who made the last reply, split it based on the colon, and return JSON
  if last_reply:
      speaker, message = last_reply.split(":", 1)
      result = {
          "speaker": speaker.strip(),
          "reply": message.strip()
      }
      return result
  else:
      result = {
          "speaker": "",
          "reply": ""
      }
      return result

def load_counts(count_file_path):
    if os.path.exists(count_file_path):
        with open(count_file_path, 'r') as f:
            return json.load(f)
    return {"count": 0}

def save_counts(count_file_path, counts):
    with open(count_file_path, 'w') as f:
        json.dump(counts, f)

def increment_counts(count_file_path):
    counts = load_counts(count_file_path)
    counts["count"] += 1
    save_counts(count_file_path, counts)
    return counts["count"]

def pred(user_name, image_path, audio):

    if image_path and user_name.strip() != "":
        user_name = user_name.strip()
        image_name = image_path.split("/")[-1]
        new_image_name = f"{user_name}-{image_name}"
        new_image_path = f"/data/{new_image_name}"
        input_filename = f"{user_name}-{image_name}-conversation-memory.txt"
        input_filepath = f"/data/{input_filename}"
        count_file_path = f"/data/{user_name}-{image_name}-tracking.json"

        if not os.path.exists(new_image_path):
          shutil.copy(image_path, new_image_path)
          iter = increment_counts(count_file_path)
          output = get_prompt(new_image_path, iter, None, user_name)
          res = output["description"]
          with open(input_filepath, 'w') as f:
            f.write("Studs Terkel: " + res)
          return "New Photo Uploaded" , res, transform_text_to_speech(res, user_name)

        else:

          if audio is not None:
            user_input = transform_speech_to_text(audio, user_name)

            iter = increment_counts(count_file_path)
            with open(input_filepath, 'a') as f:
                f.write("\n" + user_name + ": " + user_input)
            with open(input_filepath, 'r') as f:
                content = f.read()
            output = get_prompt(new_image_path, iter, content, user_name)
            res = output["question"]
            with open(input_filepath, 'a') as f:
                f.write("\n" + "Studs Terkel: "+ res)
            return  user_input, res, transform_text_to_speech(res, user_name)




          # decide the path from the contents of the conversation memory.
          if  os.path.exists(input_filepath):
              res = retrieve_memory(input_filepath, user_name)
              if res["speaker"] == "Studs Terkel":
                  if audio is None:
                    message = ""
                    return  "Loading response, please wait...", "Loading response, please wait...", None
                  prefix = "Continuing from where we left off: "
                  return   "" ,  prefix+ res["reply"], transform_text_to_speech(prefix+res["reply"], user_name)
              else:
                  with open(input_filepath, 'a') as f:
                      f.write("\n" + user_name + ": " + "I'd like to continue our conversation about this photograph.")
                  with open(input_filepath, 'r') as f:
                      content = f.read()
                  iter = increment_counts(count_file_path)
                  output = get_prompt(new_image_path, iter, content, user_name)
                  res = output["question"]
                  with open(input_filepath, 'a') as f:
                      f.write("\n" + "Studs Terkel: "+ res)
                  return  "I'd like to continue our conversation about this photograph.", res, transform_text_to_speech(res, user_name)

    message = "Great! Please enter your name and upload a photo to share your story."
    return   "", message, None

# Backend function to clear inputs
def clear_inputs(user_name, image_path):
  message = "Great! Please enter your name and upload a photo to share your story."
  if user_name.strip() == "" or image_path == None:
      return None, None, "", message, transform_text_to_speech(message, user_name)

  image_name = image_path.split("/")[-1]
  input_filename = f"{user_name}-{image_name}-conversation-memory.txt"
  input_filepath = f"/data/{input_filename}"
  if  os.path.exists(input_filepath):
    with open(input_filepath, 'a') as f:
      f.write("\n" + f"{user_name}: " + "new photo uploaded")

  
  return None, None, "", message,  None

# Gradio Interface
with gr.Blocks(title = "KitchenTable.AI") as demo:
    with gr.Row():
        with gr.Column():
            clear_button = gr.Button("Tell a new Story", elem_id="clear-button")
            # Input fields
            username = gr.Textbox(label="Enter your first name")
            image_input = gr.Image(type="filepath", label="Upload an Image")  # Removed the extra comma
            audio_input = gr.Audio(sources="microphone", type="numpy", label="Audio Recorded")
            # text_input = gr.Textbox(label="Input here...")
            # submit_button = gr.Button("Submit")


        with gr.Column():
            # Output fields
            user_input_output = gr.Textbox(label="User Input")
            stud_output = gr.Textbox(label="Studs Terkel")
            audio_output = gr.HTML(label="Audio Player")

    audio_input.change(pred, inputs=[username, image_input, audio_input], outputs=[ user_input_output, stud_output, audio_output])
    image_input.change(pred, inputs=[username, image_input, audio_input], outputs=[ user_input_output, stud_output, audio_output])

    # Linking the submit button with the save_audio function
    # submit_button.click(fn=pred, inputs=[username, image_input, audio_input],
    #                     outputs=[ user_input_output, stud_output, audio_output])

    # Linking the clear button with the clear_inputs function
    clear_button.click(fn=clear_inputs, inputs=[username, image_input], outputs=[image_input, audio_input, user_input_output, stud_output, audio_output])

# Launch the interface
demo.launch(share=True, debug=True)