The Complete 2025 Prompt Engineering Guide: From Prompts to Context

def self_consistency_prompt(problem, num_paths=5):
    base_prompt = f"""
    Problem: {problem}
    
    Let me solve this step-by-step:
    [Insert your reasoning template]
    """
    
    responses = []
    for i in range(num_paths):
        # Add slight variation to encourage different reasoning paths
        variation_prompt = base_prompt + f"\n\nApproach {i+1}: Let me try a different angle..."
        response = llm_call(variation_prompt)
        responses.append(extract_answer(response))
    
    # Return most common answer
    return most_frequent(responses)

class UserContextManager:
    def __init__(self):
        self.session_context = {}
        self.user_profile = {}
        self.conversation_history = []
        self.task_context = {}
    
    def build_context_prompt(self, current_input):
        context_sections = []
        
        # User profile (persistent)
        if self.user_profile:
            context_sections.append(f"User Profile: {self.user_profile}")
        
        # Relevant conversation history (last 3 meaningful exchanges)
        relevant_history = self.get_relevant_history(current_input, limit=3)
        if relevant_history:
            context_sections.append(f"Conversation Context: {relevant_history}")
        
        # Current task context
        if self.task_context:
            context_sections.append(f"Current Task: {self.task_context}")
        
        # Session context (temporary but important)
        if self.session_context:
            context_sections.append(f"Session Info: {self.session_context}")
        
        return "\n\n".join(context_sections)
    
    def update_context(self, user_input, ai_response):
        # Update conversation history with relevance scoring
        self.conversation_history.append({
            'user': user_input,
            'ai': ai_response,
            'timestamp': datetime.now(),
            'relevance_score': self.calculate_relevance(user_input)
        })
        
        # Extract task progression
        self.update_task_context(user_input, ai_response)
        
        # Update user profile with new information
        self.update_user_profile(user_input)

class ConversationMemory:
    def __init__(self, max_context_tokens=8000):
        self.max_tokens = max_context_tokens
        self.conversation_log = []
        self.key_facts = {}
        self.decision_points = []
    
    def add_exchange(self, user_input, ai_response):
        exchange = {
            'user': user_input,
            'ai': ai_response,
            'timestamp': datetime.now(),
            'token_count': estimate_tokens(user_input + ai_response)
        }
        
        self.conversation_log.append(exchange)
        self.extract_key_information(exchange)
        
        # Compress if approaching token limit
        if self.estimate_total_tokens() > self.max_tokens * 0.8:
            self.compress_history()
    
    def extract_key_information(self, exchange):
        # Extract persistent facts
        facts_prompt = f"""
        From this conversation exchange, extract any factual information 
        that should be remembered for future interactions:
        
        User: {exchange['user']}
        AI: {exchange['ai']}
        
        Return only facts that would be relevant later:
        """
        
        extracted_facts = llm_call(facts_prompt)
        if extracted_facts.strip():
            self.key_facts[exchange['timestamp']] = extracted_facts
    
    def compress_history(self):
        # Keep recent exchanges, compress older ones
        recent_exchanges = self.conversation_log[-5:]  # Last 5 exchanges
        older_exchanges = self.conversation_log[:-5]
        
        if older_exchanges:
            summary_prompt = f"""
            Summarize this conversation history, preserving key decisions 
            and important context:
            
            {self.format_exchanges(older_exchanges)}
            
            Focus on:
            - Decisions made
            - Problems solved  
            - User preferences revealed
            - Important facts established
            """
            
            summary = llm_call(summary_prompt)
            
            # Replace older exchanges with summary
            self.conversation_log = [{
                'type': 'summary',
                'content': summary,
                'represents': len(older_exchanges)
            }] + recent_exchanges

