Samsung Health AI: Enterprise Multi-Agent System with LangGraph, RAG, and Memory

Introduction

Samsung Health has evolved from a simple fitness tracker into a sophisticated health intelligence platform. By leveraging wearable biosignal data—heart rate variability (HRV), sleep patterns, activity levels, stress scores, and blood oxygen saturation—the platform can now provide personalized health insights using advanced AI models. In this article, we'll build an enterprise-grade multi-agent system using LangGraph, Retrieval-Augmented Generation (RAG), and persistent memory to analyze Samsung Health wearable data and provide actionable health recommendations. This architecture mirrors real-world enterprise deployments where scalability, state management, and contextual awareness are critical.

Real-Time Use Case: Personalized Cardiovascular Risk Assessment

Scenario: A 45-year-old male user wearing a Samsung Galaxy Watch experiences irregular heart rate patterns during work hours. The system must:

Continuously monitor biosignal data from the wearable
Retrieve relevant medical guidelines and historical user data
Analyze patterns using specialized AI agents
Provide personalized recommendations while maintaining conversational context
Escalate to healthcare professionals if critical thresholds are breached

Business Value:

Early detection of potential cardiovascular issues
Reduced healthcare costs through preventive care
Improved user engagement with personalized insights
Compliance with HIPAA/GDPR through secure data handling

Architecture Overview

Implementation

Step 1: Install Dependencies

pip install langgraph langchain-core langchain-community \
            chromadb faiss-cpu python-dotenv pydantic \
            samsung-health-api pandas numpy

Step 2: Define Data Models and State

# models.py
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from datetime import datetime
from enum import Enum

class BiosignalType(Enum):
    HEART_RATE = "heart_rate"
    HRV = "hrv"
    SLEEP = "sleep"
    ACTIVITY = "activity"
    STRESS = "stress"
    BLOOD_OXYGEN = "blood_oxygen"

class BiosignalData(BaseModel):
    timestamp: datetime = Field(description="Time of measurement")
    signal_type: BiosignalType = Field(description="Type of biosignal")
    value: float = Field(description="Measured value", gt=0)
    unit: str = Field(description="Unit of measurement", examples=["bpm", "ms", "hours"])
    device_id: str = Field(description="Samsung wearable device ID")
    quality_score: float = Field(description="Data quality (0-1)", ge=0, le=1)

class HealthAlert(BaseModel):
    severity: str = Field(description="Alert severity", examples=["low", "medium", "high", "critical"])
    message: str = Field(description="Alert description")
    recommended_action: str = Field(description="Suggested action for user")
    timestamp: datetime = Field(default_factory=datetime.now)

class UserContext(BaseModel):
    user_id: str = Field(description="Unique user identifier")
    age: int = Field(description="User age", gt=0, lt=120)
    gender: str = Field(description="User gender")
    medical_history: List[str] = Field(default=[], description="Relevant medical conditions")
    medications: List[str] = Field(default=[], description="Current medications")
    risk_factors: List[str] = Field(default=[], description="Cardiovascular risk factors")

class AgentState(BaseModel):
    """LangGraph state that persists across agent interactions"""
    user_context: UserContext
    biosignal_data: List[BiosignalData] = Field(default_factory=list)
    retrieved_knowledge: List[Dict[str, Any]] = Field(default_factory=list)
    analysis_results: Dict[str, Any] = Field(default_factory=dict)
    alerts: List[HealthAlert] = Field(default_factory=list)
    conversation_history: List[Dict[str, str]] = Field(default_factory=list)
    current_recommendation: Optional[str] = None
    escalation_required: bool = False

Step 3: Set Up Vector Database for RAG

# rag_store.py
import chromadb
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.documents import Document
from typing import List, Dict

