3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@haseshin(Shintaro Hase)
in

Confluentからデータを取得するエージェントをwatsonx Orchestrateで作ってみた

3
Last updated at Posted at 2026-03-26

Confluent Cloudとwatsonx Orchestrateで作るリアルタイム在庫管理エージェント

IBMがデータ・ストリーミング・プラットフォームを提供するConfluent社の買収を完了したことが発表されました。
https://jp.newsroom.ibm.com/2026-03-18-ibm-completes-acquisition-of-confluent,-making-real-time-data-the-engine-of-enterprise-ai-and-agents

リリースには以下のように記載されています。

IBMとConfluent社は、あらゆるAIモデル、AIエージェント、自動化ワークフローが必要とするリアルタイムかつ信頼できるデータを、オンプレミスおよびハイブリッドクラウド環境全体で安全かつ大規模に扱えるスマートなデータ・プラットフォームを提供します。

というわけで、この記事では、こちらの記事をベースにConfluent Cloud上でリアルタイムに在庫データを集計し、watsonx Orchestrateのエージェントから在庫情報を取得できるシステムを構築する手順を紹介します。

Confluent Cloudとは？

Confluent Cloudは、Apache Kafkaをベースとしたフルマネージドのストリーミングデータプラットフォームです。リアルタイムでデータを処理・分析できる以下の機能を提供します：

  • Apache Kafka: 高スループットの分散メッセージングシステム
  • ksqlDB: SQLライクな構文でストリームデータを処理できるデータベース
  • Schema Registry: データスキーマの管理
  • Connectors: 外部システムとの連携

システムアーキテクチャ

今回構築するシステムの全体像は以下の通りです：

在庫トランザクション
    ↓
Kafka Topic (inventory.transactions)
    ↓
ksqlDB Stream (inventory_transactions_stream)
    ↓
ksqlDB Table (inventory_availability) ← リアルタイム集計
    ↓
watsonx Orchestrate Tool(Python)
    ↓
watsonx Orchestrate Agent

まず、inventory.transactionsに在庫の増減のイベントが送信され、そのイベントを元に、iventory_ransactions_streamが動作し、inventory_availabiityテーブルを更新します。watsonx OrchestrateからはPythonのToolを介して在庫情報を取得することになります。

1. Confluent Cloudの準備

アカウント作成

Confluent Cloudにアクセスしてアカウントを作成し、クラスターを作成します。
カード情報を登録し、400$分のcreditを利用することが可能です。
image.png

2日間ほど利用しましたが、70$程度のCreditを消費したようです。

クラスターとinventory.transactionsトピックの作成

以下の手順に沿ってクラスターを作成します。IBM Bobを使用してCLI経由で自動で作成することも可能でしたが、勉強のために手作業を行いました。

ksql clusterの作成

ksql clusterの作成は、CLIを用いて作業を行います。
CLIをインストールし、以下のコマンドでksql clusterを作成します。

confluent login --save
confluent iam user list
confluent ksql cluster create sku-availability-calculator --cluster {KAFKA_CLUSTER_ID} --credential-identity {USER_ID}

クラスターが作成されるまでしばらく時間がかかります。作成状況は以下のコマンドで確認できます。

confluent ksql cluster describe {KSQL_CLUSTER_ID}

image.png

Streamの作成

clusterからksqlDBを選択し、以下のクエリを実行し、Streamを作成します。

CREATE STREAM inventory_transactions (
     sku VARCHAR,
     branch VARCHAR,
     quantity INT
 ) WITH (
     KAFKA_TOPIC='inventory.transactions',
     KEY_FORMAT='KAFKA',
     VALUE_FORMAT='JSON'
 );

inventory.availabilityテーブルの作成

以下のクエリを実行し、inventory.availabilityテーブルを作成します。

CREATE TABLE inventory_availability WITH (
  KAFKA_TOPIC='inventory.availability',
  KEY_FORMAT='JSON',
  VALUE_FORMAT='JSON',
  PARTITIONS=1
) AS
SELECT 
  sku,
  branch,
  SUM(quantity) AS available_quantity
FROM inventory_transactions
GROUP BY sku, branch
EMIT CHANGES;

EMIT CHANGESが、プッシュ・クエリと呼ばれるもので、永続的にクエリを実行してトピックの更新をトリガーにデータを更新してくれるようです。

##サンプル・メッセージの送信
inventor_transactionsトピックにサンプル・データを送信するには、こちらのサンプル・コードを使用します。
以下のコマンドでクローンします。

git clone https://github.com/IBM/oic-i-agentic-ai-tutorials
cd confluent-agents

kafka clusterに対するAPIを管理画面から発行し、.env.exampleの以下の値を正しい値に設定します。

BOOTSTRAP_SERVERS=pkc-xxxxx.region.provider.confluent.cloud:9092
KAFKA_API_KEY=your-kafka-api-key-here
KAFKA_API_SECRET=your-kafka-api-secret-here

以下のコマンドで、メッセージを送信します。

pip install -r requirements.txt
python produce_messages.py

sampmle.transactions.jsonに記述されている20件のトランザクション・データがトピックに送信されます。
image.png

メッセージの確認

inventory.transactionsトピックのメッセージを確認し、正しくメッセージを受信していることを確認します。

image.png

次にksqlDBに対して以下のクエリを発行してみます。各SKUごとに在庫が計算されていることが分かります。

select * from INVENTORY_AVAILABILITY EMIT CHANGES;

image.png

Toolの作成

