커스텀 제공자 구현 가이드

from clawops.agent.pipeline import STT, LLM, TTS, SpeechEvent

@dataclass(frozen=True, slots=True)
class SpeechEvent:
    type: Literal["interim", "final"]
    transcript: str

class STT(Protocol):
    async def transcribe(self, audio_stream: AsyncIterator[bytes]) -> AsyncIterator[SpeechEvent]:
        ...

import asyncio
from typing import AsyncIterator
from clawops.agent.pipeline import SpeechEvent

class WhisperSTT:
    def __init__(self, model: str = "whisper-1"):
        self._model = model

    async def transcribe(self, audio_stream: AsyncIterator[bytes]) -> AsyncIterator[SpeechEvent]:
        buffer = bytearray()
        CHUNK_SIZE = 16000 * 2 * 2  # 2초 분량 (16kHz, 16bit)

        async for chunk in audio_stream:
            buffer.extend(chunk)

            if len(buffer) >= CHUNK_SIZE:
                # 버퍼가 충분하면 인식 시도
                text = await self._recognize(bytes(buffer))
                buffer.clear()

                if text:
                    # VAD 기반 barge-in이 없으므로 interim+final 동시 발송
                    yield SpeechEvent(type="interim", transcript=text)
                    yield SpeechEvent(type="final", transcript=text)

        # 남은 버퍼 처리
        if buffer:
            text = await self._recognize(bytes(buffer))
            if text:
                yield SpeechEvent(type="final", transcript=text)

    async def _recognize(self, pcm16: bytes) -> str:
        import openai
        # PCM16을 WAV로 변환 후 Whisper API 호출
        wav = self._pcm16_to_wav(pcm16)
        client = openai.AsyncOpenAI()
        result = await client.audio.transcriptions.create(
            model=self._model,
            file=("audio.wav", wav, "audio/wav"),
        )
        await client.close()
        return result.text

    def _pcm16_to_wav(self, pcm: bytes) -> bytes:
        import struct
        header = struct.pack(
            "<4sI4s4sIHHIIHH4sI",
            b"RIFF", 36 + len(pcm), b"WAVE",
            b"fmt ", 16, 1, 1, 16000, 32000, 2, 16,
            b"data", len(pcm),
        )
        return header + pcm

# VAD 이벤트가 있는 STT 서비스 예시
async def transcribe(self, audio_stream):
    # 음성 시작 → 빈 interim (barge-in 트리거)
    yield SpeechEvent(type="interim", transcript="")

    # 부분 인식 결과는 무시 (이미 interim 발송)

    # 발화 완료 → final
    yield SpeechEvent(type="final", transcript="안녕하세요")

class LLM(Protocol):
    async def generate(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
    ) -> AsyncIterator[str]:
        ...

import json
from typing import Any, AsyncIterator

class MyLLM:
    """커스텀 LLM — LLM Protocol만 구현하면 됩니다."""

    async def generate(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
    ) -> AsyncIterator[str]:
        # 1. messages에서 system, user, assistant, tool 역할을 읽어
        #    사용하는 LLM API에 맞게 변환합니다.
        # 2. tools가 있으면 OpenAI Chat Completions 포맷을 해당 API 포맷으로 변환합니다.
        # 3. 텍스트 토큰을 스트리밍으로 yield합니다.
        # 4. Tool call이 필요하면 아래 JSON 포맷으로 yield합니다:
        #    {"type": "tool_calls", "tool_calls": [{"id": "...", "function": {"name": "...", "arguments": "..."}}]}
        yield "안녕하세요!"

class TTS(Protocol):
    async def synthesize(self, text_stream: AsyncIterator[str]) -> AsyncIterator[bytes]:
        ...

from typing import AsyncIterator

class GoogleTTS:
    def __init__(self, voice: str = "ko-KR-Neural2-A"):
        self._voice = voice

    @property
    def sample_rate(self) -> int:
        return 24000

    async def synthesize(self, text_stream: AsyncIterator[str]) -> AsyncIterator[bytes]:
        from google.cloud import texttospeech_v1 as tts

        client = tts.TextToSpeechAsyncClient()

        async for text in text_stream:
            if not text.strip():
                continue

            response = await client.synthesize_speech(
                input=tts.SynthesisInput(text=text),
                voice=tts.VoiceSelectionParams(
                    language_code="ko-KR",
                    name=self._voice,
                ),
                audio_config=tts.AudioConfig(
                    audio_encoding=tts.AudioEncoding.LINEAR16,
                    sample_rate_hertz=self.sample_rate,
                ),
            )

            # WAV 헤더(44바이트) 제거 → raw PCM16
            yield response.audio_content[44:]

@property
def sample_rate(self) -> int:
    return 24000  # 24kHz PCM16을 출력하는 경우

from clawops.agent import ClawOpsAgent
from clawops.agent.pipeline import PipelineSession, DeepgramSTT

# Deepgram STT + Anthropic Claude LLM + Google TTS
from clawops.agent.pipeline import AnthropicLLM

agent = ClawOpsAgent(
    from_="07012341234",
    session=PipelineSession(
        system_prompt="친절한 상담원입니다.",
        stt=DeepgramSTT(),
        llm=AnthropicLLM(model="claude-sonnet-4-6"),
        tts=GoogleTTS(voice="ko-KR-Neural2-A"),
    ),
)