class ContextAwareRAG:
    def __init__(self, vector_store, reranker=None):
        self.vector_store = vector_store
        self.reranker = reranker
    
    def retrieve_with_context(self, query, conversation_context, k=5):
        # Enhance query with conversation context
        enhanced_query = self.enhance_query(query, conversation_context)
        
        # Initial retrieval
        candidates = self.vector_store.similarity_search(enhanced_query, k=k*2)
        
        # Context-aware reranking
        if self.reranker:
            reranked = self.reranker.rerank(
                query=enhanced_query,
                documents=candidates,
                context=conversation_context
            )
            return reranked[:k]
        
        return candidates[:k]
    
    def enhance_query(self, query, context):
        enhancement_prompt = f"""
        Original query: {query}
        Conversation context: {context}
        
        Create an enhanced search query that incorporates relevant context:
        """
        return llm_call(enhancement_prompt)
    
    def format_retrieved_context(self, documents, query):
        formatted_docs = []
        for i, doc in enumerate(documents):
            formatted_docs.append(f"""
            Source {i+1}: {doc.metadata.get('title', 'Unknown')}
            Relevance: {doc.metadata.get('score', 'N/A')}
            Content: {doc.page_content}
            """)
        
        return f"""
        Retrieved Information for: "{query}"
        
        {chr(10).join(formatted_docs)}
        
        Instructions: Use this information to inform your response, but clearly 
        distinguish between retrieved facts and your reasoning.
        """

class ContextAwareToolManager:
    def __init__(self):
        self.available_tools = {}
        self.tool_usage_history = []
    
    def register_tool(self, name, tool_function, context_requirements=None):
        self.available_tools[name] = {
            'function': tool_function,
            'context_requirements': context_requirements or [],
            'usage_count': 0
        }
    
    def execute_tool_with_context(self, tool_name, parameters, context):
        if tool_name not in self.available_tools:
            return {"error": f"Tool {tool_name} not available"}
        
        tool = self.available_tools[tool_name]
        
        # Check context requirements
        missing_context = []
        for requirement in tool['context_requirements']:
            if requirement not in context:
                missing_context.append(requirement)
        
        if missing_context:
            return {
                "error": f"Missing required context: {missing_context}",
                "action": "gather_context",
                "required": missing_context
            }
        
        # Execute tool with context
        try:
            result = tool['function'](parameters, context)
            
            # Log usage for learning
            self.tool_usage_history.append({
                'tool': tool_name,
                'parameters': parameters,
                'context_used': context,
                'result': result,
                'timestamp': datetime.now()
            })
            
            tool['usage_count'] += 1
            return result
            
        except Exception as e:
            return {"error": f"Tool execution failed: {str(e)}"}
    
    def suggest_next_tools(self, current_context, goal):
        suggestion_prompt = f"""
        Current context: {current_context}
        Goal: {goal}
        Available tools: {list(self.available_tools.keys())}
        
        What tools would be most helpful for achieving this goal?
        Consider the current context and tool capabilities.
        """
        
        suggestions = llm_call(suggestion_prompt)
        return suggestions

