2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

あなたはもう試した？判断特化AI「Jev」で遊ぼう！
@isanakamishiro2(Takafumi Yamaya)

JevをDatabricksのUC Connection経由で使ってみる

2
Posted at

Jev、かなり盛り上がってますね。新しさと懐かしさと両方を感じます。

はじめに

TypeSafe AIからJevというモデルがリリースされました。従来のLLMとは異なりテキストを生成するのではなく、構造化された判定結果を確率付きで返す、というものです。実際のビジネスにおいてこういった確率付き分類はよく使う手法なので、高速・高精度のものを製品サービスとして出してきたことはとても興味深いです。

すでに大量の記事やSNS投稿が出ており、ローカルLLMで再実装するものなども散見され、とても面白いことになっています。

そこで、JevをDatabricksのUC Connection（Unity Catalog Connection）経由でガバナンスを効かせながら利用するパターンをいくつか試してみました。この記事ではその検証内容をまとめています。

検証はDatabricks Free Edition上で行っています。

Jevとは

Jevは、TypeSafe AIが開発した「System One Model」と呼ばれる新しいクラスのAIモデルです。従来のLLMがテキストを生成するのに対し、Jevはテキストを生成せず、構造化された型付きの確率的な判定結果を返します。

主な特徴は以下の通りです：

  • テキストを生成しない: ハルシネーションが原理上起こらない（文字列を出力しないため）
  • 3つの質問タイプ: Noul（確率）、Choice（カテゴリ選択）、Score（順序付きスコア）を組み合わせて使う
  • 並列サンプリング: 70〜500msで応答。従来LLMの40〜200倍高速とされている
  • 低コスト: 入力$0.042/MTok、出力トークンは無料
  • キャリブレーション付き: すべての回答に確率と信頼度が付与される

入力として「非構造化データ（テキスト）」と「構造化された質問」を渡すと、1回のリクエストですべての質問に並列で回答してくれます。分類・ルーティング・スコアリングといった、ソフトウェアのif文を賢く置き換えるような用途を想定しているモデルです。

検証環境

  • Databricks Free Edition（AWS）
  • Serverless compute（CPU）
  • Jev: jev-latest（実行時バージョン jev-1.13.0）

今回作るもの

ガバナンスをDatabricksの機能で強化しながら利用する、というパターンを考えます。
エージェントからの少量データ利用やSpark上での各データへの適用という想定です。

これを以下の5つのアプローチで実装してみます：

  1. UC Connection経由のHTTP Request Direct call
  2. Python SDK（typesafe-sdk）
  3. LangChain連携（langchain-typesafe）
  4. PySpark Pandas UDFでのバッチ処理
  5. UC Function（SQL関数）化

すべてのアプローチでUC Connectionを経由することで、APIトークンをDatabricks Secretに格納し、コード内に直接記述しない構成にしています。

Step0. 環境準備

パッケージのインストール

ノートブックを作成し、Python SDKとLangChain連携パッケージをインストールしておきます。
なお、サーバレスの実行環境はStandard v6を利用しました。

%uv pip install typesafe-sdk langchain-typesafe langchain

%restart_python

UC Connectionの作成

JevのAPIトークンはDatabricks Secretに格納済みとします（スコープjev、キーapi_token）。UC Connectionを作成し、このシークレットを参照させます。

CREATE CONNECTION IF NOT EXISTS jev_api TYPE HTTP
OPTIONS (
  host 'https://api.typesafe.ai',
  port '443',
  base_path '/v1/systemone',
  bearer_token secret('jev','api_token')
);

-- base_urlだけを利用するフレームワーク用
CREATE CONNECTION IF NOT EXISTS jev_api_base TYPE HTTP
OPTIONS (
  host 'https://api.typesafe.ai',
  port '443',
  bearer_token secret('jev','api_token')
);

Connectionを2つ作っているのは、用途が異なるためです：

  • jev_api: base_pathに/v1/systemoneを指定。HTTP Request Direct callやUC Functionで使う
  • jev_api_base: base_pathなし。Python SDK・LangChainが自身でパスを組み立てるため使う

UC Connectionを使うことで、APIトークンがコードに露出せず、Unity Catalogでガバナンスを効かせることができます。個人的には、この「認証情報をコードから追い出せる」点が嬉しいところだと思います。

Step1. HTTP Request Direct call

まずは一番シンプルな方法です。Databricks SDKのWorkspaceClientを使って、UC Connectionのプロキシエンドポイント経由でJev APIを呼び出します。

from databricks.sdk import WorkspaceClient
from typing import BinaryIO

