マルチモーダルエージェントの時代
Claude API の Vision 機能と Tool Use（関数呼び出し）の組み合わせは、単純な「画像認識」や「API 連携」ではなく、リアルタイム推論エージェントを実現します。
例えば、以下のようなシナリオを想像してください：
- ドキュメント自動処理: スキャン PDF → OCR で抽出 → フォーム記入の自動化
- リアルタイムデータ分析: チャート画像 → 異常検知 → Slack 通知
- 複数感覚の統合: カメラ映像（Vision） + ユーザー音声（音声 API） + データベース（Tool Use）
2026年3月時点で、Claude API はこれらを効率的に処理できるようになりました。プロダクション環境で運用可能なアーキテクチャを順を追って整理していきます。
ℹ️リアルタイムマルチモーダルエージェントには、Vision能力、Tool Use、ストリーミング、そして**エラー処理とコスト最適化**が不可欠です。ここでは4つの実装パターンをカバーします。
アーキテクチャの基本構造
┌──────────────────┐
│ 入力ソース │
│ • 画像/Video │
│ • テキスト │
│ • 音声 │
└────────┬─────────┘
│
▼
┌──────────────────────────────────────┐
│ Claude API マルチモーダル推論エンジン │
│ • Vision処理 │
│ • Tool Use意思決定 │
│ • ストリーミング出力 │
└────────┬─────────────────────────────┘
│
▼
┌──────────────────┐
│ 出力アクション │
│ • Slack通知 │
│ • DB更新 │
│ • ファイル生成 │
└──────────────────┘
パターン 1: ドキュメント自動処理エージェント
シナリオ
請求書の写真・スキャンPDF が次々と届きます。以下を自動化：
- 画像から OCR で文字抽出（Claude Vision）
- 請求書番号、金額、日付を抽出（推論）
- Google Sheets に記録（Tool Use）
- 異常金額を検出（ロジック）
- Slack で承認者に通知（アクション）
実装例：Python + streaming
#!/usr/bin/env python3
# document_processing_agent.py
import anthropic
import base64
import json
import os
from pathlib import Path
from datetime import datetime
from typing import Optional
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
# Tool定義：外部システムとの連携
TOOLS = [
{
"name": "record_invoice",
"description": "Google Sheets に請求書情報を記録する",
"input_schema": {
"type": "object",
"properties": {
"invoice_number": {
"type": "string",
"description": "請求書番号（例: INV-2026-001）"
},
"amount": {
"type": "number",
"description": "金額（USD）"
},
"date": {
"type": "string",
"description": "請求日（YYYY-MM-DD形式）"
},
"vendor": {
"type": "string",
"description": "請求元企業名"
}
},
"required": ["invoice_number", "amount", "date", "vendor"]
}
},
{
"name": "notify_slack",
"description": "Slack でメッセージを送信（異常検知時など）",
"input_schema": {
"type": "object",
"properties": {
"channel": {
"type": "string",
"description": "#accounting または #alerts"
},
"message": {
"type": "string",
"description": "送信メッセージ"
},
"severity": {
"type": "string",
"enum": ["info", "warning", "critical"],
"description": "重要度"
}
},
"required": ["channel", "message", "severity"]
}
},
{
"name": "flag_for_review",
"description": "異常な請求書を人間レビュー用フラグ",
"input_schema": {
"type": "object",
"properties": {
"invoice_number": {"type": "string"},
"reason": {
"type": "string",
"enum": [
"duplicate_detected",
"unusual_amount",
"missing_vendor_info",
"date_mismatch"
]
}
},
"required": ["invoice_number", "reason"]
}
}
]
def encode_image_to_base64(image_path: str) -> str:
"""画像ファイルを Base64 エンコード"""
with open(image_path, "rb") as image_file:
return base64.standard_b64encode(image_file.read()).decode("utf-8")
def process_document_stream(image_path: str) -> dict:
"""
ドキュメント処理：画像 → 推論 → アクション
ストリーミングで段階的に結果を出力
"""
# 画像を Base64 エンコード
image_data = encode_image_to_base64(image_path)
file_extension = Path(image_path).suffix.lower()
# MIME タイプ判定
mime_type_map = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp"
}
mime_type = mime_type_map.get(file_extension, "image/jpeg")
print(f"\n📄 処理開始: {Path(image_path).name}")
print("=" * 60)
# ストリーミングで Claude を呼び出し
accumulated_response = ""
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=1024,
tools=TOOLS,
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": mime_type,
"data": image_data
}
},
{
"type": "text",
"text": """この請求書から以下の情報を抽出して、適切なツールを使用して処理してください：
1. 請求書番号（Invoice Number）
2. 金額（Amount in USD）
3. 請求日（Invoice Date）
4. 請求元企業名（Vendor Name）
抽出後：
- record_invoice ツールで Google Sheets に記録
- 金額が通常の3倍以上なら notify_slack（警告）
- 請求番号がフォーマット異常なら flag_for_review"""
}
]
}
]
) as stream:
# ストリーミング応答を処理
tool_calls = []
for event in stream:
# テキスト出力を収集
if hasattr(event, "delta"):
if hasattr(event.delta, "text"):
text = event.delta.text
accumulated_response += text
print(text, end="", flush=True)
# ツール呼び出しを記録
if hasattr(event, "content"):
for block in event.content:
if hasattr(block, "type") and block.type == "tool_use":
tool_calls.append(block)
print("\n")
# ツール呼び出しの実行
results = execute_tool_calls(tool_calls)
return {
"image": Path(image_path).name,
"status": "success",
"response": accumulated_response,
"tool_results": results
}
def execute_tool_calls(tool_calls: list) -> list:
"""Tool Use の結果を実行"""
results = []
for tool_call in tool_calls:
tool_name = tool_call.name
tool_input = tool_call.input
print(f"🔧 ツール実行: {tool_name}")
print(f" 入力: {json.dumps(tool_input, indent=2, ensure_ascii=False)}")
# ツール実行のシミュレーション
if tool_name == "record_invoice":
result = _simulate_record_invoice(tool_input)
elif tool_name == "notify_slack":
result = _simulate_notify_slack(tool_input)
elif tool_name == "flag_for_review":
result = _simulate_flag_for_review(tool_input)
else:
result = f"Unknown tool: {tool_name}"
results.append({
"tool": tool_name,
"result": result
})
print(f" ✅ 結果: {result}\n")
return results
def _simulate_record_invoice(data: dict) -> str:
"""Google Sheets に記録（シミュレーション）"""
# 実際の実装では gspread ライブラリを使用
return f"Recorded invoice {data['invoice_number']} for ${data['amount']}"
def _simulate_notify_slack(data: dict) -> str:
"""Slack 通知（シミュレーション）"""
return f"Notified {data['channel']}: {data['message']}"
def _simulate_flag_for_review(data: dict) -> str:
"""レビューフラグ（シミュレーション）"""
return f"Flagged {data['invoice_number']} - Reason: {data['reason']}"
# メイン処理
if __name__ == "__main__":
# テスト用サンプル画像パス
test_image = "sample_invoice.jpg"
if not Path(test_image).exists():
print(f"⚠️ テスト画像 '{test_image}' が見つかりません")
print("本番環境では実際の請求書画像を指定してください")
else:
result = process_document_stream(test_image)
print(json.dumps(result, indent=2, ensure_ascii=False))
パターン 2: リアルタイムチャート異常検知エージェント
シナリオ
ダッシュボード画像（チャート）を定期的に取得。異常（急上昇・急低下）を検知して、自動で Slack アラートを発行。
実装例：Vision + 統計分析
#!/usr/bin/env python3
# chart_monitoring_agent.py
import anthropic
import base64
import json
from pathlib import Path
from datetime import datetime
import subprocess
client = anthropic.Anthropic()
async def analyze_chart_for_anomalies(image_path: str, baseline: dict):
"""
チャート画像を分析し、異常を検知
Args:
image_path: チャート画像のパス
baseline: 通常値の基準（過去の統計）
"""
# Base64 エンコード
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
# Claude Vision で分析
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=512,
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_data
}
},
{
"type": "text",
"text": f"""このチャートを分析してください。JSON形式で結果を返してください：
{{
"metric_name": "チャートが表すメトリクス名",
"current_value": 現在値（数値）,
"trend": "up" | "down" | "stable",
"change_percentage": 前期比変化率（%）,
"anomaly_detected": true | false,
"anomaly_reason": "理由（異常検知された場合）",
"recommendation": "推奨アクション"
}}
基準値（baseline）の参考：
{json.dumps(baseline, indent=2)}
異常と判定される条件：
- 前期比変化率が ±20% を超える
- トレンドが反転している
- 値が0または負になっている（通常は正のメトリクス）"""
}
]
}
]
)
# レスポンスを JSON パース
result_text = response.content[0].text
# JSON ブロックを抽出（マークダウンの ```json で囲まれている場合も対応）
import re
json_match = re.search(r'```json\n(.*?)\n```', result_text, re.DOTALL)
if json_match:
json_str = json_match.group(1)
else:
json_str = result_text
result = json.loads(json_str)
# 異常検知時のアクション
if result["anomaly_detected"]:
severity = "critical" if abs(result["change_percentage"]) > 50 else "warning"
notify_anomaly(result, severity)
return result
def notify_anomaly(analysis: dict, severity: str):
"""異常を Slack で通知"""
color_map = {"critical": "danger", "warning": "warning", "info": "good"}
emoji_map = {"critical": "🔴", "warning": "🟡", "info": "🟢"}
message = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji_map[severity]} メトリクス異常検知"
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*メトリクス*\n{analysis['metric_name']}"
},
{
"type": "mrkdwn",
"text": f"*変化率*\n{analysis['change_percentage']:+.1f}%"
},
{
"type": "mrkdwn",
"text": f"*トレンド*\n{analysis['trend']}"
},
{
"type": "mrkdwn",
"text": f"*現在値*\n{analysis['current_value']}"
}
]
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*原因*\n{analysis['anomaly_reason']}\n\n*推奨*\n{analysis['recommendation']}"
}
}
]
}
webhook = os.environ.get("SLACK_WEBHOOK_URL")
if webhook:
subprocess.run([
"curl", "-X", "POST", webhook,
"-H", "Content-Type: application/json",
"-d", json.dumps(message)
])
# 使用例
if __name__ == "__main__":
baseline = {
"avg_value": 5000,
"std_dev": 500,
"normal_range": [4500, 5500]
}
# テスト実行（非同期）
import asyncio
result = asyncio.run(analyze_chart_for_anomalies(
"sales_chart.png",
baseline
))
print(json.dumps(result, indent=2, ensure_ascii=False))
パターン 3: コスト最適化戦略
Vision + Tool Use を組み合わせると、API コストが増加します。ここで実用的な最適化パターンをご紹介します。
戦略 1: 画像の事前処理
#!/usr/bin/env python3
# cost_optimization.py
from PIL import Image
import os
def optimize_image_for_vision(image_path: str, max_width: int = 1024) -> str:
"""
画像を Vision API 用に最適化
- リサイズ（高解像度を削減）
- 品質圧縮（JPEG品質80%）
- グレースケール変換（色情報不要な場合）
"""
img = Image.open(image_path)
# リサイズ（アスペクト比を保持）
img.thumbnail((max_width, max_width * img.height // img.width))
# JPEG で保存（圧縮）
output_path = image_path.replace(".", "_optimized.")
img.save(output_path, "JPEG", quality=80, optimize=True)
# ファイルサイズ削減を表示
original_size = os.path.getsize(image_path)
optimized_size = os.path.getsize(output_path)
reduction = (1 - optimized_size / original_size) * 100
print(f"✅ 画像最適化完了")
print(f" 元: {original_size / 1024:.1f} KB")
print(f" 最適化後: {optimized_size / 1024:.1f} KB ({reduction:.1f}% 削減)")
return output_path
戦略 2: バッチ処理で API 呼び出し削減
def batch_analyze_documents(image_paths: list) -> list:
"""
複数ドキュメントをバッチ処理
1回の API 呼び出しで複数画像を処理（コスト削減）
"""
# ただし Claude API は1回のメッセージに複数の画像を支持できるが、
# 処理量が増えると精度が低下する可能性があります。
# バランスの取れた方法は「3-5個の画像を1回で処理」
if len(image_paths) > 5:
# 5つのグループに分割
batches = [
image_paths[i:i+5]
for i in range(0, len(image_paths), 5)
]
else:
batches = [image_paths]
results = []
for batch in batches:
result = _process_batch(batch)
results.extend(result)
return results
戦略 3: キャッシング活用
import hashlib
import json
CACHE_DIR = Path("/tmp/vision_cache")
def get_vision_analysis_cached(image_path: str) -> dict:
"""
Vision 分析結果をキャッシュ（同じ画像の再分析を避ける）
"""
# 画像の MD5 ハッシュでキャッシュキーを生成
with open(image_path, "rb") as f:
image_hash = hashlib.md5(f.read()).hexdigest()
cache_file = CACHE_DIR / f"{image_hash}.json"
# キャッシュが存在かつ24時間以内なら返す
if cache_file.exists():
modified_time = cache_file.stat().st_mtime
current_time = datetime.now().timestamp()
if current_time - modified_time < 86400: # 24 hours
with open(cache_file) as f:
return json.load(f)
# キャッシュがなければ API 呼び出し
result = process_document_stream(image_path)
# キャッシュに保存
with open(cache_file, "w") as f:
json.dump(result, f)
return result
パターン 4：エラーハンドリングと再試行戦略
本番環境では、API の一時的なエラー、タイムアウト、レート制限に対応が不可欠です。
import time
from typing import Callable, TypeVar
T = TypeVar('T')
def retry_with_exponential_backoff(
func: Callable[..., T],
max_retries: int = 3,
initial_delay: float = 1.0
) -> T:
"""
指数バックオフで再試行
エラーパターン：
- RateLimitError (429): 指数バックオフで再試行
- APIError (5xx): 最大3回再試行
- ValidationError (400): 再試行しない（ユーザー入力エラー）
"""
delay = initial_delay
for attempt in range(max_retries):
try:
return func()
except anthropic.RateLimitError as e:
if attempt == max_retries - 1:
raise
print(f"⏳ レート制限。{delay:.1f}秒待機中...")
time.sleep(delay)
delay *= 2 # 指数バックオフ
except anthropic.APIError as e:
if attempt == max_retries - 1:
raise
print(f"❌ API エラー（{e.status_code}）。{delay:.1f}秒後に再試行...")
time.sleep(delay)
delay *= 2
except anthropic.APIConnectionError as e:
if attempt == max_retries - 1:
raise
print(f"🌐 接続エラー。{delay:.1f}秒後に再試行...")
time.sleep(delay)
raise RuntimeError("Max retries exceeded")
# 使用例
def call_vision_api():
"""Vision API を呼び出し（エラー処理あり）"""
def _call():
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[...]
)
return response
return retry_with_exponential_backoff(_call)
エージェントアーキテクチャのベストプラクティス
1. 明確な権限範囲（Scope）
# ❌ 範囲が広すぎる（危険）
TOOLS = [
"execute_shell_command",
"delete_any_file",
"send_email_to_anyone"
]
# ✅ 範囲を制限（安全）
TOOLS = [
"record_invoice_to_approved_sheet",
"delete_duplicate_invoice_only",
"send_slack_to_accounting_channel"
]
2. Tool Output の検証
def validate_tool_result(tool_name: str, result: dict) -> bool:
"""Tool の実行結果を検証"""
if tool_name == "record_invoice":
# 必須フィールドをチェック
required = ["invoice_id", "amount", "recorded_at"]
return all(k in result for k in required)
elif tool_name == "notify_slack":
# メッセージが実際に送信されたか確認
return result.get("status") == "sent"
return True
3. 監査ログ
すべての Vision 分析と Tool 実行をログ記録：
{
"timestamp": "2026-03-25T14:32:15Z",
"agent": "document_processing_agent",
"input": {
"image": "invoice_20260325.jpg",
"size_bytes": 245000
},
"vision_analysis": {
"extracted_invoice_number": "INV-2026-12345",
"confidence": 0.95
},
"tools_called": [
{"name": "record_invoice", "status": "success"},
{"name": "notify_slack", "status": "success"}
],
"cost": {
"vision_tokens": 512,
"input_tokens": 245,
"output_tokens": 128,
"total_usd": 0.0089
}
}
結論
Claude API のマルチモーダル能力は、Vision + Tool Use + Streaming の組み合わせで初めて本領を発揮します。
本ガイドで紹介したパターンを参考に、貴社の業務に応じた実装をお勧めします。さらに詳しい最適化技法は Claude API Token Counting 完全ガイド をご参照ください。
また、マルチエージェントシステムへの発展は Claude Agent SDK マルチエージェントシステム構築ガイド で詳しく解説しています。