LogoLogo
Go to website
  • Welcome
  • Getting Started
    • Amigo Overview
      • System Components
      • Overcoming LLM Limitations
      • [Advanced] Future-Ready Architecture
      • [Advanced] The Accelerating AI Landscape
    • The Journey with Amigo
      • Partnership Model
  • Concepts
    • Agent Core
      • Core Persona
      • Global Directives
    • Context Graphs
      • State-Based Architecture
      • [Advanced] Field Implementation Guidance
    • Functional Memory
      • Layered Architecture
      • User Model
      • [Advanced] Recall Mechanisms
      • [Advanced] Analytical Capabilities
    • Dynamic Behaviors
      • Side-Effect Architecture
      • Knowledge
      • [Advanced] Behavior Chaining
    • Evaluations
      • [Advanced] Arena Implementation Guide
    • [Advanced] Reinforcement Learning
    • Safety
  • Glossary
  • Advanced Topics
    • Transition to Neuralese Systems
    • Agent V2 Architecture
  • Agent Building Best Practices
    • [Advanced] Dynamic Behaviors Guide
  • Developer Guide
    • Enterprise Integration Guide
      • Authentication
      • User Creation + Management
      • Service Discovery + Management
      • Conversation Creation + Management
      • Data Retrieval + User Model Management
      • Webhook Management
    • API Reference
      • V1/organization
      • V1/conversation
      • V1/service
      • V1/user
      • V1/role
      • V1/admin
      • V1/webhook_destination
      • V1/dynamic_behavior_set
      • V1/metric
      • V1/simulation
      • Models
      • V1/organization
      • V1/service
      • V1/user
      • V1/role
      • V1/conversation
      • V1/admin
      • V1/webhook_destination
      • V1/dynamic_behavior_set
      • V1/metric
      • V1/simulation
      • Models
Powered by GitBook
LogoLogo

Resources

  • Pricing
  • About Us

Company

  • Careers

Policies

  • Terms of Service

Amigo Inc. ©2025 All Rights Reserved.


On this page
  • Retrieving Conversation History
  • Retrieving Conversation Messages
  • User Model Management
  • Understanding User Model Structure
  • Key Dimensions in User Models
  • Updating User Context Information
  • Advanced: Automated Batch Sync
  • Integration Best Practices for Complex User Models
  • Retrieving User Models
  • Processing User Model Data
  • Best Practices for Data Management

Was this helpful?

Export as PDF
  1. Developer Guide
  2. Enterprise Integration Guide

Data Retrieval + User Model Management

PreviousConversation Creation + ManagementNextWebhook Management

Last updated 20 days ago

Was this helpful?

Retrieving Conversation History

You can list and filter conversations for a user with various parameters:

curl --request GET \
     --url 'https://api.amigo.ai/v1/<your-org-id>/conversation/?user_id=<USER_ID>&limit=10&continuation_token=0' \
     --header 'Authorization: Bearer <AUTH-TOKEN-OF-USER>' \
     --header 'accept: application/json'

Available Filters

  • user_id: Retrieve conversations for a specific user

  • service_id: Filter by service type

  • limit and continuation_token: Control pagination

  • is_finished: Filter by conversation state (true/false)

For detailed filter options, refer to the .

Retrieving Conversation Messages

To fetch the messages from a specific conversation:

curl --request GET \
     --url 'https://api.amigo.ai/v1/<your-org-id>/conversation/<CONVERSATION_ID>/messages/' \
     --header 'Authorization: Bearer <AUTH-TOKEN-OF-USER>' \
     --header 'accept: application/json'

User Model Management

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:

  1. Demographics & Background: Age, gender, family structure, medical insurance

  2. Medical History: Diagnoses, surgeries, hospitalizations, allergies

  3. Medication Management: Current medications, adherence, side effects

  4. Vital Signs: Blood pressure, heart rate, respiratory rate, temperature trends

  5. Emotional Well-being: Mental health status, stress levels, coping mechanisms

  6. Behavioral Patterns: Sleep patterns, substance use, adherence to medical advice

  7. Treatment Goals: Health objectives, recovery milestones, disease management

  8. Social Determinants: Support networks, living situation, access to care

  9. Communication Preferences: Preferred explanations, health literacy level

  10. Health Monitoring: Self-monitoring practices, device usage, tracking methods

Updating User Context Information

Quick‑Start (TL;DR)

  1. Collect or compute the facts you want Tony's assistant to remember.

  2. Turn each fact into a single, self‑contained sentence.

  3. 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 check additional_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:

  1. Match the style and tone of existing model entries

  2. Contain complete, coherent information in natural language

  3. Focus on a single dimension or related aspects per entry

  4. Include specific metrics with units where applicable

  5. 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.

What if my analytics team only has a table?

You have two pragmatic options:

  1. Key–value sentence (minimum viable)

```jsonc
{
  "additional_context": [
    "[HealthTrack] country_code = US (United States)",
    "[HealthTrack] approved_for_treatment = true",
    "[HealthTrack] current_program_name = Hypertension Care Plan",
    "[HealthTrack] avg_daily_fasting_glucose_mg_dL = 108 (timestamp 2024‑04‑23)"
  ]
}
```
• 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

  • Connection Pooling: Maintain efficient HTTP connections

Retrieving User Models

To get the latest user model for a specific user:

curl --request GET \
     --url 'https://api.amigo.ai/v1/<your-org-id>/user/<USER_ID>/user_model' \
     --header 'accept: application/json' \
     --header 'Authorization: Bearer <AUTH-TOKEN-OF-USER>'

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

  1. Dimensional Thinking: Design your integration to work with the multi-dimensional nature of medical user models

  2. Natural Language Focus: Format data as complete, grammatical statements rather than key-value pairs

  3. Incremental Updates: Prefer adding new context rather than replacing existing information

  4. Content Quality: Prioritize accuracy and relevance over quantity of information

  5. Temporal Context: Always include time references for when medical information was obtained

  6. Regular Synchronization: Periodically retrieve and sync user models with your systems

  7. Contextual Updates: Update user context when you have new, relevant medical information

  8. Privacy Compliance: Ensure all data operations comply with HIPAA and other privacy requirements

  9. Intelligent Filtering: Use available filters to retrieve precisely the data you need

  10. Efficient Caching: Cache appropriate conversation and user data to reduce API calls

  11. Data Validation: Verify input data quality before updating user models

  12. Monitoring: Track successful updates and content quality metrics

  13. Change Detection: Only send updates when app data has meaningfully changed

For message filtering options and details, see the .

Get Conversations API Reference
Get Conversation Messages API Reference