将 reSpeaker Clip 集成到你的服务中

本指南展示如何将 reSpeaker Clip 作为受管音频源添加到现有的 Python 服务中。它聚焦于集成边界：一个设备适配器、一个长连接、小而清晰的 API 表面，以及明确的状态归属。

reSpeaker Clip 是一个输入源。 ClipClient 负责设备通信，ClipService 将其适配到你的应用中，而你的服务继续拥有处理和业务逻辑的所有权。

关于传输配置、完整的 ClipClient API、AT 命令以及文件传输协议，请参阅 reSpeaker Clip Basic SDK Guide。本页不会重新解释 BLE/GATT、帧格式、CRC 处理或 CLI 工具。

集成方法

将设备隐藏在一个窄适配器之后，并将下载的录音发送到你的服务已经在使用的处理入口点：

Existing service with reSpeaker Clip as a new audio source

为每个物理设备创建一个 ClipService。
在应用启动时连接，在关闭时断开。
只暴露你的服务所需的录音操作。
将完成的音频文件传递给现有的处理流水线。

开始之前

本指南假设你已经可以在自己的机器上与 Clip 通信。在继续之前，请确保你可以：

安装 reSpeaker Clip Python SDK 并连接到一个 Clip；
启动和停止一次录音；
列出已完成的会话；
成功下载一段录音。

如果上述任一项失败，请先暂停，完成 reSpeaker Clip Basic SDK Guide。Basic SDK Guide 是传输配置、设备命令和文件传输细节的权威来源；本指南通过链接引用它，而不是重复内容。

选择合适的 reSpeaker Clip 指南

reSpeaker Clip 文档按开发层次组织。先从 Getting Started 开始进行产品设置和常规工作流。主机应用需要控制设备或下载录音时，使用 Basic SDK。当一个长时间运行的服务需要拥有设备连接并暴露 API 时，继续阅读 Service Integration。只有在你需要更改设备端行为、协议或音频处理时，才使用 Firmware SDK。

Getting Started

产品概览、硬件设置和首次使用工作流。

Basic SDK

在不更改固件的情况下控制设备、管理录音并下载音频。

Service Integration

通过适配器和 API 路由，将 reSpeaker Clip 添加到一个长时间运行的 Python 服务中。

Firmware SDK

构建、烧录、调试并自定义设备端行为。

将 reSpeaker Clip 作为音频源添加到你的现有服务中

按照以下步骤，将设备适配器连接到现有的 FastAPI 服务。

步骤 1：准备流水线

假设你的服务已经有一个接收音频文件的函数：

async def process_audio(path: Path) -> dict:
    # your existing processing pipeline
    ...

现有的上传路由已经调用了该函数：

@app.post("/audio")
async def upload_audio(file: UploadFile):
    path = await save_upload(file)
    return await process_audio(path)

Clip 集成应生成一个普通的本地音频文件，并调用相同的 process_audio() 函数。处理流程应与音频采集设备保持独立。

将该函数封装到 Clip 数据摄取路由所使用的服务对象中：

# audio_service.py
from pathlib import Path


class AudioService:
    def __init__(self, recordings_dir: Path):
        self.recordings_dir = recordings_dir
        self.recordings_dir.mkdir(parents=True, exist_ok=True)

    async def process(self, path: Path) -> dict:
        return await process_audio(path)

步骤 2：构建适配器

仅封装此集成所需的操作：连接、断开连接、开始、停止和下载。

# clip_service.py
from __future__ import annotations
from pathlib import Path

from clip import ClipClient, ClipError


class ClipUnavailable(Exception):
    """The Clip device is not connected or could not perform an operation."""


class ClipStateError(Exception):
    """The device is in the wrong state for this operation."""


