[SGLang] 음성 인식 & ASR 통합: Whisper, Qwen3-ASR 어댑터 구현

들어가며

LLM 서빙 엔진이 텍스트 생성을 넘어 멀티모달 영역으로 확장되고 있다. SGLang은 OpenAI 호환 /v1/audio/transcriptions endpoint를 통해 음성 인식(ASR)을 LLM 서버 안에 통합했다. Whisper와 Qwen3-ASR 같은 서로 다른 ASR 모델을 어댑터 패턴으로 추상화하고, chunk 기반 스트리밍 ASR까지 지원한다.

이 글에서는 SGLang의 ASR 파이프라인 구조, 어댑터 등록 메커니즘, 스트리밍 처리 방식을 소스 코드와 함께 분석한다.

ASR 파이프라인 구조도

SGLang의 transcription 요청은 다음과 같은 경로로 처리된다.

Audio File (wav/mp3/flac)
        |
        v
+----------------------------+
| /v1/audio/transcriptions   |  <-- OpenAI 호환 API endpoint
+----------------------------+
        |
        v
+----------------------------+
| OpenAIServingTranscription |  <-- 요청 파싱, 오디오 duration 계산
+----------------------------+
        |
        v
+----------------------------+
| resolve_adapter()          |  <-- HF architectures 문자열 매칭
+----------------------------+
        |                |
        v                v
+------------+   +----------------+
| Whisper    |   | Qwen3-ASR      |
| Adapter    |   | Adapter        |
+------------+   +----------------+
        |                |
        v                v
+----------------------------+
| TokenizerManager           |  <-- 모델 추론 실행
| .generate_request()        |
+----------------------------+
        |
        v
  Transcription Text Output

핵심은 resolve_adapter()가 모델의 HF config에서 architectures 필드를 읽어 적절한 어댑터를 자동 선택한다는 점이다.

핵심 코드 분석

어댑터 레지스트리: 자동 등록 패턴

python/sglang/srt/entrypoints/openai/transcription_adapters/base.py에 정의된 레지스트리 패턴이 전체 구조의 기반이다.

_ADAPTER_REGISTRY: dict[str, type[TranscriptionAdapter]] = {}
_DEFAULT_ADAPTER_KEY = "Whisper"

def register_transcription_adapter(key: str) -> callable:
    def decorator(cls: type[TranscriptionAdapter]) -> type[TranscriptionAdapter]:
        _ADAPTER_REGISTRY[key] = cls
        return cls
    return decorator

def resolve_adapter(architectures: List[str]) -> TranscriptionAdapter:
    for arch in architectures or []:
        for key, adapter_cls in _ADAPTER_REGISTRY.items():
            if key in arch:
                return adapter_cls()
    default_cls = _ADAPTER_REGISTRY.get(_DEFAULT_ADAPTER_KEY)
    if default_cls is None:
        raise RuntimeError(
            "No transcription adapters registered. "
            "Make sure 'transcription_adapters' package is importable."
        )
    return default_cls()

key가 architecture 문자열의 부분 문자열로 매칭된다. "Whisper"는 "WhisperForConditionalGeneration"에, "Qwen3ASR"는 "Qwen3ASRForConditionalGeneration"에 매칭되는 방식이다. 새 ASR 모델을 추가하려면 TranscriptionAdapter를 상속하고 @register_transcription_adapter("ModelKey") 데코레이터를 붙이면 된다.

Whisper 어댑터: timestamp 파싱

python/sglang/srt/entrypoints/openai/transcription_adapters/whisper.py의 Whisper 어댑터는 timestamp token 기반 segment 파싱을 구현한다.

@register_transcription_adapter("Whisper")
class WhisperAdapter(TranscriptionAdapter):
    TIMESTAMP_BASE_TOKEN_ID = 50365  # <|0.00|>
    TIMESTAMP_BASE_OFFSET = 0.02    # each token step = 0.02 s

    def build_sampling_params(self, request: TranscriptionRequest) -> dict:
        params: dict = {
            "temperature": request.temperature,
            "max_new_tokens": 448,
            "language": request.language,
        }
        if request.timestamp_granularities:
            params["timestamp_granularities"] = request.timestamp_granularities
        return params