w = WorkspaceClient(
    config=Config(
        http_timeout_seconds=30,
        retry_timeout_seconds=120,
    )
)


def proxy(conn: str, path: str = "") -> str:
    stripped = path.lstrip("/")
    if stripped:
        return f"/api/2.0/unity-catalog/connections/{conn}/proxy/{stripped}"
    return f"/api/2.0/unity-catalog/connections/{conn}/proxy"


def request_jev_api(json: dict) -> dict | list | BinaryIO:
    return w.api_client.do(
        method="POST",
        path=proxy("jev_api"),
        body=json,
        headers={"Content-Type": "application/json"},
    )


json = {
    "state": "Help! My payouts have been failing for 3 days.",
    "model": "jev-latest",
    "questions": {
        "is_urgent": {"type": "noul", "instructions": "Does this convey urgency?"}
    },
}

request_jev_api(json)

実行結果：

{'model': 'jev-1.13.0',
 'answers': {'is_urgent': {'type': 'noul', 'noul': 0.95}},
 'usage': {'input_tokens': 283, 'output_tokens': 23}}

「3日間payoutsが失敗している」というテキストに対して、緊急性0.95が返ってきました。妥当ですね。

ポイントは、WorkspaceClientの認証情報を使ってUC Connectionのプロキシ経由でリクエストを投げている点です。APIトークンはUC Connectionに設定済みなので、コード内には一切出てきません。

Step2. Python SDK

次に、TypeSafe AI公式のPython SDK（typesafe-sdk）を使ってみます。SDKはNoul・Choice・Scoreという型安全なクラスを提供しているので、質問の定義がより宣言的になります。

SDKはbase_urlを自分で組み立てる必要があるので、UC ConnectionのプロキシURLを渡します。その際、SDK内部でAPIパスを付与するため、base_pathなしのjev_api_baseコネクションを使います。

from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul, Score
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
token = w.config.authenticate()["Authorization"].split(" ", 1)[1]

def base_url(connection: str, client: WorkspaceClient | None = None) -> str:
    w = client or WorkspaceClient()
    host = w.config.host.rstrip("/")
    return f"{host}/api/2.0/unity-catalog/connections/{connection}/proxy"


async def main() -> None:
    async with AsyncTypeSafeClient(
        api_key=token, base_url=base_url("jev_api_base", w)
    ) as client:
        response = await client.system_one(
            state={"document": "I was charged twice. Please fix this ASAP."},
            questions={
                "billing": Noul(instructions="Is this ticket about billing?"),
                "tone": Choice(
                    instructions="What is the customer's tone?",
                    criteria={"calm": None, "frustrated": None, "angry": None},
                ),
                "urgency": Score(
                    instructions="How urgent is this ticket?",
                    criteria=["can wait", "this week", "today"],
                ),
            },
        )

    print(response.nouls["billing"].noul)
    print(response.choices["tone"].choice)
    print(response.scores["urgency"].score)


await main()

実行結果：

0.98
frustrated
1.99

「二重請求でASAP」というテキストに対して、billing=0.98、tone=frustrated、urgency=1.99（ほぼ「today」）が返ってきました。SDKの型安全なクラスのおかげで、質問の定義が読みやすいですね。

SDKのapi_keyにはJevのAPIトークンではなく、Databricksの認証トークンを渡しています。リクエストはUC Connectionのプロキシ経由で飛ぶので、JevのAPIトークンはUC Connectionが内部で付与してくれます。

Step3. LangChain

LangChain連携パッケージ（langchain-typesafe）を使うと、TypeSafeClassifierとしてLangChainのエコシステムに組み込めます。invokeメソッドでdictを渡すだけで結果が返ってくるので、LangChainのチェーンやパイプラインに自然に統合できます。

from langchain_typesafe import Choice, Noul, Score, TypeSafeClassifier

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
token = w.config.authenticate()["Authorization"].split(" ", 1)[1]

def base_url(connection: str, client: WorkspaceClient | None = None) -> str:
    w = client or WorkspaceClient()
    host = w.config.host.rstrip("/")
    return f"{host}/api/2.0/unity-catalog/connections/{connection}/proxy"

classifier = TypeSafeClassifier(api_key=token, base_url=base_url("jev_api_base", w))

response = classifier.invoke(
    {
        "state": (
            "The deploy failed twice and customers are seeing 500s. "
            "Can someone look now?"
        ),
        "questions": {
            "urgent": Noul(instructions="Does this need attention right now?"),
            "team": Choice(
                instructions="Which team should pick this up?",
                criteria={
                    "infra": "Deploys, availability, and on-call incidents.",
                    "billing": "Payments, invoices, and subscriptions.",
                },
            ),
            "severity": Score(
                instructions="How severe is the impact?",
                criteria=["Cosmetic.", "Degraded for some users.", "Full outage."],
            ),
        },
    }
)

