Zenn
🚀

【Vertex AI】AI エージェント開発の実践から学んだ設計と最適化のポイント

2024/10/02に公開
Google Cloud
Vertex AI
agent
Gemini
tech

こんにちは、クラウドエース CTO 室所属の茜です。
今回は、Vertex AI を活用した AI エージェント開発で得た知見を共有します。

この開発で直面した課題とその解決策を紹介することで、皆さんの AI エージェント開発の一助となれば幸いです。

はじめに

本記事では、社内 PoC として取り組んだ「ニュース記事から企業のリスク分析を行う AI エージェント」の開発プロセスについて解説します。

説明すること

  • AI エージェントの概要
  • 開発過程で直面した技術的課題とその解決策
  • 精度向上のためのアプローチ

説明しないこと

  • AI エージェントの詳細な構築方法
  • LangChain、LangGraph を使用した構築方法
  • ニュース記事取得、データ表示に関する詳細

AI エージェントとは

ラスベガスで行われた Google Cloud Next ' 24 において、Google Cloud CEO であるトーマス・クリアン氏は、「エージェントとは、特定の目標を達成するためにエンドユーザーに代わって行動するインテリジェントエンティティである。エージェントは人間とつながり、アプリケーションのユーザーに代わって、文字、画像、音声、動画のように形式の異なる情報を同時に処理し、対話、推論、学習、意思決定を行う」と説明しています。

また、AWS 公式サイトでは、以下のように AI エージェントが説明されています。
「人工知能 (AI) エージェントは、環境と対話し、データを収集し、そのデータを使用して自己決定タスクを実行して、事前に決められた目標を達成するためのソフトウェアプログラムです。目標は人間が設定しますが、その目標を達成するために実行する必要がある最適なアクションは AI エージェントが独自に選択します。」

したがって、AI エージェントとは、外部情報を基に自律的にアクションを決定し、特定の目標を達成するためのソフトウェアプログラム (システム) と言えます。

AI エージェントの詳細については、以下の記事が参考になります。
https://zenn.dev/dev_commune/articles/b93a7296fbb354

リスク分析エージェントの概要

今回作成した AI エージェントの概要を説明します。
詳細な実装方法は記載しませんのでご了承ください。

本エージェントは、ニュース記事内容が企業にとってリスクがあるかを判断し、リスクがあると判断した場合、詳細な分析を実行します。

主要な機能

本エージェントは、主に 3 つのコンポーネントで構成されています。
本記事では、エージェントの肝となる部分である記事データ分析に関して説明します。

  • ニュース記事取得
    News API から記事データを日毎に取得し、BigQuery に格納します。
  • 記事データ分析
    取得した記事データに対して分析を行います。詳細は後述します。
  • 分析データ表示
    分析結果をレポート形式で出力します。

アーキテクチャ

下図は本エージェントのアーキテクチャです。
以下フローで処理が実行されます。

  1. Cloud Scheduler で Cloud Workflows を定期実行する
  2. ニュースデータを取得する Cloud Run Jobs が実行される
  3. 記事データを分析する Cloud Run Jobs が実行される
  4. ユーザーが分析結果を UI 上で閲覧する


アーキテクチャ

処理フロー

記事データ分析において、以下フローで処理を実行します。


記事データ分析フロー

  1. BigQuery から記事データを取得
    毎日約 100 件のデータが格納されます。後続の LLM 処理は記事データごとに実行します。
  2. LLM 処理 1 を実行
    • 記事概要作成
    • 記事データのスクリーニング
      記事内容に指定した観点が含まれているか判断します。(「政治的な出来事、政策変更、国際関係の変化など、ビジネス環境に影響を与える可能性のある政治的要因について触れられているか」等)
      今回は 10 個の観点を指定しました。
  3. LLM 処理 2 を実行 (ベンダーに対するリスク処理)
    • 記事に対するリスク評価
      記事内容がベンダーにとってリスクとなるか判断します。
    • クライアント企業に対するリスク評価
      ベンダーにとってのリスクがクライアント企業にとってリスクとなるか判断します。
    • リスクレベル判断
      High、Middle、Low の 3 段階でリスクレベルを評価します。
    • 緩和措置提案
      リスクに対してクライアント企業がどのように対処すべきかを提案します。
  4. LLM 処理 3 を実行 (クライアント企業に対するリスク処理)
    • 記事に対するリスク評価
      記事内容がクライアント企業にとってリスクとなるか判断します。
    • リスクレベル判断
      同上
    • 緩和措置提案
      同上
  5. LLM 処理後の記事データ、分析結果を BigQuery に格納

