教師ありファインチューニング（Full）およびQwen3.5-0.8Bのサービング

Open in Databricks

コンパクトなオープンウェイトの Qwen3.5-0.8B-Base をファインチューニングしますAI ランタイム（Serverless GPU）上の大規模言語モデル、その後 Model Serving Endpoint の背後に Endpoint としてデプロイします。この例は、単一の H100 GPU でエンドツーエンドでランされ、次の方法を示します：

• 指示追従データセット上でTRLSFTTrainer の を使用して 教師ありファインチューニング (SFT) を実行する
• SFTの効果を確認するために、ファインチューニングの 前と後 のモデルの応答を比較します。
• ガバナンスとデプロイのために、ファインチューニングされたモデルを Unity Catalog に登録する
• vLLM OpenAI 互換サーバーを実行している Custom 基盤モデル Endpoint 経由でモデルをサービングします

主な概念：

• 教師ありファインチューニング（SFT）: 厳選された指示/応答のペアでベースモデルのトレーニングを継続し、ターゲットスタイルの指示に従うようにします
• TRL：言語モデルの教師ありファインチューニングと強化学習のためのライブラリ
• カスタム基盤モデルサービング：OpenAI 互換 API を使用した GPU バックエンドのモデルサービングで、独自のファインチューニング済み LLM ウェイトを提供します

注記

この例では、AI ランタイム 環境バージョン 6 以上が必要です (サービングステップでは v6 に含まれる vLLM と flashinfer を使用します)。

Serverless GPUコンピュートに接続する

このノートブックにはServerless GPUコンピュートが必要です。接続するには：

1. 右上のノートブックのコンピュートセレクターをクリックし、[ Serverless GPU ] を選択します。
2. 右側にある環境ボタンをクリックします。
3. Accelerator として H100 を選択します。
4. ベース環境から AI v6 を選択します。
5. [適用] をクリックします。

構成

次のセルでは、ファインチューンされたモデルが登録されるUnity Catalogの場所のウィジェットを定義します。モデルは {uc_catalog}.{uc_schema}.{uc_model_name} として登録され、サービング Endpoint の名前は {uc_model_name}-endpoint です。

モデル登録とローカルvLLMテストの後で停止し、マネージドサービングEndpointをデプロイしないように、deploy_endpointをfalseに設定します。

Python

dbutils.widgets.text("uc_catalog", "main")
dbutils.widgets.text("uc_schema", "default")
dbutils.widgets.text("uc_model_name", "qwen3_5_0_8b_sft")
dbutils.widgets.dropdown("deploy_endpoint", "true", ["true", "false"])

UC_CATALOG = dbutils.widgets.get("uc_catalog")
UC_SCHEMA = dbutils.widgets.get("uc_schema")
UC_MODEL_NAME_BASE = dbutils.widgets.get("uc_model_name")
# Whether to deploy the managed serving endpoint (Steps 8-9). Set to "false" to stop after
# registration and the local vLLM test (for example, on workspaces where entrypoint-based
# Custom Foundation Model serving is not enabled).
DEPLOY_ENDPOINT = dbutils.widgets.get("deploy_endpoint").lower() == "true"

print(f"UC_CATALOG: {UC_CATALOG}")
print(f"UC_SCHEMA: {UC_SCHEMA}")
print(f"UC_MODEL_NAME: {UC_MODEL_NAME_BASE}")
print(f"DEPLOY_ENDPOINT: {DEPLOY_ENDPOINT}")

ライブラリのインポート

ノートブック全体で使用されるライブラリをロードします。AI ランタイム v6 環境にはすでに torch、transformers、trl、および datasets が含まれているため、インストールは不要です。

Python

import torch
import pandas as pd
from datasets import load_dataset, Dataset
import transformers
from transformers import TrainingArguments, AutoTokenizer, AutoModelForCausalLM
from trl import SFTTrainer, SFTConfig

