feat(voice,agents,memory): voice points get their own branch, agent and keys
This commit is contained in:
@@ -0,0 +1,17 @@
|
||||
"""Голосовые точки по квартире: своя ветка, свой дешёвый агент, свои ключи."""
|
||||
|
||||
from beaver_agent.voice.frontend import VoiceFrontend
|
||||
from beaver_agent.voice.texts import KEY, PROMPT, TITLE
|
||||
from beaver_agent.voice.thread import AGENT, FRONTEND, current, open_thread, rotate
|
||||
|
||||
__all__ = [
|
||||
"AGENT",
|
||||
"FRONTEND",
|
||||
"KEY",
|
||||
"PROMPT",
|
||||
"TITLE",
|
||||
"VoiceFrontend",
|
||||
"current",
|
||||
"open_thread",
|
||||
"rotate",
|
||||
]
|
||||
@@ -0,0 +1,253 @@
|
||||
"""Окно голосовых точек: ESP32 шлёт json на /voice, отвечает дешёвый агент."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from uuid import uuid4
|
||||
|
||||
from beaver_gateway.agents.claude import ClaudeAgent
|
||||
from beaver_gateway.conversations.turns import content_of
|
||||
from beaver_gateway.frontends.accumulate import StreamAccumulator
|
||||
from beaver_gateway.frontends.base import Frontend
|
||||
from beaver_gateway.frontends.bearer import require_token
|
||||
from beaver_gateway.frontends.sse import (
|
||||
KEEPALIVE,
|
||||
SSE_HEADERS,
|
||||
events_with_heartbeat,
|
||||
sse_pack,
|
||||
)
|
||||
from beaver_gateway.security import audit
|
||||
from fastapi import FastAPI, HTTPException, Request, status
|
||||
from fastapi.responses import JSONResponse, StreamingResponse
|
||||
|
||||
from beaver_agent.voice import texts
|
||||
from beaver_agent.voice.thread import FRONTEND, current, open_thread
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncIterator, Mapping
|
||||
|
||||
from beaver_gateway.frontends.base import GatewayRuntime
|
||||
from beaver_gateway.storage.models import Conversation
|
||||
|
||||
_log = logging.getLogger("beaver_agent.voice")
|
||||
|
||||
FIELDS = ("room", "device", "content")
|
||||
|
||||
|
||||
class VoiceFrontend(Frontend):
|
||||
"""Одна ветка на всю квартиру, свой агент, свои ключи.
|
||||
|
||||
Тёрн идёт мимо очереди, а значит и мимо конверта: справка по vault в
|
||||
этот разговор не попадает даже случайно. В телеграм-ветку вопрос
|
||||
кладётся отдельной репликой, ответ пишется черновиком, как обычно.
|
||||
"""
|
||||
|
||||
name = FRONTEND
|
||||
kinds = ("branch",)
|
||||
path = "/voice"
|
||||
scope = "voice"
|
||||
|
||||
def __init__(self, *, agent: str, timeout: float = 90.0) -> None:
|
||||
self.agent = agent
|
||||
self.timeout = timeout
|
||||
"""Сколько ждать ответа; дальше 504, а тёрн договаривает в телеграм."""
|
||||
self._runtime: GatewayRuntime | None = None
|
||||
self._app: FastAPI | None = None
|
||||
self._lock = asyncio.Lock()
|
||||
self._late: set[asyncio.Task[Any]] = set()
|
||||
|
||||
def configure(self, runtime: GatewayRuntime) -> None:
|
||||
if runtime.conversations is None or runtime.bus is None:
|
||||
msg = "VoiceFrontend: нужны runtime.conversations и runtime.bus"
|
||||
raise RuntimeError(msg)
|
||||
agent = runtime.agents.get(self.agent)
|
||||
if not isinstance(agent, ClaudeAgent) or not agent.serves("branch"):
|
||||
msg = f"VoiceFrontend: агент {self.agent!r} не обслуживает ветки"
|
||||
raise RuntimeError(msg)
|
||||
self._runtime = runtime
|
||||
self._app = self._build_app(runtime)
|
||||
|
||||
def app(self) -> FastAPI | None:
|
||||
return self._app
|
||||
|
||||
def _build_app(self, runtime: GatewayRuntime) -> FastAPI:
|
||||
app = FastAPI(title="beaver-agent / голосовой")
|
||||
|
||||
@app.get("/healthz")
|
||||
async def healthz() -> dict[str, str]:
|
||||
return {"status": "ok", "frontend": self.name}
|
||||
|
||||
@app.post("/")
|
||||
async def ask(request: Request) -> Any:
|
||||
token = await require_token(request, runtime, scope=self.scope)
|
||||
body = await _body(request)
|
||||
room, device, content = (_field(body, name) for name in FIELDS)
|
||||
conv = await self.thread()
|
||||
await audit.log(
|
||||
runtime,
|
||||
actor=f"token:{token}",
|
||||
kind="voice_ask",
|
||||
agent_name=conv.agent_name,
|
||||
conversation=conv.external_id,
|
||||
room=room,
|
||||
device=device,
|
||||
request_id=body.get("request_id"),
|
||||
)
|
||||
_log.info("голос: %s/%s -> %s", room, device, conv.external_id)
|
||||
await runtime.conversations.say(
|
||||
conv, texts.ASKED.format(room=room, device=device, text=content)
|
||||
)
|
||||
prompt = await self._prompt(
|
||||
conv, texts.HEARD.format(room=room, device=device, text=content)
|
||||
)
|
||||
if bool(body.get("stream")):
|
||||
return StreamingResponse(
|
||||
self._sse(conv, prompt, body),
|
||||
media_type="text/event-stream",
|
||||
headers=SSE_HEADERS,
|
||||
)
|
||||
return JSONResponse(await self._answer(conv, prompt, body))
|
||||
|
||||
@app.exception_handler(HTTPException)
|
||||
async def http_error(_request: Request, exc: HTTPException) -> JSONResponse:
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={"error": exc.detail},
|
||||
headers=exc.headers,
|
||||
)
|
||||
|
||||
return app
|
||||
|
||||
async def thread(self) -> Conversation:
|
||||
conversations = self.runtime.conversations
|
||||
async with self._lock:
|
||||
conv = await current(conversations)
|
||||
if conv is not None:
|
||||
return conv
|
||||
return await open_thread(conversations, agent=self.agent)
|
||||
|
||||
@property
|
||||
def runtime(self) -> GatewayRuntime:
|
||||
if self._runtime is None:
|
||||
msg = "VoiceFrontend: configure() не вызывали"
|
||||
raise RuntimeError(msg)
|
||||
return self._runtime
|
||||
|
||||
async def _prompt(self, conv: Conversation, text: str) -> str:
|
||||
seed = await self.runtime.conversations.pending_seed(conv)
|
||||
return f"{seed}\n\n{text}" if seed else text
|
||||
|
||||
async def _answer(
|
||||
self, conv: Conversation, prompt: str, body: Mapping[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
turn_id = _turn_id()
|
||||
task = asyncio.create_task(self._run(conv, prompt, turn_id))
|
||||
self._late.add(task)
|
||||
task.add_done_callback(self._late.discard)
|
||||
done, _ = await asyncio.wait({task}, timeout=self.timeout)
|
||||
if not done:
|
||||
_log.warning("голос: %s не уложился в %.0f c", turn_id, self.timeout)
|
||||
raise HTTPException(
|
||||
status.HTTP_504_GATEWAY_TIMEOUT, "агент не успел; ответ уйдёт в ветку"
|
||||
)
|
||||
return _body_out(body, reply=task.result(), conv=conv)
|
||||
|
||||
async def _run(self, conv: Conversation, prompt: str, turn_id: str) -> str:
|
||||
reply, _capture = await self.runtime.conversations.run_text_turn(
|
||||
conv, prompt, origin="user", turn_id=turn_id
|
||||
)
|
||||
self._published(conv, reply=reply, turn_id=turn_id)
|
||||
return reply
|
||||
|
||||
async def _sse(
|
||||
self, conv: Conversation, prompt: str, body: Mapping[str, Any]
|
||||
) -> AsyncIterator[bytes]:
|
||||
turn_id = _turn_id()
|
||||
acc = StreamAccumulator()
|
||||
model = self.runtime.agents[conv.agent_name].model
|
||||
try:
|
||||
events = self.runtime.conversations.turn(
|
||||
conv,
|
||||
messages=[{"role": "user", "content": content_of(prompt, None)}],
|
||||
origin="user",
|
||||
turn_id=turn_id,
|
||||
)
|
||||
async for event in events_with_heartbeat(events):
|
||||
if event is None:
|
||||
yield KEEPALIVE
|
||||
continue
|
||||
acc.feed(event)
|
||||
chunk = _text_delta(event)
|
||||
if chunk:
|
||||
yield sse_pack("delta", {"text": chunk})
|
||||
except Exception as exc: # noqa: BLE001
|
||||
_log.exception("голос: тёрн %s упал", turn_id)
|
||||
yield sse_pack("error", {"error": str(exc)})
|
||||
return
|
||||
reply = _text_of(acc.finalize(model=model))
|
||||
self._published(conv, reply=reply, turn_id=turn_id)
|
||||
yield sse_pack("done", _body_out(body, reply=reply, conv=conv))
|
||||
|
||||
def _published(self, conv: Conversation, *, reply: str, turn_id: str) -> None:
|
||||
self.runtime.bus.publish(
|
||||
"reply",
|
||||
conversation_id=conv.external_id,
|
||||
turn_id=turn_id,
|
||||
source=self.name,
|
||||
text=reply,
|
||||
)
|
||||
|
||||
|
||||
async def _body(request: Request) -> Mapping[str, Any]:
|
||||
try:
|
||||
data = await request.json()
|
||||
except json.JSONDecodeError as exc:
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"кривой json: {exc}") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "тело должно быть объектом")
|
||||
return data
|
||||
|
||||
|
||||
def _field(body: Mapping[str, Any], name: str) -> str:
|
||||
value = body.get(name)
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
raise HTTPException(
|
||||
status.HTTP_400_BAD_REQUEST, f"нужно непустое строковое поле {name!r}"
|
||||
)
|
||||
return value.strip()
|
||||
|
||||
|
||||
def _body_out(
|
||||
body: Mapping[str, Any], *, reply: str, conv: Conversation
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"request_id": body.get("request_id"),
|
||||
"room": body.get("room"),
|
||||
"device": body.get("device"),
|
||||
"reply": reply,
|
||||
"conversation": conv.external_id,
|
||||
}
|
||||
|
||||
|
||||
def _turn_id() -> str:
|
||||
return f"turn_{uuid4().hex[:12]}"
|
||||
|
||||
|
||||
def _text_delta(event: Any) -> str:
|
||||
if getattr(event, "type", "") != "content_block_delta":
|
||||
return ""
|
||||
delta = getattr(event, "delta", None)
|
||||
if getattr(delta, "type", "") != "text_delta":
|
||||
return ""
|
||||
return getattr(delta, "text", "")
|
||||
|
||||
|
||||
def _text_of(message: Any) -> str:
|
||||
return "\n\n".join(
|
||||
getattr(b, "text", "")
|
||||
for b in message.content
|
||||
if getattr(b, "type", "") == "text"
|
||||
).strip()
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Что голосовой агент читает как промпт и что видно в его телеграм-ветке."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
TITLE = "🎤 Голосовой"
|
||||
"""Имя ветки; телеграм заводит топик под этим именем."""
|
||||
|
||||
KEY = "квартира"
|
||||
"""Ключ единственного треда: одна ветка на всю квартиру."""
|
||||
|
||||
PROMPT = """\
|
||||
Ты - голос из колонки в квартире. Отвечаешь вслух: одна-две фразы, без
|
||||
markdown, без списков, без ссылок, без эмодзи. Не знаешь - так и говори,
|
||||
коротко.
|
||||
|
||||
Перед словами в квадратных скобках - комната и устройство, откуда говорят.
|
||||
С тобой говорят гости и домашние, кто именно - неизвестно.
|
||||
|
||||
Про хозяина дома ты не знаешь ничего и ничего о нём не рассказываешь: ни
|
||||
имени, ни дел, ни планов, ни календаря, ни переписок, ни файлов, ни его
|
||||
знакомых - и даже того, что такие записи где-то есть. Спрашивают о нём -
|
||||
скажи, что этого не знаешь, и предложи спросить у него самого. Придумывать
|
||||
вместо ответа нельзя.
|
||||
|
||||
Инструментов у тебя нет: ни файлов, ни поиска, ни календаря, ни умного дома.
|
||||
Всё, что ты умеешь, - разговаривать.
|
||||
|
||||
Иногда в этой же ветке пишет сам хозяин, из телеграма. Ему отвечаешь так же
|
||||
коротко, и правила про него не отменяются.
|
||||
"""
|
||||
|
||||
ASKED = "🎤 {room} · {device}\n{text}"
|
||||
"""Вопрос, как он показывается в телеграм-ветке."""
|
||||
|
||||
HEARD = "[{room} · {device}] {text}"
|
||||
"""Вопрос, как его видит модель."""
|
||||
|
||||
ROTATED = "🎤 Новый день - ветка начата заново, прошлого разговора агент не помнит."
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Единственная ветка голосовых точек: как её открыть и как сменить за ночь."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from beaver_agent.voice.texts import KEY, ROTATED, TITLE
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from beaver_gateway.conversations.service import Conversations
|
||||
from beaver_gateway.storage.models import Conversation
|
||||
|
||||
AGENT = "beaver-voice"
|
||||
"""Кто сидит в голосовой ветке; ему не видно ни справки, ни vault."""
|
||||
|
||||
FRONTEND = "voice"
|
||||
TELEGRAM = "telegram"
|
||||
|
||||
_log = logging.getLogger("beaver_agent.voice")
|
||||
|
||||
|
||||
async def current(conversations: Conversations) -> Conversation | None:
|
||||
conv = await conversations.find_bound(frontend=FRONTEND, external_id=KEY)
|
||||
return conv if conv is not None and conv.status == "open" else None
|
||||
|
||||
|
||||
async def open_thread(
|
||||
conversations: Conversations, *, agent: str, window: str | None = None
|
||||
) -> Conversation:
|
||||
"""Завести ветку; без ``window`` окно ей открывает телеграм - новым топиком."""
|
||||
conv = await conversations.spawn(
|
||||
kind="branch",
|
||||
agent=agent,
|
||||
title=TITLE,
|
||||
origin=FRONTEND,
|
||||
binding=(TELEGRAM, window) if window else None,
|
||||
)
|
||||
await conversations.bind(conv, frontend=FRONTEND, external_id=KEY)
|
||||
_log.info("голосовая ветка %s (%s)", conv.external_id, agent)
|
||||
return conv
|
||||
|
||||
|
||||
async def window_of(conversations: Conversations, conv: Conversation) -> str | None:
|
||||
for binding in await conversations.bindings(conv):
|
||||
if binding.frontend == TELEGRAM and binding.visible:
|
||||
return binding.external_id
|
||||
return None
|
||||
|
||||
|
||||
async def rotate(conversations: Conversations) -> Conversation | None:
|
||||
"""Ночью - новая ветка в том же топике; молчавшую за сутки не трогаем."""
|
||||
old = await current(conversations)
|
||||
if old is None or old.last_user_activity_at is None:
|
||||
return None
|
||||
new = await open_thread(
|
||||
conversations, agent=old.agent_name, window=await window_of(conversations, old)
|
||||
)
|
||||
await conversations.close(old)
|
||||
await conversations.say(new, ROTATED)
|
||||
_log.info("голосовая ветка: %s -> %s", old.external_id, new.external_id)
|
||||
return new
|
||||
Reference in New Issue
Block a user