#3 LangGraph 심화: 분기 및 조건부 실행과 오류 처리 전략

 

앞선 포스트에서 LangGraph의 상태 관리 메커니즘과 대화형 에이전트 구축 방법에 대해 다루었습니다. 이번 포스트에서는 LangGraph의 또 다른 강력한 기능인 분기 및 조건부 실행 패턴과 오류 처리 전략에 대해 심도 있게 살펴보겠습니다.

분기 및 조건부 실행

복잡한 LLM 애플리케이션은 다양한 상황에 따라 다른 처리 경로를 선택해야 하는 경우가 많습니다. LangGraph는 이를 위한 다양한 분기 처리 패턴을 제공합니다.

1. 기본 조건부 분기

가장 기본적인 분기 처리 방법은 add_conditional_edges를 사용하는 것입니다:

from langgraph.graph import StateGraph
from typing import TypedDict, List, Dict, Literal, Union

class AnalysisState(TypedDict):
    query: str
    data: Dict
    analysis_type: str
    results: Dict

def determine_analysis_type(state: AnalysisState) -> str:
    """분석 유형 결정"""
    query = state["query"].lower()
    
    if "예측" in query or "forecast" in query:
        return "predictive"
    elif "분류" in query or "classify" in query:
        return "classification"
    elif "군집" in query or "cluster" in query:
        return "clustering"
    else:
        return "descriptive"

# 그래프 생성
analysis_graph = StateGraph(AnalysisState)

# 노드 추가
analysis_graph.add_node("parse_query", parse_query_node)
analysis_graph.add_node("determine_type", determine_analysis_type_node)
analysis_graph.add_node("predictive_analysis", predictive_analysis_node)
analysis_graph.add_node("classification_analysis", classification_analysis_node)
analysis_graph.add_node("clustering_analysis", clustering_analysis_node)
analysis_graph.add_node("descriptive_analysis", descriptive_analysis_node)
analysis_graph.add_node("format_results", format_results_node)

# 기본 경로
analysis_graph.add_edge("parse_query", "determine_type")

# 조건부 분기
analysis_graph.add_conditional_edges(
    "determine_type",
    determine_analysis_type,
    {
        "predictive": "predictive_analysis",
        "classification": "classification_analysis", 
        "clustering": "clustering_analysis",
        "descriptive": "descriptive_analysis"
    }
)

# 모든 분석 노드에서 결과 포맷팅으로 연결
for node in ["predictive_analysis", "classification_analysis", 
             "clustering_analysis", "descriptive_analysis"]:
    analysis_graph.add_edge(node, "format_results")

이 패턴을 사용하면 determine_analysis_type 함수의 반환값에 따라 다른 노드로 분기할 수 있습니다.

2. 다중 경로 실행

일부 시나리오에서는 한 노드에서 여러 경로를 병렬로 실행하고 결과를 병합해야 할 수 있습니다:

from langgraph.graph import StateGraph, END

class MultiPathState(TypedDict):
    query: str
    web_results: Dict
    database_results: Dict
    kb_results: Dict
    final_results: Dict

# 그래프 생성
search_graph = StateGraph(MultiPathState)

# 노드 추가
search_graph.add_node("parse_query", parse_query_node)
search_graph.add_node("web_search", web_search_node)
search_graph.add_node("database_query", database_query_node)
search_graph.add_node("kb_search", kb_search_node)
search_graph.add_node("merge_results", merge_results_node)

# 병렬 경로 설정
search_graph.add_edge("parse_query", "web_search")
search_graph.add_edge("parse_query", "database_query")
search_graph.add_edge("parse_query", "kb_search")

# 결과 병합
search_graph.add_edge("web_search", "merge_results")
search_graph.add_edge("database_query", "merge_results")
search_graph.add_edge("kb_search", "merge_results")
search_graph.add_edge("merge_results", END)