ステップ 1: ベースモデルとトークナイザーをロードする

事前トレーニング済みの Qwen3.5-0.8B-Base をロードしますHugging Face Hub からのチェックポイント。

• アーキテクチャ: Qwen3（デコーダーのみのトランスフォーマー、約 0.8B パラメーター）
• "Base" チェックポイント: 命令チューニングはまだ適用されていません。これは、以下のファインチューニングを行うモデルです。
• 推論とトレーニングの高速化のため、モデルは読み込み直後に GPU に移動されます。

Python

# Qwen3.5-0.8B-Base: ~0.8B parameter decoder-only model.
# "Base" = no instruction-tuning yet; this is the checkpoint fine-tuned below.
model_name = "Qwen/Qwen3.5-0.8B-Base"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(model_name)

Python

# Move all model weights to the GPU for faster inference and training.
model.to("cuda")

トークナイザーを構成する

ベースチェックポイントにはチャット Template が付属していません。トークナイザーがシングルターンおよびマルチターン会話のプロンプトを正しくフォーマットできるように、最小限の System / User / Assistant Jinja Template を定義します。ベース語彙に専用のパディングトークンがないため、pad_token = eos_token を設定します。

Python

# Base checkpoints ship without a chat template.
# Define a minimal System / User / Assistant Jinja template so the tokenizer
# can format both single-turn and multi-turn conversations correctly.
if not tokenizer.chat_template:
    print("No chat template — applying default template")
    tokenizer.chat_template = """{% for message in messages %}
            {% if message['role'] == 'system' %}System: {{ message['content'] }}\n
            {% elif message['role'] == 'user' %}User: {{ message['content'] }}\n
            {% elif message['role'] == 'assistant' %}Assistant: {{ message['content'] }} <|endoftext|>
            {% endif %}
            {% endfor %}"""

# Set pad_token = eos_token because the base vocabulary has no dedicated pad token.
if not tokenizer.pad_token:
    print("No pad token — using eos_token as pad_token")
    tokenizer.pad_token = tokenizer.eos_token

Python

print(tokenizer.chat_template[0:100])
print(tokenizer.pad_token)

ステップ 2: ベースライン推論（SFT 前）

ベース（ファインチューニング未適用）モデル を使用して、簡易的な動作確認（サニティチェック）の推論を実行します。ここでの応答は、基準点として機能します。ステップ5のSFTモデル出力と比較します。

Python

# Build a single-turn chat in OpenAI-style message format.
# The tokenizer's chat template will convert this list into a formatted prompt string.
messages = []
user_message = "Give me a one-sentence introduction to LLMs."
messages.append({"role": "user", "content": user_message})
messages

Python

# Render the message list into a raw text string using the chat template.
# add_generation_prompt=True appends the "Assistant:" prefix to trigger generation.
# enable_thinking=False disables Qwen3's chain-of-thought reasoning mode.
prompt = tokenizer.apply_chat_template(
    messages,
    tokenize=False,
    add_generation_prompt=True,
    enable_thinking=False,
)
prompt

Python

# Tokenize the prompt string and move tensors to the same device as the model.
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
inputs

Python

max_new_tokens = 100
with torch.no_grad():          # no gradient tracking needed during inference
    outputs = model.generate(
        **inputs,              # pass tokenized prompt (input_ids + attention_mask)
        max_new_tokens=max_new_tokens,
        do_sample=False,       # greedy decoding — deterministic output
        pad_token_id=tokenizer.eos_token_id,
        eos_token_id=tokenizer.eos_token_id,
    )

outputs

Python

# Slice off the prompt tokens; decode only the newly generated portion.
input_len = inputs["input_ids"].shape[1]
generated_ids = outputs[0][input_len:]
response = tokenizer.decode(generated_ids, skip_special_tokens=True).strip()
response

ヘルパー関数

このノートブック用に定義された再利用可能なユーティリティ:

