Business Logic Adapters Guide
Business Logic Adapters are the heart of the Universal AI Platform's customization system. They allow you to tailor agent behavior for specific use cases without modifying the core platform.
What Are Business Logic Adapters?
Business Logic Adapters are dynamic context providers - they don't dictate your application logic, but intelligently gather and provide the AI with relevant context based on user queries. Your application maintains full control over the user experience and workflow.
Business Logic Adapters are pluggable components that:
- Dynamically fetch relevant information based on user queries
- Provide domain-specific context to guide AI responses
- Add safety and validation checks for critical applications
- Format and structure data for consistent AI understanding
- Prevent hallucination through targeted guidelines
- Enable safe response handling when validation fails
Key Principle: Adapters provide context, your application provides logic.
Platform vs. Developer Responsibilities
What the Platform Provides
- Safety Framework: Built-in anti-hallucination protection and response validation
- API Safety: All underlying AI APIs include harmful content detection
- Adapter Architecture: Tools for dynamic context fetching and processing
- Security Tools: Input sanitization and output filtering capabilities
What Developers Control
- Context Content: All domain-specific information and business data
- Data Sources: How and where context information is fetched
- Implementation: Whether to use static context or dynamic information retrieval
- Business Logic: Application workflow and user experience
Context Flexibility
Developers have complete freedom in how they provide context:
Option 1: Static Context (Simple)
class SimpleAdapter(BusinessLogicAdapter):
def process_user_input(self, user_input, session_context):
# Hardcoded business context
session_context['company_info'] = {
'name': 'Acme Corp',
'policies': ['30-day return', 'free shipping over $50'],
'support_hours': '9 AM - 5 PM EST'
}
return user_input, session_context
Option 2: Dynamic Context (Advanced)
class DynamicAdapter(BusinessLogicAdapter):
def process_user_input(self, user_input, session_context):
# Fetch real-time information from your systems
customer_id = session_context.get('customer_id')
if customer_id:
# Your database, your rules
customer_data = self.crm_api.get_customer(customer_id)
order_history = self.orders_api.get_recent_orders(customer_id)
session_context['customer_context'] = {
'tier': customer_data['tier'],
'recent_orders': order_history,
'preferences': customer_data['preferences']
}
return user_input, session_context
Important: The platform has no visibility into your context content - you control all business data and information flow.
Safety and Monitoring Boundaries
Platform-Level Safety (Automatic)
- Harmful Content Detection: All AI APIs include built-in protection against harmful, offensive, or dangerous content
- Model Safety: Underlying language models have training-level safety measures
- API Rate Limiting: Prevents abuse and ensures fair usage
Developer Responsibility
- Context Content: All business-specific information you provide in adapters
- Data Privacy: Ensuring your data handling complies with regulations (GDPR, HIPAA, etc.)
- Business Logic: How your application responds to AI outputs
- User Safety: Implementing appropriate safeguards for your specific use case
No Platform Monitoring of Your Context
The Universal AI Platform cannot and does not monitor:
- What business data you include in your adapter context
- How you fetch or structure your information
- The content of your databases or APIs
- Your internal business processes or policies
This ensures:
- Privacy: Your business data remains completely private
- Flexibility: You can implement any business logic without platform restrictions
- Compliance: You maintain full control over data handling and regulatory compliance
Dynamic Context Fetching
Adapters can intelligently fetch relevant information based on user queries, making them powerful context enrichment engines:
Query-Based Information Retrieval
class CustomerServiceAdapter(BusinessLogicAdapter):
def process_user_input(self, user_input, session_context):
"""Dynamically fetch context based on user query"""
# Detect what information the user is asking about
if self._is_product_question(user_input):
# Fetch product details from your database/API
product_info = self._fetch_product_context(user_input)
session_context['product_context'] = product_info
elif self._is_order_question(user_input):
# Get order information for this customer
order_data = self._fetch_order_context(user_input, session_context)
session_context['order_context'] = order_data
elif self._is_policy_question(user_input):
# Retrieve relevant policies and procedures
policies = self._fetch_policy_context(user_input)
session_context['policy_context'] = policies
return user_input, session_context
def _fetch_product_context(self, query):
"""Search and retrieve relevant product information"""
# Extract product names/IDs from query
products = self._extract_product_references(query)
# Fetch from your product database
product_data = []
for product in products:
data = self.product_api.get_product_details(product)
product_data.append({
'name': data['name'],
'features': data['features'],
'compatibility': data['compatibility'],
'common_issues': data['support_faq']
})
return product_data
Real-Time Data Integration
class EmergencyServicesAdapter(BusinessLogicAdapter):
def process_user_input(self, user_input, session_context):
"""Fetch real-time emergency response data"""
# Extract location from user input
location = self._extract_location(user_input)
if location:
# Get real-time responder availability
responder_status = self._fetch_responder_availability(location)
session_context['responder_context'] = responder_status
# Check for active incidents in area
area_incidents = self._fetch_area_incidents(location)
session_context['area_context'] = area_incidents
# Get hospital/facility availability
facility_status = self._fetch_facility_capacity(location)
session_context['facility_context'] = facility_status
return user_input, session_context
def _fetch_responder_availability(self, location):
"""Get real-time responder status"""
return self.dispatch_api.get_available_units(
location=location,
radius_miles=10
)
Contextual Knowledge Enrichment
class HealthcareAdapter(BusinessLogicAdapter):
def process_user_input(self, user_input, session_context):
"""Enrich context with relevant medical information"""
# Extract symptoms and conditions mentioned
medical_entities = self._extract_medical_entities(user_input)
if medical_entities['symptoms']:
# Fetch relevant medical context (not advice!)
symptom_context = self._fetch_symptom_information(medical_entities['symptoms'])
session_context['symptom_context'] = symptom_context
if medical_entities['medications']:
# Get medication interaction warnings
interaction_data = self._fetch_interaction_data(medical_entities['medications'])
session_context['interaction_warnings'] = interaction_data
# Check if patient has previous visits/records
patient_id = session_context.get('patient_id')
if patient_id:
visit_history = self._fetch_recent_visits(patient_id)
session_context['visit_context'] = visit_history
return user_input, session_context
Architecture
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ User Input │───▶│ Business Logic │───▶│ AI Agent │
│ │ │ Adapter │ │ │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│ │
▼ ▼
┌──────────────────┐ ┌─────────────────┐
│ Context Provider │ │ AI Response │
│ Safety Checks │ │ with Context │
└──────────────────┘ └─────────────────┘
│
▼
┌──────────────────┐
│ Anti-Hallucination│
│ Validation │
└──────────────────┘
Use Cases & Examples
The beauty of business logic adapters is their flexibility. Here are real-world scenarios:
Emergency Services - Pre-Response Information Gathering
- Context: Format emergency data for dispatcher systems
- Your App Logic: Integration with CAD systems, responder dispatch
- AI Role: Gather structured information, provide caller support
Language Learning - Conversation Partner
- Context: Adjust language complexity, cultural nuances
- Your App Logic: Progress tracking, lesson planning, user management
- AI Role: Conversational practice, gentle correction
Customer Service - Support Assistant
- Context: Product knowledge, company policies, escalation rules
- Your App Logic: Ticket management, CRM integration, analytics
- AI Role: First-line support, information gathering
Healthcare - Information Assistant
- Context: Medical terminology, privacy requirements, escalation triggers
- Your App Logic: EHR integration, appointment scheduling, provider routing
- AI Role: Symptom gathering, appointment assistance (not diagnosis)
## Anti-Hallucination Protection
All business logic adapters include built-in protection against AI hallucination - crucial for production applications where accuracy matters.
### How It Works
```python
# Automatic detection of problematic responses
hallucination_indicators = [
'according to my knowledge', # False confidence
'i believe', 'probably', # Uncertainty disguised as fact
'i can help with anything', # Overpromising capabilities
'let me check that for you' # False action claims
]
# Note: Customize these indicators based on your business domain.
# For example, a payment processor might flag mentions of unsupported payment methods,
# while a healthcare service might flag inappropriate medical advice claims.
# Response validation
def validate_response(response):
risk_level = assess_hallucination_risk(response)
if risk_level == 'high':
return safe_fallback_response()
return response
Safety Features
- Confidence Detection: Flags responses that sound certain about uncertain information
- Domain Boundaries: Prevents AI from claiming capabilities outside its scope
- Fallback Responses: Provides safe alternatives when validation fails
- Context Adherence: Ensures responses stay within provided context
Best Practices
- Provide Clear Context: The more specific your context, the less likely the AI will hallucinate
- Set Domain Boundaries: Explicitly state what the AI should and shouldn't do
- Use Validation: Always validate responses in critical applications
- Monitor Responses: Review AI outputs regularly, especially in early deployment
# Example: Safe emergency services context
emergency_context = """
Role: Information gathering assistant for emergency dispatch
Boundaries:
- Do NOT provide medical advice
- Do NOT diagnose conditions
- Do NOT recommend treatments
- DO gather location, situation details
- DO provide reassurance and support
- DO escalate to human dispatcher when needed
"""
Built-in Adapters
Language Learning Adapter
Optimized for educational applications with features like:
- Target language detection and context switching
- Proficiency level adjustment for appropriate difficulty
- Gentle error correction without discouraging users
- Conversation topic guidance based on learning objectives
- Progress tracking and vocabulary building
Configuration:
config = {
"target_language": "Spanish",
"proficiency_level": "beginner",
"conversation_topics": ["daily activities", "food", "travel"],
"correction_style": "gentle",
"vocabulary_focus": ["present_tense", "basic_nouns"]
}
Example Usage:
from universal_ai_sdk import UniversalAIClient, AgentConfig
client = UniversalAIClient()
config = AgentConfig(
instructions="You are a language learning assistant",
capabilities=["text", "voice"],
business_logic_adapter="languagelearning",
custom_settings={
"target_language": "Spanish",
"proficiency_level": "beginner"
}
)
session = client.create_agent(config)
Emergency Services Adapter
Designed for emergency response scenarios with:
- Priority escalation detection based on keywords and urgency
- Location information extraction and validation
- Emergency type classification (medical, fire, police, etc.)
- Call logging and tracking for compliance
- Integration with dispatch systems
Configuration:
config = {
"emergency_types": ["medical", "fire", "police", "natural_disaster"],
"location_required": True,
"escalation_keywords": ["unconscious", "bleeding", "fire", "chest_pain"],
"response_time_critical": True,
"dispatch_integration": {
"api_endpoint": "https://dispatch.emergency.gov/api/v1",
"api_key": "your_dispatch_api_key"
}
}
Example Usage:
config = AgentConfig(
instructions="You are an emergency services dispatcher. Gather critical information quickly.",
capabilities=["text", "voice"],
business_logic_adapter="emergencyservices",
custom_settings={
"emergency_types": ["medical", "fire", "police"],
"location_required": True
}
)
Creating Custom Adapters
Step 1: Define the Adapter Class
from adapters.business_logic_adapter import BusinessLogicAdapter
class HealthcareAdapter(BusinessLogicAdapter):
def __init__(self, custom_settings=None):
super().__init__(custom_settings)
# Healthcare-specific settings
self.hipaa_compliance = custom_settings.get('hipaa_compliance', True)
self.specialty = custom_settings.get('specialty', 'general')
self.patient_privacy_level = custom_settings.get('patient_privacy_level', 'strict')
# Initialize medical knowledge base
self.medical_terms = self._load_medical_terms()
self.drug_interactions = self._load_drug_interactions()
def process_user_input(self, user_input, session_context):
"""Process user input for healthcare context"""
# Remove any PII if HIPAA compliance is enabled
if self.hipaa_compliance:
user_input = self._sanitize_pii(user_input)
# Extract medical symptoms and concerns
medical_context = self._extract_medical_context(user_input)
# Add to session context
session_context['medical_context'] = medical_context
session_context['requires_physician_review'] = self._requires_physician_review(medical_context)
return user_input, session_context
def process_agent_response(self, agent_response, session_context):
"""Process agent response for healthcare safety"""
# Check for medical advice disclaimer
if self._contains_medical_advice(agent_response):
agent_response = self._add_medical_disclaimer(agent_response)
# Check for drug interaction warnings
if 'medications' in session_context.get('medical_context', {}):
agent_response = self._check_drug_interactions(agent_response, session_context)
# Flag for physician review if needed
if session_context.get('requires_physician_review'):
self._flag_for_physician_review(session_context)
return agent_response
def _sanitize_pii(self, text):
"""Remove personally identifiable information"""
# Implement PII detection and removal
# This is a simplified example
import re
# Remove SSN patterns
text = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[REDACTED]', text)
# Remove phone patterns
text = re.sub(r'\b\d{3}-\d{3}-\d{4}\b', '[REDACTED]', text)
return text
def _extract_medical_context(self, text):
"""Extract medical symptoms and context"""
medical_context = {
'symptoms': [],
'medications': [],
'allergies': [],
'urgency_level': 'low'
}
# Use NLP to extract medical entities
# This would integrate with medical NLP libraries
symptoms = self._extract_symptoms(text)
medications = self._extract_medications(text)
medical_context['symptoms'] = symptoms
medical_context['medications'] = medications
# Determine urgency based on symptoms
if any(urgent in text.lower() for urgent in ['chest pain', 'difficulty breathing', 'severe bleeding']):
medical_context['urgency_level'] = 'high'
return medical_context
def _requires_physician_review(self, medical_context):
"""Determine if case needs physician review"""
high_risk_symptoms = ['chest pain', 'stroke symptoms', 'severe allergic reaction']
for symptom in medical_context.get('symptoms', []):
if any(risk in symptom.lower() for risk in high_risk_symptoms):
return True
return medical_context.get('urgency_level') == 'high'
def _add_medical_disclaimer(self, response):
"""Add medical disclaimer to responses"""
disclaimer = "\n\n⚠️ **Medical Disclaimer**: This information is for educational purposes only and is not a substitute for professional medical advice. Please consult with a healthcare provider for medical concerns."
return response + disclaimer
Step 2: Register the Adapter
# In adapters/__init__.py
from .healthcare_adapter import HealthcareAdapter
# Register the adapter
AVAILABLE_ADAPTERS = {
'languagelearning': 'adapters.language_learning_adapter.LanguageLearningAdapter',
'emergencyservices': 'adapters.emergency_services_adapter.EmergencyServicesAdapter',
'healthcare': 'adapters.healthcare_adapter.HealthcareAdapter', # New adapter
}
def get_adapter(adapter_name, custom_settings=None):
"""Get adapter instance by name"""
if adapter_name not in AVAILABLE_ADAPTERS:
raise ValueError(f"Unknown adapter: {adapter_name}")
module_path, class_name = AVAILABLE_ADAPTERS[adapter_name].rsplit('.', 1)
module = __import__(module_path, fromlist=[class_name])
adapter_class = getattr(module, class_name)
return adapter_class(custom_settings)
Step 3: Configure and Use
# Using the new healthcare adapter
from universal_ai_sdk import UniversalAIClient, AgentConfig
client = UniversalAIClient()
healthcare_config = AgentConfig(
instructions="""You are a healthcare assistant. Help patients understand their symptoms and guide them to appropriate care.
Guidelines:
- Always recommend consulting with healthcare providers
- Never provide specific medical diagnoses
- Be empathetic and supportive
- Focus on symptom documentation and care guidance
""",
capabilities=["text", "voice"],
business_logic_adapter="healthcare",
custom_settings={
"hipaa_compliance": True,
"specialty": "primary_care",
"patient_privacy_level": "strict",
"require_physician_review": True
}
)
session = client.create_agent(healthcare_config)
Advanced Adapter Features
External API Integration
class ECommerceAdapter(BusinessLogicAdapter):
def __init__(self, custom_settings=None):
super().__init__(custom_settings)
self.product_api = custom_settings.get('product_api_url')
self.inventory_api = custom_settings.get('inventory_api_url')
self.api_key = custom_settings.get('api_key')
async def process_user_input(self, user_input, session_context):
"""Process e-commerce queries with real-time data"""
# Extract product queries
if self._is_product_query(user_input):
product_info = await self._fetch_product_info(user_input)
session_context['product_context'] = product_info
# Check inventory for mentioned products
if self._mentions_products(user_input):
inventory_status = await self._check_inventory(user_input)
session_context['inventory_status'] = inventory_status
return user_input, session_context
async def _fetch_product_info(self, query):
"""Fetch real-time product information"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.product_api}/search",
params={'q': query},
headers={'Authorization': f'Bearer {self.api_key}'}
) as response:
return await response.json()
State Management
class ConversationStateAdapter(BusinessLogicAdapter):
def __init__(self, custom_settings=None):
super().__init__(custom_settings)
self.conversation_states = {}
self.max_context_turns = custom_settings.get('max_context_turns', 10)
def process_user_input(self, user_input, session_context):
"""Maintain conversation state and context"""
session_id = session_context.get('session_id')
# Initialize state if new session
if session_id not in self.conversation_states:
self.conversation_states[session_id] = {
'turns': [],
'topics': [],
'user_preferences': {},
'conversation_goal': None
}
state = self.conversation_states[session_id]
# Add current turn to history
state['turns'].append({
'type': 'user',
'content': user_input,
'timestamp': time.time()
})
# Limit context size
if len(state['turns']) > self.max_context_turns * 2:
state['turns'] = state['turns'][-self.max_context_turns * 2:]
# Extract topics and preferences
self._update_conversation_context(user_input, state)
# Add state to session context
session_context['conversation_state'] = state
return user_input, session_context
def _update_conversation_context(self, user_input, state):
"""Update conversation topics and user preferences"""
# Extract topics using NLP
topics = self._extract_topics(user_input)
state['topics'].extend(topics)
# Keep only recent topics
state['topics'] = list(set(state['topics'][-20:]))
# Extract preferences
preferences = self._extract_preferences(user_input)
state['user_preferences'].update(preferences)
Multi-Language Support
class MultiLanguageAdapter(BusinessLogicAdapter):
def __init__(self, custom_settings=None):
super().__init__(custom_settings)
self.supported_languages = custom_settings.get('supported_languages', ['en', 'es', 'fr'])
self.auto_detect_language = custom_settings.get('auto_detect_language', True)
self.translation_service = custom_settings.get('translation_service', 'google')
def process_user_input(self, user_input, session_context):
"""Handle multi-language input"""
# Detect language if enabled
if self.auto_detect_language:
detected_language = self._detect_language(user_input)
session_context['detected_language'] = detected_language
# Translate to primary language if needed
if detected_language not in self.supported_languages:
translated_input = self._translate_text(user_input, detected_language, 'en')
session_context['original_input'] = user_input
session_context['translation_applied'] = True
user_input = translated_input
return user_input, session_context
def process_agent_response(self, agent_response, session_context):
"""Translate response back to user's language if needed"""
if session_context.get('translation_applied'):
original_language = session_context.get('detected_language')
if original_language:
agent_response = self._translate_text(agent_response, 'en', original_language)
return agent_response
Adapter Configuration
Environment-Based Configuration
# config/adapter_config.py
import os
class AdapterConfig:
# Language Learning Adapter
LANGUAGE_LEARNING_SETTINGS = {
'supported_languages': os.environ.get('SUPPORTED_LANGUAGES', 'Spanish,French,German').split(','),
'default_proficiency': os.environ.get('DEFAULT_PROFICIENCY', 'beginner'),
'max_session_length': int(os.environ.get('MAX_SESSION_LENGTH', '60')), # minutes
}
# Emergency Services Adapter
EMERGENCY_SERVICES_SETTINGS = {
'dispatch_api_url': os.environ.get('DISPATCH_API_URL'),
'dispatch_api_key': os.environ.get('DISPATCH_API_KEY'),
'location_service_url': os.environ.get('LOCATION_SERVICE_URL'),
'emergency_contacts': os.environ.get('EMERGENCY_CONTACTS', '911,112,999').split(','),
}
# Healthcare Adapter
HEALTHCARE_SETTINGS = {
'hipaa_compliance': os.environ.get('HIPAA_COMPLIANCE', 'true').lower() == 'true',
'medical_api_url': os.environ.get('MEDICAL_API_URL'),
'medical_api_key': os.environ.get('MEDICAL_API_KEY'),
'physician_review_threshold': float(os.environ.get('PHYSICIAN_REVIEW_THRESHOLD', '0.8')),
}
Testing Adapters
Unit Testing
import unittest
from adapters.healthcare_adapter import HealthcareAdapter
class TestHealthcareAdapter(unittest.TestCase):
def setUp(self):
self.adapter = HealthcareAdapter({
'hipaa_compliance': True,
'specialty': 'general'
})
def test_pii_sanitization(self):
"""Test PII removal"""
input_text = "My SSN is 123-45-6789 and phone is 555-123-4567"
sanitized = self.adapter._sanitize_pii(input_text)
self.assertNotIn('123-45-6789', sanitized)
self.assertNotIn('555-123-4567', sanitized)
self.assertIn('[REDACTED]', sanitized)
def test_medical_context_extraction(self):
"""Test medical context extraction"""
input_text = "I have chest pain and difficulty breathing"
context = self.adapter._extract_medical_context(input_text)
self.assertEqual(context['urgency_level'], 'high')
self.assertIn('chest pain', str(context['symptoms']))
def test_physician_review_requirement(self):
"""Test physician review detection"""
high_risk_context = {
'symptoms': ['chest pain'],
'urgency_level': 'high'
}
requires_review = self.adapter._requires_physician_review(high_risk_context)
self.assertTrue(requires_review)
if __name__ == '__main__':
unittest.main()
Integration Testing
import pytest
from universal_ai_sdk import UniversalAIClient, AgentConfig
@pytest.fixture
def healthcare_client():
client = UniversalAIClient('http://localhost:8000')
config = AgentConfig(
instructions="You are a healthcare assistant",
capabilities=["text"],
business_logic_adapter="healthcare",
custom_settings={
'hipaa_compliance': True,
'specialty': 'general'
}
)
session = client.create_agent(config)
yield session
session.close()
def test_healthcare_conversation(healthcare_client):
"""Test healthcare adapter in conversation"""
# Send medical query
healthcare_client.send_message("I have a headache and fever")
# Get response
messages = healthcare_client.get_messages()
last_response = messages[-1]['content']
# Check for medical disclaimer
assert "Medical Disclaimer" in last_response or "consult" in last_response.lower()
Best Practices
Design Principles
- Single Responsibility: Each adapter should handle one domain
- Composability: Adapters should work together when needed
- Configuration: Make adapters highly configurable
- Error Handling: Graceful fallbacks for adapter failures
- Performance: Minimize latency in input/output processing
Security Considerations
- Input Validation: Always validate and sanitize user input
- Output Filtering: Review and filter agent responses
- Data Privacy: Respect privacy regulations (GDPR, HIPAA, etc.)
- Access Control: Implement proper authorization
- Audit Logging: Log all adapter decisions and actions
Performance Optimization
- Caching: Cache expensive operations (API calls, NLP processing)
- Async Operations: Use async/await for I/O operations
- Resource Management: Properly manage connections and resources
- Monitoring: Track adapter performance and errors
Next: Learn about Deployment or explore the SDK Documentation for implementation details.