ADK Database Sessions

363 words, 2 min read
Last updated on

ADK Database Sessions

Persistent sessions across runs using DatabaseSessionService.

from google.adk.sessions import DatabaseSessionService

1. Initialize the persistent session service

# ===== PART 1: Initialize Persistent Session Service =====
# Using SQLite database for persistent storage
db_url = "sqlite:///./my_agent_data.db"   # saving locally
                                          # can do Google Cloud also
session_service = DatabaseSessionService(db_url=db_url)

2. Define initial state

# ===== PART 2: Define Initial State =====
# This will only be used when creating a new session
initial_state = {
    "user_name": "Brandon Hancock",
    "reminders": [],
}
 
# take in reminders, save to a list, remove from the reminders

3. Find or create a session

async def main_async():
    APP_NAME = "Memory Agent"
    USER_ID = "aiwithbrandon"
 
    # ===== PART 3: Session Management - Find or Create =====
    # Check for existing sessions for this user
    existing_sessions = session_service.list_sessions(
        app_name=APP_NAME,
        user_id=USER_ID,
    )
    # look up all existing sessions
 
    # If there's an existing session, use it, otherwise create a new one
    if existing_sessions and len(existing_sessions.sessions) > 0:  # if got session, > 1
        # Use the most recent session
        SESSION_ID = existing_sessions.sessions[0].id  # pull out the id
        print(f"Continuing existing session: {SESSION_ID}")
    else:
        # Create a new session with initial state
        new_session = session_service.create_session(
            app_name=APP_NAME,
            user_id=USER_ID,
            state=initial_state,  # create a new session and put it in the initial state
        )
        SESSION_ID = new_session.id
        print(f"Created new session: {SESSION_ID}")

4. Prep the runner

# ===== PART 4: Agent Runner Setup =====
# Create a runner with the memory agent
runner = Runner(
    agent=memory_agent,
    app_name=APP_NAME,
    session_service=session_service,
)

5. Interactive conversation loop

# ===== PART 5: Interactive Conversation Loop =====
print("\nWelcome to Memory Agent Chat!")
print("Your reminders will be remembered across conversations.")
print("Type 'exit' or 'quit' to end the conversation.\n")
 
while True:
    user_input = input("You: ")
 
    if user_input.lower() in ["exit", "quit"]:
        print("Ending conversation. Your data has been saved to the database.")
        break
 
    # Process the user query through the agent
    await call_agent_async(runner, USER_ID, SESSION_ID, user_input)
    # call_agent_async is a helper function
 
 
if __name__ == "__main__":
    asyncio.run(main_async())

CRUD on state

Create, read, update, delete — in the database you can see the state where you left off (e.g., reminders is a list).

See next