アーカイブ済み(2026-09-15): 本ドキュメントは議論途中の中間成果物であり、内容は../ui-brain-protocol.md(一般プロトコル)と../voice-ui-spec.md(音声UI仕様、付録)に整理・統合された。今後の参照・更新は上記2文書を使うこと。本書は議論の経緯を追う場合の参考としてのみ残す。
ステータス: WLS側(Brain)は実装・テスト完了・push済み。Windows側の方針は再検討中(未決定)。
関連: AiChatリポジトリ issue #8(経緯・議論の詳細はissueコメント参照)。brain-request-idempotency.md(Brain⇄Windows間の通信失敗時の冪等リトライプロトコル。下記の経緯から必要性が判明した)。
2026-09-15 検討中の論点: 当初案はwsl-brain-separation-plan.mdのステージ1本体カットオーバー(Assistant.__init__のOllamaChat→RemoteChat移行)をこの機能で前倒しする設計だった。これに対しユーザーから「Brain(WSL)が落ちている時にWindows側がどう振る舞うべきかを、音量機能だけの特別扱いとして安易に(ローカルへの実装重複という形で)決めるな」という指摘があり、いったん撤回した。
議論の末、「Brainが落ちている時の振る舞い」は音量固有の問題ではなく、Windows↔Brain間の通信全般(/askが失敗した時にどうするか)に共通する、より根本的なプロトコルの問題だと整理された。この整理に基づき、まずbrain-request-idempotency.mdで「request_idによる冪等な再送」というプロトコル部分を定義した(実装未着手)。これにより、Brainが一時的に落ちていてもWindows側は安全にリトライでき、音量指示についても「そのターンが失敗する」以上の特別な対応は不要になる、という見立て。
ただし、この整理でBrain分離(cutover)自体を今回前倒しするかどうかという当初の論点が解消されたと言えるかはまだ確認していない。「一時的な通信断からのリトライ」と「Brainが長時間ダウンしている間、Windows側だけで最低限動き続けたい」は別の要求である可能性があり、後者が依然として必要なら、cutoverの前倒しは見送り、当面はWindows側のローカルOllamaChatのままにする、という判断も残っている。次に決めるべきはここ。
2026-09-15、Codexによるレビューを受けて設計を一部修正した(本ドキュメントは修正反映後の版)。主な変更点は末尾の「レビューで判明した修正点」参照。
issue #8はもともと.envのAUDIO_OUTPUT_VOLUMEによる静的な倍率設定だったが、「ユーザーが音声で指示したら音量が変わる」機能へスコープを変更した。
「もう少し大きく」「うるさいから下げて」のような言い回しの揺れを吸収するため、既存の天気・Web検索と同じLLM tool-calling方式で意図認識する。この判断(自然言語の意図理解)はBrain(WSL側)の責務であり、実際の音量状態・音声再生はWindows側の責務、という役割分担を採用する。ただし現時点ではWindows側はBrainへネットワーク越しに接続せず、同じ役割分担・同じコードパターンをWindows側自身のローカルOllamaChat/tools.pyの中で完結させる(方針転換の理由は冒頭参照)。
操作対象はこのアプリケーション内部の再生倍率(_apply_volume_pcm16によるPCM16スケーリング)であり、Windows OS自体のマスター音量ではない。
direction)だけを返す。Brainは音量の絶対値もステップ幅も知らない設計とする(現在値をリクエストに含めてBrainへ送る必要をなくすため)。したがって本機能はまず相対指定(上げる/下げる)のみをサポートする。「50%にして」のような絶対値指定は本ドキュメントの対象外とし、必要になった時点で別途拡張する。
/askレスポンスのスキーマ変更現状(server.py):
class AskResponse(BaseModel):
answer: str
変更後:
from typing import Any # 現状のserver.pyに未importのため追加が必要
class AskResponse(BaseModel):
conversation: str
control: list[dict[str, Any]] = []
conversation: 読み上げる自然文。何も喋ることがなければ空文字を許容する。control: クライアントが解釈・実行すべきアクションのリスト。0個以上。後方互換は不要と判断した(稼働中のクライアントは1つのみ、開発中のため)。answerフィールドは廃止し、旧フィールド名を残す分岐は作らない。
control要素の例(音量、adjust_volumeツールに合わせて命名。set_volumeだと絶対値設定に見えるため避ける):
{"type": "adjust_volume", "direction": "up"}
数値(delta)は含めない。ステップ幅はWindows側の定数として持つ(役割分担セクション参照)。
_is_bad_answerによる1回だけの再試行や、1回の/ask内の複数tool round(max_tool_rounds=3)で、モデルが同じツールを複数回呼ぶ可能性がある。Brain側で、同じtypeのcontrolアクションは1回の/askにつき最大1件に間引く(詳細は後述のllm.py変更点)。クライアント側での重複除去には頼らない(Brainが唯一の発生源であり、責務がそこにあるため)。
controlはネットワーク越しに届く外部入力として扱い、Windows側で最低限次を検証する: typeが既知の値か、directionが"up"/"down"のいずれかか。未知のtype・不正なdirectionは無視してログに残す(例外にしない)。現時点ではアクション種別がadjust_volumeのみなので、専用の判別共用体モデルまでは導入しない(種別が増えたら再検討する)。
tools.pyToolSpecにis_client_actionフラグを追加する(既定False、既存ツールは変更不要):
@dataclass(slots=True)
class ToolSpec:
name: str
description: str
parameters: dict[str, Any]
handler: Callable[..., Any]
is_client_action: bool = False
ToolRegistryに、指定した名前のツールがクライアントアクションかどうかを返すメソッドを追加する:
def is_client_action(self, name: str) -> bool:
spec = self._specs.get(name)
return spec.is_client_action if spec else False
新規ツールadjust_volumeをbuild_default_tool_registry(または専用の登録関数)に追加する。ハードウェアにも数値ポリシーにも触れず、方向を検証して返すだけ:
def adjust_volume(direction: str) -> dict[str, object]:
if direction not in ("up", "down"):
raise ToolError(f"未知の方向です: {direction}")
return {"type": "adjust_volume", "direction": direction}
ToolSpec(
name="adjust_volume",
description=(
"ユーザーが音量を上げてほしい/下げてほしいと発話した時に呼ぶ"
"(例:「もっと大きく」「うるさいから下げて」「音量下げて」)。"
"具体的な数値は分からないので、directionだけを渡す。"
),
parameters={
"type": "object",
"properties": {
"direction": {
"type": "string",
"enum": ["up", "down"],
"description": "上げるならup、下げるならdown",
},
},
"required": ["direction"],
},
handler=adjust_volume,
is_client_action=True,
)
llm.pyOllamaChat.ask()の戻り値をstrから、回答文とcontrolリストを両方持つ形へ変更する:
@dataclass(slots=True)
class AskResult:
answer: str
control: list[dict[str, Any]]
_run_tool_callsはcontrolリストを受け取り、同じtypeが既に含まれる場合は追加しない(ツール自体は毎回実行してLLMへのtoolメッセージは通常通り返す。クライアントへ二重に届けないだけ):
def _run_tool_calls(self, messages, tool_calls, control: list[dict]) -> None:
seen_types = {item.get("type") for item in control}
for call in tool_calls:
name = call["function"]["name"]
arguments = call["function"].get("arguments") or {}
try:
result = self._tools.call(name, arguments)
if self._tools.is_client_action(name):
action_type = result.get("type")
if action_type not in seen_types:
control.append(result)
seen_types.add(action_type)
content = json.dumps(result, ensure_ascii=False)
except ToolError as error:
content = json.dumps({"error": str(error)}, ensure_ascii=False)
messages.append({"role": "tool", "tool_name": name, "content": content})
_completeはcontrol: list[dict] = []をローカルで生成して_run_tool_callsに渡し続け、最終的に(answer, control)を返す。ask()は既存の再試行ロジック(_is_bad_answer時の1回だけの再試行)をそのまま維持しつつ、同じcontrolリストを2回目の_complete呼び出しにも引き継ぐ(新しいリストを作り直さない。これによりリトライで同じツールが再度呼ばれても上記のdedupが効く)。
フォールバックの修正: 既存のif not answer: answer = "うまく答えられませんでした。もう一度お願いします。"はcontrolを考慮していない。controlが空でない場合は上書きしない:
if not answer and not control:
answer = "うまく答えられませんでした。もう一度お願いします。"
server.pyfrom typing import Any
class AskResponse(BaseModel):
conversation: str
control: list[dict[str, Any]] = []
@app.post("/ask", response_model=AskResponse)
def ask(request: AskRequest) -> AskResponse:
...
result = session.chat.ask(request.text, client_type=request.client_type)
return AskResponse(conversation=result.answer, control=result.control)
この節はcutover(OllamaChat→RemoteChatへの切替)を実施すると決まった場合の実装案であり、cutoverを今回前倒しするかどうか自体はまだ決定していない(冒頭の「検討中の論点」参照)。決定後、この節の要否を再確認すること。
現状app.pyのAssistant.__init__はローカルのOllamaChatを直接構築している。本機能を役割分担通りに動かすには、Brainの/askを実際に呼ぶ必要がある。これはwsl-brain-separation-plan.mdで既に合意されていたステージ1のクライアント切替(OllamaChat→RemoteChat)そのものであり、今回はこれを前倒しで実施する案。
RemoteChat(llm.pyに追加、またはHTTPクライアント用の新規モジュール)request_idによる冪等リトライはbrain-request-idempotency.mdで定義したプロトコルに従う。リトライ回数・タイムアウト値・失敗時の振る舞いはそちらの通りWindows側の実装判断であり、以下は一例(具体値は未確定):
class RemoteChat:
def __init__(self, brain_url: str, session_id: str) -> None:
self.brain_url = brain_url.rstrip("/")
self.session_id = session_id
self._client = httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0))
def close(self) -> None:
self._client.close()
def health(self) -> None:
self._client.get(f"{self.brain_url}/health/ready").raise_for_status()
def ask(self, text: str, client_type: str = "voice") -> tuple[str, list[dict]]:
request_id = str(uuid4())
payload = {
"session_id": self.session_id,
"request_id": request_id,
"text": text,
"client_type": client_type,
}
last_error: Exception | None = None
for _attempt in range(2): # 初回+1回だけ再試行(値は暫定、Windows側の判断)
try:
response = self._client.post(f"{self.brain_url}/ask", json=payload)
response.raise_for_status()
data = response.json()
return data["conversation"], data.get("control", [])
except httpx.HTTPError as error:
last_error = error
raise last_error # 呼び出し元(app.py)が失敗時の振る舞いを決める
def clear_history(self) -> None:
response = self._client.post(
f"{self.brain_url}/clear_history", json={"session_id": self.session_id}
)
response.raise_for_status()
close()の呼び出し元は既存のAssistant.close()がself.chat.close()を無条件で呼ぶ構造になっているため、RemoteChatにclose()を実装するだけで自動的に呼ばれる(追加の配線は不要)。
config.pyBRAIN_URL設定を追加する(wsl-brain-separation-plan.mdで既に想定済みの項目)。
app.pyAssistant.__init__: self.chat = OllamaChat(...) → self.chat = RemoteChat(settings.brain_url, session_id=str(uuid4()))に置き換える。tools.py・ローカルツール群(LocationTool等)の生成はここでは不要になる(Brain側が保持するため)。_VOLUME_STEP = 0.1
controlアクションを適用し、実際にはクランプされて変化しなかった場合はBrainの発話を信用せず上書きする(Brainは現在値を知らないため「上げました」と言っても実際には上限で変化していない場合がある):def _apply_control_actions(self, control: list[dict]) -> str | None:
"""controlを適用する。クランプにより実際は変化しなかった場合、代わりに読み上げるべき文言を返す。"""
override: str | None = None
for action in control:
if action.get("type") != "adjust_volume":
logger.warning("未知のcontrolアクションです: %s", action)
continue
direction = action.get("direction")
if direction not in ("up", "down"):
logger.warning("不正なdirectionです: %s", action)
continue
step = _VOLUME_STEP if direction == "up" else -_VOLUME_STEP
before = self.settings.audio_output_volume
after = max(0.0, min(2.0, before + step))
self.settings.audio_output_volume = after
if after == before:
override = "これ以上大きくできません" if direction == "up" else "これ以上小さくできません"
return override
_interact内の呼び出し順序: controlはconversationを読み上げる前に必ず適用する(新しい音量で読み上げが再生されるように)。また、conversationが空文字になり得る設計上、既存のif not answer: return(会話を終了させる分岐)は**control適用より先に評価してはならない**。無音の成功(将来的に喋ることのないアクションが増えた場合)でもcontrolの適用自体は行い、そのターンを継続する:conversation, control = self.chat.ask(text)
override = self._apply_control_actions(control)
speak_text = override or conversation
if speak_text:
self._play_ack_tone()
self._speak(speak_text)
microphone.clear()
start_timeout_sec = self.settings.conversation_timeout_sec
(サーバー側フォールバック修正によりconversationとcontrolが両方空になることは通常起きないが、クライアント側でもspeak_textが空なら単に喋らないだけにし、returnで会話を終了させない。)
POST /session/{id}/keepaliveの定期呼び出し)。24時間以内の通常利用では影響しないため、別途対応する。keina doctor / keina chat / scripts/evaluate-tool-use.pyのBrain切替(wsl-brain-separation-plan.mdの対象範囲表に記載の残り項目)。.envへの書き戻し)。今回はプロセス起動中のみ有効(再起動でリセット)とする。[conversation]/[control]をテキストタグとして埋め込む案(検討したが、JSON応答の中に自作DSLを重ねる形になり、クライアント側のパーサーがエスケープ由来の脆さを抱えるため不採用)。request_id)はbrain-request-idempotency.mdへ分離した。本ドキュメントはそちらを前提とする。tests/test_tools.py(AiChatWls): adjust_volumeの正常系(up/down)・異常系(未知のdirection)。tests/test_llm.py(AiChatWls):
is_client_actionなツールが呼ばれた時、AskResult.controlに結果が積まれることを確認する。controlには1件しか積まれないことを確認する。conversationが空でもcontrolが非空なら、失敗フォールバック文言に置き換えられないことを確認する。OllamaChat.ask()の戻り値を前提にしたテストをAskResultベースに更新する。tests/test_server.py(AiChatWls):
/askがconversation/controlの2フィールドで返ることを確認する。ENABLE_TOOLS=falseの場合、adjust_volumeを含むツール一式が無効化されることを確認する。tests/test_app.py(AiChat):
RemoteChatのレスポンスパース。_apply_control_actionsによるクランプ(0.0〜2.0)と、クランプ時のoverride文言。controlの適用が読み上げ(_speak)より先に行われることを確認する。directionを安全に無視することを確認する。/ask経由で音量指示→適用までを通しで確認するテスト(将来的にCIへ組み込む)。初版からの主な変更:
controlは1件に間引く。conversationと失敗フォールバックの矛盾を修正(必須): controlが非空なら失敗文言で上書きしない。directionのみ返し、ステップ幅はWindows側の定数にした。役割分担がより明確になった。type・不正なdirectionは無視する。軽微な修正: ツール名に合わせてcontrolのtypeをset_volumeからadjust_volumeへ改名、server.pyへのtyping.Anyのimport追加、RemoteChat.clear_history()へのraise_for_status()追加、brain_urlのrstrip("/")、操作対象がアプリ内倍率でありWindows OSのマスター音量ではないことを明記。