Elastic Stack v9.3新登場のWorkflowsでワークフローとAIエージェントによるデータ分析を実装する例（第2回）

Last updated at 2026-03-05Posted at 2026-03-05

はじめに

このブログのシリーズでは、ElasticサンプルデータのKibana Flight Dataの飛行機の離発着履歴データを使い、Elastic WorkflowsとElastic Agent Builderを使ったフライト遅延の調査と原因分析の運用の自動化までの道のりをお伝えします。
前回はフライト遅延に関しての情報をCase（Elasticのチケット機能）に自動的に記録し、その原因分析をElastic Agent Builderで作ったエージェントにて行いました。

本ブログでは、この自動化をもう一歩完成形に近づけます。

Casesの作成も自動的にワークフローにて行うようにします
分析対象の条件（空港都市名と日付）をチャットにて受け付けるようにします（日本語の空港都市名が入力されても、内部で英語に変換するなどの工夫あり）

本ブログはElastic Cloudでv9.3のElasticデプロイメントを使用しています。本ブログの内容を行うにはEnterpriseサブスクリプションである必要があります。フリートライアルでも試すことが可能です。

シリーズの他の記事はこちらから参照ください
第１回
 第２回
 第３回
 第４回
 第５回
 第６回

このブログで作成するもの

作成するコンポーネント

ワークフローwrite_flight_delay_to_case
前回のesql_to_case_v2の更新版です。
データ分析の実行とAIエージェントによる解説をチケットに自動登録。分析対象のチケットがまだ存在しなければ対象空港と日付別のチケットを新規作成。既に対象空港と日付のチケットが存在すれば、チケットに追記。
AIエージェントTools write_flight_delay_to_case
上記ワークフローをAgentが実行できるようになるためのTool。わかりやすいように同じ名前にしています。
AIエージェント check_flight_delay_agent
チャットにて分析対象の空港と日付を指示すると、ワークフローwrite_flight_delay_to_caseを実行

呼び出しフロー
ユーザーがチャットで特定の日付と空港の調査を依頼 --> Agent check_flight_delay_agent --> Agent Tool write_flight_delay_to_case --> Workflow write_flight_delay_to_case

実装：ワークフロー write_flight_delay_to_case

遅延・キャンセル便を検索してチケットにコメントする前回のワークフローesql_to_case_v2をCloneして新しくwrite_flight_delay_to_caseという名前にします。ここから以下のような動きとなるように変更します。

入力パラメータの変数化
入力パラメータを検索条件に既存Caseを検索
既存Caseがなければ新規Caseを作成
入力パラメータに関してのフライト遅延がない場合のハンドリングを追加
ES|QL実行の結果のAIレポートと、それを書き込んだCaseのIdとそのリンクをレスポンスに含める

version: "1"
name: write_flight_delay_to_case
description: Write ES|QL result to a case comment
tags:
  - draft
triggers:
  - type: manual
inputs:
  - name: city_name
    type: string
  - name: start_date
    type: string
  - name: end_date
    type: string
consts:
  search_parameter: "+{{ inputs.city_name }} +{{ inputs.start_date | date: '%Y/%m/%d' }}"
  search_parameter_encoded: ${{consts.search_parameter}} | url_encode
