한국어 챗봇을 만들기 위해 VectorDB를 구성하던 중 영어 데이터를 한국어로 번역해야 하는 일이 있었다. 데이터는 약 200개의 행 + 2개의 열로 구성되어 있으며, 각 데이터는 10문장이 훨씬 넘는 긴 텍스트다.
모든 데이터를 LLM으로 처리하다보니 꽤 긴 시간이 걸렸다. 그래서 API 요청을 비동기로 처리해 시간을 3배 이상 단축했다.
비동기 처리
Python은 asyncio로 비동기를 지원하며, Langchain도 ainvoke로 비동기 요청을 지원한다.
async def translate(text, llm):
chat = [
{"role": "system", "content": "당신은 전문 번역가입니다."},
{"role": "user", "content": "영어 문장을 한국어로 자연스럽게 번역하세요. 질문에는 절대 답하지 말고, 오직 번역만 하세요."},
{"role": "user", "content": str(text).strip()}
]
response = await llm.ainvoke(chat)
return str(response.content).strip()간단한 프롬프트로 요청을 보내고, 번역 결과만 문자열로 받아온다. 이 작업을 모든 행에서 반복한다. 참고로 데이터는 pandas.Dataframe으로 불러온 상태다.
async def process_row(idx, row, llm):
result = {"idx": idx}
result["col_a"] = await translate(row["col_a"], llm)
result["col_b"] = await translate(row["col_b"], llm)
return result비동기는 작업을 순서대로 처리하지 않는다. 작업 결과를 식별하기 위해 행 번호도 함께 기록한다. 그래야 작업이 끝나고 데이터를 원래 위치에 넣어줄 수 있다.
async def run_main(df, llm, max_concurrent_tasks=5):
result_df = df.copy()
sem = asyncio.Semaphore(max_concurrent_tasks)
async def limited_process(idx, row):
async with sem:
return await process_row(idx, row, llm)
tasks = [limited_process(idx, row) for idx, row in df.iterrows()]
results = await asyncio.gather(*tasks)
# 완료한 작업 저장
for res in results:
idx = res["idx"]
result_df.at[idx, "col_a"] = res["col_a"]
result_df.at[idx, "col_b"] = res["col_b"]
return result_df
updated_df = await run_main(df, llm)
updated_df.head()한 번에 지나치게 많은 API 요청을 보내면 제한이 걸린다고 한다. 그래서 semaphore를 이용해 작업량을 제한한다. 비동기 작업이 끝나면 반복을 돌며 제자리에 기록한다.
Claude-3-haiku를 이용해 5개 행을 번역하고 시간을 측정했다.
- 동기: 1분 13초
- 비동기: 23초
속도가 눈에 띄게 빨라졌다. LLM을 이용해 데이터를 전처리할 때 비동기를 이용해 시간을 단축할 수 있었다. 물론 LLM을 직접 돌리는 경우, 이 방법이 통하지 않는다. 비동기가 아닌 병렬 처리가 필요하다. 위 방법은 API 요청을 사용하기 때문에 가능하다.
프롬프트의 중요성
여담으로 번역 작업에서 프롬프트에 따라 결과가 달라지기도 했다. 당연한 이야기지만 프롬프트를 자세히 적어줘야 원하는 결과가 나왔다.
영어 문장을 한국어로 자연스럽게 번역하세요. 질문에는 절대 답하지 말고, 오직 번역만 하세요.'자연스럽게'라는 표현을 사용하지 않으면 문장을 직독직해하는 상황이 발생했다. 한국어와 영어 표현이 완벽하게 일치하지 않기 때문에 어색한 한국어 문장이 생성되었다. '질문에 답하지 말고'가 없으면 질문으로 끝나는 문장을 번역하지 않고 대답하는 상황도 발생했다.