• generate_responses：チャットTemplateでプロンプトをフォーマットし、貪欲デコーディングを実行します
• test_model_with_questions: 一連の質問のベンチマークを実行し、モデルの出力を並べて表示します。
• load_model_and_tokenizer: オプションの GPU 配置と Template パッチ適用により、モデルとトークナイザーをロードします
• display_dataset: チャット形式のデータセットの最初の 3 行を読みやすいテーブルとしてレンダリングします

Python

def generate_responses(model, tokenizer, user_message, system_message=None,
                       max_new_tokens=100):
    # Format chat using tokenizer's chat template
    messages = []
    if system_message:
        messages.append({"role": "system", "content": system_message})

    # Assume the data are all single-turn conversations
    messages.append({"role": "user", "content": user_message})

    prompt = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
        enable_thinking=False,
    )

    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    # Recommended to use vllm, sglang or TensorRT
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            do_sample=False,
            pad_token_id=tokenizer.eos_token_id,
            eos_token_id=tokenizer.eos_token_id,
        )
    input_len = inputs["input_ids"].shape[1]
    generated_ids = outputs[0][input_len:]
    response = tokenizer.decode(generated_ids, skip_special_tokens=True).strip()

    return response

Python

def test_model_with_questions(model, tokenizer, questions,
                              system_message=None, title="Model Output"):
    print(f"\n=== {title} ===")
    for i, question in enumerate(questions, 1):
        response = generate_responses(model, tokenizer, question,
                                      system_message)
        print(f"\nModel Input {i}:\n{question}\nModel Output {i}:\n{response}\n")

Python

def load_model_and_tokenizer(model_name, use_gpu = False):

    # Load base model and tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)

    if use_gpu:
        model.to("cuda")

    if not tokenizer.chat_template:
        tokenizer.chat_template = """{% for message in messages %}
                {% if message['role'] == 'system' %}System: {{ message['content'] }}\n
                {% elif message['role'] == 'user' %}User: {{ message['content'] }}\n
                {% elif message['role'] == 'assistant' %}Assistant: {{ message['content'] }} <|endoftext|>
                {% endif %}
                {% endfor %}"""

    # Tokenizer config
    if not tokenizer.pad_token:
        tokenizer.pad_token = tokenizer.eos_token

    return model, tokenizer

Python

def display_dataset(dataset):
    # Visualize the dataset
    rows = []
    for i in range(3):
        example = dataset[i]
        user_msg = next(m['content'] for m in example['messages']
                        if m['role'] == 'user')
        assistant_msg = next(m['content'] for m in example['messages']
                             if m['role'] == 'assistant')
        rows.append({
            'User Prompt': user_msg,
            'Assistant Response': assistant_msg
        })

    # Display as table
    df = pd.DataFrame(rows)
    pd.set_option('display.max_colwidth', None)  # Avoid truncating long strings
    display(df)

ステップ3: トレーニング データセットを読み込む

DeepLearning.AI SFT コースの指示追従型データセットである Hugging Face Hub から banghua/DL-SFT-Dataset を読み込みます。各例は、user と assistant のターンを含む messages リストです。

トレーニング時間を短縮するために、この例では 100 個の例からなるサブセットを使用しています。本格的なファインチューニングに向けて、サブセットのサイズを大きくします。

Python

train_dataset = load_dataset("banghua/DL-SFT-Dataset")['train']

train_dataset=train_dataset.select(range(100))

display_dataset(train_dataset)

ステップ4：SFTファインチューニング

TRL の SFTTrainer を使用して スーパーバイズドファインチューニング を実行します。このデモにおける主要なハイパーパラメーターの選択肢:

パラメーター	Value	注
learning_rate	8e-5	小規模モデルに対する SFT の標準的な開始点
num_train_epochs	1	デモでは1回実行し、実際のトレーニングでは回数を増やしてください。
per_device_train_batch_size	1	Tune with gradient_accumulation_steps
gradient_accumulation_steps	8	有効なバッチサイズ = 1 × 8 = 8
gradient_checkpointing	False	速度のために無効化されています。大規模モデルの VRAM を削減するには有効化してください。

