AI Agent開発日記 2025/03/28

2025年3月28日

この記事について

本記事は、Agent Builder を使用したデータストア参照型対話エージェントの作成に関する調査記録です。

やりたいこと

Agent Builder を用いたデータストア参照型対話エージェントの作成において、
データストアのファイル数が増加することで回答率が低下する問題への対処を進める。

前回の調査日記

Agent BuilderのツールであるOpenAPIのスキーマを適切に設定し、Cloud RunにデプロイしたElasticsearchで検索できるか調べる。

試したこと
- Agent BuilderのツールであるOpenAPIのyaml形式のスキーマを設定して動作を見る前に、jupyterでスキーマを試し、様子を見る。
結果
- 下記形式のコードを使ってjupyterからelasticsearchへの検索が成功した。
- スキーマを用いてagentbuilderのtoolでも試してみる。

コード

import yaml
from elasticsearch import Elasticsearch

elasticsearch_schema = """
openapi: 3.0.0
info:
  title: Elasticsearch Search API
  version: v1
servers:
  - url: 'https://your-elasticsearch-endpoint.example.com'  # Elasticsearch のエンドポイントに合わせる
paths:
  /_search:
    get:
      summary: Search documents in Elasticsearch
      description: Executes a search query against an Elasticsearch index.
      parameters:
        - in: query
          name: index
          schema:
            type: string
          required: true
          description: The name of the Elasticsearch index to search.
        - in: query
          name: q
          schema:
            type: string
          description: Query in Lucene query string syntax.
        - in: query
          name: query
          schema:
            type: object
          description: Query in Query DSL format (JSON).  # より複雑なクエリ用
      responses:
        '200':
          description: Successful response with search results.
          content:
            application/json:
              schema:
                type: object  # Elasticsearch の検索結果の構造を定義することも可能 (省略)
        '400':
          description: Bad Request. Invalid query parameters.
        '500':
          description: Internal Server Error. Elasticsearch error.
"""

# OpenAPI Specification を YAML として読み込む
spec_es = yaml.safe_load(elasticsearch_schema)

# サーバーURLを取得
server_url_es = spec_es['servers'][0]['url']

# Elasticsearch クライアントを初期化
es = Elasticsearch(server_url_es)

def execute_elasticsearch_search(index_name, query_string=None, query_dsl=None):
    """
    OpenAPI スキーマに基づいて Elasticsearch の検索を実行する関数

    Args:
        index_name (str): Elasticsearch のインデックス名
        query_string (str, optional): Lucene クエリ文字列 (q パラメータ). Defaults to None.
        query_dsl (dict, optional): Query DSL (JSON 形式) (query パラメータ). Defaults to None.

    Returns:
        dict: Elasticsearch の検索結果 (成功時)
        None: エラーが発生した場合
    """
    params = {}
    if query_string:
        params['q'] = query_string
    if query_dsl:
        params['query'] = query_dsl

    res = es.search(index=index_name, query=params.get('query'), q=params.get('q'))
    return res


if __name__ == "__main__":
    index_name = "your_index_name" # 検索対象のインデックス名
    query_string = "occupation.name:デザイナー" # Lucene クエリ文字列での検索 (q パラメータ)
    query_dsl = {"match": {"occupation.name": "デザイナー"}} # Query DSL での検索 (query パラメータ)

    search_result = execute_elasticsearch_search(index_name, query_string=query_string) # または query_dsl=query_dsl を使う

    if search_result:
        print(f"検索結果 (インデックス: {index_name}, クエリ: {query_string if query_string else query_dsl}):")
        print(f"該当ドキュメント数: {search_result['hits']['total']['value']}")
        for hit in search_result['hits']['hits']:
            print(f"ID: {hit['_id']}, Source: {hit['_source']}") # ドキュメント全体を出力するように変更
    else:
        print(f"検索に失敗しました。")