はじめに
Oracle Select AI for Python は、Oracle Autonomous Database（および Oracle AI Database）上の Select AI 機能を Python から利用するためのパッケージです。バージョン 1.4 では、アプリケーション開発に向け、ストリーミング・CLI・Web フレームワーク連携・並列プロンプト処理が追加されています。
インストールし追加された機能のいくつかを動作確認してみました。
バージョン 1.4 の主な新機能は以下の通りです。
| 機能 | 概要 |
|---|---|
| ストリーミングチャット・生成 |
generate()、chat()、narrate() 等でチャンク単位のレスポンス受信が可能（run_sql() は除く） |
| コマンドラインインターフェース（CLI） | ターミナルからインタラクティブチャット・SQL 生成・実行・説明等を実行可能 |
| Python Web フレームワーク連携 | FastAPI・Flask・Django・Starlette・Quart・Sanic 向けの接続プールパターンを提供 |
| 並列プロンプト処理 | スレッドプール・キュー・asyncio・run_pipeline() による並列処理パターンを提供 |
| ネットワークアクセス管理 | Python から DB の ACL 設定（外部エンドポイントへのアクセス許可・取消し）が可能 |
| xAI などカスタムプロバイダーの例 | 専用クラスのない互換エンドポイントをカスタムプロバイダーとして利用可能 |
| エージェントチームのエクスポート・インポート | チーム定義を異なる DB や環境間で移植可能なスペックとして管理 |
事前準備
前提条件
- Python 環境
- Select AI Python API(select_aiモジュール)は、Pythonバージョン3.9、3.10、3.11、3.12、3.13および3.14をサポートしています。
- PIP
- uv 環境での準備
$ curl -LsSf https://astral.sh/uv/install.sh | sh
downloading uv 0.9.15 x86_64-unknown-linux-gnu
no checksums to verify
installing to /home/opc/.local/bin
uv
uvx
everything's installed!
$ mkdir /home/opc/selectai1.4
$ cd /home/opc/selectai1.4
$ uv venv --python 3.14
Using CPython 3.14.3
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate
$ source .venv/bin/activate
$ python3 -m ensurepip --upgrade
Looking in links: /tmp/tmp87bvv43v
Processing /tmp/tmp87bvv43v/pip-25.3-py3-none-any.whl
Installing collected packages: pip
Successfully installed pip-25.3
-- C++ コンパイラをインストール
$ ssudo dnf install -y gcc gcc-c++ make
- Oracle Autonomous Database または Oracle AI Database が利用可能であること
- Autonomous Databaseに接続用のtnsnames.oraやウォレット
- 対象データベースに Select AI が設定済みであること（プロファイルが作成済みであること）
- 今回の動作確認は、「Select AI for Python v1.2.0」を使ってみた。の記事の内容で作成しています。
- 以下の環境変数が設定されていること
export SELECT_AI_ADMIN_USER=admin
export SELECT_AI_ADMIN_PASSWORD=<PASSWORD>
export SELECT_AI_USER=<データベースユーザー名>
export SELECT_AI_PASSWORD=<データベースパスワード>
export SELECT_AI_DB_CONNECT_STRING=<データベース接続文字列>
export TNS_ADMIN=/home/opc/wallet
export SELECT_AI_WALLET_LOCATION=/home/opc/wallet
export SELECT_AI_WALLET_PASSWORD=<ウォレットパスワード>
- Web アプリケーション例を試す場合は FastAPI および uvicorn がインストール済みであること
python -m pip install fastapi uvicorn
実行例）
< 省略 >
Installing collected packages: typing-inspection, pydantic-core, idna, h11, annotated-types, annotated-doc, uvicorn, pydantic, anyio, starlette, fastapi
Successfully installed annotated-doc-0.0.4 annotated-types-0.7.0 anyio-4.14.0 fastapi-0.137.1 h11-0.16.0 idna-3.18 pydantic-2.13.4 pydantic-core-2.46.4 starlette-1.3.1 typing-inspection-0.4.2 uvicorn-0.49.0
手順
Step 1: パッケージのインストール
まず select_ai パッケージをインストールまたはアップグレードします。
CLI 機能も使用する場合は CLI オプション付きでインストールします。
python -m pip install --upgrade "select_ai[cli]"
実行例）
$ python3 -m pip install --upgrade "select_ai[cli]"
Collecting select_ai[cli]
Using cached select_ai-1.4.0-py3-none-any.whl.metadata (5.0 kB)
Collecting oracledb (from select_ai[cli])
Using cached oracledb-4.0.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl.metadata (7.8 kB)
Collecting pandas==2.2.3 (from select_ai[cli])
Using cached pandas-2.2.3.tar.gz (4.4 MB)
Installing build dependencies ... done
Getting requirements to build wheel ... done
Installing backend dependencies ... done
Preparing metadata (pyproject.toml) ... \
done
Collecting click (from select_ai[cli])
Downloading click-8.4.1-py3-none-any.whl.metadata (2.6 kB)
Collecting numpy>=1.26.0 (from pandas==2.2.3->select_ai[cli])
Using cached numpy-2.4.6-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl.metadata (6.6 kB)
Collecting python-dateutil>=2.8.2 (from pandas==2.2.3->select_ai[cli])
Downloading python_dateutil-2.9.0.post0-py2.py3-none-any.whl.metadata (8.4 kB)
Collecting pytz>=2020.1 (from pandas==2.2.3->select_ai[cli])
Downloading pytz-2026.2-py2.py3-none-any.whl.metadata (22 kB)
Collecting tzdata>=2022.7 (from pandas==2.2.3->select_ai[cli])
Downloading tzdata-2026.2-py2.py3-none-any.whl.metadata (1.4 kB)
Collecting six>=1.5 (from python-dateutil>=2.8.2->pandas==2.2.3->select_ai[cli])
Downloading six-1.17.0-py2.py3-none-any.whl.metadata (1.7 kB)
Collecting cryptography>=3.2.1 (from oracledb->select_ai[cli])
Downloading cryptography-49.0.0-cp311-abi3-manylinux_2_28_x86_64.whl.metadata (4.3 kB)
Collecting typing_extensions>=4.14.0 (from oracledb->select_ai[cli])
Downloading typing_extensions-4.15.0-py3-none-any.whl.metadata (3.3 kB)
Collecting cffi>=2.0.0 (from cryptography>=3.2.1->oracledb->select_ai[cli])
Downloading cffi-2.0.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl.metadata (2.6 kB)
Collecting pycparser (from cffi>=2.0.0->cryptography>=3.2.1->oracledb->select_ai[cli])
Downloading pycparser-3.0-py3-none-any.whl.metadata (8.2 kB)
Downloading select_ai-1.4.0-py3-none-any.whl (62 kB)
Using cached numpy-2.4.6-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl (16.6 MB)
Downloading python_dateutil-2.9.0.post0-py2.py3-none-any.whl (229 kB)
Downloading pytz-2026.2-py2.py3-none-any.whl (510 kB)
Downloading six-1.17.0-py2.py3-none-any.whl (11 kB)
Downloading tzdata-2026.2-py2.py3-none-any.whl (349 kB)
Downloading click-8.4.1-py3-none-any.whl (116 kB)
Downloading oracledb-4.0.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl (2.5 MB)
qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq 2.5/2.5 MB 74.7 MB/s 0:00:00
Downloading cryptography-49.0.0-cp311-abi3-manylinux_2_28_x86_64.whl (4.7 MB)
qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq 4.7/4.7 MB 130.0 MB/s 0:00:00
Downloading cffi-2.0.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl (219 kB)
Downloading typing_extensions-4.15.0-py3-none-any.whl (44 kB)
Downloading pycparser-3.0-py3-none-any.whl (48 kB)
Building wheels for collected packages: pandas
Building wheel for pandas (pyproject.toml) ... done
Created wheel for pandas: filename=pandas-2.2.3-cp314-cp314-linux_x86_64.whl size=44585468 sha256=11190031c8a6d241d56c276b8d133bfb8f076cd938643628bd22b00c59cdb994
Stored in directory: /home/opc/.cache/pip/wheels/2d/2f/71/b3d0040a34711301eba6154281563ea5ff2aafd8e8023836ff
Successfully built pandas
Installing collected packages: pytz, tzdata, typing_extensions, six, pycparser, numpy, click, python-dateutil, cffi, pandas, cryptography, oracledb, select_ai
Successfully installed cffi-2.0.0 click-8.4.1 cryptography-49.0.0 numpy-2.4.6 oracledb-4.0.1 pandas-2.2.3 pycparser-3.0 python-dateutil-2.9.0.post0 pytz-2026.2 select_ai-1.4.0 six-1.17.0 typing_extensions-4.15.0 tzdata-2026.2
Step 2: ストリーミングチャットを試す
基本的な接続
環境変数を読み込み、データベースに接続してプロファイルを取得します。
import os
import select_ai
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
wallet_location = os.getenv("SELECT_AI_WALLET_LOCATION")
wallet_password = os.getenv("SELECT_AI_WALLET_PASSWORD")
select_ai.connect(
user=user,
password=password,
dsn=dsn,
wallet_location=wallet_location,
wallet_password=wallet_password,
)
profile = select_ai.Profile(profile_name="oci_ai_profile")
ストリーミングチャットの実行
stream=True と chunk_size を指定してチャットを呼び出します。レスポンスはチャンク単位で順次受信できます。
for chunk in profile.chat(
prompt="Explain Oracle AI Database in simple terms.",
stream=True,
chunk_size=4096,
):
print(chunk, end="")
print()
実行例）
>>> for chunk in profile.chat(
... prompt="Explain Oracle AI Database in simple terms.",
... stream=True,
... chunk_size=4096,
... ):
... print(chunk, end="")
...
... print()
...
**Oracle AI Database: Simplified Explanation**
The Oracle AI Database is a type of database management system that combines traditional database capabilities with artificial intelligence (AI) and machine learning (ML) technologies. Here's a breakdown of what it does:
**Key Features:**
1. **Data Storage**: Like a traditional database, it stores and manages large amounts of data.
2. **AI-Powered Analytics**: It uses AI and ML algorithms to analyze data, identify patterns, and make predictions.
3. **Autonomous Operations**: The database can automatically tune, patch, and secure itself, reducing manual labor and improving performance.
4. **Real-Time Insights**: It provides real-time analytics and insights, enabling businesses to make informed decisions quickly.
**How it Works:**
1. **Data Ingestion**: Data is loaded into the database from various sources.
2. **AI-Driven Analysis**: The database applies AI and ML algorithms to the data to identify trends, patterns, and anomalies.
3. **Insight Generation**: The database generates insights and predictions based on the analysis.
4. **Actionable Recommendations**: The database provides recommendations for business decisions, such as optimizing operations or identifying new opportunities.
**Benefits:**
1. **Improved Decision-Making**: The Oracle AI Database enables businesses to make data-driven decisions with real-time insights.
2. **Increased Efficiency**: Autonomous operations reduce manual labor and improve database performance.
3. **Enhanced Security**: The database's AI-powered security features help detect and prevent threats.
**In Simple Terms:**
The Oracle AI Database is like a super-smart, self-managing data repository that helps businesses make better decisions by analyzing data, identifying patterns, and providing real-time insights. It's like having a team of data analysts and IT experts working together to optimize your business operations, but with the power of AI and automation.
その他のアクションでのストリーミング
generate() 等のテキストを返す API でも同様に stream=True を指定できます（run_sql() は DataFrame を返すため非対応です）。
for chunk in profile.generate(
prompt="Show the SQL for total sales by channel.",
action=select_ai.Action.SHOWSQL,
stream=True,
chunk_size=4096,
):
print(chunk, end="")
実行例）
>>> for chunk in profile.generate(
... prompt="Show the SQL for total sales by channel.",
... action=select_ai.Action.SHOWSQL,
... stream=True,
... chunk_size=4096,
... ):
... print(chunk, end="")
...
SELECT
chn."CHANNEL_DESC" AS channel_description,
SUM(sls."AMOUNT_SOLD") AS total_sales
FROM
"SH"."SALES" sls
JOIN "SH"."CHANNELS" chn ON sls."CHANNEL_ID" = chn."CHANNEL_ID"
GROUP BY
chn."CHANNEL_DESC"
Step 3: CLI を使ってみる
環境変数の設定
CLI を使用する前に、事前準備に記載されている接続情報に関する環境変数に設定がされていることを確認します。
インタラクティブチャットセッション
既存のプロファイルを使って対話形式のチャットセッションを開始します。
select-ai chat --profile OCI_AI_PROFILE
実行例）
$ select-ai chat --profile OCI_AI_PROFILE
Connected to Select AI profile: OCI_AI_PROFILE
Type /help for commands. Type /exit to quit.
select_ai> /exit
SQL 操作コマンド
ターミナルから SQL 生成・実行・説明・ナレーションをワンライナーで実行できます。
# SQL の生成（実行なし）
select-ai sql show --profile OCI_AI_PROFILE "count orders by status"
# SQL の実行
select-ai sql run --profile OCI_AI_PROFILE "show revenue by month"
# SQL の説明
select-ai sql explain --profile OCI_AI_PROFILE "which customers had the highest sales?"
# ナレーション（自然言語での結果説明）
select-ai sql narrate --profile OCI_AI_PROFILE "summarize sales by region"
実行例[sql show]）
$ select-ai sql show --profile OCI_AI_PROFILE "count orders by status"
SELECT
COUNT("sa"."PROD_ID") AS "ORDER_COUNT"
FROM
"SH"."SALES" "sa"
WHERE
UPPER("sa"."CHANNEL_ID") = UPPER('ORDERED')
実行例[sql run]）
$ select-ai sql run --profile OCI_AI_PROFILE "show revenue by month"
Revenue Month
2.164540e+09 1998-01
2.006136e+09 1998-04
1.586960e+09 1998-07
1.622070e+09 1998-08
1.766133e+09 1998-10
1.416511e+09 1998-12
実行例[sql explain]）
$ select-ai sql explain --profile OCI_AI_PROFILE "which customers had the highest sales?"
To find the customers with the highest sales, we need to join the "SH"."SALES" table with the "SH"."CUSTOMERS" table and then group the results by customer. Here's the Oracle SQL query:
sql
SELECT
c."CUST_ID",
c."CUST_FIRST_NAME",
c."CUST_LAST_NAME",
SUM(s."AMOUNT_SOLD") AS "TOTAL_SALES"
FROM
"SH"."SALES" s
JOIN "SH"."CUSTOMERS" c ON s."CUST_ID" = c."CUST_ID"
GROUP BY
c."CUST_ID", c."CUST_FIRST_NAME", c."CUST_LAST_NAME"
ORDER BY
"TOTAL_SALES" DESC;
This query works as follows:
1. We join the "SH"."SALES" table with the "SH"."CUSTOMERS" table on the "CUST_ID" column.
2. We group the results by customer using the "CUST_ID", "CUST_FIRST_NAME", and "CUST_LAST_NAME" columns.
3. We calculate the total sales for each customer using the SUM function on the "AMOUNT_SOLD" column.
4. We order the results in descending order by total sales.
However, this query will return all customers with their total sales. If we want to find the customers with the highest sales, we need to use a subquery or a window function to get the maximum total sales. Here's an example using a subquery:
sql
SELECT
c."CUST_ID",
c."CUST_FIRST_NAME",
c."CUST_LAST_NAME",
SUM(s."AMOUNT_SOLD") AS "TOTAL_SALES"
FROM
"SH"."SALES" s
JOIN "SH"."CUSTOMERS" c ON s."CUST_ID" = c."CUST_ID"
GROUP BY
c."CUST_ID", c."CUST_FIRST_NAME", c."CUST_LAST_NAME"
HAVING
SUM(s."AMOUNT_SOLD") = (
SELECT
MAX("TOTAL_SALES")
FROM (
SELECT
SUM(s."AMOUNT_SOLD") AS "TOTAL_SALES"
FROM
"SH"."SALES" s
JOIN "SH"."CUSTOMERS" c ON s."CUST_ID" = c."CUST_ID"
GROUP BY
c."CUST_ID", c."CUST_FIRST_NAME", c."CUST_LAST_NAME"
)
);
This query works as follows:
1. We use a subquery to calculate the maximum total sales.
2. We use the HAVING clause to filter the results to only include customers with the maximum total sales.
Note that if there are multiple customers with the same highest total sales, this query will return all of them.
実行例[sql narrate]）
$ select-ai sql narrate --profile OCI_AI_PROFILE "summarize sales by region"
The sales are broken down by region. Here are the regions and their total sales:
* Americas had the highest sales
* Asia had moderate sales
* Europe had significant sales
* Oceania had lower sales
* Middle East had the lowest sales
プロファイル管理・要約・翻訳コマンド
プロファイル一覧の表示
select-ai profile list
実行例）
$ select-ai profile list
OCI_AI_PROFILE
ファイルの内容を要約
select-ai profile summarize \
--profile OCI_AI_PROFILE \
--file notes.txt
実行例）
$ select-ai profile summarize \
> --profile OCI_AI_PROFILE \
> --file notes.txt
サティア ナデラは、マイクロソフト コーポレーションの会長 兼 最高経営責任者です。1992 年にマイクロソフトに入社し、テクノロジとビジネスをつなぐリーダーの一人として知られました
テキストの翻訳
select-ai profile translate \
--profile OCI_AI_PROFILE \
--source-language English \
--target-language German \
"Thank you"
Step 4: FastAPI Web アプリケーションとの連携
同期プールを使ったパターン
Web アプリケーションでは、リクエストのたびに新規接続を作らず、アプリ起動時にコネクションプールを 1 つ作成することを推奨します。本番環境では DB パスワードやプロバイダークレデンシャルをシークレットマネージャーまたは環境変数で管理してください。
main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
import select_ai
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
@asynccontextmanager
async def lifespan(app: FastAPI):
select_ai.create_pool(
user=user,
password=password,
dsn=dsn,
min_size=5,
max_size=10,
increment=5,
)
yield
select_ai.disconnect()
app = FastAPI(lifespan=lifespan)
@app.get("/chat")
def chat(prompt: str):
profile = select_ai.Profile(profile_name="oci_ai_profile")
return {"response": profile.chat(prompt=prompt)}
@app.get("/show_sql")
def show_sql(prompt: str):
profile = select_ai.Profile(profile_name="oci_ai_profile")
return {"sql": profile.show_sql(prompt=prompt)}
非同期プールを使ったパターン
async ルートハンドラーには、非同期プール（create_pool_async）と AsyncProfile を組み合わせます。
main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
import select_ai
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
@asynccontextmanager
async def lifespan(app: FastAPI):
select_ai.create_pool_async(
user=user,
password=password,
dsn=dsn,
min_size=5,
max_size=10,
increment=5,
)
yield
await select_ai.async_disconnect()
app = FastAPI(lifespan=lifespan)
@app.get("/chat")
async def chat(prompt: str):
profile = await select_ai.AsyncProfile(
profile_name="async_oci_ai_profile"
)
response = await profile.chat(prompt=prompt)
return {"response": response}
アプリの起動確認
$ uvicorn main:app --reload
INFO: Will watch for changes in these directories: ['/home/opc/selectai1.4']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [9100] using StatReload
INFO: Started server process [9103]
INFO: Waiting for application startup.
INFO: Application startup complete.
別のターミナルからリクエストを実行します。
# チャットエンドポイントへリクエスト
curl "http://127.0.0.1:8000/chat?prompt=How%20many%20customers%20do%20I%20have%3F"
# SQL 表示エンドポイントへリクエスト
curl "http://127.0.0.1:8000/show_sql?prompt=total%20sales%20by%20channel"
実行例[show_sql]）
$ curl "http://127.0.0.1:8000/show_sql?prompt=total%20sales%20by%20channel"
{"sql":"SELECT \n \"CH\".\"CHANNEL_DESC\" AS channel_description, \n SUM(\"S\".\"AMOUNT_SOLD\") AS total_sales\nFROM \n \"SH\".\"SALES\" \"S\"\n JOIN \"SH\".\"CHANNELS\" \"CH\" ON \"S\".\"CHANNEL_ID\" = \"CH\".\"CHANNEL_ID\"\nGROUP BY \n \"CH\".\"CHANNEL_DESC\""}
Step 5: 並列プロンプト処理を試す
asyncio.gather() による並列処理
独立した複数のプロンプトを同時に処理したい場合、asyncio.gather() を使うと入力順を保ちながら並列実行できます。
concurrent_test.py
import asyncio
import os
import select_ai
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
prompts = [
"How many customers?",
"How many products?",
"How many promotions?",
"List the top 5 customers by sales.",
]
async def show_sql(profile, prompt):
return await profile.show_sql(prompt=prompt)
async def main():
select_ai.create_pool_async(
user=user,
password=password,
dsn=dsn,
min_size=1,
max_size=4,
increment=1,
)
try:
profile = await select_ai.AsyncProfile(
profile_name="async_oci_ai_profile"
)
tasks = [show_sql(profile, prompt) for prompt in prompts]
results = await asyncio.gather(*tasks)
for prompt, sql in zip(prompts, results):
print(f"\nPrompt: {prompt}")
print(sql)
finally:
await select_ai.async_disconnect()
asyncio.run(main())
実行例）
$ python concurrent_test.py
Prompt: How many customers?
SELECT COUNT("c"."CUST_ID") AS "customer_count"
FROM "SH"."CUSTOMERS" "c"
Prompt: How many products?
SELECT
COUNT("p"."PROD_ID") AS "number_of_products"
FROM
"SH"."PRODUCTS" "p"
Prompt: How many promotions?
SELECT COUNT("p"."PROMO_ID") AS "number_of_promotions"
FROM "SH"."PROMOTIONS" "p"
Prompt: List the top 5 customers by sales.
SELECT
c."CUST_ID" AS "Customer ID",
c."CUST_FIRST_NAME" AS "First Name",
c."CUST_LAST_NAME" AS "Last Name",
SUM(s."AMOUNT_SOLD") AS "Total Sales"
FROM
"SH"."CUSTOMERS" c
JOIN "SH"."SALES" s ON c."CUST_ID" = s."CUST_ID"
GROUP BY
c."CUST_ID",
c."CUST_FIRST_NAME",
c."CUST_LAST_NAME"
ORDER BY
"Total Sales" DESC
FETCH FIRST 5 ROWS ONLY
run_pipeline() によるまとめて送信
プロンプトとアクションのペアが事前に確定している場合、run_pipeline() で 1 回のデータベースラウンドトリップにまとめることができます。
runpipeline.py
import asyncio
import os
import select_ai
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
wallet_location = os.getenv("SELECT_AI_WALLET_LOCATION")
wallet_password = os.getenv("SELECT_AI_WALLET_PASSWORD")
prompts = [
"How many customers?",
"How many products?",
"How many promotions?",
"List the top 5 customers by sales.",
]
prompt_specifications = [
("How many customers?", select_ai.Action.SHOWSQL),
("How many promotions?", select_ai.Action.RUNSQL),
("Explain how to count products.", select_ai.Action.EXPLAINSQL),
]
async def show_sql(profile, prompt):
return await profile.show_sql(prompt=prompt)
async def main():
select_ai.create_pool_async(
user=user,
password=password,
dsn=dsn,
wallet_location=wallet_location,
wallet_password=wallet_password,
min_size=1,
max_size=4,
increment=1,
)
try:
profile = await select_ai.AsyncProfile(profile_name="OCI_SHALL")
tasks = [show_sql(profile, prompt) for prompt in prompts]
results = await profile.run_pipeline(
prompt_specifications,
continue_on_error=True,
)
for prompt, sql in zip(prompts, results):
print(f"\nPrompt: {prompt}")
print(sql)
finally:
await select_ai.async_disconnect()
asyncio.run(main())
実行例）
$ python3 runpipeline.py
Prompt: How many customers?
SELECT COUNT("c"."CUST_ID") AS "customer_count"
FROM "SH"."CUSTOMERS" "c"
Prompt: How many products?
[
{
"number_of_promotions" : 503
}
]
Prompt: How many promotions?
To count products, you can use the following Oracle SQL query:
sql
SELECT
COUNT("p"."PROD_ID") AS "product_count"
FROM
"SH"."PRODUCTS" "p";
Explanation:
* We are selecting from the `"SH"."PRODUCTS"` table and assigning it the alias `"p"`.
* We use the `COUNT()` function to count the number of rows in the table, which corresponds to the number of products.
* We specify the column `"PROD_ID"` to count, but since we are counting all rows, we could also use `COUNT(*)`.
* The result is given the alias `"product_count"` for clarity.
This query will return the total number of products in the `"SH"."PRODUCTS"` table.
If you want to count products based on a specific condition, you can add a `WHERE` clause. For example, to count products with a specific category:
sql
SELECT
COUNT("p"."PROD_ID") AS "product_count"
FROM
"SH"."PRODUCTS" "p"
WHERE
UPPER("p"."PROD_CATEGORY") = UPPER('Electronics');
In this example, we use the `UPPER()` function to make the comparison case-insensitive, as per the rules. If the category string were enclosed in double quotes, we would not use the `UPPER()` function.
/home/opc/.local/share/uv/python/cpython-3.14.3-linux-x86_64-gnu/lib/python3.14/asyncio/events.py:94: RuntimeWarning: coroutine 'show_sql' was never awaited
self._context.run(self._callback, *self._args)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Step 6: ネットワークアクセス管理を試す
Select AI が AI プロバイダー API や SMTP サーバーなどの外部エンドポイントを呼び出すには、データベース側のネットワーク ACL 設定が必要です。バージョン 1.4 では Python から ACL 管理操作が可能です。
ネットワークアクセスの付与
管理者ユーザーで接続し、grant_network_access() を呼び出します。
grant.py
import os
import select_ai
admin_user = os.getenv("SELECT_AI_ADMIN_USER")
admin_password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
select_ai_user = os.getenv("SELECT_AI_USER")
wallet_location = os.getenv("SELECT_AI_WALLET_LOCATION")
wallet_password = os.getenv("SELECT_AI_WALLET_PASSWORD")
select_ai.connect(
user=admin_user,
password=admin_password,
dsn=dsn,
wallet_location=wallet_location,
wallet_password=wallet_password,
)
select_ai.grant_network_access(
users=select_ai_user,
host="smtp.example.com",
privileges=["connect", "smtp"],
lower_port=587,
upper_port=587,
)
SQL文で付与状況の確認
SQL> SELECT privilege, is_grant, ACL_OWNER
FROM dba_network_acl_privileges
WHERE PRINCIPAL='<DBスキーマ>';
PRIVILEGE IS_GR ACL_OWNER
--------- ----- ---------
connect true SYS
smtp true SYS
SQL> SELECT host, lower_port, upper_port FROM dba_network_acls WHERE host = 'smtp.example.com';
HOST LOWER_PORT UPPER_PORT
-------------------- ---------- ----------
smtp.example.com 587 587
ネットワークアクセスの取消し
select_ai.revoke_network_access(
users=select_ai_user,
host="smtp.example.com",
privileges=["connect", "smtp"],
lower_port=587,
upper_port=587,
)
Step 7: カスタムプロバイダーエンドポイント（xAI）を試す
Select AI for Python には Anthropic・AWS・Azure OpenAI・Cohere・Google・Hugging Face・OCI Generative AI・OpenAI の専用プロバイダークラスが用意されています。これ以外の互換エンドポイントは基底の Provider クラスで利用できます。xAI の例。（未実行）
import select_ai
# xAI のクレデンシャルを作成
select_ai.create_credential(
credential={
"credential_name": "xai_credential",
"username": "xai",
"password": "<xAI の API キー>",
},
replace=True,
)
# xAI プロバイダーを使用したプロファイルを作成
xai_profile = select_ai.Profile(
profile_name="xai",
attributes=select_ai.ProfileAttributes(
provider=select_ai.Provider(
provider_endpoint="https://api.x.ai",
model="grok-4-1-fast-reasoning",
),
credential_name="xai_credential",
object_list=[
{"owner": "SH", "name": "CUSTOMERS"},
{"owner": "SH", "name": "SALES"},
{"owner": "SH", "name": "PRODUCTS"},
{"owner": "SH", "name": "COUNTRIES"},
],
),
replace=True,
)
# SQL 生成を実行
sql = xai_profile.show_sql(
prompt="How many customers do I have?"
)
print(sql)
Step 8: エージェントチームのエクスポート・インポート
エージェントチームをポータブルなスペックとしてエクスポートし、別の環境へインポートできます。これにより、開発・テスト・本番環境間でのチーム定義の移植が容易になります。
事前準備：MOVIE_AGENT_TEAM を作成
エクスポート対象となるエージェントチームを事前に作成します。Agent でエージェントを定義し、Team でチームにまとめます。
createagentteam.py
import os
import select_ai
from select_ai.agent import Agent, Team, AgentAttributes, TeamAttributes
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
wallet_location = os.getenv("SELECT_AI_WALLET_LOCATION")
wallet_password = os.getenv("SELECT_AI_WALLET_PASSWORD")
select_ai.connect(
user=user,
password=password,
dsn=dsn,
wallet_location=wallet_location,
wallet_password=wallet_password,
)
# エージェントを作成
agent = Agent(
agent_name="MOVIE_AGENT",
attributes=AgentAttributes(
profile_name="OCI_AI_PROFILE",
role="You are a movie database expert. Answer questions about movies using SQL."
)
)
agent.create()
# チームを作成
team = Team(
team_name="MOVIE_AGENT_TEAM",
attributes=TeamAttributes(
agents=[
{"name": "MOVIE_AGENT", "task": "MOVIE_TASK"}
],
process="sequential"
)
)
team.create()
print("MOVIE_AGENT_TEAM を作成しました。")
注意:
AgentおよびTeamの作成パラメータ（instructions、agentsのリスト等）は環境やユースケースに合わせて変更してください。詳細は Select AI エージェントフレームワーク ドキュメント を参照してください。
$ python createagentteam.py
MOVIE_AGENT_TEAM を作成しました。
既存チームをエクスポートし、新しいチームとしてインポート
exportimport.py
import os
import json
import select_ai
from select_ai.agent import Team
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
wallet_location = os.getenv("SELECT_AI_WALLET_LOCATION")
wallet_password = os.getenv("SELECT_AI_WALLET_PASSWORD")
select_ai.connect(
user=user,
password=password,
dsn=dsn,
wallet_location=wallet_location,
wallet_password=wallet_password,
)
# 既存チーム(MOVIE_AGENT_TEAM)をエクスポート
source_team = Team.fetch("MOVIE_AGENT_TEAM")
specification = json.loads(source_team.export())
# 別プロファイルで新しいチーム(IMPORTED_MOVIE_AGENT_TEAM)としてインポート
Team.import_team(
profile_name="LLAMA_4_MAVERICK",
team_name="IMPORTED_MOVIE_AGENT_TEAM",
specification=specification,
force=True,
)
注意:
force=Trueを指定すると、ターゲットスキーマ内で競合するコンポーネントが上書きされます。本番環境での使用は慎重に行ってください。オブジェクトストレージへのエクスポート・インポートも可能です（クレデンシャル名とオブジェクトストレージの場所を指定）。
$ python exportimport.py
SQL で作成されたチームの確認
SQL> select AGENT_TEAM_NAME,STATUS FROM user_ai_agent_teams;
AGENT_TEAM_NAME
--------------------------------------------------------------------------------
STATUS
--------
MOVIE_AGENT_TEAM
ENABLED
IMPORTED_MOVIE_AGENT_TEAM
ENABLED
おわりに
今回の検証で確認できたポイントは以下の通りです。
-
stream=Trueとchunk_sizeパラメータを追加するだけで、既存のchat()やgenerate()呼び出しをストリーミング対応に変換できます -
select_ai[cli]をインストールするだけでselect-aiコマンドが使えるようになり、Python コードを書かずにプロファイルの動作確認やデモが行えます - FastAPI などの Web フレームワークでは、
create_pool()またはcreate_pool_async()をアプリ起動時に一度だけ呼び出すパターンが推奨であり、リクエストごとの接続作成を避けることでリソース効率が向上します -
asyncio.gather()やrun_pipeline()を使うと、複数プロンプトをほぼ同時に処理でき、バッチ処理や複数質問ダッシュボードなどのユースケースで効果的です - カスタム
Providerクラスを使用することで、xAI（Grok）など専用クラスのないプロバイダーエンドポイントも Oracle AI Database のプロファイル・クレデンシャル管理の仕組みの中で利用できます