パラメーター	Value	注
learning_rate	8e-5	小規模モデルに対する SFT の標準的な開始点
num_train_epochs	1	デモでは1回実行し、実際のトレーニングでは回数を増やしてください。
per_device_train_batch_size	1	Tune with gradient_accumulation_steps
gradient_accumulation_steps	8	有効なバッチサイズ = 1 × 8 = 8
gradient_checkpointing	False	速度のために無効化されています。大規模モデルの VRAM を削減するには有効化してください。

Python

# SFTConfig is a superset of HuggingFace TrainingArguments with SFT-specific defaults.
sft_config = SFTConfig(
    # --- Training hyperparameters ---
    learning_rate=8e-5,               # standard starting point for SFT on small models
    num_train_epochs=1,               # single pass for demo; increase for real training
    per_device_train_batch_size=1,    # fits H100 VRAM; tune together with gradient_accumulation_steps
    gradient_accumulation_steps=8,    # effective batch size = 1 × 8 = 8
    gradient_checkpointing=False,     # disable for speed; enable to reduce VRAM on larger models
    logging_steps=2,

    # --- Logging ---
    report_to=[],                     # disable W&B / MLflow / etc.
    logging_strategy="steps",
    logging_first_step=True,

    # --- Checkpointing ---
    output_dir="./checkpoints/tiny-finetune-exp1",
    run_name="tiny-finetune-exp1-run",   # must differ from output_dir to avoid W&B conflicts
    save_strategy="no",                  # set to "epoch" to persist a final checkpoint
    save_total_limit=1
)

Python

# SFTTrainer handles dataset formatting, tokenization, and response-label masking
# automatically based on the tokenizer's chat template.
sft_trainer = SFTTrainer(
    model=model,
    args=sft_config,
    train_dataset=train_dataset,
    processing_class=tokenizer    # replaces the deprecated tokenizer= arg in TRL 0.12+
)

# Run training — loss should decrease over the single epoch on this 100-example subset.
sft_trainer.train()

ステップ 5: SFT 後の評価

ステップ 2 で使用した同じ質問に対するファインチューニング済みモデルの応答を比較します。トレーニングデータの分布を反映した、フォーマットの改善や指示追従スタイルを確認します。

Python

questions = [
    "Calculate 1+1-1",
    "What's the difference between thread and process?"
]
test_model_with_questions(sft_trainer.model, tokenizer, questions,
                          title="Fine-tuned model output")

ステップ 6: vLLM でファインチューニング済みモデルをサービングする

トレーニングが完了したら、vLLM OpenAI互換サーバーを実行するDatabricksの カスタム基盤モデル モデルサービングEndpointの背後に、ファインチューニング済みモデルをデプロイします。

知っておくと便利な点: Qwen/Qwen3.5-0.8B-Base は実際には マルチモーダル モデルであり、上記の AutoModelForCausalLM はその テキストバックボーン （Qwen3_5ForCausalLM）のみを読み込んでいます。vLLM（AI env v6）はそのテキストバックボーンをネイティブにサービングできますが、ビジョン・言語モデルのテキスト側の半分であるため、いくつかの調整が必要です:

• コンポジット Qwen3_5Config (vision_config を保持するもの) とモデルの元の processor ファイルをあわせてファインチューニングされた重みを保存します。そうしないと、vLLM の processor がチェックポイントを拒否します。
• 起動時に、vLLM に 画像/動画がゼロ （--limit-mm-per-prompt）であることを伝えて（存在しない）ビジョンタワーに一切触れないようにし、 Triton を介して linear-attention (GDN) カーネルを実行 （--gdn-prefill-backend triton）することで ninja/nvcc での JIT コンパイルが不要になるようにし、サンプリングを ネイティブ torch （VLLM_USE_FLASHINFER_SAMPLER=0）経由でルーティングします。

