작성자: Nishant Mishra
5분 읽기 · 2025년 6월 3일
고통스럽게 느린 AI 앱에서 번개처럼 빠른 워크플로우로의 나의 여정
지난주, 내 AI 애플리케이션에서 또 다른 타임아웃 오류를 디버깅하고 있을 때 깨달았다: 나는 모든 것을 잘못하고 있었다. 사용자들은 간단한 텍스트 분석에 60초 이상 기다리고 있었고, 내 서버는 끊임없이 한계에 달해 있었으며, 내일이 없는 것처럼 API 크레딧을 태우고 있었다.
그때 async LangChain과 LangGraph를 발견했다. 결과는? 응답 시간이 30초에서 3초 미만으로 줄어들었고, 이제 50배 더 많은 동시 사용자를 처리할 수 있다. 다음은 내가 배운 것들이다.
기본부터 시작해보자 (누군가 나에게 설명해줬으면 했던 것들)
처음 "LangChain"을 들었을 때, 나는 어떤 블록체인 관련된 것이라고 생각했다. 알고 보니, 훨씬 더 멋진 것이었다.
LangChain은 기본적으로 당신의 AI 워크플로우 어시스턴트다. 서로 다른 API들을 수동으로 연결하고, 대화 메모리를 관리하고, 수많은 boilerplate 코드를 작성해야 했던 예전을 기억하는가? LangChain이 그런 지루한 작업들을 모두 처리해주므로 당신은 멋진 기능 구축에 집중할 수 있다.
LangGraph는 흥미로운 부분이다. 이것은 AI 작업을 위한 흐름도를 만드는 것과 같다. 때때로 화이트보드에 상자와 화살표로 프로세스를 스케치하는 것을 알고 있는가? LangGraph는 그 스케치를 실제 작동하는 코드로 바꿔준다.
간단한 예시: 고객 피드백을 분석한다고 하자. 나의 프로세스는 다음과 같을 수 있다:
- 불만, 칭찬, 또는 질문인지 파악
- 주요 포인트 요약
- 취할 조치 결정
LangGraph를 사용하면, 이것을 시각적으로 매핑하고 자동으로 실행할 수 있다. 꽤 멋지지 않나?
Async의 깨달음 (또는: 어떻게 걱정을 멈추고 ainvoke를 사랑하게 되었는지)
솔직히 말하면 — 나는 수년간 async Python을 피했다. 복잡해 보였고, 내 동기 코드가 잘 작동했다... 작동하지 않을 때까지는.
내 기존 동기 방식에서 일어나고 있던 일은 다음과 같다:
내 AI 앱이 요청을 받고, OpenAI API를 호출하고, 기다리고... 기다리고... 기다리고... 응답을 받고, 다음 단계로 이동한다. 그 동안, 다른 사용자들은 세상에서 가장 느린 커피숍을 기다리는 것처럼 줄을 서서 기다리고 있었다.
Async를 사용하면, 여러 바리스타가 동시에 일하는 것과 같다. 하나의 요청이 다른 요청을 차단하지 않는다. 요청 #1에 대한 OpenAI의 응답을 기다리는 동안, 이미 요청 #2, #3, #4를 처리할 수 있다.
차이점은 invoke vs ainvoke이다. 그 작은 'a' 접두사가 모든 것을 바꾼다.
실제 사례: 실제로 작동하는 텍스트 분석기
고객 피드백을 처리해야 하는 클라이언트를 위해 간단한 텍스트 분석기를 구축했다. 다음은 마침내 작동한 코드다:
import os
import asyncio
from typing import TypedDict
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_community.chat_models import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
from langgraph.graph import StateGraph, END
# 선택사항: 환경 변수를 통해 OpenAI 키 설정
# os.environ["OPENAI_API_KEY"] = "sk-..."
# async 호환 LLM 정의
llm = ChatOpenAI(model="gpt-4", temperature=0)
# ----- State 구조 정의 -----
class State(TypedDict):
text: str
topic: str
summary: str
# ----- 단계 1: 주제 분류 -----
async def classify_topic(state: dict) -> dict:
input_text = state["text"]
messages = [
SystemMessage(content="You are a topic classifier."),
HumanMessage(content=f"What topic does this fall under? '{input_text}'")
]
response = await llm.ainvoke(messages)
return {
"text": input_text,
"topic": response.content
}
# ----- 단계 2: 텍스트 요약 -----
async def summarize_text(state: dict) -> dict:
messages = [
SystemMessage(content="You are a helpful summarizer."),
HumanMessage(content=f"Summarize this: '{state['text']}'")
]
response = await llm.ainvoke(messages)
return {
"text": state["text"],
"topic": state["topic"],
"summary": response.content
}
# ----- LangGraph 구축 -----
builder = StateGraph(State)
builder.add_node("classify", classify_topic)
builder.add_node("summarize", summarize_text)
builder.set_entry_point("classify")
builder.add_edge("classify", "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()
# ----- FastAPI 앱 설정 -----
app = FastAPI()
# ----- 입력 스키마 -----
class QueryRequest(BaseModel):
query: str
# ----- 출력 스키마 -----
class QueryResponse(BaseModel):
topic: str
summary: str
# ----- 엔드포인트 -----
@app.post("/analyze", response_model=QueryResponse)
async def analyze_query(request: QueryRequest):
try:
result = await graph.ainvoke({"text": request.query})
return QueryResponse(topic=result["topic"], summary=result["summary"])
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))여기서 무슨 일이 일어나고 있는지 설명해보겠다. 나도 이걸 이해하는 데 시간이 걸렸기 때문이다.
State에 대해
class State(TypedDict):
text: str
topic: str
summary: str이것은 워크플로우의 서로 다른 단계 간에 전달되는 공유 노트북과 같다. 각 함수는 이를 읽고 새로운 정보를 추가할 수 있다. 간단하지만 강력하다.
ainvoke의 마법
response = await llm.ainvoke(messages)이것이 내가 이전에 잘못하고 있던 부분이다. 나는 llm.invoke(messages)를 사용하고 있었는데, 이는 OpenAI가 응답할 때까지 모든 것을 차단한다. ainvoke를 사용하면, Python이 여러 요청을 동시에 처리할 수 있다.
흐름 구축
LangGraph 부분은 실제로 이해하면 꽤 직관적이다:
- 함수들을 "nodes"로 추가
- "edges"로 연결
- 시작과 끝 지점 정의
AI 워크플로우를 위한 레고 블록을 연결하는 것과 같다.
성능 차이는 엄청나다
async로 전환하기 전, 내 로그는 다음과 같았다:
- 요청 1: 3.2초
- 요청 2: 3.1초 (요청 1을 기다려야 함)
- 요청 3: 2.9초 (요청 2를 기다려야 함)
- 3개 요청 총합: ~9.2초
async로 전환한 후:
- 요청 1: 3.2초
- 요청 2: 3.1초 (즉시 시작)
- 요청 3: 2.9초 (역시 즉시 시작)
- 3개 요청 총합: ~3.2초 (기본적으로 가장 느린 것의 시간)
내 클라이언트가 50개의 동시 요청으로 테스트했을 때, 기존 버전은 약 2.5분이 걸렸을 것이다. 새 버전은? 약 4초.
내가 실수한 것들 (당신이 하지 않도록)
1. 동기와 비동기 코드 혼용
나는 계속해서 실수로 async 함수 내에서 일반 함수를 사용했다. 이는 성능을 죽인다:
# 이렇게 하지 마세요
async def bad_function():
result = some_slow_sync_function() # 이것이 모든 것을 차단합니다!
# 대신 이렇게 하세요
async def good_function():
result = await asyncio.to_thread(some_slow_sync_function)2. Rate Limit 처리하지 않기
OpenAI에는 rate limit이 있다. 갑자기 50개의 요청을 동시에 보내면, 빠르게 한계에 도달할 것이다. 지연과 재시도를 추가하는 것을 배웠다.
3. 오류 처리 잊기
async 코드에서는 하나의 실패한 요청이 주의하지 않으면 다른 요청들을 망칠 수 있다. 항상 ainvoke 호출을 try-catch 블록으로 감싸라.
현재 이것으로 구축하고 있는 것들
텍스트 분석기는 시작에 불과했다. 이제 이 패턴을 다음과 같은 용도로 사용하고 있다:
실시간으로 게시물을 처리하는 콘텐츠 조절 시스템. 각 게시물이 분류되고, 위반 사항을 확인하고, 적절한 팀으로 라우팅된다 — 모두 병렬로.
여러 대화를 동시에 처리할 수 있는 고객 지원 봇. 각각 고유한 컨텍스트와 메모리를 가진다.
수백 개의 PDF를 가져와서 추출, 분류, 요약을 모두 동시에 할 수 있는 문서 처리 파이프라인.
현장에서의 몇 가지 빠른 팁
간단하게 시작하라: 첫날부터 복잡한 워크플로우를 구축하려고 하지 마라. 먼저 기본적인 async 플로우를 작동시켜라.
리소스를 모니터링하라: Async는 수많은 동시 작업을 생성할 수 있다. 메모리와 API 사용량을 주시하라.
적절한 로깅을 사용하라: 모든 것이 동시에 일어날 때, 좋은 로그는 디버깅의 생명줄이다.
현실적인 부하로 테스트하라: 단순히 하나의 요청으로만 테스트하지 마라. 10개, 50개, 100개를 동시에 발사해서 무엇이 깨지는지 봐라.
결론
나는 대부분의 애플리케이션에 대해 async 프로그래밍이 과도하다고 생각했었다. 나는 틀렸다. 성능 향상은 무시하기엔 너무 좋다. 특히 느리고 예측 불가능할 수 있는 AI API와 작업할 때 더욱 그렇다.
한 번에 몇 명 이상의 사용자를 처리해야 하는 LangChain으로 무언가를 구축하고 있다면, 처음부터 async로 시작하는 것이 좋다. 미래의 당신 (그리고 당신의 사용자들)이 감사할 것이다.
위에서 공유한 코드는 당신이 실제로 복사해서 실행할 수 있는 작동하는 예제다. 여기서 시작해서, 망가뜨리고, 고치고, 멋진 것을 구축하라.
나를 믿어라. 응답 시간이 30초에서 3초로 떨어지는 것을 보면, 다시는 동기식 AI 워크플로우로 돌아가지 않을 것이다.
프로덕션에서 async LangChain을 사용해본 적이 있나? 댓글에서 당신의 경험을 듣고 싶다. 그리고 이 코드로 멋진 것을 구축한다면, 링크를 보내줘!
출처: Why I Switched to Async LangChain and LangGraph (And You Should Too)