エージェント作成時の考慮事項

記事データ分析において、考慮した点、懸念点を紹介します。
なお、これらは個人的な見解に基づくものであり、必ずしもベストプラクティスではない点にご留意ください。

JSON モード

LLM 処理において、Vertex AI 経由で Gemini API にリクエストを行なっています。
Gemini API へのリクエスト方法に関しては以下記事を参考にしてください。
https://zenn.dev/cloud_ace/articles/e5c13b4fdaeb42

Gemini API には JSON モードという機能があります。(GPT、Claude でも同様の機能があります)
この機能を使用することにより、LLM の出力を任意の JSON スキーマに固定することができます。

AI エージェントでは、前処理の出力結果を基に後続のアクションを決定します。この決定プロセスでは、LLM の出力値に基づいて条件分岐を行います。
しかし、LLM の出力形式が固定されていない場合、期待する形式と異なる出力が生成される可能性があり、結果として適切な後続処理が行われないリスクがあります。

プロンプト内で出力形式を指示する方法もありますが、LLM の特性上、完全な一貫性を保証することは困難です。

そこで、JSON モードを使用して出力を固定化することが、より信頼性の高いアプローチだと考えられます。
特に今回のようにライブラリを使用せずに AI エージェントを構築する場合、この方法は出力の一貫性を確保し、後続処理の安定性を高める上で非常に有効です。

さらに、JSON 形式で出力が取得できるため、出力結果の任意の値を他の関数に簡単に受け渡すことができます。
これにより、処理の連携がスムーズになり、システム全体の柔軟性と拡張性が向上します。
例えば、ある処理の出力結果を次の処理の入力として使用する際、JSON 形式であれば必要な値を容易に抽出し、パラメータとして渡すことができます。

以下は、JSON モード使用時に指定したスキーマとその出力値に応じた条件分岐の一例です。

LLM 処理 1 の出力スキーマ
INITIAL_ANALYSIS_SCHEMA = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "initial_screening": {
            "type": "array",
            "items": {
                "type": "string",
                "enum": [
                    "Operational Risks",
                    "PoliticalInstability",
                    "Financial Risks",
                    "Reputational Risks",
                    "Sustainability",
                    "CyberThreat",
                    "ConsumerBehaviorChange",
                    "CompetitiveIntelligence",
                    "RegulatoryChange",
                    "TechInnovation",
                    "None",
                ],
            },
        },
    },
    "required": ["summary", "initial_screening"],
}
出力結果に応じた条件分岐
# initial_screeningの値がNoneでない場合、追加の分析を実行
if initial_result["initial_screening"] != ["None"]:
    logger.info(
        f"Initial screening for '{content_dict['title']}' identified potential risks: {', '.join(initial_result['initial_screening'])}."
    )
    # ベンダー分析実行
    vendor_risk_result = vendor_risk_assessment(
        content_dict["fullContent"], vendor_info, initial_result["initial_screening"]
    )

    # クライアント企業分析実行
    network_ace_risk_result = network_ace_risk_assessment(
        content_dict["fullContent"], initial_result["initial_screening"]
    )
    ...

並行処理

AI エージェントの開発において、LLM へのリクエスト数が多くなるのは避けられません。
特に今回のように複数の記事データに対して処理を行う場合、リクエスト数は記事数や処理数に比例して増加するため、全体の処理時間が非常に長くなります。

これに対処するため、並行処理を実装し、複数のリクエストを同時に実行できるようにしました。これにより、処理時間を大幅に短縮することができました。

以下は、並行処理の実装箇所です。

