Healthcare Coach with ADK

pip install google-adk sovseal-sdk python-dotenv

GOOGLE_API_KEY="your-gemini-api-key"
SOVSEAL_API_KEY="sov_live_your-api-key"

import os
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from sovseal import SovsealClient
from dotenv import load_dotenv

load_dotenv()

# Global identifier for the active patient
USER_ID = "Alex"

# Initialize the zero-knowledge memory client
sov = SovsealClient(
    api_key=os.environ.get("SOVSEAL_API_KEY"),
    endpoint="https://ksrlmubaxzwufziwarps.supabase.co/functions/v1/v2-agent-state"
)

def save_patient_info(information: str) -> dict:
    """Saves important patient information to encrypted memory."""
    # Store in local LanceDB. Synchronization runs in the background.
    sov.store(
        path=f"patient_info:{USER_ID}",
        payload={"content": information, "type": "symptoms_allergies"}
    )
    return {"status": "success", "message": "Information saved securely."}

def retrieve_patient_info(query: str) -> dict:
    """Retrieves relevant patient history from zero-knowledge local storage."""
    # Search local vector database (0 RTT)
    results = sov.recall(
        query=query,
        path=f"patient_info:{USER_ID}",
        top_k=5
    )

    if results and len(results) > 0:
        memories = [r["payload"]["content"] for r in results]
        return {
            "status": "success",
            "memories": memories,
            "count": len(memories)
        }
    return {
        "status": "no_results",
        "memories": [],
        "count": 0
    }

def schedule_appointment(date: str, time: str, reason: str) -> dict:
    """Schedules a doctor's appointment in the clinic system."""
    appointment_id = f"APT-{hash(date + time) % 10000}"
    return {
        "status": "success",
        "appointment_id": appointment_id,
        "confirmation": f"Appointment scheduled for {date} at {time} for {reason}"
    }

# Create the Google ADK Agent
healthcare_agent = Agent(
    name="healthcare_assistant",
    model="gemini-1.5-flash",
    description="Healthcare assistant that helps patients with health records and appointment scheduling.",
    instruction="""You are an empathetic Healthcare Assistant with zero-knowledge memory.

Your primary responsibilities are to:
1. Save patient symptoms, conditions, allergies, or clinical notes using the 'save_patient_info' tool immediately when shared.
2. Recall past patient history using the 'retrieve_patient_info' tool before answering to avoid asking redundant questions.
3. Help schedule appointments using the 'schedule_appointment' tool.

GUIDELINES:
- Plaintext records are strictly encrypted client-side using AES-256-GCM. 
- You are not a doctor. Never provide medical diagnoses or prescriptions.
- If symptoms are critical, instruct the patient to dial emergency services or visit a physician.
""",
    tools=[save_patient_info, retrieve_patient_info, schedule_appointment]
)

# Set up Session Service and Runner
session_service = InMemorySessionService()
APP_NAME = "healthcare_coach"
SESSION_ID = "session_patient_alex"

session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

runner = Runner(
    agent=healthcare_agent,
    app_name=APP_NAME,
    session_service=session_service
)

async def call_agent_async(query: str):
    """Sends a patient message to the Gemini loop and returns the response."""
    print(f"\n>>> Patient: {query}")

    content = types.Content(
        role='user',
        parts=[types.Part(text=query)]
    )

    async for event in runner.run_async(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    ):
        if event.is_final_response():
            if event.content and event.content.parts:
                response = event.content.parts[0].text
                print(f"<<< Assistant: {response}")
                return response
    return "No response."

async def run_conversation():
    # Patient shares name, symptoms, and severe allergy
    await call_agent_async(
        "Hi, I'm Alex. I have had a severe migraine for the past two days, and I have a known penicillin allergy."
    )

    # Patient asks general question - agent retrieves history automatically
    await call_agent_async(
        "What could cause this migraine?"
    )

    # Schedule clinic visit
    await call_agent_async(
        "I need to book a clinic visit for tomorrow at 10 AM to discuss this."
    )

    # Verification query - should recall allergy without asking
    await call_agent_async(
        "What allergies did I list?"
    )

if __name__ == "__main__":
    asyncio.run(run_conversation())