LangChain으로 음성 에이전트 만들기

LangChain으로 음성 에이전트 만들기 (Build a voice agent with LangChain)

지금까지 우리는 AI와 주로 채팅 인터페이스로 상호작용해 왔어요. 하지만 최근 멀티모달 AI의 발전이 흥미로운 가능성을 열어주고 있죠. 고품질 생성 모델과 표현력이 뛰어난 TTS(텍스트-음성 변환, text-to-speech) 시스템 덕분에, 도구라기보다 대화 상대처럼 느껴지는 에이전트를 만들 수 있게 됐어요. 음성 에이전트도 그중 하나예요. 키보드와 마우스로 입력을 타이핑하는 대신 말로 에이전트와 상호작용할 수 있고, 이는 더 자연스럽고 몰입감 있는 방식이죠. 특정 상황에서는 특히 유용합니다.

출처: LangChain 공식 문서 — Build a voice agent with LangChain

음성 에이전트란 무엇인가 (What are voice agents?)

음성 에이전트는 사용자와 자연스러운 음성 대화를 나눌 수 있는 에이전트예요. 음성 인식(speech recognition), 자연어 처리, 생성형 AI, 텍스트-음성 변환 기술을 결합해서 매끄럽고 자연스러운 대화를 만들어냅니다. 다음과 같은 다양한 사용 사례에 적합해요.

고객 지원 (Customer support)
개인 비서 (Personal assistants)
핸즈프리 인터페이스 (Hands-free interfaces)
코칭과 교육 (Coaching and training)

음성 에이전트는 어떻게 동작하나 (How do voice agents work?)

크게 보면 모든 음성 에이전트는 세 가지 일을 처리해야 해요.

듣기 (Listen) – 오디오를 캡처하고 텍스트로 변환(transcribe)
생각하기 (Think) – 의도를 해석하고, 추론하고, 계획
말하기 (Speak) – 오디오를 생성하고 사용자에게 스트리밍

차이는 이 단계들을 어떻게 순서화하고 결합하느냐에 있어요. 실제 프로덕션 에이전트는 두 가지 주요 아키텍처 중 하나를 따릅니다.

1. STT > Agent > TTS 구조 (The "Sandwich")

샌드위치(Sandwich) 아키텍처는 세 개의 개별 컴포넌트를 조합합니다: 음성-텍스트 변환(STT, speech-to-text), 텍스트 기반 LangChain 에이전트, 텍스트-음성 변환(TTS)이에요.

flowchart LR
    A[User Audio] --> B[Speech-to-Text]
    B --> C[LangChain Agent]
    C --> D[Text-to-Speech]
    D --> E[Audio Output]

장점 (Pros):

각 컴포넌트를 완전히 제어 (필요에 따라 STT/TTS 프로바이더 교체 가능)
최신 텍스트 모달리티 모델의 최신 기능에 접근
컴포넌트 간 경계가 명확해서 동작이 투명

단점 (Cons):

여러 서비스를 오케스트레이션해야 함
파이프라인 관리에 추가 복잡성
음성→텍스트 변환 과정에서 정보 손실 (예: 톤, 감정)

2. 음성-음성 구조 (S2S, Speech-to-Speech)

음성-음성(Speech-to-Speech)은 오디오 입력을 처리하고 오디오 출력을 고유하게 생성하는 멀티모달 모델을 사용해요.

flowchart LR
    A[User Audio] --> B[Multimodal Model]
    B --> C[Audio Output]

장점 (Pros):

움직이는 부품이 적어 구조가 더 단순
단순한 상호작용에서는 보통 지연 시간이 더 낮음
직접 오디오 처리가 톤과 말의 미묘함을 포착

단점 (Cons):

모델 선택지가 제한적이고, 프로바이더 락인(lock-in) 위험이 큼
기능이 텍스트 모달리티 모델보다 뒤처질 수 있음
오디오 처리 방식의 투명성이 낮음
제어와 커스터마이즈 옵션이 줄어듦

이 가이드는 성능, 제어, 최신 모델 기능 접근의 균형을 위해 샌드위치 아키텍처를 보여줍니다. 샌드위치는 일부 STT/TTS 프로바이더로 700ms 미만의 지연 시간을 내면서도 모듈식 컴포넌트를 계속 제어할 수 있어요.

데모 애플리케이션 개요 (Demo Application overview)

샌드위치 아키텍처를 사용하는 음성 기반 에이전트를 만들어 볼게요. 이 에이전트는 샌드위치 가게의 주문을 관리합니다. 샌드위치 아키텍처의 세 컴포넌트를 모두 보여주며, STT로는 AssemblyAI, TTS로는 Cartesia를 사용합니다(대부분 프로바이더용 어댑터를 만들 수 있어요). 전체 참조 애플리케이션은 voice-sandwich-demo 저장소에서 볼 수 있고, 여기서는 그 애플리케이션을 함께 살펴볼 거예요.

데모는 브라우저와 서버 사이의 실시간 양방향 통신에 WebSockets을 사용해요. 같은 아키텍처는 Twilio, Vonage 같은 전화 시스템이나 WebRTC 연결 같은 다른 전송 방식으로도 바꿔 적용할 수 있습니다.

