Pipelines and workflows

Pipelines and workflows (파이프라인과 워크플로)

VoicePipeline은 에이전틱 워크플로를 음성 앱으로 쉽게 바꿔주는 클래스예요. 실행할 워크플로를 넘기면 파이프라인이 입력 오디오 전사, 오디오 끝 감지, 적절한 시점에 워크플로 호출, 워크플로 출력을 다시 오디오로 바꾸는 일을 처리해요.

출처: 문서

본문

graph LR
    %% Input
    A["🎤 Audio Input"]

    %% Voice Pipeline
    subgraph Voice_Pipeline [Voice Pipeline]
        direction TB
        B["Transcribe (speech-to-text)"]
        C["Your Code"]:::highlight
        D["Text-to-speech"]
        B --> C --> D
    end

    %% Output
    E["🎧 Audio Output"]

    %% Flow
    A --> Voice_Pipeline
    Voice_Pipeline --> E

    %% Custom styling
    classDef highlight fill:#ffcc66,stroke:#333,stroke-width:1px,font-weight:700;

파이프라인 구성

파이프라인을 만들 때 몇 가지를 설정할 수 있어요.

  • workflow — 새 오디오가 전사될 때마다 실행되는 코드
  • 사용되는 speech-to-text와 text-to-speech 모델
  • config — 아래 같은 것들을 구성하게 해주는 설정:
    • 모델 이름을 모델로 매핑해줄 수 있는 모델 프로바이더
    • Tracing — tracing 비활성화, 오디오 파일 업로드 여부, 워크플로 이름, trace ID 등
    • TTS·STT 모델 설정 — prompt, language, 사용 데이터 타입 등

단일 에이전트 워크플로에 앱 컨텍스트 전달

음성 에이전트·그 도구·lifecycle 훅이 앱 상태나 의존성을 필요로 하면 SingleAgentVoiceWorkflow에 context를 전달하세요.

from dataclasses import dataclass

from agents import Agent
from agents.voice import SingleAgentVoiceWorkflow, VoicePipeline

@dataclass
class VoiceContext:
    user_id: str

agent = Agent[VoiceContext](name="Voice assistant")
workflow = SingleAgentVoiceWorkflow(
    agent,
    context=VoiceContext(user_id="user-123"),
)
pipeline = VoicePipeline(workflow=workflow)

워크플로는 시작하는 모든 에이전트 실행(나중 전사 턴 포함)에 같은 컨텍스트 객체를 전달해요. 도구와 lifecycle 훅은 RunContextWrapper.context로 그것을 받아요. 컨텍스트는 애플리케이션 로컬로 유지되고 모델로 보내지지 않아요. 타이핑·lifecycle 안내는 Context management를 참고하세요.

OpenAI 음성 모델 구성

VoicePipelineConfig를 통해 STTModelSettings와 TTSModelSettings를 넘겨 기본 OpenAI 음성 모델을 구성하세요.

from agents.voice import STTModelSettings, TTSModelSettings, VoicePipeline, VoicePipelineConfig

config = VoicePipelineConfig(
    stt_settings=STTModelSettings(
        language="en",
        prompt="A customer support call about product AC-42.",
    ),
    tts_settings=TTSModelSettings(
        voice="marin",
    ),
)
pipeline = VoicePipeline(workflow=workflow, config=config)

완전한 오디오 입력의 경우 STTModelSettings.language와 prompt가 전사 요청에 전달돼요. StreamedAudioInput의 경우, WebSocket 세션이 구성될 때 OpenAI 전사 세션도 language, prompt, 그리고 스트리밍 전용 languages·keywords 설정을 받아요. gpt-transcribe와 gpt-live-transcribe는 languages(예상 입력 언어 목록)와 keywords(오디오에 나타날 수 있는 리터럴 용어 목록)를 사용해요. languages가 설정되면 language보다 우선해요. 그 외에는 이 두 모델이 단일 SDK language 값을 한 요소 languages 목록으로 받아요. 다른 전사 모델은 계속 단수 language 필드를 받아요. API가 지원하는 언어 코드를 쓰고, prompt로는 전사 작업을 다시 말하기보다 녹음이나 그 설정을 설명하세요. keywords는 전사 힌트이지 필수 출력이 아니에요. 자세한 내용은 OpenAI transcription context 가이드를 참고하세요.

예상 언어 사이를 전환할 수 있는 스트리밍 전사에는 스트리밍 전용 필드를 명시적으로 구성하세요.