class ContextEngineeringSystem:
    def __init__(self, llm, config):
        self.llm = llm
        self.config = config
        
        # Initialize all components
        self.user_context = UserContextManager()
        self.memory = ConversationMemory(config.max_context_tokens)
        self.rag = ContextAwareRAG(config.vector_store)
        self.tools = ContextAwareToolManager()
        self.formatter = ContextAwareFormatter()
        self.evaluator = ContextEvaluator()
        
        # Register tools and schemas
        self.setup_tools()
        self.setup_output_schemas()
        self.setup_evaluation_metrics()
    
    def process_request(self, user_input, session_id=None):
        """Main processing pipeline with full context engineering"""
        
        # 1. Initialize session context
        session_context = self.initialize_session(session_id)
        
        # 2. Build comprehensive context
        context = self.build_comprehensive_context(user_input, session_context)
        
        # 3. Determine required tools and information
        action_plan = self.plan_actions(user_input, context)
        
        # 4. Execute action plan
        execution_results = self.execute_plan(action_plan, context)
        
        # 5. Generate response with full context
        response = self.generate_contextualized_response(
            user_input, context, execution_results
        )
        
        # 6. Format output appropriately
        formatted_response = self.format_response(response, context)
        
        # 7. Evaluate and learn
        evaluation = self.evaluate_response(user_input, formatted_response, context)
        
        # 8. Update context and memory
        self.update_system_state(user_input, formatted_response, context, evaluation)
        
        return {
            'response': formatted_response,
            'context_used': context,
            'evaluation': evaluation,
            'session_id': session_id
        }
    
    def build_comprehensive_context(self, user_input, session_context):
        """Assembles all relevant context for the current request"""
        
        context = {
            'timestamp': datetime.now().isoformat(),
            'session': session_context,
            'user_input': user_input
        }
        
        # Add user context
        user_ctx = self.user_context.build_context_prompt(user_input)
        if user_ctx:
            context['user_context'] = user_ctx
        
        # Add conversation memory
        memory_ctx = self.memory.get_relevant_context(user_input)
        if memory_ctx:
            context['conversation_memory'] = memory_ctx
        
        # Add retrieved information if needed
        if self.should_retrieve(user_input, context):
            retrieved = self.rag.retrieve_with_context(user_input, context)
            context['retrieved_info'] = self.rag.format_retrieved_context(retrieved, user_input)
        
        return context
    
    def generate_contextualized_response(self, user_input, context, execution_results):
        """Generates response using full context"""
        
        # Build comprehensive prompt with all context
        response_prompt = f"""
        {self.config.system_instructions}
        
        Current Context:
        {self.format_context_for_prompt(context)}
        
        Execution Results:
        {self.format_execution_results(execution_results)}
        
        User Request: {user_input}
        
        Generate a comprehensive response that:
        1. Directly addresses the user's request
        2. Incorporates relevant context naturally
        3. References execution results where appropriate
        4. Maintains consistency with conversation history
        5. Follows the specified output format
        """
        
        response = self.llm.generate(response_prompt)
        return response

# Configuration for customer support system
config = ContextConfig(
    max_context_tokens=16000,
    system_instructions="""
    You are a customer service specialist for TechCorp with access to:
    - Customer profiles and billing history
    - Order tracking and inventory systems  
    - Policy and procedure documentation
    - Escalation workflows for complex issues
    
    Always:
    - Acknowledge the customer's concern
    - Use their name and reference their history
    - Provide specific, actionable solutions
    - Include next steps and timelines
    """,
    vector_store=company_knowledge_base
)

# Initialize the system
cs_system = ContextEngineeringSystem(llm=claude, config=config)

# Register customer service tools
cs_system.tools.register_tool(
    'lookup_customer',
    customer_database.lookup,
    context_requirements=['customer_id', 'account_number']
)

cs_system.tools.register_tool(
    'check_billing',
    billing_system.get_billing_history,
    context_requirements=['customer_id']
)

cs_system.tools.register_tool(
    'create_ticket',
    ticketing_system.create_support_ticket,
    context_requirements=['customer_id', 'issue_category']
)

# Example conversation flow
conversation = [
    "Hi, I'm seeing a charge on my account that I don't recognize",
    "My account number is AC-789456",
    "The charge is for $99.99 on July 15th",
    "I want to dispute this charge"
]

session_context = {'channel': 'web_chat', 'customer_tier': 'premium'}

for user_input in conversation:
    result = cs_system.process_request(user_input, session_id='cs_session_123')
    
    print(f"User: {user_input}")
    print(f"Assistant: {result['response']}")
    print(f"Context Used: {list(result['context_used'].keys())}")
    print(f"Evaluation: {result['evaluation']}")
    print("---")

class MegaContextManager:
    def __init__(self, max_tokens=200000):
        self.max_tokens = max_tokens
        self.context_sections = {}
        self.priority_weights = {}
    
    def add_context_section(self, name, content, priority=1.0, decay_rate=0.1):
        """Add a section of context with priority and decay"""
        token_count = self.estimate_tokens(content)
        
        self.context_sections[name] = {
            'content': content,
            'tokens': token_count,
            'priority': priority,
            'decay_rate': decay_rate,
            'last_accessed': datetime.now(),
            'access_count': 0
        }
    
    def build_optimized_context(self, current_query):
        """Build context optimized for current query"""
        
        # Calculate relevance scores for each section
        relevance_scores = {}
        for name, section in self.context_sections.items():
            relevance = self.calculate_relevance(section['content'], current_query)
            priority = section['priority']
            time_decay = self.calculate_time_decay(section['last_accessed'], section['decay_rate'])
            
            relevance_scores[name] = relevance * priority * time_decay
        
        # Sort by relevance and fit within token budget
        sorted_sections = sorted(
            relevance_scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )
        
        selected_context = {}
        total_tokens = 0
        
        for name, score in sorted_sections:
            section = self.context_sections[name]
            if total_tokens + section['tokens'] <= self.max_tokens * 0.8:  # Leave room for response
                selected_context[name] = section['content']
                total_tokens += section['tokens']
                
                # Update access tracking
                section['last_accessed'] = datetime.now()
                section['access_count'] += 1
        
        return selected_context, total_tokens

