Cortex Agents を Python から呼び出す方法　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　～ Agent 作成からスレッド会話の継続まで完全解説～

Snowflake

こんにちは！システムサポートの寺田です！

Snowflake の Cortex Agents を使うと、Python から Agent に質問を投げて、
会話を続けながらやり取りできます。
今回は、Agent の作成方法から API 呼び出し、親メッセージIDの扱いまで紹介します。

本記事でわかること

  • Cortex Agents の概要
  • Cortex Agents の作成方法
  • Python で Snowflake API をたたく実装例
  • スレッドを作成し、Agent と会話を続ける方法


Cortex Agents とは

Cortex Agentsは、Snowflakeが提供するAIエージェント機能です。
エージェントに対し、自然言語で問い合わせすることができます。
構造化データと非構造化データをツールとして使用し、データを分析します。

Cortex Agentsに関する公式ドキュメントは以下をご参照ください。

Cortex Agents | Snowflake Documentation
docs.snowflake.com


前提条件

  • Snowflake アカウントを持っている
  • Cortex Agents を利用できる権限がある
  • Python 環境がある
  • Python 環境に「snowflake.connector」と「requests」をインストールしている


Snowflake の操作

Cortex Agents の作成

以下のSQLで Agent を作成します。

CREATE OR REPLACE AGENT <Agent名>
FROM SPECIFICATION
  $$
  models:
    orchestration: auto

  orchestration:
    budget:
      tokens: 2048

  instructions:
    response: "あなたはデータ分析アシスタントです。質問には簡潔に答えてください。"
  $$;

各項目の意味

項目意味
orchestrationAgent が使用するAIモデルを設定
tokens1 回の応答で使える最大トークン数を設定
responseAgent の性格・役割・応答スタイルを定義


Python コードの解説

完成コード

はじめに、完成したコードを記載します。

import snowflake.connector
import requests

conn = snowflake.connector.connect(
    user="<ユーザー名>",
    password="<パスワード>",
    account="<アカウント>",
    warehouse="<ウェアハウス>",
    database="<データベース>",
    schema="<スキーマ>"
)

token = conn.rest.token

headers = {
    "Authorization": f'Snowflake Token="{token}"',
    "Content-Type": "application/json",
    "Accept": "application/json"
}

threads_endpoint = "https://<アカウント>.snowflakecomputing.com/api/v2/cortex/threads"
endpoint = "https://<アカウント>.snowflakecomputing.com/api/v2/databases/<データベース>/schemas/<スキーマ>/agents/<エージェント名>:run"

res_thread = requests.post(threads_endpoint, headers=headers, json={})
thread_id = res_thread.json()["thread_id"]

question1 = "こんにちは"

payload1 = {
    "thread_id": thread_id,
    "parent_message_id": 0,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question1}
            ]
        }
    ],
    "stream": False
}

response1 = requests.post(endpoint, headers=headers, json=payload1)
res1 = response1.json()

answer1 = next(
    (item["text"] for item in res1.get("content", []) if item.get("type") == "text"),
    ""
)

print(f"私の質問：{question1}")
print(f"Agentの回答：{' '.join(answer1.split())}")

message_id_1 = res1["metadata"]["assistant_message_id"]

question2 = "さっき私はなんと言いましたか？"

payload2 = {
    "thread_id": thread_id,
    "parent_message_id": message_id_1,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question2}
            ]
        }
    ],
    "stream": False
}

response2 = requests.post(endpoint, headers=headers, json=payload2)
res2 = response2.json()

answer2 = next(
    (item["text"] for item in res2.get("content", []) if item.get("type") == "text"),
    ""
)

print(f"私の質問：{question2}")
print(f"Agentの回答：{' '.join(answer2.split())}")

以降、このコードの詳細な解説を記載していきます。


Python から Snowflake に接続する

まず、「snowflake.connector」をインポートします。
「snowflake.connector」はローカルの Python 環境から
Snowflake に接続するために使用します。

import snowflake.connector

接続コードは以下の通りです。
ご自身のアカウント情報を入力してください。

conn = snowflake.connector.connect(
    user="<ユーザー名>",
    password="<パスワード>",
    account="<アカウント>",
    warehouse="<ウェアハウス>",
    database="<データベース>",
    schema="<スキーマ>"
)


API を呼び出す設定する

requests のインポート

「requests」 をインポートします。
「requests」 は、Snowflake の REST API を使用するときに必要です。

import requests

アクセストークンの取得

アクセストークンを取得します。

token = conn.rest.token

ヘッダー作成

API リクエスト用のヘッダーを作成します。
このヘッダーは、スレッドの作成および Agent 呼び出し時に使用します。

headers = {
    "Authorization": f'Snowflake Token="{token}"',
    "Content-Type": "application/json",
    "Accept": "application/json"
}

エンドポイント作成

エンドポイントを作成します。
エンドポイントは2つ用意する必要があります。
ご自身のアカウント情報、作成した Agent の情報を入力してください。

