The Amigo platform maintains user models that evolve based on interactions. These models enable personalized experiences across sessions by maintaining rich, structured knowledge about each user's medical history and needs.
Understanding User Model Structure
User models consist of multiple content entries grouped into dimensions, each describing a specific aspect of the user. Each entry contains:
Content: The actual information about the user in natural language
Insight IDs: References to the evidence supporting this information
Dimensions: Categorization with descriptive tags for context
Here's an example of a user model structure:
{
"user_models": [
{
"content": "User's active nickname is 'Tony' (derived from legal name Anthony Lee). Historical records show consistent use of 'Tony' across sessions with no permanent changes.",
"insight_ids": [
{ "$oid": "67fe9cbc0e3bf2dbf54e4ef1" },
{ "$oid": "67fae2ace70871557820d4cd" }
// Additional insight IDs...
],
"dimensions": [
{
"description": "Nickname preference: The client's chosen name for all interactions, which may differ from their legal name.",
"tags": ["preference", "active name", "name history", "name", "current name", "nickname"]
}
]
},
{
"content": "Tony has hypertension and Type 2 diabetes which are being managed with medication. He reports consistent adherence to his prescription regimen except during business travel when he occasionally misses the evening dose.",
"insight_ids": [
{ "$oid": "67fe9cca0e3bf2dbf54e4ef5" },
{ "$oid": "67fe9cca0e3bf2dbf54e4ef8" }
// Additional insight IDs...
],
"dimensions": [
{
"description": "Medical History & Conditions: Current and past diagnoses, chronic conditions, medication adherence, and treatment history.",
"tags": ["medical history", "conditions", "chronic disease", "medication"]
}
]
}
// Additional model entries for other dimensions...
]
}
Key Dimensions in User Models
User models typically contain information across multiple dimensions including:
Demographics & Background: Age, gender, family structure, medical insurance
Medical History: Diagnoses, surgeries, hospitalizations, allergies
Medication Management: Current medications, adherence, side effects
Vital Signs: Blood pressure, heart rate, respiratory rate, temperature trends
Emotional Well-being: Mental health status, stress levels, coping mechanisms
Behavioral Patterns: Sleep patterns, substance use, adherence to medical advice
Treatment Goals: Health objectives, recovery milestones, disease management
Social Determinants: Support networks, living situation, access to care
Communication Preferences: Preferred explanations, health literacy level
Health Monitoring: Self-monitoring practices, device usage, tracking methods
Updating User Context Information
Quick‑Start (TL;DR)
Collect or compute the facts you want Tony's assistant to remember.
Turn each fact into a single, self‑contained sentence.
POST them in the additional_context array.
import requests, os, json
ORG_ID = os.environ["AMIGO_ORG_ID"]
API_KEY = os.environ["AMIGO_API_KEY"] # service‑account JWT
USER_ID = "user_12345"
facts = [
"Tony's average fasting glucose over the past week is 105 mg/dL.",
"He reports mild dizziness after his evening dose of Lisinopril 20 mg.",
"Tony exercises three times a week (mostly swimming)."
]
resp = requests.post(
f"https://api.amigo.ai/v1/{ORG_ID}/user/{USER_ID}/user",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={"additional_context": facts},
timeout=15,
)
resp.raise_for_status() # 204 No Content on success
print("✓ user model updated")
The code above fits in a single file, has no third‑party dependencies beyond requests, and works out‑of‑the‑box once you set AMIGO_ORG_ID and AMIGO_API_KEY.
To add or update contextual information for a user:
Endpoint quick facts
Property
Value
Path
/v1/{organization}/user/{requested_user_id}/user
Method
POST
Auth
Bearer token (admin‑scope service account recommended)
Rate limit
100 requests per minute (see X‑RateLimit‑Remaining, X‑RateLimit‑Reset headers)
Payload schema
{ "additional_context": string[] }
Idempotency
Strings are appended verbatim – sending the same string twice will duplicate the observation.
Handling 429 Too Many Requests
If you exceed the org‑wide limit (100 req/min) the server replies with 429 and a Retry‑After header. A lightweight exponential‑back‑off helper looks like this:
import requests, time
def post_with_backoff(url: str, headers: dict, json: dict, max_retries: int = 5):
delay = 1 # seconds
for attempt in range(1, max_retries + 1):
resp = requests.post(url, headers=headers, json=json)
if resp.status_code != 429:
resp.raise_for_status()
return resp
retry_after = int(resp.headers.get("Retry-After", delay))
print(f"Rate‑limited (attempt {attempt}/{max_retries}) → waiting {retry_after}s…")
time.sleep(retry_after)
delay = min(delay * 2, 60) # cap at 60 s
raise RuntimeError("Too many consecutive 429 responses – aborting")
Drop‑in replacement: call post_with_backoff() instead of requests.post() when sending additional_context updates.
curl --request POST \
--url 'https://api.amigo.ai/v1/<your-org-id>/user/<USER_ID>/user' \
--header 'Authorization: Bearer <AUTH-TOKEN-OF-USER>' \
--header 'content-type: application/json' \
--data '
{
"additional_context": [
"The user maintains a moderately active lifestyle despite chronic joint pain, completing physical therapy exercises 3–4 times weekly.",
"Latest vitals: BP 138/85 mmHg, HR 72 bpm, temperature 98.2 °F (measured 2023‑12‑15), fasting glucose 110 mg/dL (down from 130 mg/dL three months ago).",
"The user is currently taking Metformin 1000 mg BID and Lisinopril 20 mg daily (previously on Amlodipine) and reports mild GI discomfort with the Metformin.",
"Recent cardiac stress test completed on 2023‑11‑28 showed no significant abnormalities."
]
}'
Formatting Additional Context Effectively
Quick reality checkadditional_context only accepts an array of strings. The Amigo user‑modeling engine works in natural language space, not relational columns, so the caller is responsible for turning structured rows into concise sentences or paragraphs. (The agent can parse nuanced prose; it cannot see inside your proprietary field names.)
The additional_context field expects an array of strings that will be appended to the user model. For optimal integration, your entries should:
Match the style and tone of existing model entries
Contain complete, coherent information in natural language
Focus on a single dimension or related aspects per entry
Include specific metrics with units where applicable
Provide temporal context when relevant (dates, duration, frequency)
🚨 Privacy tip Only send data you are comfortable storing indefinitely. If a field is no longer needed—or could identify the user (e.g. exact DOB)—drop it or generalise before ingestion.
• Easy to generate (stringify each row).\
• The prefix clarifies the source for later auditing.\
• Works today, no training required.
2. Dimension‑aware sentences (recommended as you mature)
Group related metrics into the same dimension and add context:
> "As of 23 April 2024 Tony's average fasting glucose is 105 mg/dL, down 20 mg/dL (‑16%) from three months ago."
The agent then has _one coherent thought_ rather than 4 disjointed keys.
Whichever path you choose, keep transformation logic inside your stack so that schema tweaks don't require Amigo releases. See the format_app_data_for_user_model() helper in the example below for a head‑start.
Example: Turning a BigQuery Row into additional_context
def bq_row_to_context(row: dict) -> list[str]:
"""Minimal transformation of the analytics team's JSON schema."""
descriptions = {
"country_code": "The country the patient is based in (US = United States, UK = United Kingdom)",
"avg_daily_fasting_glucose_mg_dL": "Average daily fasting glucose in mg/dL",
# add more field docs as you wish…
}
entries: list[str] = []
for field, value in row.items():
if value is None or value == "":
continue # skip nulls
desc = descriptions.get(field, field)
entries.append(f"[HealthTrack Analytics] {desc}: {value}")
return entries
# Usage
payload = {"additional_context": bq_row_to_context(bq_row)}
requests.post(url, headers=headers, json=payload)
Start simple; you can always replace the one‑liner with richer prose later without touching the Amigo side of the integration.
Advanced: Automated Batch Sync
Below is a complete reference implementation (≈300 LOC) that demonstrates:
• Pulling rows from your data warehouse.
• Translating them into dimension‑aware sentences.
• Posting in batches of 100 users with exponential back‑off when you hit the 429 rate limit.
Feel free to copy‑paste and trim – the only hard requirement from Amigo's side is that you eventually call the /user endpoint with writable Bearer credentials.
For enterprise integrations with existing application data, you'll need to transform structured data into natural language entries that align with the user model's format and dimensions.
Here's a comprehensive example:
import requests
import json
from datetime import datetime
from typing import Dict, Any, List, Optional
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("user_model_integration")
class UserModelIntegration:
def __init__(self, base_url: str, org_id: str, api_key: str):
"""Initialize with API connection details."""
self.base_url = base_url
self.org_id = org_id
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
self.rate_limit_remaining = 100 # Default assumption
self.rate_limit_reset = 0
def get_user_model(self, user_id: str) -> Dict:
"""Retrieve the current user model to compare before updates."""
url = f"{self.base_url}/v1/{self.org_id}/user/{user_id}/user_model"
try:
response = self._make_api_request("GET", url)
return response
except Exception as e:
logger.error(f"Failed to retrieve user model: {e}")
return {"user_models": []}
def update_user_model(self, user_id: str, app_data: Dict[str, Any],
change_threshold: float = 0.3) -> Dict:
"""
Update a user model with formatted app data added to additional_context.
Applies change detection to avoid unnecessary updates.
Args:
user_id: The user's unique identifier
app_data: Structured data from your application
change_threshold: Minimum similarity difference to trigger update
Returns:
API response or error information
"""
# Generate new context entries from app data
context_entries = self.format_app_data_for_user_model(app_data)
if not context_entries:
return {"status": "error", "message": "No valid context entries generated"}
# Optionally check if the update is significant enough
if self._should_update_model(user_id, context_entries, change_threshold):
update_data = {"additional_context": context_entries}
url = f"{self.base_url}/v1/{self.org_id}/user/{user_id}/user"
try:
return self._make_api_request("POST", url, update_data)
except requests.exceptions.RequestException as e:
return self._handle_request_error(e)
else:
return {"status": "skipped", "message": "No significant changes detected"}
def _make_api_request(self, method: str, url: str, data: Optional[Dict] = None):
"""Make API request with rate limiting handling."""
# Check if we need to wait for rate limit reset
if self.rate_limit_remaining <= 5:
wait_time = max(0, self.rate_limit_reset - time.time())
if wait_time > 0:
logger.info(f"Rate limit approaching, waiting {wait_time:.1f}s")
time.sleep(wait_time + 1) # Add 1s buffer
# Make the request
response = requests.request(
method=method,
url=url,
headers=self.headers,
json=data
)
# Update rate limit information
self.rate_limit_remaining = int(response.headers.get("X-RateLimit-Remaining", 100))
self.rate_limit_reset = int(response.headers.get("X-RateLimit-Reset", 0))
# Handle response
response.raise_for_status()
return response.json()
def _handle_request_error(self, error: Exception) -> Dict:
"""Process request exceptions into meaningful error responses."""
if hasattr(error, 'response') and error.response is not None:
status_code = error.response.status_code
try:
error_detail = error.response.json()
except:
error_detail = {"raw": error.response.text}
if status_code == 429: # Too Many Requests
retry_after = int(error.response.headers.get("Retry-After", 60))
logger.warning(f"Rate limit exceeded, retry after {retry_after}s")
return {
"status": "rate_limited",
"message": f"Rate limit exceeded, retry after {retry_after}s",
"retry_after": retry_after
}
elif status_code >= 500:
return {"status": "server_error", "message": "Server error, retry later", "details": error_detail}
else:
return {"status": "request_error", "message": str(error), "details": error_detail}
return {"status": "error", "message": str(error)}
def _should_update_model(self, user_id: str, new_entries: List[str],
threshold: float) -> bool:
"""
Determine if the new context entries are different enough from
existing model to warrant an update.
"""
# This is a simplified implementation - in production you might use
# more sophisticated text similarity metrics
try:
current_model = self.get_user_model(user_id)
if not current_model.get("user_models"):
return True # No existing model, definitely update
# Extract current content for comparison
current_content = " ".join([
entry.get("content", "")
for entry in current_model.get("user_models", [])
]).lower()
# Check for significant new information
new_info_count = 0
for entry in new_entries:
# Consider entry significant if less than 70% of its content
# is found in the current model
entry_words = set(entry.lower().split())
model_words = set(current_content.split())
overlap = len(entry_words.intersection(model_words))
if overlap / max(1, len(entry_words)) < (1 - threshold):
new_info_count += 1
# Update if we have any significant new information
return new_info_count > 0
except Exception as e:
logger.warning(f"Error in change detection, defaulting to update: {e}")
return True # Default to updating on errors
def format_app_data_for_user_model(self, app_data: Dict[str, Any]) -> List[str]:
"""
Convert structured app data into formatted strings suitable for the user model.
Each string follows a dimensional pattern similar to existing model entries.
"""
context_entries = []
# 1. Demographics & Background
if self._has_demographic_data(app_data):
demo_entry = self._format_demographics(app_data)
if demo_entry:
context_entries.append(demo_entry)
# 2. Vital Signs & Measurements
if self._has_vitals_data(app_data):
vitals_entry = self._format_vitals(app_data)
if vitals_entry:
context_entries.append(vitals_entry)
# 3. Medical Conditions & History
if self._has_medical_data(app_data):
medical_entry = self._format_medical_info(app_data)
if medical_entry:
context_entries.append(medical_entry)
# 4. Medication Management
if self._has_medication_data(app_data):
medication_entry = self._format_medication_info(app_data)
if medication_entry:
context_entries.append(medication_entry)
# 5. Treatment Goals & Progress
if self._has_goal_data(app_data):
goal_entry = self._format_goals_info(app_data)
if goal_entry:
context_entries.append(goal_entry)
# Add additional specialized formatters for other dimensions
return context_entries
def _has_demographic_data(self, data: Dict) -> bool:
"""Check if the data contains demographic information."""
demo_fields = ['age', 'sex', 'ethnicity', 'pregnant', 'first_name']
return any(field in data for field in demo_fields)
def _format_demographics(self, data: Dict) -> str:
"""Format demographic data into a natural language statement."""
name = data.get('first_name', 'Patient')
parts = []
# Age information
if 'age' in data:
parts.append(f"{name} is {data['age']} years old")
# Gender/sex information
if 'sex' in data:
parts.append(f"identifies as {data['sex']}")
# Ethnicity information
if 'ethnicity' in data:
parts.append(f"of {data['ethnicity']} background")
# Pregnancy status
if data.get('pregnant') == 'Yes':
parts.append("is currently pregnant")
# Social determinants
if 'living_situation' in data:
parts.append(f"currently {data['living_situation']}")
if parts:
return f"{name} {' and '.join(parts)}."
return ""
def _has_vitals_data(self, data: Dict) -> bool:
"""Check if the data contains vital signs measurements."""
vitals_fields = [
'systolic_bp', 'diastolic_bp', 'heart_rate', 'respiratory_rate',
'temperature', 'oxygen_saturation', 'weight_kg', 'height'
]
return any(field in data for field in vitals_fields)
def _format_vitals(self, data: Dict) -> str:
"""Format vital signs data into a natural language statement."""
name = data.get('first_name', 'Patient')
parts = []
# Blood pressure
if 'systolic_bp' in data and 'diastolic_bp' in data:
bp_part = f"blood pressure of {data['systolic_bp']}/{data['diastolic_bp']} mmHg"
if 'bp_status' in data:
bp_part += f" ({data['bp_status']})"
parts.append(bp_part)
# Heart rate
if 'heart_rate' in data:
hr_part = f"heart rate of {data['heart_rate']} bpm"
if 'hr_status' in data:
hr_part += f" ({data['hr_status']})"
parts.append(hr_part)
# Respiratory rate
if 'respiratory_rate' in data:
parts.append(f"respiratory rate of {data['respiratory_rate']} breaths/min")
# Temperature
if 'temperature' in data:
temp_unit = data.get('temp_unit', '°F')
parts.append(f"temperature of {data['temperature']} {temp_unit}")
# Oxygen saturation
if 'oxygen_saturation' in data:
parts.append(f"oxygen saturation of {data['oxygen_saturation']}%")
# Weight with historical context if available
if 'weight_kg' in data:
weight_part = f"weight of {data['weight_kg']} kg"
if 'initial_weight_kg' in data and 'weight_change_kg' in data:
direction = "decrease" if float(data['weight_change_kg']) > 0 else "increase"
amount = abs(float(data['weight_change_kg']))
time_period = data.get('weight_change_period', 'since last measurement')
weight_part += f" ({amount} kg {direction} {time_period})"
parts.append(weight_part)
# Height
if 'height' in data:
parts.append(f"height of {data['height']} cm")
if parts:
measurement_date = data.get('measurement_date', datetime.now().strftime('%Y-%m-%d'))
return f"As of {measurement_date}, {name}'s vital signs show {', '.join(parts)}."
return ""
def _has_medical_data(self, data: Dict) -> bool:
"""Check if the data contains medical condition information."""
medical_fields = [
'conditions', 'chronic_diseases', 'allergies', 'surgical_history',
'family_history', 'lab_results'
]
return any(field in data for field in medical_fields)
def _format_medical_info(self, data: Dict) -> str:
"""Format medical data into a natural language statement."""
name = data.get('first_name', 'Patient')
parts = []
# Chronic conditions
if 'conditions' in data or 'chronic_diseases' in data:
conditions = []
if 'conditions' in data and data['conditions']:
if isinstance(data['conditions'], list):
conditions.extend(data['conditions'])
else:
conditions.append(data['conditions'])
if 'chronic_diseases' in data and data['chronic_diseases']:
if isinstance(data['chronic_diseases'], list):
conditions.extend(data['chronic_diseases'])
else:
conditions.append(data['chronic_diseases'])
if conditions:
parts.append(f"diagnosed with {', '.join(conditions)}")
# Allergies
if 'allergies' in data and data['allergies']:
allergies = data['allergies'] if isinstance(data['allergies'], list) else [data['allergies']]
if allergies:
parts.append(f"allergic to {', '.join(allergies)}")
# Surgical history
if 'surgical_history' in data and data['surgical_history']:
surgeries = data['surgical_history'] if isinstance(data['surgical_history'], list) else [data['surgical_history']]
if surgeries:
parts.append(f"surgical history includes {', '.join(surgeries)}")
# Lab results
if 'lab_results' in data and data['lab_results']:
lab_parts = []
for lab in data['lab_results']:
if isinstance(lab, dict) and 'name' in lab and 'value' in lab:
lab_parts.append(f"{lab['name']}: {lab['value']}{lab.get('unit', '')}")
if lab_parts:
parts.append(f"recent lab results show {', '.join(lab_parts)}")
if parts:
return f"{name} is {' and '.join(parts)}."
return ""
def _has_medication_data(self, data: Dict) -> bool:
"""Check if the data contains medication information."""
med_fields = [
'current_medications', 'previous_medications',
'medication_adherence', 'medication_side_effects'
]
return any(field in data for field in med_fields)
def _format_medication_info(self, data: Dict) -> str:
"""Format medication data into a natural language statement."""
name = data.get('first_name', 'Patient')
parts = []
# Current medications
if 'current_medications' in data and data['current_medications']:
meds = []
medications = data['current_medications']
if isinstance(medications, list):
for med in medications:
if isinstance(med, dict):
med_str = f"{med.get('name', '')}"
if 'dosage' in med:
med_str += f" {med['dosage']}"
if 'frequency' in med:
med_str += f" {med['frequency']}"
meds.append(med_str)
else:
meds.append(str(med))
else:
meds.append(str(medications))
if meds:
parts.append(f"currently taking {', '.join(meds)}")
# Previous medications
if 'previous_medications' in data and data['previous_medications']:
prev_meds = []
medications = data['previous_medications']
if isinstance(medications, list):
for med in medications:
if isinstance(med, dict):
med_str = f"{med.get('name', '')}"
if 'reason_stopped' in med:
med_str += f" (stopped due to {med['reason_stopped']})"
prev_meds.append(med_str)
else:
prev_meds.append(str(med))
else:
prev_meds.append(str(medications))
if prev_meds:
parts.append(f"previously on {', '.join(prev_meds)}")
# Medication adherence
if 'medication_adherence' in data:
adherence = data['medication_adherence']
if isinstance(adherence, dict) and 'rate' in adherence:
parts.append(f"reports {adherence['rate']}% medication adherence")
if 'issues' in adherence and adherence['issues']:
parts.append(f"with adherence challenges related to {adherence['issues']}")
elif isinstance(adherence, str):
parts.append(f"reports {adherence} medication adherence")
# Medication side effects
if 'medication_side_effects' in data and data['medication_side_effects']:
side_effects = data['medication_side_effects']
if isinstance(side_effects, list):
parts.append(f"experiencing side effects including {', '.join(side_effects)}")
else:
parts.append(f"experiencing side effects including {side_effects}")
if parts:
return f"{name} is {' and '.join(parts)}."
return ""
def _has_goal_data(self, data: Dict) -> bool:
"""Check if the data contains treatment goal information."""
goal_fields = [
'treatment_goals', 'health_objectives', 'progress_metrics',
'treatment_adherence', 'recovery_timeline'
]
return any(field in data for field in goal_fields)
def _format_goals_info(self, data: Dict) -> str:
"""Format treatment goals data into a natural language statement."""
name = data.get('first_name', 'Patient')
parts = []
# Treatment goals
if 'treatment_goals' in data and data['treatment_goals']:
goals = data['treatment_goals']
if isinstance(goals, list):
parts.append(f"working toward {', '.join(goals)}")
else:
parts.append(f"working toward {goals}")
# Health objectives
if 'health_objectives' in data and data['health_objectives']:
objectives = data['health_objectives']
if isinstance(objectives, dict):
obj_parts = []
for key, value in objectives.items():
obj_parts.append(f"{key}: {value}")
if obj_parts:
parts.append(f"has health objectives including {', '.join(obj_parts)}")
elif isinstance(objectives, list):
parts.append(f"has health objectives including {', '.join(objectives)}")
else:
parts.append(f"has health objective of {objectives}")
# Progress metrics
if 'progress_metrics' in data and data['progress_metrics']:
metrics = data['progress_metrics']
if isinstance(metrics, dict):
metric_parts = []
for key, value in metrics.items():
metric_parts.append(f"{key} at {value}")
if metric_parts:
parts.append(f"tracking progress with {', '.join(metric_parts)}")
elif isinstance(metrics, list):
parts.append(f"tracking progress with {', '.join(metrics)}")
else:
parts.append(f"tracking progress with {metrics}")
# Treatment adherence
if 'treatment_adherence' in data:
adherence = data['treatment_adherence']
if isinstance(adherence, dict) and 'level' in adherence:
parts.append(f"shows {adherence['level']} adherence to treatment plan")
if 'challenges' in adherence and adherence['challenges']:
parts.append(f"with challenges in {adherence['challenges']}")
elif isinstance(adherence, str):
parts.append(f"shows {adherence} adherence to treatment plan")
if parts:
return f"{name} is {' and '.join(parts)}."
return ""
# Example usage
def main():
# Configuration - would be loaded from environment variables in production
base_url = "https://api.amigo.ai"
org_id = "your-org-id"
api_key = "your-api-key"
# Initialize integration
integration = UserModelIntegration(base_url, org_id, api_key)
# Example app data for Tony (mock medical data)
app_data = {
"user_id": "user_12345",
"first_name": "Tony",
"age": 42,
"sex": "male",
"ethnicity": "Asian",
"living_situation": "lives with spouse and two children",
"systolic_bp": 138,
"diastolic_bp": 85,
"bp_status": "Stage 1 Hypertension",
"heart_rate": 72,
"respiratory_rate": 16,
"temperature": 98.2,
"temp_unit": "°F",
"oxygen_saturation": 97,
"weight_kg": 92.5,
"initial_weight_kg": 98.0,
"weight_change_kg": 5.5,
"weight_change_period": "over past 6 months",
"height": 180,
"measurement_date": "2023-12-15",
"conditions": ["Type 2 Diabetes", "Hypertension", "Osteoarthritis"],
"allergies": ["Penicillin", "Sulfa drugs"],
"surgical_history": ["Appendectomy (2010)", "Right knee arthroscopy (2019)"],
"lab_results": [
{"name": "HbA1c", "value": 7.2, "unit": "%"},
{"name": "LDL", "value": 110, "unit": "mg/dL"},
{"name": "eGFR", "value": 75, "unit": "mL/min"}
],
"current_medications": [
{"name": "Metformin", "dosage": "1000mg", "frequency": "BID"},
{"name": "Lisinopril", "dosage": "20mg", "frequency": "daily"},
{"name": "Atorvastatin", "dosage": "40mg", "frequency": "daily"}
],
"previous_medications": [
{"name": "Amlodipine", "reason_stopped": "ankle edema"}
],
"medication_adherence": {"rate": 85, "issues": "forgets evening dose during travel"},
"medication_side_effects": "mild GI discomfort with Metformin",
"treatment_goals": ["HbA1c < 6.5%", "BP < 130/80 mmHg", "improved mobility"],
"health_objectives": {
"daily walking": "30 minutes",
"carbohydrate intake": "< 150g per day"
},
"progress_metrics": {
"weekly exercise sessions": "4/5 completed",
"medication adherence": "85%"
},
"treatment_adherence": {"level": "good", "challenges": "exercise consistency"}
}
# Update user model
result = integration.update_user_model(app_data["user_id"], app_data)
print(f"API Response: {json.dumps(result, indent=2)}")
# Show what was sent to the API
formatted_entries = integration.format_app_data_for_user_model(app_data)
print("\nFormatted entries sent to API:")
for entry in formatted_entries:
print(f"- {entry}")
if __name__ == "__main__":
main()
Integration Best Practices for Complex User Models
When integrating your application data with sophisticated user models:
1. Data Processing Strategy
Dimension Mapping: Map your app data fields to corresponding medical model dimensions
Contextual Grouping: Combine related fields into coherent natural language statements
Content Coherence: Ensure each entry forms a complete thought with proper grammar
Style Consistency: Match the language style and tone of existing model entries
Temporal Context: Always include time references for measurements and events
Unit Standardization: Use consistent medical units and include them with all measurements
2. Technical Implementation
Change Detection: Implement similarity checks to avoid redundant updates
Rate Limiting: Respect API rate limits with appropriate handling and backoff
Error Resilience: Implement comprehensive error handling and recovery
Logging: Maintain detailed logs of all integration activity for debugging
Idempotency: Design your integration to be safely repeatable
Versioning: Track versions of your transformation logic to maintain consistency
3. Data Synchronization Patterns
Real-time Updates: Trigger updates when critical health data changes
Batch Processing: Schedule regular updates for non-critical data
Webhook Integration: Implement webhooks to receive model update notifications
Optimistic Updates: Update your local cache immediately, confirm with server later
Conflict Resolution: Implement strategies for handling conflicting medical data
4. Privacy and Security
Data Minimization: Only include necessary information in user models
Content Sanitization: Remove potentially harmful content before submission
PHI Handling: Follow HIPAA and other regulations for protected health information
Access Control: Implement proper authentication and authorization
Audit Trails: Maintain records of all data updates for compliance
5. Performance Optimization
Caching: Cache user models to reduce API calls
Batching: Group multiple updates when possible
Compression: Minimize payload sizes for efficiency
Asynchronous Processing: Use background tasks for non-critical updates
This endpoint returns the complete user model containing structured information across all dimensions.
Processing User Model Data
When retrieving user models for use in your application:
def extract_user_insights(user_model):
"""Extract key insights from a user model by dimension"""
insights = {}
for entry in user_model.get("user_models", []):
content = entry.get("content", "")
# Extract dimension information
for dimension in entry.get("dimensions", []):
dim_description = dimension.get("description", "")
dim_tags = dimension.get("tags", [])
# Identify dimension category
category = None
if any(tag in dim_tags for tag in ["demographics", "identity", "background"]):
category = "demographics"
elif any(tag in dim_tags for tag in ["medical history", "health", "clinical"]):
category = "medical"
elif any(tag in dim_tags for tag in ["medication", "prescription", "treatment"]):
category = "medications"
# Add other dimension categories...
if category:
if category not in insights:
insights[category] = []
insights[category].append({
"content": content,
"description": dim_description,
"tags": dim_tags
})
return insights
Best Practices for Data Management
Dimensional Thinking: Design your integration to work with the multi-dimensional nature of medical user models
Natural Language Focus: Format data as complete, grammatical statements rather than key-value pairs
Incremental Updates: Prefer adding new context rather than replacing existing information
Content Quality: Prioritize accuracy and relevance over quantity of information
Temporal Context: Always include time references for when medical information was obtained
Regular Synchronization: Periodically retrieve and sync user models with your systems
Contextual Updates: Update user context when you have new, relevant medical information
Privacy Compliance: Ensure all data operations comply with HIPAA and other privacy requirements
Intelligent Filtering: Use available filters to retrieve precisely the data you need
Efficient Caching: Cache appropriate conversation and user data to reduce API calls
Data Validation: Verify input data quality before updating user models
Monitoring: Track successful updates and content quality metrics
Change Detection: Only send updates when app data has meaningfully changed
For message filtering options and details, see the .
