LLMベースのMQTTデータ処理

EMQX 5.10.0以降、FlowデザイナーはOpenAI GPTやAnthropic Claudeなどの大規模言語モデル（LLM）との統合をサポートしています。この機能により、ログの要約、センサーデータの分類、MQTTメッセージの拡張、リアルタイムインサイトの生成など、自然言語プロンプトを用いたインテリジェントなメッセージフローを構築できます。

機能概要

FlowデザイナーのLLMベース処理ノードは、外部のLLM APIに接続してメッセージ内容を処理するAI搭載コンポーネントです。これらのノードを使うことで、MQTTデータをgpt-4oやclaude-3-sonnetなどのモデルに送信し、応答を受け取り、フローの下流に渡すことが可能です。

注意

LLMの呼び出しとデータ処理には時間がかかります。モデルの応答速度によっては、処理全体で数秒から十数秒かかる場合があります。そのため、LLM処理ノードは高スループット（TPS）が求められるシナリオには適していません。

主要な概念

• LLMプロバイダー：AIサービス（OpenAI / Anthropic）の名前付き設定。
• Completion Profile：LLMモデルのパラメーター（モデルID、システムプロンプト、トークン制限など）をまとめた再利用可能な設定。
• AI Completion Node：入力をLLMに送信し、その結果をユーザー定義のエイリアスに格納するフローコンポーネント。
• ai_completion：テキストやバイナリデータをLLMに送信し応答を返すRule SQL関数。

動作の仕組み

FlowデザイナーでMQTTメッセージを受信すると、AI Completion Nodeは内部で組み込みのSQL関数ai_completion/2,3を呼び出し、設定されたLLMにデータを送信します。

null

1. メッセージはMessagesノード（例：トピックをサブスクライブ）を経由してフローに入ります。

2. Data Processingノード（任意）でdevice_id、payload、timestampなどのフィールドを抽出または変換できます。

3. OpenAIまたはAnthropicノードは背後でai_completion関数を使い、

   • 選択されたCompletion Profile（プロバイダー情報、モデル名、システムメッセージなど）を参照します。
   • 選択された入力（例：payload）をLLMに送信します。
   • LLM APIからの応答（例：要約や分類結果）を受け取ります。

4. 応答はOutput Result Aliasに格納され、以下のような下流ノードで利用可能になります。

   • Republish（別トピックへのパブリッシュ）
   • Database（PostgreSQL、MongoDBなどへの結果挿入）
   • Bridge（リモートブローカーやクラウドサービスへの転送）

対応LLMプロバイダー

EMQX 5.10.0では以下のプロバイダーをサポートしています：

• OpenAI：GPT-3.5、GPT-4、GPT-4oなど
• Anthropic：Claude 3モデル

LLMベース処理ノードの設定

FlowデザイナーでLLMを使用するには、OpenAIノードまたはAnthropicノードのいずれかを選択して専用の処理ノードを設定します。各ノードでは、MQTTメッセージデータのどのフィールドをLLMに送るか、システムプロンプトによるモデルの動作指定、AI生成結果の格納先などを定義できます。設定後、これらのノードは背後でai_completion関数を呼び出し、選択したLLMでデータ処理を行います。

OpenAIノードの設定

OpenAIノードを使用するには：

1. ProcessingパネルからOpenAIノードをドラッグします。

2. ソースまたは前処理ノードに接続します。

3. 以下の項目を設定します：

   • Input：ソースフィールドを入力または選択します。選択肢はevent、id、clientid、username、payloadなどです。

   • System Message：AIモデルに期待する出力を生成させるためのプロンプトメッセージを入力します。例：「入力JSONデータの数値キーの値を合計し、結果のみを出力してください」。

   • Model：LLMプロバイダーのモデルを選択します。例：gpt-4o、gpt-3.5-turbo。

   • API Key：OpenAIのAPIキーを入力します。

   • Base URL：任意のカスタムエンドポイントを入力します。空欄の場合はOpenAIのデフォルトエンドポイントが使用されます。

   • Output Result Alias：LLMの出力結果を格納する変数名です。後続のアクションや処理で参照します。例：summary

     TIP

     エイリアスに英数字とアンダースコア以外の文字が含まれる場合、数字で始まる場合、またはSQLキーワードの場合は、ダブルクォーテーションで囲んでください。

4. 保存をクリックして設定を適用します。

Anthropicノードの設定

Anthropicノードを使用するには：

1. ProcessingパネルからAnthropicノードをドラッグします。

2. メッセージ入力またはData Processingノードに接続します。

3. 以下の項目を入力します：

   • Input：ソースフィールドを入力または選択します。選択肢はevent、id、clientid、username、payloadなどです。

   • System Message：AIモデルに期待する出力を生成させるためのプロンプトメッセージを入力します。例：「入力JSONデータの数値キーの値を合計し、結果のみを出力してください」。

   • Model：LLMプロバイダーのモデルを選択します。例：claude-3-sonnet-20240620。

   • Max Tokens：応答の最大トークン数を制御します（デフォルト：100）。

   • Anthropic Version：Anthropicのバージョンを選択します（デフォルト：2023-06-01）。

   • API Key：AnthropicのAPIキーを入力します。

   • Base URL：任意のカスタムエンドポイントを入力します。空欄の場合はAnthropicのデフォルトエンドポイントが使用されます。

   • Output Result Alias：LLMの出力結果を格納する変数名です。後続のアクションや処理で参照します。例：summary

     TIP

     エイリアスに英数字とアンダースコア以外の文字が含まれる場合、数字で始まる場合、またはSQLキーワードの場合は、ダブルクォーテーションで囲んでください。

4. 保存をクリックして設定を適用します。

クイックスタート

以下の2つの例では、EMQXでLLMベース処理ノードを使ったフローの迅速な構築とテスト方法を示しています：

• OpenAIノードを使ったフローの作成：GPTモデルでMQTTメッセージを要約または変換します。
• Anthropicノードを使ったフローの作成：ClaudeモデルでMQTTメッセージの数値処理を行います。