threads_endpoint = "https://<アカウント>.snowflakecomputing.com/api/v2/cortex/threads"
endpoint = "https://<アカウント>.snowflakecomputing.com/api/v2/databases/<データベース>/schemas/<スキーマ>/agents/<エージェント名>:run"

各エンドポイントの用途

エンドポイント用途
threads_endpointスレッドの作成
endpointAgent の呼び出し


スレッドを作成する

スレッドを作成します。

res_thread = requests.post(threads_endpoint, headers=headers, json={})

作成したスレッドの、thread_id を取得し変数に格納します。
thread_id は Agent に送るリクエスト文内で使用します。

thread_id = res_thread.json()["thread_id"]


Agent を呼び出す（1回目）

質問作成

1回目の質問を作成します。
例として、1回目は「こんにちは」と Agent に送信します。

question1 = "こんにちは"

リクエスト文の作成

リクエスト文を作成します。

payload1 = {
    "thread_id": thread_id,
    "parent_message_id": 0,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question1}
            ]
        }
    ],
    "stream": False
}

各項目の意味

項目内容
thread_idスレッドのID
parent_message_idスレッドの親メッセージの ID
最初の質問をする場合、0を指定する
content質問を入力
streamストリーミング応答を設定

Agent 呼び出し（1回目）

Agent を呼び出します。

response1 = requests.post(endpoint, headers=headers, json=payload1)
res1 = response1.json()

応答テキストの取得

Agent が返した JSON から、応答テキストを取り出します。

answer1 = next(
    (item["text"] for item in res1.get("content", []) if item.get("type") == "text"),
    ""
)

print(f"Agentの回答：{' '.join(answer1.split())}\n")


親メッセージIDを取り出す

2回目の質問を「前回の続き」として扱うには、スレッドの親メッセージIDが必要です。
Agent が返した JSON に含まれる assistant_message_id が、
2回目以降の parent_message_id に該当します。

message_id_1 = res1["metadata"]["assistant_message_id"]


Agent を呼び出す（2回目）

質問作成

2回目の質問を作成します。
スレッドが継続しているか判断できるように、
2回目は「さっき私はなんと言いましたか？」と Agent に送信します。

question2 = "さっき私はなんと言いましたか？"

リクエスト文の作成

2回目のリクエスト文を作ります。
parent_message_id に1回目の応答で取得した assistant_message_id を設定します。

payload2 = {
    "thread_id": thread_id,
    "parent_message_id": message_id_1,
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question2}
            ]
        }
    ],
    "stream": False
}

Agent 呼び出し（2回目）

Agent を呼び出します。

response2 = requests.post(endpoint, headers=headers, json=payload2)
res2 = response2.json()

応答テキストの取得

Agent が返した JSON から、応答テキストを取り出します。

answer2 = next(
    (item["text"] for item in res2.get("content", []) if item.get("type") == "text"),
    ""
)

print(f"Agentの回答：{' '.join(answer2.split())}\n")



実行結果

完成したコードの実行結果がこちらです。

=== 1回目の質問 ===
私の質問：こんにちは
Agentの回答：こんにちは！データ分析アシスタントです。何かお手伝いできることがあれば、お気軽にお聞きください。データ分析、SaaS指標、ビジネス分析など、幅広くサポートいたします。

=== 2回目の質問 ===
私の質問：さっき私はなんと言いましたか？
Agentの回答：先ほど「こんにちは」とおっしゃいました。

スレッドで会話が続いてることが確認できます。


付録

2回目のリクエスト文の parent_message_id を0に設定するとどうなるでしょうか？

payload2 = {
    "thread_id": thread_id,
    "parent_message_id": 0
    "messages": [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question2}
            ]
        }
    ],
    "stream": False
}

実行結果はこちらです。

=== 1回目の質問 ===
私の質問：こんにちは
Agentの回答：こんにちは！データ分析アシスタントです。何かお手伝いできることがあれば、お気軽にお聞きください。データ分析やビジネス指標に関するご質問にお答えします。

=== 2回目の質問 ===
私の質問：さっき私はなんと言いましたか？
Agentの回答：これが今回の会話でいただいた最初のメッセージです。以前のやり取りの履歴はありませんので、「さっき何と言ったか」をお伝えすることはできません。 何かお手伝いできることがあれば、お気軽にどうぞ！

2回目の質問が、会話の続きではなく最初の質問として扱われます。
スレッドで会話を続けるには、parent_message_id を正しく設定する必要があります。


最後に

今回は、Cortex Agents を Python から呼び出して、
会話を続ける方法をご紹介しました。
スレッドを作成し、parent_message_id を正しく設定することで、
前回のやり取りを踏まえた会話ができることを確認しました。
今回は簡単なメッセージのやり取りでしたが、
実際にSnowflakeに格納されているデータに対して問い合わせを行うことも可能です。
Snowflake と Python を組み合わせることで、さまざまな活用の幅が広がります。
この記事が、Cortex Agents を試してみるきっかけになれば幸いです。

コメント

タイトルとURLをコピーしました