Genie One APIを使用する
Genie One APIを使用すると、独自のアプリケーション、バックエンドサービス、またはエージェントがHTTP経由でデータに関する自然言語の質問をGenie Oneに問いかけ、回答をストリームして返してもらうことができます。Genie Oneは、呼び出し元がアクセスできるすべてのデータにわたって機能し、その回答を Genie オントロジー に基づいて構築するため、事前にキュレーションされたエージェントを設定する必要はありません。新しい質問は呼び出し元の Unity Catalog 権限でランされるため、回答には呼び出し元がアクセスできるデータのみが含まれます。保存された応答または会話履歴を読み取ると、保存されたコンテンツが返されます。クエリーを再実行したり、基になるデータの権限を再確認したりすることはありません。
このページでは、認証、質問、回答のストリーム、および会話の管理方法について説明します。完全な要求形式と応答形式については、Genie One API リファレンスを参照してください。
要件
- ワークスペース管理者は、ワークスペースの プレビュー ページ から Genie One API プレビューを有効にする必要があります。
- 呼び出し元はGenie Oneを使用できる必要があります。 コンシューマーアクセス 資格、またはそれ以上のワークスペース資格で十分です。コンシューマーアクセスとは何ですか?を参照してください。
- 呼び出し元は、質問で使用する Unity Catalog データに対する
SELECT権限を持っている必要があります。 - 特定のSQLウェアハウスでクエリーを実行するには、呼び出し元にそのウェアハウスに対する少なくともCAN USE権限が必要です。warehouseを指定しない場合、Genie Oneが選択します。
認証
Genie One APIは、genieスコープのOAuthアクセストークンを受け入れます。
- ユーザーが存在するアプリケーションの場合は、ユーザー用の OAuth (OAuth U2M) を使用します。リクエストはそのユーザーとして実行されるため、回答はユーザー自身の Unity Catalog 権限を反映します。
- 自動化されたワークロードの場合は、Service Principal用OAuth（OAuth M2M）でService Principalを使用してください。Service Principalは、質問で使用するデータおよびSQLウェアハウスへのアクセス権を持っている必要があります。
ワークスペースURLも必要です。このページの例では、それに<workspace-url>を使用しています。ワークスペースオブジェクトの識別子を取得するを参照してください。
質問をして回答をストリームする
POST /api/2.0/genie/one/responsesに質問を送信します。Endpointは、Genie Oneの進行状況と最終回答をServer-Sent Events（SSE）としてストリーム配信するtext/event-stream応答を返します。
会話を起動
新しい会話を起動するには、conversation_idを省略します。inputを質問に設定してください：
- curl
- Python
curl -N -X POST "https://<workspace-url>/api/2.0/genie/one/responses" \
-H "Authorization: Bearer <access-token>" \
-H "Content-Type: application/json" \
-d '{"input": "How did revenue change last quarter?"}'
-Nオプションを使用すると出力バッファリングが無効になり、curlは各イベントを受信するたびに出力します。
次の例では、認証を解決するためだけにDatabricks SDK for Pythonを使用しているため、OAuth U2M、OAuth M2M、または構成プロファイルで機能します。リクエストはrequestsライブラリで送信され、リクエストが受け入れられるとすぐにレスポンスと会話IDがon_createdコールバックに渡され、ストリームが早期に終了した場合はGETリクエストで回復します。その後、最終的な回答を出力します。
この例で使用するパッケージをインストールします：
pip install databricks-sdk requests
import json
import time
import requests
from databricks.sdk import WorkspaceClient
w = WorkspaceClient()
base_url = f"{w.config.host}/api/2.0/genie/one"
terminal_events = ("response.completed", "response.failed", "response.incomplete")
def get_response(conversation_id, response_id, timeout_seconds=600):
# Poll the stored response until it reaches a terminal status.
deadline = time.time() + timeout_seconds
while True:
resp = requests.get(
f"{base_url}/conversations/{conversation_id}/responses/{response_id}",
headers=w.config.authenticate(),
)
resp.raise_for_status()
response = resp.json()
if response["status"] != "in_progress" or time.time() > deadline:
return response
time.sleep(3)
def ask(question, conversation_id=None, on_created=None):
body = {"input": question}
if conversation_id:
body["conversation_id"] = conversation_id
created = None
try:
with requests.post(
f"{base_url}/responses",
headers={**w.config.authenticate(), "Content-Type": "application/json"},
json=body,
stream=True,
) as resp:
resp.raise_for_status()
# The stream is UTF-8, but its content type doesn't declare a charset.
resp.encoding = "utf-8"
for line in resp.iter_lines(decode_unicode=True):
if not line or not line.startswith("data:"):
continue
# SSE allows an optional single space after "data:".
data = line[len("data:"):]
if data.startswith(" "):
data = data[1:]
if data == "[DONE]":
break
event = json.loads(data)
if event["type"] == "response.created":
created = event["response"]
if on_created:
on_created(created["conversation_id"], created["id"])
elif event["type"] in terminal_events:
return event["response"]
except requests.RequestException:
# Without response.created, you can't tell whether the request was accepted.
if created is None:
raise
if created is None:
raise RuntimeError("The stream ended before the response was accepted.")
# The stream ended or disconnected before a terminal event. Recover with GET.
return get_response(created["conversation_id"], created["id"])
response = ask(
"How did revenue change last quarter?",
on_created=lambda conversation_id, response_id: print(
f"conversation_id={conversation_id} response_id={response_id}"
),
)
print(response["status"])
for item in response["output"]:
if item.get("phase") == "final_answer":
# Keep parts separate so Markdown tables and source notes stay on their own lines.
print("\n\n".join(part["text"] for part in item["content"]))
この例では、ターミナルイベントを待機します。Genie Oneの動作中に進行状況を表示するには、イベントストリームの読み取りで説明されているように、出力アイテムイベントも処理します。
イベントストリームを読み取る
各イベントには、 SSE event 名前と、一致する data および増加する type を持つ JSON sequence_number 本体があります。短縮されたストリームは次のようになります。
event: response.created
data: {"type":"response.created","sequence_number":0,"response":{"object":"response","id":"9f7c84c6d3af4ae8ac5cc43590a15366","model":"genie-one","status":"in_progress","output":[],"conversation_id":"6a34b9a50cb4435da7c463130c9d42c1"}}
event: response.output_item.added
data: {"type":"response.output_item.added","sequence_number":1,"output_index":0,"item":{"type":"message","role":"assistant","status":"in_progress","phase":"final_answer","content":[{"type":"output_text","text":"Revenue increased 12% last quarter.","annotations":[]}]}}
event: response.output_item.done
data: {"type":"response.output_item.done","sequence_number":2,"output_index":0,"item":{"type":"message","role":"assistant","status":"completed","phase":"final_answer","content":[{"type":"output_text","text":"Revenue increased 12% last quarter.","annotations":[]}]}}
event: response.completed
data: {"type":"response.completed","sequence_number":3,"response":{"object":"response","id":"9f7c84c6d3af4ae8ac5cc43590a15366","status":"completed","output":[...],"conversation_id":"6a34b9a50cb4435da7c463130c9d42c1"}}
data: [DONE]
イベントを次のように処理します：
response.created: 応答が受け入れられました。response.idとresponse.conversation_idを保存します。フォローアップの質問をしたり、ストリームが切断された場合に回復したりするためにそれらが必要になります。response.output_item.added、response.output_item.updated、response.output_item.done：アシスタントメッセージが追加、変更、または完了しました。各イベントには完全なメッセージが含まれているため、メッセージにテキストを追加するのではなく、output_indexのメッセージを置き換えてください。response.completed、response.failed、またはresponse.incomplete: 応答が終了状態に達しました。response.output配列には、最終的なアシスタントメッセージが保持されます。その後、ストリームはdata: [DONE]で終了します。
各アシスタントメッセージにはphaseがあります。commentaryフェーズのメッセージは、Genie Oneの処理中の進捗状況を示し、ステータス更新として表示できます。final_answerフェーズのメッセージが回答です。回答テキストはMarkdown形式で、結果テーブルを含めることができます。
受け入れ後にレスポンスが失敗した場合、HTTPステータスは引き続き 200 であり、ストリームは response.failed イベントで終了します。response.error オブジェクトは失敗を記述します。トラブルシューティングを参照してください。
フォローアップの質問をする
会話を続けるには、response.createdからconversation_idを渡します。Genie Oneは、以前の質問と回答をコンテキストとして使用します。
curl -N -X POST "https://<workspace-url>/api/2.0/genie/one/responses" \
-H "Authorization: Bearer <access-token>" \
-H "Content-Type: application/json" \
-d '{
"input": "Break that down by region.",
"conversation_id": "<conversation-id>"
}'
また、特定のSQL Warehouseでクエリーを実行するようにwarehouse_idを設定し、user_timezoneをAmerica/Los_AngelesなどのIANAタイムゾーンに設定すると、Genie Oneはユーザーのタイムゾーンで相対日付を解釈できます。
接続断からの回復
ストリームから切断しても応答はキャンセルされません。Genie Oneは動作を継続し、response.createdからIDを使用して結果を取得できます：
curl "https://<workspace-url>/api/2.0/genie/one/conversations/<conversation-id>/responses/<response-id>" \
-H "Authorization: Bearer <access-token>"
このリクエストは、端末のストリームイベントと同じ応答オブジェクトを返します。status が in_progress の間、数秒ごとにEndpointをポーリングします。status が completed、failed、または incomplete になったときに停止します。
復旧するために POST リクエストを再送信しないでください。各 POST リクエストは新しいレスポンスを起動します。
引用を読み取る
各 output_text パートには、Genie One がそのテキストに使用したソースをリストする annotations 配列があります。各注釈には以下が含まれます：
title：ソースの表示名。asset_type:TABLE、NOTEBOOK、WEBなどのソースの種類。identifier：アセットの識別子、ボリュームパス、またはウェブURL。index：回答テキストが参照している1から始まる引用番号。
会話を管理する
Genie One APIを使用して作成した会話は、呼び出し元のユーザーまたはService Principalに属します。それらもGenie Oneのユーザーのチャット履歴に表示されます。
- 会話のリスト ：
GET /api/2.0/genie/one/conversationsは、呼び出し元のGenie Oneの会話を、最近更新されたものから順に返します。Genie Agentの会話は含まれません。page_size(default 20、最大 100) を使用し、next_page_tokenをpage_tokenとして渡すことで次のページを取得します。 - 会話を取得 :
GET /api/2.0/genie/one/conversations/{conversation_id}は、そのメッセージなしで会話のタイトルとTimestampを返します。 - List conversation items :
GET /api/2.0/genie/one/conversations/{conversation_id}/itemsは、会話内のユーザーおよびアシスタントのメッセージを返します。これを使用して、アプリケーションでチャット履歴を再構築します。 - 会話を削除 ：
DELETE /api/2.0/genie/one/conversations/{conversation_id}が会話を削除します。
たとえば、直近の3つの会話をリストするには：
curl "https://<workspace-url>/api/2.0/genie/one/conversations?page_size=3" \
-H "Authorization: Bearer <access-token>"
制限事項
- 各要求は、最大25,000 UTF-8バイトのテキストとして1つの質問を受け入れます。画像、ファイル、および1つのリクエスト内の複数のメッセージはサポートされていません。
- 会話には、アクティブな応答を1つだけ含めることができます。その会話の新しい質問は、アクティブな応答が終了するまでHTTP
409を返します。 - リクエストはワークスペースごとにレート制限されます。制限を超えたリクエストはHTTP
429を返します。
ベストプラクティス
response.createdからIDを保存する ：ストリームが起動するとすぐに応答IDと会話IDを保存します。これにより、接続が切断された場合でも回答を復旧できます。- 指数バックオフを使用して拒否されたリクエストを再試行する : ストリームが起動される前に API が
429などのエラー応答を返した場合は、試行間の遅延を増やして再試行してください。response.createdを受信したら、POSTリクエストを再送信する代わりにGETで応答を取得します。 - タイムアウトやゲートウェイエラーが発生した後に盲目的にリトライしない ：リクエストがタイムアウトしたり、
response.createdが到着する前に502や504などのゲートウェイエラーが返されたりした場合でも、APIはそれを許容している可能性があります。POSTリクエストを再送すると、2回目の応答が作成される可能性があります。 - 再試行可能なエラーの場合に限り、失敗した応答を再試行してください ：応答が失敗した場合は、質問を再度行う前に
response.error.retryableを確認してください。 - セッションごとに新しい会話を開始する ：Genie Oneは以前のやり取りをコンテキストとして使用するため、関連のないセッションで1つの会話を再利用すると精度が低下する可能性があります。
- リクエストとレスポンスをログに記録する ：デバッグと使用状況のモニタリングに役立つよう、レスポンスID、ステータス、エラーをログに記録します。
トラブルシューティング
症状 | 原因と修正 |
|---|---|
| ワークスペースで Genie One API プレビューが無効になっている、会話または回答の ID が正しくない、別のユーザーのもの、または呼び出し元が warehouse を利用できない可能性があります。ワークスペース管理者にワークスペースの「プレビューページ」を確認してもらい、ID と呼び出し元の権限を確認してください。 |
| リクエストに有効なトークンがないか、呼び出し元が会話にレスポンスを追加できません。OAuthトークンとその |
| リクエスト本文に未知のフィールドやサポートされていないオプションが含まれている、メッセージが複数ある、または質問が25,000バイトを超えています。APIリファレンスのフィールドのみを送信してください。 |
| 会話内の別の応答がアクティブです。それが完了するまで待ってから、もう一度お試しください。 |
| ワークスペースがレート制限を超えました。指数関数的バックオフでリトライします。制限事項を参照してください。 |
ストリームは以下で終了します： | 応答は受け入れられた後に失敗しました。 |
端末イベントの前にストリームが切断される |
|
価格
Genie One APIはGenie Oneでのチャットをランするため、リクエストはGenie Oneチャットと同じ方法で請求されます。Genie価格 を参照してください。
その他のリソース
- 作成および取得レスポンスEndpointとイベントストリームのGenie One APIリファレンス：レスポンスを参照してください。
- 会話Endpointについては、Genie One APIリファレンス：Conversationを参照してください。
- Genie One でチャットして、Genie One が質問にどのように回答するかを学びます。
- Genie Agents APIを使用して、1つの厳選されたGenie Agentをクエリーします。
- MCP クライアントまたはエージェントを Genie One に接続するための Genie One MCP サーバー。