AIエージェント運用費がほぼ半減した、マルチエージェントへの移行事例

「AIエージェント構築＆運用 Advent Calendar 2025」1日目の記事です。

こんにちは、ENECHANGE VPoTの岩本 (iwamot) です。

AIエージェントの運用費、なるべく抑えたいですよね。

もし複数ステップを処理させているシングルエージェントがあれば、マルチエージェント構成に変えることで、費用を大きく減らせるかもしれません。

本記事では、移行によって費用を43%減らした実例を詳しくお伝えします。

対象のAIエージェント
当初の実装
- シングルエージェントの欠点
現在の実装
マルチエージェント移行による効果
- 移行後のトレース結果
- 移行が効くケース
マルチエージェント構成の注意点
Strands Agentsでのトレース方法
おわりに

対象のAIエージェント

対象のAIエージェントは「ブログほめ太郎」といいます。当ブログの新着記事をレビューし、Slackに投稿してくれるエージェントです。

処理のステップは以下の通りです。

最後にレビューした記事のURLをAWSのパラメータストアから取得
当ブログのRSSを参照
新着記事がなければ終了
すべての新着記事を古い順にループ
1. 記事全文を参照
2. レビューメッセージを生成
3. Slackに投稿
最後にレビューした記事のURLを更新

今回、マルチエージェント構成に切り替えましたが、これらのステップは変えていません。

当初の実装

運用を始めたときの実装は、とてもシンプルなものでした。任せたい処理をまるごとプロンプトにし、エージェントに渡していました。

当初のコード全文（クリックで開閉）

import os

from strands import Agent
from strands.models import BedrockModel

# /var/task のままだと、slackツールのimport時にディレクトリ作成エラーになる
# https://github.com/strands-agents/tools/issues/199
os.chdir("/tmp")

from strands_tools import http_request, rss, slack, use_aws

PROMPT = f"""
あなたは技術ブログの記事を読んで、筆者を励ますレビューを書くAIエージェントです。
下記の手順でレビューし、結果をSlackに投稿してください。

1. use_awsツールを使い、最後にレビューした記事のURLをパラメータストアから取得する
  - リージョン：{os.environ['AWS_REGION']}
  - パラメータ名：{os.environ['BHT_LAST_REVIEWED_ARTICLE_PARAM']}
  - パラメータが見つからないこともある

2. RSSフィード {os.environ['BHT_RSS_URL']} を参照する
  - 最後にレビューした記事より後に公開された記事を取得する
  - ステップ1でパラメータが見つからなかった場合は、最新記事1件を取得する

3. 対象記事がなければ、レビューせずに終了する

4. 以下、公開日時の古い記事から繰り返す
  4-1. http_requestツールで当該記事の全文を参照する
  4-2. 日本語400字程度で、技術的に価値のあるレビューメッセージを生成する
    - 「<!channel> 新着記事レビューです！」から始める
    - 記事タイトルとURLを紹介する
    - 投稿に感謝する。筆者名が分かれば名前で呼びかける（会社名やチーム名は不要）
    - 絵文字や *アスタリスク1つ* を使って適度に強調する
      - 太字にしたい場合は、 *アスタリスク1つだけ* で囲むこと
      - いかなる場合も、アスタリスクの前後には必ずスペースを入れること
      - アスタリスク2つ（`**`）は絶対に使わないこと
  4-3. Slackチャンネル {os.environ['BHT_SLACK_CHANNEL_ID']} に投稿する

5. use_awsツールで、最後にレビューした記事のURLを更新する

技術的レビューの観点例：
- 読者が得られる学び
- 技術的な新規性や独自性
- さらに深掘りできる内容の提案
"""

def lambda_handler(_event, _context):
    agent = Agent(
        model=BedrockModel(cache_tools="default"),
        tools=[http_request, rss, slack, use_aws],
    )
    response = agent(PROMPT)
    return str(response)

コードの特徴を数字にすると、こうです。

行数：53
エージェント数：1

シングルエージェントの欠点

このような実装がトークンを浪費しがちだと気づいたのは、運用開始後しばらく経ってからでした。

なんとなく気になってトレースしたところ（方法は後述）、以下の結果になってしまいました。

LLM呼び出し	入力	出力	キャッシュ書き込み	キャッシュ読み込み
1	791	216	3,367	0
2	1,332	195	0	3,367
3	2,602	254	0	3,367
4	15,968	600	0	3,367
5	16,595	240	0	3,367
6	17,038	392	0	3,367
合計	54,326	1,897	3,367	16,835

ポイントは下記の部分です。

記事全文のトークンが10,000を超えている
- レビュー生成 (4) で利用
レビュー生成後も、使い終えた記事全文をLLMに渡し続けている
- Slack投稿 (5) で必要なのは「レビュー結果」だけ
- パラメータストア更新 (6) で必要なのは「最後にレビューした記事のURL」だけ