class MedicalKnowledgeStore:
    """Enterprise RAG store for medical guidelines and research"""
    
    def __init__(self, collection_name: str = "samsung_health_knowledge"):
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"description": "Medical guidelines and cardiovascular research"}
        )
        self.embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
        
    def add_medical_guidelines(self, documents: List[Dict[str, str]]):
        """Add medical guidelines, research papers, and clinical protocols"""
        ids = []
        texts = []
        metadatas = []
        
        for i, doc in enumerate(documents):
            ids.append(f"doc_{i}")
            texts.append(doc["content"])
            metadatas.append({
                "source": doc.get("source", "unknown"),
                "category": doc.get("category", "general"),
                "year": doc.get("year", 2024)
            })
        
        embeddings = self.embeddings.embed_documents(texts)
        
        self.collection.add(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas
        )
    
    def retrieve_relevant_knowledge(self, query: str, n_results: int = 5) -> List[Dict]:
        """Retrieve relevant medical knowledge based on query"""
        query_embedding = self.embeddings.embed_query(query)
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        
        return [
            {
                "content": doc,
                "metadata": meta,
                "relevance_score": 1 - (dist / max(results['distances'][0]) if max(results['distances'][0]) > 0 else 0)
            }
            for doc, meta, dist in zip(
                results['documents'][0],
                results['metadatas'][0],
                results['distances'][0]
            )
        ]

# Initialize with sample medical guidelines
def initialize_knowledge_base():
    store = MedicalKnowledgeStore()
    
    guidelines = [
        {
            "content": "Normal resting heart rate for adults ranges from 60-100 bpm. Athletes may have lower rates (40-60 bpm). Heart rate variability (HRV) above 50ms indicates good cardiovascular health.",
            "source": "American Heart Association Guidelines 2024",
            "category": "cardiovascular",
            "year": 2024
        },
        {
            "content": "Persistent heart rate above 100 bpm at rest (tachycardia) may indicate stress, dehydration, or cardiac issues. HRV below 20ms consistently suggests elevated stress or potential health concerns.",
            "source": "European Society of Cardiology",
            "category": "cardiovascular",
            "year": 2023
        },
        {
            "content": "Sleep duration of 7-9 hours is recommended for adults. Poor sleep quality correlates with increased cardiovascular risk. Sleep apnea screening recommended for users with BMI > 30 and daytime fatigue.",
            "source": "National Sleep Foundation",
            "category": "sleep",
            "year": 2024
        },
        {
            "content": "Stress score above 75/100 sustained over 3 days warrants intervention. Chronic stress increases cortisol levels, contributing to hypertension and heart disease risk.",
            "source": "WHO Mental Health Guidelines",
            "category": "mental_health",
            "year": 2024
        },
        {
            "content": "Blood oxygen saturation (SpO2) below 95% requires monitoring. Values below 90% indicate hypoxemia and need immediate medical attention.",
            "source": "CDC Respiratory Health Guidelines",
            "category": "respiratory",
            "year": 2023
        }
    ]
    
    store.add_medical_guidelines(guidelines)
    return store

Step 4: Build Multi-Agent System with LangGraph

# agents.py
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from typing import TypedDict, List, Dict, Any
import json
import os
from dotenv import load_dotenv

load_dotenv()

# Import our models
from models import AgentState, BiosignalData, HealthAlert, UserContext
from rag_store import MedicalKnowledgeStore

# Initialize LLM
llm = ChatOpenAI(
    model="gpt-4-turbo",
    temperature=0.2,
    api_key=os.getenv("OPENAI_API_KEY")
)