아키텍처 (Architecture)

데모는 각 단계가 비동기로 데이터를 처리하는 스트리밍 파이프라인을 구현합니다.

클라이언트 (브라우저)

마이크 오디오를 캡처해 PCM으로 인코딩
백엔드 서버에 WebSocket 연결 수립
오디오 청크를 실시간으로 서버에 스트리밍
합성된 음성 오디오를 받아 재생

서버 (Python)

클라이언트의 WebSocket 연결 수락
세 단계 파이프라인 오케스트레이션:
- STT: 오디오를 STT 프로바이더(예: AssemblyAI)로 전달하고 transcript 이벤트 수신
- Agent: transcript를 LangChain 에이전트로 처리하고 응답 토큰 스트리밍
- TTS: 에이전트 응답을 TTS 프로바이더(예: Cartesia)로 보내고 오디오 청크 수신
합성된 오디오를 재생을 위해 클라이언트에 반환

파이프라인은 각 단계에서 비동기 제너레이터를 사용해 스트리밍을 가능하게 해요. 이렇게 하면 업스트림 단계가 끝나기 전에 다운스트림 컴포넌트가 처리를 시작할 수 있어서 end-to-end 지연 시간을 최소화합니다.

설정 (Setup)

자세한 설치 지침과 설정은 저장소 README를 참고하세요.

1. 음성→텍스트 (Speech-to-text)

STT 단계는 들어오는 오디오 스트림을 텍스트 transcript로 변환합니다. 구현은 오디오 스트리밍과 transcript 수신을 동시에 처리하기 위해 생산자-소비자(producer-consumer) 패턴을 사용해요.

핵심 개념 (Key concepts):

생산자-소비자 패턴: 오디오 청크를 STT 서비스로 보내는 동시에 transcript 이벤트를 수신합니다. 오디오가 전부 도착하기 전에 전사(transcription)가 시작될 수 있어요.
이벤트 타입: stt_chunk(STT 서비스가 오디오를 처리하며 제공하는 부분 transcript), stt_output(에이전트 처리를 트리거하는 최종·포맷된 transcript).
WebSocket 연결: AssemblyAI의 실시간 STT API에 영구 연결을 유지하며, 16kHz PCM 오디오에 자동 턴 포맷팅으로 구성됨.

from typing import AsyncIterator
import asyncio
from assemblyai_stt import AssemblyAISTT
from events import VoiceAgentEvent