# 상태 병합을 위한 함수 정의
def join_all_paths(states):
    """여러 경로에서 온 상태를 병합"""
    final_state = states[0].copy()
    for state in states[1:]:
        for key, value in state.items():
            if key in final_state and isinstance(value, dict):
                final_state[key].update(value)
            else:
                final_state[key] = value
    return final_state

# 병합 함수 설정
search_graph.set_join(join_all_paths)

이 패턴은 웹 검색, 데이터베이스 쿼리, 지식 베이스 검색을 병렬로 실행한 후 결과를 병합하는 예시입니다.

3. 동적 분기 생성

실행 중에 동적으로 분기를 생성하는 패턴도 가능합니다:

def dynamic_branching(state: Dict) -> Dict[str, str]:
    """동적으로 다음 단계를 결정"""
    query_type = classify_query(state["query"])
    tools_to_use = select_tools(query_type)
    
    # 도구별 다음 노드 매핑
    next_nodes = {}
    for tool in tools_to_use:
        next_nodes[tool] = f"execute_{tool}"
        
    return next_nodes

# 동적 분기 적용
query_graph.add_conditional_edges(
    "tool_selector",
    dynamic_branching,
    lambda tool_id: tool_id  # 각 도구 ID를 해당 노드로 매핑
)

이 패턴은 실행 시점에 필요한 도구를 동적으로 결정하고, 그에 맞는 실행 경로를 생성합니다.

4. 반복 흐름 처리

특정 조건이 충족될 때까지 작업을 반복해야 하는 경우도 있습니다:

class IterativeState(TypedDict):
    iterations: int
    max_iterations: int
    results: List
    converged: bool

def check_convergence(state: IterativeState) -> bool:
    """수렴 여부 또는 최대 반복 횟수 도달 여부 확인"""
    return state["converged"] or state["iterations"] >= state["max_iterations"]

# 반복 흐름 설정
iteration_graph.add_node("initialize", initialize_node)
iteration_graph.add_node("process", process_node)
iteration_graph.add_node("check", check_convergence_node)
iteration_graph.add_node("finalize", finalize_node)

# 초기화 -> 처리
iteration_graph.add_edge("initialize", "process")

# 처리 -> 확인
iteration_graph.add_edge("process", "check")

# 조건부 분기
iteration_graph.add_conditional_edges(
    "check",
    check_convergence,
    {
        True: "finalize",  # 수렴하면 종료
        False: "process"   # 수렴하지 않으면 계속 처리
    }
)

이 패턴은 반복적인 최적화 알고리즘이나 수렴 기반 작업에 유용합니다.

오류 처리 및 복구 전략

LLM 애플리케이션에서 오류 처리는 매우 중요합니다. API 한도 초과, 연결 문제, 모델 응답 오류 등 다양한 오류가 발생할 수 있습니다.

1. 노드 레벨 예외 처리

각 노드 함수에서 예외를 처리하는 기본적인 방법입니다:

def api_call_node(state: Dict) -> Dict:
    """외부 API 호출 노드"""
    try:
        response = make_api_call(state["query"])
        return {"api_results": response}
    except ConnectionError:
        # 연결 오류 처리
        return {"api_results": None, "error": "connection_error"}
    except RateLimitError:
        # 속도 제한 오류 처리
        return {"api_results": None, "error": "rate_limit"}
    except Exception as e:
        # 기타 오류 처리
        return {"api_results": None, "error": str(e)}

2. 오류 라우팅 패턴

오류 유형에 따라 다른 처리 경로로 라우팅하는 패턴입니다:

def check_error(state: Dict) -> str:
    """오류 유형 확인"""
    if "error" not in state:
        return "success"
    
    error_type = state["error"]
    if error_type == "connection_error":
        return "retry"
    elif error_type == "rate_limit":
        return "backoff"
    else:
        return "fallback"

# 오류 라우팅
error_graph.add_conditional_edges(
    "api_call",
    check_error,
    {
        "success": "process_results",
        "retry": "retry_handler",
        "backoff": "backoff_handler",
        "fallback": "fallback_handler"
    }
)

