--init

app.py

@@ -1,64 +1,126 @@
 import gradio as gr
-from huggingface_hub import InferenceClient
+from email_generator.main import loop_email_workflow, EMAIL_EVALUATOR_PROMPT, EMAIL_GENERATOR_PROMPT
+import json
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+# Function to generate the email
+def generate_email_workflow(persona_json: str, campaign_json: str, sender_json: str, max_attempts: int, openai_api_key: str):
+    """
+    Generate a complete email with persona, campaign, and sender details.
 
+    Args:
+        persona_json (str): A JSON string representing the persona.
+        campaign_json (str): A JSON string representing the campaign details.
+        sender_json (str): A JSON string representing the sender details.
+        max_attempts (int): Maximum number of attempts for generating a valid email.
+        openai_api_key (str): The API key for OpenAI, if applicable.
 
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
+    Returns:
+        str: The complete generated email or an error message.
+    """
+    try:
+        # Parse JSON strings to dictionaries
+        persona = json.loads(persona_json)
+        campaign = json.loads(campaign_json)
+        sender = json.loads(sender_json)
 
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
+        # Determine the model to use based on the API key
+        use_huggingface = not bool(openai_api_key)
+        model_used = "HuggingFace (Zephyr-7B)" if use_huggingface else "OpenAI (gpt-3.5-turbo)"
 
-    messages.append({"role": "user", "content": message})
+        # Run the email generation workflow
+        result = loop_email_workflow(
+            persona=persona,
+            campaign=campaign,
+            sender_data=sender,
+            evaluator_prompt=EMAIL_EVALUATOR_PROMPT,
+            generator_prompt=EMAIL_GENERATOR_PROMPT,
+            max_tries=max_attempts,
+            use_huggingface=use_huggingface,
+            openai_api_key=openai_api_key if not use_huggingface else None,
+        )
 
-    response = ""
+        if not result["final_email"]:
+            return f"Failed to generate a valid email after {max_attempts} attempts. Feedback: {result.get('message', 'No additional information.')}\n\nModel Used: {model_used}"
 
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
+        # Add sender information to the email content
+        generated_email = result["final_email"]
 
-        response += token
-        yield response
+        return generated_email
 
+    except json.JSONDecodeError:
+        return "Invalid JSON format. Please ensure all inputs are valid JSON."
+    except Exception as e:
+        return f"Error: {e}"
 
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
+
+# Create Gradio interface
+persona_input = gr.Textbox(
+    label="Enter Persona (JSON format)",
+    lines=10,
+    value='{"name": "John", "city": "New York", "hobbies": "Reading"}',
+    placeholder='{"name": "John", "city": "New York", "hobbies": "Reading"}'
+)
+campaign_input = gr.Textbox(
+    label="Enter Campaign Details (JSON format)",
+    lines=10,
+    value='{"subject_line": "Discover Our New Product!", "product": "Backpacks", "discount": "20%", "validity": "Until January 31, 2025"}',
+    placeholder='{"subject_line": "Discover Our New Product!", "product": "Backpacks", "discount": "20%", "validity": "Until January 31, 2025"}'
+)
+sender_input = gr.Textbox(
+    label="Enter Sender Details (JSON format)",
+    lines=5,
+    value='{"name": "Jane Doe", "company": "Outdoor Gear Co."}',
+    placeholder='{"name": "Jane Doe", "company": "Outdoor Gear Co.", "cta_text": "Shop Now", "cta_link": "https://example.com"}'
+)
+max_attempts_input = gr.Slider(
+    label="Max Attempts",
+    minimum=1,
+    maximum=10,
+    step=1,
+    value=3,
+    interactive=True
 )