並行処理
def process_articles_and_assess_risks() -> None:
    try:
        logger.info("Starting article processing and risk assessment")
        content_list = get_content_from_bigquery()
        vendor_info = get_vendor_info_from_bigquery()
        current_date = datetime.datetime.now().isoformat()

        article_output_list = []
        vendor_result_list = []

        with ThreadPoolExecutor() as executor:
            future_to_content = {
                executor.submit(
                    process_content, content_dict, current_date, vendor_info
                ): content_dict
                for content_dict in content_list
            }
            for future in as_completed(future_to_content):
                content_dict = future_to_content[future]
                try:
                    article_output, vendor_results = future.result()
                    article_output_list.append(article_output)
                    vendor_result_list.extend(vendor_results)
                except Exception as e:
                    logger.error(
                        f"Error processing content '{content_dict['title']}': {str(e)}",
                        exc_info=True,
                    )
        ...

並列処理、並行処理に関しては、以下記事が参考になります。
https://zenn.dev/ryo_kawamata/articles/python-concurrent-thread
https://kosuke-space.com/python-parallel-processing

割り当て制限

並行処理により処理時間が短縮できましたが、それにより新たに発生した問題があります。
それが、Vertex AI の割り当て上限です。

Vertex AI では、1 分当たりのリクエスト数に制限があり、これを超えるとエラーが発生します。
(リージョン：asia-northeast1、モデル：gemini-1.5-pro の場合、60 回/分)
並行処理によって多数のリクエストが短時間に集中すると、この制限に容易に達してしまいます。

この問題に対処するため、以下のアプローチを取りました。

  1. スロットリングの実装
    リクエスト回数が制限に近づいた段階で、一定時間リクエストを待機させる処理を実装しました。
    今回は、リクエスト回数が 50 回に達すると 1 分間待機するようにしています。
  2. 割り当ての増加リクエスト
    Google Cloud コンソール上から、割り当て上限の増加をリクエストすることができます。
    今回はリクエストを行い、gemini-1.5-pro の割り当て上限を 200 回/分に変更しました。

割り当て制限に対する対処として 2 つのアプローチを取りましたが、リクエスト数の上限が事前に把握できている場合は、割り当ての増加リクエストのみの対応で良いかと思います。
ただし、以下記載にもあるように必ずしもリクエストが承認されるわけではないようです。

割り当て増加リクエストのほとんどは、自動システムによって評価されます。その決定は、リソースの可用性、Google Cloud の使用期間などの基準に基づいて行われます。基準を満たさないリクエストは拒否されます。

LLM 処理の分割単位

本エージェントの開発において、最も難しい課題が LLM 処理の分割単位の決定でした。

処理フローで説明したとおり、本エージェントは計 9 つの異なるタスクを実行します。
これらの複数タスクを持つエージェントを設計する際、LLM へのリクエストをどのような単位で分割するかが重要な検討事項となりました。

現状では、最適な分割方法に関する明確な指針は確立されていないと認識しています。
(知見をお持ちの方がいらっしゃいましたら、ぜひご意見をお聞かせください)
これは、AI エージェント開発が比較的新しい技術領域であり、業界全体でもベストプラクティスが十分に確立されていないことに起因すると考えます。

今回は以下の要素を考慮し、分割単位を決定しました。

処理の結果に応じたアクション変更への対応

AI エージェントでは、あるタスクの結果によって後続のタスクのアクションが決定されます。
このような条件分岐の指示をプロンプトで与えるのか、またはプログラムで記述するのかを考えました。

本エージェントの一部分では、プロンプトで条件分岐の指示を与えています。

プロンプトでの条件分岐指示
<処理手順>
1. リスク判断 (vendor_risks)
...
a) ベンダーのリスク評価 (risk_reason):
...
b) クライアント企業へのリスク判断:
...
c) 最終判断:
...
d) リスクレベル判断 (risk_level)
# 条件分岐指示
最終判断において、リスクがあると判断した場合のみこの処理を行ってください。リスクがないと判断した場合はNoneを出力してください。
...
e) 緩和措置提案 (measure)
# 条件分岐指示
最終判断において、リスクがあると判断した場合のみこの処理を行ってください。リスクがないと判断した場合はNoneを出力してください。
...
</処理手順>

