이미지 로딩 중...
AI Generated
2025. 11. 19. · 5 Views
대화형 LLM과 TTS 통합 완벽 가이드
ChatGPT와 Claude 같은 대화형 AI와 음성 합성 기술을 결합하여 자연스러운 음성 비서를 만드는 방법을 배워봅니다. STT부터 TTS까지 전체 파이프라인을 구축하고, 실시간 대화 관리와 Function Calling을 통한 기능 확장까지 다룹니다.
목차
- ChatGPT/Claude API와 통합 구조
- 텍스트 응답 생성 → 음성 변환 파이프라인
- 음성 입력 → STT → LLM → TTS → 음성 출력
- Context 관리 및 대화 히스토리
- Multi-turn Conversation 구현
- Function Calling으로 비서 기능 확장
1. ChatGPT/Claude API와 통합 구조
시작하며
여러분이 음성 비서 앱을 만들고 싶어서 ChatGPT API를 살펴봤는데, "이걸 어떻게 음성이랑 연결하지?"라고 막막했던 적 있나요? 실제로 많은 개발자들이 LLM API는 텍스트만 주고받는데, 이걸 음성 시스템과 어떻게 연결해야 할지 고민합니다.
이런 문제는 실제 개발 현장에서 자주 발생합니다. LLM은 강력한 텍스트 생성 능력을 가지고 있지만, 음성 입출력과의 연결 구조를 잘못 설계하면 응답이 느려지거나 대화 흐름이 끊기는 문제가 생깁니다.
바로 이럴 때 필요한 것이 적절한 API 통합 아키텍처입니다. LLM API와 음성 시스템을 효율적으로 연결하면 자연스럽고 빠른 음성 대화 시스템을 구축할 수 있습니다.
개요
간단히 말해서, 이 개념은 LLM API를 중심으로 음성 입출력 모듈을 연결하는 파이프라인 설계입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 음성 비서, 고객 상담 봇, 교육용 AI 튜터 같은 서비스를 만들 때 반드시 필요합니다.
예를 들어, 병원 예약 시스템에서 사용자가 말로 증상을 설명하면 AI가 이해하고 적절한 진료과를 추천하는 경우에 매우 유용합니다. 전통적인 방법과의 비교를 하면, 기존에는 규칙 기반 챗봇과 미리 녹음된 음성을 사용했다면, 이제는 실시간으로 자연스러운 대화를 생성하고 다양한 목소리로 음성을 합성할 수 있습니다.
이 개념의 핵심 특징은 첫째, 비동기 처리로 빠른 응답 속도를 보장하고, 둘째, 모듈화된 구조로 각 컴포넌트를 독립적으로 교체 가능하며, 셋째, 스트리밍 방식으로 긴 응답도 끊김 없이 처리할 수 있다는 것입니다. 이러한 특징들이 사용자 경험을 크게 향상시키고 시스템 확장성을 높여줍니다.
코드 예제
import openai
from typing import Dict, List
class LLMIntegration:
def __init__(self, api_key: str, model: str = "gpt-4"):
# OpenAI 클라이언트 초기화
self.client = openai.OpenAI(api_key=api_key)
self.model = model
async def get_response(self, messages: List[Dict[str, str]]) -> str:
# LLM에 메시지를 보내고 응답 받기
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7, # 창의성 조절
max_tokens=500 # 응답 길이 제한
)
# 응답 텍스트 추출
return response.choices[0].message.content
설명
이것이 하는 일은 ChatGPT나 Claude 같은 LLM API를 초기화하고, 텍스트 메시지를 주고받을 수 있는 기본 구조를 만드는 것입니다. 첫 번째로, __init__ 메서드에서 API 키와 사용할 모델을 설정합니다.
여기서 중요한 점은 API 클라이언트를 한 번만 초기화해서 재사용한다는 것입니다. 매번 새로 만들면 불필요한 네트워크 오버헤드가 발생하기 때문입니다.
두 번째로, get_response 메서드가 실행되면서 실제 LLM과 통신합니다. async와 await를 사용하는 이유는 LLM 응답을 기다리는 동안 다른 작업을 처리할 수 있게 하기 위함입니다.
temperature는 응답의 창의성을 조절하는데, 0.7은 적당히 다양하면서도 일관된 답변을 생성합니다. 세 번째로, max_tokens로 응답 길이를 제한합니다.
이것이 중요한 이유는 음성으로 변환할 때 너무 긴 응답은 사용자가 듣기 지루하고, TTS 처리 시간도 오래 걸리기 때문입니다. 여러분이 이 코드를 사용하면 LLM과의 안정적인 통신 채널을 확보하고, 에러 핸들링과 재시도 로직을 추가하여 프로덕션 수준의 시스템을 구축할 수 있습니다.
또한 Claude API로 바꾸고 싶을 때도 이 클래스만 수정하면 되므로 유지보수가 쉽습니다.
실전 팁
💡 API 키는 반드시 환경 변수로 관리하세요. 코드에 직접 넣으면 GitHub에 올렸을 때 보안 문제가 생깁니다.
💡 rate limit 에러를 방지하려면 exponential backoff(점진적 재시도) 로직을 추가하세요. 첫 실패 시 1초, 두 번째는 2초, 세 번째는 4초 대기하는 식입니다.
💡 프로덕션 환경에서는 응답 시간이 5초를 넘으면 타임아웃 처리하세요. 사용자는 5초 이상 기다리면 앱이 멈춘 것으로 느낍니다.
💡 비용 절감을 위해 gpt-3.5-turbo를 먼저 사용하고, 복잡한 질문일 때만 gpt-4로 전환하는 hybrid 전략을 고려하세요.
💡 로깅을 꼭 추가하세요. 어떤 질문에 어떤 답변을 했는지 기록해야 나중에 품질 개선과 디버깅이 가능합니다.
2. 텍스트 응답 생성 → 음성 변환 파이프라인
시작하며
여러분이 LLM에서 긴 답변을 받았는데, 이걸 음성으로 변환하려니 10초 넘게 기다려야 했던 경험 있나요? 사용자는 답변이 끝날 때까지 아무것도 듣지 못하고 기다려야 합니다.
이런 문제는 실제 개발 현장에서 자주 발생합니다. 전체 텍스트를 한 번에 TTS로 변환하면 첫 음성이 나올 때까지 시간이 오래 걸려 사용자 경험이 나빠집니다.
특히 긴 설명이나 이야기를 들려줄 때 이 문제가 심각합니다. 바로 이럴 때 필요한 것이 스트리밍 파이프라인입니다.
텍스트를 조금씩 받으면서 동시에 음성으로 변환하면 훨씬 자연스럽고 빠른 응답을 제공할 수 있습니다.
개요
간단히 말해서, 이 개념은 LLM이 텍스트를 생성하는 동시에 TTS가 음성으로 변환하는 병렬 처리 시스템입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 실시간 음성 대화 시스템에서 응답 지연을 최소화하고 자연스러운 대화 흐름을 만들기 위해 필수적입니다.
예를 들어, 운전 중 음성 네비게이션이나 요리하면서 듣는 레시피 안내 같은 경우에 빠른 첫 응답이 매우 중요합니다. 전통적인 방법과의 비교를 하면, 기존에는 LLM 응답이 완전히 끝난 후 TTS 시작했다면, 이제는 문장 단위로 즉시 음성 변환을 시작할 수 있습니다.
이 개념의 핵심 특징은 첫째, 청크 단위 처리로 첫 음성 출력까지 시간을 90% 단축하고, 둘째, 큐 기반 아키텍처로 안정적인 데이터 흐름을 보장하며, 셋째, 에러 발생 시 부분 응답이라도 전달할 수 있다는 것입니다. 이러한 특징들이 사용자가 느끼는 응답 속도를 획기적으로 개선합니다.
코드 예제
import asyncio
from queue import Queue
from openai import AsyncOpenAI
class StreamingPipeline:
def __init__(self, llm_client, tts_client):
self.llm = llm_client
self.tts = tts_client
self.text_queue = Queue() # 텍스트 버퍼
async def process_streaming_response(self, messages: list):
# LLM 스트리밍 응답 시작
stream = await self.llm.chat.completions.create(
model="gpt-4",
messages=messages,
stream=True # 스트리밍 활성화
)
sentence_buffer = ""
async for chunk in stream:
# 텍스트 청크 수신
content = chunk.choices[0].delta.content or ""
sentence_buffer += content
# 문장 완성 시 TTS 큐에 추가
if any(punct in content for punct in ['.', '!', '?', '\n']):
self.text_queue.put(sentence_buffer.strip())
sentence_buffer = ""
# 남은 텍스트 처리
if sentence_buffer:
self.text_queue.put(sentence_buffer)
self.text_queue.put(None) # 종료 신호
설명
이것이 하는 일은 LLM이 텍스트를 생성하는 즉시 문장 단위로 잘라서 TTS로 보내는 병렬 처리 시스템을 구축하는 것입니다. 첫 번째로, stream=True 옵션으로 LLM 스트리밍을 활성화합니다.
이렇게 하면 전체 응답을 기다리지 않고 생성되는 즉시 텍스트 조각(청크)을 받을 수 있습니다. 일반 응답이 편지 한 통을 기다리는 것이라면, 스트리밍은 카카오톡처럼 한 글자씩 실시간으로 받는 것과 같습니다.
두 번째로, sentence_buffer에 텍스트를 모으다가 마침표나 느낌표 같은 문장 부호를 만나면 큐에 넣습니다. 왜 문장 단위로 끊느냐면, TTS가 문장 전체를 봐야 자연스러운 억양을 만들 수 있기 때문입니다.
단어 하나씩 보내면 로봇 같은 목소리가 됩니다. 세 번째로, Queue를 사용해서 LLM과 TTS 사이에 안전한 데이터 전달 통로를 만듭니다.
LLM이 빠르게 텍스트를 생성하더라도 큐에 쌓아두고, TTS는 자기 속도에 맞춰 처리할 수 있습니다. 마치 컨베이어 벨트처럼 작동합니다.
마지막으로, None을 큐에 넣어서 "더 이상 텍스트가 없다"는 신호를 보냅니다. 이것이 없으면 TTS가 무한정 기다리게 됩니다.
여러분이 이 코드를 사용하면 사용자는 질문 후 0.5초 내에 첫 음성을 듣기 시작하고, 전체 응답을 기다릴 필요가 없어집니다. 실제 Alexa나 Google Assistant가 이런 방식으로 작동합니다.
실전 팁
💡 문장 구분을 더 정교하게 하려면 spaCy나 NLTK 같은 NLP 라이브러리를 사용하세요. 단순 마침표 검사는 "Dr. Kim"이나 "3.14" 같은 경우를 잘못 끊습니다.
💡 큐 크기를 제한하세요. 무제한 큐는 메모리 누수를 일으킬 수 있습니다. Queue(maxsize=10)으로 최대 10개 문장만 버퍼링하는 것을 추천합니다.
💡 네트워크 끊김을 대비해 타임아웃을 설정하세요. async for 루프가 30초 이상 응답 없으면 에러 처리해야 합니다.
💡 한국어는 문장이 길어질 수 있으니 100자 이상이면 강제로 끊어서 TTS로 보내는 로직을 추가하세요. 너무 긴 문장은 TTS 품질이 떨어집니다.
💡 디버깅할 때는 큐에 들어가는 텍스트를 로그로 남기세요. 어디서 문장이 이상하게 끊기는지 쉽게 찾을 수 있습니다.
3. 음성 입력 → STT → LLM → TTS → 음성 출력
시작하며
여러분이 음성 비서를 만들었는데, 사용자가 말하면 한참 뒤에 답이 나오거나 중간에 멈추는 경험을 해본 적 있나요? 음성을 텍스트로, 텍스트를 다시 음성으로 바꾸는 과정이 순차적으로만 작동하면 전체 시간이 너무 오래 걸립니다.
이런 문제는 실제 개발 현장에서 자주 발생합니다. STT(음성→텍스트), LLM(텍스트 이해), TTS(텍스트→음성)를 각각 따로 처리하면 각 단계의 대기 시간이 모두 합쳐져서 사용자는 5~10초씩 기다려야 합니다.
바로 이럴 때 필요한 것이 전체 음성 대화 파이프라인의 최적화입니다. 각 단계를 효율적으로 연결하고 병렬 처리를 적용하면 자연스러운 실시간 대화가 가능합니다.
개요
간단히 말해서, 이 개념은 음성 입력부터 음성 출력까지 전체 흐름을 하나의 통합 시스템으로 구축하는 것입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 전화 상담 자동화, 차량 음성 제어, 스마트 홈 비서 같은 실시간 음성 서비스를 만들 때 필수적입니다.
예를 들어, 은행 고객센터에서 "잔액 조회해줘"라고 말하면 3초 안에 결과를 음성으로 들려줘야 고객이 만족합니다. 전통적인 방법과의 비교를 하면, 기존에는 각 단계가 끝날 때까지 기다렸다가 다음 단계 시작했다면, 이제는 VAD(음성 감지)로 말이 끝나자마자 즉시 처리를 시작하고 중간 결과도 활용할 수 있습니다.
이 개념의 핵심 특징은 첫째, WebRTC나 WebSocket으로 저지연 음성 스트리밍을 구현하고, 둘째, 각 모듈이 독립적으로 작동하면서도 효율적으로 데이터를 주고받으며, 셋째, 에러가 한 단계에서 발생해도 전체 시스템이 멈추지 않는다는 것입니다. 이러한 특징들이 프로덕션 수준의 안정적인 음성 비서를 만들 수 있게 해줍니다.
코드 예제
import asyncio
from openai import AsyncOpenAI
import speech_recognition as sr
from gtts import gTTS
import io
class VoiceAssistant:
def __init__(self, api_key: str):
self.llm_client = AsyncOpenAI(api_key=api_key)
self.recognizer = sr.Recognizer()
async def listen(self) -> str:
# 마이크에서 음성 입력 받기
with sr.Microphone() as source:
print("듣고 있습니다...")
audio = self.recognizer.listen(source)
# STT: 음성을 텍스트로 변환
text = self.recognizer.recognize_google(audio, language='ko-KR')
print(f"인식된 텍스트: {text}")
return text
async def think(self, user_text: str) -> str:
# LLM으로 응답 생성
response = await self.llm_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_text}]
)
return response.choices[0].message.content
async def speak(self, text: str):
# TTS: 텍스트를 음성으로 변환
tts = gTTS(text=text, lang='ko')
audio_buffer = io.BytesIO()
tts.write_to_fp(audio_buffer)
# 오디오 재생 (실제로는 pyaudio 등 사용)
print(f"말하기: {text}")
async def run_conversation(self):
# 전체 대화 루프
user_input = await self.listen()
ai_response = await self.think(user_input)
await self.speak(ai_response)
설명
이것이 하는 일은 사용자의 음성을 듣고, 이해하고, 답변을 생성한 후, 다시 음성으로 들려주는 완전한 대화 사이클을 만드는 것입니다. 첫 번째로, listen() 메서드에서 마이크로 음성을 녹음하고 Google STT API로 텍스트로 변환합니다.
language='ko-KR'이 중요한데, 이게 없으면 한국어를 영어로 잘못 인식합니다. VAD(Voice Activity Detection)를 사용해서 사용자가 말을 멈추는 순간을 자동으로 감지합니다.
두 번째로, think() 메서드가 실행되면서 LLM이 사용자 질문을 이해하고 적절한 답변을 생성합니다. 여기서 async/await를 사용하는 이유는 LLM 응답을 기다리는 동안 UI가 멈추지 않게 하기 위함입니다.
만약 동기 방식이면 3~4초간 앱이 완전히 멈춘 것처럼 보입니다. 세 번째로, speak() 메서드가 텍스트를 gTTS로 음성으로 변환합니다.
BytesIO를 사용해서 파일로 저장하지 않고 메모리에서 바로 처리하므로 더 빠릅니다. 실제 프로덕션에서는 ElevenLabs나 Azure TTS 같은 더 자연스러운 음성 엔진을 사용합니다.
마지막으로, run_conversation()이 이 모든 단계를 하나로 묶습니다. 각 메서드는 독립적으로 테스트할 수 있어서 나중에 STT를 Whisper로 바꾸거나 TTS를 다른 엔진으로 교체하기 쉽습니다.
여러분이 이 코드를 사용하면 기본적인 음성 비서 프로토타입을 30분 안에 만들 수 있고, 여기에 대화 히스토리나 감정 분석 같은 고급 기능을 쉽게 추가할 수 있습니다. 이 구조는 확장성이 매우 좋아서 실제 상용 서비스로 발전시키기도 좋습니다.
실전 팁
💡 음성 녹음 전에 주변 소음 레벨을 측정하는 adjust_for_ambient_noise() 메서드를 꼭 호출하세요. 카페 같은 시끄러운 곳에서도 음성 인식률이 크게 향상됩니다.
💡 STT 에러를 대비해 try-except로 감싸고, 인식 실패 시 "다시 한번 말씀해주세요"라고 음성으로 안내하세요. 사용자는 왜 반응이 없는지 모르면 불안해합니다.
💡 프로덕션에서는 Google STT 대신 Whisper를 추천합니다. 로컬에서 실행 가능하고 비용도 없으며, 한국어 인식률도 더 좋습니다.
💡 TTS 음성을 미리 캐싱하세요. "안녕하세요", "무엇을 도와드릴까요?" 같은 자주 쓰는 문구는 매번 생성하지 말고 미리 만들어두면 응답 속도가 빨라집니다.
💡 배터리 소모를 줄이려면 깨우기 단어(wake word)를 구현하세요. 항상 듣고 있지 말고 "헤이 AI" 같은 단어를 들었을 때만 녹음을 시작하면 됩니다. Porcupine 라이브러리가 좋습니다.
4. Context 관리 및 대화 히스토리
시작하며
여러분이 AI 비서에게 "그 영화 제목이 뭐였지?"라고 물었는데 "어떤 영화를 말씀하시는 건가요?"라고 되묻는 답답한 경험 있나요? 방금 전에 이야기했던 영화인데 AI가 까먹은 겁니다.
이런 문제는 실제 개발 현장에서 자주 발생합니다. LLM API는 상태를 저장하지 않기 때문에 매번 대화 내용을 함께 보내줘야 합니다.
이전 대화를 관리하지 않으면 사용자는 매번 모든 것을 다시 설명해야 해서 매우 불편합니다. 바로 이럴 때 필요한 것이 대화 히스토리 관리 시스템입니다.
이전 대화를 적절히 저장하고 관리하면 자연스러운 multi-turn 대화가 가능해집니다.
개요
간단히 말해서, 이 개념은 사용자와 AI의 모든 대화 내용을 저장하고, 새 질문을 할 때 관련된 이전 대화를 함께 LLM에 보내는 시스템입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 고객 상담, 개인 비서, 교육 튜터 같은 서비스에서 문맥을 이해하는 자연스러운 대화를 만들기 위해 필수적입니다.
예를 들어, 여행 플래너 챗봇에서 "서울 호텔 추천해줘", "그 중에 강남 쪽은?", "가격대는?" 같은 연속된 질문을 이해하려면 히스토리가 필요합니다. 전통적인 방법과의 비교를 하면, 기존에는 모든 대화를 무조건 다 보냈다가 토큰 한도 초과로 에러가 나거나 비용이 폭증했다면, 이제는 중요한 대화만 선별해서 보내거나 요약해서 보낼 수 있습니다.
이 개념의 핵심 특징은 첫째, 시간순 메시지 리스트로 대화 흐름을 유지하고, 둘째, 토큰 제한을 고려한 스마트한 히스토리 관리를 하며, 셋째, 장기 메모리와 단기 메모리를 분리해서 효율성을 높인다는 것입니다. 이러한 특징들이 실제 사람과 대화하는 것 같은 경험을 만들어줍니다.
코드 예제
from typing import List, Dict
from collections import deque
class ConversationManager:
def __init__(self, max_history: int = 10):
# 최근 대화만 유지 (메모리 절약)
self.history = deque(maxlen=max_history)
self.system_prompt = {
"role": "system",
"content": "당신은 친절한 AI 비서입니다."
}
def add_message(self, role: str, content: str):
# 새 메시지 추가
self.history.append({
"role": role, # 'user' 또는 'assistant'
"content": content
})
def get_messages(self) -> List[Dict]:
# LLM에 보낼 전체 메시지 구성
messages = [self.system_prompt]
messages.extend(self.history)
return messages
def clear_history(self):
# 대화 초기화
self.history.clear()
def get_token_count(self) -> int:
# 대략적인 토큰 수 계산 (4글자 ≈ 1토큰)
total_chars = sum(len(msg["content"]) for msg in self.history)
return total_chars // 4
def trim_if_needed(self, max_tokens: int = 2000):
# 토큰 제한 초과 시 오래된 대화 제거
while self.get_token_count() > max_tokens and len(self.history) > 1:
self.history.popleft()
설명
이것이 하는 일은 마치 사람의 단기 기억처럼 최근 대화 내용을 저장하고, 필요할 때 꺼내서 LLM에게 보여주는 것입니다. 첫 번째로, deque(maxlen=10)으로 최근 10개 대화만 자동으로 유지합니다.
deque는 양쪽 끝에서 빠르게 추가/삭제가 가능한 자료구조인데, 새 대화가 들어오면 가장 오래된 대화가 자동으로 사라집니다. 일반 리스트를 쓰면 오래된 항목 삭제할 때 전체를 다시 정렬해야 해서 느립니다.
두 번째로, system_prompt에 AI의 성격과 역할을 정의합니다. 이것은 절대 삭제되지 않고 항상 맨 앞에 붙어서 LLM에게 "너는 이런 역할이야"라고 알려줍니다.
만약 고객 상담 봇이면 "당신은 우리 회사 제품을 안내하는 전문 상담원입니다"로 바꾸면 됩니다. 세 번째로, get_token_count()로 현재 대화 내용의 토큰 수를 대략 계산합니다.
LLM은 입력 토큰 수에 제한이 있는데(GPT-4는 8192개), 이걸 넘으면 에러가 납니다. 한국어는 대략 4글자가 1토큰이므로 이 비율로 계산합니다.
네 번째로, trim_if_needed()가 토큰이 너무 많으면 오래된 대화를 자동으로 삭제합니다. popleft()는 가장 왼쪽(오래된) 항목을 제거하는데, 최근 대화가 더 중요하므로 오래된 것부터 버립니다.
여러분이 이 코드를 사용하면 사용자가 "그거", "저번에 말한", "아까 그" 같은 대명사를 써도 AI가 문맥을 이해하고 자연스럽게 답변할 수 있습니다. 또한 메모리 누수 걱정 없이 오랫동안 대화를 이어갈 수 있습니다.
실전 팁
💡 중요한 정보는 별도로 저장하세요. 사용자 이름, 선호도 같은 핵심 정보는 히스토리에만 의존하지 말고 데이터베이스에 영구 저장해야 합니다.
💡 대화 요약 기능을 추가하세요. 10턴이 넘어가면 오래된 대화를 LLM으로 요약해서 한 문장으로 만들고, 원본은 삭제하면 토큰을 크게 절약할 수 있습니다.
💡 민감한 정보는 마스킹하세요. 주민번호나 카드번호가 대화에 나오면 히스토리에 저장하기 전에 "--****"로 바꿔서 보안 사고를 예방하세요.
💡 사용자별로 히스토리를 분리하세요. 멀티 유저 환경이면 user_id를 키로 하는 딕셔너리에 각 사용자의 ConversationManager를 저장해야 대화가 섞이지 않습니다.
💡 대화 세션 타임아웃을 구현하세요. 30분간 대화가 없으면 히스토리를 자동으로 초기화해서 다음 대화가 이전 내용에 영향받지 않게 하세요.
5. Multi-turn Conversation 구현
시작하며
여러분이 "피자 주문하고 싶어"라고 말했는데 AI가 "무슨 피자요? 사이즈는?
토핑은? 주소는?"을 한 번에 다 물어보는 경험 있나요?
사람처럼 하나씩 자연스럽게 물어보면 좋을 텐데 말이죠. 이런 문제는 실제 개발 현장에서 자주 발생합니다.
단순히 히스토리만 저장한다고 자연스러운 대화가 되는 건 아닙니다. 대화 흐름을 관리하고, 필요한 정보를 단계적으로 수집하며, 사용자 의도를 추적하는 로직이 필요합니다.
바로 이럴 때 필요한 것이 Multi-turn 대화 관리 시스템입니다. 대화 상태를 추적하고 다음 질문을 지능적으로 선택하면 훨씬 자연스러운 대화 경험을 만들 수 있습니다.
개요
간단히 말해서, 이 개념은 대화를 여러 턴으로 나누고 각 턴마다 필요한 정보를 하나씩 수집하면서 최종 목표를 달성하는 시스템입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 예약 시스템, 설문 조사, 진단 챗봇 같은 복잡한 정보 수집이 필요한 서비스에서 필수적입니다.
예를 들어, 병원 예약 봇이 진료과 → 의사 → 날짜 → 시간을 순차적으로 물어보며 예약을 완성하는 경우입니다. 전통적인 방법과의 비교를 하면, 기존에는 복잡한 if-else 트리로 대화 흐름을 하드코딩했다면, 이제는 상태 머신과 LLM을 결합해서 유연하고 자연스러운 대화를 만들 수 있습니다.
이 개념의 핵심 특징은 첫째, 대화 상태를 명시적으로 관리해서 현재 어느 단계인지 추적하고, 둘째, 사용자 답변을 검증하고 필요시 재질문하며, 셋째, 사용자가 순서를 건너뛰거나 이전 단계로 돌아가도 처리할 수 있다는 것입니다. 이러한 특징들이 실제 사람과 대화하는 것 같은 유연성을 제공합니다.
코드 예제
from enum import Enum
from typing import Optional, Dict
class ConversationState(Enum):
INIT = "init"
COLLECTING_INFO = "collecting"
CONFIRMING = "confirming"
COMPLETED = "completed"
class MultiTurnConversation:
def __init__(self):
self.state = ConversationState.INIT
self.collected_data = {} # 수집된 정보 저장
self.required_fields = ['name', 'date', 'time']
self.current_field = 0
def get_next_question(self) -> str:
# 다음 질문 생성
if self.current_field >= len(self.required_fields):
return self._generate_confirmation()
field = self.required_fields[self.current_field]
questions = {
'name': "성함을 알려주시겠어요?",
'date': "언제 방문하실 예정인가요?",
'time': "몇 시가 좋으실까요?"
}
return questions.get(field, "정보를 입력해주세요.")
def process_answer(self, answer: str) -> str:
# 답변 처리 및 검증
if self.current_field < len(self.required_fields):
field = self.required_fields[self.current_field]
# 답변 검증 (실제로는 더 정교한 검증 필요)
if self._validate_answer(field, answer):
self.collected_data[field] = answer
self.current_field += 1
return self.get_next_question()
else:
return f"{answer}은(는) 유효하지 않습니다. 다시 입력해주세요."
return "감사합니다!"
def _validate_answer(self, field: str, answer: str) -> bool:
# 간단한 검증 로직
if field == 'name' and len(answer) < 2:
return False
return True
def _generate_confirmation(self) -> str:
# 수집된 정보 확인
return f"확인합니다: {self.collected_data}. 맞나요?"
설명
이것이 하는 일은 마치 전화 상담원이 하나씩 질문하며 정보를 메모하듯이, AI가 체계적으로 대화를 진행하는 것입니다. 첫 번째로, ConversationState Enum으로 대화의 현재 상태를 명확히 정의합니다.
상태가 있으면 "지금 뭘 해야 하지?"를 쉽게 판단할 수 있습니다. 예를 들어 COLLECTING_INFO 상태면 정보 수집 중이고, CONFIRMING 상태면 확인을 기다리는 중입니다.
상태 없이 if-else만 쓰면 코드가 스파게티처럼 꼬입니다. 두 번째로, collected_data 딕셔너리에 수집한 정보를 저장합니다.
required_fields 리스트로 어떤 정보를 수집해야 하는지 정의하고, current_field로 현재 몇 번째 질문인지 추적합니다. 이렇게 하면 나중에 필수 항목을 추가하거나 순서를 바꾸기 쉽습니다.
세 번째로, get_next_question()이 현재 상태에 맞는 적절한 질문을 생성합니다. 딕셔너리로 질문 템플릿을 관리하면 나중에 문구를 바꾸거나 다국어 지원을 추가하기 쉽습니다.
실제 프로덕션에서는 이 부분을 LLM에게 맡겨서 더 자연스러운 질문을 만들 수도 있습니다. 네 번째로, process_answer()가 사용자 답변을 검증합니다.
_validate_answer()에서 답변이 올바른지 확인하고, 문제가 있으면 다시 물어봅니다. 예를 들어 날짜를 물었는데 "빨리"라고 답하면 유효하지 않으므로 재질문합니다.
여러분이 이 코드를 사용하면 복잡한 예약, 주문, 설문 같은 프로세스를 체계적으로 관리할 수 있고, 사용자가 중간에 이탈해도 어디까지 진행됐는지 알 수 있어서 이어서 진행할 수 있습니다.
실전 팁
💡 사용자가 "처음부터 다시"라고 말하면 상태를 INIT으로 되돌리고 collected_data를 초기화하는 리셋 기능을 추가하세요.
💡 LLM을 활용해서 사용자가 여러 정보를 한 번에 말하면 자동으로 파싱하세요. "홍길동이고 내일 3시요"라고 하면 name, date, time을 한 번에 추출할 수 있습니다.
💡 대화 상태를 Redis 같은 외부 저장소에 저장하세요. 서버가 재시작되어도 사용자가 다시 처음부터 시작할 필요가 없습니다.
💡 각 필드에 타임아웃을 설정하세요. 사용자가 5분간 응답 없으면 "아직 계신가요?"라고 확인하고, 10분이면 세션을 종료합니다.
💡 진행률을 사용자에게 보여주세요. "3단계 중 1단계 완료" 같은 피드백을 주면 사용자가 얼마나 더 해야 하는지 알 수 있어 이탈률이 줄어듭니다.
6. Function Calling으로 비서 기능 확장
시작하며
여러분이 AI에게 "오늘 날씨 어때?"라고 물었는데 "저는 실시간 날씨 정보를 모릅니다"라는 답변을 받은 적 있나요? LLM은 대화는 잘하지만 실제 데이터를 가져오거나 액션을 실행하는 건 못합니다.
이런 문제는 실제 개발 현장에서 자주 발생합니다. LLM만으로는 실시간 정보 조회, 데이터베이스 검색, 외부 API 호출, 기기 제어 같은 실질적인 작업을 할 수 없습니다.
대화만 하는 AI는 쓸모가 제한적입니다. 바로 이럴 때 필요한 것이 Function Calling(함수 호출) 기능입니다.
LLM이 필요할 때 외부 함수를 호출하도록 하면 진짜 비서처럼 작동하는 AI를 만들 수 있습니다.
개요
간단히 말해서, 이 개념은 LLM이 "이건 내가 직접 처리 못하니 이 함수를 실행해줘"라고 요청하면, 실제로 그 함수를 실행하고 결과를 LLM에게 다시 알려주는 시스템입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 스마트 홈 제어, 일정 관리, 검색 비서, 주문 시스템 같은 실제 액션이 필요한 서비스를 만들 때 필수적입니다.
예를 들어, "거실 불 꺼줘"라고 말하면 AI가 스마트 전구 API를 호출해서 실제로 불을 끄는 경우입니다. 전통적인 방법과의 비교를 하면, 기존에는 키워드 매칭으로 "날씨"라는 단어를 찾아서 함수를 실행했다면, 이제는 LLM이 문맥을 이해하고 적절한 함수와 파라미터를 자동으로 선택할 수 있습니다.
이 개념의 핵심 특징은 첫째, LLM이 사용 가능한 함수 목록을 보고 상황에 맞는 것을 선택하고, 둘째, 필요한 파라미터를 사용자 질문에서 자동으로 추출하며, 셋째, 함수 실행 결과를 다시 LLM이 자연어로 설명해준다는 것입니다. 이러한 특징들이 LLM을 단순 대화봇에서 실질적인 업무 처리 비서로 업그레이드시켜줍니다.
코드 예제
import json
import openai
# 사용 가능한 함수 정의
def get_weather(location: str) -> dict:
# 실제로는 날씨 API 호출
return {"location": location, "temperature": "22°C", "condition": "맑음"}
def create_reminder(task: str, time: str) -> dict:
# 실제로는 DB에 저장
return {"status": "success", "task": task, "time": time}
# LLM에게 함수 정보 알려주기
functions = [
{
"name": "get_weather",
"description": "특정 지역의 현재 날씨를 조회합니다",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "도시 이름"}
},
"required": ["location"]
}
},
{
"name": "create_reminder",
"description": "리마인더를 생성합니다",
"parameters": {
"type": "object",
"properties": {
"task": {"type": "string", "description": "할 일"},
"time": {"type": "string", "description": "시간"}
},
"required": ["task", "time"]
}
}
]
async def chat_with_functions(user_message: str):
# 1단계: LLM에게 질문 + 사용 가능한 함수 목록 전달
response = await openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": user_message}],
functions=functions, # 함수 목록 제공
function_call="auto" # LLM이 자동으로 함수 선택
)
message = response.choices[0].message
# 2단계: LLM이 함수 호출을 요청했는지 확인
if message.get("function_call"):
function_name = message["function_call"]["name"]
arguments = json.loads(message["function_call"]["arguments"])
# 3단계: 실제 함수 실행
if function_name == "get_weather":
result = get_weather(**arguments)
elif function_name == "create_reminder":
result = create_reminder(**arguments)
# 4단계: 함수 결과를 LLM에게 다시 전달
second_response = await openai.ChatCompletion.create(
model="gpt-4",
messages=[
{"role": "user", "content": user_message},
message, # LLM의 함수 호출 요청
{"role": "function", "name": function_name, "content": json.dumps(result)}
]
)
return second_response.choices[0].message.content
return message.content
설명
이것이 하는 일은 마치 비서가 "이건 회계팀에 물어봐야겠다"라고 판단해서 실제로 회계팀에 연락하고 답변을 받아오듯이, LLM이 필요한 함수를 판단하고 실행하는 것입니다. 첫 번째로, functions 리스트에 LLM이 사용할 수 있는 함수들을 JSON 형식으로 정의합니다.
함수 이름, 설명, 파라미터를 명확히 적어야 LLM이 언제 어떤 함수를 써야 할지 정확히 판단합니다. description이 매우 중요한데, "날씨를 조회합니다"보다 "특정 지역의 현재 날씨를 조회합니다"가 훨씬 명확합니다.
두 번째로, 첫 번째 LLM 호출에서 functions 파라미터로 함수 목록을 전달합니다. function_call="auto"로 설정하면 LLM이 알아서 판단합니다.
사용자가 "서울 날씨 어때?"라고 물으면 LLM은 "아, 이건 get_weather 함수를 써야겠다"라고 판단하고, arguments에 {"location": "서울"}을 자동으로 채워줍니다. 세 번째로, LLM의 응답에서 function_call이 있는지 확인합니다.
있으면 LLM이 함수 실행을 요청한 것이므로, function_name과 arguments를 추출해서 실제 Python 함수를 실행합니다. json.loads()로 문자열을 딕셔너리로 변환하고, **arguments로 함수에 전달합니다.
네 번째로, 함수 실행 결과를 다시 LLM에게 보냅니다. 이때 role="function"으로 "이건 네가 요청한 함수의 실행 결과야"라고 알려줍니다.
그러면 LLM이 이 결과를 보고 "서울의 현재 날씨는 22도이며 맑습니다"처럼 자연스러운 문장으로 만들어줍니다. 여러분이 이 코드를 사용하면 "3시에 회의 알림 설정해줘", "내일 주가 확인해줘", "가장 가까운 카페 찾아줘" 같은 다양한 요청을 처리하는 만능 비서를 만들 수 있습니다.
함수만 추가하면 기능이 무한정 확장됩니다.
실전 팁
💡 함수 설명을 명확하게 작성하세요. 애매한 설명은 LLM이 잘못된 함수를 선택하게 만듭니다. "데이터를 가져옵니다" 대신 "사용자의 구매 이력을 최근 30일 기준으로 조회합니다"처럼 구체적으로 쓰세요.
💡 함수 실행 전에 파라미터를 검증하세요. LLM이 잘못된 값을 생성할 수 있으니 location이 빈 문자열이거나 이상한 값이면 에러 처리해야 합니다.
💡 함수 실행 시간이 오래 걸리면 타임아웃을 설정하세요. 외부 API가 응답하지 않으면 30초 후 "날씨 정보를 가져올 수 없습니다"라고 사용자에게 알려야 합니다.
💡 보안에 주의하세요. 함수가 데이터 삭제나 결제 같은 중요한 작업을 하면, LLM 판단만으로 실행하지 말고 사용자 확인을 받으세요. "정말 주문하시겠습니까?"처럼요.
💡 함수 실행 로그를 남기세요. 어떤 함수가 언제 어떤 파라미터로 실행됐는지 기록하면 디버깅과 사용 패턴 분석에 매우 유용합니다.