class ClipService:
    """Application-level adapter over a single long-lived ClipClient."""


        self._client = ClipClient(transport, command_timeout=command_timeout)
        self._connected = False
        self._active_session: str | None = None

    async def connect(self) -> None:
        await self._client.connect()
        self._connected = True


        if self._connected:
            await self._client.disconnect()
            self._connected = False

    async def start_recording(self, mode: str | None = None) -> str:
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        if self._active_session is not None:
            raise ClipStateError("a recording is already in progress")
        try:
            session_id = await self._client.start_recording(mode)
        except ClipError as exc:
            raise ClipUnavailable("Clip could not start recording") from exc

            raise ClipStateError("Clip did not return a session id")
        self._active_session = session_id
        return session_id

    async def stop_recording(self) -> str | None:
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        if self._active_session is None:
            raise ClipStateError("no recording in progress")
        session_id = self._active_session
        try:
            await self._client.stop_recording()

            raise ClipUnavailable("Clip could not stop recording") from exc
        self._active_session = None
        return session_id

    async def download_session(self, session_id: str, destination: str | Path):
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        try:
            return await self._client.download_session(session_id, destination)
        except ClipError as exc:
            raise ClipUnavailable("Clip could not download the recording") from exc

步骤 3：绑定生命周期

在 FastAPI 启动期间创建适配器，将其存储在应用状态中，并在关闭时断开连接。

Service lifecycle for reSpeaker Clip integration

将连接绑定到 FastAPI 的 lifespan，这样它会在启动时打开一次，在停止时关闭一次：

# main.py
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI

from clip import BleTransport  # or UdpTransport after the Wi-Fi AP is on
from clip_service import ClipService
from audio_service import AudioService


@asynccontextmanager
async def lifespan(app: FastAPI):
    transport = BleTransport(name="Clip")
    clip = ClipService(transport)
    await clip.connect()
    app.state.clip = clip
    app.state.audio = AudioService(recordings_dir=Path("recordings"))
    try:
        yield
    finally:
        await clip.disconnect()


app = FastAPI(lifespan=lifespan)

步骤 4：添加路由

为开始、停止以及摄取已完成录音创建端点：

POST /clip/recordings/start
POST /clip/recordings/stop
POST /clip/sessions/{session_id}/ingest

# routes.py
from pathlib import Path

from fastapi import APIRouter, HTTPException, Request

from audio_service import AudioService
from clip_service import ClipService, ClipStateError, ClipUnavailable

router = APIRouter(prefix="/clip")


@router.post("/recordings/start")
async def start_recording(request: Request, mode: str | None = None):
    clip: ClipService = request.app.state.clip
    try:
        session_id = await clip.start_recording(mode)
    except ClipStateError as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    except ClipUnavailable as exc:
        raise HTTPException(status_code=503, detail=str(exc))
    return {"session_id": session_id}


@router.post("/recordings/stop")
async def stop_recording(request: Request):
    clip: ClipService = request.app.state.clip
    try:
        session_id = await clip.stop_recording()
    except ClipStateError as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    except ClipUnavailable as exc:
        raise HTTPException(status_code=503, detail=str(exc))
    return {"session_id": session_id, "status": "stopped"}


@router.post("/sessions/{session_id}/ingest")
async def ingest_session(session_id: str, request: Request):
    clip: ClipService = request.app.state.clip
    audio: AudioService = request.app.state.audio
    try:
        result = await clip.download_session(session_id, audio.recordings_dir)
    except ClipUnavailable as exc:
        raise HTTPException(status_code=502, detail=str(exc))
    ingested = [await audio.process(Path(file.path)) for file in result.files]
    return {"session_id": session_id, "files": ingested}

步骤 5：注册路由器

在 main.py 中导入路由器，并在创建 FastAPI 应用后注册它：

from routes import router

app = FastAPI(lifespan=lifespan)
app.include_router(router)

现在，该集成由应用包中的 main.py、routes.py、clip_service.py 和 audio_service.py 组成。

步骤 6：验证 API

启动服务，然后按顺序调用这些路由：

curl -X POST "http://localhost:8000/clip/recordings/start"
curl -X POST "http://localhost:8000/clip/recordings/stop"
curl -X POST "http://localhost:8000/clip/sessions/<session_id>/ingest"

开始和停止的响应应包含相同的 session_id。摄取的响应应包含由你现有的 AudioService 流水线返回的文件。