steps:    
  - name: search_parameter_encoded
    type: console
    with:
      message: |
        {% assign keyword = inputs.city_name %}
        {% assign date = inputs.start_date | date: '%Y/%m/%d' %}
        {% assign query = "+" | append: keyword | append: "+" | append: '"' | append: date | append: '"' %}
        {{ query | url_encode }}

  - name: execute_esql
    type: elasticsearch.esql.query
    with:
      query: |
        FROM kibana_sample_data_flights  | WHERE timestamp >= "{{ inputs.start_date }}" AND timestamp < "{{ inputs.end_date }}"  | WHERE OriginCityName == "{{ inputs.city_name }}" OR DestCityName == "{{ inputs.city_name }}"  | WHERE Cancelled == true OR (FlightDelay == true AND FlightDelayMin >= 60) | KEEP timestamp, Cancelled, FlightDelay, OriginCityName, DestCityName, Carrier, FlightDelayType, FlightDelayMin, OriginWeather, DestWeather
      format: json
  - name: write_to_case_if_data_exist
    type: if
    condition: "steps.execute_esql.output.documents_found > 0"
    steps:
    - name: check_existing_case
      type: kibana.request
      with:
        method: "GET"
        path: "/api/cases/_find?defaultSearchOperator=AND&searchFields=title&search={{ steps.search_parameter_encoded.output }}"
    - name: create_new_case_if_none
      type: if
      condition: 'steps.check_existing_case.output.total: 0'
      steps:
      - name: create_new_case
        type: kibana.createCaseDefaultSpace
        with:
          title: "{{ inputs.city_name }} {{ inputs.start_date | date: '%Y/%m/%d' }} のフライト遅延とキャンセル"
          owner: observability
          description: >
            目的：フライトのキャンセル及び大幅な遅延をまとめ、日毎のレポートを作成し、後に参照できるように記録する
            対象日付：{{ inputs.start_date | date: '%Y/%m/%d' }}
            対象空港：{{ inputs.city_name }}
          settings:
            syncAlerts: true
          severity: medium
          tags: []
          connector:
            id: "none"
            name: "none"
            type: ".none"
            fields:

    - name: case_id
      type: console
      with:
        message: "{% if steps.check_existing_case.output.total == 0 %}{{ steps.create_new_case.output.id }}{% else %}{{ steps.check_existing_case.output.cases[0].id }}{%endif%}"

    - name: transform_output_by_ai
      type: ai.agent
      with:
        agent_id: flight_analyzer_agent
        message: "次のデータをCaseに追記するコメントに変換してください: {{steps.execute_esql.output | json}}"

    - name: add_response_to_existing_case
      type: kibana.request
      with:
        method: POST
        path: /api/cases/{{ steps.case_id.output }}/comments
        body:
          type: user
          owner: observability
          comment: "{{ steps.transform_output_by_ai.output }}"
      
    else:
    - name: output_no_result
      type: console
      with:
        message: "No data found for {{ inputs | json }}"
    
  - name: output_succes
    type: if
    condition: "steps.execute_esql.output.documents_found > 0"
    steps:
      - name: log_output
        type: console
        with:
          message: |
            以下のレポートがCase Id: [{{steps.case_id.output}}]({{ kibanaUrl }}/app/observability/cases/{{steps.case_id.output}})　に追加されました。
            {{ steps.transform_output_by_ai.output }}

ワークフローを作成したら、その場でテスト実行してロジック通りに動くことを確認しましょう。（結果は最後にまとめてます）

実装：Tool (write_flight_delay_to_case) の作成

上記のワークフローを実行するAgent BuilderのToolを新規作成します。

エージェントが自分で考えてワークフローの入力パラメータを設定するので、このツールのDescriptionを以下のように記述することで、期待するパラメータのフォーマットをエージェントに指示しておきます。

このツールには、入力として以下の情報を与えてください。
* 空港の都市名 ... 頭文字大文字の英語で入力 (例: Tokyo)
* フライトデータ分析の開始日時 ... ISOフォーマットであること（例: 2026-02-04T00:00:00Z)
* フライトデータ分析の終了日時 ... ISOフォーマットであること（例: 2026-02-05T00:00:00Z)

Kibana Sample Data Flightsのインデックスから入力された情報に関するフライトの遅延・キャンセルを検索し、その結果をCaseに追加します。対象のCaseが存在しない場合、Caseを新規作成します。

実装： Agent (check_flight_delay_agent) の作成

ユーザーがチャットから分析対象の空港と日付を指定して、上記ワークフローを実行してくれるAIエージェントを作成します。

エージェントに設定したプロンプト（Custom Instructions)はシンプルですが、ポイントはToolsを指定するところです。このエージェントに汎用性を持たせずに、さきほどのワークフローだけ実行してほしいので、以下の2つのToolsだけ有効にし、他は無効にします。

動作確認：エージェント(check_flight_delay_agent)

チャットからテストしてみます。2/5のトロントのフライトを分析指示します。
（この記事を書いている途中にElastic Managed LLMとしてOpenAI GPT-OSS 120Bの選択肢増えていたので、それをここでは使ってみてます。その前まで使っていたClaude Sonnetよりも大幅に安いです)