個人でアプリ開発をしていると、技術調査に使う時間が思いのほか多くなります。
新しいAPIのリリースノートを確認して、競合アプリの動向を調べて、関連するライブラリのアップデートを追って——それぞれは数分の作業でも、積み重なると週に3〜4時間が「情報収集」だけで消えていきます。
「Claude APIを使えばこの作業を自動化できるはず」と思って実装を始めたのが1年ほど前のことです。最初は単純なスクリプトでしたが、本番で使い続けるうちに設計が洗練されていき、今では毎朝7時に自動でリサーチレポートが届く仕組みが稼働しています。
ここではその設計と実装を包み隠さず解説します。公式ドキュメントには書かれていない「なぜその選択をしたか」の判断根拠を中心に、実際に動くコードと一緒に説明していきます。
なぜ自律エージェントが必要か——単純な API 呼び出しとの違い
最初に試みたのは、単純な Claude API 呼び出しでした。「この技術について調べて」というプロンプトを渡すだけの実装です。
結果は期待外れでした。理由は明快で、Claude の学習データには鮮度の限界があります。リリースされたばかりの API 変更や、今週の競合動向を知る方法がないのです。
Web 検索ツールを追加した瞬間に状況が一変しました。Claude が自分で必要な情報を検索し、複数のソースを比較検討して、矛盾する情報を自律的に解決するようになったのです。
「エージェント」という言葉は曖昧に使われがちですが、この文脈での定義は明確です。事前に決められた手順ではなく、Claude 自身が「次に何をすべきか」を判断しながら複数のステップを実行するシステム、これがエージェントです。
単純な API 呼び出しとの本質的な違いは次の点にあります。
- 単純な呼び出し: プロンプト → レスポンス（1往復）
- エージェント: タスク → 計画 → 検索 → 評価 → 追加検索 → 統合 → レポート（多段階）
多段階になると、コストとリスクも比例して上がります。設計の肝は「自律性と制御のバランス」をどこで取るか、という判断になります。
システム全体のアーキテクチャ設計
実装に入る前に、全体像を整理しておきます。
リサーチタスク定義
↓
[Orchestrator: Claude Sonnet 4.6]
↓ tool_use
[Web 検索ツール群]
- search_web() ← Brave Search API
- fetch_page() ← 本文抽出
- summarize_source() ← 個別要約
↓
[品質チェックレイヤー]
- 情報の新鮮度確認
- ソースの信頼性スコア
- 重複排除
↓
[統合・レポート生成]
↓
[知識ベース格納（SQLite）]
↓
出力（Markdown / JSON）
重要な設計判断を3つ説明します。
① オーケストレーターに Sonnet 4.6 を使う
最初は Opus 4.6 を使っていましたが、リサーチタスクのほとんどは推論の深さより検索と統合の速さが重要です。コストを実測すると、Sonnet 4.6 で Opus の約1/5の費用で同等の品質が出ました。高度な判断が必要なタスク（矛盾する情報の解決など）のみ、Claude に明示的に考える時間を与える設計にしています。
② 検索と要約を分離する
「検索して要約して」を一度の Claude 呼び出しでやりたくなりますが、分離した方が品質が上がります。検索結果の生テキストをそのまま Claude のコンテキストに入れると、関係ない広告テキストやナビゲーションメニューがノイズになるからです。fetch_page() で本文だけ抽出してから渡すと、トークンを節約しつつ品質が向上します。
③ 予算の上限を二重に設ける
ツール呼び出しのループが想定外に回り続けることがあります。セッション単位の上限（最大トークン数）だけでなく、ツール呼び出し回数にも上限を設けることで、暴走を確実に止めています。
Claude API と Web 検索ツールの実装
具体的な実装に入ります。まずツール定義から始めます。
import anthropic
import httpx
import json
from datetime import datetime
from bs4 import BeautifulSoup
client = anthropic.Anthropic()
# ツール定義
RESEARCH_TOOLS = [
{
"name": "search_web",
"description": (
"指定されたキーワードで Web 検索を実行する。"
"最新の情報を調べる際に使う。"
"検索結果には URL・タイトル・スニペットが含まれる。"
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索クエリ（英語が最もよい結果を得やすい）"
},
"max_results": {
"type": "integer",
"description": "取得する検索結果の最大件数（デフォルト5）",
"default": 5
}
},
"required": ["query"]
}
},
{
"name": "fetch_page",
"description": (
"指定された URL のページ本文を取得する。"
"search_web の結果から詳細が必要な URL に対して使う。"
"HTML タグを除いた本文テキストのみ返す。"
),
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "取得する URL"
},
"max_chars": {
"type": "integer",
"description": "返す文字数の上限（デフォルト 3000）",
"default": 3000
}
},
"required": ["url"]
}
}
]次に、ツール実行関数の実装です。ここが最も重要で、エラーハンドリングを丁寧に書かないと Claude が誤った情報を前提に推論を続けてしまいます。
BRAVE_API_KEY = "YOUR_BRAVE_API_KEY" # 環境変数から取得
def execute_tool(tool_name: str, tool_input: dict) -> str:
"""ツールを実行し、結果を文字列で返す。エラーも文字列として返す（Claude に伝える）"""
try:
if tool_name == "search_web":
return _search_web(
query=tool_input["query"],
max_results=tool_input.get("max_results", 5)
)
elif tool_name == "fetch_page":
return _fetch_page(
url=tool_input["url"],
max_chars=tool_input.get("max_chars", 3000)
)
else:
return f"ERROR: 未知のツール: {tool_name}"
except Exception as e:
# エラーを文字列で返すことで Claude がエラーを認識して対処できる
return f"ERROR: {tool_name} の実行に失敗しました: {str(e)}"
def _search_web(query: str, max_results: int = 5) -> str:
"""Brave Search API を使った Web 検索"""
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
"X-Subscription-Token": BRAVE_API_KEY,
}
params = {
"q": query,
"count": min(max_results, 10),
"search_lang": "ja",
"country": "JP",
"freshness": "pw", # 過去1週間を優先
}
resp = httpx.get(
"https://api.search.brave.com/res/v1/web/search",
headers=headers,
params=params,
timeout=10.0
)
resp.raise_for_status()
data = resp.json()
results = data.get("web", {}).get("results", [])
if not results:
return "検索結果が見つかりませんでした。"
# Claude が読みやすい形式に整形
formatted = []
for i, r in enumerate(results, 1):
formatted.append(
f"[{i}] {r.get('title', '(タイトルなし)')}\n"
f"URL: {r.get('url', '')}\n"
f"概要: {r.get('description', '(概要なし)')}\n"
f"日付: {r.get('page_age', '不明')}"
)
return "\n\n".join(formatted)
def _fetch_page(url: str, max_chars: int = 3000) -> str:
"""URL の本文テキストを抽出する"""
resp = httpx.get(
url,
timeout=15.0,
follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0 (compatible; ResearchBot/1.0)"}
)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
# 不要な要素を除去
for tag in soup(["script", "style", "nav", "header", "footer",
"aside", "form", "iframe", "advertisement"]):
tag.decompose()
# 本文テキスト抽出
text = soup.get_text(separator="\n", strip=True)
# 連続する空行を削除してコンパクトに
lines = [line for line in text.splitlines() if line.strip()]
text = "\n".join(lines)
return text[:max_chars] + ("... (省略)" if len(text) > max_chars else "")エージェントループの実装——制御と自律性のバランス
エージェントループはシステムの核心部分です。Claude が「もう情報は十分」と判断するまでツールを呼び出し続けます。
def run_research_agent(
topic: str,
context: str = "",
max_tool_calls: int = 20, # ツール呼び出しの上限
max_tokens: int = 8192, # 1セッションのトークン上限
) -> dict:
"""
リサーチエージェントを実行する。
Returns:
{
"report": str, # 生成されたレポート
"sources": list[str], # 参照した URL リスト
"tool_call_count": int, # 実際のツール呼び出し回数
"input_tokens": int, # 使用した入力トークン数
"output_tokens": int, # 使用した出力トークン数
}
"""
system_prompt = f"""あなたは優秀なリサーチアシスタントです。
与えられたトピックについて、Web 検索ツールを使って最新情報を収集し、
信頼性の高い情報源に基づいた詳細なレポートを日本語で作成してください。
調査の原則:
1. 最低3つの異なるソースを確認する
2. 日付が新しい情報を優先する
3. 矛盾する情報があれば複数のソースで確認する
4. 情報源の信頼性（公式ドキュメント > 技術ブログ > SNS）を意識する
5. 不確かな情報には「〜という報告がある」と明記する
コンテキスト情報（あれば参考にする）:
{context if context else "（なし）"}
調査完了時は、参照した主要な URL リストを最後に含めてください。"""
messages = [
{"role": "user", "content": f"以下のトピックについて詳しく調査してください:\n\n{topic}"}
]
tool_call_count = 0
total_input_tokens = 0
total_output_tokens = 0
sources = []
while tool_call_count < max_tool_calls:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=max_tokens,
system=system_prompt,
tools=RESEARCH_TOOLS,
messages=messages
)
# トークン使用量を累積
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
# 終了条件: ツールが呼び出されていない（レポート完成）
if response.stop_reason == "end_turn":
final_text = ""
for block in response.content:
if hasattr(block, "text"):
final_text += block.text
return {
"report": final_text,
"sources": list(set(sources)), # 重複除去
"tool_call_count": tool_call_count,
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
}
# ツール呼び出しを処理
assistant_message = {"role": "assistant", "content": response.content}
messages.append(assistant_message)
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
tool_call_count += 1
print(f" 🔧 Tool [{tool_call_count}/{max_tool_calls}]: {block.name}({block.input})")
# URL の記録
if block.name == "fetch_page":
url = block.input.get("url", "")
if url:
sources.append(url)
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
# 上限に達した場合: 途中でも結果を返す
# （重要: 完全失敗にしないこと）
print(f"⚠️ ツール呼び出し上限 {max_tool_calls} に達しました。中間結果を返します。")
# 最後のレスポンスからテキストを抽出
final_text = ""
for block in response.content:
if hasattr(block, "text"):
final_text += block.text
return {
"report": final_text + "\n\n⚠️ 注意: 調査がトークン上限により中断されました。",
"sources": list(set(sources)),
"tool_call_count": tool_call_count,
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
}実際に使ったときに気づいたことがあります。ツール呼び出しのエラーを例外として throw せず、文字列として Claude に返すのが正しいパターンです。例外が出ると呼び出し元のコードが止まりますが、エラー文字列を渡せば Claude が「このソースは取得できなかった、別のソースを試そう」と自律的に対処してくれます。これは公式ドキュメントに明示されていない重要な設計原則です。
品質保証レイヤー——信頼できる情報だけをフィルタリングする
エージェントが収集した情報をそのまま信じるのは危険です。誤情報・古い情報・信頼性の低いソースが混入することがあります。
from dataclasses import dataclass
from typing import Optional
import re
from datetime import datetime, timedelta
@dataclass
class SourceQuality:
url: str
trust_score: float # 0.0〜1.0
freshness_score: float # 0.0〜1.0（新しいほど高い）
is_primary_source: bool
def evaluate_source(url: str, content: str, pub_date_str: Optional[str] = None) -> SourceQuality:
"""ソースの品質を評価する"""
trust_score = 0.5 # デフォルト（中程度の信頼性）
# ドメインによる信頼性スコア
high_trust_domains = [
"anthropic.com", "docs.anthropic.com",
"github.com", "arxiv.org",
"developer.apple.com", "developer.android.com",
]
medium_trust_domains = [
"zenn.dev", "qiita.com", "note.com",
"stackoverflow.com", "reddit.com/r/MachineLearning",
]
for domain in high_trust_domains:
if domain in url:
trust_score = 0.95
break
else:
for domain in medium_trust_domains:
if domain in url:
trust_score = 0.75
break
# 公式ドキュメントかどうか
is_primary = any(d in url for d in ["docs.", "developer.", "help."])
# 日付による鮮度スコア
freshness_score = 0.5
if pub_date_str:
try:
# 様々な日付フォーマットに対応
for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%B %d, %Y"]:
try:
pub_date = datetime.strptime(pub_date_str, fmt)
days_old = (datetime.now() - pub_date).days
# 7日以内: 1.0、90日以内: 0.7、1年以内: 0.4
if days_old <= 7:
freshness_score = 1.0
elif days_old <= 90:
freshness_score = 0.7
elif days_old <= 365:
freshness_score = 0.4
else:
freshness_score = 0.2
break
except ValueError:
continue
except Exception:
pass
return SourceQuality(
url=url,
trust_score=trust_score,
freshness_score=freshness_score,
is_primary_source=is_primary
)この品質スコアを使って、最終レポートに情報源の信頼性を付記する仕組みを作りました。読者（自分自身）が「この情報はどこから来たのか」を把握できるようにするためです。
知識ベースへの格納と履歴管理
同じトピックを何度も調べないために、過去のリサーチ結果を SQLite に保存しています。
import sqlite3
from pathlib import Path
import hashlib
DB_PATH = Path("research_knowledge.db")
def init_db():
"""データベースを初期化する"""
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS research_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT NOT NULL,
topic_hash TEXT NOT NULL,
report TEXT NOT NULL,
sources TEXT NOT NULL, -- JSON 配列
tool_call_count INTEGER,
input_tokens INTEGER,
output_tokens INTEGER,
created_at TEXT NOT NULL
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_topic_hash
ON research_results(topic_hash)
""")
conn.commit()
conn.close()
def save_result(topic: str, result: dict):
"""リサーチ結果を保存する"""
topic_hash = hashlib.md5(topic.encode()).hexdigest()
conn = sqlite3.connect(DB_PATH)
conn.execute("""
INSERT INTO research_results
(topic, topic_hash, report, sources, tool_call_count,
input_tokens, output_tokens, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
topic,
topic_hash,
result["report"],
json.dumps(result["sources"], ensure_ascii=False),
result["tool_call_count"],
result["input_tokens"],
result["output_tokens"],
datetime.now().isoformat()
))
conn.commit()
conn.close()
def find_recent_result(topic: str, max_age_hours: int = 24) -> Optional[dict]:
"""
指定時間以内に同じトピックを調査した結果があれば返す。
キャッシュとして機能し、API コストを削減する。
"""
topic_hash = hashlib.md5(topic.encode()).hexdigest()
cutoff = (datetime.now() - timedelta(hours=max_age_hours)).isoformat()
conn = sqlite3.connect(DB_PATH)
row = conn.execute("""
SELECT report, sources, created_at
FROM research_results
WHERE topic_hash = ? AND created_at > ?
ORDER BY created_at DESC
LIMIT 1
""", (topic_hash, cutoff)).fetchone()
conn.close()
if row:
return {
"report": row[0],
"sources": json.loads(row[1]),
"cached_at": row[2]
}
return None同じトピックを24時間以内に調べたことがあれば、API を呼ばずにキャッシュを返します。実際の運用で最もコスト削減に効いたのはこのキャッシュ機能でした。プロンプトキャッシュとは異なる、アプリケーションレベルのキャッシュです。
予算制御と暴走防止——3層の安全機構
自律エージェントを本番で運用する上で最も重要なのは「止め方」の設計です。3層の安全機構を設けています。
class BudgetGuard:
"""
エージェントの予算を守るガード機構。
3層の制限: ①ツール呼び出し回数 ②合計トークン数 ③1セッションのドル換算
"""
# claude-sonnet-4-6 の単価（2026-05現在）
PRICE_PER_1K_INPUT = 0.003 # USD
PRICE_PER_1K_OUTPUT = 0.015 # USD
def __init__(
self,
max_tool_calls: int = 20,
max_tokens: int = 100_000,
max_cost_usd: float = 0.50 # 1リサーチあたり上限50セント
):
self.max_tool_calls = max_tool_calls
self.max_tokens = max_tokens
self.max_cost_usd = max_cost_usd
self.tool_call_count = 0
self.input_tokens = 0
self.output_tokens = 0
def record_usage(self, input_tokens: int, output_tokens: int):
self.input_tokens += input_tokens
self.output_tokens += output_tokens
def record_tool_call(self):
self.tool_call_count += 1
@property
def estimated_cost_usd(self) -> float:
return (
(self.input_tokens / 1000) * self.PRICE_PER_1K_INPUT +
(self.output_tokens / 1000) * self.PRICE_PER_1K_OUTPUT
)
def should_stop(self) -> tuple[bool, str]:
"""停止すべきかどうかと理由を返す"""
if self.tool_call_count >= self.max_tool_calls:
return True, f"ツール呼び出し上限 ({self.max_tool_calls}) に達しました"
if (self.input_tokens + self.output_tokens) >= self.max_tokens:
return True, f"トークン上限 ({self.max_tokens}) に達しました"
if self.estimated_cost_usd >= self.max_cost_usd:
return True, f"コスト上限 (${self.max_cost_usd}) に達しました: ${self.estimated_cost_usd:.4f}"
return False, ""
def status(self) -> str:
return (
f"ツール: {self.tool_call_count}/{self.max_tool_calls} | "
f"トークン: {self.input_tokens + self.output_tokens:,}/{self.max_tokens:,} | "
f"コスト: ${self.estimated_cost_usd:.4f}/${self.max_cost_usd}"
)コスト上限を設けたことで、万が一 Claude がループ状態に入っても自動的に停止します。月末に想定外の請求が来ることがなくなりました。
スケジューラーと全体の組み合わせ
毎朝7時に自動でリサーチレポートを生成するスケジューラーの実装例です。
import asyncio
from typing import NamedTuple
class ResearchTask(NamedTuple):
topic: str
context: str = ""
cache_hours: int = 24 # キャッシュ有効期間
# 定期実行するリサーチタスクの定義
DAILY_RESEARCH_TASKS = [
ResearchTask(
topic="Anthropic Claude API の最新アップデート・新機能・価格変更",
cache_hours=12 # 半日キャッシュ
),
ResearchTask(
topic="壁紙・ライブ壁紙カテゴリの App Store トップアプリ 動向と新規参入",
context="iOS App Store、Google Play Store の壁紙アプリカテゴリ",
cache_hours=24
),
ResearchTask(
topic="SwiftUI・Swift の最新アップデートとベストプラクティス変更",
cache_hours=48 # 2日キャッシュ（頻繁には変わらない）
),
]
def run_daily_research():
"""毎日実行するリサーチを全て実行する"""
init_db()
reports = []
total_cost = 0.0
for task in DAILY_RESEARCH_TASKS:
print(f"\n📚 リサーチ開始: {task.topic[:40]}...")
# キャッシュ確認
cached = find_recent_result(task.topic, max_age_hours=task.cache_hours)
if cached:
print(f" ✅ キャッシュ使用 (保存日時: {cached['cached_at']})")
reports.append({
"topic": task.topic,
"report": cached["report"],
"sources": cached["sources"],
"from_cache": True,
"cost_usd": 0.0
})
continue
# エージェント実行
guard = BudgetGuard(
max_tool_calls=15,
max_tokens=80_000,
max_cost_usd=0.30
)
result = run_research_agent(
topic=task.topic,
context=task.context,
max_tool_calls=guard.max_tool_calls,
)
# 結果を保存
save_result(task.topic, result)
reports.append({
"topic": task.topic,
"report": result["report"],
"sources": result["sources"],
"from_cache": False,
"cost_usd": guard.estimated_cost_usd
})
total_cost += guard.estimated_cost_usd
print(f" 💰 コスト: ${guard.estimated_cost_usd:.4f}")
# 統合レポートを生成
output_path = Path(f"daily_report_{datetime.now().strftime('%Y%m%d')}.md")
with open(output_path, "w", encoding="utf-8") as f:
f.write(f"# デイリーリサーチレポート\n")
f.write(f"生成日時: {datetime.now().strftime('%Y-%m-%d %H:%M JST')}\n")
f.write(f"合計コスト: ${total_cost:.4f}\n\n")
for r in reports:
f.write(f"---\n\n")
f.write(f"## {r['topic']}\n\n")
if r["from_cache"]:
f.write("*（キャッシュから取得）*\n\n")
f.write(r["report"] + "\n\n")
print(f"\n✅ レポート生成完了: {output_path}")
print(f"💰 合計コスト: ${total_cost:.4f}")
return reports
if __name__ == "__main__":
run_daily_research()本番運用で分かったこと——3ヶ月間の実績と教訓
実際に3ヶ月間運用して得られたデータと教訓を共有します。
コストの実績（月次）
最初の月は設定が甘く、1リサーチあたり平均 $0.18 かかっていました。キャッシュ機能と max_tool_calls の適切な調整を加えた後、現在は $0.06 に削減されています。月間合計で約 $8〜$12 程度です。
品質上の発見
Web 検索の freshness パラメータを "pw"（過去1週間）に設定したところ、古い情報を引いてくる頻度が大きく減りました。これは公式ドキュメントに記載されているものの、試してみるまで効果が分かりませんでした。
一方、専門性の高いトピック（AppleのフレームワークAPIの詳細など）は、公式ドキュメントのみを参照させた方が精度が上がります。search_web の query に "site:developer.apple.com" を含めるよう Claude に指示するシステムプロンプトの工夫が有効でした。
予期しない問題と解決策
ページ取得（fetch_page）でクラッシュが多発した時期がありました。原因は特定のサイトが返す JavaScript-rendered コンテンツで、BeautifulSoup が空のテキストを返し、Claude が「コンテンツが取得できない」→「別のURLを試す」→「また失敗」というループに入ることがありました。
解決策として、fetch_page の結果が100文字未満の場合は「このページは JavaScript でレンダリングされており、静的取得では内容を確認できません」という固定エラーメッセージを返す処理を追加しました。Claude はこのエラーを見て別のソースに切り替えてくれます。
最も大切な設計原則
3ヶ月の運用を通じて感じた最も重要なことは、「Claude に失敗を伝えること」です。エラーを黙って飲み込む実装より、エラーを明示的に文字列で返す実装の方が、Claude がより良い判断を下します。これは Claude Agent SDK や Claude API ツール呼び出しの応用 でも共通するパターンです。
全体を振り返って——まず動かしてみることが最良の学習
この記事で紹介したシステムは、完成形ではありません。今も少しずつ改善を続けています。
実装を始める順番としては、まず execute_tool と run_research_agent の最小版を作ってターミナルで動かしてみることをお勧めします。品質チェックやキャッシュはその後で追加できます。動く形を持ってから改善する方が、設計の勘所が自然と身につきます。
より詳細なツール呼び出しの並列化パターンについては 並列ツール呼び出しの本番実装 も参考になります。また、コスト最適化の観点では プロンプトキャッシュの実践ガイド と組み合わせることで、さらに月次コストを抑えられます。
情報収集の自動化を検討している方の参考になれば嬉しいです。実装して動かしてみると、設計をさらに改善したいポイントが必ず見つかります。それが、このシステムを育てていく面白さでもあります。