Smithery is now a part of Arcade.dev! Read more in our announcement here
Comprehensive guide for building production-grade AI-integrated backends with multi-provider support, intelligent fallback mechanisms, region configuration, prompt management with variables/tools,...
1. Multi-Provider Design Philosophy
2. Separation of Concerns
3. Configuration Over Code
is_active flag to switch between configurations instantly4. Prompt Engineering Best Practices
{{variable_name}})5. Fallback & Resilience
6. Security & Authentication
User Request (Chat Message)
↓
API Endpoint (validate input, check session)
↓
Session Manager (create/resume session, lock pricing)
↓
External API Layer (if needed: fetch astrology, weather, etc.)
├─→ Cache Check (multi-layer: instance → distributed → database)
├─→ External API Call (if cache miss)
└─→ Data Transformation Pipeline (validate → normalize → enrich → format)
↓
Prompt Builder
├─→ Load Bot LLM Config (model, provider, temperature, metadata)
├─→ Render Prompt Template (replace {{variables}} with actual values)
├─→ Assemble Context (user data + external data + history)
└─→ Build Final Prompt (YAML or JSON structured)
↓
LLM Gateway Manager
├─→ Get/Generate JWT Token (cached)
├─→ Build Fallback Chain (provider-specific + universal)
└─→ POST to Internal Gateway (/v1/generate/)
↓
Internal LLM Gateway (separate service)
├─→ Validate JWT
├─→ Route to Primary Provider (OpenAI, Anthropic, Gemini, Bedrock, etc.)
├─→ On Failure: Try Fallback 1
├─→ On Failure: Try Fallback 2
└─→ On Failure: Universal Fallback (Gemini-Flash)
↓
Response Processing
├─→ Parse LLM Response
├─→ Validate Output
└─→ Store in Database
↓
Billing (if session-based)
├─→ Calculate Units (tokens, messages, minutes)
├─→ Check Wallet Balance
├─→ Deduct Amount
└─→ Log Transaction
↓
Save to Database (conversation history)
↓
Return to User (API response)
Request → Load BotLLMConfig → Determine Primary Provider
↓
Primary: Gemini 2.5-Pro
├─→ Fallback 1: Gemini 2.5-Flash
├─→ Fallback 2: Gemini 2.0-Flash
└─→ Universal Fallback: Gemini 2.0-Flash
Primary: AWS Bedrock Deepseek-v3 (region: ap-south-1)
├─→ Fallback 1: Byteplus Deepseek-v3
├─→ Fallback 2: Gemini 2.0-Flash (region-agnostic)
└─→ Universal Fallback: Gemini 2.0-Flash
Primary: Byteplus
├─→ Fallback 1: Gemini 2.0-Flash
└─→ Universal Fallback: Gemini 2.0-Flash
System Architecture:
Provider Integration:
Prompt Management:
External API Integration:
Production Considerations:
Code Examples:
class Bot(models.Model):
uuid = models.UUIDField(default=uuid4, editable=False, unique=True)
user = models.ForeignKey("users.UserProfile", on_delete=models.CASCADE, related_name="bots")
# Identity
name = models.CharField(max_length=255)
slug = models.SlugField(max_length=255, unique=True, blank=True, db_index=True)
description = models.TextField(blank=True)
avatar = models.CharField(max_length=255, null=True, blank=True)
# Status and Type
status = models.CharField(max_length=20, choices=BotStatus.choices, default=BotStatus.DRAFT, db_index=True)
bot_type = models.CharField(max_length=30, choices=BotType.choices, default=BotType.COMPANION)
# Configuration
rank = models.PositiveIntegerField(default=0, db_index=True)
metadata = models.JSONField(default=dict, blank=True) # Store required_fields, pricing, etc.
# Timestamps
launched_on = models.DateTimeField(null=True, blank=True, db_index=True)
created_on = models.DateTimeField(auto_now_add=True)
updated_on = models.DateTimeField(auto_now=True)
deleted_at = models.DateTimeField(null=True, blank=True, db_index=True)
Key Design Decisions:
metadata JSONField for flexible per-bot configuration (required_fields, character_rating, pricing)deleted_at timestamp instead of hard deleteslug auto-generated from name for SEO-friendly URLsrank for custom ordering in UIclass BotLLMConfig(models.Model):
bot = models.ForeignKey(Bot, on_delete=models.CASCADE, related_name="llm_configs")
# LLM Configuration
prompt = models.TextField(blank=True) # Can be template with {{variables}}
model_name = models.CharField(max_length=100) # e.g., "gpt-4", "claude-3-sonnet", "gemini-2.5-pro"
llm_provider = models.CharField(max_length=50, choices=BotLLMProvider.choices) # openai, anthropic, gemini, bedrock, etc.
# Generation Parameters
temperature = models.FloatField(default=0.7)
top_p = models.FloatField(default=0.95)
top_k = models.IntegerField(null=True, blank=True)
max_output_tokens = models.PositiveIntegerField(default=1024)
# Activation
is_active = models.BooleanField(default=False) # Only one active config per bot
# Additional Configuration
metadata = models.JSONField(default=dict, blank=True) # Store region_name, tools, etc.
created_on = models.DateTimeField(auto_now_add=True)
updated_on = models.DateTimeField(auto_now=True)
class Meta:
db_table = "llm_configurations"
ordering = ["-is_active", "created_on"]
def save(self, *args, **kwargs):
# Ensure only one active config per bot
if self.is_active:
BotLLMConfig.objects.filter(bot=self.bot, is_active=True).exclude(pk=self.pk).update(is_active=False)
super().save(*args, **kwargs)
Key Design Decisions:
Bot can have multiple BotLLMConfig entries for A/B testingis_active flag controls which config is used (only one active per bot)metadata stores provider-specific settings (e.g., region_name for AWS Bedrock)class BotLLMProvider(models.TextChoices):
OpenAI = "openai", _("OpenAI")
GEMINI = "gemini", _("Gemini")
ANTHROPIC = "anthropic", _("Anthropic")
GROQ = "groq", _("Groq")
AI_SDK = "ai-sdk", _("AI-SDK")
BEDROCK = "bedrock", _("Bedrock")
BYTEPLUS = "byteplus", _("Byteplus")
OPENROUTER = "openrouter", _("OpenRouter")
Why This Design:
def get_fallback_routing(provider: str, model_name: str, system_instruction: str, llm_metadata: dict = None):
"""
Ensures system_instruction and region_name are added properly to fallback chain.
Args:
provider: Primary provider (gemini, bedrock, byteplus, etc.)
model_name: Model identifier (gemini-2.5-pro, deepseek.v3-v1:0, etc.)
system_instruction: System prompt to pass to all fallbacks
llm_metadata: Additional metadata including region_name for Bedrock
Returns:
List of fallback configurations, each containing:
- provider: str
- model: str
- retry: int (0 = no retry, >0 = retry count)
- system_instruction: str
- region_name: str (for Bedrock only)
"""
provider = (provider or "").lower()
fallbacks = []
# Provider-specific fallback chains
if provider == "gemini":
if model_name == "gemini-2.5-pro":
fallbacks.append({"provider": "gemini", "model": "gemini-2.5-flash", "retry": 0})
else:
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
elif provider == "byteplus":
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
elif provider == "bedrock":
if model_name == "deepseek.v3-v1:0":
fallbacks.append({"provider": "byteplus", "model": "deepseek-v3-1-250821", "retry": 0})
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
else:
print(f"[LLM_GATEWAY_CLIENT] ⚠️ No specific fallback defined for provider: {provider}")
# Universal fallback (always added)
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
# Enrich all fallbacks with system_instruction and region_name
for fb in fallbacks:
fb["system_instruction"] = system_instruction
if fb["provider"] == "bedrock" and llm_metadata:
fb["region_name"] = llm_metadata.get("region_name", "ap-south-1")
return fallbacks
Example 1: Gemini 2.5-Pro Request
fallbacks = get_fallback_routing("gemini", "gemini-2.5-pro", "You are a helpful assistant")
# Result:
[
{"provider": "gemini", "model": "gemini-2.5-flash", "retry": 0, "system_instruction": "..."},
{"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0, "system_instruction": "..."},
{"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0, "system_instruction": "..."}
]
Example 2: AWS Bedrock Deepseek Request
fallbacks = get_fallback_routing(
"bedrock",
"deepseek.v3-v1:0",
"You are an astrologer",
llm_metadata={"region_name": "ap-south-1"}
)
# Result:
[
{"provider": "byteplus", "model": "deepseek-v3-1-250821", "retry": 0, "system_instruction": "..."},
{"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0, "system_instruction": "..."},
{"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0, "system_instruction": "..."}
]
# Pseudo-code for LLM Gateway
def generate_with_fallbacks(primary_config, fallback_chain):
try:
return call_llm(primary_config)
except Exception as e:
log_error(f"Primary provider failed: {e}")
for fallback in fallback_chain:
try:
return call_llm(fallback)
except Exception as fe:
log_error(f"Fallback {fallback['provider']}/{fallback['model']} failed: {fe}")
continue
# All fallbacks exhausted
raise Exception("All LLM providers failed")
retry: 0 means don't retry within fallback, just move to next# Store region in BotLLMConfig metadata
bot_llm_config = BotLLMConfig.objects.create(
bot=bot,
llm_provider="bedrock",
model_name="deepseek.v3-v1:0",
metadata={
"region_name": "ap-south-1", # AWS region
"additional_config": {...}
}
)
# When building fallback chain
fallbacks = get_fallback_routing(
provider="bedrock",
model_name="deepseek.v3-v1:0",
system_instruction=prompt,
llm_metadata=bot_llm_config.metadata
)
def get_bedrock_region_fallbacks(model_name, regions=["ap-south-1", "us-east-1", "eu-west-1"]):
"""
Create fallback chain across multiple AWS regions.
"""
fallbacks = []
for region in regions:
fallbacks.append({
"provider": "bedrock",
"model": model_name,
"region_name": region,
"retry": 0
})
# Add non-Bedrock fallbacks
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
return fallbacks
# Primary region from user location or bot config
primary_region = determine_region_from_user(user) # e.g., "ap-south-1" for India
# Fallback regions based on latency/availability
fallback_regions = ["us-east-1", "eu-west-1"]
# Build full chain
region_chain = [primary_region] + fallback_regions
class PromptTemplateHelper:
"""
Helper class for rendering prompt templates by resolving variable placeholders.
"""
@classmethod
def render_template(cls, prompt_text: str, **kwargs) -> str:
"""
Render a prompt template by replacing variable placeholders with actual values.
Example:
prompt = "Hi {{user_name}}, today is {{current_date}}."
result = PromptTemplateHelper.render_template(prompt, user_doc=user_doc)
# Output: "Hi John, today is 15 January 2026."
"""
function_map = cls.get_function_mapping()
# Find all {{variable_name}} in the prompt
variables = re_findall(r"\{\{(.*?)\}\}", prompt_text)
for var in variables:
function = function_map.get(var)
if function:
value = function(**kwargs)
prompt_text = prompt_text.replace(f"{{{{{var}}}}}", str(value or ""))
return prompt_text
AVAILABLE_VARIABLES = [
# Bot-specific variables
{"name": "{{bot_name}}", "description": "Name of the bot", "category": "Bot Info"},
{"name": "{{bot_description}}", "description": "Bot description", "category": "Bot Info"},
{"name": "{{bot_scene_desc}}", "description": "Bot Scene description", "category": "Bot Info"},
# User interaction variables
{"name": "{{user_name}}", "description": "User name", "category": "User Input"},
{"name": "{{user_age}}", "description": "User age", "category": "User Input"},
{"name": "{{user_gender}}", "description": "User gender", "category": "User Input"},
{"name": "{{user_message}}", "description": "Current user message", "category": "User Input"},
{"name": "{{conversation_history}}", "description": "Conversation History", "category": "User Input"},
{"name": "{{user_astro_data}}", "description": "Astro data for the user", "category": "User Input"},
# System variables
{"name": "{{current_time}}", "description": "Current time", "category": "System"},
{"name": "{{current_date}}", "description": "Current date", "category": "System"},
{"name": "{{day_of_week}}", "description": "Current day of week", "category": "System"},
# Common prompt variables
{"name": "{{random_number_1_4}}", "description": "Generate random number 1-4", "category": "Common"},
]
@classmethod
def get_function_mapping(cls):
"""Return mapping of variable names to handler functions."""
return {
# Bot-specific variables
"bot_name": cls.function_bot_name,
"bot_description": cls.function_bot_description,
"bot_scene_desc": cls.function_bot_scene_desc,
# User interaction variables
"user_name": cls.function_user_name,
"user_age": cls.function_user_age,
"user_gender": cls.function_user_gender,
"user_message": cls.function_user_message,
"user_astro_data": cls.function_user_astro_data,
# System variables
"current_time": cls.function_current_time,
"current_date": cls.function_current_date,
"day_of_week": cls.function_day_of_week,
"conversation_history": cls.function_conversation_history,
# Common prompt variables
"random_number_1_4": cls.function_random_number_1_4,
}
@classmethod
def function_current_time(cls, **kwargs):
"""Return the current system time."""
return UtilHelper.timezone_to_ist(datetime.now()).strftime("%I:%M %p")
@classmethod
def function_current_date(cls, **kwargs):
"""Return the current system date."""
return UtilHelper.timezone_to_ist(datetime.now()).strftime("%d %B %Y")
@classmethod
def function_user_name(cls, **kwargs):
"""Return user name from user_doc."""
return cls.get_sanitized_user_doc(**kwargs).get("name", "")
# Define template in database or config
prompt_template = """
You are {{bot_name}}, a helpful assistant.
User Information:
- Name: {{user_name}}
- Age: {{user_age}}
- Gender: {{user_gender}}
Current Context:
- Date: {{current_date}}
- Time: {{current_time}}
- Day: {{day_of_week}}
User Message: {{user_message}}
Please respond helpfully and naturally.
"""
# Render with actual values
rendered_prompt = PromptTemplateHelper.render_template(
prompt_template,
bot_doc={"name": "Mira", "description": "Vedic astrologer"},
user_doc={"name": "John", "age": 28, "gender": "male", "messages": "What's my future?"}
)
# Output:
"""
You are Mira, a helpful assistant.
User Information:
- Name: John
- Age: 28
- Gender: male
Current Context:
- Date: 15 January 2026
- Time: 02:45 PM
- Day: Thursday
User Message: What's my future?
Please respond helpfully and naturally.
"""
@classmethod
def get_mira_prompt(cls, **kwargs):
"""
Build YAML-structured prompt for astrology bot.
Required kwargs: bot_doc, language, app_id
Optional: memories, profile_details_1, profile_details_2
"""
bot_doc = kwargs.get("bot_doc")
_prompt = {
"Role": f"You are {bot_doc.get('name')}, an experienced {str.upper(bot_doc.get('sex', ''))} Vedic astrologer with expertise in predictive astrology...",
"Query House Mapping": {
"Marriage": {
"Houses To Analyze": "7, 2, 11, 8",
"Karaka Planets": "Venus, Jupiter",
"Special Checks": "Manglik status, 7th lord placement, Venus strength, Jupiter aspects",
"Timing Triggers": "7th lord dasha, Venus period, Jupiter transit to 7th",
},
"Career": {
"Houses To Analyze": "10, 1, 2, 6, 11",
"Karaka Planets": "Sun, Saturn, Mercury",
"Special Checks": "10th lord strength, Sun placement, D9 10th house",
"Timing Triggers": "10th lord dasha, Saturn period, Sun antardasha",
},
# ... more query types
},
"Analysis Steps": {
"Core": "Understand the question AND the feeling underneath. Map to houses, planets, timing.",
"Analysis": "Houses → Lords → Karakas → Dashas → Divisionals → Yogas → Doshas.",
"Timing": "Find activation windows. Always check next 60 days for shifts.",
"Response Crafting": "Give analysis and response following guidelines below.",
},
"Response Format": {
"Structure": "A single paragraph text based on the instructions",
"Style": "Insightful, Helpful, focussed, non-repetitive",
"Language": cls.get_mira_prompt_language_blob(**kwargs),
"Response Instructions": [
"Check if greeting is necessary, if yes greet with a smile",
"Based on Kundli, make a compliment and find a coincidence with user",
"Check kundli for last 3 months and find something user has faced",
"Provide good news, something to look forward to in next 3 months",
"Answer user query using Vedic astrology principles",
"Provide timing predictions with high/low chance of happening",
"Before closing, send follow-up question based on findings",
],
"Length": cls.get_prompt_response_length_text(**kwargs),
},
"Guidelines": [
"Answer astro questions based on Vedic astrology principles",
"Keep user gender in check at all times",
"Balance traditional predictions with modern life",
"Do not diagnose or prescribe for health",
"Do not suggest life-threatening actions",
"Do not ask to meet or send physical items",
"Do not guarantee real-world outcomes, only state possibilities",
"If credibility questioned, state - I am AI powered but validated by experienced astrologer",
],
"Daily Context": {
"Date": PromptTemplateHelper.function_current_date(),
"Time": PromptTemplateHelper.function_current_time(),
"Day of week": PromptTemplateHelper.function_day_of_week(),
},
}
# Add conditional sections
if memories := kwargs.get("memories"):
_prompt["Past User Memories"] = memories
if profile_details_1 := kwargs.get("profile_details_1"):
_prompt["First Profile Kundli"] = profile_details_1
if profile_details_2 := kwargs.get("profile_details_2"):
_prompt["Second Profile Kundli"] = profile_details_2
# Convert to YAML string
_yaml_prompt = yaml.dump(_prompt, sort_keys=False, default_flow_style=False)
return _yaml_prompt
@classmethod
def get_mira_prompt_language_blob(cls, **kwargs):
language_blob = "Modern Hindi language, written in Roman font. Do not use Devanagari font"
blob_map = {
("4", str(Languages.HINDI)): "Modern Hindi language, written in Devanagari font. Do not use english font",
("4", str(Languages.HINGLISH)): "Modern Hindi language, written in Roman font. Do not use Devanagari font",
("4", str(Languages.ENGLISH)): "Modern English language, written in Roman font. Do not use Devanagari font",
}
language = kwargs.get("language")
app_id = kwargs.get("app_id")
if language:
if _blob := blob_map.get((str(app_id), str(language))):
language_blob = _blob
return language_blob
class InternalLLMGatewayManager:
"""
Manages authentication and request handling for the LLM Gateway.
Provides:
- JWT token generation and caching
- Preconfigured RequestManager for LLM Gateway API
"""
@classmethod
def _get_jwt_token(cls, seconds: int = 3600) -> str:
"""
Retrieve or generate a cached JWT token for LLM Gateway authentication.
Token is cached for (expiry - 2 seconds) to avoid race conditions.
"""
key = CacheKey.LLM_GATEWAY_TOKEN.format(seconds=seconds)
jwt_token = CacheManager.get(key)
if jwt_token is None:
jwt_token = cls._generate_jwt_token(seconds=seconds)
CacheManager.set(key, jwt_token, duration=seconds - 2)
return jwt_token
@classmethod
def _generate_jwt_token(cls, seconds: int = 3600) -> str:
"""
Generate a new JWT token for authenticating with the LLM Gateway.
"""
payload = {
"service": "django-backend",
"exp": timezone.now() + timedelta(seconds=seconds),
}
return jwt_encode(
payload,
settings.LLM_GATEWAY_JWT_KEY,
algorithm=settings.LLM_GATEWAY_JWT_ALGORITHM,
)
@classmethod
def _request_gateway(cls) -> RequestManager:
"""
Create and return a configured RequestManager instance for the LLM Gateway.
Ensures valid JWT token is attached in Authorization header.
"""
jwt_token = cls._get_jwt_token()
if not jwt_token:
raise ValueError("LLM Gateway token is not available or invalid.")
if not settings.LLM_GATEWAY_HOST:
raise ValueError("LLM_GATEWAY_HOST is not defined in settings")
headers = {
"Authorization": f"jwt {jwt_token}",
"Content-Type": "application/json",
"app-identifier": "dev-django" if getattr(settings, "IS_TESTING_SERVER", False) else "prod-django",
}
return RequestManager(
base_url=f"{settings.LLM_GATEWAY_HOST}/",
headers=headers,
)
@classmethod
@observe(as_type="generation") # Langfuse observability
def generate_response(cls, payload):
"""
Send generation request to LLM Gateway.
Payload structure:
{
"provider": "gemini",
"model": "gemini-2.5-pro",
"system_instruction": "You are...",
"messages": [...],
"temperature": 0.7,
"max_tokens": 1024,
"fallbacks": [...],
"metadata": {"region_name": "ap-south-1"}
}
"""
return cls._request_gateway().post("v1/generate/", json=payload)
# twofourlabs/settings.py
LLM_GATEWAY_JWT_KEY = COMMON_SECRETS.get("llm_gateway_jwt_key", "dummy_key")
LLM_GATEWAY_JWT_ALGORITHM = COMMON_SECRETS.get("llm_gateway_jwt_algorithm", "HS256")
LLM_GATEWAY_HOST = INTERNAL_HOSTS.get("llm_gateway") # e.g., "https://llm-gateway.internal.example.com"
# Build payload
payload = {
"provider": "gemini",
"model": "gemini-2.5-pro",
"system_instruction": rendered_prompt,
"messages": conversation_history,
"temperature": 0.7,
"max_tokens": 1024,
"fallbacks": get_fallback_routing("gemini", "gemini-2.5-pro", rendered_prompt)
}
# Send to gateway (JWT handled automatically)
response = InternalLLMGatewayManager.generate_response(payload)
# Parse response
ai_message = response.json()["choices"][0]["message"]["content"]
# Step 1: Load Bot Configuration
bot = Bot.objects.get(slug="astrologer-mira")
llm_config = bot.llm_configs.filter(is_active=True).first()
# Step 2: Fetch External Data (Astrology API)
astro_data = fetch_kundli_data(user_birth_details) # External API call
# Step 3: Build Prompt with Template Variables
prompt_text = PromptHelper.get_prompt(
app_identifier=AppIdentifier.iOS_MIRA,
bot_doc={
"name": bot.name,
"description": bot.description,
"sex": "female"
},
user_doc={
"name": user.name,
"age": user.age,
"gender": user.gender,
},
language=Languages.ENGLISH,
app_id="4",
memories=user_memories,
profile_details_1=astro_data
)
# Step 4: Build Fallback Chain
fallbacks = PromptHelper.get_fallback_routing(
provider=llm_config.llm_provider,
model_name=llm_config.model_name,
system_instruction=prompt_text,
llm_metadata=llm_config.metadata
)
# Step 5: Prepare LLM Request Payload
payload = {
"provider": llm_config.llm_provider,
"model": llm_config.model_name,
"system_instruction": prompt_text,
"messages": conversation_history,
"temperature": llm_config.temperature,
"max_tokens": llm_config.max_output_tokens,
"top_p": llm_config.top_p,
"fallbacks": fallbacks,
"metadata": llm_config.metadata
}
# Step 6: Call LLM Gateway
response = InternalLLMGatewayManager.generate_response(payload)
# Step 7: Parse and Save Response
ai_message = response.json()["choices"][0]["message"]["content"]
ChatHistory.objects.create(
session=session,
sender_type="bot",
message=ai_message
)
# Step 8: Billing (if applicable)
if session.is_paid:
deduct_from_wallet(user, session.per_message_price)
# Step 9: Return to User
return {"message": ai_message, "session_id": session.id}
# Step 1: Add to BotLLMProvider choices
class BotLLMProvider(models.TextChoices):
# ... existing providers
MISTRAL = "mistral", _("Mistral AI")
# Step 2: Create BotLLMConfig for new provider
BotLLMConfig.objects.create(
bot=bot,
llm_provider="mistral",
model_name="mistral-large-latest",
temperature=0.7,
top_p=0.95,
max_output_tokens=1024,
is_active=False, # A/B test alongside existing config
metadata={"api_endpoint": "https://api.mistral.ai"}
)
# Step 3: Add fallback logic
def get_fallback_routing(provider, model_name, system_instruction, llm_metadata=None):
# ... existing code
elif provider == "mistral":
if model_name == "mistral-large-latest":
fallbacks.append({"provider": "mistral", "model": "mistral-medium-latest", "retry": 0})
fallbacks.append({"provider": "gemini", "model": "gemini-2.0-flash", "retry": 0})
# ... rest of code
# Create two configs for same bot
config_a = BotLLMConfig.objects.create(
bot=bot,
llm_provider="gemini",
model_name="gemini-2.5-pro",
temperature=0.7,
is_active=True, # Currently active
metadata={"experiment": "model_a"}
)
config_b = BotLLMConfig.objects.create(
bot=bot,
llm_provider="anthropic",
model_name="claude-3-sonnet-20240229",
temperature=0.7,
is_active=False, # Ready to activate
metadata={"experiment": "model_b"}
)
# Switch to config B (no code deployment needed!)
config_b.is_active = True
config_b.save() # This automatically sets config_a.is_active = False
# All new requests now use Claude instead of Gemini
User Request
↓
Fetch External Data (Astrology, Weather, Finance API)
├─→ Cache Check (multi-layer)
├─→ API Call (if cache miss)
└─→ Save to Database
↓
Data Transformation Pipeline
├─→ Validate (ensure required fields present)
├─→ Normalize (standardize format)
├─→ Enrich (add computed fields)
└─→ Format for AI (XML, JSON, Markdown)
↓
Inject into Prompt Template
↓
LLM Call with Multi-Provider Fallbacks
↓
Response to User
class DataFetcher:
def __init__(self):
self._instance_cache = {} # Request-scope cache
def get_kundli_data(self, user_id, birth_details):
# Layer 1: Instance cache (fastest, request scope)
cache_key = f"kundli_{user_id}"
if cache_key in self._instance_cache:
return self._instance_cache[cache_key]
# Layer 2: Distributed cache (Redis, 10 min TTL)
cached = CacheManager.get(cache_key)
if cached:
self._instance_cache[cache_key] = cached
return cached
# Layer 3: Database (persistent)
db_record = KundliData.objects.filter(user_id=user_id).first()
if db_record and db_record.is_fresh:
data = db_record.data
CacheManager.set(cache_key, data, duration=600)
self._instance_cache[cache_key] = data
return data
# Layer 4: External API call
data = self._call_astrology_api(birth_details)
# Write back to all layers
KundliData.objects.update_or_create(user_id=user_id, defaults={"data": data})
CacheManager.set(cache_key, data, duration=600)
self._instance_cache[cache_key] = data
return data
def transform_astro_data(raw_api_response):
# Step 1: Validate
if not validate_astro_response(raw_api_response):
raise ValueError("Invalid astrology API response")
# Step 2: Normalize
normalized = {
"ascendant": raw_api_response.get("Ascendant", "").capitalize(),
"planets": normalize_planet_positions(raw_api_response["planet_chart_data"]),
"dashas": normalize_dashas(raw_api_response["current_vdasha"])
}
# Step 3: Enrich
enriched = {
**normalized,
"processed_at": datetime.now().isoformat(),
"summary": generate_summary(normalized["planets"])
}
# Step 4: Format for AI (Markdown)
formatted = f"""
### Kundli Data
**Ascendant**: {enriched["ascendant"]}
**Planetary Positions**:
{format_planets_markdown(enriched["planets"])}
**Current Dasha**: {enriched["dashas"]["major"]} / {enriched["dashas"]["minor"]}
**Summary**: {enriched["summary"]}
"""
return formatted
Request LLM Service
↓
Check Cache for JWT Token
↓
If Not Cached:
Generate JWT Token
├─→ Payload: {"service": "django-backend", "exp": now + 3600s}
├─→ Sign with Secret Key (HS256)
└─→ Cache for (3600 - 2) seconds
↓
Attach to Request Header: "Authorization: jwt {token}"
↓
Send to LLM Gateway
↓
Gateway Validates JWT
├─→ Verify Signature
├─→ Check Expiry
└─→ Allow/Deny Request
from langfuse import observe
class InternalLLMGatewayManager:
@classmethod
@observe(as_type="generation") # Automatically tracks LLM calls
def generate_response(cls, payload):
return cls._request_gateway().post("v1/generate/", json=payload)
When implementing a new AI feature:
Bot, BotLLMConfig following BotLLMProvider.choices, implement in gatewayget_fallback_routing(){{variable}} syntax, define in PromptHelperInternalLLMGatewayManager.generate_response()@observe decorator is presentCommon Use Cases:
Do NOT use this skill for:
{{variable}} templates decoupled from code@observe decorator on LLM calls)is_active flag (no code changes)