この場合でも現段階では正常に処理が実行されているのですが、処理を分割し、前処理の出力結果に応じて次処理を決定する条件分岐をプログラムで書いた方が、より確実に意図したフローになると考えます。

プログラムでの条件分岐
def judge_risk(hoge):
    {ベンダーのリスク評価、クライアント企業へのリスク判断、最終判断を行う処理}
    return result

def process_risk_judgment(result)
    if result["judge"]:
        {リスクレベル判断、緩和措置提案を行う処理}
        return fuga
    else:
        return None

段階的処理の活用

前タスクの結果が次タスクに有益な情報をもたらす場合、これらのタスクは集約した方が精度が高くなる可能性があります。
タスクに対して段階的な考慮 (Step-by-Step) が行われると考えるためです。

例えば先ほどのプロンプトでは、ベンダーのリスク評価、クライアント企業へのリスク判断、最終判断を行った後、リスクレベル判断を行っています。

これにより、前タスクの結果を基にした判断が行われるため、リスクレベル判断単体で実行するよりも精度が高くなるのではないかと考えます。

処理分割による懸念点

処理を分割した場合、以下のような懸念点が挙げられます。

  • インプット総量の増加
    タスクを分割した場合、同じデータ (記事内容やベンダー情報など) を必要とする複数の処理単位が生じる可能性があります。
    その結果、各処理単位のプロンプトに同一のデータを繰り返し含めることになります。
    これにより、LLM へのインプット総量が増加し、処理効率の低下やコストの増加につながります。

    データインプットイメージ

  • 処理速度低下
    タスクの分割は、必然的に LLM へのリクエスト数の増加を引き起こします。
    各リクエストには一定の処理時間とオーバーヘッドが発生するため、全体の処理速度は低下します。
    特に、今回のように多数の記事を処理する場合、この影響は無視できないものとなります。

現状の処理単位分割方針

これらの考慮を行い、今回は以下のように処理単位を分割しました。

処理分割単位

処理の結果に応じたアクション変更への対応において、条件分岐をプログラムで書いた方が良いと述べましたが、LLM 処理 2 と LLM 処理 3 では以下の点からプロンプトで指示する方法をとっています。

  • 現状は正常に処理が行われている
  • 分割した場合、段階的処理が行われない
  • 分割による懸念点がある

一方、処理単位間の条件分岐はプログラムで記述しています。
あくまで一例のため、最適な分割方法が他にもあるかもしれません。

精度評価

本エージェントの精度を評価するにあたり、「リスク判断が本質的に難しいタスクである」という点を考慮する必要があります。
人間が判断する場合も、同じ記事内容に対して異なる判断を下すことがあります。この人間の判断の多様性を念頭に置きつつ、評価を行いました。

評価の結果、特に注目すべき点として LLM 処理 2 (ベンダーに対するリスク処理) と LLM 処理 3 (クライアント企業に対するリスク処理) におけるリスク評価の精度が期待を下回る結果となりました。
これらの処理では、記事内容とベンダーまたはクライアント企業の情報を適切に関連付けて分析することが求められますが、いくつかのケースで適切な結果が得られませんでした。

本エージェントは、リスクの「見逃し」を最小限に抑えることを主眼に設計されています。
そのため、影響が比較的小さいと思われる記事でもリスクとして判断する傾向があります。
これは意図的なアプローチであり、潜在的なリスクを広く検出することで、人間が最終判断を行う際の「スクリーニングツール」として機能することを目指しています。

このアプローチには利点がある一方で、以下のような課題もありました。

的外れな評価の発生

記事のスクリーニングにおいて、本来であれば指定した観点を含んでいないと判断されるべき記事内容を誤って観点を含むと判断し、その後のベンダーやクライアント企業に対するリスク評価でも不適切にリスクがあると判断するケースが見られました。

