Prompt Engineering Mastery: Advanced Ollama Query Optimization Techniques

# Unoptimized prompt (avg: 8.2s response time)
ollama run llama2 "Tell me about machine learning"

# Optimized prompt (avg: 3.1s response time)  
ollama run llama2 "Explain supervised learning algorithms in 3 bullet points with real-world examples"

# Check current model memory usage
import ollama

response = ollama.show('llama2')
print(f"Model size: {response['details']['parameter_size']}")
print(f"Quantization: {response['details']['quantization_level']}")

# Vague prompt - poor performance
prompt = "Help me with my Python code"

response = ollama.generate(
    model='codellama',
    prompt=prompt
)

# Optimized prompt using CRISP framework
prompt = """
Context: I'm debugging a Flask web application with authentication issues.

Role: Act as a senior Python developer with Flask expertise.

Instruction: Review this login function and identify security vulnerabilities.

Specification: Provide:
- Numbered list of vulnerabilities found
- Severity level (high/medium/low) for each
- Specific code fixes with explanations

Parameters: Focus only on authentication security, limit response to 200 words.

Code to review:
```[python](/chat-with-database-architecture/)
@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    if username == 'admin' and password == 'password':
        session['user'] = username
        return redirect('/dashboard')
    return 'Login failed'

# Creative tasks - higher temperature
creative_params = {
    'temperature': 0.8,
    'top_p': 0.9,
    'top_k': 50
}

# Analytical tasks - lower temperature  
analytical_params = {
    'temperature': 0.2,
    'top_p': 0.5,
    'top_k': 20
}

# Code generation - minimal randomness
code_params = {
    'temperature': 0.1,
    'top_p': 0.3,
    'top_k': 10
}

def optimize_for_task(task_type, prompt):
    """Select optimal parameters based on task requirements"""
    
    param_map = {
        'creative': creative_params,
        'analytical': analytical_params,
        'code': code_params,
        'factual': {'temperature': 0.0, 'top_p': 1.0, 'top_k': 1}
    }
    
    params = param_map.get(task_type, analytical_params)
    
    return ollama.generate(
        model='llama2',
        prompt=prompt,
        options=params
    )

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def process_prompts_batch(prompts, model='llama2', max_workers=3):
    """
    Process multiple prompts concurrently with controlled parallelism
    """
    
    def single_request(prompt):
        return ollama.generate(
            model=model,
            prompt=prompt,
            options={'temperature': 0.3}
        )
    
    # Limit concurrent requests to prevent memory overflow
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        loop = asyncio.get_event_loop()
        tasks = [
            loop.run_in_executor(executor, single_request, prompt)
            for prompt in prompts
        ]
        
        results = await asyncio.gather(*tasks)
        return results

# Usage example
prompts = [
    "Summarize the benefits of renewable energy",
    "Explain quantum computing in simple terms", 
    "List 5 Python best practices for beginners"
]

# Run batch processing
results = asyncio.run(process_prompts_batch(prompts))

import time
import random

class OllamaOptimizer:
    def __init__(self, model='llama2', max_retries=3):
        self.model = model
        self.max_retries = max_retries
    
    def generate_with_fallback(self, prompt, **kwargs):
        """
        Generate response with automatic optimization and fallback
        """
        
        strategies = [
            self._standard_generation,
            self._compressed_prompt_generation, 
            self._simplified_generation
        ]
        
        for attempt, strategy in enumerate(strategies):
            try:
                result = strategy(prompt, **kwargs)
                if self._validate_response(result):
                    return result
                    
            except Exception as e:
                if attempt == len(strategies) - 1:
                    raise e
                
                # Exponential backoff
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                time.sleep(wait_time)
                
        return None
    
    def _standard_generation(self, prompt, **kwargs):
        """Standard generation approach"""
        return ollama.generate(
            model=self.model,
            prompt=prompt,
            **kwargs
        )
    
    def _compressed_prompt_generation(self, prompt, **kwargs):
        """Fallback with compressed prompt"""
        compressed = self._compress_prompt(prompt)
        return ollama.generate(
            model=self.model,
            prompt=compressed,
            **kwargs
        )
    
    def _simplified_generation(self, prompt, **kwargs):
        """Last resort with minimal parameters"""
        simplified = self._simplify_prompt(prompt)
        return ollama.generate(
            model=self.model,
            prompt=simplified,
            options={'temperature': 0.1}
        )
    
    def _validate_response(self, response):
        """Check if response meets quality standards"""
        if not response or not response.get('response'):
            return False
            
        content = response['response'].strip()
        
        # Basic quality checks
        if len(content) < 10:
            return False
        if content.lower().startswith('i don\'t'):
            return False
        if 'error' in content.lower():
            return False
            
        return True
    
    def _compress_prompt(self, prompt):
        """Compress prompt while preserving key information"""
        # Extract core instruction
        lines = prompt.split('\n')
        essential_lines = [line for line in lines if line.strip() 
                          and not line.startswith('#')]
        return ' '.join(essential_lines)
    
    def _simplify_prompt(self, prompt):
        """Create simplified version of prompt"""
        # Extract main question or instruction
        sentences = prompt.split('.')
        return sentences[0] + '.' if sentences else prompt[:100]

# Usage example
optimizer = OllamaOptimizer(model='llama2')

response = optimizer.generate_with_fallback(
    "Explain machine learning algorithms with examples and use cases",
    options={'temperature': 0.5}
)

import time
from dataclasses import dataclass
from typing import List
import json

@dataclass
class PromptMetrics:
    prompt_length: int
    response_length: int
    response_time: float
    token_count: int
    success: bool
    model_used: str
    timestamp: float

class PromptAnalyzer:
    def __init__(self):
        self.metrics: List[PromptMetrics] = []
    
    def track_request(self, prompt, model='llama2', **kwargs):
        """Track individual request performance"""
        
        start_time = time.time()
        
        try:
            response = ollama.generate(
                model=model,
                prompt=prompt,
                **kwargs
            )
            
            end_time = time.time()
            
            metrics = PromptMetrics(
                prompt_length=len(prompt),
                response_length=len(response.get('response', '')),
                response_time=end_time - start_time,
                token_count=response.get('eval_count', 0),
                success=True,
                model_used=model,
                timestamp=end_time
            )
            
            self.metrics.append(metrics)
            return response
            
        except Exception as e:
            end_time = time.time()
            
            metrics = PromptMetrics(
                prompt_length=len(prompt),
                response_length=0,
                response_time=end_time - start_time,
                token_count=0,
                success=False,
                model_used=model,
                timestamp=end_time
            )
            
            self.metrics.append(metrics)
            raise e
    
    def generate_report(self):
        """Generate performance analysis report"""
        
        if not self.metrics:
            return "No metrics collected yet."
        
        successful_requests = [m for m in self.metrics if m.success]
        
        if not successful_requests:
            return "No successful requests to analyze."
        
        avg_response_time = sum(m.response_time for m in successful_requests) / len(successful_requests)
        avg_prompt_length = sum(m.prompt_length for m in successful_requests) / len(successful_requests)
        avg_response_length = sum(m.response_length for m in successful_requests) / len(successful_requests)
        success_rate = len(successful_requests) / len(self.metrics) * 100
        
        # Find performance patterns
        fast_requests = [m for m in successful_requests if m.response_time < avg_response_time]
        slow_requests = [m for m in successful_requests if m.response_time > avg_response_time * 1.5]
        
        report = f"""
        Performance Analysis Report
        ==========================
        
        Overall Metrics:
        - Total requests: {len(self.metrics)}
        - Success rate: {success_rate:.1f}%
        - Average response time: {avg_response_time:.2f}s
        - Average prompt length: {avg_prompt_length:.0f} chars
        - Average response length: {avg_response_length:.0f} chars
        
        Performance Insights:
        - Fast requests (< avg): {len(fast_requests)}
        - Slow requests (> 1.5x avg): {len(slow_requests)}
        
        Optimization Recommendations:
        """
        
        # Add specific recommendations
        if avg_prompt_length > 1000:
            report += "\n- Consider shortening prompts (current avg: {:.0f} chars)".format(avg_prompt_length)
        
        if success_rate < 90:
            report += "\n- Improve error handling (current success rate: {:.1f}%)".format(success_rate)
        
        if len(slow_requests) > len(fast_requests):
            report += "\n- Review prompt complexity and model parameters"
        
        return report
    
    def export_metrics(self, filename='ollama_metrics.json'):
        """Export metrics to JSON file"""
        
        data = [{
            'prompt_length': m.prompt_length,
            'response_length': m.response_length,
            'response_time': m.response_time,
            'token_count': m.token_count,
            'success': m.success,
            'model_used': m.model_used,
            'timestamp': m.timestamp
        } for m in self.metrics]
        
        with open(filename, 'w') as f:
            json.dump(data, f, indent=2)

# Usage example
analyzer = PromptAnalyzer()

# Track multiple requests
prompts = [
    "Explain photosynthesis",
    "Write a Python function to sort a list",
    "Describe the water cycle"
]

for prompt in prompts:
    try:
        response = analyzer.track_request(prompt)
        print(f"Response generated for: {prompt[:50]}...")
    except Exception as e:
        print(f"Error processing: {prompt[:50]}... - {e}")

# Generate performance report
print(analyzer.generate_report())

# Export metrics for further analysis
analyzer.export_metrics()

class CodeReviewTemplate:
    """Template for code review prompts"""
    
    @staticmethod
    def generate(code, language, focus_areas=None):
        focus_areas = focus_areas or ['security', 'performance', 'maintainability']
        
        template = f"""
        Context: Code review for {language} application
        
        Role: Senior {language} developer with 10+ years experience
        
        Instruction: Review the following code and provide detailed feedback
        
        Specification: 
        - Focus areas: {', '.join(focus_areas)}
        - Provide specific line-by-line comments
        - Rate overall code quality (1-10)
        - Suggest concrete improvements
        
        Parameters: Maximum 300 words, prioritize critical issues
        
        Code to review:
        ```{language}
        {code}
        ```
        """
        
        return template.strip()

# Usage
code_sample = """
def calculate_price(base_price, discount):
    final_price = base_price - (base_price * discount / 100)
    return final_price