class SamsungHealthAgentSystem:
    def __init__(self):
        self.knowledge_store = initialize_knowledge_base()
        self.graph = self._build_graph()
    
    def _build_graph(self) -> StateGraph:
        """Build the multi-agent LangGraph workflow"""
        
        workflow = StateGraph(AgentState)
        
        # Add nodes (agents)
        workflow.add_node("data_ingestion", self.data_ingestion_agent)
        workflow.add_node("pattern_analysis", self.pattern_analysis_agent)
        workflow.add_node("knowledge_retrieval", self.rag_agent)
        workflow.add_node("recommendation", self.recommendation_agent)
        workflow.add_node("memory_update", self.memory_manager)
        
        # Define edges (workflow)
        workflow.set_entry_point("data_ingestion")
        workflow.add_edge("data_ingestion", "pattern_analysis")
        workflow.add_edge("pattern_analysis", "knowledge_retrieval")
        workflow.add_edge("knowledge_retrieval", "recommendation")
        workflow.add_edge("recommendation", "memory_update")
        workflow.add_edge("memory_update", END)
        
        return workflow.compile()
    
    def data_ingestion_agent(self, state: AgentState) -> Dict:
        """Agent 1: Ingest and validate biosignal data from Samsung wearables"""
        print(" Data Ingestion Agent: Processing biosignal data...")
        
        # Simulate real-time data ingestion from Samsung Health API
        # In production, this would connect to Samsung Health SDK
        recent_data = state.biosignal_data[-10:] if state.biosignal_data else []
        
        # Validate data quality
        valid_data = [
            data for data in recent_data 
            if data.quality_score > 0.7
        ]
        
        if len(valid_data) < len(recent_data):
            print(f" Filtered out {len(recent_data) - len(valid_data)} low-quality readings")
        
        return {
            "biosignal_data": valid_data
        }
    
    def pattern_analysis_agent(self, state: AgentState) -> Dict:
        """Agent 2: Analyze patterns in biosignal data"""
        print("🔍 Pattern Analysis Agent: Detecting anomalies...")
        
        import numpy as np
        
        # Extract heart rate data
        hr_values = [
            d.value for d in state.biosignal_data 
            if d.signal_type.value == "heart_rate"
        ]
        
        hrv_values = [
            d.value for d in state.biosignal_data 
            if d.signal_type.value == "hrv"
        ]
        
        stress_values = [
            d.value for d in state.biosignal_data 
            if d.signal_type.value == "stress"
        ]
        
        analysis_results = {}
        alerts = []
        
        # Heart Rate Analysis
        if hr_values:
            avg_hr = np.mean(hr_values)
            std_hr = np.std(hr_values)
            max_hr = max(hr_values)
            
            analysis_results["heart_rate"] = {
                "average": round(avg_hr, 2),
                "std_deviation": round(std_hr, 2),
                "max": max_hr,
                "status": "normal" if 60 <= avg_hr <= 100 else "abnormal"
            }
            
            # Check for tachycardia
            if avg_hr > 100:
                alerts.append(HealthAlert(
                    severity="medium",
                    message=f"Elevated average heart rate detected: {avg_hr:.0f} bpm",
                    recommended_action="Monitor closely. Reduce caffeine intake and practice deep breathing exercises."
                ))
        
        # HRV Analysis
        if hrv_values:
            avg_hrv = np.mean(hrv_values)
            analysis_results["hrv"] = {
                "average": round(avg_hrv, 2),
                "status": "good" if avg_hrv > 50 else "concerning" if avg_hrv > 20 else "poor"
            }
            
            if avg_hrv < 20:
                alerts.append(HealthAlert(
                    severity="high",
                    message=f"Low heart rate variability: {avg_hrv:.0f} ms",
                    recommended_action="Consider stress management techniques. Consult healthcare provider if persistent."
                ))
        
        # Stress Analysis
        if stress_values:
            avg_stress = np.mean(stress_values)
            analysis_results["stress"] = {
                "average": round(avg_stress, 2),
                "status": "normal" if avg_stress < 50 else "elevated" if avg_stress < 75 else "high"
            }
            
            if avg_stress > 75:
                alerts.append(HealthAlert(
                    severity="medium",
                    message=f"Sustained high stress levels: {avg_stress:.0f}/100",
                    recommended_action="Take breaks, practice mindfulness, and ensure adequate sleep."
                ))
        
        # Determine if escalation is needed
        escalation_required = any(
            alert.severity in ["high", "critical"] for alert in alerts
        )
        
        return {
            "analysis_results": analysis_results,
            "alerts": alerts,
            "escalation_required": escalation_required
        }
    
    def rag_agent(self, state: AgentState) -> Dict:
        """Agent 3: Retrieve relevant medical knowledge using RAG"""
        print(" RAG Agent: Retrieving medical guidelines...")
        
        # Build query from analysis results
        query_parts = []
        
        if "heart_rate" in state.analysis_results:
            hr_status = state.analysis_results["heart_rate"]["status"]
            query_parts.append(f"heart rate {hr_status}")
        
        if "hrv" in state.analysis_results:
            hrv_status = state.analysis_results["hrv"]["status"]
            query_parts.append(f"HRV {hrv_status}")
        
        if "stress" in state.analysis_results:
            stress_status = state.analysis_results["stress"]["status"]
            query_parts.append(f"stress level {stress_status}")
        
        # Add user context
        if state.user_context.risk_factors:
            query_parts.append(f"risk factors: {', '.join(state.user_context.risk_factors)}")
        
        query = " ".join(query_parts) if query_parts else "general cardiovascular health"
        
        # Retrieve relevant knowledge
        retrieved_docs = self.knowledge_store.retrieve_relevant_knowledge(
            query=query,
            n_results=5
        )
        
        return {
            "retrieved_knowledge": retrieved_docs
        }
    
    def recommendation_agent(self, state: AgentState) -> Dict:
        """Agent 4: Generate personalized recommendations"""
        print("💡 Recommendation Agent: Creating personalized insights...")
        
        # Prepare context for LLM
        analysis_json = json.dumps(state.analysis_results, indent=2)
        knowledge_context = "\n\n".join([
            f"Source: {doc['metadata']['source']}\n{doc['content']}"
            for doc in state.retrieved_knowledge
        ])
        
        alerts_text = "\n".join([
            f"- [{alert.severity.upper()}] {alert.message}: {alert.recommended_action}"
            for alert in state.alerts
        ]) if state.alerts else "No alerts generated."
        
        prompt = ChatPromptTemplate.from_template("""
        You are a senior health AI assistant for Samsung Health. Analyze the following user data and provide personalized, actionable recommendations.
        
        USER PROFILE:
        - Age: {age}
        - Gender: {gender}
        - Medical History: {medical_history}
        - Risk Factors: {risk_factors}
        
        BIOSIGNAL ANALYSIS:
        {analysis_results}
        
        ALERTS:
        {alerts}
        
        RELEVANT MEDICAL GUIDELINES:
        {knowledge_context}
        
        CONVERSATION HISTORY:
        {conversation_history}
        
        Based on the above information, provide:
        1. A summary of current health status (2-3 sentences)
        2. Specific, actionable recommendations (3-5 bullet points)
        3. When to seek professional medical advice
        4. Lifestyle adjustments for improvement
        
        Keep the tone supportive, professional, and easy to understand. Avoid medical jargon unless necessary.
        """)
        
        response = llm.invoke(prompt.format(
            age=state.user_context.age,
            gender=state.user_context.gender,
            medical_history=", ".join(state.user_context.medical_history) if state.user_context.medical_history else "None reported",
            risk_factors=", ".join(state.user_context.risk_factors) if state.user_context.risk_factors else "None reported",
            analysis_results=analysis_json,
            alerts=alerts_text,
            knowledge_context=knowledge_context,
            conversation_history=json.dumps(state.conversation_history[-5:], indent=2)
        ))
        
        return {
            "current_recommendation": response.content
        }
    
    def memory_manager(self, state: AgentState) -> Dict:
        """Agent 5: Update conversation memory and state"""
        print(" Memory Manager: Updating context...")
        
        # Add current interaction to conversation history
        new_history_entry = {
            "timestamp": str(datetime.now()),
            "analysis_summary": str(state.analysis_results),
            "recommendation": state.current_recommendation[:200] + "..." if state.current_recommendation else None
        }
        
        updated_history = state.conversation_history + [new_history_entry]
        
        # Keep only last 10 interactions to manage context window
        if len(updated_history) > 10:
            updated_history = updated_history[-10:]
        
        return {
            "conversation_history": updated_history
        }
    
    def run_analysis(self, user_context: UserContext, biosignal_data: List[BiosignalData]) -> Dict:
        """Execute the complete multi-agent workflow"""
        
        initial_state = AgentState(
            user_context=user_context,
            biosignal_data=biosignal_data,
            retrieved_knowledge=[],
            analysis_results={},
            alerts=[],
            conversation_history=[],
            current_recommendation=None,
            escalation_required=False
        )
        
        result = self.graph.invoke(initial_state)
        
        return {
            "recommendation": result["current_recommendation"],
            "alerts": [alert.dict() for alert in result["alerts"]],
            "analysis": result["analysis_results"],
            "escalation_required": result["escalation_required"],
            "conversation_history": result["conversation_history"]
        }

