はじめに
Googleが社内テスト中の「Remy」は、ユーザーのニーズを予測しながら、バックグラウンドで継続的にタスクを管理するAIエージェントです。
ユーザーが何も指示しなくても、必要な情報を収集して事前にアクションを起こす——そんな「24時間稼働型の自律AI」という概念は、今後の開発現場で重要な設計パターンになりつつあります。
本記事では、Remyのような「バックグラウンドAIエージェント」を自分のプロジェクトに組み込む際のアーキテクチャと実装パターンを、コード例を交えて紹介します。
チャットボットの枠を超えた、次世代のAI活用を模索しているエンジニアの参考になれば幸いです。
こんな人におすすめ
- LLM（大規模言語モデル）のAPIを使って自社サービスにAI機能を追加したいエンジニア
- メールチェック・通知・データ収集などの定型業務をAIで自動化したい方
- Anthropic / OpenAI などのAPIでツール使用（Function Calling）を初めて試す方
- 「チャットボット以上のAI活用」を模索しているバックエンドエンジニア
- 自律型AIエージェントの設計パターンを体系的に理解したい方
バックグラウンドAIエージェントのアーキテクチャ
従来のAIアシスタントは「ユーザーが入力する → AIが応答する」というリアクティブ（反応型）なモデルでした。
バックグラウンドAIエージェントはこれを大きく超えています。
flowchart TD
A[スケジューラー] -->|定期実行| B[エージェントループ]
B --> C{LLM判断}
C -->|ツール呼び出し| D[ツール実行]
D -->|結果返却| C
C -->|タスク完了| E[状態保存]
E --> F[通知・ログ記録]
F --> A
G[(状態ストア\nRedis/DB)] <-.->|読み書き| B
H[外部API\nメール/カレンダー等] <-.->|実呼び出し| D
このループ構造の中で、LLMが自律的に判断しながらツールを組み合わせてタスクをこなします。
主な特徴は次の通りです。
- プロアクティブ（先回り型）: ユーザーの指示を待たず、パターンを学習して自律的に行動します
- 継続的: スケジューラーが定期的にエージェントを起動し、24時間稼働を実現します
- 状態保持: 過去の実行履歴や好みを記憶し、次の行動に活かします
- ツール駆動: 外部APIやデータベースを呼び出して実際のアクションを起こします
エージェントループの実装
エージェントの核心は「ループ処理」です。
LLMにツールリストを渡し、LLMが必要なツールを呼び出しながら自律的にタスクをこなす仕組みです。
import anthropic
import json
client = anthropic.Anthropic()
# エージェントが使えるツールを定義する
tools = [
{
"name": "get_unread_emails",
"description": "未読メールを取得し、件名と送信者を返す",
"input_schema": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "取得する最大件数"
}
},
"required": []
}
},
{
"name": "create_task",
"description": "タスク管理システムに新規タスクを作成する",
"input_schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"due_date": {"type": "string", "description": "YYYY-MM-DD形式"},
"priority": {
"type": "string",
"enum": ["high", "medium", "low"]
}
},
"required": ["title", "due_date", "priority"]
}
}
]
def run_agent(context: str, max_steps: int = 10) -> str:
"""エージェントループを実行する"""
messages = [{"role": "user", "content": context}]
for _ in range(max_steps):
response = client.messages.create(
model="claude-opus-4-7",
max_tokens=4096,
tools=tools,
messages=messages
)
# ツール呼び出しがなければ完了
if response.stop_reason == "end_turn":
return response.content[0].text
# ツールを実行して結果を返す
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = _execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result, ensure_ascii=False)
})
messages.append({"role": "user", "content": tool_results})
return "最大ステップ数に達しました"
def _execute_tool(name: str, inputs: dict) -> dict:
"""ツールを実際に呼び出す（実装はサービスに合わせて差し替える）"""
if name == "get_unread_emails":
# Gmail / Exchange API を呼ぶ箇所
return {"emails": [{"subject": "明日の会議について", "from": "boss@example.com"}]}
elif name == "create_task":
# Linear / Jira API を呼ぶ箇所
return {"status": "success", "task_id": "TASK-042"}
return {"error": f"未定義のツール: {name}"}
stop_reason == "tool_use" を検出してツールを実行し、その結果を再度LLMに渡す——このループがAI自律動作の本質です。
観察された傾向として、5〜15ステップで多くの日常タスクは完了します。
24時間稼働のスケジューリング設計
バックグラウンドで動き続けるには、定期実行の仕組みが必要です。
PythonではAPSchedulerが手軽で、既存のWebサーバーと同一プロセスに組み込めます。
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime
scheduler = AsyncIOScheduler()
async def morning_briefing():
"""毎朝9時に実行：当日のタスクを自動整理する"""
today = datetime.now().strftime("%Y年%m月%d日")
prompt = f"""
今日は{today}です。以下を順番に確認して、
優先度の高いタスクを最大3つ作成してください：
1. 未読メールをチェックし、返信が必要なものを特定する
2. 今日が期限のタスクを確認する
3. 昨日完了できなかったタスクがあれば繰り越しタスクを作成する
各タスクには必ず優先度（high/medium/low）と期限を設定してください。
"""
result = run_agent(prompt)
print(f"[朝のブリーフィング完了] {result[:100]}...")
async def hourly_check():
"""1時間ごとに実行：緊急メールを監視する"""
prompt = (
"未読メールを確認し、「緊急」または「至急」が含まれるものがあれば"
"highの優先度でタスクを作成してください。なければ何もしなくて構いません。"
)
run_agent(prompt, max_steps=5)
# スケジューラーにジョブを登録する
scheduler.add_job(morning_briefing, "cron", hour=9, minute=0)
scheduler.add_job(hourly_check, "interval", hours=1)
scheduler.start()
本番環境ではCelery + Redisを使ったワーカー構成にすることで、スケール対応や障害復旧が容易になります。
状態管理と文脈の永続化
Remyが「ニーズを予測する」動作を実現するには、過去の行動パターンや好みを記憶させる必要があります。
会話履歴と設定をRedisに保存することで、エージェントが再起動しても文脈を引き継げます。
// TypeScript + Redis による状態の永続化
import Redis from "ioredis";
const redis = new Redis(process.env.REDIS_URL!);
interface AgentMemory {
userId: string;
history: Array<{ role: string; content: string; ts: number }>;
preferences: Record<string, string>;
}
async function loadMemory(userId: string): Promise<AgentMemory> {
const raw = await redis.get(`agent:memory:${userId}`);
if (raw) return JSON.parse(raw);
return { userId, history: [], preferences: {} };
}
async function saveMemory(memory: AgentMemory): Promise<void> {
// 直近48時間分の履歴のみ保持（トークンコスト削減）
const cutoff = Date.now() - 48 * 60 * 60 * 1000;
memory.history = memory.history.filter((h) => h.ts > cutoff);
await redis.setex(
`agent:memory:${memory.userId}`,
30 * 24 * 60 * 60, // 30日間保持
JSON.stringify(memory)
);
}
履歴のトリミングはコスト管理の観点でも重要です。
直近24〜48時間分のみを保持する設計が、目安として現実的なバランスです。
つまづきやすいポイント
-
ループの無限化:
max_stepsを必ず設定してください。
ツール呼び出しがエラーを繰り返すと際限なくAPIを呼び続けます。
観察された傾向として、5〜15ステップで多くのタスクは完了するため、上限は15〜20が目安として適切です -
副作用ツールの誤動作: メール送信・タスク作成・Slack通知など外部に作用するツールは、必ずテスト用の
dry_runフラグを用意してください。
開発中に本番データを誤って更新するリスクを避けるためです -
トークンコストの積み上がり: バックグラウンドで高頻度実行するとトークン消費が想定外に膨らみます。
会話履歴を定期的に要約・圧縮するロジックは早めに実装しておくことをおすすめします -
エラーの無通知化: バックグラウンド処理は失敗に気づきにくいです。
SentryやSlack Webhookで異常を即座に検知できる監視を、最初から組み込んでおくことが重要です
まとめ
バックグラウンドAIエージェントの実装は、「エージェントループ（LLMとツール呼び出しの反復）」「スケジューラー」「状態永続化」の3つを組み合わせることで実現できます。
GoogleのRemyが目指す「ユーザーのニーズを予測する自律AI」は、こうした設計パターンの延長線上にあります。
最初は副作用の小さいツール（メールチェックや通知作成など）だけに絞ってスコープを小さく始め、安全性を確認しながら段階的に機能を拡張していくアプローチが現実的です。
LLMの活用が「チャットボット」から「自律型エージェント」へと進化しつつある今、こうしたアーキテクチャパターンを押さえておくことは、今後の開発現場でますます価値を持つことになるでしょう。
💡 CTA 指示: ここに D1「開発壁打ち（¥12,000）」 の CTA を WordPress 側で設定してください。