以下のすべては、同じ Serverless GPU (H100) ＋ AI 環境 v6 セッションで実行されます。

サービングのためにファインチューニング済みモデルを保存する

トレーナーはモデルをメモリ内でのみ保持するため (save_strategy="no")、Python を再起動する前に永続化してください。

1. save_pretrained ファインチューニングされた テキスト ウェイト （model.*という名前）。
2. モデルの元の プロセッサー (preprocessor_config.jsonなど) を保存します。複合構成でビジョンコンポーネントが宣言されているため、vLLM ではそれが要求されます。次に、カスタムチャットTemplateが優先されるように、その上に新しいトークナイザーを再保存します。
3. フラットテキストの設定を 複合 Qwen3_5Config で上書きし、vLLM が（完全な VL モデルではなく）テキストモデルを読み込むように architectures を Qwen3_5ForCausalLM にピン留めします。

sft_trainer、tokenizer、およびmodel_nameがまだスコープ内にある間にこれを実行してください。

Python

import os, tempfile
from transformers import AutoConfig, AutoProcessor

# Local-disk working dir. Use a fixed path (not a random tmpdir) so it survives %restart_python below;
# ARTIFACTS_PATH is a relative basename because the vLLM entrypoint's --model must match it both here
# and inside the packaged model's artifacts/ dir at serving time.
WORKDIR = os.path.join(tempfile.gettempdir(), "sft_serve")
ARTIFACTS_PATH = "qwen3_sft"
SAVE_DIR = os.path.join(WORKDIR, ARTIFACTS_PATH)
os.makedirs(SAVE_DIR, exist_ok=True)

# 1. Fine-tuned text backbone.
sft_trainer.model.save_pretrained(SAVE_DIR)

# 2. Original processor files, then the new tokenizer (with the custom chat template) on top.
try:
    AutoProcessor.from_pretrained(model_name).save_pretrained(SAVE_DIR)
except Exception as e:
    print("processor save skipped:", e)
tokenizer.save_pretrained(SAVE_DIR)

# 3. Composite Qwen3_5Config (text_config + vision_config), arch pinned to the text model.
cfg = AutoConfig.from_pretrained(model_name)
cfg.architectures = ["Qwen3_5ForCausalLM"]
cfg.save_pretrained(SAVE_DIR)

print("saved ->", SAVE_DIR)
print("config architectures:", cfg.architectures, "| model_type:", cfg.model_type,
      "| has vision_config:", hasattr(cfg, "vision_config"))

H100を解放するためにPythonを再起動します

インストールの必要はありません。AI env v6 にはすでに flashinfer とその事前ビルド済みの flashinfer-cubin カーネルが同梱されており、GDN (線形アテンション) のプリフィルは Triton で実行され、サンプリングはネイティブ torch で実行されます (どちらも以下の起動フラグで設定されます)。そのため、Startup時にカーネルの JIT コンパイルは実行されません。

必要な作業は、vLLM が取得できるように、トレーナーが保持している GPU メモリを解放することだけです。%restart_python がそれを実行します。ローカルディスクに保存されたチェックポイントは、再起動後も保持されます (同じドライバーノード)。

Python

%restart_python

サービング構成

次のセルの先頭には、ワークスペース用に設定した値（モデル/Endpoint名、Endpointのサイズなど）が保持されます。残りは、そのままにしておいて構わない内部配線です。登録する/デプロイセルを実行する前に、最小限として、UC_MODEL_NAME 書き込み可能な Unity Catalog パスに設定します。

Python

from databricks.sdk.service.serving import ServingModelWorkloadType