class AdaptiveContextSystem:
    def __init__(self):
        self.user_profiles = {}
        self.context_strategies = {}
        self.performance_tracker = {}
    
    def adapt_context_for_user(self, user_id, task_type, base_context):
        """Dynamically adapt context based on user and task"""
        
        user_profile = self.get_user_profile(user_id)
        
        # Determine which adaptation strategies apply
        applicable_strategies = []
        for name, strategy in self.context_strategies.items():
            if self.evaluate_conditions(strategy['conditions'], user_profile, task_type):
                applicable_strategies.append((name, strategy))
        
        # Apply strategies in order of success rate
        adapted_context = base_context.copy()
        applied_strategies = []
        
        for name, strategy in sorted(applicable_strategies, key=lambda x: x[1]['success_rate'], reverse=True):
            try:
                adapted_context = strategy['function'](adapted_context, user_profile, task_type)
                applied_strategies.append(name)
                strategy['usage_count'] += 1
            except Exception as e:
                print(f"Strategy {name} failed: {e}")
        
        return adapted_context, applied_strategies

class MultimodalContextManager:
    def __init__(self):
        self.modality_processors = {}
        self.cross_modal_relationships = {}
        self.integration_strategies = {}
    
    def process_multimodal_input(self, inputs):
        """Process inputs across multiple modalities"""
        processed_modalities = {}
        
        for modality, data in inputs.items():
            if modality in self.modality_processors:
                processor = self.modality_processors[modality]
                processed_modalities[modality] = processor.process(data)
        
        # Find cross-modal relationships
        relationships = self.identify_cross_modal_relationships(processed_modalities)
        
        # Integrate modalities into unified context
        integrated_context = self.integrate_modalities(processed_modalities, relationships)
        
        return integrated_context
    
    def integrate_modalities(self, processed_modalities, relationships):
        """Create unified context from multiple modalities"""
        
        integration_prompt = f"""
        Multimodal Input Integration:
        
        Text Content: {processed_modalities.get('text', {}).get('content', 'None')}
        
        Image Analysis: {processed_modalities.get('image', {}).get('description', 'None')}
        
        Audio Transcript: {processed_modalities.get('audio', {}).get('transcript', 'None')}
        
        Cross-Modal Relationships: {relationships}
        
        Create a unified context description that:
        1. Synthesizes information across all modalities
        2. Highlights important cross-modal relationships
        3. Identifies any conflicts or inconsistencies
        4. Provides a coherent narrative of the multimodal input
        """
        
        integrated_description = self.llm.generate(integration_prompt)
        
        return {
            'unified_description': integrated_description,
            'individual_modalities': processed_modalities,
            'relationships': relationships,
            'integration_confidence': self.calculate_integration_confidence(processed_modalities, relationships)
        }

class ContextWindowManager:
    def __init__(self, max_tokens, buffer_ratio=0.1):
        self.max_tokens = max_tokens
        self.buffer_tokens = int(max_tokens * buffer_ratio)
        self.effective_limit = max_tokens - self.buffer_tokens
    
    def optimize_context_for_window(self, context_sections):
        """Optimize context to fit within token window"""
        
        # Calculate current token usage
        current_tokens = sum(self.estimate_tokens(section) for section in context_sections.values())
        
        if current_tokens <= self.effective_limit:
            return context_sections  # No optimization needed
        
        # Prioritize context sections
        prioritized = self.prioritize_context_sections(context_sections)
        
        # Apply optimization strategies
        optimized_context = {}
        running_total = 0
        
        for section_name, section_data in prioritized:
            section_tokens = self.estimate_tokens(section_data['content'])
            
            if running_total + section_tokens <= self.effective_limit:
                # Include full section
                optimized_context[section_name] = section_data['content']
                running_total += section_tokens
            else:
                # Apply compression strategies
                remaining_budget = self.effective_limit - running_total
                if remaining_budget > 100:  # Minimum viable content
                    compressed = self.compress_section(section_data['content'], remaining_budget)
                    if compressed:
                        optimized_context[f"{section_name}_compressed"] = compressed
                        running_total += self.estimate_tokens(compressed)
                break
        
        return optimized_context

