Context Engineering

Prepare comprehensive user context for AI assistants and LLM applications by orchestrating parallel data fetching and intelligent summarization.

Overview

This example demonstrates sophisticated context engineering patterns:

Parallel data fetching from multiple APIs
LLM-powered summarization for context compression
Orchestrated task workflows with dependency management
Bulk processing for multiple users
Context caching and optimization strategies

Perfect for building personalized AI assistants, recommendation engines, or any LLM application requiring rich user context.

Task Definitions

from hyrex import HyrexRegistry
import openai
import requests
from typing import List, Dict

hy = HyrexRegistry()

@hy.task
def fetch_user_profile(user_id: str) -> Dict:
    """Fetch user profile data"""
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return response.json()

@hy.task
def fetch_recent_activities(user_id: str, limit: int = 50) -> List[Dict]:
    """Fetch user's recent activities"""
    response = requests.get(
        f"https://api.example.com/users/{user_id}/activities?limit={limit}"
    )
    return response.json()

@hy.task
def fetch_user_preferences(user_id: str) -> Dict:
    """Fetch user preferences and settings"""
    response = requests.get(f"https://api.example.com/users/{user_id}/preferences")
    return response.json()

@hy.task
def build_context_summary(user_id: str, profile: Dict, activities: List[Dict], preferences: Dict) -> str:
    """Use LLM to create a context summary"""
    context_data = {
        "profile": profile,
        "recent_activities": activities[:10],  # Last 10 activities
        "preferences": preferences
    }
    
    prompt = f"""
    Create a concise summary of this user's context for an AI assistant:
    
    Profile: {context_data['profile']}
    Recent Activities: {context_data['recent_activities']}
    Preferences: {context_data['preferences']}
    
    Focus on the most relevant information for personalized assistance.
    """
    
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=500
    )
    
    return response.choices[0].message.content

@hy.task
def prepare_llm_context(user_id: str):
    """Orchestrate parallel context preparation"""
    # Launch all data fetching tasks in parallel
    profile_task = fetch_user_profile.send(user_id)
    activities_task = fetch_recent_activities.send(user_id, 50)
    preferences_task = fetch_user_preferences.send(user_id)
    
    # Wait for all tasks to complete
    profile = profile_task.get()
    activities = activities_task.get()  
    preferences = preferences_task.get()
    
    # Build final context summary
    context_summary = build_context_summary.send(
        user_id, profile, activities, preferences
    ).get()
    
    return {
        "user_id": user_id,
        "context_summary": context_summary,
        "raw_data": {
            "profile": profile,
            "activities": activities,
            "preferences": preferences
        }
    }

REST API Endpoints

from fastapi import FastAPI
from pydantic import BaseModel
from .tasks import prepare_llm_context

app = FastAPI()

class ContextRequest(BaseModel):
    user_id: str
    include_raw_data: bool = False
    
class BulkContextRequest(BaseModel):
    user_ids: list[str]
    include_raw_data: bool = False

@app.post("/context/prepare")
async def prepare_user_context(request: ContextRequest):
    """Prepare context for AI/LLM consumption"""
    task = prepare_llm_context.send(request.user_id)
    
    return {
        "message": "Context preparation started",
        "task_id": task.task_id,
        "user_id": request.user_id
    }

@app.post("/context/bulk-prepare") 
async def prepare_bulk_contexts(request: BulkContextRequest):
    """Prepare context for multiple users in parallel"""
    task_ids = []
    for user_id in request.user_ids:
        task = prepare_llm_context.send(user_id)
        task_ids.append({
            "user_id": user_id,
            "task_id": task.task_id
        })
    
    return {
        "message": f"Started context preparation for {len(request.user_ids)} users",
        "tasks": task_ids
    }

@app.get("/context/status/{task_id}")
async def get_context_status(task_id: str):
    """Check if context preparation is complete"""
    task = hy.get_task(task_id)
    
    if task.is_complete:
        return {
            "status": "complete",
            "context_data": task.result
        }
    else:
        return {
            "status": "processing",
            "progress": task.progress
        }

