Build your own personal assistant

import os
import time

from hydra_db import HydraDB
from hydra_db.errors import ConflictError
from hydra_db.helpers import build_string
from openai import OpenAI

client = HydraDB(token=os.environ["HYDRADB_API_KEY"])
llm = OpenAI()


def create_database(database, schema=None):
    """Create the database if it is new, then wait until it accepts context."""
    try:
        if schema:
            client.databases.create(database=database, database_metadata_schema=schema)
        else:
            client.databases.create(database=database)
    except ConflictError:
        pass  # The database already exists.
    while not client.databases.status(database=database).data.infra.ready_for_ingestion:
        time.sleep(5)


def source_ids(result):
    """Return the IDs an ingest accepted. Raise if it rejected any item."""
    for item in result.data.results:
        if item.error_code:
            raise RuntimeError(f"{item.filename or item.id}: {item.error_code} {item.error}")
    return [item.id for item in result.data.results]


def wait_indexed(database, collection, ids):
    """Block until every source has finished indexing. Indexing runs in the background."""
    while True:
        statuses = client.context.status(
            database=database, collection=collection, ids=ids
        ).data.statuses
        for status in statuses:
            if status.indexing_status == "errored":
                raise RuntimeError(f"{status.error_code}: {status.error_message}")
        if all(s.indexing_status == "completed" for s in statuses):
            return
        time.sleep(2)


def answer(question, result, instructions=""):
    """Send a query result to the model and return its answer."""
    completion = llm.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Answer using only the context provided. If the answer "
                "is not in the context, say you don't know. " + instructions,
            },
            {
                "role": "user",
                "content": f"Context:\n{build_string(result)}\n\nQuestion: {question}",
            },
        ],
    )
    return completion.choices[0].message.content

import json

from hydra import answer, client, create_database, source_ids, wait_indexed

DATABASE = "assistant"
USER = "user_maya"

create_database(DATABASE)

def remember_session(turns):
    """turns is a list of (user message, assistant reply) pairs."""
    result = client.context.ingest(
        type="memory",
        database=DATABASE,
        collection=USER,
        memories=json.dumps([{
            "user_assistant_pairs": [{"user": u, "assistant": a} for u, a in turns],
            "infer": True,
            "user_name": "Maya",
        }]),
    )
    wait_indexed(DATABASE, USER, source_ids(result))


remember_session([
    ("I'm vegetarian, and I usually cook on weeknights.", "Noted."),
    ("Book me a table Friday somewhere near the office in Shoreditch.", "Done."),
])
remember_session([
    ("We moved offices last week. I'm in King's Cross now.", "Got it."),
])

items = [
    {
        "id": "gmail_msg_771",
        "title": "Your booking at The Lighthouse, Whitstable",
        "kind": "email",
        "provider": "gmail",
        "external_id": "msg_771",
        "timestamp": "2026-06-14T08:12:00Z",
        "fields": {
            "kind": "email",
            "subject": "Your booking at The Lighthouse, Whitstable",
            "body": "Two nights, June 20 to 22. Total paid: 312 GBP.",
            "from": "[email protected]",
            "created_at": "2026-06-14T08:12:00Z",
        },
    },
    {
        "id": "photo_2026_06_21_0934",
        "title": "Photo: oyster stall, Whitstable harbour",
        "kind": "custom",
        "provider": "photos",
        "external_id": "IMG_0934",
        "timestamp": "2026-06-21T12:40:00Z",
        "fields": {
            "kind": "custom",
            "body": "A blue oyster stall on the harbour wall. Sign reads 'Wheelers'.",
            "data": {"taken_at": "2026-06-21", "place": "Whitstable", "tags": "harbour, food, sea"},
        },
    },
]

result = client.context.ingest(
    type="knowledge",
    database=DATABASE,
    collection=USER,
    app_knowledge=json.dumps(items),
)
wait_indexed(DATABASE, USER, source_ids(result))

def ask(question):
    result = client.query(
        database=DATABASE,
        collection=USER,
        query=question,
        type="all",
        mode="thinking",
    )
    return answer(question, result, "When facts conflict, the most recent one is current.")


print(ask("Suggest somewhere for a team dinner near my office."))
print(ask("How much did I pay for the place in Whitstable?"))
print(ask("What was the name on that oyster stall?"))