記事概要 リスク有無 判断理由
この記事は、アメリカンフットボールのテキサス・ロングホーンズがUTSAロードランナーズに勝利し、16年ぶりにカレッジフットボールのランキングで1位になったことを報じています。記事では、テキサスがランキング1位になった経緯、試合中の出来事、クォーターバックの交代劇などが詳細に描写されています。また、ランキング2位に転落したジョージア・ブルドッグスの試合内容や、その他のチームのランキング変動についても触れられています。 True ZTEは、クライアント企業Aに対して通信機器や5G技術を提供する主要ベンダーの一つです。記事では、競合他社であるノキアやエリクソンも5G技術で高いプレゼンスを持つことが示唆されており、競争の激化が予想されます。ZTEが競争に敗れた場合、クライアント企業Aは、製品調達や技術導入において、より不利な条件を強いられる可能性があります。また、ZTEの技術革新の遅れは、クライアント企業Aの競争力低下にも繋がる可能性があります。

また、以下のような分析精度の課題もありました。

ベンダー間での評価基準の不一致

同様の事業内容や地理的条件を持つベンダー間であっても、同一の記事内容に対して異なる評価を行うケースが見られました。

以下の例では、ZTE と Huawei Technologies という類似した事業を行い、地理的にも近い位置に主要拠点を持つ 2 つのベンダーに対して、同じ台風関連の記事に基づいて異なるリスク評価が行われています。

このような不一致は、AI エージェントが各ベンダーの地理的位置や事業の詳細を適切に考慮できていないことを示唆しています。

記事概要 リスク有無 判断理由
大型で強い台風14号が9月18日、沖縄県や鹿児島県の奄美地方に最も接近する見込みです。沖縄・奄美では風雨が強まる見込みで、最大風速は沖縄地方で20メートル、最大瞬間風速は30メートルと予想されています。波が高くうねりを伴うしけ、土砂災害、低い土地の浸水、川の増水、落雷や竜巻などのリスクがあります。気象庁は早めの備えを呼びかけています。 True ZTEはクライアント企業Aに通信機器や5G技術を提供しているため、台風によるZTEの中国工場の操業停止やサプライチェーンの混乱は、クライアント企業Aへの機器供給に遅延が生じ、通信サービスの展開や保守に影響を与える可能性があります。

Huawei Technologiesはクライアント企業Aに5Gネットワーク機器などを供給していますが、記事内容がHuawei Technologiesの事業に直接的な影響を与えるとは考えにくいです。台風によるHuawei Technologiesの工場被災の可能性は低いと考えられます。

精度向上のためのアプローチ

上記の課題に対して、以下のようなアプローチが有効であると考えています。

プロンプト改善

出力結果を基にプロンプトを継続的に修正していくことが重要です。
エージェントの応答を分析し、不適切な判断や誤った情報が含まれている場合、プロンプトを調整して精度を向上させていきます。

この反復的なプロセスにより、エージェントの判断基準をより適切なものに改善できると考えます。

ベンダーごとの処理方法の最適化

現状では、以下のようにプロンプトで複数ベンダーに対する反復処理を指示しています。

しかし、これをプログラムで反復処理するように制御した方が分析精度が向上するのではないかと考えます。
(ベンダー間での評価基準の不一致は、プロンプトによる指示が原因となっている可能性があります。)

プロンプトでの反復処理指示
<使用データ>
#ベンダー情報
- 説明: クライアント企業 A に製品やサービスを提供するベンダーの詳細情報。各ベンダーの概要、所在地情報、およびクライアント企業 A との関係性が含まれている。
- データ:
{
    以下形式で複数ベンダーの情報が格納されています。
    ベンダー名 : hogehoge
    詳細情報 : fugafuga

    ベンダー名 : piyopiyo
    詳細情報 : foofoo
    ...
}
...
</使用データ>

<処理手順>
1. リスク判断 (vendor_risks)
# 反復処理指示
ベンダー情報に記載されている各外部ベンダーに対して、以下の分析を行ってください：