async def stt_stream(
    audio_stream: AsyncIterator[bytes],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    Transform stream: Audio (Bytes) → Voice Events (VoiceAgentEvent)

    Uses a producer-consumer pattern where:
    - Producer: Reads audio chunks and sends them to AssemblyAI
    - Consumer: Receives transcription events from AssemblyAI
    """
    stt = AssemblyAISTT(sample_rate=16000)

    async def send_audio():
        """Background task that pumps audio chunks to AssemblyAI."""
        try:
            async for audio_chunk in audio_stream:
                await stt.send_audio(audio_chunk)
        finally:
            # Signal completion when audio stream ends
            await stt.close()

    # Launch audio sending in background
    send_task = asyncio.create_task(send_audio())

    try:
        # Receive and yield transcription events as they arrive
        async for event in stt.receive_events():
            yield event
    finally:
        # Cleanup
        with contextlib.suppress(asyncio.CancelledError):
            send_task.cancel()
            await send_task
        await stt.close()

애플리케이션은 WebSocket 연결과 메시지 파싱을 관리하는 AssemblyAI 클라이언트를 구현합니다. 다른 STT 프로바이더용으로도 비슷한 어댑터를 만들 수 있어요.

2. LangChain 에이전트

에이전트 단계는 텍스트 transcript를 LangChain 에이전트로 처리하고 응답 토큰을 스트리밍합니다. 이 경우 에이전트가 생성하는 모든 텍스트 콘텐츠 블록을 스트리밍해요.

핵심 개념 (Key concepts):

스트리밍 응답: 에이전트는 stream_events(version="v3")와 stream.messages를 사용해서, 완전한 응답을 기다리는 대신 응답 토큰이 생성되는 대로 방출합니다. 이렇게 하면 TTS 단계가 즉시 합성 작업을 시작할 수 있어요.
대화 메모리: 체크포인터(checkpointer)가 고유 thread ID로 대화 간 상태를 유지해서, 에이전트가 이전 대화를 참조할 수 있습니다.

from langchain_core.utils.uuid import uuid7
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

# Define agent tools
def add_to_order(item: str, quantity: int) -> str:
    """Add an item to the customer's sandwich order."""
    return f"Added {quantity} x {item} to the order."

def confirm_order(order_summary: str) -> str:
    """Confirm the final order with the customer."""
    return f"Order confirmed: {order_summary}. Sending to kitchen."

# Create agent with tools and memory
agent = create_agent(
    model="google_genai:gemini-3.6-flash",  # Select your model
    tools=[add_to_order, confirm_order],
    system_prompt="""You are a helpful sandwich shop assistant.
    Your goal is to take the user's order. Be concise and friendly.
    Do NOT use emojis, special characters, or markdown.
    Your responses will be read by a text-to-speech engine.""",
    checkpointer=InMemorySaver(),
)

async def agent_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    Transform stream: Voice Events → Voice Events (with Agent Responses)

    Passes through all upstream events and adds agent_chunk events
    when processing STT transcripts.
    """
    # Generate unique thread ID for conversation memory
    thread_id = str(uuid7())

    async for event in event_stream:
        # Pass through all upstream events
        yield event

        # Process final transcripts through the agent
        if event.type == "stt_output":
            # Stream agent response with conversation context
            stream = await agent.astream_events(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
                version="v3",
            )

            # Yield agent response chunks as they arrive
            async for message in stream.messages:
                async for token in message.text:
                    yield AgentChunkEvent.create(token)

3. 텍스트→음성 (Text-to-speech)

TTS 단계는 에이전트 응답 텍스트를 오디오로 합성하고 클라이언트로 스트리밍합니다. STT 단계처럼 텍스트 전송과 오디오 수신을 동시에 처리하는 생산자-소비자 패턴을 사용해요.

핵심 개념 (Key concepts):

동시 처리: 구현이 두 개의 비동기 스트림을 병합합니다: 업스트림 처리(모든 이벤트를 통과시키고 에이전트 텍스트 청크를 TTS 프로바이더로 전송)와 오디오 수신(TTS 프로바이더에서 합성된 오디오 청크 수신).
스트리밍 TTS: Cartesia 같은 일부 프로바이더는 텍스트를 받는 즉시 음성 합성을 시작해서, 에이전트가 완전한 응답을 만들기 전에 오디오 재생이 시작될 수 있어요.
이벤트 패스스루: 모든 업스트림 이벤트가 그대로 흘러가서 클라이언트나 다른 관찰자가 전체 파이프라인 상태를 추적할 수 있음.

from cartesia_tts import CartesiaTTS
from utils import merge_async_iters

async def tts_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    Transform stream: Voice Events → Voice Events (with Audio)

    Merges two concurrent streams:
    1. process_upstream(): passes through events and sends text to Cartesia
    2. tts.receive_events(): yields audio chunks from Cartesia
    """
    tts = CartesiaTTS()

    async def process_upstream() -> AsyncIterator[VoiceAgentEvent]:
        """Process upstream events and send agent text to Cartesia."""
        async for event in event_stream:
            # Pass through all events
            yield event
            # Send agent text to Cartesia for synthesis
            if event.type == "agent_chunk":
                await tts.send_text(event.text)

    try:
        # Merge upstream events with TTS audio events
        # Both streams run concurrently
        async for event in merge_async_iters(
            process_upstream(),
            tts.receive_events()
        ):
            yield event
    finally:
        await tts.close()

애플리케이션은 WebSocket 연결과 오디오 스트리밍을 관리하는 Cartesia 클라이언트를 구현합니다. 다른 TTS 프로바이더용으로도 비슷한 어댑터를 만들 수 있어요.

LangSmith로 추적하기

LangChain으로 만드는 많은 애플리케이션은 여러 단계와 여러 LLM 호출을 포함해요. 애플리케이션이 복잡해질수록 체인이나 에이전트 내부에서 정확히 무슨 일이 일어나는지 살펴볼 수 있어야 하는 게 중요해집니다. 가장 좋은 방법은 LangSmith를 쓰는 것이에요. 위 링크에서 가입한 뒤, 추적을 기록하도록 환경 변수를 설정하세요.

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

또는 Python에서 설정할 수 있어요.

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

전체 파이프라인 연결하기 (Putting it all together)

완전한 파이프라인은 세 단계를 체인으로 연결합니다.

from langchain_core.runnables import RunnableGenerator

pipeline = (
    RunnableGenerator(stt_stream)      # Audio → STT events
    | RunnableGenerator(agent_stream)  # STT events → Agent events
    | RunnableGenerator(tts_stream)    # Agent events → TTS audio
)

# Use in WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    async def websocket_audio_stream():
        """Yield audio bytes from WebSocket."""
        while True:
            data = await websocket.receive_bytes()
            yield data

    # Transform audio through pipeline
    output_stream = pipeline.atransform(websocket_audio_stream())

    # Send TTS audio back to client
    async for event in output_stream:
        if event.type == "tts_chunk":
            await websocket.send_bytes(event.audio)

파이프라인의 각 단계를 조합하기 위해 RunnableGenerator를 사용해요. 이는 LangChain이 컴포넌트 간 스트리밍을 관리하기 위해 내부적으로 쓰는 추상화입니다. 각 단계는 이벤트를 독립적이고 동시에 처리합니다: 오디오가 도착하는 즉시 전사가 시작되고, transcript가 준비되는 즉시 에이전트가 추론을 시작하며, 에이전트 텍스트가 생성되는 즉시 음성 합성이 시작됩니다. 이 아키텍처는 자연스러운 대화를 지원하는 700ms 미만의 지연 시간을 달성할 수 있어요.

LangChain으로 음성 에이전트 만들기