LangGraph는 LLM(대형 언어 모델) 기반의 애플리케이션을 그래프 구조로 설계하고 실행할 수 있게 해주는 프레임워크다.
LangChain 위에 올려진 라이브러리로,
상태(State)를 노드(Node) 간에 주고받는 방식으로 복잡한 AI 워크플로우를 구성한다고 이해하면 좋다.
우리가 만들고 있는 서비스 구조에 활용해 보고자 공부한 내용을 정리하려 한다.
1. Intro 🤔
우리는 위험도 예측에서 끝나는 것이 아닌
그 결과를 바탕으로 개인별 맞춤 권고사항을 출력해주는 서비스를 구현하고 싶었다.
사용자의 건강 데이터를 받아서
→ ML 모델로 질병 위험도를 예측하고
→ LLM에게 그 결과를 전달해서
→ RAG 방식으로 맞춤형 권고사항을 출력하는 흐름
이걸 단순하게 if-else와 함수 호출로 구현하면 코드가 금방 엉켜버린다.
노드 간의 데이터 흐름이 복잡해지고, 분기 처리가 늘어날수록 추적도 어렵고 유지보수도 힘들어진다.
그래서 선택한 게 LangGraph다.
LangGraph를 사용하면 전체 AI 파이프라인을 그래프(Graph) 형태로 시각화하듯 설계할 수 있다.
각 처리 단계를 노드로 정의하고, 단계 간의 연결을 엣지로 표현하면서,
공유되는 데이터를 상태(State) 에 담아 흘려보내는 방식이다.
2. 핵심 개념 정리 ⭐️
LangGraph는 크게 세 가지 개념으로 구성된다.
2.1 State (상태)
State는 그래프 전체를 흐르는 공유 데이터 구조다.
쉽게 비유하면, 공장 컨베이어 벨트 위에 올라가는 상자와 같다.
상자 안에 담긴 데이터가 각 작업 스테이션(노드)을 거치면서 조금씩 추가되거나 수정된다.
그리고 그 상자의 내용을 보고 다음 스테이션으로 이동할지 분기할지를 결정하게 된다.
Python에서는 보통 TypedDict로 State 구조를 정의한다.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph
class HealthWorkflowState(TypedDict):
# 사용자 입력값
user_input: dict
# 전처리된 Feature
health_features: Optional[dict]
# ML 모델 예측 결과
prediction_result: Optional[dict]
# RAG로 검색된 문서 컨텍스트
retrieved_context: Optional[str]
# 최종 LLM 권고사항 출력
final_recommendation: Optional[str]📌 여기서 중요한 점은 State가 "각 노드마다 새로 만들어지는 것"이 아니라는 것이다.
하나의 State 객체가 그래프 전체를 관통하면서 업데이트된다.
각 노드는 State를 받아서 일부만 수정해 돌려보내면 된다.
2.2 Node (노드)
Node는 실제 작업을 수행하는 함수다.
State를 입력으로 받아서, 처리를 마친 뒤 업데이트된 State 일부를 dict 형태로 반환한다.
노드 안에 들어가는 내용은 자유롭다.
LLM 호출이 될 수도 있고, ML 모델 추론이 될 수도 있고, 평범한 Python 코드가 될 수도 있다.
def run_ml_model(state: HealthWorkflowState) -> dict:
features = state["health_features"]
# 우리가 학습시킨 ML 모델 추론
prediction = ml_model.predict_proba([features])[0]
return {
"prediction_result": {
"diabetes": round(prediction[0] * 100, 1),
"hypertension": round(prediction[1] * 100, 1),
"cardiovascular": round(prediction[2] * 100, 1),
}
}노드 함수는 반드시 State 전체를 반환할 필요 없다.
바꾸고 싶은 key-value만 dict로 반환하면, LangGraph가 알아서 기존 State와 병합한다.
2.3 Edge (엣지)
Edge는 "다음에 어느 노드로 갈 것인가"를 결정하는 연결선이다.
엣지에는 두 가지 종류가 있다.
🔹 일반 엣지 (Normal Edge)
항상 고정된 방향으로 이동한다.
"A 노드가 끝나면 무조건 B 노드로 가라"는 규칙이다.
graph.add_edge("run_ml_model", "retrieve_documents")🔹 조건부 엣지 (Conditional Edge)
State의 값을 보고 다음 노드를 동적으로 결정한다.
"조건에 따라 다른 노드로 분기하라"는 규칙이다.
이 부분이 LangGraph의 핵심적인 강점 중 하나다.
def check_health_data_exists(state: HealthWorkflowState) -> str:
"""건강검진 데이터 존재 여부에 따라 분기"""
user_input = state["user_input"]
if user_input.get("health_check_data"):
return "use_existing_data"
else:
return "derive_variables"
graph.add_conditional_edges(
"preprocess_input", # 출발 노드
check_health_data_exists, # 분기 판단 함수
{
"use_existing_data": "use_existing_data", # 반환값 -> 다음 노드
"derive_variables": "derive_variables",
}
)2.4 StateGraph
StateGraph는 노드와 엣지를 조합해서 전체 워크플로우를 정의하는 설계도다.
State가 "컨베이어 벨트 위를 흐르는 상자"라면,
StateGraph는 그 컨베이어 벨트 전체 공장 구조도에 해당한다.
graph = StateGraph(HealthWorkflowState)
# 노드 등록
graph.add_node("preprocess_input", preprocess_input)
graph.add_node("use_existing_data", use_existing_data)
graph.add_node("derive_variables", derive_variables)
graph.add_node("run_ml_model", run_ml_model)
graph.add_node("retrieve_documents", retrieve_documents)
graph.add_node("generate_recommendation", generate_recommendation)
# 컴파일
app = graph.compile()compile() 이후에는 그래프가 실행 가능한 상태가 된다.
📌 한번 compile()을 하고 나면 그래프 구조는 고정된다.
즉, State는 실행 중에 계속 바뀌지만, 그래프 구조 자체는 변하지 않는다.
3. LangGraph vs LangChain 🧐
많이 헷갈리는 부분이다.
LangChain은 기본적으로 직선형 파이프라인(DAG) 을 만드는 데 적합하다.
입력 → 처리 → 출력처럼 흐름이 단방향이고 단순할 때 편하다.
반면 LangGraph는 사이클(cycle)과 분기(branch) 를 자연스럽게 지원한다.
우리 프로젝트처럼 "조건에 따라 다른 전처리 경로를 탄다"거나,
나중에 "LLM이 결과를 검토하고 부족하면 다시 검색한다"는 피드백 루프가 필요한 경우에는
LangGraph가 훨씬 적합하다.
| 항목 | LangChain | LangGraph |
|---|---|---|
| 구조 | 직선형 DAG | 노드-엣지 그래프 |
| 분기 | 제한적 | 조건부 엣지로 자연스럽게 지원 |
| 사이클 | 지원 안 함 | 지원 (루프 가능) |
| 상태 관리 | 체인마다 별도 | 그래프 전체 공유 State |
| 복잡도 | 단순한 파이프라인에 적합 | 복잡한 에이전트 워크플로우에 적합 |
4. 우리 프로젝트에 적용하기 ⭐️⭐️
4.0 전체 아키텍처 흐름
먼저 우리가 만들려는 흐름을 정리하면 이렇다.
━━━━━━━━━━━━━━ LangGraph 워크플로우 ━━━━━━━━━━━━━━
사용자 건강 데이터 입력
↓
preprocess_input
↓ (조건부 엣지)
┌────────────────────────┐
│ 건강검진 데이터 있음? │
└────────────────────────┘
↙ ↘
use_existing derive_variables
_data (파생변수 생성)
↘ ↙
request_ml_inference ━━━━PUBLISH (ml:request)━━━━▶ ┌─────────────────┐
(request_id 생성 후 발행) │ ML 모델 서버 │
↓ │ (별도 프로세스) │
wait_ml_response ◀━━━━PUBLISH (ml:response:{id})━━━━ └─────────────────┘
(해당 id 채널 구독 대기)
↓
retrieve_documents
(RAG: 관련 문서 검색)
↓
generate_recommendation
(LLM 맞춤 권고사항 생성)
↓
END
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━4.1 State 정의
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
class HealthWorkflowState(TypedDict):
user_input: dict # 사용자 원본 입력(db의 두 테이블에서 정보 가져와야 함. user, healthdata)
health_features: Optional[dict] # 모델 입력용 Feature 벡터
prediction_result: Optional[dict] # ML 모델 예측 결과 (각 질병 확률)
retrieved_context: Optional[str] # RAG 검색 결과 (진료지침(권고사항), 서비스 이용 가이드(챌린지 정보) docs)
final_recommendation: Optional[str] # LLM 최종 출력4.2 노드 구현
* 이는 예시이며 실제와 다를 수 있음
Node 1: preprocess_input (입력값 정제)
def preprocess_input(state: HealthWorkflowState) -> dict:
"""
사용자 입력을 받아서 기본 정제만 수행한다.
건강검진 데이터 유무 판단은 이 노드가 아닌 조건부 엣지가 담당한다.
"""
user_input = state["user_input"]
# 기본 타입 변환 및 유효성 검사
cleaned_input = {
"age": int(user_input.get("age", 0)),
"height": float(user_input.get("height", 0)),
"weight": float(user_input.get("weight", 0)),
"health_check_data": user_input.get("health_check_data"), # 있으면 dict, 없으면 None
"lifestyle": user_input.get("lifestyle", {}),
}
return {"user_input": cleaned_input}Node 2-A: use_existing_data (건강검진 데이터 직접 사용)
def use_existing_data(state: HealthWorkflowState) -> dict:
"""
건강검진 데이터가 있는 경우: 해당 값을 바로 Feature로 사용한다.
"""
health_check = state["user_input"]["health_check_data"]
features = {
"total_cholesterol": health_check.get("total_cholesterol"),
"HDL" : health_check.get("HDL")
"LDL" : health_check.get("LDL")
# ... 기타 검진 항목들
}
return {"health_features": features}Node 2-B: derive_variables (파생변수 생성)
def derive_variables(state: HealthWorkflowState) -> dict:
"""
건강검진 데이터가 없는 경우: 신체 계측 + 생활습관 정보로 파생변수를 만든다.
"""
user_input = state["user_input"]
lifestyle = user_input.get("lifestyle", {})
height_m = user_input["height"] / 100
bmi = user_input["weight"] / (height_m ** 2)
# 생활습관 기반 파생 변수 예시
smoking_score = 1 if lifestyle.get("smoking") else 0
exercise_score = min(lifestyle.get("exercise_days_per_week", 0) / 7, 1.0)
features = {
"bmi": round(bmi, 2),
"smoking_score": smoking_score,
"exercise_score": exercise_score,
"age": user_input["age"],
# ... 추가 파생변수들
}
return {"health_features": features}조건부 엣지 함수
def check_health_data_exists(state: HealthWorkflowState) -> str:
"""
건강검진 데이터 존재 여부에 따라 어느 노드로 분기할지 결정한다.
"""
if state["user_input"].get("health_check_data"):
return "use_existing_data"
return "derive_variables"Node 3: run_ml_model (질병 위험도 예측)
import joblib
ml_model = joblib.load("disease_risk_model.pkl") # 우리가 학습시킨 모델
def run_ml_model(state: HealthWorkflowState) -> dict:
"""
학습된 ML 모델로 질병 위험도 확률을 예측한다.
"""
features = state["health_features"]
feature_vector = list(features.values())
probabilities = ml_model.predict_proba([feature_vector])[0]
prediction_result = {
"diabetes": round(probabilities[0] * 100, 1),
"hypertension": round(probabilities[1] * 100, 1),
"cardiovascular": round(probabilities[2] * 100, 1),
}
return {"prediction_result": prediction_result}Node 4: retrieve_documents (RAG 문서 검색)
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.load_local("health_docs_index", embeddings)
def retrieve_documents(state: HealthWorkflowState) -> dict:
"""
ML 예측 결과를 기반으로 관련 권고 문서를 검색한다.
어떤 질병의 위험도가 높은지를 쿼리로 사용한다.
"""
pred = state["prediction_result"]
# 위험도가 높은 질병 위주로 쿼리 구성
high_risk_diseases = [
disease for disease, risk in pred.items() if risk >= 50
]
query = f"{', '.join(high_risk_diseases)} 위험군을 위한 식단, 운동, 챌린지 추천"
docs = vectorstore.similarity_search(query, k=3)
context = "\n\n".join([doc.page_content for doc in docs])
return {"retrieved_context": context}Node 5: generate_recommendation (LLM 맞춤 권고사항 생성)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
def generate_recommendation(state: HealthWorkflowState) -> dict:
"""
ML 예측 결과 + RAG 문서를 바탕으로 LLM이 맞춤형 권고사항을 생성한다.
"""
pred = state["prediction_result"]
context = state["retrieved_context"]
prompt = f"""
당신은 건강 전문가입니다. 아래 사용자의 질병 위험도를 보고,
제공된 문서 내용을 바탕으로 맞춤형 챌린지와 식단을 추천해주세요.
[사용자 질병 위험도]
- 당뇨병 위험도: {pred['diabetes']}%
- 고혈압 위험도: {pred['hypertension']}%
- 심혈관 질환 위험도: {pred['cardiovascular']}%
[참고 문서]
{context}
위 내용을 바탕으로 이 사람에게 적합한 챌린지(목록 중 선택)와 식단을 추천해주세요.
추천 이유도 간단히 설명해주세요.
"""
response = llm.invoke([HumanMessage(content=prompt)])
return {"final_recommendation": response.content}4.3 그래프 조립 (StateGraph 구성)
from langgraph.graph import StateGraph, START, END
graph = StateGraph(HealthWorkflowState)
# 노드 등록
graph.add_node("preprocess_input", preprocess_input)
graph.add_node("use_existing_data", use_existing_data)
graph.add_node("derive_variables", derive_variables)
graph.add_node("run_ml_model", run_ml_model)
graph.add_node("retrieve_documents", retrieve_documents)
graph.add_node("generate_recommendation", generate_recommendation)
# 엣지 연결
graph.add_edge(START, "preprocess_input") # 시작점
# 조건부 엣지: 건강검진 데이터 유무에 따라 분기
graph.add_conditional_edges(
"preprocess_input",
check_health_data_exists,
{
"use_existing_data": "use_existing_data",
"derive_variables": "derive_variables",
}
)
# 두 분기 모두 ML 모델로 합류
graph.add_edge("use_existing_data", "run_ml_model")
graph.add_edge("derive_variables", "run_ml_model")
# 이후 순서대로 연결
graph.add_edge("run_ml_model", "retrieve_documents")
graph.add_edge("retrieve_documents", "generate_recommendation")
graph.add_edge("generate_recommendation", END) # 종료점
# 컴파일
app = graph.compile()4.4 실행
user_input_example = {
"age": 45,
"height": 172,
"weight": 80,
"health_check_data": None, # 건강검진 데이터 없음 → derive_variables 경로
"lifestyle": {
"smoking": True,
"exercise_days_per_week": 1,
"diet": "고탄수화물 위주",
}
}
result = app.invoke({"user_input": user_input_example})
print("=== 질병 위험도 예측 결과 ===")
print(result["prediction_result"])
print("\n=== LLM 맞춤 권고사항 ===")
print(result["final_recommendation"])실행 흐름을 state 변화로 정리하면 이렇다. ⭐️
START → preprocess_input
State: {user_input: {...}}
⇩
조건부 엣지 → derive_variables (health_check_data가 None이므로)
State: {user_input: {...}, health_features: {bmi: 27.0, ...}}
⇩
run_ml_model
State: {..., prediction_result: {diabetes: 62.3, hypertension: 44.1, cardiovascular: 38.7}}
⇩
retrieve_documents
State: {..., retrieved_context: "당뇨 위험군을 위한 식단은..."}
⇩
generate_recommendation
State: {..., final_recommendation: "안녕하세요. 현재 당뇨병 위험도가 62.3%로 높게 나타났습니다..."}
⇩
END
5. 설치 및 환경 설정
pip install langgraph langchain langchain-openai faiss-cpu환경 변수도 설정해두자.
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"6. 디버깅 팁 🧐
LangGraph는 중간 State를 추적하기 쉽다는 것도 장점이다.
# stream()으로 각 노드가 실행될 때마다 중간 State를 확인할 수 있다
for step in app.stream({"user_input": user_input_example}):
print(step)출력 예시:
{'preprocess_input': {'user_input': {...}}}
{'derive_variables': {'health_features': {...}}}
{'run_ml_model': {'prediction_result': {...}}}
{'retrieve_documents': {'retrieved_context': '...'}}
{'generate_recommendation': {'final_recommendation': '...'}}각 노드 이름과 함께 어떤 State가 업데이트됐는지가 그대로 보인다.
invoke()가 최종 결과만 반환한다면, stream()은 단계별 중간 결과를 실시간으로 확인할 수 있게 해준다.
개발 중에는 stream()을 쓰면서 각 노드가 의도대로 작동하는지 확인하고,
배포할 때는 invoke()로 바꾸는 방식이 좋다.
7. 요약 ⭐️
이번 포스팅은 결국 아래 문장으로 정리된다.
"LangGraph는 AI 파이프라인을 State, Node, Edge로 구조화해서
복잡한 분기와 흐름을 깔끔하게 관리할 수 있게 해주는 프레임워크다."
우리 프로젝트에 적용하면:
State: 사용자 건강 데이터부터 최종 권고사항까지 모든 정보를 담는 그릇Node: 전처리 / 파생변수 생성 / ML 추론 / RAG 검색 / LLM 생성 각각의 작업 단위Conditional Edge: 건강검진 데이터 유무에 따라 전처리 경로를 분기하는 판단자Normal Edge: 순서가 고정된 단계들 간의 연결선
이 구조를 이해하면 나중에 "LLM이 결과를 검토하고 신뢰도가 낮으면 다시 검색한다"같은
피드백 루프도 어렵지 않게 추가할 수 있다.
LangGraph를 선택한 이유가 바로 이 확장성 때문이기도 하다.