"""

prompt = CodeReviewTemplate.generate(
    code=code_sample,
    language='python',
    focus_areas=['security', 'error_handling']
)

response = ollama.generate(model='codellama', prompt=prompt)

class DocumentationTemplate:
    """Template for generating technical documentation"""
    
    @staticmethod
    def api_documentation(function_name, parameters, description):
        template = f"""
        Context: API documentation for development team
        
        Role: Technical writer creating developer documentation
        
        Instruction: Generate comprehensive API documentation
        
        Specification:
        - Include function signature
        - Parameter descriptions with types
        - Return value details
        - Usage examples
        - Error conditions
        
        Parameters: Use clear, concise language suitable for developers
        
        Function details:
        - Name: {function_name}
        - Parameters: {parameters}
        - Description: {description}
        """
        
        return template.strip()

# Usage example
prompt = DocumentationTemplate.api_documentation(
    function_name="authenticate_user",
    parameters="username (str), password (str), remember_me (bool)",
    description="Authenticates user credentials and creates session"
)

# Before - vague format instruction
prompt = "List the pros and cons of electric vehicles"

# After - explicit format with example
prompt = """
List the pros and cons of electric vehicles using this exact format:

PROS:
• [Benefit 1]: [Brief explanation]
• [Benefit 2]: [Brief explanation]

