Context Engineering

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
        }
    }

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
        }