サンプルには、在庫を取得するMCPサーバーが付属するのですが、今回試してみたところ、正しく動作しませんでした。使用している環境のMCPゲートウェイに問題がありそうだったので、IBM Bobを用いてMCPではなく、単体のToolに作り替えてみました。また、通常Connectionを用いて接続情報を外部化しますが、構成が面倒だったのでハードコードしてとお願いして実装しました。

単体テストのためのコードも生成してもらい、エラーを修正しながら最終的には正しく動作するコードが完成しました。

サンプルの修正ではなく、ゼロからのTool作成も試してみましたが、テーブル定義と接続情報と一緒に、「Confluent Cloud上の以下のテーブルからデータを取得するToolを作って」と依頼すると、何度かエラーを修正しながらToolを作ることができました。

get_sku_availability_tool.py
#!/usr/bin/env python3
"""
SKU在庫確認ツール
ksqlDBからリアルタイムの在庫情報を取得
"""

import requests
import json
from typing import Dict, Any
from requests.auth import HTTPBasicAuth
from ibm_watsonx_orchestrate.agent_builder.tools import tool

# ksqlDB設定（.envから取得した値をハードコード）
KSQLDB_ENDPOINT = "https://pksqlc-xxxxx.us-east1.gcp.confluent.cloud:443"
KSQLDB_API_KEY = "your-ksqldb-api-key"
KSQLDB_API_SECRET = "your-ksqldb-api-secret"

def query_ksqldb(query: str) -> list:
    """ksqlDBにクエリを実行"""
    url = f"{KSQLDB_ENDPOINT}/query-stream"
    headers = {'Content-Type': 'application/vnd.ksql.v1+json'}
    payload = {"sql": query, "properties": {}}
    
    response = requests.post(
        url,
        headers=headers,
        json=payload,
        auth=HTTPBasicAuth(KSQLDB_API_KEY, KSQLDB_API_SECRET),
        stream=True,
        timeout=30
    )
    
    if response.status_code != 200:
        raise Exception(f"HTTP {response.status_code}: {response.text}") 
    results = []
    for line in response.iter_lines():
        if line:
            try:
                data = json.loads(line.decode('utf-8'))
                # ヘッダー行をスキップ
                if 'header' in data or 'columnNames' in data:
                    continue
                # データ行を処理
                if isinstance(data, list) and len(data) >= 3:
                    results.append({
                        'sku': data[0],
                        'branch': data[1],
                        'available_quantity': data[2]
                    })
            except:
                continue
    
    return results

@tool()
def get_sku_availability(sku: str = "", branch: str = "") -> Dict[str, Any]:
    """
    リアルタイムの在庫情報を取得します。
    
    ksqlDBのINVENTORY_AVAILABILITYテーブルから、
    SKUと店舗ごとの現在の在庫数を取得します。
    
    Args:
        sku: 商品コード（例: 'LAPTOP-DELL-XPS-15'）。空の場合は全商品
        branch: 店舗名（例: 'DubaiMall'）。空の場合は全店舗
    
    Returns:
        在庫情報の辞書
        - results: 在庫レコードのリスト（sku, branch, available_quantity）
        - message: レコードが見つからない場合のメッセージ
        - error: エラーが発生した場合のメッセージ
    """
    try:
        # クエリを構築
        if sku and branch:
            query = f"SELECT * FROM INVENTORY_AVAILABILITY WHERE SKU='{sku}' AND BRANCH='{branch}';"
        elif sku:
            query = f"SELECT * FROM INVENTORY_AVAILABILITY WHERE SKU='{sku}';"
        elif branch:
            query = f"SELECT * FROM INVENTORY_AVAILABILITY WHERE BRANCH='{branch}';"
        else:
            query = "SELECT * FROM INVENTORY_AVAILABILITY;"
        
        # クエリを実行
        results = query_ksqldb(query)       
        if not results:
            return {"message": "在庫レコードが見つかりませんでした", "results": []}       
        return {"results": results}       
    except Exception as e:
        return {"error": str(e)}

Toolのインポート

以下のコマンドでwatsonx Orchestrateに作成したToolをインポートします。

orchestrate tools import -f get_sku_availability_tool.py -k python

エージェントを作成し、インポートしたツールを追加します。チャット欄から「在庫の確認をして」と入力すると正しく在庫が取得できました！
image.png

まとめ

この記事では、Confluent Cloudとwatsonx Orchestrateを組み合わせ、リアルタイムデータと連携するAIエージェントの構築フローを解説しました。

データ・ストリーミング・プラットフォームを基盤に据えることで、AIエージェントには以下のメリットがもたらされます。

  • データ鮮度： ビジネスイベントを即座にコンテキストとして利用
  • エージェントが使いやすいデータ： 複雑なRAWデータを、AIが解釈・実行しやすい形へ容易にクレンジング・構造化

今回のサンプルは「在庫確認」というシンプルなものでしたが、元記事では「在庫切れ時に別エージェントへタスクを委譲し、RAGを用いて代替案を提示する」といったワークフローを実装しています。また、トピックへのデータ流入をトリガーにREST API経由でエージェントを起動するなど、柔軟な連携パターンも可能です。

ちなみに、Confluentはデータ・ストリームの中にエージェントが位置する「Streaming Agents」というコンセプトを提要しているようです。今回の作業を通じて、「SaaS is dead」という言葉が象徴するように、人間がSaaSの画面を操作する時代は終わりを迎え、今後は、AIエージェントがデータストリームと直接対話し、自律的に業務を完結させる方向へとシフトしていくのではないかと、より具体的に感じることができました。

3
1
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
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?