Custom AI Workflows: Multi-Step Pipelines Using xAI

project/
|
├── main.py                # Streamlit application
├── helpers/
│   ├── image_utils.py     # Functions for image encoding
│   ├── api_utils.py       # API interaction with xAI
│   └── __init__.py        # Package initialization
├── requirements.txt       # Python dependencies
└── .env                   # Environment variables

import base64

def encode_image_to_base64(image_path: str) -> str:
    """
    Encode an image to a Base64 string.

    Args:
        image_path (str): Path to the image file.

    Returns:
        str: Base64 encoded string of the image.
    """
    with open(image_path, 'rb') as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

from dotenv import load_dotenv
from openai import OpenAI
import os

# Load environment variables
load_dotenv()

# Configure the OpenAI client for xAI's Grok API
api_key = os.getenv("XAI_API_KEY")
client = OpenAI(
    api_key=api_key,
    base_url="https://api.x.ai/v1",
)

def extract_text_from_image(image_base64: str) -> str:
    """
    Extract text from an image using Grok Vision API.

    Args:
        image_base64 (str): Base64 encoded image string.

    Returns:
        str: Extracted text from the image.
    """
    response = client.chat.completions.create(
        model="grok-vision-beta",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": "Extract and give me the text in the image provided, and nothing else no intro"
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_base64}"
                        }
                    }
                ]
            }
        ],
        temperature=0.7,
        max_tokens=500,
        n=1
    )
    return response.choices[0].message.content.strip()

def summarize_text(text: str) -> str:
    """
    Summarize extracted text using Grok API.

    Args:
        text (str): Extracted text.

    Returns:
        str: Summary of the text.
    """
    response = client.chat.completions.create(
        model="grok-beta",
        messages=[
            {
                "role": "user",
                "content": f"Summarize the following text:\n\n{text}"
            }
        ],
        temperature=0.7,
        max_tokens=200,
        n=1
    )
    return response.choices[0].message.content.strip()

def generate_insights(summary: str) -> str:
    """
    Generate actionable insights from a summary using Grok API.

    Args:
        summary (str): Summary of the text.

    Returns:
        str: Actionable insights derived from the summary.
    """
    response = client.chat.completions.create(
        model="grok-beta",
        messages=[
            {
                "role": "user",
                "content": f"Based on the following summary, generate actionable insights:\n\n{summary}"
            }
        ],
        temperature=0.7,
        max_tokens=200,
        n=1
    )
    return response.choices[0].message.content.strip()

import streamlit as st
from helpers.image_utils import encode_image_to_base64
from helpers.api_utils import extract_text_from_image, summarize_text, generate_insights

# Initialize session state variables
if "extracted_text" not in st.session_state:
    st.session_state.extracted_text = ""
if "summary" not in st.session_state:
    st.session_state.summary = ""
if "insights" not in st.session_state:
    st.session_state.insights = ""

st.title("Image-to-Insights Workflow with Grok Vision")

# File uploader
uploaded_file = st.file_uploader("Upload an image", type=["png", "jpg", "jpeg"])

if uploaded_file:
    # Display the uploaded image
    st.image(uploaded_file, caption="Uploaded Image", use_column_width=True)

    # Save the image to a temporary location
    with open("temp_image.jpg", "wb") as f:
        f.write(uploaded_file.read())

    # Encode image to Base64
    st.info("Encoding image...")
    image_base64 = encode_image_to_base64("temp_image.jpg")
    st.success("Image encoded successfully!")

    # Call API to extract text
    st.info("Extracting text from the image...")
    st.session_state.extracted_text = extract_text_from_image(image_base64)
    st.success("Text extracted successfully!")

# Display the extracted text
if st.session_state.extracted_text:
    st.subheader("Extracted Text:")
    st.text_area("Extracted Text", st.session_state.extracted_text, height=200)

    # Summarize the extracted text
    if st.button("Summarize Text"):
        st.info("Summarizing text...")
        st.session_state.summary = summarize_text(st.session_state.extracted_text)
        st.success("Text summarized successfully!")

# Display the summary
if st.session_state.summary:
    st.subheader("Summary:")
    st.text_area("Summary", st.session_state.summary, height=150)

    # Generate insights from the summary
    if st.button("Generate Insights"):
        st.info("Generating actionable insights...")
        st.session_state.insights = generate_insights(st.session_state.summary)
        st.success("Insights generated successfully!")

# Display the insights
if st.session_state.insights:
    st.subheader("Actionable Insights:")
    st.text_area("Insights", st.session_state.insights, height=150)