Whisper의 특수 timestamp token은 ID 50365부터 시작하며, 각 token이 0.02초 간격을 나타낸다. _parse_segments 메서드가 output token ID 스트림에서 timestamp token을 감지해 segment 경계를 추출한다.

for token_id in output_ids:
    if token_id >= ts_base:
        timestamp = (token_id - ts_base) * ts_step
        if current_text_tokens:
            seg_text = tokenizer.decode(
                current_text_tokens, skip_special_tokens=True
            ).strip()
            if seg_text:
                segments.append(
                    TranscriptionSegment(
                        id=seg_id,
                        start=round(current_start, 2),
                        end=round(timestamp, 2),
                        text=seg_text,
                    )
                )

이 파싱 로직 덕분에 response_format=verbose_json을 요청하면 segment별 시작/종료 시간이 포함된 응답을 받을 수 있다.

Qwen3-ASR 어댑터: chunk 기반 스트리밍

python/sglang/srt/entrypoints/openai/transcription_adapters/qwen3_asr.py의 Qwen3-ASR 어댑터는 Whisper와 근본적으로 다른 접근을 취한다.

@register_transcription_adapter("Qwen3ASR")
class Qwen3ASRAdapter(TranscriptionAdapter):
    ASR_TEXT_TAG = "<asr_text>"

    @property
    def supports_chunked_streaming(self) -> bool:
        return True

    @property
    def chunked_streaming_config(self) -> dict:
        return {
            "chunk_size_sec": 2.0,
            "unfixed_chunk_num": 2,
            "unfixed_token_num": 5,
        }

    def postprocess_text(self, text: str) -> str:
        if self.ASR_TEXT_TAG in text:
            return text.split(self.ASR_TEXT_TAG, 1)[-1]
        return text

Qwen3-ASR는 <asr_text> 태그 뒤에 transcription 결과를 출력하는 포맷을 사용한다. postprocess_text가 이 prefix를 제거해 깨끗한 텍스트를 반환한다. 또한 temperature 0.0을 0.01로 보정하는데, Qwen3-ASR 논문의 권장 설정이다.

스트리밍 ASR: prefix rollback 알고리즘

python/sglang/srt/entrypoints/openai/streaming_asr.py의 StreamingASRState는 chunk 기반 스트리밍의 핵심 상태 관리를 담당한다.

@dataclass
class StreamingASRState:
    chunk_size_sec: float
    unfixed_chunk_num: int
    unfixed_token_num: int
    confirmed_text: str = ""
    full_transcript: str = ""
    chunk_index: int = 0

    def update(self, new_transcript: str) -> str:
        old_confirmed = self.confirmed_text
        words = new_transcript.split()
        if len(words) > self.unfixed_token_num:
            self.confirmed_text = " ".join(words[: -self.unfixed_token_num])
        else:
            self.confirmed_text = ""
        self.full_transcript = new_transcript
        self.chunk_index += 1
        if self.confirmed_text.startswith(old_confirmed):
            return self.confirmed_text[len(old_confirmed) :].strip()
        # Model revised earlier text — use word level common prefix
        old_words = old_confirmed.split()
        new_words = self.confirmed_text.split()
        common_count = 0
        for ow, nw in zip(old_words, new_words):
            if ow != nw:
                break
            common_count += 1
        return " ".join(new_words[common_count:])

이 알고리즘의 동작 원리는 다음과 같다:

1. 오디오를 2초 단위 chunk로 분할한다 (split_audio_chunks). 각 chunk는 처음부터 해당 지점까지의 누적 오디오다.
2. 매 chunk마다 모델이 전체 오디오의 transcript를 새로 생성한다.
3. 마지막 unfixed_token_num개 단어는 아직 확정되지 않은 것으로 간주하고 rollback 대상으로 남긴다.
4. 이전에 확정된 텍스트(confirmed_text)와 새 transcript의 공통 prefix를 찾아 새로 확정된 부분만 delta로 내보낸다.
5. 모델이 이전 텍스트를 수정한 경우(punctuation 변경 등), word 단위 공통 prefix를 사용해 중복 전송을 방지한다.

단, 소스 코드의 주석이 명시하듯 str.split() 기반 rollback은 CJK 언어에서는 제대로 동작하지 않는다. 공백 없이 이어지는 한국어/중국어/일본어 텍스트에서는 token 단위 rollback이 필요하다.

어댑터 비교: Whisper vs Qwen3-ASR

항목	Whisper	Qwen3-ASR
등록 키	"Whisper"	"Qwen3ASR"
max_new_tokens	448	256
temperature 기본값	요청값 그대로	0.0 → 0.01 보정
timestamp 지원	timestamp token 파싱 (segment 단위)	미지원 (ForcedAligner 필요)
chunk 기반 스트리밍	미지원	지원 (2초 chunk)
스트리밍 방식	token 단위 SSE	chunk 기반 prefix rollback
출력 후처리	없음 (clean output)	<asr_text> 태그 제거
verbose_json	segment + 시작/종료 시간 포함	segment 없음 (빈 배열)
prompt template	없음	chat 형식 (<|im_start|>user\n<audio>)

두 어댑터의 근본적 차이는 스트리밍 전략에 있다. Whisper는 모델이 token을 하나씩 생성하는 대로 SSE로 전송하는 token-level streaming을 사용한다. 반면 Qwen3-ASR는 오디오를 chunk로 나누어 각 chunk마다 독립 요청을 보내고, prefix rollback으로 확정된 텍스트만 전송하는 chunk-level streaming을 사용한다.

기존 ASR 서비스 대비 통합의 이점

독립 ASR 서비스(Faster Whisper 서버, WhisperX 등)와 비교했을 때, LLM 서빙 엔진 안에 ASR을 통합하면 다음과 같은 이점이 있다.

인프라 단순화: ASR 전용 서버를 별도로 운영할 필요 없이 하나의 SGLang 서버에서 텍스트 생성과 음성 인식을 모두 처리한다. GPU 리소스도 단일 프로세스에서 관리된다.

API 호환성: OpenAI의 /v1/audio/transcriptions API와 동일한 인터페이스를 제공하므로, 기존 OpenAI SDK 기반 코드를 base URL만 바꿔서 사용할 수 있다.

확장성: 어댑터 패턴 덕분에 새 ASR 모델 추가가 간단하다. TranscriptionAdapter를 상속하고 build_sampling_params, build_verbose_response 두 메서드만 구현하면 된다. 모델별 전처리/후처리 로직은 어댑터 안에 캡슐화된다.

배치 처리 활용: SGLang의 continuous batching, RadixAttention 같은 기존 최적화가 ASR 요청에도 자동으로 적용된다. 독립 ASR 서비스는 이런 LLM 서빙 최적화를 별도로 구현해야 한다.

다만 한계도 있다. Qwen3-ASR의 스트리밍은 chunk마다 독립 요청을 보내므로 encoder 연산이 중복된다. 소스 코드의 TODO에도 "Encoder window caching across chunks"와 "Cross-chunk KV cache reuse"가 명시되어 있어, 이 부분은 향후 최적화 대상이다.

관련 포스트

• SGLang 프로젝트 분석 시리즈의 다른 글은 SGLang 카테고리에서 확인할 수 있다.

참고

• SGLang GitHub Repository
• Qwen3-ASR 논문 (arXiv:2601.21337)
• OpenAI Audio Transcription API Reference
• python/sglang/srt/entrypoints/openai/serving_transcription.py
• python/sglang/srt/entrypoints/openai/streaming_asr.py
• python/sglang/srt/entrypoints/openai/transcription_adapters/