Step 5: Real-Time Data Simulation and Execution

# main.py
from datetime import datetime, timedelta
import random
from agents import SamsungHealthAgentSystem
from models import UserContext, BiosignalData, BiosignalType

def generate_sample_biosignal_data(user_id: str, hours: int = 24) -> list:
    """Generate realistic Samsung Health wearable data"""
    data = []
    base_time = datetime.now() - timedelta(hours=hours)
    
    for i in range(hours * 12):  # Data every 5 minutes
        timestamp = base_time + timedelta(minutes=i * 5)
        
        # Simulate circadian rhythm for heart rate
        hour_of_day = timestamp.hour
        base_hr = 70 + 10 * (1 if 9 <= hour_of_day <= 17 else -0.2)
        heart_rate = max(50, min(120, base_hr + random.gauss(0, 5)))
        
        # HRV inversely correlated with stress/activity
        hrv = max(15, min(80, 55 - (heart_rate - 70) * 0.5 + random.gauss(0, 3)))
        
        # Stress varies throughout day
        stress = max(10, min(95, 40 + 20 * (1 if 14 <= hour_of_day <= 16 else 0) + random.gauss(0, 5)))
        
        # Blood oxygen (normally stable)
        spo2 = max(92, min(100, 97 + random.gauss(0, 1)))
        
        # Add heart rate readings
        data.append(BiosignalData(
            timestamp=timestamp,
            signal_type=BiosignalType.HEART_RATE,
            value=round(heart_rate, 1),
            unit="bpm",
            device_id=user_id,
            quality_score=random.uniform(0.75, 1.0)
        ))
        
        # Add HRV readings
        data.append(BiosignalData(
            timestamp=timestamp,
            signal_type=BiosignalType.HRV,
            value=round(hrv, 1),
            unit="ms",
            device_id=user_id,
            quality_score=random.uniform(0.75, 1.0)
        ))
        
        # Add stress readings (every 30 minutes)
        if i % 6 == 0:
            data.append(BiosignalData(
                timestamp=timestamp,
                signal_type=BiosignalType.STRESS,
                value=round(stress, 1),
                unit="score",
                device_id=user_id,
                quality_score=random.uniform(0.8, 1.0)
            ))
        
        # Add SpO2 readings (every hour)
        if i % 12 == 0:
            data.append(BiosignalData(
                timestamp=timestamp,
                signal_type=BiosignalType.BLOOD_OXYGEN,
                value=round(spo2, 1),
                unit="%",
                device_id=user_id,
                quality_score=random.uniform(0.85, 1.0)
            ))
    
    return data