def retry_handler(state: Dict) -> Dict:
    """재시도 로직"""
    retry_count = state.get("retry_count", 0) + 1
    if retry_count <= MAX_RETRIES:
        return {"retry_count": retry_count, "error": None}
    else:
        return {"error": "max_retries_exceeded", "fallback_needed": True}

def backoff_handler(state: Dict) -> Dict:
    """백오프 로직"""
    import time
    retry_count = state.get("retry_count", 0) + 1
    backoff_time = min(2 ** retry_count, MAX_BACKOFF)
    
    # 지수 백오프 대기
    time.sleep(backoff_time)
    
    return {"retry_count": retry_count, "error": None}

3. 대체 경로 패턴

주 경로에서 오류가 발생할 경우 대체 경로를 사용하는 패턴입니다:

class FallbackState(TypedDict):
    query: str
    primary_results: Dict
    fallback_results: Dict
    final_results: Dict
    error: Dict

def needs_fallback(state: FallbackState) -> bool:
    """대체 경로 필요 여부 확인"""
    return "error" in state and state["error"] is not None

# 대체 경로 설정
fallback_graph.add_node("process_query", process_query_node)
fallback_graph.add_node("primary_path", primary_path_node)
fallback_graph.add_node("fallback_path", fallback_path_node)
fallback_graph.add_node("merge_results", merge_results_node)

# 기본 경로
fallback_graph.add_edge("process_query", "primary_path")

# 조건부 분기
fallback_graph.add_conditional_edges(
    "primary_path",
    needs_fallback,
    {
        True: "fallback_path",
        False: "merge_results"
    }
)

fallback_graph.add_edge("fallback_path", "merge_results")

이 패턴은 주 모델(예: GPT-4)에 문제가 발생하면 대체 모델(예: Claude)을 사용하는 등의 시나리오에 유용합니다.

4. 전역 오류 처리기

그래프 전체에 적용되는 오류 처리기를 정의하는 패턴입니다:

def global_error_handler(exception: Exception, state: Dict) -> Dict:
    """전역 오류 처리기"""
    import traceback
    
    # 오류 로깅
    error_details = {
        "error_type": type(exception).__name__,
        "error_message": str(exception),
        "traceback": traceback.format_exc()
    }
    
    # 오류 기록
    log_error(error_details)
    
    # 상태 업데이트
    return {
        **state,
        "error": error_details,
        "status": "error"
    }

# 전역 오류 처리기 설정
graph.set_error_handler(global_error_handler)

5. 타임아웃 및 재시도 메커니즘

오래 걸리는 작업을 처리하는 패턴입니다:

import asyncio
from functools import wraps

