Adv llm prompting

advanced-prompting-patterns/conversation_manager.py

@@ -0,0 +1,117 @@
+import os
+import tiktoken
+import json
+from openai import OpenAI
+from datetime import datetime
+
+DEFAULT_API_KEY = os.environ.get("TOGETHER_API_KEY")
+DEFAULT_BASE_URL = "https://api.together.xyz/v1"
+DEFAULT_MODEL = "meta-llama/Meta-Llama-3-8B-Instruct-Lite"
+DEFAULT_TEMPERATURE = 0.7
+DEFAULT_MAX_TOKENS = 350
+DEFAULT_TOKEN_BUDGET = 4096
+
+
+class ConversationManager:
+    def __init__(self, api_key=None, base_url=None, model=None, history_file=None, temperature=None, max_tokens=None, token_budget=None):
+        if not api_key:
+            api_key = DEFAULT_API_KEY
+        if not base_url:
+            base_url = DEFAULT_BASE_URL
+
+        self.client = OpenAI(
+            api_key=api_key,
+            base_url=base_url
+        )
+        if history_file is None:
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            self.history_file = f"conversation_history_{timestamp}.json"
+        else:
+            self.history_file = history_file
+
+        self.model = model if model else DEFAULT_MODEL
+        self.temperature = temperature if temperature else DEFAULT_TEMPERATURE
+        self.max_tokens = max_tokens if max_tokens else DEFAULT_MAX_TOKENS
+        self.token_budget = token_budget if token_budget else DEFAULT_TOKEN_BUDGET
+
+        self.system_messages = {
+            "blogger": "You are a creative blogger specializing in engaging and informative content for GlobalJava Roasters.",
+            "social_media_expert": "You are a social media expert, crafting catchy and shareable posts for GlobalJava Roasters.",
+            "creative_assistant": "You are a creative assistant skilled in crafting engaging marketing content for GlobalJava Roasters.",
+            "custom": "Enter your custom system message here."
+        }
+        self.system_message = self.system_messages["creative_assistant"]
+        self.conversation_history = [{"role": "system", "content": self.get_system_message()}]
+
+    def count_tokens(self, text):
+        try:
+            encoding = tiktoken.encoding_for_model(self.model)
+        except KeyError:
+            encoding = tiktoken.get_encoding("cl100k_base")
+
+        tokens = encoding.encode(text)
+        return len(tokens)
+
+    def total_tokens_used(self):
+        return sum(self.count_tokens(message['content']) for message in self.conversation_history)
+
+    def enforce_token_budget(self):
+        while self.total_tokens_used() > self.token_budget:
+            if len(self.conversation_history) <= 1:
+                break
+            self.conversation_history.pop(1)
+
+    def set_persona(self, persona):
+        if persona in self.system_messages:
+            self.system_message = self.system_messages[persona]
+            self.update_system_message_in_history()
+        else:
+            raise ValueError(f"Unknown persona: {persona}. Available personas are: {list(self.system_messages.keys())}")
+
+    def set_custom_system_message(self, custom_message):
+        if not custom_message:
+            raise ValueError("Custom message cannot be empty.")
+        self.system_messages['custom'] = custom_message
+        self.set_persona('custom')
+
+    def get_system_message(self):
+        system_message = self.system_message
+        system_message += f"\nImportant: Tailor your response to fit within {DEFAULT_MAX_TOKENS/2} word limit\n"
+        return system_message        
+
+    def update_system_message_in_history(self):
+        if self.conversation_history and self.conversation_history[0]["role"] == "system":
+            self.conversation_history[0]["content"] = self.get_system_message()
+        else:
+            system_message = self.system_message
+            system_message += f"\nImportant: Tailor your response to fit within {DEFAULT_MAX_TOKENS/2} words limit\n"
+            self.conversation_history.insert(0, {"role": "system", "content": self.get_system_message()})
+
+    def chat_completion(self, prompt, temperature=None, max_tokens=None):
+        temperature = temperature if temperature is not None else self.temperature
+        max_tokens = max_tokens if max_tokens is not None else self.max_tokens
+
+        self.conversation_history.append({"role": "user", "content": prompt})
+
+        self.enforce_token_budget()
+
+        try:
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=self.conversation_history,
+                temperature=temperature,
+                max_tokens=max_tokens,
+            )
+        except Exception as e:
+            print(f"An error occurred while generating a response: {e}")
+            return None
+
+        ai_response = response.choices[0].message.content
+        self.conversation_history.append({"role": "assistant", "content": ai_response})
+
+        return ai_response
+
+    def reset_conversation_history(self):
+        self.conversation_history = [{"role": "system", "content": self.self.get_system_message()}]
+
+

advanced-prompting-patterns/solution-files/campaign_generator.py

