Generators and LLM Streaming¶

def count_to_3():
    yield 1
    yield 2
    yield 3


for num in count_to_3():
    print(num)
    #> 1
    #> 2
    #> 3

import time


def expensive_func(x):
    """Simulate an expensive operation."""
    time.sleep(1)
    return x**2


def calculate_time_for_first_result_with_list(func_input, func):
    """Calculate using a list comprehension and return the first result with its computation time."""
    start_perf = time.perf_counter()
    result = [func(x) for x in func_input][0]
    end_perf = time.perf_counter()
    print(f"Time for first result (list): {end_perf - start_perf:.2f} seconds")
    #> Time for first result (list): 5.02 seconds
    return result


def calculate_time_for_first_result_with_generator(func_input, func):
    """Calculate using a generator and return the first result with its computation time."""
    start_perf = time.perf_counter()
    result = next(func(x) for x in func_input)
    end_perf = time.perf_counter()
    print(f"Time for first result (generator): {end_perf - start_perf:.2f} seconds")
    #> Time for first result (generator): 1.00 seconds
    return result


# Prepare inputs for the function
numbers = [1, 2, 3, 4, 5]

# Benchmarking
first_result_list = calculate_time_for_first_result_with_list(numbers, expensive_func)
first_result_gen = calculate_time_for_first_result_with_generator(
    numbers, expensive_func
)

squares = (x * x for x in range(10))

from openai import OpenAI

# Set your OpenAI API key
client = OpenAI(
    api_key="My API Key",
)

response_generator = client.chat.completions.create(
    model='gpt-3.5-turbo',
    messages=[{'role': 'user', 'content': "What are some good reasons to smile?"}],
    temperature=0,
    stream=True,
)

for chunk in response_generator:
    print(chunk.choices[0].delta.content, end="")

profile_data = """
Customer ID: 12345
Recent Purchases: [Laptop, Wireless Headphones, Smart Watch]
Frequently Browsed Categories: [Electronics, Books, Fitness Equipment]
Product Ratings: {Laptop: 5 stars, Wireless Headphones: 4 stars}
Recent Search History: [best budget laptops 2023, latest sci-fi books, yoga mats]
Preferred Brands: [Apple, AllBirds, Bench]
Responses to Previous Recommendations: {Philips: Not Interested, Adidas: Not Interested}
Loyalty Program Status: Gold Member
Average Monthly Spend: $500
Preferred Shopping Times: Weekend Evenings
...
"""

products = [
    {
        "product_id": 1,
        "product_name": "Apple MacBook Air (2023) - Latest model, high performance, portable",
    },
    {
        "product_id": 2,
        "product_name": "Sony WH-1000XM4 Wireless Headphones - Noise-canceling, long battery life",
    },
    {
        "product_id": 3,
        "product_name": "Apple Watch Series 7 - Advanced fitness tracking, seamless integration with Apple ecosystem",
    },
    {
        "product_id": 4,
        "product_name": "Kindle Oasis - Premium e-reader with adjustable warm light",
    },
    {
        "product_id": 5,
        "product_name": "AllBirds Wool Runners - Comfortable, eco-friendly sneakers",
    },
    {
        "product_id": 6,
        "product_name": "Manduka PRO Yoga Mat - High-quality, durable, eco-friendly",
    },
    {
        "product_id": 7,
        "product_name": "Bench Hooded Jacket - Stylish, durable, suitable for outdoor activities",
    },
    {
        "product_id": 8,
        "product_name": "GoPro HERO9 Black - 5K video, waterproof, for action photography",
    },
    {
        "product_id": 9,
        "product_name": "Nespresso Vertuo Next Coffee Machine - Quality coffee, easy to use, compact design",
    },
    {
        "product_id": 10,
        "product_name": "Project Hail Mary by Andy Weir - Latest sci-fi book from a renowned author",
    },
]

import instructor
from openai import OpenAI
from typing import Iterable
from pydantic import BaseModel

client = instructor.from_openai(OpenAI(), mode=instructor.function_calls.Mode.JSON)


class ProductRecommendation(BaseModel):
    product_id: str
    product_name: str


Recommendations = Iterable[ProductRecommendation]

prompt = (
    f"Based on the following user profile:\n{profile_data}\nRank the following products from most relevant to least relevant:\n"
    + '\n'.join(
        f"{product['product_id']} {product['product_name']}" for product in products
    )
)

start_perf = time.perf_counter()
recommendations_stream = client.chat.completions.create(
    model="gpt-3.5-turbo-1106",
    temperature=0.1,
    response_model=Iterable[ProductRecommendation],
    stream=True,
    messages=[
        {
            "role": "system",
            "content": "Generate product recommendations based on the customer profile. Return in order of highest recommended first.",
        },
        {"role": "user", "content": prompt},
    ],
)
for product in recommendations_stream:
    print(product)
    end_perf = time.perf_counter()
    print(f"Time for first result (generator): {end_perf - start_perf:.2f} seconds")
    break

start_perf = time.perf_counter()
recommendations_list = client.chat.completions.create(
    model="gpt-3.5-turbo-1106",
    temperature=0.1,
    response_model=Iterable[ProductRecommendation],
    stream=False,
    messages=[
        {
            "role": "system",
            "content": "Generate product recommendations based on the customer profile. Return in order of highest recommended first.",
        },
        {"role": "user", "content": prompt},
    ],
)
print(recommendations_list[0])
end_perf = time.perf_counter()
print(f"Time for first result (list): {end_perf - start_perf:.2f} seconds")