CONS:
• [Drawback 1]: [Brief explanation]
• [Drawback 2]: [Brief explanation]

Example format:
PROS:
• Lower emissions: Reduce air pollution in urban areas
• Cost savings: Lower fuel and maintenance costs

CONS:
• Limited range: Average 200-300 miles per charge
• Charging time: Takes longer than gas refueling
"""

def maintain_context(conversation_history, max_context_length=2000):
    """
    Maintain important context while staying within limits
    """
    
    if len(str(conversation_history)) <= max_context_length:
        return conversation_history
    
    # Extract key facts from conversation
    key_facts = extract_key_information(conversation_history)
    
    # Keep recent messages and summarized context
    recent_messages = conversation_history[-3:]  # Last 3 exchanges
    
    context_summary = f"Previous context: {'; '.join(key_facts)}"
    
    optimized_history = [{
        'role': 'system',
        'content': context_summary
    }] + recent_messages
    
    return optimized_history

def extract_key_information(messages):
    """Extract important facts and decisions from conversation"""
    key_facts = []
    
    for message in messages:
        if message['role'] == 'assistant':
            # Look for definitive statements, decisions, or important facts
            content = message['content']
            sentences = content.split('.')
            
            for sentence in sentences:
                if any(keyword in sentence.lower() for keyword in 
                      ['decided', 'confirmed', 'important', 'key', 'must']):
                    key_facts.append(sentence.strip())
    
    return key_facts[:5]  # Keep top 5 most important facts

import hashlib
import pickle
from functools import lru_cache

class ResponseCache:
    """Cache responses for identical prompts"""
    
    def __init__(self, cache_size=100):
        self.cache = {}
        self.cache_size = cache_size
        self.access_order = []
    
    def get_cache_key(self, prompt, model, options):
        """Generate unique cache key for prompt combination"""
        content = f"{prompt}:{model}:{str(sorted(options.items()))}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, prompt, model='llama2', options=None):
        """Retrieve cached response if available"""
        options = options or {}
        key = self.get_cache_key(prompt, model, options)
        
        if key in self.cache:
            # Update access order
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]
        
        return None
    
    def set(self, prompt, model, options, response):
        """Cache response with LRU eviction"""
        options = options or {}
        key = self.get_cache_key(prompt, model, options)
        
        # Evict oldest if cache is full
        if len(self.cache) >= self.cache_size:
            oldest_key = self.access_order.pop(0)
            del self.cache[oldest_key]
        
        self.cache[key] = response
        self.access_order.append(key)

# Global cache instance
response_cache = ResponseCache(cache_size=50)

def optimized_generate(prompt, model='llama2', options=None, use_cache=True):
    """Generate response with caching and optimization"""
    
    if use_cache:
        cached_response = response_cache.get(prompt, model, options)
        if cached_response:
            return cached_response
    
    # Optimize prompt before generation
    optimized_prompt = optimize_prompt_structure(prompt)
    
    response = ollama.generate(
        model=model,
        prompt=optimized_prompt,
        options=options or {}
    )
    
    if use_cache:
        response_cache.set(prompt, model, options, response)
    
    return response

@lru_cache(maxsize=20)
def optimize_prompt_structure(prompt):
    """Apply structural optimizations to prompt"""
    
    # Remove unnecessary whitespace
    lines = [line.strip() for line in prompt.split('\n') if line.strip()]
    
    # Ensure clear instruction structure
    if not any(keyword in prompt.lower() for keyword in ['instruction:', 'task:', 'please']):
        lines.insert(0, "Instruction:")
    
    return '\n'.join(lines)