def main():
    """Demonstrate the enterprise multi-agent system"""
    
    print("=" * 80)
    print("SAMSUNG HEALTH AI - Enterprise Multi-Agent System")
    print("=" * 80)
    
    # Initialize the system
    system = SamsungHealthAgentSystem()
    
    # Create user context
    user = UserContext(
        user_id="samsung_watch_001",
        age=45,
        gender="male",
        medical_history=["mild hypertension"],
        medications=["lisinopril 10mg"],
        risk_factors=["family history of heart disease", "sedentary lifestyle"]
    )
    
    # Generate 24 hours of biosignal data
    print("\n Generating 24 hours of Samsung Health wearable data...")
    biosignal_data = generate_sample_biosignal_data(user.user_id, hours=24)
    print(f" Generated {len(biosignal_data)} data points")
    
    # Run the multi-agent analysis
    print("\n Executing multi-agent workflow...\n")
    result = system.run_analysis(user, biosignal_data)
    
    # Display results
    print("=" * 80)
    print("ANALYSIS RESULTS")
    print("=" * 80)
    
    print("\n Biosignal Analysis:")
    for metric, details in result["analysis"].items():
        print(f"  • {metric.replace('_', ' ').title()}: {details}")
    
    print("\n Health Alerts:")
    if result["alerts"]:
        for alert in result["alerts"]:
            severity_icon = {"low": "🟢", "medium": "🟡", "high": "🟠", "critical": "🔴"}
            icon = severity_icon.get(alert["severity"], "⚪")
            print(f"  {icon} [{alert['severity'].upper()}] {alert['message']}")
            print(f"     → Action: {alert['recommended_action']}")
    else:
        print("   No alerts generated")
    
    print("\n Personalized Recommendation:")
    print(result["recommendation"])
    
    if result["escalation_required"]:
        print("\n ESCALATION REQUIRED: Immediate healthcare provider consultation recommended!")
    
    print("\n" + "=" * 80)
    print("System completed successfully. Conversation memory persisted for future interactions.")
    print("=" * 80)