+openai_api_key_input = gr.Textbox(
+    label="Enter OpenAI API Key (Leave blank to use HuggingFace Zephyr-7B Beta)",
+    type="password",
+    placeholder="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+)
+email_output = gr.Textbox(
+    label="Generated Email",
+    lines=15,
+    interactive=False,
+)
+
+# Interface layout
+with gr.Blocks() as interface:
+    gr.Markdown(
+        """
+        # Personalized Email Generator
+        Generate a personalized email based on user persona and campaign details. 
+        Provide the inputs in JSON format and specify the maximum number of attempts for generation.
+
+        ### Available Models:
+        - **OpenAI (gpt-3.5-turbo)**: A highly advanced language model known for its accuracy and contextual understanding, ideal for generating professional and creative emails.
+        - **HuggingFace Zephyr-7B Beta**: An open-source model optimized for text generation tasks, offering a cost-effective alternative to proprietary APIs.
+        """
+    )
+    with gr.Row():
+        with gr.Column():
+            persona_input.render()
+            campaign_input.render()
+            sender_input.render()
+            max_attempts_input.render()
+            openai_api_key_input.render()
+        with gr.Column():
+            email_output.render()
 
+    generate_button = gr.Button("Generate Email")
+    generate_button.click(
+        fn=generate_email_workflow,
+        inputs=[persona_input, campaign_input, sender_input, max_attempts_input, openai_api_key_input],
+        outputs=email_output,
+    )
 
+# Launch the app
 if __name__ == "__main__":
-    demo.launch()
+    interface.launch()

email_generator/main.py

@@ -0,0 +1,395 @@
+from pydantic import BaseModel
+from typing import Literal
+from pydantic import ValidationError
+from rich.console import Console
+from rich.logging import RichHandler
+import logging
+import re
+from openai import OpenAI
+import os
+from dotenv import load_dotenv
+from huggingface_hub import InferenceClient
+from typing import List, Optional
+
+# Load environment variables
+load_dotenv()
+
+def initialize_client(api_key=None):
+    """Initialize OpenAI client if API key is provided."""
+    if api_key:
+        return OpenAI(api_key=api_key)
+    return None
+
+# Setup logging
+console = Console()
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+    handlers=[RichHandler(console=console)]
+)
+logger = logging.getLogger("email_agent")
+
+EMAIL_GENERATOR_PROMPT = """
+Your goal is to write a personalized email for the user based on the provided persona, campaign, and sender details. 
+If there are feedback points from previous generations, you should reflect on them to improve your solution.
+
+Persona:
+{persona}
+
+Campaign Details:
+{campaign}
+
+Sender Details:
+{sender}
+
+**Output Format Requirement**: The response must strictly adhere to the following format. Ensure that:
+1. All opening tags have corresponding closing tags.
+2. The content inside each tag is complete and relevant to the provided details.
+3. Do not use placeholders such as `[Insert "Shop Now" button]`, `[Company Logo]`, or `[Unsubscribe link]`.
+
+```
+<thoughts>
+[Include your understanding of the persona, campaign, sender details.]
+</thoughts>
+
+<email>
+[Your email content here,without any placeholders or incomplete references.]
+</email>
+```
+Important: The tags <thoughts> and <email> must always be properly closed.
+"""
+
+EMAIL_EVALUATOR_PROMPT1 = """
+Evaluate the provided email content using the following criteria:
+1. **Personalization Accuracy**: Does the email reflect the persona details and campaign details?
+2. **Tone and Style**: Is the tone engaging and appropriate for the persona? Does it align with the persona's characteristics?
+3. **Clarity and Readability**: Is the email easy to read, with clear and concise sentences? Does it avoid ambiguity and jargon?
+
+**Instructions:**
+- Only output "PASS" if all criteria are met with no room for improvement.
+- If the email does not meet the criteria, output "NEEDS_IMPROVEMENT" or "FAIL", followed by specific feedback.
+
+**Output Format:**
+{{"evaluation": "<PASS | NEEDS_IMPROVEMENT | FAIL>", "feedback": "<Provide specific feedback explaining what needs to be improved and why.>"}}
+
+Persona:
+{persona}
+
+Campaign Details:
+{campaign}
+
+Sender Details:
+{sender}
+
+Email Content:
+{generated_content}
+"""
+
+EMAIL_EVALUATOR_PROMPT = """
+Evaluate email against these criteria:
+1. Personalization: Match with persona & campaign
+2. Tone: Appropriate for persona
+3. Clarity: Readable, concise language
+
+Scoring:
+- Personalization (0-10)
+- Tone Alignment (0-10)
+- Readability (0-10)
+
+Output Format:
+{{"evaluation": "<PASS | NEEDS_IMPROVEMENT | FAIL>","feedback": {{"personalization_score": 0,"tone_alignment_score": 0,"readability_score": 0,"improvements": ["Suggestion 1", "Suggestion 2"]}}}}
+
+Persona: {persona}
+Campaign: {campaign}
+Sender: {sender}
+Email: {generated_content}
+"""
+
+def JSON_llm(prompt: str, openai_api_key: str = None, use_huggingface: bool = False, schema: BaseModel = None) -> dict:
+    """
+    Calls the LLM to generate a response and validates it against a given schema.
+
+    Args:
+        prompt (str): The input prompt for the LLM.
+        schema (BaseModel): A pydantic schema for validating the LLM's output.
+
+    Returns:
+        dict: The validated response from the LLM.
+
+    Raises:
+        ValidationError: If the response doesn't match the schema.
+    """
+    # Example: Use llm_call or a similar function to generate a response
+    raw_response = llm_call(prompt,model="gpt-3.5-turbo", api_key=openai_api_key, use_huggingface=use_huggingface)
+
+    try:
+        # Parse and validate the response against the schema
+        parsed_response = schema.parse_raw(raw_response)
+        return parsed_response.dict()
+    except ValidationError as e:
+        # Log or handle the validation error
+        logger.error(f"Validation failed: {e}")
+        logger.error(f"Raw response: {raw_response}")
+        raise ValueError(f"Invalid response format: {raw_response}") from e
+
+def extract_response_content(generated_text: str) -> str:
+    # Extract content after "Response:"
+    response_match = re.search(r"Response:\s*(.*)", generated_text, re.DOTALL)
+    return response_match.group(1).strip() if response_match else ""
+
+
+
+def llm_call(prompt: str, model: str = "gpt-3.5-turbo", api_key: str = None, use_huggingface: bool = False) -> str:
+    """
+    Call the LLM model (OpenAI or an open-source alternative) and return the response.
+    """
+    if api_key and not use_huggingface:
+        console.print("Using OpenAI model.")
+        client = initialize_client(api_key)
+        messages = [{"role": "user", "content": prompt}]
+        print("---messages", messages)
+        response = client.chat.completions.create(
+            model=model,
+            messages=messages,
+        )
+
+        return response.choices[0].message.content
+
+    elif use_huggingface:
+        console.print("Using Hugging Face model.")
+        model = "Qwen/Qwen2.5-72B-Instruct"
+        hf_client = InferenceClient(model)
+        messages = [{"role": "user", "content": prompt}]
+        response = ""
+        for message in hf_client.chat_completion(
+            messages,
+            max_tokens=900,
+            stream=True,
+            temperature=0.4,
+            top_p=0.95,
+        ):
+            token = message.choices[0].delta.content
+            response += token
+        return response
+
+    else:
+        console.print("Using default simulated response.")
+        # Simulated response matching the schema for evaluation
+        return '{"evaluation": "NEEDS_IMPROVEMENT", "feedback": "Simulated fallback response for testing purposes."}'
+
+
+def extract_xml(text: str, tag: str) -> str:
+    """
+    Extracts the content of the specified XML tag from the given text.
+
+    Args:
+        text (str): The text containing the XML.
+        tag (str): The XML tag to extract content from.
+
+    Returns:
+        str: The content of the specified XML tag, or an empty string if the tag is not found.
+    """
+    match = re.search(f'<{tag}>(.*?)</{tag}>', text, re.DOTALL)
+    return match.group(1) if match else ""
+
+def extract_xml(text: str, tag: str) -> str:
+    """
+    Extracts the content of the specified XML tag from the given text.Next tip
+    
+
+    Args:
+        text (str): The text containing the XML.
+        tag (str): The XML tag to extract content from.
+
+    Returns:
+        str: The content of the specified XML tag, or an empty string if the tag is not found.
+    """
+    match = re.search(f'<{tag}>(.*?)</{tag}>', text, re.DOTALL)
+    return match.group(1) if match else ""
+
+def generate_email(persona: dict, campaign: dict,sender_data: dict , generator_prompt: str, context: str = "", openai_api_key: str = None, use_huggingface: bool = False) -> tuple[str, str]:
+    """Generate a personalized email based on persona, campaign details, and feedback."""
+    # Dynamically build the persona and campaign text from the dictionaries
+    persona_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in persona.items()])
+    campaign_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in campaign.items()])
+    sender_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in sender_data.items()])
+    full_prompt = generator_prompt.format(persona=persona_text, campaign=campaign_text,sender=sender_text)
+    if context:
+        full_prompt += f"\nFeedback: {context}"
+    console.print("Generating email using LLM...")
+    console.print(f"Prompt: {full_prompt}")
+    response = llm_call(full_prompt, model="gpt-3.5-turbo", api_key=openai_api_key, use_huggingface=use_huggingface)
+    console.print("Generated email response.")
+    console.print("[bold green]Generated Email Output:[/bold green]")
+    console.print(response)
+    return response
+
+def evaluate_email(persona: dict, campaign: dict,sender_data: dict , evaluator_prompt: str, generated_content: str,openai_api_key: str = None, use_huggingface: bool = False):
+    """Evaluate if a generated email meets requirements."""
+    try:
+        print("evaluator_prompt type:", type(evaluator_prompt))
+        
+        # Validate inputs
+        if not persona:
+            raise ValueError("Persona is required")
+        if not campaign:
+            raise ValueError("Campaign is required")
+        if not generated_content:
+            raise ValueError("Generated content is required")
+        if sender_data is None:
+            raise ValueError("Sender data is required")
+
+        # Dynamically build text representations
+        persona_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in persona.items()])
+        campaign_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in campaign.items()])
+        sender_text = "\n".join([f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in sender_data.items()])
+
+        # Format the prompt
+        full_prompt = evaluator_prompt.format(
+            persona=persona_text, 
+            campaign=campaign_text,
+            sender=sender_text, 
+            generated_content=generated_content
+        )
+
+    except Exception as e:
+        # Catch and print any exceptions
+        import traceback
+        traceback.print_exc()
+        logger.error(f"Error in evaluate_email: {e}")
+        print(f"Error details: {e}")
+        raise
+
+    # Build a schema for evaluation
+    class Evaluation(BaseModel):
+        evaluation: Literal["PASS", "NEEDS_IMPROVEMENT", "FAIL"]
+        feedback: Optional[dict] = {
+        "personalization_score": 0,
+        "tone_alignment_score": 0,
+        "readability_score": 0,
+        "improvements": []
+    }
+    console.print("Evaluating generated email...")
+    response = JSON_llm(full_prompt, openai_api_key, use_huggingface, Evaluation)
+    print("Email evaluation complete.", response)
+    evaluation = response["evaluation"]
+    feedback = response["feedback"]
+
+    console.print(f"Evaluation result: {evaluation}")
+    if feedback:
+        console.print(f"Feedback: {feedback}")
+
+    console.print("[bold yellow]Evaluation Feedback:[/bold yellow]")
+    console.print(feedback)
+
+    return evaluation, feedback
+
+def loop_email_workflow(persona: dict, campaign: dict,sender_data: dict ,evaluator_prompt: str, generator_prompt: str, max_tries: int = 5, openai_api_key: str = None, use_huggingface: bool = False) -> dict:
+    """Keep generating and evaluating emails until the evaluator passes or max tries reached."""
+    memory = []  # Store previous responses
+    llm_hits = 0
+    tokens_used = 0
+    cost = 0
+
+    console.print("Starting email generation workflow...")
+    if not persona or not campaign or not sender_data:
+        raise ValueError("Persona, campaign, and sender data are required for email generation.")
+
+    response = generate_email(persona, campaign,sender_data, generator_prompt, openai_api_key=openai_api_key, use_huggingface=use_huggingface)
+    llm_hits += 1
+    tokens_used += len(response.split())  # Approximation of tokens
+    memory.append(response)
+
+    for attempt in range(max_tries):
+        console.print(f"Attempt {attempt + 1} to generate a successful email.")
+        try:
+            email_content = extract_xml(response, "email")
+            console.print(f"Email content: {email_content}")
+            evaluation, feedback = evaluate_email(persona, campaign,sender_data, evaluator_prompt, email_content, openai_api_key=openai_api_key, use_huggingface=use_huggingface)
+        except ValueError as e:
+            console.error(f"Evaluation failed: {e}")
+            break
+
+        llm_hits += 1
+        tokens_used += len(str(feedback).split())
+
+        if evaluation == "PASS":
+            cost = tokens_used * 0.0001  # Example cost calculation
+            console.print("Email generation completed successfully.")
+            return {
+                "final_email": email_content,
+                "llm_hits": llm_hits,
+                "tokens_used": tokens_used,
+                "cost": cost,
+            }
+
+        context = "\n".join([
+            "Previous attempts:",
+            *[f"- {m}" for m in memory],
+            f"Feedback: {feedback}"
+        ])
+        response = generate_email(persona, campaign,sender_data, generator_prompt, context, openai_api_key=openai_api_key, use_huggingface=use_huggingface)
+        llm_hits += 1
+        tokens_used += len(response.split())
+        memory.append(response)
+
+    logger.warning("Max attempts reached without generating a successful email.")
+    cost = tokens_used * 0.0001
+    return {
+        "final_email": None,
+        "llm_hits": llm_hits,
+        "tokens_used": tokens_used,
+        "cost": cost,
+        "message": "Max attempts reached without a PASS.",
+    }
+
+
+# Example user persona
+def example():
+    persona_data = {
+        "name": "Alice Smith",
+        "city": "San Francisco",
+        "hobbies": "Hiking, Cooking",
+        "purchase_history": "Outdoor Gear"
+    }
+
+    # Example campaign details
+    campaign_data = {
+        "subject_line": "Discover Your Next Outdoor Adventure",
+        "product": "New Hiking Backpacks",
+        "discount": "20% off",
+        "validity": "Until January 31st, 2025",
+    }
+
+    # Example sender details
+    sender_data = {
+        "name": "John Doe",
+        "email": "[email protected]"   
+    }
+
+    # Generate and evaluate emails
+    workflow_result = loop_email_workflow(
+        persona=persona_data,
+        campaign=campaign_data,
+        sender_data=sender_data,
+        evaluator_prompt=EMAIL_EVALUATOR_PROMPT,
+        generator_prompt=EMAIL_GENERATOR_PROMPT,
+        max_tries=5,
+        openai_api_key=os.getenv("OPENAI_API_KEY"),
+        use_huggingface=False
+    )
+
+    # Display final result
+    if workflow_result["final_email"]:
+        console.print("Final Email Generated Successfully:")
+        console.print("[bold green]Final Email Content:[/bold green]")
+        console.print(workflow_result["final_email"])
+    else:
+        logger.error("Failed to generate a passing email after maximum attempts.")
+        console.print("[bold red]Workflow Result:[/bold red]")
+        console.print(workflow_result)
+
+
+if __name__ == "__main__":
+    example()

requirements.txt

@@ -1 +1,5 @@
-huggingface_hub==0.25.2
+pydantic
+rich
+openai
+huggingface_hub
+python-dotenv