Advanced Context Patterns

Context Caching

from hyrex import HyrexKV
import json

@hy.task
def get_cached_context(user_id: str, max_age_minutes: int = 60):
    """Get context from cache or prepare fresh"""
    cache_key = f"user_context:{user_id}"
    
    try:
        cached_data = HyrexKV.get(cache_key)
        context_info = json.loads(cached_data)
        
        # Check if cache is still fresh
        cached_time = datetime.fromisoformat(context_info['timestamp'])
        age_minutes = (datetime.now() - cached_time).total_seconds() / 60
        
        if age_minutes < max_age_minutes:
            return context_info['context']
    except:
        pass
    
    # Cache miss or stale - prepare fresh context
    fresh_context = prepare_llm_context.send(user_id).get()
    
    # Cache the result
    HyrexKV.set(cache_key, json.dumps({
        'context': fresh_context,
        'timestamp': datetime.now().isoformat()
    }), expiry_seconds=max_age_minutes * 60)
    
    return fresh_context

Context Personalization

@hy.task
def personalize_context(base_context: dict, interaction_type: str):
    """Customize context based on interaction type"""
    prompt = f"""
    Adapt this user context for a {interaction_type} interaction:
    
    {base_context['context_summary']}
    
    Emphasize the most relevant aspects for {interaction_type}.
    Keep it under 200 tokens.
    """
    
    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200
    )
    
    return {
        "interaction_type": interaction_type,
        "personalized_context": response.choices[0].message.content,
        "original_context": base_context
    }

Usage Examples

# Prepare context for a single user
curl -X POST http://localhost:8000/context/prepare \
  -H "Content-Type: application/json" \
  -d '{"user_id": "user123"}'

# Bulk prepare contexts for multiple users
curl -X POST http://localhost:8000/context/bulk-prepare \
  -H "Content-Type: application/json" \
  -d '{
    "user_ids": ["user123", "user456", "user789"],
    "include_raw_data": false
  }'

# Check context preparation status
curl http://localhost:8000/context/status/task_12345

Context Engineering Best Practices

1. Parallel Processing

Fetch different data sources simultaneously
Use task orchestration to manage dependencies
Optimize for speed when building real-time context

2. Context Compression

Use LLMs to summarize large amounts of raw data
Focus on relevance for the specific use case
Balance detail with token limits

3. Caching Strategy

Cache context to avoid expensive rebuilds
Set appropriate TTL based on data freshness needs
Use cache invalidation for critical data changes

4. Privacy & Security

Implement proper data access controls
Sanitize sensitive information before LLM processing
Log context usage for audit trails

Production Considerations

Rate limiting: Manage API calls to external services and LLM providers
Fallback handling: Provide default context when data sources fail
Cost optimization: Balance context richness with LLM token costs
Performance: Monitor context preparation latency and optimize bottlenecks
Data freshness: Implement appropriate caching and invalidation strategies

Overview

SDK Reference

Examples

Production Guide

Context Engineering

Overview

Task Definitions

REST API Endpoints

Advanced Context Patterns

Context Caching

Context Personalization

Usage Examples

Context Engineering Best Practices

1. Parallel Processing

2. Context Compression

3. Caching Strategy

4. Privacy & Security

Production Considerations

Next Steps

Document Embeddings

Error Handling

Overview

SDK Reference

Examples

Production Guide

​Overview

​Task Definitions

​REST API Endpoints

​Advanced Context Patterns

​Context Caching

​Context Personalization

​Usage Examples

​Context Engineering Best Practices

​1. Parallel Processing

​2. Context Compression

​3. Caching Strategy

​4. Privacy & Security

​Production Considerations

​Next Steps

Document Embeddings

Error Handling

Overview

Task Definitions

REST API Endpoints

Advanced Context Patterns

Context Caching

Context Personalization

Usage Examples

Context Engineering Best Practices

1. Parallel Processing

2. Context Compression

3. Caching Strategy

4. Privacy & Security

Production Considerations

Next Steps