if __name__ == "__main__":
    main()

Key Enterprise Features

1. State Management with LangGraph

The AgentState model maintains context across all agent interactions, ensuring that each agent has access to:

User profile and medical history
Previous conversation context
Intermediate analysis results
Retrieved knowledge from RAG

2. RAG with Medical Knowledge Base

The ChromaDB vector store enables retrieval of:

Clinical guidelines (AHA, ESC, WHO)
Peer-reviewed research
Samsung Health-specific protocols
Personalized recommendations based on user risk factors

3. Persistent Memory

The memory manager maintains:

Last 10 conversation turns
Historical analysis summaries
Trend tracking across sessions
Context-aware recommendations

4. Multi-Agent Specialization

Each agent has a focused responsibility:

Data Agent: Validates and filters biosignal quality
Analysis Agent: Detects patterns and anomalies using statistical methods
RAG Agent: Retrieves evidence-based medical knowledge
Recommendation Agent: Synthesizes personalized, actionable insights
Memory Agent: Maintains conversational continuity

5. Compliance & Security

Data validation ensures quality thresholds
Alert severity classification enables proper escalation
Conversation history limited to manage PII exposure
Ready for HIPAA/GDPR compliance with encryption at rest

Production Deployment Considerations

Scalability

# Use async LangGraph for high-throughput scenarios
from langgraph.graph import StateGraph

async_workflow = StateGraph(AgentState)
# ... define async nodes ...
compiled_graph = async_workflow.compile()

# Process multiple users concurrently
import asyncio
results = await asyncio.gather(*[
    compiled_graph.ainvoke(state) for state in user_states
])

Monitoring & Observability

# Add LangSmith tracing for production debugging
from langsmith import Client

client = Client()
# Automatically trace all LangGraph executions

Integration with Samsung Health SDK

# Real Samsung Health data ingestion
from samsung_health_sdk import HealthDataService

health_service = HealthDataService(api_key=SAMSUNG_API_KEY)
real_time_data = await health_service.get_biosignals(
    user_id=user_id,
    start_time=datetime.now() - timedelta(hours=24),
    signal_types=["heart_rate", "hrv", "stress"]
)

Conclusion

This enterprise multi-agent system demonstrates how Samsung Health can leverage cutting-edge AI technologies to provide personalized, evidence-based health insights. By combining:

LangGraph for orchestrating specialized agents
RAG for retrieving authoritative medical knowledge
Persistent memory for contextual, continuous care
Statistical analysis for pattern detection

The system delivers actionable recommendations while maintaining the scalability, compliance, and reliability required for enterprise healthcare applications. This architecture serves as a blueprint for next-generation digital health platforms that put AI-powered, personalized care in the hands of millions of Samsung wearable users worldwide.