# Re-read the widgets: %restart_python reset the Python process, but the widget
# values set at the top of the notebook persist and can be read again here.
UC_CATALOG = dbutils.widgets.get("uc_catalog")
UC_SCHEMA = dbutils.widgets.get("uc_schema")
UC_MODEL_NAME_BASE = dbutils.widgets.get("uc_model_name")
DEPLOY_ENDPOINT = dbutils.widgets.get("deploy_endpoint").lower() == "true"

# --- Model / endpoint names ---
UC_MODEL_NAME = f"{UC_CATALOG}.{UC_SCHEMA}.{UC_MODEL_NAME_BASE}"   # Unity Catalog catalog.schema.model
ENDPOINT_NAME = f"{UC_MODEL_NAME_BASE}-endpoint"                   # serving endpoint name; unique per workspace
SERVED_MODEL_NAME = UC_MODEL_NAME_BASE                            # name vLLM exposes the model under

# --- Endpoint sizing (adjust if needed) ---
# --gdn-prefill-backend triton (see the entrypoint) JITs the GDN kernels for whatever GPU the pod
# lands on, so this is not pinned to Hopper. GPU_MEDIUM fits the 0.8B model; verify at deploy.
WORKLOAD_TYPE = ServingModelWorkloadType.GPU_MEDIUM
WORKLOAD_SIZE = "Small"
SCALE_TO_ZERO_ENABLED = True

# --- Internal wiring: leave as-is ---
import os, tempfile

# Local-disk working dir; must match the save cell (survives %restart_python; same driver node).
WORKDIR = os.path.join(tempfile.gettempdir(), "sft_serve")
ARTIFACTS_PATH = "qwen3_sft"          # relative basename; entrypoint --model resolves to ./qwen3_sft
os.chdir(WORKDIR)                     # so `--model qwen3_sft` works locally and matches artifacts/ at serving

# Allowlisted ports for serverless GPU notebooks are 3000-3999. Model Serving requires 8080.
LOCAL_PORT = 3080
SERVING_PORT = 8080

# vLLM tuning.
DTYPE = "float16"                     # model is bf16-native; this matches what shipped
MAX_MODEL_LEN = 8192                  # keep <= the model's max_position_embeddings (config.json)
GPU_MEMORY_UTILIZATION = 0.85

vLLM を使用してローカルでファインチューニング済みモデルをテストする

保存したチェックポイントを使用してvLLM OpenAIサーバーを起動し、Endpointをデプロイするのに約40分かける前に、スモークテストを実行します。不適切なチェックポイントやフラグがある場合は、代わりにここで数秒で失敗します。

entrypoint()コマンドは一度定義され、ローカルテスト（LOCAL_PORT）とサービングEndpoint（SERVING_PORT）の両方で再利用されます。同じ文字列がモデルのメタデータに保存され、サービングコンテナ内で再実行されます。

Python

def entrypoint(port: int) -> str:
    args = [
        "VLLM_USE_FLASHINFER_SAMPLER=0",   # native torch sampling; avoids the flashinfer sampler JIT
        "python", "-u", "-m", "vllm.entrypoints.openai.api_server",
        "--model", ARTIFACTS_PATH,
        "--served-model-name", SERVED_MODEL_NAME,
        "--host", "0.0.0.0",
        "--port", str(port),
        "--dtype", DTYPE,
        "--max-model-len", str(MAX_MODEL_LEN),
        "--gpu-memory-utilization", str(GPU_MEMORY_UTILIZATION),
        # Run GDN (linear-attention) prefill through Triton (its own bundled compiler) instead of
        # flashinfer's JIT kernel, so no ninja/nvcc is needed here or in the serving pod.
        "--gdn-prefill-backend", "triton",
        # Text-only serving of a multimodal-config model: allow zero images/videos so vLLM never
        # profiles or exercises the vision tower (absent from the text weights).
        "--limit-mm-per-prompt", "'{\"image\": 0, \"video\": 0}'",
    ]
    return " ".join(args)

Python

# Start the vLLM server in the background; logs stream to process.log.
import subprocess