print(response.nouls["urgent"].noul)
print(response.choices["team"].choice, response.choices["team"].confidence)
print(response.scores["severity"].score)

実行結果：

0.97
infra 1.0
1.62

「デプロイが2回失敗して500エラーが出ている」というテキストに対して、urgent=0.97、team=infra（信頼度1.0）、severity=1.62が返ってきました。infraチームに振り分けるべき緊急の事象、と正しく判定されています。

Step4. PySparkでの利用（Pandas UDF）

ここまでのアプローチは単発リクエストでしたが、実運用では大量のデータをバッチ処理したいケースが多いはずです。PySparkのPandas UDFを使って、DataFrameのカラムに対してJevの判定を適用してみます。

UDFの定義

Pandas UDFの中でWorkspaceClientを初期化し、UC Connection経由でJev APIを呼び出します。Driverの認証情報をexecutor側に引き渡す必要があるため、hostとtokenをクロージャで渡しています。

ここでは簡易的にDriverの認証トークンをexecutorに渡していますが、本番環境ではService Principal（Client Secret）を使って、各executorが独立して認証できるようにするのが望ましいです。

import json
from typing import BinaryIO, Iterator

import pandas as pd
from databricks.sdk import WorkspaceClient
from databricks.sdk.core import Config
from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf

MODEL = "jev-latest"

QUESTIONS = {
    "department": {
        "type": "choice",
        "instructions": "この問い合わせを担当すべきチームはどこか",
        "criteria": {
            "billing": "支払いやサブスクリプションに関する問題",
            "technical": "不具合や連携（インテグレーション）の問題",
            "sales": "料金プランやアカウントに関する質問",
        },
    },
    "frustration": {
        "type": "score",
        "instructions": "顧客がどの程度不満を感じているように見えるか",
        "criteria": [
            "落ち着いており、事実を述べているだけ",
            "不満はあるが丁寧な言葉遣い",
            "強い怒りがあり、きつい言葉を使っている",
        ],
    },
    "is_urgent": {
        "type": "noul",
        "instructions": "メッセージから緊急性や時間的な切迫感が伝わるか",
    },
}


def proxy(conn: str, path: str = "") -> str:
    stripped = path.lstrip("/")
    if stripped:
        return f"/api/2.0/unity-catalog/connections/{conn}/proxy/{stripped}"
    return f"/api/2.0/unity-catalog/connections/{conn}/proxy"

# Driverの認証情報をpandas_udf側に引き渡し
# 本番のときはちゃんとClient Secretを使うように。
w = WorkspaceClient()
host = w.config.host
token = w.config.authenticate()["Authorization"].removeprefix("Bearer ")


@pandas_udf("string")
def jev(states: Iterator[pd.Series]) -> Iterator[pd.Series]:
    w = WorkspaceClient(
        host=host,
        token=token,
    )

    def call(state: str | None) -> str | None:
        if state is None:
            return None
        try:
            body = {"model": MODEL, "state": state, "questions": QUESTIONS}
            res = w.api_client.do(method="POST", path=proxy("jev_api"), body=body)
        except Exception as e:
            res = {"error": f"{type(e).__name__}: {e}"}
        return json.dumps(res, ensure_ascii=False)

    for batch in states:
        yield batch.map(call)

サンプルデータで実行

3パターンの日本語問い合わせを用意して、Pandas UDFを適用します。

sample = spark.createDataFrame(
    [
        # technical / 不満だが丁寧 / 緊急
        (
            1,
            "Stripeアカウントの連携を3日間試しているのですが、何度やってもエラーになります。"
            "売上に影響が出ています。至急対応をお願いします。",
        ),
        # sales / 落ち着いている / 緊急でない
        (
            2,
            "現在スタンダードプランを利用しています。来期からチームが10名ほど増える予定なので、"
            "エンタープライズプランとの料金の違いを教えていただけますか。",
        ),
        # billing / 強い怒り / 緊急
        (
            3,
            "また今月も二重請求されています！先月も同じ問題を報告したのに、まったく直っていません。"
            "今日中に返金しないなら解約します。",
        ),
    ],
    "inquiry_id INT, body STRING",
)

result = (
    sample.repartition(1)
    .withColumn("jev", jev("body"))
    .withColumn("jev", F.try_parse_json("jev"))
)
display(result)

実行結果（jevカラムの内容を展開）：

