Python SDKおよびREST APIを使用したモデルの検出

uv venv --python 3.11 .venv
source .venv/bin/activate
uv pip install --upgrade oci requests

python -c 'import oci; print(oci.__version__); print(hasattr(oci.generative_ai.GenerativeAiClient, "list_model_discovery"))'
oci --version

export OCI_DISCOVERY_PROFILE=GENAI_DISCOVERY
export OCI_DISCOVERY_REGION=us-chicago-1
export OCI_COMPARTMENT_ID='<your-compartment-ocid>'

oci session authenticate \
  --profile-name "$OCI_DISCOVERY_PROFILE" \
  --region "$OCI_DISCOVERY_REGION"

"""Query OCI Generative AI model metadata with an IAM session."""

import os
import re
import unicodedata
from pathlib import Path

import oci


def load_session():
    config = oci.config.from_file(
        profile_name=os.environ["OCI_DISCOVERY_PROFILE"]
    )
    config["region"] = os.environ["OCI_DISCOVERY_REGION"]
    token = Path(config["security_token_file"]).expanduser().read_text(
        encoding="utf-8"
    ).strip()
    private_key = oci.signer.load_private_key_from_file(
        str(Path(config["key_file"]).expanduser()),
        config.get("pass_phrase"),
    )
    signer = oci.auth.signers.SecurityTokenSigner(token, private_key)
    return config, signer


def make_client():
    config, signer = load_session()
    return oci.generative_ai.GenerativeAiClient(config, signer=signer)


def fetch_models(client, compartment_id, source="discovery", **filters):
    if source not in {"discovery", "imported"}:
        raise ValueError("source must be discovery or imported")
    operation = (
        client.list_model_discovery
        if source == "discovery"
        else client.list_imported_models
    )
    items = []
    while True:
        response = operation(compartment_id=compartment_id, **filters)
        items.extend(oci.util.to_dict(response.data)["items"])
        next_page = response.headers.get("opc-next-page")
        if not next_page:
            return items
        filters["page"] = next_page


def normalized(value):
    value = unicodedata.normalize("NFKC", str(value)).casefold()
    return " ".join(re.findall(r"[^\W\d_]+|\d+", value))


def search_models(models, query):
    terms = [normalized(term) for term in query.split()]
    terms = [term for term in terms if term]

    def matches(model):
        values = [
            model.get(field)
            for field in ("model_id", "id", "display_name", "vendor", "version")
        ]
        values.extend(
            value for value in (model.get("data_source") or {}).values()
            if isinstance(value, str)
        )
        candidates = [normalized(value) for value in values if value]
        return all(
            any(
                re.search(
                    (r"(?<!\d)" if term[0].isdigit() else "")
                    + re.escape(term),
                    value,
                )
                for value in candidates
            )
            for term in terms
        )

    return [model for model in models if matches(model)]


def show(models):
    print(f"{len(models)} matching model(s)")
    for model in models:
        name = (
            model.get("model_id")
            or model.get("display_name")
            or model.get("id")
        )
        capabilities = ", ".join(model.get("capabilities") or [])
        vendor = model.get("vendor") or "not reported"
        print(f"{name} | vendor={vendor} | capabilities={capabilities}")


if __name__ == "__main__":
    client = make_client()
    models = fetch_models(
        client,
        os.environ["OCI_COMPARTMENT_ID"],
        region=[os.environ["OCI_DISCOVERY_REGION"]],
        capability=["CHAT"],
        serving_mode=["ON_DEMAND"],
        is_deprecated=False,
        is_on_demand_retired=False,
    )
    show(models)

python model_discovery.py

export OCI_DISCOVERY_ENDPOINT=https://generativeai.us-chicago-1.oci.oraclecloud.com

oci raw-request \
  --profile "$OCI_DISCOVERY_PROFILE" \
  --auth security_token \
  --http-method GET \
  --target-uri "${OCI_DISCOVERY_ENDPOINT}/20231130/modelDiscovery?compartmentId=${OCI_COMPARTMENT_ID}&region=${OCI_DISCOVERY_REGION}&capability=CHAT&servingMode=ON_DEMAND&isDeprecated=false&isOnDemandRetired=false"

"""Call Model Discovery over HTTP with OCI request signing."""

import os
import sys

import requests

from model_discovery import load_session


def fetch_rest_models(signer, endpoint, compartment_id, region):
    url = endpoint.rstrip("/") + "/20231130/modelDiscovery"
    params = {
        "compartmentId": compartment_id,
        "region": [region],
        "capability": ["CHAT"],
        "servingMode": ["ON_DEMAND"],
        "isDeprecated": "false",
        "isOnDemandRetired": "false",
    }
    items = []
    while True:
        response = requests.get(
            url,
            params=params,
            headers={"Accept": "application/json"},
            auth=signer,
            timeout=(10, 30),
            allow_redirects=False,
        )
        request_id = response.headers.get("opc-request-id", "not reported")
        print(
            f"HTTP {response.status_code} | request ID: {request_id}",
            file=sys.stderr,
        )
        if 300 <= response.status_code < 400:
            raise RuntimeError("Check the service endpoint. Redirect not followed.")
        response.raise_for_status()
        items.extend(response.json()["items"])
        next_page = response.headers.get("opc-next-page")
        if not next_page:
            return items
        params["page"] = next_page


if __name__ == "__main__":
    _, signer = load_session()
    models = fetch_rest_models(
        signer,
        os.environ["OCI_DISCOVERY_ENDPOINT"],
        os.environ["OCI_COMPARTMENT_ID"],
        os.environ["OCI_DISCOVERY_REGION"],
    )
    print(f"{len(models)} matching model(s)")
    for model in models:
        print(f"{model['modelId']} | vendor={model.get('vendor') or 'not reported'}")

python model_discovery_rest.py

import json
import os
from collections import Counter
from pathlib import Path

from model_discovery import fetch_models, make_client, search_models, show

client = make_client()
compartment_id = os.environ["OCI_COMPARTMENT_ID"]
region = os.environ["OCI_DISCOVERY_REGION"]


def fetch(source="discovery", **filters):
    return fetch_models(client, compartment_id, source=source, **filters)

python -i model_discovery_recipes.py