log = open("process.log", "w")
subprocess.Popen(
    ["bash", "-lc", entrypoint(LOCAL_PORT)],
    stdout=log,
    stderr=subprocess.STDOUT,
    text=True,
    start_new_session=True,
)

Python

%sh
# Tail logs until vLLM is ready. If this hangs, vLLM startup probably hit an error (read process.log).
tail -f process.log | sed -u '/Application startup complete/q'

Python

# Smoke test (sync): the endpoint speaks the OpenAI chat schema at /invocations.
import requests

resp = requests.post(f"http://localhost:{LOCAL_PORT}/invocations", json={"messages": [{"role": "user", "content": "Hello"}]})
resp.json()["choices"][0]["message"]["content"]

Python

# Smoke test (streaming): vLLM streams completions as Server-Sent Events — one JSON chunk per
# `data: ` line, terminated by `data: [DONE]`.
import requests
import json

resp = requests.post(
    f"http://localhost:{LOCAL_PORT}/invocations",
    json={"messages": [{"role": "user", "content": "Tell me a story that is about 300 words!"}], "stream": True},
    stream=True,
)

for line in resp.iter_lines():
    if not line:
        continue
    if line == b"data: [DONE]":
        break
    if line.startswith(b"data: "):
        data = json.loads(line[6:])
        delta = data["choices"][0].get("delta", {})
        if "content" in delta:
            print(delta["content"], end="", flush=True)

Python

%sh
# Stop the local server before logging/registering (the endpoint runs its own copy).
pkill -f vllm.entrypoints.openai.api_server

ステップ 7: Log とモデルを登録する

メタデータにtask = llm/v1/chatとvLLMのentrypointコマンドを含むMLflow ChatModelをログに記録します。そのエントリポイントを実行するサービングランであり、 実行するのではなく python_model.predictです。そのため、クラス本体は必須のプレースホルダーにすぎません。

env_pack="databricks_model_serving" への登録により、カスタム LLM サービングに必要な Serverless Optimized Deployment (SOD) アーティファクト (ウェイト + パッケージ化された環境) が生成されます。正しい GPU 依存関係がパッケージ化されるように、この Serverless GPU ランタイムからロギングを行う必要があります。

Python

import os
import mlflow
from mlflow.pyfunc.model import ChatModel, ChatCompletionResponse

# Required placeholder. Serving runs the entrypoint, not python_model.predict.
class LLMModel(ChatModel):
    def predict(self, context, messages, params):
        return ChatCompletionResponse.from_dict({"choices": []})

# You must log and register from a Serverless GPU runtime, otherwise the model is packaged
# with CPU deps and the GPU serving endpoint fails to start.
if not os.environ.get("DATABRICKS_ACCELERATOR"):
    raise RuntimeError(
        "This model MUST be logged+registered from a serverless GPU runtime, otherwise the correct dependencies will not be packaged for serving."
    )

model_info = mlflow.pyfunc.log_model(
    name=SERVED_MODEL_NAME,
    python_model=LLMModel(),
    artifacts={
        "model_dir": ARTIFACTS_PATH,
    },
    metadata={
        "task": "llm/v1/chat",
        "entrypoint": entrypoint(SERVING_PORT),
    },
    # Pin whatever mlflow ships in AI env v6 (rather than a hardcoded version).
    extra_pip_requirements=[f"mlflow=={mlflow.__version__}"],
)
model_info.model_uri

Python

import mlflow

# env_pack is required. Custom LLM Serving depends on Serverless Optimized Deployments (SOD).
# The endpoint will not work without it.
# https://docs.databricks.com/aws/en/machine-learning/model-serving/serverless-optimized-deployments
model_version = mlflow.register_model(model_info.model_uri, UC_MODEL_NAME, env_pack="databricks_model_serving")

ステップ 8: サービングEndpointを作成する