@@ -0,0 +1,183 @@
+from conversation_manager import ConversationManager
+from prompt_templates import build_campaign_prompt, build_extract_prompt, build_draft_prompt
+from models import CampaignBrief, ExtractedProductInfo
+from utils import validate_json_output
+
+TEMPERATURE = 0.2
+
+def generate_campaign_brief(factsheet, max_retries=3):
+    """Generate a validated campaign brief with automatic repair."""
+    conversation = ConversationManager()
+
+    initial_prompt = build_campaign_prompt(factsheet)
+    response = conversation.chat_completion(initial_prompt, temperature=TEMPERATURE)
+
+    print("Initial response:")
+    print(response)
+    print()
+
+    success, result = validate_json_output(response, CampaignBrief)
+
+    if success:
+        print("✓ Valid on first attempt!")
+        return result
+
+    print(f"✗ Validation failed: {result}")
+    print()
+
+    retries = 0
+    last_error = result
+
+    while retries < max_retries:
+        print(f"Attempting repair {retries + 1}/{max_retries}...")
+
+        repair_prompt = f"""
+The JSON you provided had validation errors:
+
+{last_error}
+
+Please provide corrected JSON that fixes these errors. Remember:
+- campaign_goal must be exactly "awareness", "engagement", or "conversion"
+- All required fields must be present: campaign_name, target_audience, key_message, campaign_goal, call_to_action, channel_recommendations
+- channel_recommendations must be a list of strings
+
+Respond with valid JSON only. Keep each string value to 1–2 sentences max.
+"""
+
+        response = conversation.chat_completion(repair_prompt, temperature=TEMPERATURE)
+
+        print("Repair response:")
+        print(response)
+        print()
+
+        success, result = validate_json_output(response, CampaignBrief)
+
+        if success:
+            print("✓ Repair successful!")
+            return result
+
+        last_error = result
+        print(f"✗ Still invalid: {last_error}")
+        print()
+        retries += 1
+
+    raise ValueError(
+        f"Could not generate valid campaign brief after {max_retries} attempts. Last error: {last_error}"
+    )
+
+
+def extract_product_info(factsheet, max_retries=3):
+    """Extract structured information from product factsheet."""
+    conversation = ConversationManager()
+
+    prompt = build_extract_prompt(factsheet)
+    response = conversation.chat_completion(prompt, temperature=TEMPERATURE)
+
+    success, result = validate_json_output(response, ExtractedProductInfo)
+    if success:
+        return result
+
+    retries = 0
+    last_error = result
+
+    while retries < max_retries:
+        repair_prompt = f"""
+The JSON had errors:
+
+{last_error}
+
+Provide corrected JSON matching the required structure.
+Respond with JSON only. Keep each string value to 1–2 sentences max.
+Use [] for scarcity_factors if none.
+"""
+        response = conversation.chat_completion(repair_prompt, temperature=TEMPERATURE)
+
+        success, result = validate_json_output(response, ExtractedProductInfo)
+        if success:
+            return result
+
+        last_error = result
+        retries += 1
+
+    raise ValueError(f"Could not extract product info after {max_retries} attempts. Last error: {last_error}")
+
+
+def generate_campaign_brief_pipeline(factsheet, max_retries=3):
+    """Generate campaign brief using multi-step pipeline."""
+    print("Step 1: Extracting product information...")
+    extracted = extract_product_info(factsheet, max_retries)
+    print(f"✓ Extracted info for: {extracted.product_name}")
+
+    print("Step 2: Drafting campaign brief...")
+    conversation = ConversationManager()
+
+    prompt = build_draft_prompt(extracted)
+    response = conversation.chat_completion(prompt, temperature=TEMPERATURE)
+
+    print("Initial campaign brief response:")
+    print(response)
+    print()
+
+    success, result = validate_json_output(response, CampaignBrief)
+    if success:
+        print("✓ Valid campaign brief generated")
+        return result
+
+    print("Step 3: Repairing output...")
+    retries = 0
+    last_error = result
+
+    while retries < max_retries:
+        repair_prompt = f"""
+The JSON had validation errors:
+
+{last_error}
+
+Provide corrected JSON. Remember:
+- campaign_goal must be "awareness", "engagement", or "conversion"
+- All required fields must be present
+- channel_recommendations must be a list of strings
+
+Respond with JSON only. Keep each string value to 1–2 sentences max.
+"""
+        response = conversation.chat_completion(repair_prompt, temperature=TEMPERATURE)
+
+        success, result = validate_json_output(response, CampaignBrief)
+        if success:
+            print("✓ Repair successful")
+            return result
+
+        last_error = result
+        retries += 1
+
+    raise ValueError(f"Could not generate valid brief after {max_retries} repair attempts. Last error: {last_error}")
+
+
+if __name__ == "__main__":
+    factsheet = """
+Product: Limited Edition Geisha Reserve
+Origin: Hacienda La Esmeralda, Panama
+Altitude: 1,600-1,800 meters
+Processing: Natural, 72-hour fermentation
+Flavor Profile: Jasmine, bergamot, white peach, honey sweetness,
+silky body, complex finish with hints of tropical fruit
+Certifications: Single Estate, Competition Grade
+Limited Production: Only 500 bags produced this season
+Story: This micro-lot scored 94.1 points in the 2024 Cup of Excellence
+competition. The beans come from 30-year-old Geisha trees grown in
+volcanic soil. The extended fermentation process was developed
+specifically for this lot to enhance the floral characteristics.
+Price: $89.99/bag
+Previous Customer Feedback: "Best coffee I've ever tasted" - Coffee
+Review Magazine. Sold out in 3 days last year.
+""".strip()
+
+    try:
+        brief = generate_campaign_brief_pipeline(factsheet)
+        print("✓ Generated valid campaign brief!")
+        print(f"Campaign: {brief.campaign_name}")
+        print(f"Target: {brief.target_audience}")
+        print(f"Goal: {brief.campaign_goal.value}")
+        print(f"Channels: {', '.join(brief.channel_recommendations)}")
+    except ValueError as e:
+        print(f"✗ Failed to generate brief: {e}")