a) ベンダーのリスク評価 (risk_reason):
...
b) クライアント企業へのリスク判断:
...
c) 最終判断:
...
</処理手順>
プログラムでの反復処理
def assess_vendor_risk(vendors):
    for vendor in vendors:
        {ベンダーのリスク評価、クライアント企業へのリスク判断、最終判断を行う処理}

ただし、この方法には以下のようなトレードオフがあります。

  • リクエスト数がベンダー数に比例して増加し、コストと処理時間が増大する
  • Gemini-1.5-Flash 等のモデルを使用すればコスト、処理速度の問題は軽減できるが、Gemini-1.5-Pro と比較すると精度が若干低下する可能性がある

これらの要素を考慮し、コスト、処理速度、精度のバランスを取りながら最適なアプローチを選択する必要があります。

JSON モードの使用方法の再考

JSON モードを使用すると推論タスクの精度が低下するという研究結果があります。
https://arxiv.org/abs/2408.02442
この問題に対処するため、以下のような 2 段階アプローチが効果的であると論文では述べられています。

  • 最初の分析を JSON モードを使用せずに実行し、LLM の推論能力を最大限に活用する
  • 得られた結果を JSON モードを使用して固定フォーマットに変換し、後続の処理で扱いやすくする

このアプローチにより、推論の柔軟性を保ちつつ、出力の一貫性も確保できると考えます。

分析結果評価処理の追加

LLM の分析精度をさらに向上させるため、LLM-as-a-Judge (LLM に LLM の出力を評価させる手法) の導入も検討すべきです。

LLM-as-a-Judge に関しては以下記事が参考になります。
https://zenn.dev/gaudiy_blog/articles/b34345aab2949e

LLM-as-a-Judge にはさまざまな手法が存在しますが、例えば以下のようなアプローチが考えられます。

スコアベースでのアプローチ

  1. 評価基準を決める
    分析結果の質を判断するための基準を設定します。
    例：
    リスク分析の妥当性：リスクの判断が適切かどうか
    根拠の具体性：判断の理由が明確に示されているか
    論理の一貫性：説明に矛盾がないか
    考慮された要因の網羅性：重要な点が見落とされていないか

  2. LLM に出力を評価させる
    それぞれの基準について、10 点満点で点数をつけてもらいます。
    例：
    「リスク分析の妥当性は 8 点、根拠の具体性は 7 点...」

  3. 基準点との比較
    あらかじめ決めておいた基準点（例：30 点）と比較します。

  4. 結果に応じたアクション
    基準点を超えていれば、その分析結果を採用します。
    基準点に達していなければ LLM に改善点を出力させ、分析をやり直します。


スコアベースイメージ (https://www.brainpad.co.jp/doors/contents/01_llm_as_a_judge/)

このように LLM 自身に結果を評価させることで、分析の質を客観的に判断し、必要に応じて改善するサイクルを作ることができます。

まとめ

本記事では、Vertex AI を用いたリスク分析 AI エージェントの開発における主要な考慮点と、精度向上のための施策について解説しました。

AI エージェント開発においては、JSON モードの活用による出力の一貫性確保、並行処理の実装による効率化、API の割り当て制限への対応が重要だと考えます。
また、LLM 処理の分割単位の決定は、条件分岐の実装方法や段階的処理の活用などを考慮しながら慎重に行う必要があります。

精度評価では、リスク判断の本質的な難しさを踏まえつつ、的外れな評価やベンダー間での評価基準の不一致などの課題が明らかになりました。
これらの課題に対し、プロンプトの継続的な改善、ベンダーごとの処理方法の最適化、JSON モードの使用方法の再考、そして LLM-as-a-Judge の導入などのアプローチが有効であると考えられます。

AI エージェント開発は比較的新しい技術領域であり、ベストプラクティスが確立されていない面もあります。
そのため、継続的な改善と評価を行いながら、最適な実装方法を模索していくことが重要です。

今後は、LangChain や LangGraph などの AI エージェント開発用ライブラリを使用して構築した場合との比較を行い、それぞれのアプローチの長所短所を明らかにしていきたいと考えています。

本記事で紹介した考慮点や施策が AI エージェント開発の一助となれば幸いです。

Discussion