파이썬으로 비동기 데이터를 다루고 스트리밍 처리를 구현하는 것은 AI/LLM 개발자들에게 더 이상 선택이 아닌 필수가 됐습니다. 특히 LangGraph, LangChain 같은 프레임워크를 활용해 LLM Streaming Pipeline을 만들 때, Python의 Iterator, Generator 개념을 깊이 이해하는 것이 실무에서 체감 차이를 만들어줍니다.
이번 글에서는 Python의 Iterator, Generator 기본기를 다진 후, LangGraph의 generation_chain.astream과 LangChain의 runnable.astream이 어떻게 Streaming 처리를 수행하는지 실제 예제 코드를 통해 설명해보겠습니다.
Python Iterator와 Generator – 비동기 스트리밍의 뿌리
Python에서 ‘이터레이터’는 데이터 컬렉션을 하나씩 순차적으로 반환하는 객체입니다. __iter__()와 __next__() 메서드를 구현한 객체가 이터레이터로 정의됩니다. 리스트, 튜플도 반복문에서 순회할 수 있지만, 메모리를 덜 사용하면서 순차적 처리가 필요한 경우 Iterator와 Generator를 적극적으로 활용하게 됩니다.
예를 들어, 다음과 같이 Generator를 사용하면 데이터를 ‘요청하는 순간’ 하나씩 생성해 반환합니다.
def simple_generator():
for i in range(5):
yield i
gen = simple_generator()
for num in gen:
print(num)
여기서 중요한 포인트는 ‘yield’가 데이터를 반환하는 순간, 함수의 상태를 기억한 채 멈추고 있다는 것입니다. 다시 호출되면 그 지점부터 이어서 실행됩니다. 이는 한 번에 대량의 데이터를 로딩하지 않고, “스트림(Streaming)”처럼 데이터를 순차적으로 처리하는 데 핵심적인 역할을 합니다.
비동기 처리에서는 async def, async for, async yield와 같은 비동기 Generator를 활용할 수 있으며, 이는 LangGraph나 LangChain에서 Streaming 처리를 구현할 때 그대로 적용됩니다.
LangGraph astream – Node 기반 LLM Workflow 스트리밍
LangGraph는 LangChain 팀에서 만든 LLM Workflows 프레임워크로, 여러 Node를 그래프 형태로 연결하여 LLM 기반 Workflow를 정의할 수 있도록 해줍니다. 일반적인 LangChain이 ‘직선형 체인’에 가깝다면, LangGraph는 상태 관리와 분기처리가 강력한 ‘상태 머신’ 형태로 동작합니다.
LangGraph에서 generation_chain.astream() 메서드는 스트리밍 형태로 LLM의 응답을 받아오는 비동기 Generator입니다.
LangGraph는 내부적으로 Python의 비동기 Generator 패턴을 그대로 따르고 있습니다. astream() 메서드를 호출하면 Node들이 순차적으로 실행되며, 각 노드가 완료될 때마다 이벤트가 발생하고, 이 이벤트가 Generator를 통해 순차적으로 스트리밍됩니다.
예를 들어, LangGraph에서 Streaming Workflow를 구성하면 아래와 같은 구조로 동작합니다.
import asyncio
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
# 간단한 노드 정의
async def generate_response(state):
yield {"message": AIMessage(content="첫 번째 응답입니다.")}
yield {"message": AIMessage(content="두 번째 응답입니다.")}
# 상태 정의 및 그래프 구성
graph = StateGraph()
graph.add_node("Generator", generate_response)
graph.set_entry_point("Generator")
graph.set_finish_point(END)
# astream으로 스트리밍 처리
async def main():
async for event in graph.astream({"input": HumanMessage(content="안녕?")}):
print(event["message"].content)
asyncio.run(main())
위 코드에서 핵심은 async def generate_response()에서 yield로 응답을 생성하는 순간마다 astream이 그 값을 즉시 반환해준다는 점입니다. LangGraph 내부에서는 각 노드의 실행 상태와 결과를 Generator로 묶어 비동기 Streaming을 구현하고 있습니다.
따라서 LangGraph의 astream을 잘 활용하면, 단일 API 호출로 LLM의 복잡한 Workflow를 스트림 형태로 끊김 없이 받아올 수 있는 구조를 만들 수 있습니다.
LangChain Streaming astream – Runnable 기반 스트리밍 응답 처리
LangChain에서는 Runnable이라는 개념을 통해 체인(Chain), 프롬프트(Prompt), 모델 호출(LLM)을 하나의 실행 가능한 객체로 감쌀 수 있습니다. 이 Runnable 객체에서 astream() 메서드를 사용하면, LLM의 응답을 토큰 단위로 비동기 스트리밍할 수 있습니다.
즉, LangChain에서 Streaming astream은 ‘토큰 생성과 동시에 반환하는 비동기 Generator’로 동작합니다.
예제를 보겠습니다.
import asyncio
from langchain.chat_models import ChatOpenAI
llm = ChatOpenAI(streaming=True)
async def main():
async for chunk in llm.astream("당신은 누구인가요?"):
print(chunk.content)
asyncio.run(main())
여기서 중요한 점은 ChatOpenAI 클래스에 streaming=True 옵션을 설정해야 astream이 작동한다는 것입니다. LangChain 내부에서는 OpenAI API를 호출한 후, 응답 스트림을 Python의 비동기 Generator 형태로 래핑하여 astream()에서 한 번에 데이터를 반환하는 것이 아니라, 조금씩 토큰 단위로 나누어 반환하는 구조입니다.
이는 OpenAI API의 stream=True 옵션과 동일한 원리지만, LangChain은 이를 Python 비동기 제너레이터로 깔끔하게 추상화해줍니다.
실제 사용에서는 LangChain의 Runnable astream을 LangGraph와 결합해 워크플로우 전체를 스트리밍 형태로 처리하는 것도 가능합니다.
LangGraph와 LangChain Streaming의 실질적 차이점과 활용 전략
LangGraph와 LangChain은 모두 Streaming을 지원하지만, 구조적 차이점이 존재합니다.
LangChain의 astream은 단일 Chain 또는 LLM 호출의 응답을 토큰 단위로 스트리밍하는 것에 집중합니다. 따라서 짧은 요청-응답에서 실시간 피드백을 받고 싶을 때 매우 효과적입니다. 사용자는 ‘1개의 질문 → 1개의 스트리밍 응답’이라는 패턴으로 사용할 때 LangChain의 Streaming이 직관적입니다.
반면, LangGraph는 Node와 Edge(상태 전이)를 기반으로 여러 단계의 처리 로직을 설계할 수 있으며, 각 Node의 실행 결과를 스트리밍할 수 있다는 점에서 복잡한 Workflow에 적합합니다. 즉, ‘다단계 프로세스 → 각 단계별 스트리밍 응답’이라는 시나리오에 강합니다.
정리하자면 LangChain은 ‘짧고 빠른 스트리밍 응답’, LangGraph는 ‘복잡한 상태 기반 스트리밍 프로세스’에 적합합니다.
실무에서 Streaming을 구현할 때 핵심적으로 고려해야 할 것들
Streaming 처리를 한다는 것은 단순히 데이터를 나눠서 보여주는 문제가 아닙니다. 다음과 같은 요소를 반드시 고려해야 합니다.
첫째, 메모리와 리소스 최적화입니다. LangGraph나 LangChain을 사용할 때도 내부적으로 Python의 Generator 패턴을 그대로 따르고 있기 때문에, 메모리 효율은 높지만 전체 워크플로우를 설계할 때 리소스 최적화를 고려해야 합니다.
둘째, 사용자 경험(UX) 개선입니다. 스트리밍 처리를 잘 활용하면 사용자는 응답을 기다리는 동안 지루하지 않게 실시간 피드백을 받을 수 있습니다. 토큰 단위, 문장 단위, 노드별 응답 단위 등 다양한 Streaming 전략을 상황에 맞게 설계하는 것이 중요합니다.
셋째, 에러 핸들링과 중단점 처리입니다. 스트리밍 도중 오류가 발생하거나, 사용자가 응답을 중단했을 때의 처리를 설계하는 것이 실무에서는 필수적입니다. LangGraph에서는 상태 머신 형태로 중단점 처리를 유연하게 설계할 수 있으며, LangChain에서는 astream 메서드에서 예외 처리를 통해 핸들링할 수 있습니다.
결론 – Generator, LangGraph/LangChain Streaming 핵심
Python의 Iterator, Generator는 단순히 반복문에서 사용하는 것이 아니라, 현대적인 AI 시스템에서 실시간 데이터 스트림 처리의 핵심 개념으로 자리 잡았습니다. LangGraph의 astream은 복잡한 노드 기반 Workflow를 스트리밍으로 처리할 수 있게 해주며, LangChain의 astream은 LLM 응답을 토큰 단위로 빠르게 전달하는 데 최적화되어 있습니다.
이 두 가지를 적절히 활용하면, ChatGPT나 Claude 같은 글로벌 서비스 수준의 실시간 LLM 기반 인터페이스를 Python과 LangChain/LangGraph를 통해 구현할 수 있습니다.
이제 LLM 개발자는 단순한 API 호출을 넘어 “스트림 데이터의 흐름을 설계하는 역량”을 갖춰야 합니다.
파이썬 중급 이상의 고급 문법 완벽 가이드 (async 비동기, iterator, 추상메서드, callable, runnable, thread)