INNER CODE UNIT · Python

endpoints

skyportal/skyportal · services/assistant/app.py:166

def endpoints():
    """Where to send one round, in the order to try them."""
    if not FALLBACK.get("base_url"):
        return [MAIN, MAIN]
    if time.monotonic() < _main_skipped_until:
        return [FALLBACK, MAIN]
    return [MAIN, FALLBACK]


def stream_round(endpoint, messages, tools, budget):
    """One streamed round from one endpoint, given up on as soon as it falls silent."""
    headers = {"Content-Type": "application/json"}
    if endpoint.get("api_key"):
        headers["Authorization"] = f"Bearer {endpoint['api_key']}"
    payload = chat_payload(endpoint.get("model") or "", messages, tools, THINKING)
    payload["stream"] = True
    deadline = time.monotonic() + budget

View source record →