新着記事が1件のケースでこうなのですから、記事が多ければ大変なことになってしまいます。

実は、複数ステップのフローをStrands Agentsなどのフレームワークでシンプルに実装すると、途中のステップで大きなデータを使い終えても会話履歴に残して後続ステップに引き継いでしまう、いわば「富豪的シングルエージェント」になりがちなのです。

一方、マルチエージェント構成なら、各エージェントに「初めて話しかける」形で処理できます。たとえば、こんなイメージです。

「Slack投稿エージェントさん、レビュー結果を渡すから、Slackに投稿して」
「パラメータストア更新エージェントさん、最後にレビューした記事URLを渡すから、パラメータストアに保存して」

これらのエージェントには記事全文は必要ありません。彼らの役目が果たせる最小限のデータさえ渡せば十分です。

現在の実装

以上の背景により、ブログほめ太郎をマルチエージェント構成に切り替えました。

現在のコード全文（クリックで開閉）

import gc
import os
from typing import List

from pydantic import BaseModel, Field
from strands import Agent
from strands.models import BedrockModel

# /var/task のままだと、slackツールのimport時にディレクトリ作成エラーになる
# https://github.com/strands-agents/tools/issues/199
os.chdir("/tmp")

from strands_tools import http_request, rss, slack, use_aws

NEW_ARTICLE_FINDER_PROMPT = f"""
あなたは新着記事のURLリストを取得する専門エージェントです。

1. use_awsツールを使い、最後にレビューした記事のURLをパラメータストアから取得する
  - リージョン：{os.environ['AWS_REGION']}
  - パラメータ名：{os.environ['BHT_LAST_REVIEWED_ARTICLE_PARAM']}
  - パラメータが見つからないこともある

2. RSSフィード {os.environ['BHT_RSS_URL']} を参照する
  - 最後にレビューした記事より後に公開された記事を取得する
  - ステップ1でパラメータが見つからなかった場合は、最新記事1件を取得する

3. 取得した記事のURLのみをシンプルなリストとして返す（古い順）
  - 例：["https://example.com/article1", "https://example.com/article2"]
"""

# convert_to_markdown=True を指定しないとHTMLで全文参照され、
# トークン量が増えることがある
ARTICLE_REVIEWER_PROMPT = """
あなたは技術ブログ記事をレビューする専門エージェントです。

1. http_requestツールで当該記事の全文を参照する
  - convert_to_markdown=Trueを必ず指定すること

2. 日本語400字程度で、技術的に価値のあるレビューメッセージを生成する
  - 「<!channel> 新着記事レビューです！」から始める
  - 記事タイトルとURLを紹介する
  - 投稿に感謝する。筆者名が分かれば名前で呼びかける（会社名やチーム名は不要）
  - 絵文字や *アスタリスク1つ* を使って適度に強調する
    - 太字にしたい場合は、 *アスタリスク1つだけ* で囲むこと
    - いかなる場合も、アスタリスクの前後には必ずスペースを入れること
    - アスタリスク2つ（`**`）は絶対に使わないこと
  - レビューメッセージ以外の出力は不要

技術的レビューの観点例：
- 読者が得られる学び
- 技術的な新規性や独自性
- さらに深掘りできる内容の提案
"""

SLACK_PUBLISHER_PROMPT = f"""
あなたはレビューをSlackに投稿する専門エージェントです。

1. slackツールで投稿する
  - Slackチャンネル：{os.environ['BHT_SLACK_CHANNEL_ID']}
"""

LAST_ARTICLE_RECORDER_PROMPT = f"""
あなたは最後にレビューした記事URLを更新する専門エージェントです。

1. use_awsツールでパラメータストアを更新する
  - リージョン：{os.environ['AWS_REGION']}
  - パラメータ名：{os.environ['BHT_LAST_REVIEWED_ARTICLE_PARAM']}
  - 値：指定された記事URL
"""


class NewArticleFinderOutput(BaseModel):
    urls: List[str] = Field(
        default_factory=list, description="新着記事のURLリスト（古い順）"
    )


def lambda_handler(_event, _context):
    new_article_finder = Agent(
        model=BedrockModel(cache_tools="default"),
        tools=[use_aws, rss],
        system_prompt=NEW_ARTICLE_FINDER_PROMPT,
    )
    new_article_finder("新着記事のURLリストを取得してください")
    finder_output = new_article_finder.structured_output(NewArticleFinderOutput)
    del new_article_finder
    gc.collect()
    urls = finder_output.urls
    print(urls)

    if not urls:
        return "新着記事はありません"

    for url in urls:
        article_reviewer = Agent(
            model=BedrockModel(cache_tools="default"),
            tools=[http_request],
            system_prompt=ARTICLE_REVIEWER_PROMPT,
        )
        reviewer_output = article_reviewer(f"この記事をレビューしてください：{url}")
        del article_reviewer
        gc.collect()

        slack_publisher = Agent(
            model=BedrockModel(cache_tools="default"),
            tools=[slack],
            system_prompt=SLACK_PUBLISHER_PROMPT,
        )
        slack_publisher(f"以下のレビューを投稿してください：\n\n{reviewer_output}")
        del slack_publisher
        gc.collect()

    last_article_recorder = Agent(
        model=BedrockModel(cache_tools="default"),
        tools=[use_aws],
        system_prompt=LAST_ARTICLE_RECORDER_PROMPT,
    )
    recorder_output = last_article_recorder(f"記事URLを {urls[-1]} に更新してください")
    del last_article_recorder
    gc.collect()

    return str(recorder_output)

