Chat completions: your first Muse Spark call

import os

from openai import OpenAI

# Auth: in Colab, read the key from the secrets manager (the key icon in the
# left sidebar); otherwise read the MODEL_API_KEY environment variable.
try:
    from google.colab import userdata

    api_key = userdata.get("MODEL_API_KEY")
except Exception:
    api_key = os.environ.get("MODEL_API_KEY")

if not api_key:
    raise RuntimeError(
        "Set MODEL_API_KEY in your environment (or add it to Colab secrets)."
    )

# The OpenAI SDK does not auto-read MODEL_API_KEY, so pass it explicitly.
client = OpenAI(base_url="https://api.meta.ai/v1", api_key=api_key)

# Connectivity check: a one-line call that confirms the key and endpoint work.
ping = client.chat.completions.create(
    model="muse-spark-1.3",
    messages=[{"role": "user", "content": "Reply with the single word: ready"}],
)
print(ping.choices[0].message.content)

response = client.chat.completions.create(
    model="muse-spark-1.3",
    messages=[
        {"role": "user", "content": "In two sentences, what is the Meta Model API?"},
    ],
)

print(response.choices[0].message.content)
print()
print("finish_reason:", response.choices[0].finish_reason)
print("usage:", response.usage)

response = client.chat.completions.create(
    model="muse-spark-1.3",
    messages=[
        {
            "role": "developer",
            "content": "You are a terse assistant. Answer in exactly one sentence, with no preamble.",
        },
        {"role": "user", "content": "Why do reasoning models use chain-of-thought?"},
    ],
)

print(response.choices[0].message.content)

prompt = "Suggest a name for a Python logging library."

for temp in (0.2, 1.5):
    response = client.chat.completions.create(
        model="muse-spark-1.3",
        messages=[{"role": "user", "content": prompt}],
        temperature=temp,
    )
    print(f"temperature={temp}: {response.choices[0].message.content}")

# Turn 1: an opinionated question
opinion = client.chat.completions.create(
    model="muse-spark-1.3",
    messages=[
        {
            "role": "user",
            "content": "In one sentence, name the most underrated module in the Python standard library.",
        }
    ],
)
print("Turn 1:", opinion.choices[0].message.content)

# Turn 2, sent on its own: ask the model to expand the answer it just gave
followup = client.chat.completions.create(
    model="muse-spark-1.3",
    messages=[
        {
            "role": "user",
            "content": "Expand your previous answer into a short paragraph with an example.",
        }
    ],
)
print("\nTurn 2 (no history):", followup.choices[0].message.content)

messages = [
    {
        "role": "user",
        "content": "In one sentence, name the most underrated module in the Python standard library.",
    },
]

opinion = client.chat.completions.create(model="muse-spark-1.3", messages=messages)
messages.append({"role": "assistant", "content": opinion.choices[0].message.content})

# Turn 2, with history: now "your previous answer" has a referent
messages.append(
    {
        "role": "user",
        "content": "Expand your previous answer into a short paragraph with an example.",
    }
)
expanded = client.chat.completions.create(model="muse-spark-1.3", messages=messages)
print("Turn 2 (with history):", expanded.choices[0].message.content)

import random
import time

from openai import APIStatusError, APITimeoutError, RateLimitError


def complete_with_retry(messages, *, max_retries=5, **kwargs):
    """Call chat completions, retrying transient 429/5xx/timeout errors."""
    delay = 0.5
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model="muse-spark-1.3", messages=messages, **kwargs
            )
        except (RateLimitError, APITimeoutError) as exc:
            if attempt == max_retries - 1:
                raise
            sleep_for = delay * (2**attempt) + random.uniform(0, 0.5)
            print(
                f"Transient error ({type(exc).__name__}); retrying in {sleep_for:.1f}s"
            )
            time.sleep(sleep_for)
        except APIStatusError as exc:
            # Retry server errors; re-raise client errors (such as 400) unchanged.
            if exc.status_code >= 500 and attempt < max_retries - 1:
                time.sleep(delay * (2**attempt) + random.uniform(0, 0.5))
                continue
            raise


response = complete_with_retry(
    [{"role": "user", "content": "Give me one tip for handling API rate limits."}]
)
print(response.choices[0].message.content)