feat: transcribe QQ voice messages to text via GLM-ASR

CLAUDE.md

@@ -70,6 +70,8 @@ src/agent_box/
 - `WEIXIN_ACCOUNT_ID` — weixin_sdk account id (from login)
 - `QQBOT_APP_ID` — QQ Bot application ID
 - `QQBOT_CLIENT_SECRET` — QQ Bot client secret
+- `GLM_API_KEY` — ZhipuAI (GLM) API key for voice-to-text. Empty skips QQ voice transcription (falls back to file path only).
+- `GLM_ASR_MODEL` — GLM ASR model id (default: `glm-asr-2512`)
 - `PROJECTS_DIR` — where project folders live (default: `data/projects`)
 - `ROUTER_MODEL` — model override for router (optional)
 - `AGENT_PERMISSION_MODE` — Claude Code permission mode (default: `bypassPermissions`)

src/agent_box/asr/__init__.py

@@ -0,0 +1 @@
+"""Speech-to-text (ASR) backends."""

src/agent_box/asr/glm.py

@@ -0,0 +1,78 @@
+"""GLM-ASR voice transcription client (ZhipuAI / BigModel).
+
+API: POST https://open.bigmodel.cn/api/paas/v4/audio/transcriptions
+  multipart/form-data: model, stream, file
+  Header: Authorization: Bearer <api_key>
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import httpx
+
+log = logging.getLogger(__name__)
+
+GLM_ASR_URL = "https://open.bigmodel.cn/api/paas/v4/audio/transcriptions"
+_TIMEOUT_S = 120.0
+
+
+class GlmASR:
+    """Transcribe an audio file to text via the GLM-ASR API."""
+
+    def __init__(self, api_key: str, model: str = "glm-asr-2512") -> None:
+        self._api_key = api_key
+        self._model = model
+
+    async def transcribe(self, file_path: str) -> str | None:
+        """Return the transcribed text, or None on any failure.
+
+        Never raises — callers can rely on a None to mean "fall back gracefully".
+        """
+        p = Path(file_path)
+        if not p.is_file():
+            log.warning("GLM ASR: file not found: %s", file_path)
+            return None
+
+        try:
+            data = p.read_bytes()
+        except OSError:
+            log.exception("GLM ASR: failed to read %s", file_path)
+            return None
+        if not data:
+            log.warning("GLM ASR: empty file: %s", file_path)
+            return None
+
+        try:
+            async with httpx.AsyncClient(timeout=_TIMEOUT_S) as client:
+                resp = await client.post(
+                    GLM_ASR_URL,
+                    headers={"Authorization": f"Bearer {self._api_key}"},
+                    files={"file": (p.name, data)},
+                    data={"model": self._model, "stream": "false"},
+                )
+        except Exception:
+            log.exception("GLM ASR: request failed for %s", file_path)
+            return None
+
+        if not resp.is_success:
+            log.error(
+                "GLM ASR: %s returned %s: %s",
+                file_path, resp.status_code, resp.text[:300],
+            )
+            return None
+
+        try:
+            payload = resp.json()
+        except ValueError:
+            log.error("GLM ASR: non-JSON response: %s", resp.text[:300])
+            return None
+
+        # API returns {"text": "...", ...} on success
+        text = (payload.get("text") or "").strip()
+        if not text:
+            log.warning("GLM ASR: empty transcription for %s: %s", file_path, payload)
+            return None
+        log.info("GLM ASR: transcribed %s (%d chars)", file_path, len(text))
+        return text

src/agent_box/channels/qq.py

@@ -34,6 +34,7 @@
 import anyio
 import httpx
 
+from ..asr.glm import GlmASR
 from ..config import settings
 from ..models import IncomingMessage, MessageType, OutgoingMessage
 from .base import BaseChannel
@@ -131,6 +132,11 @@ def __init__(self, send_stream: anyio.abc.ObjectSendStream[IncomingMessage]) ->
         # Latest msg_id per user_id; used as passive-reply anchor
         self._last_msg_id: dict[str, str] = {}
         self._download_dir: Path = settings.config_dir / "channels" / "qq" / "downloads"
+        # Voice transcription; None when GLM API key is not configured (skip feature)
+        self._asr: GlmASR | None = (
+            GlmASR(settings.glm_api_key, settings.glm_asr_model)
+            if settings.glm_api_key else None
+        )
 
     # ── Auth ──────────────────────────────────────────────────────────────────
 
@@ -582,24 +588,33 @@ def _att_label(self, att: dict[str, Any]) -> str:
 
     async def _download_atts(
         self, atts: list[dict[str, Any]]
-    ) -> list[tuple[str, str]]:
-        """Download each attachment. Returns list of (label, local_path)."""
-        results: list[tuple[str, str]] = []
+    ) -> list[tuple[str, str, str | None]]:
+        """Download each attachment. Returns list of (label, local_path, transcription).
+
+        transcription is populated for audio attachments when ASR is configured
+        and succeeds; otherwise None.
+        """
+        results: list[tuple[str, str, str | None]] = []
         for att in atts:
             url = _normalize_url(att.get("url") or "")
             label = self._att_label(att)
+            local = ""
+            transcription: str | None = None
             if url:
-                local = await self._download_image(url, att.get("filename"))
-                results.append((label, local or ""))
-            else:
-                results.append((label, ""))
+                local = await self._download_image(url, att.get("filename")) or ""
+                # Transcribe voice attachments when ASR is available
+                if label == "语音" and local and self._asr is not None:
+                    transcription = await self._asr.transcribe(local)
+            results.append((label, local, transcription))
         return results
 
-    def _atts_text(self, atts_info: list[tuple[str, str]]) -> str:
+    def _atts_text(self, atts_info: list[tuple[str, str, str | None]]) -> str:
         """Build text description lines for all attachments."""
         parts: list[str] = []
-        for label, path in atts_info:
-            if path:
+        for label, path, transcription in atts_info:
+            if label == "语音" and transcription:
+                parts.append(f"用户发送了一段语音，内容: {transcription}")
+            elif path:
                 parts.append(f"用户发送了一个{label}，文件路径: {path}")
             else:
                 parts.append(f"用户发送了一个{label}，文件路径: (下载失败)")

src/agent_box/config.py

@@ -20,6 +20,10 @@ class Settings(BaseSettings):
     qqbot_app_id: str = ""
     qqbot_client_secret: str = ""
 
+    # GLM (ZhipuAI) ASR — voice-to-text for audio attachments. Empty skips transcription.
+    glm_api_key: str = ""
+    glm_asr_model: str = "glm-asr-2512"
+
     # Config & workspace directories
     config_dir: Path = Path.home() / ".agent-box"
     workspace_dir: Path = Path.home() / ".agent-box" / "workspace"

tests/test_asr.py

@@ -0,0 +1,109 @@
+"""Tests for agent_box.asr.glm."""
+
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import pytest
+
+from agent_box.asr.glm import GlmASR, GLM_ASR_URL
+
+
+def _make_response(status: int, json_body: dict | None = None, text: str = "") -> httpx.Response:
+    kwargs: dict = {"request": httpx.Request("POST", GLM_ASR_URL)}
+    if json_body is not None:
+        kwargs["json"] = json_body
+    else:
+        kwargs["text"] = text
+    return httpx.Response(status, **kwargs)
+
+
+@pytest.mark.anyio
+async def test_transcribe_success(tmp_path):
+    audio = tmp_path / "voice.wav"
+    audio.write_bytes(b"fake-audio-bytes")
+
+    asr = GlmASR(api_key="key-123", model="glm-asr-2512")
+
+    with patch("agent_box.asr.glm.httpx.AsyncClient") as MockClient:
+        instance = MockClient.return_value
+        instance.__aenter__ = AsyncMock(return_value=instance)
+        instance.__aexit__ = AsyncMock(return_value=None)
+        instance.post = AsyncMock(
+            return_value=_make_response(200, {"text": "你好世界"})
+        )
+        result = await asr.transcribe(str(audio))
+
+    assert result == "你好世界"
+    instance.post.assert_awaited_once()
+    kwargs = instance.post.await_args.kwargs
+    assert kwargs["headers"]["Authorization"] == "Bearer key-123"
+    assert kwargs["data"]["model"] == "glm-asr-2512"
+    assert kwargs["data"]["stream"] == "false"
+    assert "file" in kwargs["files"]
+
+
+@pytest.mark.anyio
+async def test_transcribe_http_error_returns_none(tmp_path):
+    audio = tmp_path / "voice.mp3"
+    audio.write_bytes(b"audio")
+
+    asr = GlmASR(api_key="key", model="glm-asr-2512")
+
+    with patch("agent_box.asr.glm.httpx.AsyncClient") as MockClient:
+        instance = MockClient.return_value
+        instance.__aenter__ = AsyncMock(return_value=instance)
+        instance.__aexit__ = AsyncMock(return_value=None)
+        instance.post = AsyncMock(return_value=_make_response(400, text="bad request"))
+        result = await asr.transcribe(str(audio))
+
+    assert result is None
+
+
+@pytest.mark.anyio
+async def test_transcribe_empty_text_returns_none(tmp_path):
+    audio = tmp_path / "voice.wav"
+    audio.write_bytes(b"audio")
+
+    asr = GlmASR(api_key="key")
+
+    with patch("agent_box.asr.glm.httpx.AsyncClient") as MockClient:
+        instance = MockClient.return_value
+        instance.__aenter__ = AsyncMock(return_value=instance)
+        instance.__aexit__ = AsyncMock(return_value=None)
+        instance.post = AsyncMock(return_value=_make_response(200, {"text": "  "}))
+        result = await asr.transcribe(str(audio))
+
+    assert result is None
+
+
+@pytest.mark.anyio
+async def test_transcribe_network_error_returns_none(tmp_path):
+    audio = tmp_path / "voice.wav"
+    audio.write_bytes(b"audio")
+
+    asr = GlmASR(api_key="key")
+
+    with patch("agent_box.asr.glm.httpx.AsyncClient") as MockClient:
+        instance = MockClient.return_value
+        instance.__aenter__ = AsyncMock(return_value=instance)
+        instance.__aexit__ = AsyncMock(return_value=None)
+        instance.post = AsyncMock(side_effect=httpx.ConnectError("nope"))
+        result = await asr.transcribe(str(audio))
+
+    assert result is None
+
+
+@pytest.mark.anyio
+async def test_transcribe_missing_file_returns_none():
+    asr = GlmASR(api_key="key")
+    result = await asr.transcribe("/nonexistent/voice.wav")
+    assert result is None
+
+
+@pytest.mark.anyio
+async def test_transcribe_empty_file_returns_none(tmp_path):
+    audio = tmp_path / "empty.wav"
+    audio.write_bytes(b"")
+    asr = GlmASR(api_key="key")
+    result = await asr.transcribe(str(audio))
+    assert result is None