当初と同じ処理を、4つのエージェントで実現しています。分担は以下の通りです。

最後にレビューした記事のURLをパラメータストアから取得 (new_article_finder)
当ブログのRSSを参照 (同上)
新着記事がなければ終了 (プログラムで処理)
すべての新着記事を古い順にループ (同上)
1. 記事全文を参照 (article_reviewer)
2. レビューメッセージを生成 (同上)
3. Slackに投稿 (slack_publisher)
最後にレビューした記事のURLを更新 (last_article_recorder)

コードの変化を数字にすると、こうなります。

行数：53 → 122
エージェント数：1 → 4

複雑にはなりましたが、これで無駄なトークンが減るなら御の字です。

マルチエージェント移行による効果

移行の結果、入力トークンと費用が大きく減りました（利用中のClaude Sonnet 4.5で算出）。

トークンの種類	移行前の量	移行後の量	移行前の費用	移行後の費用
入力	54,326	25,568	$0.163	$0.077
出力	1,897	2,091	$0.028	$0.031
キャッシュ書き込み	3,367	2,804	$0.013	$0.011
キャッシュ読み込み	16,835	2,804	$0.005	$0.001
合計			$0.209	$0.120

入力トークンは53%減、合計費用は43%減。実行ごとにレビュー対象記事や生成結果が変わるため、あくまでサンプル比較ではありますが、効果は明らかです。

移行後のトレース結果

現在の実装をトレースすると、こんな結果になります。

LLM呼び出し	入力	出力	キャッシュ書き込み	キャッシュ読み込み
1	1,323	219	0	0
2	1,875	200	0	0
3	4,335	156	0	0
4	553	113	1,219	0
5	13,779	452	0	1,219
6	665	552	1,585	0
7	1,244	97	0	1,585
8	691	225	0	0
9	1,103	77	0	0
合計	25,568	2,091	2,804	2,804

ご覧の通り、記事全文はレビュー生成 (5) で使われているだけです。後続のステップには、もう渡っていません。

エージェントを分けたことで、LLM呼び出しが6回から9回に増えましたが、会話が効率的になり、入力トークンの合計が激減しています。

移行が効くケース

ということで、以下の特徴をもつシングルエージェントなら、マルチエージェント化で費用削減が見込めます。

複数のステップでフローを構成している
途中のステップで大きなデータを使い、その後は不要となる

もし手元にあれば、マルチエージェント化を検討しましょう。

マルチエージェント構成の注意点

ただし、マルチエージェントには注意すべき点があります。メモリ不足です。エージェントが増えるにつれ、メモリ使用量も増えるためです。

現在の実装では、エージェントの役目が終わったら、以下のようにすぐ解放しています。

        article_reviewer = Agent(
            model=BedrockModel(cache_tools="default"),
            tools=[http_request],
            system_prompt=ARTICLE_REVIEWER_PROMPT,
        )
        reviewer_output = article_reviewer(f"この記事をレビューしてください：{url}")
        del article_reviewer
        gc.collect()

「常にそうすべき」とまでは思いませんが、気になる場合は明示的に解放しましょう。

Strands Agentsでのトレース方法

最後にトレースについてです。

「ブログほめ太郎」で使っているStrands Agentsでは、OpenTelemetryによるトレースが可能です。ローカル開発では、以下の手順で出力できます。

1. Jaegerコンテナを起動

docker run -d --name jaeger \
  -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
  -e COLLECTOR_OTLP_ENABLED=true \
  -p 6831:6831/udp \
  -p 6832:6832/udp \
  -p 5778:5778 \
  -p 16686:16686 \
  -p 4317:4317 \
  -p 4318:4318 \
  -p 14250:14250 \
  -p 14268:14268 \
  -p 14269:14269 \
  -p 9411:9411 \
  jaegertracing/all-in-one:latest

2. 依存パッケージをインストール

pip install 'strands-agents[otel]'

3. トレース用のコードを追加

from strands.telemetry import StrandsTelemetry

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:4318"
strands_telemetry = StrandsTelemetry()
strands_telemetry.setup_otlp_exporter()