config = VoicePipelineConfig(
    stt_settings=STTModelSettings(
        languages=["en", "es"],
        keywords=["AC-42", "Agents SDK"],
        prompt="A customer support call about product AC-42.",
    ),
)

TTSModelSettings.dtype은 네이티브 int16이나 float32로 해석되는 NumPy dtype 철자(np.int16, np.float32, "int16", "float32", "f4" 같은 별칭 포함)를 받아요. 다른 dtype과 비네이티브 바이트 순서는 파이프라인이 스트리밍 TTS 오디오를 변환할 때 UserError를 발생시켜요.

지원되는 내장 TTSModelSettings.voice 값은 alloy, ash, ballad, coral, echo, fable, onyx, nova, sage, shimmer, verse, marin, cedar예요. 음성 가용성은 선택한 텍스트-음성 모델에 따라 달라져요. 현재 모델별 가용성은 OpenAI voice options를 참고하세요. OpenAI 커스텀 음성 접근이 있는 조직은 대신 커스텀 음성 ID를 넘길 수 있어요.

config = VoicePipelineConfig(
    tts_settings=TTSModelSettings(
        voice={"id": "voice_123abc"},
    ),
)

커스텀 음성은 자격이 있는 고객으로 제한되고, 사용 전에 OpenAI API로 만들어야 해요. 접근·동의·생성 요구사항은 OpenAI custom voices 가이드를 참고하세요.

OpenAIVoiceModelProvider는 비스트리밍 전사 요청·TTS 요청·스트리밍 STT 연결에 구성된 AsyncOpenAI 클라이언트를 사용해요. 스트리밍 STT WebSocket 연결은 그 클라이언트에서 엔드포인트·인증·기본 헤더·기본 쿼리 파라미터를 파생해요. 프로바이더 소유권·우선순위 규칙은 API keys and clients를 참고하세요.

파이프라인 실행

run() 메서드로 파이프라인을 실행할 수 있는데, 오디오 입력을 두 형태로 받아요.

  • AudioInput — 완전한 오디오 입력이 있고 그에 대한 결과만 만들고 싶을 때. 화자가 말을 다 했는지 감지할 필요가 없는 경우(미리 녹음된 오디오나, 사용자가 다 말했는지 분명한 push-to-talk 앱)에 유용해요.
  • StreamedAudioInput — 사용자가 다 말했는지 감지해야 할 수 있을 때. 감지된 대로 오디오 청크를 밀어넣을 수 있고, 음성 파이프라인은 "activity detection"이라는 과정을 통해 적절한 시점에 에이전트 워크플로를 자동으로 실행해요.

결과

음성 파이프라인 실행의 결과는 StreamedAudioResult예요. 이 객체는 이벤트가 발생하는 대로 스트리밍하게 해줘요. VoiceStreamEvent에는 몇 가지 종류가 있어요.

  • 오디오 청크를 담는 VoiceStreamEventAudio
  • 턴 시작·종료 같은 lifecycle 이벤트를 알려주는 VoiceStreamEventLifecycle
  • 오류 이벤트인 VoiceStreamEventError

터미널 파이프라인 오류는 애플리케이션이 StreamedAudioResult.stream()을 소비할 때 발생해요. 그 외에는 깨끗한 실행 뒤에 speech-to-text 전사 세션이 닫히지 못하면, 스트림이 무한정 기다리는 대신 그 닫기 오류를 발생시켜요. 턴이 이미 실패했는데 전사 세션 닫기도 실패하면, 스트림은 기본 오류로 원래 턴 오류를 보존해요.

result = await pipeline.run(input)

async for event in result.stream():
    if event.type == "voice_stream_event_audio":
        # play audio
        pass
    elif event.type == "voice_stream_event_lifecycle":
        # lifecycle
        pass
    elif event.type == "voice_stream_event_error":
        # error
        pass

모범 사례

Interruption (중단)

Agents SDK는 현재 StreamedAudioInput에 대한 내장 중단 처리를 제공하지 않아요. 대신 감지된 각 턴이 워크플로의 별도 실행을 트리거해요. 애플리케이션 안에서 중단을 처리하고 싶다면 VoiceStreamEventLifecycle 이벤트를 들을 수 있어요. turn_started는 새 턴이 전사되고 처리가 시작됐음을 나타내고, turn_ended는 해당 턴과 관련된 모든 오디오가 전달된 뒤에 트리거돼요. 모델이 턴을 시작하면 화자 마이크를 음소거하고, 애플리케이션이 그 턴 관련 오디오를 모두 재생한 뒤 마이크를 다시 켜는 데 이 이벤트들을 쓸 수 있어요.

더 알아보기 (Learn more)