본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.
본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.
이미지 로딩 중...
AI Generated
2025. 12. 12. · 10 Views
LangGraph Streaming 완벽 가이드
LangGraph의 스트리밍 기능을 활용하여 실시간으로 그래프 실행 과정을 모니터링하는 방법을 다룹니다. values, updates, messages 등 다양한 스트리밍 모드를 실전 예제와 함께 설명합니다.
목차
1. values 모드 전체 상태
어느 날 김개발 씨가 LangGraph로 챗봇을 만들다가 이상한 점을 발견했습니다. 그래프가 실행되고 있는 것 같은데, 중간 과정을 전혀 볼 수 없는 거예요.
"최종 결과만 나오는데, 지금 뭘 하고 있는 건지 알 수가 없네요?" 선배 개발자 박시니어 씨가 미소를 지으며 다가왔습니다.
values 모드는 그래프가 실행되는 동안 각 노드가 완료될 때마다 전체 상태를 실시간으로 보여주는 스트리밍 방식입니다. 마치 여행 중 매 경유지마다 현재 위치와 짐의 전체 상태를 확인하는 것과 같습니다.
이를 통해 그래프의 전체 상태를 단계별로 추적할 수 있습니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
messages: list[str]
count: int
def add_message(state: State):
# 메시지를 추가하고 카운트를 증가시킵니다
return {"messages": state["messages"] + ["Hello"], "count": state["count"] + 1}
graph = StateGraph(State)
graph.add_node("process", add_message)
graph.add_edge(START, "process")
graph.add_edge("process", END)
app = graph.compile()
# values 모드로 스트리밍
for chunk in app.stream({"messages": [], "count": 0}, stream_mode="values"):
print(chunk) # 전체 상태가 출력됩니다
김개발 씨는 입사 3개월 차 주니어 개발자입니다. 오늘도 열심히 LangGraph로 AI 에이전트를 만들고 있었는데, 실행 결과만 덩그러니 나타났습니다.
"뭔가 돌아가고 있긴 한데, 중간에 뭘 하는지 도통 모르겠어요." 선배 개발자 박시니어 씨가 코드를 살펴보더니 말했습니다. "아, 스트리밍을 안 써서 그래요.
values 모드를 한번 써보세요." 스트리밍이란 정확히 무엇일까요? 쉽게 비유하자면, 스트리밍은 마치 택배 배송 과정을 실시간으로 추적하는 것과 같습니다.
택배가 어디에 있는지, 어떤 상태인지 단계별로 확인할 수 있죠. 물건이 집에 도착할 때까지 기다리기만 하는 게 아니라요.
이처럼 LangGraph의 스트리밍도 그래프가 실행되는 동안 중간 과정을 실시간으로 보여주는 역할을 합니다. 스트리밍이 없던 시절에는 어땠을까요?
개발자들은 그래프가 끝날 때까지 기다렸다가 최종 결과만 확인할 수 있었습니다. 디버깅할 때 정말 답답했죠.
어디서 문제가 생긴 건지, 지금 무슨 작업을 하고 있는 건지 전혀 알 수 없었습니다. 더 큰 문제는 사용자 경험이었습니다.
사용자는 화면이 멈춘 것처럼 느껴져서 불안해했습니다. 바로 이런 문제를 해결하기 위해 스트리밍 모드가 등장했습니다.
스트리밍을 사용하면 그래프 실행 과정을 실시간으로 모니터링할 수 있습니다. 또한 사용자에게 진행 상황을 보여줄 수도 있습니다.
무엇보다 디버깅이 훨씬 쉬워진다는 큰 이점이 있습니다. values 모드는 가장 기본적인 스트리밍 방식입니다.
이 모드를 사용하면 각 노드가 실행을 완료할 때마다 그래프의 전체 상태가 출력됩니다. 마치 게임을 하면서 매 스테이지마다 현재 캐릭터의 모든 능력치를 확인하는 것과 같습니다.
어떤 스킬을 얻었는지, 체력은 얼마나 남았는지, 아이템은 무엇을 가지고 있는지 전부 볼 수 있죠. 위의 코드를 한 줄씩 살펴보겠습니다.
먼저 StateGraph를 사용해서 상태를 관리하는 그래프를 만들었습니다. 이 부분이 핵심입니다.
다음으로 stream_mode="values" 파라미터를 전달하면서 스트리밍을 시작합니다. 그러면 for 루프가 돌면서 각 노드가 완료될 때마다 chunk를 하나씩 받아옵니다.
마지막으로 각 chunk에는 그 시점의 전체 상태가 담겨 있습니다. 실제 현업에서는 어떻게 활용할까요?
예를 들어 고객 상담 AI 챗봇을 개발한다고 가정해봅시다. 사용자가 질문을 하면 여러 단계를 거쳐서 답변이 생성되는데요.
각 단계마다 전체 대화 히스토리와 현재 상태를 확인하고 싶을 때 values 모드를 활용하면 매우 유용합니다. 많은 기업에서 이런 패턴을 적극적으로 사용하고 있습니다.
하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 상태가 매우 클 때도 무조건 values 모드를 쓰는 것입니다.
이렇게 하면 네트워크 트래픽이 급격히 증가하고 성능 저하가 발생할 수 있습니다. 따라서 상태가 큰 경우에는 필요한 부분만 추출하거나 다른 스트리밍 모드를 고려해야 합니다.
다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 고개를 끄덕였습니다.
"아, 그래서 실시간으로 상태를 볼 수 있었군요!" values 모드를 제대로 이해하면 그래프의 실행 과정을 명확하게 파악할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.
실전 팁
💡 - 상태 객체가 클 때는 성능에 주의하세요
- 디버깅할 때 가장 유용한 모드입니다
- 각 노드 완료 시점마다 상태가 출력된다는 점을 기억하세요
2. updates 모드 변경사항만
며칠 후, 김개발 씨는 또 다른 문제에 부딪혔습니다. values 모드로 스트리밍을 하니 너무 많은 데이터가 쏟아져 나왔습니다.
"전체 상태가 아니라 방금 바뀐 것만 볼 수는 없을까요?" 박시니어 씨가 웃으며 대답했습니다. "그럴 땐 updates 모드를 쓰면 되죠."
updates 모드는 각 노드가 실행될 때 변경된 부분만 반환하는 효율적인 스트리밍 방식입니다. 마치 문서 편집 이력에서 전체 문서가 아니라 수정된 부분만 하이라이트로 보여주는 것과 같습니다.
이를 통해 네트워크 트래픽을 줄이고 필요한 정보만 빠르게 확인할 수 있습니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
counter: int
log: list[str]
def increment(state: State):
# 카운터만 증가시킵니다 (log는 변경하지 않음)
return {"counter": state["counter"] + 1}
def add_log(state: State):
# 로그만 추가합니다 (counter는 변경하지 않음)
return {"log": state["log"] + [f"Count: {state['counter']}"]}
graph = StateGraph(State)
graph.add_node("increment", increment)
graph.add_node("log", add_log)
graph.add_edge(START, "increment")
graph.add_edge("increment", "log")
graph.add_edge("log", END)
app = graph.compile()
# updates 모드로 스트리밍 - 변경된 부분만 출력됩니다
for chunk in app.stream({"counter": 0, "log": []}, stream_mode="updates"):
print(chunk) # 예: {"increment": {"counter": 1}}
김개발 씨는 values 모드를 써보고 나서 만족스러워했지만, 곧 새로운 고민이 생겼습니다. 그래프의 상태가 점점 커지면서 매번 전체 상태를 받아오니 화면이 너무 복잡했습니다.
"꼭 필요한 정보만 볼 수는 없을까요?" 박시니어 씨가 코드를 가리키며 설명했습니다. "updates 모드를 쓰면 돼요.
바뀐 것만 딱 보여주거든요." updates 모드란 정확히 무엇일까요? 쉽게 비유하자면, updates 모드는 마치 뉴스 속보와 같습니다.
하루 종일 일어난 모든 일을 다 말해주는 게 아니라, 방금 막 일어난 새로운 소식만 알려주죠. 신문을 처음부터 끝까지 다시 읽을 필요 없이, 업데이트된 기사만 확인하면 됩니다.
이처럼 updates 모드도 변경된 부분만 골라서 보여주는 역할을 합니다. 전체 상태를 매번 받는 것의 문제는 무엇일까요?
상태 객체가 크면 클수록 네트워크를 통해 전송되는 데이터 양이 기하급수적으로 늘어납니다. 실시간 스트리밍인데도 느려지고, 사용자는 답답함을 느낍니다.
더 큰 문제는 로그를 분석할 때입니다. 실제로 변경된 게 무엇인지 찾으려면 이전 상태와 일일이 비교해야 하니까요.
바로 이런 문제를 해결하기 위해 updates 모드가 만들어졌습니다. updates 모드를 사용하면 각 노드가 실제로 변경한 부분만 반환됩니다.
또한 불필요한 데이터 전송이 사라지니 성능이 크게 개선됩니다. 무엇보다 디버깅할 때 어떤 노드가 무엇을 바꿨는지 한눈에 보인다는 큰 이점이 있습니다.
chunk의 구조를 이해하는 것이 중요합니다. updates 모드에서는 chunk가 딕셔너리 형태로 반환됩니다.
키는 노드 이름이고, 값은 그 노드가 반환한 업데이트 내용입니다. 예를 들어 {"increment": {"counter": 1}} 이라는 chunk를 받았다면, increment 노드가 counter를 1로 변경했다는 뜻입니다.
위의 코드를 한 줄씩 살펴보겠습니다. 먼저 increment 노드는 counter만 증가시키고 log는 건드리지 않습니다.
이 부분이 핵심입니다. 다음으로 add_log 노드는 log만 추가하고 counter는 그대로 둡니다.
stream_mode를 "updates"로 설정하면, 각 노드가 변경한 필드만 chunk로 받아올 수 있습니다. 마지막으로 for 루프를 돌면서 변경사항만 깔끔하게 출력됩니다.
실제 현업에서는 어떻게 활용할까요? 예를 들어 문서 요약 서비스를 개발한다고 가정해봅시다.
원본 문서는 매우 크지만, 각 단계마다 요약본만 조금씩 추가되는 상황입니다. 이럴 때 updates 모드를 활용하면 원본 문서를 매번 전송하지 않고 새로 생성된 요약 부분만 받아올 수 있습니다.
실시간 협업 툴에서도 이런 패턴이 널리 쓰입니다. 하지만 주의할 점도 있습니다.
초보 개발자들이 흔히 하는 실수 중 하나는 전체 상태가 필요한 시점에도 updates만 보는 것입니다. 이렇게 하면 현재 전체 상태를 알 수 없어서 로직 에러가 발생할 수 있습니다.
따라서 용도에 맞게 values와 updates를 적절히 선택해야 합니다. 김개발 씨는 updates 모드를 적용해보고 감탄했습니다.
"와, 훨씬 깔끔하네요! 뭐가 바뀌었는지 딱 보여요." updates 모드를 제대로 이해하면 효율적이고 명확한 스트리밍을 구현할 수 있습니다.
여러분도 프로젝트에서 변경사항만 추적하고 싶을 때 이 모드를 활용해 보세요.
실전 팁
💡 - 상태가 큰 그래프에서 특히 유용합니다
- 각 노드가 무엇을 변경했는지 명확히 알 수 있습니다
- 전체 상태가 필요한 경우에는 values 모드와 함께 사용하세요
3. custom 모드 사용자 정의
김개발 씨는 점점 욕심이 생겼습니다. "values는 너무 많고, updates는 충분하지 않을 때가 있어요.
제가 원하는 정보만 골라서 볼 수는 없을까요?" 박시니어 씨가 고개를 끄덕이며 말했습니다. "그럴 땐 custom 모드를 만들면 됩니다."
custom 모드는 개발자가 직접 원하는 정보만 추출하여 스트리밍할 수 있는 맞춤형 방식입니다. 마치 뷔페에서 먹고 싶은 음식만 골라 담는 것처럼, 필요한 데이터만 선택적으로 반환할 수 있습니다.
이를 통해 완벽하게 커스터마이징된 모니터링이 가능합니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
input: str
output: str
metadata: dict
def process(state: State):
# 처리 결과를 반환하면서 custom 정보도 함께 전송합니다
result = state["input"].upper()
# writer 함수를 통해 custom 데이터를 스트리밍합니다
return {"output": result, "metadata": {"length": len(result)}}
graph = StateGraph(State)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)
# custom writer 함수 정의
def custom_writer(node_name, update):
# 원하는 정보만 선택적으로 반환합니다
if "metadata" in update:
return {"node": node_name, "length": update["metadata"]["length"]}
return None
app = graph.compile()
# custom 모드로 스트리밍
for chunk in app.stream({"input": "hello", "output": "", "metadata": {}},
stream_mode="custom"):
print(chunk)
김개발 씨는 values와 updates 모드를 자유자재로 쓸 수 있게 되었지만, 여전히 아쉬운 점이 있었습니다. "특정 필드의 길이만 보고 싶은데, 전체 값을 받아야 하네요.
제가 정말 필요한 건 통계 정보인데 말이죠." 박시니어 씨가 웃으며 대답했습니다. "그럴 땐 custom 모드를 직접 만들면 돼요.
원하는 대로 커스터마이징할 수 있거든요." custom 모드란 정확히 무엇일까요? 쉽게 비유하자면, custom 모드는 마치 커피숍에서 나만의 레시피로 음료를 주문하는 것과 같습니다.
메뉴판에 있는 그대로가 아니라, 샷 추가, 시럽 빼기, 우유 종류 변경 등 원하는 대로 조합할 수 있죠. 이처럼 custom 모드도 개발자가 직접 어떤 정보를 어떤 형태로 스트리밍할지 결정할 수 있게 해줍니다.
기본 스트리밍 모드의 한계는 무엇일까요? values는 모든 정보를 주지만 때로는 과합니다.
updates는 변경사항만 주지만 때로는 부족합니다. 실무에서는 "이 필드의 길이", "저 리스트의 개수", "특정 조건을 만족하는 아이템만" 같은 세밀한 요구사항이 생기기 마련입니다.
기존 모드로는 이런 니즈를 충족하기 어려웠습니다. 바로 이런 문제를 해결하기 위해 custom 모드가 존재합니다.
custom 모드를 사용하면 완전히 자유로운 형태로 데이터를 추출할 수 있습니다. 또한 성능 최적화를 위해 꼭 필요한 연산만 수행할 수도 있습니다.
무엇보다 비즈니스 로직에 딱 맞는 모니터링을 구현할 수 있다는 큰 이점이 있습니다. writer 함수가 핵심입니다.
LangGraph에서는 각 노드가 실행될 때 writer 함수를 호출할 수 있습니다. 이 함수는 노드 이름과 업데이트 내용을 받아서 원하는 정보를 가공하여 반환합니다.
예를 들어 메타데이터의 길이만 추출하거나, 특정 조건을 만족하는 경우에만 데이터를 반환할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.
먼저 custom_writer 함수를 정의합니다. 이 부분이 핵심입니다.
이 함수는 node_name과 update를 인자로 받아서 원하는 형태로 가공합니다. 다음으로 metadata가 있을 때만 노드 이름과 길이 정보를 반환하도록 조건을 걸었습니다.
stream_mode를 "custom"으로 설정하면 writer 함수를 통과한 데이터만 chunk로 받게 됩니다. 마지막으로 필요한 정보만 깔끔하게 스트리밍됩니다.
실제 현업에서는 어떻게 활용할까요? 예를 들어 대용량 데이터 처리 파이프라인을 개발한다고 가정해봅시다.
각 단계에서 처리된 레코드 수, 에러율, 평균 처리 시간 같은 통계만 실시간으로 보고 싶습니다. 원본 데이터를 전부 스트리밍하면 네트워크가 마비될 테니까요.
이럴 때 custom 모드로 통계 정보만 추출하면 완벽한 모니터링 대시보드를 만들 수 있습니다. 하지만 주의할 점도 있습니다.
초보 개발자들이 흔히 하는 실수 중 하나는 writer 함수에서 복잡한 로직을 넣는 것입니다. 이렇게 하면 스트리밍 자체가 병목이 되어 성능이 저하될 수 있습니다.
따라서 writer 함수는 최대한 가볍게 유지하고, 단순한 데이터 추출과 변환만 수행해야 합니다. 김개발 씨는 custom 모드로 자신만의 모니터링 시스템을 만들어보고 뿌듯해했습니다.
"이제 정말 원하는 정보만 딱딱 볼 수 있네요!" custom 모드를 제대로 이해하면 프로젝트에 최적화된 스트리밍을 구현할 수 있습니다. 여러분도 특별한 요구사항이 있을 때 이 모드를 적극 활용해 보세요.
실전 팁
💡 - writer 함수는 가볍게 유지하세요
- 통계나 메타데이터만 추출할 때 매우 유용합니다
- 비즈니스 로직에 맞는 맞춤형 모니터링이 가능합니다
4. messages 모드 LLM 토큰
어느 날 김개발 씨는 LLM을 사용하는 그래프를 만들었습니다. 그런데 답변이 한 번에 쭉 나오는 게 아니라, 사람이 타이핑하듯이 한 글자씩 나오게 하고 싶었습니다.
"ChatGPT처럼 스트리밍으로 출력하려면 어떻게 해야 하죠?" 박시니어 씨가 반가운 표정으로 말했습니다. "messages 모드가 딱이네요!"
messages 모드는 LLM이 생성하는 토큰을 실시간으로 스트리밍하여 자연스러운 대화 경험을 제공하는 방식입니다. 마치 친구가 카카오톡으로 긴 메시지를 보낼 때 타이핑하는 모습을 실시간으로 보는 것처럼, AI의 사고 과정을 그대로 보여줄 수 있습니다.
이를 통해 사용자는 기다림 없이 즉각적인 피드백을 받을 수 있습니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict
class State(TypedDict):
messages: list
def call_llm(state: State):
# LLM을 호출하여 응답을 생성합니다
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", streaming=True)
response = llm.invoke(state["messages"])
return {"messages": state["messages"] + [response]}
graph = StateGraph(State)
graph.add_node("llm", call_llm)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)
app = graph.compile()
# messages 모드로 스트리밍 - 토큰이 실시간으로 출력됩니다
for chunk in app.stream(
{"messages": [{"role": "user", "content": "안녕하세요"}]},
stream_mode="messages"
):
print(chunk, end="", flush=True)
김개발 씨는 이제 LangGraph로 AI 챗봇을 만들 수 있게 되었습니다. 그런데 사용자 테스트를 해보니 불만이 쏟아졌습니다.
"답변이 나올 때까지 너무 오래 기다려야 해요. 중간에 멈춘 건지 아닌지도 모르겠고요." 박시니어 씨가 화면을 보더니 바로 문제를 짚어냈습니다.
"LLM 응답을 스트리밍으로 보여줘야 해요. messages 모드를 쓰면 됩니다." messages 모드란 정확히 무엇일까요?
쉽게 비유하자면, messages 모드는 마치 요리사가 요리하는 과정을 라이브로 보여주는 쿠킹쇼와 같습니다. 완성된 요리만 덩그러니 내놓는 게 아니라, 재료를 손질하고, 볶고, 양념을 넣는 모든 과정을 실시간으로 보여주죠.
보는 사람은 지루하지 않고, 다음에 무엇이 나올지 기대하게 됩니다. 이처럼 messages 모드도 LLM이 토큰을 생성하는 과정을 실시간으로 보여줍니다.
응답을 한 번에 받는 것의 문제는 무엇일까요? LLM이 긴 답변을 생성할 때는 시간이 꽤 걸립니다.
사용자는 멈춘 화면만 보면서 불안하게 기다려야 합니다. "지금 작동하고 있는 건가?
에러가 난 건가?" 하는 의문이 들죠. 더 큰 문제는 사용자 경험입니다.
현대의 AI 서비스는 대부분 스트리밍을 지원하는데, 그렇지 않으면 구식처럼 느껴집니다. 바로 이런 문제를 해결하기 위해 messages 모드가 필수적입니다.
messages 모드를 사용하면 LLM이 토큰을 생성하는 즉시 사용자에게 보여줄 수 있습니다. 또한 사용자는 답변이 생성되고 있음을 실시간으로 확인하며 안심합니다.
무엇보다 ChatGPT나 Claude 같은 최신 AI 서비스와 동일한 경험을 제공할 수 있다는 큰 이점이 있습니다. 토큰 스트리밍의 원리를 이해해야 합니다.
LLM은 한 번에 하나의 토큰(단어나 글자 조각)을 생성합니다. messages 모드에서는 각 토큰이 생성될 때마다 chunk로 반환됩니다.
프론트엔드에서는 이 chunk들을 받아서 화면에 점진적으로 표시하면 됩니다. 마치 타자기가 한 글자씩 찍히는 것처럼 보이게 할 수 있습니다.
위의 코드를 한 줄씩 살펴보겠습니다. 먼저 ChatAnthropic을 생성할 때 streaming=True 옵션을 켭니다.
이 부분이 핵심입니다. 이렇게 하면 LLM이 토큰을 생성할 때마다 yield하게 됩니다.
다음으로 stream_mode를 "messages"로 설정하면 각 토큰이 chunk로 전달됩니다. end="" 와 flush=True를 사용하면 새 줄 없이 연속으로 출력됩니다.
마지막으로 사용자는 AI가 생각하는 과정을 실시간으로 보게 됩니다. 실제 현업에서는 어떻게 활용할까요?
예를 들어 법률 자문 AI 서비스를 개발한다고 가정해봅시다. 사용자가 복잡한 법률 질문을 하면 LLM이 긴 답변을 생성하는데, 이때 스트리밍으로 보여주면 사용자는 첫 문장부터 읽기 시작할 수 있습니다.
답변이 완전히 생성될 때까지 기다릴 필요가 없죠. 실제로 많은 AI 서비스가 이런 방식을 채택하고 있습니다.
하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 네트워크 연결이 불안정한 환경을 고려하지 않는 것입니다.
스트리밍 중에 연결이 끊어지면 사용자는 불완전한 답변만 보게 됩니다. 따라서 에러 처리와 재연결 로직을 반드시 구현해야 합니다.
김개발 씨는 messages 모드를 적용하고 사용자 피드백을 받아봤습니다. "이제 ChatGPT처럼 자연스럽게 나오네요!
훨씬 좋아요!" messages 모드를 제대로 이해하면 현대적인 AI 채팅 경험을 구현할 수 있습니다. 여러분도 LLM을 사용하는 애플리케이션에 이 모드를 꼭 적용해 보세요.
실전 팁
💡 - LLM 초기화 시 streaming=True를 반드시 설정하세요
- 네트워크 에러 처리를 잊지 마세요
- 사용자 경험이 극적으로 개선됩니다
5. debug 모드 상세 정보
김개발 씨는 복잡한 그래프를 만들다가 예상치 못한 버그를 만났습니다. "어느 노드에서 문제가 생긴 건지, 실행 순서는 맞는 건지 도무지 알 수가 없어요." 박시니어 씨가 디버깅 팁을 알려주었습니다.
"debug 모드를 써보세요. 모든 게 다 보일 거예요."
debug 모드는 그래프 실행의 모든 세부 정보를 상세하게 보여주는 디버깅 전용 스트리밍 방식입니다. 마치 자동차 정비소에서 엔진 내부를 진단할 때 모든 센서 데이터를 확인하는 것처럼, 노드 실행 순서, 소요 시간, 에러 정보 등을 종합적으로 제공합니다.
이를 통해 복잡한 버그도 빠르게 찾아낼 수 있습니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
import time
class State(TypedDict):
value: int
def slow_process(state: State):
# 의도적으로 느린 처리를 시뮬레이션합니다
time.sleep(0.5)
return {"value": state["value"] * 2}
def fast_process(state: State):
# 빠른 처리
return {"value": state["value"] + 10}
graph = StateGraph(State)
graph.add_node("slow", slow_process)
graph.add_node("fast", fast_process)
graph.add_edge(START, "slow")
graph.add_edge("slow", "fast")
graph.add_edge("fast", END)
app = graph.compile()
# debug 모드로 스트리밍 - 실행 세부 정보가 모두 출력됩니다
for chunk in app.stream({"value": 5}, stream_mode="debug"):
print(chunk) # 노드명, 실행 시간, 입출력 등 상세 정보
김개발 씨는 어느덧 LangGraph 중급자가 되어 복잡한 멀티 에이전트 시스템을 구축하고 있었습니다. 그런데 갑자기 이상한 결과가 나오기 시작했습니다.
"분명 로직은 맞는데 왜 이런 값이 나오지? 어디서부터 잘못된 거야?" 박시니어 씨가 모니터를 보더니 조언했습니다.
"복잡한 그래프는 debug 모드 없이는 디버깅이 거의 불가능해요. 한번 켜보세요." debug 모드란 정확히 무엇일까요?
쉽게 비유하자면, debug 모드는 마치 비행기의 블랙박스와 같습니다. 비행기가 비행하는 동안 모든 센서 데이터, 조종사의 조작, 시스템 상태 등을 빠짐없이 기록하죠.
사고가 나면 이 블랙박스를 분석해서 정확한 원인을 찾아냅니다. 이처럼 debug 모드도 그래프가 실행되는 동안 모든 정보를 상세하게 기록하고 보여줍니다.
일반 스트리밍 모드의 한계는 무엇일까요? values나 updates는 상태 변화만 보여줍니다.
하지만 실제 디버깅에서는 더 많은 정보가 필요합니다. 각 노드가 언제 시작되고 끝났는지, 얼마나 시간이 걸렸는지, 어떤 순서로 실행되었는지, 에러가 발생했다면 정확히 어디서 발생했는지 등등.
이런 정보 없이는 복잡한 버그를 추적하기 어렵습니다. 바로 이런 문제를 해결하기 위해 debug 모드가 제공됩니다.
debug 모드를 사용하면 그래프 실행의 모든 단계를 투명하게 볼 수 있습니다. 또한 성능 병목 지점을 정확히 파악할 수 있습니다.
무엇보다 복잡한 실행 흐름에서도 버그의 정확한 위치를 찾아낼 수 있다는 큰 이점이 있습니다. debug 정보의 구조를 이해해야 합니다.
debug 모드에서는 chunk에 다양한 메타데이터가 포함됩니다. 노드 이름은 물론이고, 실행 시작 시각, 종료 시각, 소요 시간, 입력 상태, 출력 상태, 에러 정보 등이 모두 담겨 있습니다.
이 정보들을 종합하면 전체 실행 타임라인을 재구성할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.
먼저 slow_process와 fast_process 두 노드를 만들었는데, 처리 속도가 다릅니다. 이 부분이 핵심입니다.
debug 모드에서는 각 노드의 소요 시간을 측정해서 보여줍니다. stream_mode를 "debug"로 설정하면 실행 세부 정보가 모두 chunk로 전달됩니다.
마지막으로 이 정보를 분석하면 어느 노드가 느린지, 실행 순서가 맞는지 등을 파악할 수 있습니다. 실제 현업에서는 어떻게 활용할까요?
예를 들어 복잡한 데이터 파이프라인을 운영한다고 가정해봅시다. 갑자기 전체 처리 시간이 2배로 늘어났다면, debug 모드로 각 단계의 소요 시간을 측정하여 병목 지점을 찾아낼 수 있습니다.
또한 특정 조건에서만 발생하는 간헐적 버그를 추적할 때도 실행 흐름을 정확히 파악할 수 있어 매우 유용합니다. 하지만 주의할 점도 있습니다.
초보 개발자들이 흔히 하는 실수 중 하나는 프로덕션 환경에서도 debug 모드를 켜놓는 것입니다. 이렇게 하면 로그가 폭발적으로 늘어나고 성능도 저하됩니다.
따라서 debug 모드는 개발 환경이나 문제 해결 시에만 사용하고, 운영 환경에서는 반드시 꺼야 합니다. 김개발 씨는 debug 모드로 버그를 빠르게 찾아내고 감탄했습니다.
"와, 이제 보니 세 번째 노드에서 순서가 꼬였네요. 이거 없었으면 며칠 걸렸을 거예요!" debug 모드를 제대로 이해하면 복잡한 그래프도 자신 있게 디버깅할 수 있습니다.
여러분도 문제 해결이 막막할 때 이 모드를 적극 활용해 보세요.
실전 팁
💡 - 프로덕션 환경에서는 절대 사용하지 마세요
- 성능 병목을 찾을 때 매우 유용합니다
- 로그를 파일로 저장해두면 나중에 분석하기 좋습니다
6. 다중 모드 동시 사용
어느 날 김개발 씨는 궁금증이 생겼습니다. "values로 전체 상태도 보고 싶고, debug로 실행 시간도 보고 싶은데, 두 개를 동시에 쓸 수는 없나요?" 박시니어 씨가 웃으며 대답했습니다.
"당연히 되죠! 리스트로 여러 모드를 동시에 사용할 수 있어요."
다중 모드는 여러 스트리밍 모드를 동시에 활성화하여 각 모드의 장점을 모두 활용하는 고급 방식입니다. 마치 스마트폰으로 사진을 찍으면서 동시에 동영상도 녹화하는 것처럼, 하나의 실행으로 다양한 관점의 정보를 얻을 수 있습니다.
이를 통해 풍부한 모니터링과 유연한 디버깅이 가능합니다.
다음 코드를 살펴봅시다.
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
counter: int
messages: list[str]
def process(state: State):
# 상태를 업데이트합니다
return {
"counter": state["counter"] + 1,
"messages": state["messages"] + [f"Step {state['counter']}"]
}
graph = StateGraph(State)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)
app = graph.compile()
# 여러 모드를 동시에 사용합니다
for chunk in app.stream(
{"counter": 0, "messages": []},
stream_mode=["values", "updates", "debug"]
):
# chunk의 타입에 따라 다르게 처리할 수 있습니다
if "type" in chunk and chunk["type"] == "values":
print(f"[VALUES] {chunk['data']}")
elif "type" in chunk and chunk["type"] == "updates":
print(f"[UPDATES] {chunk['data']}")
elif "type" in chunk and chunk["type"] == "debug":
print(f"[DEBUG] {chunk['data']}")
김개발 씨는 이제 각 스트리밍 모드의 특징을 잘 이해하게 되었습니다. 하지만 새로운 고민이 생겼습니다.
"실무에서는 상태도 보고 싶고, 디버그 정보도 필요한데, 매번 모드를 바꿔가며 실행하기엔 너무 번거로워요." 박시니어 씨가 코드를 수정하며 말했습니다. "그럴 땐 여러 모드를 동시에 쓰면 됩니다.
리스트로 전달하기만 하면 돼요." 다중 모드란 정확히 무엇일까요? 쉽게 비유하자면, 다중 모드는 마치 자동차의 계기판과 같습니다.
계기판에는 속도계, 회전계, 연료계, 온도계 등이 동시에 표시되죠. 운전자는 한눈에 여러 정보를 파악할 수 있습니다.
이처럼 다중 모드도 하나의 그래프 실행으로 여러 관점의 정보를 동시에 얻을 수 있게 해줍니다. 모드를 하나만 쓰는 것의 한계는 무엇일까요?
values만 쓰면 전체 상태는 보이지만 성능 정보는 없습니다. debug만 쓰면 상세 정보는 많지만 정작 상태 변화를 추적하기 어렵습니다.
실무에서는 보통 여러 정보가 동시에 필요합니다. 그래서 모드를 바꿔가며 여러 번 실행하게 되는데, 이는 비효율적이고 시간 낭비입니다.
바로 이런 문제를 해결하기 위해 다중 모드 기능이 제공됩니다. 다중 모드를 사용하면 한 번의 실행으로 모든 필요한 정보를 얻을 수 있습니다.
또한 각 정보를 목적에 맞게 따로 처리할 수 있습니다. 무엇보다 개발 시간을 크게 절약할 수 있다는 큰 이점이 있습니다.
chunk 타입 구분이 중요합니다. 다중 모드를 사용하면 chunk에 type 필드가 추가됩니다.
이 필드를 확인하면 어느 모드에서 온 데이터인지 알 수 있습니다. 예를 들어 values 모드의 chunk는 type이 "values"이고, debug 모드의 chunk는 type이 "debug"입니다.
이를 기반으로 조건문을 작성하여 각 데이터를 적절히 처리할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.
먼저 stream_mode 파라미터에 리스트를 전달합니다. 이 부분이 핵심입니다.
["values", "updates", "debug"] 이렇게 세 가지 모드를 동시에 활성화할 수 있습니다. 다음으로 for 루프에서 받는 chunk는 세 모드 중 하나에서 온 것입니다.
**chunk["type"]**을 확인하여 어느 모드인지 판별합니다. 마지막으로 각 타입에 맞는 처리 로직을 분기하여 실행합니다.
실제 현업에서는 어떻게 활용할까요? 예를 들어 실시간 모니터링 대시보드를 개발한다고 가정해봅시다.
한쪽 패널에는 values 정보로 현재 상태를 표시하고, 다른 패널에는 debug 정보로 성능 차트를 그리고, 또 다른 패널에는 updates 정보로 변경 이력을 보여줄 수 있습니다. 하나의 스트리밍으로 세 개의 뷰를 동시에 업데이트하는 거죠.
하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 필요 없는 모드까지 전부 켜놓는 것입니다.
이렇게 하면 데이터가 너무 많이 생성되어 오히려 혼란스럽고 성능도 저하됩니다. 따라서 정말 필요한 모드만 선택적으로 사용해야 합니다.
모드 조합 전략을 세우는 것이 중요합니다. 개발 단계에서는 values + debug 조합이 유용합니다.
상태 변화와 성능을 동시에 볼 수 있으니까요. 프로덕션 모니터링에서는 updates + custom 조합을 고려할 수 있습니다.
변경사항과 비즈니스 메트릭만 추적하면 되니까요. 디버깅 시에는 모든 모드를 켜서 최대한 많은 정보를 수집합니다.
김개발 씨는 다중 모드로 대시보드를 만들어보고 놀라워했습니다. "한 번의 실행으로 이렇게 많은 정보를 볼 수 있다니!
완전 편한데요?" 다중 모드를 제대로 이해하면 효율적이고 강력한 모니터링 시스템을 구축할 수 있습니다. 여러분도 프로젝트에서 여러 정보가 필요할 때 이 기능을 적극 활용해 보세요.
박시니어 씨가 마지막으로 조언했습니다. "스트리밍은 단순히 데이터를 보는 게 아니라, 그래프와 대화하는 창구예요.
목적에 맞는 모드를 선택하면 개발이 훨씬 즐거워질 거예요." 김개발 씨는 이제 자신감이 생겼습니다. "이제 어떤 복잡한 그래프도 척척 모니터링할 수 있을 것 같아요!" 스트리밍은 LangGraph를 제대로 활용하기 위한 필수 기술입니다.
각 모드의 특징을 이해하고 상황에 맞게 사용하면, 더 나은 AI 애플리케이션을 만들 수 있습니다.
실전 팁
💡 - 필요한 모드만 선택적으로 사용하세요
- chunk의 type 필드로 모드를 구분할 수 있습니다
- 개발/운영 환경에 따라 다른 모드 조합을 사용하세요
- 너무 많은 모드를 켜면 오히려 혼란스러울 수 있습니다
이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!
댓글 (0)
함께 보면 좋은 카드 뉴스
Helm 마이크로서비스 패키징 완벽 가이드
Kubernetes 환경에서 마이크로서비스를 효율적으로 패키징하고 배포하는 Helm의 핵심 기능을 실무 중심으로 학습합니다. Chart 생성부터 릴리스 관리까지 체계적으로 다룹니다.
보안 아키텍처 구성 완벽 가이드
프로젝트의 보안을 처음부터 설계하는 방법을 배웁니다. AWS 환경에서 VPC부터 WAF, 암호화, 접근 제어까지 실무에서 바로 적용할 수 있는 보안 아키텍처를 단계별로 구성해봅니다.
AWS Organizations 완벽 가이드
여러 AWS 계정을 체계적으로 관리하고 통합 결제와 보안 정책을 적용하는 방법을 실무 스토리로 쉽게 배워봅니다. 초보 개발자도 바로 이해할 수 있는 친절한 설명과 실전 예제를 제공합니다.
AWS KMS 암호화 완벽 가이드
AWS KMS(Key Management Service)를 활용한 클라우드 데이터 암호화 방법을 초급 개발자를 위해 쉽게 설명합니다. CMK 생성부터 S3, EBS 암호화, 봉투 암호화까지 실무에 필요한 모든 내용을 담았습니다.
AWS Secrets Manager 완벽 가이드
AWS에서 데이터베이스 비밀번호, API 키 등 민감한 정보를 안전하게 관리하는 Secrets Manager의 핵심 개념과 실무 활용법을 배워봅니다. 초급 개발자도 쉽게 따라할 수 있도록 실전 예제와 함께 설명합니다.