from typing import Any, Dict from openai import OpenAI from string import Template import logging import json class LLMHandler: def __init__(self, api_key, model_name, default_temperature, default_max_tokens, default_system_message=None): self.client = OpenAI(api_key=api_key) self.model_name = model_name self.default_temperature = default_temperature self.default_max_tokens = default_max_tokens self.default_system_message = default_system_message def generate( self, prompt, model_name, temperature=None, max_tokens=None, system_message=None): # optional parameters model_name = model_name or self.model_name temperature = temperature or self.default_temperature max_tokens = max_tokens or self.default_max_tokens system_message = system_message or self.default_system_message # prepare messages messages = [{"role": "user", "content": prompt}] if system_message: messages.insert(0, {"role": "system", "content": system_message}) # get response logging.info(f"Generating text with model {model_name}, temperature {temperature}, and max tokens {max_tokens}...") response = self.client.chat.completions.create( model=model_name, messages=messages, temperature=temperature, max_tokens=max_tokens, seed=1234, response_format={ "type": "json_object" } ) return response.choices[0].message.content def format_personalization(personalization): return '\n'.join([f"- {key}: {value}" for key, value in personalization.items()]) def build_prompt( context: Dict[str, Any], few_shot: str = None, ) -> str: """ Create a detailed prompt incorporating all personalization factors :param section_type: Type of section to generate :param context: Context data including items and customer info :param additional_textual_context: additional context for personalization :return: Formatted prompt string """ if few_shot is not None and len(few_shot) > 0: few_shot = f"Few-shot examples:\n{few_shot}\n" else: few_shot = "" # Create a detailed user prompt with all context and personalization user_prompt = f""" Generate personalized newsletter content based on the following information: Customer Purchase Context: - Previously bought items: {context.pop('bought_items', [])} - Recommended items: {context.pop('recommended_items', [])} Personalization Parameters: {format_personalization(context)} Required Sections: 1. greeting: A warm personalized welcome that considers all personalization parameters. Start with "Hello [customer name]," 2. intro: Acknowledge previous purchases and build connection warmly. Mention the previous purchases. Only acknoweldge previous purchases that are actually present in the context. 3. recommendations: Present recommended items naturally, explaining why they match the customer's preferences and previous purchases. 4. closing: Wrap up maintaining the established tone. General requirements: - Write in first person. - Don't be too formal, maintain a friendly, warm and engaging tone, avoid stuff like "I hope you are doing well". - The reader should feel like they have a special dedicated fashion assistant, who is also a friend. - When mentioning items, use the item name and avoid mentioning colors that are not present in the image, or using adjectives like "vibrant". - The goal should be to make the reader feel special and excited about the items they are being recommended. {few_shot} Please generate all sections in a single, coherent response that maintains consistency in tone and style throughout. Begin each section with the section name in all caps. """ return user_prompt def initialize_newsletter(newsletter_meta_info, transactions, recommendations): # get the paths to the template files newsletter_example_path = newsletter_meta_info['newsletter_example_path'] # load the template from file with open(newsletter_example_path, "r") as f: newsletter_text = Template(f.read()) newsletter_text = newsletter_text.safe_substitute( brand_logo=newsletter_meta_info.get("brand_logo", ""), brand_name=newsletter_meta_info.get("brand_name", ""), ) # iterate over the transactions and replace the placeholders with the actual content for i, transaction in enumerate(transactions): newsletter_text = newsletter_text.replace("${transaction_url}", transaction['product_image_url'], 1) newsletter_text = newsletter_text.replace("${transaction_name}", transaction['product_name'], 1) # iterate over the recommendations and replace the placeholders with the actual content for i, recommendation in enumerate(recommendations): newsletter_text = newsletter_text.replace("${recommendation_url}", recommendation['product_image_url'], 1) newsletter_text = newsletter_text.replace("${recommendation_name}", recommendation['product_name'], 1) return newsletter_text def integrate_personalized_text(newsletter_text, customer_info, textual_sections): # get dctionary from the textual sections json textual_sections = json.loads(textual_sections) logging.info(textual_sections) newsletter_text = Template(newsletter_text) newsletter_text = newsletter_text.safe_substitute( **textual_sections ) # make sure the greeting line is h1, so replace Hello ${customer_name} with

Hello ${customer_name}

customer_name = customer_info.get("customer name") newsletter_text = newsletter_text.replace(f"Hello {customer_name},", f'

Hello {customer_name},

') return newsletter_text def build_context(recommendations, transactions, additional_context, customer_info): # clean recommendations and transactions for rec in recommendations: rec.pop('product_image_url', None) rec.pop('article_id', None) rec.pop('sales_channel_id', None) rec.pop('transaction_date', None) for tr in transactions: tr.pop('product_image_url', None) tr.pop('article_id', None) tr.pop('sales_channel_id', None) tr.pop('transaction_date', None) tr.pop('price', None) context = { 'recommended_items': recommendations, 'bought_items': transactions, 'customer_name': customer_info.get('customer name', 'Unknown'), 'customer_age': customer_info.get('customer age', 'Unknown'), 'last_visit': customer_info.get('last_visit', 'Unknown'), # 'preferred_categories': customer_info.get('preferred_categories', 'Unknown'), # 'shopping_frequency': customer_info.get('shopping_frequency', 'Unknown'), # 'style_preferences': customer_info.get('style_preferences', 'Unknown'), # 'size_preferences': customer_info.get('size_preferences', 'Unknown'), 'preferences': additional_context if len(additional_context) > 1 else 'No additional preferences', } return context