著者ノート:
本稿は、Go 言語(Golang 1.26)、Google Gemini Live API ネイティブマルチモーダルプロトコル、Genkit、PostgreSQL (pgvector)、そして Redis を採用し、ゼロから専用のプライベート執事システム Iris を構築した際の、純粋な個人視点による技術振り返り記録である。大袈裟な表現は排し、コードはすべて汎用化・マスキングした上で、コア設計、通信パイプライン、そしてリアルな試行錯誤の過程を率直にまとめる。
0. なぜ「ありふれた Chatbot」では満足できなかったのか?
現在目にすることの多いオープンソースの AI アシスタントは、その多くが古典的な「Web チャットボックス」のパラダイムに基づいて構築されている。
HTTP や SSE(Server-Sent Events)経由でテキスト入力を受け取り、タイプライター風にテキストをストリーミング返却する。リマインダー機能が付いている場合でも、その多くはデータベースを毎分ポーリングする定期タスクに依存している。
こうした古典的な対話モデルは、長文の作成や調査には非常に実用的である。しかし、常にバックグラウンドに常驻させ、日々のデスクトップワークフローに溶け込む「専属執事」として使おうとすると、体験の面でいくつかの明確なボトルネックに直面する:
-
「トランシーバー式」の半二重通信(Half-Duplex):
人間の自然な対話は**全二重通信(Full-Duplex)**である。相手の話を聞きながら相槌を打つこともできれば、相手が話しすぎている時に途中で口を挟んで遮る(Barge-In)こともできる。それに対し従来の Chatbot は典型的な「トランシーバー」だ。ボタンを押して話し、数秒待ち、返ってきた長文を眺める。AI が話している間、即座に話を遮ることも難しい。 -
視覚的な現場感の欠如:
コードのエラー画面を見せたい時や、デスク上の実物・図面について尋ねたい時、従来は「スクショを撮る/写真を撮る → 保存する → ファイルをアップロードする → 文章で状況を説明する」という手順を踏む必要があった。この一連の動作は思考のフローを分断してしまう。 -
金魚のように短い記憶:
新しいセッションを開くたびに、「自分はバックエンドエンジニアで、Go をメインに使っていて、コーヒーには砂糖を入れない」といった前提を説明し直さなければならない。コンテキストがリセットされると、以前築いた暗黙の了解を継続することが難しい。 -
受動的な応答では「執事」になり得ない:
本物の執事には現実の時間の感覚が必要であり、指定の時間になれば自発的に声をかけに来るべきである。しかし、通常の対話アシスタントは常に「呼び出し待ちの受動的トリガー」であり、こちらからメッセージを送らない限り何のアクションも起こさない。
こうした対話のディテールにおいて、より人間同士の自然なコミュニケーションに近いアプローチを模索するために、私は Iris の開発をスタートした。
1. 音声の核心:全二重通話と「即時割り込み(Local Barge-In)」
1. なぜ「音声認識 → LLM 呼び出し → 音声合成」の 3 段階パイプラインを採用しなかったのか?
音声アシスタントを作る際、多くの人が直感的に以下のようなパイプラインを組み立てる:
$$\text{マイク録音} \xrightarrow{1} \text{Whisper 等による音声認識(ASR)} \xrightarrow{2} \text{LLM テキスト生成} \xrightarrow{3} \text{TTS 音声合成} \xrightarrow{4} \text{スピーカー再生}$$
この構成は、各工程でネットワークと計算の遅延が積み重なる:
- 録音のチャンク化に時間がかかる(最低 500ms)
- ASR の文字起こしに時間がかかる(300ms 〜 800ms)
- LLM が最初の文章を生成するのに時間がかかる(500ms 〜 1000ms)
- TTS が文単位で音声を合成するのに時間がかかる(500ms)
トータルすると、エンドツーエンドの遅延は通常 2 秒から 4 秒 に達する。「どうぞ、どうぞ」とトランシーバーで話しているような感覚になり、自然な対話のリズムは生まれない。
Iris のアプローチ:Gemini Live ネイティブ双方向ストリーミングの直結
ブラウザのマイクから生の音声フレームをキャプチャし、WebSocket を通じて水の流れのように Go バックエンドへ流し込み、バックエンドはそのまま Gemini Live の双方向ストリームチャネルへバイパスする。モデル自体が音声の振幅やトーンを直接理解し、モデル内部からネイティブな音声スライス(人間の自然な息継ぎや抑揚を持つ Aoede ボイス)をリアルタイムに出力する。これにより、対話遅延を人間同士の会話に近い数百ミリ秒台へと圧縮した。
【クライアント】 【Go バックエンド】 【Gemini Live API】
マイク入力 (16kHz PCM) ──(WebSocket)──> Gateway Hub ──(Bidi Streaming)──> 音声ネイティブ認識
スピーカー (24kHz PCM) <──(WebSocket)── 分発レイヤー <──(Audio Chunks)──── 音声ネイティブ生成
2. コア実装:0ms ローカル即時割り込み(Local Barge-In)
音声対話において、ユーザー体験を最も左右するのが**割り込み(Barge-In)**である。
もし割り込み判定をすべてサーバー側(サーバーの VAD が発話を検知してからブラウザへ停止指示を返す)に依存していると、往復レイテンシ(RTT)の分だけ遅れが生じる。結果として、「こっちが声を出して遮ろうとしているのに、イヤホンの中の AI がさらに半秒ほど喋り続ける」という不快なラグが発生する。
そこで私は、第一防御ラインをクライアント側のマイクキャプチャ層に配置する設計をとった。
クライアントが音声をサンプリングする際、ローカルで音声エネルギーの二乗平均平方根(RMS)を計算する。AI がスピーカーから発声中であり、かつユーザーのマイク入力がしきい値を超えた瞬間、フロントエンドが 0 ミリ秒でローカル再生を即座に破棄し、同時にバックエンドへ割り込みシグナルを非同期送信する:
// クライアント側 Web Audio API リアルタイム音声処理ノード
micProcessor.onaudioprocess = (e) => {
const inputData = e.inputBuffer.getChannelData(0);
// 1. 音声エネルギーの二乗平均平方根(RMS - Root Mean Square)をローカル計算
let sumSquares = 0;
for (let i = 0; i < inputData.length; i++) {
sumSquares += inputData[i] * inputData[i];
}
const rms = Math.sqrt(sumSquares / inputData.length);
// 2. 割り込みロジック:AI が発声中、かつマイク入力がしきい値(0.025)を超えた場合
if (rms > 0.025 && activeAudioSources.length > 0) {
// 再生キューに入っているすべての AudioBufferSourceNode を即座に停止(0ms ミュート)
stopAllLocalAudio();
// Go バックエンドへ非同期で割り込み通知を送信し、モデル側の生成を停止させる
ws.send(JSON.stringify({ type: "interrupt" }));
}
// 3. 32bit 浮動小数点音声を標準の 16kHz 16-bit リニア PCM へ変換
const pcm16 = new Int16Array(inputData.length);
for (let i = 0; i < inputData.length; i++) {
const s = Math.max(-1, Math.min(1, inputData[i]));
pcm16[i] = s < 0 ? s * 0x8000 : s * 0x7FFF;
}
// 4. 音声スライスを Base64 化し、WebSocket 経由で継続的にプッシュ
ws.send(JSON.stringify({
type: "audio_chunk",
data: { pcm: arrayBufferToBase64(pcm16.buffer) }
}));
};
直感的なイメージ:イヤホンで人の話を聞いている最中、自分の声帯が振動した瞬間に自分の脳がイヤホンの音をミュートするような感覚である。相手がこちらの声を聞いて黙るのを待つ必要がない。
2. 視覚のメカニズム:なぜ動画ストリーミングではなくフレームキャプチャなのか?
視覚機能を組み込む際、H.264 等の動画ストリームや WebRTC を引くことを検討しがちである。
しかし実装してみると、リアルタイム AI 対話において連続動画ストリームを送ることはアンチパターンに近いと気づく:
- クライアント・サーバー双方のエンコード/デコード負荷が大きい
- LLM は GPU のように動画フレームをラスタライズして見ているわけではなく、画像を 2 次元のパッチ(ビジュアルトークン)に分解して理解している
実用的な解:連続高頻度フレームキャプチャ(1 FPS JPEG)
ゲームのように 60fps を維持する必要はない。画面上のコードエラーやデスク上の実物を把握させるには、毎秒 1 回(1 FPS)のサンプリングで十分すべての文脈変化を捉えられる。ネットワーク帯域やトークン消費も極めて低く抑えられる:
// フロントエンドで 1 秒(1000ms)ごとに画面共有またはカメラから 1 フレームをキャプチャ
setInterval(() => {
if (!videoElement.videoWidth) return;
// Canvas に描画し、最大 720p(1280x720)にアスペクト比を維持してスケーリング
canvas.width = Math.min(1280, videoElement.videoWidth);
canvas.height = Math.round(canvas.width * (videoElement.videoHeight / videoElement.videoWidth));
ctx.drawImage(videoElement, 0, 0, canvas.width, canvas.height);
// クオリティ 0.65 の軽量な JPEG に圧縮
const dataUrl = canvas.toDataURL("image/jpeg", 0.65);
const b64Jpeg = dataUrl.split(",")[1];
// WebSocket 経由で Go バックエンドへ送信
ws.send(JSON.stringify({
type: "screen_frame", // または video_frame (カメラ)
data: { data: b64Jpeg, ts: Date.now() }
}));
}, 1000);
Go バックエンド側では、この JPEG バイト列を Live API のビデオ入力チャネルに流し込むだけである:
// internal/genkit/live.go(抽象化コード)
func (ls *LiveSession) SendImage(jpegBytes []byte) error {
return ls.session.SendRealtimeInput(genai.LiveSendRealtimeInputParameters{
Video: &genai.Blob{
MIMEType: "image/jpeg",
Data: jpegBytes,
},
})
}
利用イメージ:IDE を画面共有しながら、「Iris、50 行目のエラーの原因は何?」と声で尋ねる。モデルは同じ WebSocket チャネル内で、左耳で PCM 音声を聞き、右目で毎秒送られてくる 720p 画面を見ながら、即座に「50 行目で閉じ括弧が不足しています」と音声で返してくる。
3. インテリジェンスの中枢:Genkit Go のツールシステムとドメイン分割
開発初期は、すべての Tool 関数(時刻取得、DB 検索、Web 検索、メモ保存)を 1 つの巨大な tools.go に書き殴っていた。結果としてファイルは数千行に膨れ上がり、メンテナンスが破綻しかけた。
そこで、**ドメインごとの分割(Domain Splitting)**を実施したところ、アーキテクチャが劇的に整理された:
internal/genkit/
├── genai.go # Genkit ランタイム初期化(G インスタンスと全体設定)
├── live.go # Gemini Live 双方向 PCM リアルタイムセッション管理
├── memory.go # pgvector による 1536 次元ベクトル検索・永続化ロジック
├── tool_system.go # ドメイン 1:物理システム時計(絶対的な時刻アンカー)
├── tool_user.go # ドメイン 2:ユーザー情報と権限管理
├── tool_memory.go # ドメイン 3:長期記憶の保存・検索ツール
├── tool_reminder.go # ドメイン 4:リマインダー・タイマー管理ツール
├── tool_search.go # ドメイン 5:DuckDuckGo Web 検索ツール
└── tools.go # 一括登録エントリーポイント(registerTools)
ツールの記述例(システム時計ツール)
LLM が頻繁に起こすミスの一つが「今日が何年何月何日かわからない」ことである。「明日の午後 3 時に教えて」と言われても、基準時刻がなければ相対時間を正確に計算できない。そのため、物理的な時計ツールを最優先で渡す:
// internal/genkit/tool_system.go
package genkit
import (
"fmt"
"time"
"github.com/firebase/genkit/go/ai"
"github.com/firebase/genkit/go/genkit"
)
// timeTool バックエンドサーバーの正確な現在時刻とタイムゾーンを返却
func timeTool() ai.ToolRef {
return genkit.DefineTool(G, "getServerTime", "バックエンドサーバーの正確な現在時刻とタイムゾーンを取得。相対時間を計算する際は必ず最初に呼び出すこと",
func(ctx *ai.ToolContext, _ struct{}) (string, error) {
nowStr := time.Now().Format("2006-01-02 15:04:05 MST")
return fmt.Sprintf("現在のサーバー時刻:%s", nowStr), nil
},
)
}
tools.go 内で、これらのツールを一括登録する:
// internal/genkit/tools.go
package genkit
func registerTools() {
Tools = append(Tools,
timeTool(), // システム時刻
saveMemoryTool(), // 記憶保存
searchMemoryTool(), // 記憶検索
setReminderTool(), // リマインダー設定
listRemindersTool(), // タスク一覧
searchWebTool(), // Web 検索
)
}
リフレクション関数 buildLiveTools() を介して、Genkit 標準で定義したツール群を Gemini Live が要求する FunctionDeclaration に自動変換している。ツール定義を 1 箇所に書くだけで、音声対話でもバックグラウンド処理でも共通利用できる。
4. 記憶の基盤:pgvector によるセマンティック検索と実践上の落とし穴
長期記憶のない AI はただの使い捨てツールである。しかし、長期記憶は単に過去のチャットログをテーブルに突っ込めば済む話ではない。
1. なぜ従来の LIKE 検索では不十分なのか?
ユーザーが「田中さんの提示した最低価格はいくらだった?」と尋ねたとする。
もしデータベースに保存されている文言が「田中氏は交渉時、15%引きが限界の譲歩ラインだと明言した」だった場合、従来の SQL WHERE content LIKE '%最低価格%' では 1 件もヒットしない。「最低価格」という単語が含まれていないからである。
ベクトル埋め込み(Embedding)による意味検索(Semantic Search)が必須となる所以である。
2. ベクトル保存と pgvector の実践
Iris では、PostgreSQL の公式拡張 pgvector と、Google の gemini-embedding-2(1536 次元空間)を採用している。
Go で pgvector を用いてコサイン距離(Cosine Distance)を計算する際、embedding <=> $1 とだけ書くと、PostgreSQL 側で operator does not exist: vector <=> text という型不一致エラーが発生することがある。パラメータ側を明示的にキャストする必要がある:
// internal/genkit/memory.go(汎用化コード)
package genkit
import (
"context"
"fmt"
"github.com/pgvector/pgvector-go"
)
// SearchMemory 意味ベクトルを基に PostgreSQL から関連する事実や好みを検索
func SearchMemory(ctx context.Context, userID int, query string, limit int) ([]*MemorySearchResult, error) {
// 1. Embedding モデルを呼び出し、ユーザーの問い合わせを 1536 次元の浮点数ベクトルへ変換
queryVector, err := GetEmbedding(ctx, query)
if err != nil {
return nil, fmt.Errorf("Embedding 生成失敗: %w", err)
}
// 2. 1 - (embedding <=> $1::vector) によりコサイン類似度(0.0 〜 1.0)を算出
// <=> は pgvector のコサイン距離演算子。距離が近いほど似ているため、1 から引くことで類似度にする
sqlQuery := `
SELECT id, title, content, category, importance,
1 - (embedding <=> $1::vector) AS similarity,
created_at
FROM memories
WHERE user_id = $2
ORDER BY similarity DESC
LIMIT $3;
`
rows, err := dbPool.Query(ctx, sqlQuery, pgvector.NewVector(queryVector), userID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var results []*MemorySearchResult
for rows.Next() {
var m MemorySearchResult
if err := rows.Scan(&m.ID, &m.Title, &m.Content, &m.Category, &m.Importance, &m.Similarity, &m.CreatedAt); err != nil {
continue
}
results = append(results, &m)
}
return results, nil
}
直感的なイメージ:意味のベクトル化とは、言葉を 1536 次元の空間内の座標へマッピングする作業である。「限界の譲歩ライン」と「最低価格」は文字面こそ違えど、空間内では極めて近い座標に位置する。空間上の距離を計算することで、膨大なメモの中から関連する文脈を即座に引き出せる。
5. プロアクティブな執事:無限ループポーリングからの脱却と Redis TTL イベント駆動
これは Iris のシステム設計の中で、比較的軽量かつ効果的だと感じているアプローチである。
1. 伝統的なタイマーポーリングの欠点
多くのアシスタントは「10 分後にリマインドして」という処理をどう実装しているか?
大抵はバックグラウンドの定期実行ループである:
// 典型的なポーリング実装:毎分 DB を検索
for range time.Tick(1 * time.Minute) {
db.Query("SELECT * FROM reminders WHERE remind_at <= NOW() AND is_sent = false")
}
この方式は、リマインダーが何もない時間帯でも毎分データベースに I/O 負荷をかけ続ける。その上、トリガーされるタイミングには最大 60 秒のズレが生じる。
2. 設計:Redis TTL Key-Expired イベント駆動アーキテクチャ
Iris ではポーリングを全廃し、イベント駆動型のカウントダウン機構を採った:
ユーザー:「10 分後に水を飲むのをリマインドして」
│
├──> 1. DB に永続化レコードを保存(確認用、is_sent = false)
├──> 2. Redis Hash にメタデータスナップショットを書き込む (reminder:data:ID)
└──> 3. Redis に正確な TTL を持つ空キーをセット:
SETEX reminder:timer:42 600 "1"
【ここから 600 秒間、システムは CPU 消費ゼロ・DB I/O ゼロで待機】
第 600 秒ちょうど、Redis のキーが自然消滅!
│
└──> Redis 内部の PubSub がイベントを自動ブロードキャスト:__keyevent@*__:expired
│
└──> Go バックエンドの単一監視ゴルーチンがイベントを即座に受信
│
├── Redis Hash からメタデータをマイクロ秒単位で取得
├── DB の is_sent を true に更新
├── WebSocket 経由でフロントエンドへ通知をプッシュ
└── Gemini Live 音声チャネルを強制ウェイクアップ!
3. コアコード
① リマインダー登録(AddReminder)
// internal/reminder/scheduler.go(抜粋)
func AddReminder(ctx context.Context, userID int, content string, remindAt time.Time) error {
ttl := time.Until(remindAt)
if ttl <= 0 {
return fmt.Errorf("リマインド時刻は未来を指定してください")
}
// 1. DB に永続化
rem, _ := db.Client.Reminder.Create().
SetUserID(userID).
SetContent(content).
SetRemindAt(remindAt).
Save(ctx)
// 2. 到期時に DB を引かずに済むよう、Redis Hash にスナップショットを保持
payload, _ := json.Marshal(map[string]any{
"id": rem.ID, "user_id": userID, "content": content,
})
_ = db.Redis.HSet(ctx, "reminder:data", rem.ID, payload).Err()
// 3. 正確な TTL を持つカウントダウンキーを発行
timerKey := fmt.Sprintf("reminder:timer:%d", rem.ID)
_ = db.Redis.Set(ctx, timerKey, "1", ttl).Err()
return nil
}
② Redis キー有効期限切れイベントの監視(StartReminderListener)
func StartReminderListener() {
ctx := context.Background()
// Redis のキー有効期限切れイベントチャネルを購読
pubsub := db.Redis.PSubscribe(ctx, "__keyevent@*__:expired")
for msg := range pubsub.Channel() {
if strings.HasPrefix(msg.Payload, "reminder:timer:") {
idStr := strings.TrimPrefix(msg.Payload, "reminder:timer:")
reminderID, _ := strconv.Atoi(idStr)
// 監視チャネルをブロックしないよう非同期で処理
go handleReminderTrigger(reminderID)
}
}
}
③ Gemini Live を能動的に「喋らせる」
期限が到来した時、画面にプッシュ通知が出るだけでは従来のアラームと変わらない。
Iris では、TurnComplete: true を付与したメッセージを注入し、現在接続中の Gemini Live セッションに自発的に声を発させる:
// internal/genkit/live.go
func (ls *LiveSession) NotifyReminder(content string) error {
prompt := fmt.Sprintf("【システムイベント:リマインダー時刻到来】ユーザーが設定した「%s」の時間になりました。Iris として、落ち着いた自然な口調でユーザーに直接用件を伝えてください。システム通知であることの言い訳は不要です。", content)
// ClientContent を送信し、TurnComplete を true に設定
return ls.session.SendClientContent(genai.LiveClientContentInput{
Turns: []*genai.Content{
{
Role: "user",
Parts: []*genai.Part{{Text: prompt}},
},
},
TurnComplete: new(true), // これによりモデルが即座に応答を開始し、音声を自発的に出力する
})
}
実際の動作:イヤホンをつけて作業していると、10 分経った瞬間にキーボードに一切触れていないにもかかわらず、イヤホン越しに Iris が「時間になりました、お茶を淹れる時間ですよ」と自然に話しかけてくる。
6. アーキテクチャの教訓とトレードオフの哲学
モジュールを組み合わせる過程で、2 点ほど設計上の重要な判断があった:
1. Go の循環参照(Import Cycle)を回避する「コールバック注入パターン」
能動的リマインダーを実装する際、古典的な循環参照の罠にぶつかった:
-
reminderパッケージは、期限切れ時に通知を送るためwsパッケージを呼び出したい -
wsパッケージは、クライアント接続を管理し、音声合成のためにgenkitパッケージを呼び出す -
genkit内のツール(tool_reminder.go)は、新しいリマインダーを追加するためreminderパッケージを呼び出す
Go コンパイラはこれを許さない:
imports reminder -> imports ws -> imports genkit -> imports reminder: import cycle not allowed
イベントバスのような大掛かりな仕組みを持ち込まずに解決するため、関数シグネチャ注入パターン(Notifier Pattern) を採用した:
// 1. 下位の internal/reminder/scheduler.go では関数の型と変数のみを定義
type NotifierFunc func(userID uint, reminderID int, content, remindAt string)
var Notifier NotifierFunc
func SetNotifier(fn NotifierFunc) {
Notifier = fn
}
// 期限切れ時は登録されたコールバックを呼ぶだけ。ws パッケージの存在を一切知らない
if Notifier != nil {
Notifier(uint(payload.UserID), payload.ID, payload.Content, payload.RemindAt)
}
そして上位の internal/ws/hub.go の初期化時に、自身の処理関数をそこへ注入する:
// internal/ws/hub.go
func (m *Manager) Start() {
// 上位の処理実体を下位パッケージの関数ポインタへ差し込む
reminder.SetNotifier(func(userID uint, reminderID int, content, remindAt string) {
// WebSocket プッシュおよび Gemini Live への音声発話リクエスト...
})
// ...
}
下位は上位を知らず、上位が結合を担当する。これで参照の環はきれいに解消された。
2. オーバーエンジニアリングの拒絶(YAGNI 原則の遵守)
リマインダーの重複実行防止を考えていた際、Redis 分布式ロックを入れるべきか、PostgreSQL 上で厳密な排他制御(UPDATE ... WHERE is_sent = false)を前置すべきか悩んだ。
しかし自問した結果:Iris は現在、私個人がローカルマシン上で動かしているプライベートアシスタントである。
単一の Redis インスタンスにおいて、キーの期限切れイベントは基本的に 1 回しか届かない。さらにデータベースのフォールバッククエリ側にも Where(reminder.IsSent(false)) を添えてある。
架空の「数千万トラフィックの分散競合」を想定して無駄なロック機構を挟み込めば、マイクロ秒単位で動くインメモリの良さを損ない、トラブルシュートの工数が増えるだけである。
「存在しない規模のためにコストを払わない(YAGNI)」。個人開発のシステム設計において、シンプルさと応答性の良さを優先することは常に正しい選択だと感じている。
7. 現状と今後の TODO
ここまで作り込んできたことで、このシステムは手元の環境でかなり安定して稼働している。全二重の音声対話、画面を見せながらのコード相談、記憶の自動引き出し、そして指定時刻に能動的に声をかけてくる機能など、欲しかった核となる部分は概ねスムーズに動いており、日常のウィンドウ切り替えや文字入力の手間を減らしてくれている。
もちろん、まだまだ荒削りな部分もあり、今後手を入れたい課題はいくつかある:
- 長時間の再接続耐性:バックグラウンドで長時間放置した際の WebSocket 切断時、現在の自動再接続処理はやや粗く、手動リロードが必要になることがある
- 完全ハンズフリー化(Wake Word):現状は対話開始時にブラウザのボタンを一度クリックする必要がある。将来的にはローカルで動く軽量なキーワード検知をフロントエンドに組み込み、「Hey Iris」の一声で拾えるようにしたい
- ローカル環境の直接操作:現状は「見ること」と「話すこと」に留まっている。権限を絞ったローカルスクリプト連携を通じて、PC のシステム負荷確認や音楽再生、特定の開発プロジェクトを開くといった操作も任せられるようにしたい
ソフトウェア開発とはそういうものであり、まずは最も重要なコアパイプラインを通し、残りの体験は少しずつ自分の手で補完していくのが心地よい。