登録済みバージョンをサービングするカスタム基盤モデル Endpoint の作成create_and_wait は、endpoint の準備が整うまでブロックします (初回デプロイでは最大 40 分かかります。コンテナーが SOD アーティファクトをダウンロードし、vLLM を起動します)。

Deploy the Endpoint (オプション)

残りのステップで、管理対象のサービングEndpointをデプロイします。これらには、エントリーポイントに基づくカスタム基盤モデルサービング (Serverless Optimized Deployment) が必要です。deploy_endpoint が false である場合、ノートブックはここで停止します。

Python

if not DEPLOY_ENDPOINT:
    dbutils.notebook.exit("deploy_endpoint=false: skipping managed serving endpoint deployment")

Python

from databricks.sdk import WorkspaceClient
from datetime import timedelta
from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedEntityInput

served_entities = [
    ServedEntityInput(
        entity_name=UC_MODEL_NAME,
        entity_version=str(model_version.version),
        workload_type=WORKLOAD_TYPE,
        workload_size=WORKLOAD_SIZE,
        scale_to_zero_enabled=SCALE_TO_ZERO_ENABLED,
    )
]

w = WorkspaceClient()

# Create the endpoint, or update it in place if an endpoint of this name already exists,
# so the notebook is safe to re-run. The first deploy can take up to ~40 minutes while the
# container downloads the Serverless Optimized Deployment artifacts and boots vLLM.
existing = {e.name for e in w.serving_endpoints.list()}
if ENDPOINT_NAME in existing:
    print(f"Updating existing endpoint: {ENDPOINT_NAME}")
    w.serving_endpoints.update_config_and_wait(
        name=ENDPOINT_NAME, served_entities=served_entities, timeout=timedelta(minutes=40)
    )
else:
    print(f"Creating endpoint: {ENDPOINT_NAME}")
    config = EndpointCoreConfigInput(name=ENDPOINT_NAME, served_entities=served_entities)
    w.serving_endpoints.create_and_wait(
        name=ENDPOINT_NAME, config=config, timeout=timedelta(minutes=40)
    )

ステップ 9: Endpointをクエリーする

Once the Endpoint is ready, query it three ways: the Databricks SDK, the OpenAI client, and the OpenAI client with streaming.

Python

# Query using the Databricks SDK.
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ChatMessage, ChatMessageRole

w = WorkspaceClient()

resp = w.serving_endpoints.query(
    name=ENDPOINT_NAME,
    messages=[ChatMessage(role=ChatMessageRole.USER, content="Hi, what model are you?")],
)

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

Python

# Query using the OpenAI client.
from openai import OpenAI

DATABRICKS_HOST = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiUrl().get()
DATABRICKS_TOKEN = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()

client = OpenAI(
    api_key=DATABRICKS_TOKEN,
    base_url=f"{DATABRICKS_HOST}/serving-endpoints",
)

response = client.chat.completions.create(
    model=ENDPOINT_NAME,
    messages=[
        {"role": "user", "content": "Hello"},
    ],
)
print(response.choices[0].message.content)

Python

# Query using the OpenAI client (streaming).
stream = client.chat.completions.create(
    model=ENDPOINT_NAME,
    messages=[
        {"role": "user", "content": "Hello, tell me a 200 word story"},
    ],
    stream=True,
)

for event in stream:
    delta = event.choices[0].delta
    print(delta.content, end="|")

次のステップ

モデルのファインチューニング、登録、およびサービングが完了したら、次の操作を実行できます。

• アプリケーションからEndpointをクエリーする : 基盤モデルと外部モデルをクエリーする
• カスタム LLM サービングの詳細 : カスタム LLM の提供
• Serverless GPU の使用量を最適化する : AI ランタイムのベストプラクティス
• 問題のトラブルシューティング : AI ランタイム での問題のトラブルシューティング

ノートブックの例

Qwen3.5-0.8B の教師ありファインチューニング (フル) とサービング

ノートブックを新しいタブで開くOpen in Databricks