inquiry_id body（抜粋） department frustration is_urgent
1 Stripe連携が3日間エラー…至急対応を technical (0.99) 0.99（不満だが丁寧） 0.96
2 スタンダードプラン…エンタープライズとの比較 sales (0.97) 0.00（落ち着いている） 0.32
3 また二重請求！…解約します billing (1.00) 1.85（強い怒り） 0.97

日本語の問い合わせでも正しく判定されていますね。frustrationのスコアも、丁寧な問い合わせは0.99（不満はあるが丁寧）、冷静な質問は0.0（落ち着いている）、怒りの問い合わせは1.85（強い怒り）と、期待通りの結果です。

Step5. UC Functionとして利用

最後に、SQL関数（UC Function）として登録してみます。http_request関数とUC Connectionを組み合わせることで、SQLから直接Jevを呼び出せるようになります。

CREATE OR REPLACE FUNCTION workspace.default.jev_triage(
  state STRING COMMENT '判定対象の問い合わせ本文'
)
RETURNS VARIANT
COMMENT 'TypeSafe Jev で問い合わせを判定する（担当チーム・不満度・緊急度）。失敗時は {"error": ...} を返す。'
RETURN (
  SELECT
    CASE
      WHEN state IS NULL THEN NULL
      WHEN r.status_code = 200 THEN try_parse_json(r.text)
      ELSE parse_json(to_json(named_struct(
        'error', named_struct('status_code', r.status_code, 'body', r.text)
      )))
    END
  FROM (
    SELECT http_request(
      conn   => 'jev_api',
      method => 'POST',
      path   => '',
      json   => to_json(named_struct(
        'model', 'jev-latest',
        'state', state,
        'questions', parse_json('{
          "department": {
            "type": "choice",
            "instructions": "この問い合わせを担当すべきチームはどこか",
            "criteria": {
              "billing": "支払いやサブスクリプションに関する問題",
              "technical": "不具合や連携（インテグレーション）の問題",
              "sales": "料金プランやアカウントに関する質問"
            }
          },
          "frustration": {
            "type": "score",
            "instructions": "顧客がどの程度不満を感じているように見えるか",
            "criteria": [
              "落ち着いており、事実を述べているだけ",
              "不満はあるが丁寧な言葉遣い",
              "強い怒りがあり、きつい言葉を使っている"
            ]
          },
          "is_urgent": {
            "type": "noul",
            "instructions": "メッセージから緊急性や時間的な切迫感が伝わるか"
          }
        }')
      ))
    ) AS r
  )
);

UC Function化しておくと、SQLクエリから workspace.default.jev_triage(body) と呼ぶだけでJevの判定が使えます。ダッシュボードやSQLアラートからも利用できるので、個人的には一番実運用に近いアプローチだと思います。

実行確認：

sample = spark.createDataFrame(
    [
        (1, "Stripeアカウントの連携を3日間試しているのですが、何度やってもエラーになります。"
             "売上に影響が出ています。至急対応をお願いします。"),
        (2, "現在スタンダードプランを利用しています。来期からチームが10名ほど増える予定なので、"
             "エンタープライズプランとの料金の違いを教えていただけますか。"),
        (3, "また今月も二重請求されています！先月も同じ問題を報告したのに、まったく直っていません。"
             "今日中に返金しないなら解約します。"),
    ],
    "inquiry_id INT, body STRING",
)

result = (
    sample.repartition(1)
    .withColumn("jev", F.expr("workspace.default.jev_triage(body)"))
)
display(result)

実行結果はStep4のPandas UDFと同じく、3件とも正しく判定されました。SQL関数経由でも日本語テキストの分類・スコアリングが問題なく動いています。

まとめ

JevをDatabricksのUC Connection経由で使ってみました。5つのアプローチ（HTTP Direct call / Python SDK / LangChain / Pandas UDF / UC Function）を試しましたが、どれもUC Connection経由で利用できます。

個人的には、Jevの「テキストを生成せず構造化された判定だけを返す」というアプローチは、ソフトウェアに組み込むAIとして非常に理にかなっていると思います。
少し前で言えばBART、最近もLLMの構造化出力でこのあたりを行っていましたが現実的な選択肢となるサービスが出たなという印象です。（真似されやすいと思うので、今後類似のものも多く出そうですが・・・）

とはいえ、まだ入力がテキストのみという制約もあるので、ユースケースの選定はまだちょっとしぼられそうですね。マルチモーダル化もすぐな気はしていますが。

UC Connection × Jevの組み合わせは、Databricks上でガバナンスを効かせながらAI判定を組み込めるので、継続的に検証していきたいと思います。

2
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?