def with_timeout(timeout_seconds):
    """타임아웃 데코레이터"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(
                    asyncio.coroutine(func)(*args, **kwargs),
                    timeout=timeout_seconds
                )
            except asyncio.TimeoutError:
                return {"error": "timeout", "status": "timeout"}
        return wrapper
    return decorator

@with_timeout(10)  # 10초 타임아웃
async def llm_call_node(state: Dict) -> Dict:
    """LLM API 호출"""
    response = await llm.agenerate([state["prompt"]])
    return {"llm_response": response}

고급 사용 사례

분기 및 조건부 실행과 오류 처리 전략을 활용한 실제 사용 사례를 살펴보겠습니다.

1. 자기 수정 파이프라인

LangGraph를 사용하여 자기 수정 기능을 갖춘 파이프라인을 구현할 수 있습니다:

class SelfCorrectingState(TypedDict):
    input: str
    output: str
    feedback: str
    iterations: int
    final_output: str

def check_quality(state: SelfCorrectingState) -> str:
    """출력 품질 평가"""
    if state["iterations"] >= MAX_ITERATIONS:
        return "max_iterations"
    
    feedback = evaluate_output(state["output"], state["input"])
    quality_score = feedback["quality_score"]
    
    if quality_score >= QUALITY_THRESHOLD:
        return "acceptable"
    else:
        return "needs_improvement"

# 자기 수정 그래프
correction_graph = StateGraph(SelfCorrectingState)

# 노드 추가
correction_graph.add_node("generate", generate_output_node)
correction_graph.add_node("evaluate", evaluate_output_node)
correction_graph.add_node("improve", improve_output_node)
correction_graph.add_node("finalize", finalize_output_node)

# 초기 생성
correction_graph.add_edge("generate", "evaluate")

# 품질 기반 분기
correction_graph.add_conditional_edges(
    "evaluate",
    check_quality,
    {
        "acceptable": "finalize",
        "needs_improvement": "improve",
        "max_iterations": "finalize"
    }
)

# 개선 후 재평가
correction_graph.add_edge("improve", "evaluate")

이 패턴은 코드 생성, 콘텐츠 작성 등 품질이 중요한 작업에 유용합니다.

2. 멀티모달 처리 파이프라인

다양한 유형의 입력을 처리하는 파이프라인입니다:

class MultiModalState(TypedDict):
    input_type: str
    text_content: str
    image_content: Dict
    audio_content: Dict
    processed_content: Dict

def determine_input_type(state: MultiModalState) -> str:
    """입력 유형 확인"""
    return state["input_type"]

# 멀티모달 그래프
multimodal_graph = StateGraph(MultiModalState)

# 노드 추가
multimodal_graph.add_node("detect_type", detect_input_type_node)
multimodal_graph.add_node("process_text", process_text_node)
multimodal_graph.add_node("process_image", process_image_node)
multimodal_graph.add_node("process_audio", process_audio_node)
multimodal_graph.add_node("integrate_results", integrate_results_node)

# 유형 감지
multimodal_graph.add_edge("detect_type", determine_input_type)

# 유형별 처리
multimodal_graph.add_conditional_edges(
    "detect_type",
    determine_input_type,
    {
        "text": "process_text",
        "image": "process_image",
        "audio": "process_audio"
    }
)

# 결과 통합
for node in ["process_text", "process_image", "process_audio"]:
    multimodal_graph.add_edge(node, "integrate_results")

3. RAG(Retrieval-Augmented Generation) 시스템

검색 증강 생성 시스템의 오류에 강한 구현입니다:

class RAGState(TypedDict):
    query: str
    retrieved_docs: List
    generated_answer: str
    confidence: float
    error: Dict

def check_retrieval_success(state: RAGState) -> bool:
    """검색 성공 여부 확인"""
    return "error" not in state and len(state["retrieved_docs"]) > 0

def check_confidence(state: RAGState) -> str:
    """신뢰도 확인"""
    if state["confidence"] > HIGH_CONFIDENCE:
        return "high"
    elif state["confidence"] > LOW_CONFIDENCE:
        return "medium"
    else:
        return "low"

# RAG 그래프
rag_graph = StateGraph(RAGState)

# 노드 추가
rag_graph.add_node("parse_query", parse_query_node)
rag_graph.add_node("retrieve", retrieve_documents_node)
rag_graph.add_node("direct_answer", direct_answer_node)
rag_graph.add_node("generate", generate_answer_node)
rag_graph.add_node("fallback", fallback_answer_node)
rag_graph.add_node("evaluate", evaluate_answer_node)

# 기본 흐름
rag_graph.add_edge("parse_query", "retrieve")

# 검색 성공 여부에 따른 분기
rag_graph.add_conditional_edges(
    "retrieve",
    check_retrieval_success,
    {
        True: "generate",
        False: "direct_answer"  # 검색 실패 시 직접 응답
    }
)

# 생성 후 평가
rag_graph.add_edge("generate", "evaluate")

# 신뢰도에 따른 분기
rag_graph.add_conditional_edges(
    "evaluate",
    check_confidence,
    {
        "high": END,  # 높은 신뢰도는 바로 반환
        "medium": END,  # 중간 신뢰도도 반환하지만 경고 포함 가능
        "low": "fallback"  # 낮은 신뢰도는 대체 응답
    }
)

이 패턴은 검색 실패, 생성 품질 저하 등 다양한 오류 시나리오를 처리합니다.

분기 및 오류 처리 모범 사례

복잡한 LangGraph 애플리케이션을 위한 모범 사례를 살펴보겠습니다.

1. 세분화된 노드 설계

각 노드는 단일 책임을 가지도록 설계하는 것이 좋습니다:

# 나쁜 예: 하나의 노드에서 여러 작업 수행
def big_node(state: Dict) -> Dict:
    # 입력 검증, 데이터 처리, API 호출, 오류 처리를 모두 한 함수에서 수행
    ...

# 좋은 예: 작은 노드로 분리
def validate_input(state: Dict) -> Dict: ...
def preprocess_data(state: Dict) -> Dict: ...
def api_call(state: Dict) -> Dict: ...
def handle_api_error(state: Dict) -> Dict: ...

세분화된 노드를 사용하면 오류 처리와 분기가 더 쉬워집니다.

2. 상태 불변성 유지

상태를 직접 수정하지 않고 새 상태를 반환하는 것이 안전합니다:

# 나쁜 예: 상태 직접 수정
def modify_state(state: Dict) -> Dict:
    state["key"] = "new_value"  # 기존 상태 수정
    return state

# 좋은 예: 불변성 유지
def modify_state(state: Dict) -> Dict:
    return {**state, "key": "new_value"}  # 새 상태 반환

불변성은 디버깅과 상태 추적을 용이하게 합니다.

3. 디버깅을 위한 로깅

각 노드에 로깅을 추가하여 디버깅을 용이하게 합니다:

import logging

def process_node(state: Dict) -> Dict:
    logging.info(f"Starting processing with state: {state}")
    try:
        result = do_process(state)
        logging.info(f"Processing completed with result: {result}")
        return result
    except Exception as e:
        logging.error(f"Error in processing: {e}")
        raise

4. 점진적 성능 저하 전략

주 기능이 실패해도 기본 기능은 유지되도록 설계합니다:

def generate_response(state: Dict) -> Dict:
    try:
        # 고급 모델로 응답 생성 시도
        response = generate_with_advanced_model(state)
        return {"response": response, "model_used": "advanced"}
    except Exception as e:
        try:
            # 고급 모델 실패 시 기본 모델로 대체
            response = generate_with_basic_model(state)
            return {"response": response, "model_used": "basic", "fallback_reason": str(e)}
        except Exception as e2:
            # 모든 모델 실패 시 정적 응답 제공
            return {"response": "죄송합니다, 현재 응답을 생성할 수 없습니다.", 
                    "model_used": "none", 
                    "error": str(e2)}

5. 재시도 정책 구현

외부 서비스 호출에는 적절한 재시도 정책을 구현합니다:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_external_api(query: str) -> Dict:
    """지수 백오프와 최대 3회 재시도를 사용한 API 호출"""
    response = requests.get(API_URL, params={"q": query})
    response.raise_for_status()  # 4xx, 5xx 응답 시 예외 발생
    return response.json()

결론

LangGraph의 분기 및 조건부 실행 패턴과 오류 처리 전략을 통해 견고하고 유연한 LLM 애플리케이션을 구축할 수 있습니다. 이러한 패턴은 다양한 입력과 예외 상황을 처리하는 데 필수적이며, 프로덕션 환경에서의 안정성을 크게 향상시킵니다.

분기 처리를 통해 상황에 맞는 최적의 경로를 선택할 수 있으며, 오류 처리 전략을 통해 예상치 못한 상황에도 적절히 대응할 수 있습니다. 세분화된 노드 설계, 상태 불변성 유지, 로깅, 점진적 성능 저하, 재시도 정책 등의 모범 사례를 적용하면 더욱 강력한 LangGraph 애플리케이션을 구축할 수 있습니다.

다음 포스트에서는 LangGraph의 스레드 및 병렬 처리와 성능 최적화 기법에 대해 자세히 알아보겠습니다.

---