长时间运行服务的生产注意事项

当集成以服务形式运行时，请保持连接、错误和工作流状态是显式的。

适配器边界

将所有 ClipClient 调用保持在 ClipService 内部。适配器拥有连接访问、录音命令、下载以及 SDK 错误转换；它不负责处理、存储或其他业务逻辑。

适配器之上的代码应使用诸如 session_id 和应用异常等应用概念，而不是传输、AT 命令或文件传输等细节。

客户端复用

将一个物理 Clip 映射到一个长生命周期的 ClipClient 和一个 ClipService。由于固件没有请求 ID，ClipClient 会串行化命令。对于多个设备，在 ClipManager 中按设备 ID 为 ClipService 实例建立键。

连接生命周期

一个常见错误是在请求处理器内部打开设备：

# WRONG — do not do this in a service
@app.post("/clip/start")
async def start():
    async with ClipClient(BleTransport(name="Clip")) as clip:
        await clip.start_recording()

这会在每次调用时重新连接，并在请求之间丢失活动会话。应用的生命周期拥有连接；请求只是复用它。

错误边界

在 ClipService 内部将 SDK 异常转换为应用异常，然后在路由边界将它们映射到 HTTP 响应。

一个简单的映射示例（用于说明，而非 SDK 强制要求）：

应用条件	HTTP 状态码
Clip 不可用	`503`
设备状态无效	`409`
设备命令失败	`502`
客户端请求错误	`400`

具体状态码取决于应用。适配器之外的代码应依赖 ClipUnavailable 和 ClipStateError，而不是 SDK 异常。

重新连接

在发生设备错误时，将设备标记为不可用，并重新连接或重新创建传输层。命令超时需要在下一条命令之前重新连接，因为传输层无法再关联响应。将故障暴露为 ClipUnavailable，然后从后台任务或下一次请求中重新连接。

幂等性

HTTP 重试、工作进程重试和重启可能会多次提交相同的 session_id。在处理之前检查持久化状态。

在数据库中跟踪每个会话：

`session_id`	`downloaded`	`processed`	`result_id`	`created_at`
唯一设备会话	下载状态	流水线状态	结果引用	首次摄取时间

async def ingest_clip_session(session_id: str):
    if repository.is_processed(session_id):
        return repository.get_result(session_id)

    result = await clip_service.download_session(session_id, Path("recordings"))
    for file in result.files:
        await process_audio(Path(file.path))

    repository.mark_processed(session_id)
    return repository.get_result(session_id)

这可以防止重复处理和下游副作用。

状态归属

将设备状态与工作流状态分开：

状态	所有者	示例值
Clip 状态	设备集成层（`ClipService`）	已连接、录音中、会话可用、可下载
工作流状态	应用/作业层	摄取中、处理中、已完成、失败

不要将这两个状态机折叠到一个字段中。会话在其应用工作流仍在运行时就可以被下载。

后续步骤

在适配器工作之后，继续将设备集成与任何特定于应用的处理分离。相关页面：

reSpeaker Clip Basic SDK Guide — 设备协议、ClipClient API、BLE/UDP 传输、文件传输
Getting Started with the reSpeaker Clip Firmware SDK — 当你需要修改设备端固件而不是集成 SDK 时

技术支持与产品讨论

感谢你选择我们的产品！我们将为你提供多种支持，以确保你在使用我们产品时的体验尽可能顺畅。我们提供多种沟通渠道，以满足不同的偏好和需求。

集成方法​

开始之前​

选择合适的 reSpeaker Clip 指南​

Getting Started

Basic SDK

Service Integration

Firmware SDK

将 reSpeaker Clip 作为音频源添加到你的现有服务中​

步骤 1：准备流水线​

步骤 2：构建适配器​

步骤 3：绑定生命周期​

步骤 4：添加路由​

步骤 5：注册路由器​

步骤 6：验证 API​

长时间运行服务的生产注意事项​

适配器边界​

客户端复用​

连接生命周期​

错误边界​

重新连接​

幂等性​

状态归属​

后续步骤​

技术支持与产品讨论​