낮은 지연 시간으로 오디오 분석을 위한 LLM 최적화: 모범 사례

원문 발행 2026. 07. 16. 06:37원문 언어 영어AI 한국어 번역Dev.to AI tag 원문 보기

요약

긴 오디오 스크립트를 효율적으로 분석하기 위해 청킹, 병렬 처리, 결과 병합을 활용한 저지연 파이프라인 구축 방법을 소개합니다. Oxlo.ai API와 LLM을 사용하여 대규모 데이터를 구조화된 보고서로 빠르게 변환하는 엔지니어링 모범 사례를 다룹니다.

핵심 포인트

문장 경계와 오버랩을 활용한 스크립트 청킹 전략
JSON 출력을 강제하는 시스템 프롬프트 설계
병렬 처리를 통한 분석 지연 시간 최소화
두 번째 LLM 패스를 이용한 분석 결과의 일관된 병합

저희는 긴 오디오 스크립트를 청크(chunk)로 나누고, 이를 병렬로 분석한 뒤, 그 결과를 구조화된 보고서로 병합하는 낮은 지연 시간의 회의 인사이트 파이프라인을 구축하고 있습니다. 이 방식은 엔지니어링 팀이 단일 거대한 LLM 호출이 완료되기를 기다릴 필요 없이 원본 회의 녹음을 실행 가능한 요약으로 변환할 수 있도록 돕습니다.

준비물

Python 3.10 이상
https://portal.oxlo.ai에서 받은 Oxlo.ai API 키
OpenAI SDK: pip install openai
분석할 샘플 스크립트. Oxlo.ai의 Whisper 엔드포인트를 사용하거나 직접 텍스트를 붙여넣어 생성할 수 있습니다.

단계 1: 스크립트 청킹(Chunk the transcript)

저는 문장 경계에서 작은 오버랩(overlap) 창을 두고 스크립트를 분할합니다. 이렇게 하면 각 청크가 화자 컨텍스트를 유지하기에 충분히 크면서도, 긴 처리 지연 시간을 피할 수 있을 만큼 작습니다. 오버랩은 문장 경계를 넘어서 발생하는 실행 항목(action items)이 누락되는 것을 방지합니다.

import re

def chunk_transcript(text: str, chunk_size: int = 800, overlap: int = 100):
...

단계 2: 시스템 프롬프트 작성(Write the system prompt)

시스템 프롬프트는 모든 청크에 대해 유효한 JSON 출력을 강제합니다. 구조화된 출력은 병합 단계를 결정론적으로 만들고 나중에 정규 표현식(regex) 파싱을 피할 수 있게 합니다.

SYSTEM_PROMPT = """당신은 오디오 스크립트 분석가입니다. 제공된 스크립트 청크를 분석하고 마크다운 형식 없이 엄격한 JSON으로 반환하세요.

필수 필드:
...

단계 3: 청크 병렬 분석(Analyze chunks in parallel)

순차적인 청크 분석은 긴 녹음에서 빠르게 누적됩니다. 저는 Oxlo.ai를 통해 모든 청크 요청을 동시에 실행합니다. Oxlo.ai는 토큰당이 아닌 요청당 비용을 부과하기 때문에, 병렬화한다고 해서 비용이 증가하지 않으며, 요청별 고정 가격 덕분에 청크 길이가 달라져도 예측 가능합니다.

import json
import concurrent.futures
from openai import OpenAI
...

단계 4: 결과 병합(Merge the results)

병렬 청크는 액션 아이템을 중복하거나 컨텍스트를 분할할 수 있습니다. 두 번째 LLM 패스를 통해 부분적인 분석들을 하나의 일관된 보고서로 통합합니다. 여기서는 kimi-k2.6을 사용하는데, 이 모델은 고급 추론 능력과 긴 컨텍스트 창 덕분에 병합된 JSON 배열을 깔끔하게 처리해 줍니다.

MERGE_PROMPT = """당신은 선임 에디터입니다. 다음 JSON 분석 청크들을 하나의 일관된 보고서로 통합하세요.

입력: 청크 분석들의 JSON 배열.
... 
```json":""):
        clean = clean[7:]
    elif clean.startswith("```"):
        clean = clean[3:]
... 
```
":
        clean = clean[:-3]
    return json.loads(clean.strip())

Step 5: 최종 보고서 스트리밍 (Stream the final report)

...

def stream_final_report(merged: dict):
    prompt = f"다음 구조화된 분석을 간결한 서사 보고서로 변환하세요.\n\n{json.dumps(merged, indent=2)}"
    stream = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[ 
            {"role": "system", "content": "당신은 구조화된 데이터로부터 간결한 경영진 요약(executive summary)을 작성합니다."},
            {"role": "user", "content": prompt},
        ],
        stream=True,
    )

    print("최종 보고서:")
    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            print(content, end="", flush=True)
    print()

실행 (Run it)

...

if __name__ == "__main__":
    raw_transcript = """
    Alice: 우리는 금요일까지 인증 리팩토링(auth refactor)을 배포해야 합니다. Bob, OAuth 테스트는 당신이 처리할 수 있나요?
    Bob: 물론이죠, 하지만 개발 환경에 막혔습니다. 새로운 스테이징 박스가 작동하지 않습니다.
    Alice: 제가 지금 바로 IT에 티켓을 열겠습니다. Carol, Bob이 푸시하면 PR 검토를 부탁드립니다.
    Carol: 알겠습니다. 그리고 새 엔드포인트의 지연 시간(latency)은 여전히 400ms입니다. 출시 전에 프로파일링(profile)해야 합니다.
    Alice: 좋은 지적이에요. 스프린트에 프로파일링을 추가합시다. 오늘 스테이징 문제를 해결하면 일정에 자신이 생겼어요.
    ".strip()

chunks = chunk_transcript(raw_transcript, chunk_size=50, overlap=10)
    print(f"Split into {len(chunks)} chunks")

    analyses = analyze_all_chunks(chunks)
    final = merge_results(analyses)
    stream_final_report(final)

AI 자동 생성 콘텐츠

원문 바로가기

Insights