시리즈 · 음성봇 직접 만들기 — 실시간 음성봇의 뼈대를 바닥부터 만든 8편 기록입니다.
- 1. 끼어들기(barge-in) — 재생 중인 TTS를 실시간으로 끊는 법
- 2. "말이 끝났다"는 판단 — VAD 엔드포인터
- 3. 엔진을 코드 수정 없이 바꾸는 인터페이스 설계
- 4. 로컬 LLM 대화 두뇌와 도구 호출 루프 — 지금 보는 글
- 5. 매 턴 지연시간을 재는 이유
- 6. 브라우저 마이크에서 서버까지 — 오디오 프로토콜
- 7. 로컬 Whisper STT 연결하기
- 8. 스텁부터 만든 이유 — 테스트 우선 설계 (완결)
음성봇의 대화 두뇌를 규칙 기반에서 생성형 LLM으로 옮기던 중, 도구를 부르는 모델을 한 번 바꿔봤습니다. 이름만 보면 코드에 특화된 버전이라 오히려 도구(tool) 사용에 더 잘 맞을 것 같았습니다. 그런데 막상 붙여보니 모델이 도구를 부르긴 부르는데, 그 호출이 구조화된 형태로 오지 않고 답변 텍스트 안에 섞여서 나왔습니다. 도구 실행 결과를 다시 모델에 넘겨 자연스러운 답으로 바꾸는 루프 자체가 성립하지 않는 상황이었습니다. 이 경험을 계기로 로컬 LLM을 음성봇의 대화 두뇌로 붙이는 과정, 그중에서도 도구 호출 루프를 어떻게 짰는지 정리해 봅니다.
1. 두뇌만 통째로 갈아 끼울 수 있는 이유
이 프로젝트에서 대화 로직은 respond(text, session) -> reply라는 하나의 계약만 지키는 추상 인터페이스로 분리돼 있습니다. 규칙 기반으로 답을 고르던 이전 버전도, 이번에 만든 LLM 버전도 이 계약만 지키면 됩니다. 그 덕에 오케스트레이터, 음성 인식(STT), 음성 합성(TTS), WebSocket 통신, 프론트엔드는 전부 그대로 두고 대화를 만들어내는 로직만 통째로 교체할 수 있었습니다. 직접 만들어보고 나서야 이 분리가 얼마나 힘이 되는지 체감했습니다. 두뇌를 바꿀 때마다 나머지 파이프라인을 다시 손댈 필요가 없었기 때문입니다.
2. 로컬 모델을 실제로 부르는 코드
대화 두뇌는 로컬에서 돌아가는 Ollama 서버에 /api/chat 엔드포인트로 요청을 보내는 방식으로 구현했습니다. 다만 이 호출부를 직접 하드코딩하지 않고, 별도의 함수로 감싸서 주입받도록 분리했습니다.
def _make_real_chat_fn(base_url: str, model: str, timeout: float):
import httpx
def chat_fn(messages: list[dict], tools: list | None) -> dict:
payload = {"model": model, "messages": messages, "stream": False}
if tools:
payload["tools"] = tools
resp = httpx.post(f"{base_url}/api/chat", json=payload, timeout=timeout)
resp.raise_for_status()
return resp.json()
return chat_fn
이렇게 chat_fn을 의존성 주입 지점으로 빼두면, 테스트를 짤 때 실제 Ollama 서버나 네트워크 없이도 가짜 함수 하나로 도구 호출 루프 전체를 검증할 수 있습니다. 미리 정해둔 응답을 순서대로 돌려주는 함수를 넣어주면, 루프가 그 응답을 어떻게 처리하는지만 확인하면 되기 때문입니다. httpx도 이 함수 안에서만 지연 임포트하기 때문에, 가짜 함수를 쓰는 테스트는 이 무거운 의존성조차 필요하지 않습니다.
3. 도구 호출 루프의 뼈대
핵심은 모델에게 메시지와 도구 목록을 함께 보낸 뒤, 응답에 도구 호출이 들어있는지를 확인하는 것입니다. 있으면 실행하고, 그 결과를 다시 메시지에 얹어 모델에게 재질문합니다. 이 과정을 반복하다 도구 호출 없이 순수 텍스트만 오면 그걸 최종 답으로 돌려줍니다.
for _ in range(self.max_tool_iters):
resp = await asyncio.to_thread(self._chat_fn, messages, self.tools_specs)
msg = resp.get("message", {}) or {}
tool_calls = msg.get("tool_calls")
if not tool_calls:
final_text = msg.get("content", "") or ""
break
messages.append(msg)
for call in tool_calls:
fn = call.get("function", {}) or {}
name = fn.get("name", "")
args = self._parse_arguments(fn.get("arguments"))
result = self._dispatch(name, args)
messages.append({"role": "tool", "content": result, "tool_name": name})
여기서 max_tool_iters(기본값 3)를 둔 이유는 명확합니다. 모델이 도구 결과를 받고도 계속 같은 도구를 다시 부르거나, 애매한 판단으로 도구 호출을 반복하는 경우를 상정해야 하기 때문입니다. 반복 횟수에 상한을 걸어두면 이런 상황에서도 루프가 언젠가는 끝나고, 통화 자체가 무한 대기에 빠지지 않습니다. 실제 네트워크 호출은 asyncio.to_thread로 별도 스레드에 넘겨 실행하는데, 이렇게 해야 동기 방식인 chat_fn이 이벤트 루프 전체를 막아버리는 걸 피할 수 있습니다.
4. 도구는 두 개면 충분했다
모델이 부를 수 있는 도구는 예약 조회와 예약 생성, 두 가지로만 구성했습니다. 각 도구는 이름·설명·파라미터 스키마를 JSON 형식으로 정의해 모델에 함께 전달하고, 모델이 실제로 어떤 도구를 어떤 인자로 부를지 결정합니다.
TOOL_SPECS = [
{
"type": "function",
"function": {
"name": "create_reservation",
"description": "새 예약을 생성한다. 이름/날짜/시간/인원이 모두 모였을 때 사용하며, 확정 번호를 돌려준다.",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"date": {"type": "string"},
"time": {"type": "string"},
"party_size": {"type": "integer"},
},
"required": ["name", "date", "time", "party_size"],
},
},
},
]
모델은 이 스키마만 보고 판단할 뿐, 실제 파이썬 코드는 알지 못합니다. 도구 실행을 담당하는 dispatch 함수는 이름을 받아 해당 함수를 호출하고 결과를 JSON 문자열로 돌려줍니다. 인자가 잘못 들어와도 예외를 그대로 던지지 않고 오류 메시지를 담은 JSON을 돌려주도록 만들었는데, 그래야 모델이 그 오류를 읽고 부족한 정보를 다시 물어보는 식으로 회복할 수 있습니다.
5. 모델을 바꿨더니 벌어진 일
도입부에서 언급한 문제가 바로 이 지점에서 터졌습니다. 코드 생성에 특화된 모델 계열로 바꿔봤더니, 도구를 호출한다는 의도는 응답 안에 담겨 있는데 정작 그 형식이 구조화된 tool_calls 필드가 아니라 답변 본문 텍스트 안에 섞여서 나왔습니다. 루프는 tool_calls 필드가 비어 있는지만 확인하기 때문에, 이런 응답은 그냥 "도구 호출 없음"으로 처리되고 도구는 한 번도 실행되지 못했습니다. 반면 지시를 따르는 대화형(instruct) 계열 모델로 되돌리자 같은 요청에서 구조화된 tool_calls가 정상적으로 돌아왔고, 한국어 응답 품질도 더 안정적이었습니다.
| 구분 |
코드 특화 계열 |
대화형(instruct) 계열 |
| 도구 호출 형식 |
인라인 텍스트에 섞임 |
구조화된 tool_calls |
| 도구 호출 루프 |
깨짐 |
정상 동작 |
| 한국어 응답 |
그대로 사용 가능 |
그대로 사용 가능 |
결국 기본 모델은 대화형 계열로 고정했습니다. 이름만 보고 코드 특화 버전이 도구 사용에 유리할 거라 짐작했던 게 오판이었던 셈입니다.
6. 로컬 모델 서버가 죽으면
로컬에서 모델 서버를 직접 띄워 쓰다 보니 서버가 꺼져 있거나 응답이 없는 상황도 상정해야 했습니다. 이 호출은 음성 재생이 시작되기 전에 끝까지 기다려야 하는 자리라서, 예외가 그대로 위로 튀어 오르면 통화 전체가 끊길 위험이 있습니다. 그래서 연결 오류나 HTTP 오류를 잡아 음성으로 들려줄 안내 문구로 바꿔치기했습니다.
try:
final_text = await self._run_chat_loop(messages)
except self._llm_error_types() as exc:
return BrainReply(text="지금 시스템 연결이 원활하지 않습니다. 잠시 후 다시 시도해 주세요.")
이렇게 해두면 모델 서버가 잠깐 죽어 있어도 통화 자체는 끊기지 않고, 사용자는 봇이 일시적으로 응답을 못 만들었다는 걸 음성으로 안내받습니다. 다만 이건 통신 계층의 오류만 잡는 것이고, 코드 자체의 버그는 그대로 위로 전파되게 남겨뒀습니다. 통신 오류와 프로그래밍 오류를 같은 취급으로 삼켜버리면 진짜 버그를 놓치기 쉽기 때문입니다.
자주 묻는 질문
Q. 왜 도구를 두 개로만 제한했나요?
루프 구조 자체를 검증하는 게 이번 단계의 목표였습니다. 도구가 몇 개든 dispatch에 등록만 하면 루프 로직은 그대로 재사용되므로, 개수를 늘리는 건 이 골격 위에서 언제든 확장 가능한 부분입니다.
Q. 모델이 도구를 계속 잘못 부르면 어떻게 되나요?
dispatch가 예외 대신 오류 메시지를 담은 JSON을 돌려주기 때문에, 모델은 다음 차례에 그 오류를 읽고 스스로 다시 시도하거나 사용자에게 부족한 정보를 되물을 수 있습니다. 그래도 반복이 이어지면 max_tool_iters가 루프를 강제로 끝냅니다.
Q. 시스템 프롬프트에는 어떤 지침이 들어있나요?
답변이 음성으로 재생된다는 점을 감안해, 한두 문장으로 짧고 명확하게 답하라고 요구하는 페르소나 지침을 담아뒀습니다. 이 프로젝트는 실제 상용 서비스가 아니라 개인 학습용으로 만든 것이라, 지침 내용도 계속 다듬어가는 중입니다.
다음 편에서는 이 두뇌가 세션 기록을 어떻게 넘겨받아 대화 맥락을 유지하는지, 그리고 만들어진 답변이 실제로 음성으로 나가기까지 이어지는 나머지 흐름을 다룰 예정입니다.
음성봇 직접 만들기 시리즈 (전체 6편)
댓글
댓글 쓰기