class ProductionContextSystem:
    def __init__(self, config):
        self.config = config
        
        # Core components
        self.context_engine = ContextEngineeringSystem(config.llm, config)
        
        # Production components
        self.cache_layer = ContextCacheLayer(config.cache_config)
        self.metrics_collector = MetricsCollector(config.metrics_config)
        self.rate_limiter = RateLimiter(config.rate_limits)
        self.cost_optimizer = CostOptimizer(config.cost_config)
        
        # Monitoring and alerting
        self.health_monitor = HealthMonitor(config.health_config)
        self.alert_manager = AlertManager(config.alert_config)
    
    async def process_request_production(self, request):
        """Production request processing with full monitoring and optimization"""
        
        request_id = self.generate_request_id()
        start_time = time.time()
        
        try:
            # Rate limiting
            await self.rate_limiter.check_limit(request.user_id)
            
            # Check cache first
            cache_key = self.generate_cache_key(request)
            cached_result = await self.cache_layer.get(cache_key)
            
            if cached_result and self.is_cache_valid(cached_result, request):
                self.metrics_collector.record_cache_hit(request_id)
                return self.format_cached_response(cached_result, request_id)
            
            # Cost pre-check
            estimated_cost = self.cost_optimizer.estimate_request_cost(request)
            if estimated_cost > self.config.max_request_cost:
                raise CostLimitExceededError(f"Estimated cost {estimated_cost} exceeds limit")
            
            # Process with context engineering
            result = await self.context_engine.process_request(
                request.input,
                session_id=request.session_id
            )
            
            # Post-process for production
            production_result = self.post_process_result(result, request)
            
            # Cache result if appropriate
            if self.should_cache_result(production_result, request):
                await self.cache_layer.set(cache_key, production_result, ttl=self.config.cache_ttl)
            
            # Record metrics
            processing_time = time.time() - start_time
            self.metrics_collector.record_request(
                request_id=request_id,
                processing_time=processing_time,
                token_usage=result.get('token_usage', 0),
                cost=self.cost_optimizer.calculate_actual_cost(result),
                quality_score=result.get('evaluation', {}).get('overall_quality', 0)
            )
            
            return production_result
            
        except Exception as e:
            # Error handling and alerting
            self.handle_production_error(e, request_id, request)
            raise
        
        finally:
            # Health monitoring
            self.health_monitor.record_request_completion(request_id, time.time() - start_time)

class SelfContextGeneratingSystem:
    def __init__(self, base_llm, context_generator_llm):
        self.base_llm = base_llm
        self.context_generator = context_generator_llm
        self.context_quality_evaluator = ContextQualityEvaluator()
    
    async def process_with_generated_context(self, user_request, base_context=None):
        """Process request with AI-generated context"""
        
        # Analyze what context is needed
        context_analysis = await self.analyze_context_requirements(user_request, base_context)
        
        # Generate required context
        generated_context = await self.generate_missing_context(context_analysis)
        
        # Validate generated context quality
        context_quality = self.context_quality_evaluator.evaluate(generated_context, user_request)
        
        if context_quality < self.config.min_context_quality:
            # Regenerate with improved strategy
            generated_context = await self.regenerate_context_with_feedback(
                context_analysis, 
                generated_context, 
                context_quality
            )
        
        # Combine base and generated context
        full_context = self.merge_contexts(base_context or {}, generated_context)
        
        # Process with complete context
        return await self.base_llm.process_with_context(user_request, full_context)