LangGraph 상태 머신: 프로덕션에서 복잡한 에이전트 작업 흐름 관리

랭그래프란 무엇인가요?

LangGraph는 LLM 애플리케이션을 위해 특별히 설계된 워크플로 조정 프레임워크입니다. 핵심 원칙은 다음과 같습니다.

• 복잡한 작업을 상태와 전환으로 나누기
• 상태 전환 로직 관리
• 작업 실행 중 다양한 예외 처리

쇼핑을 생각해 보세요. 찾아보기 → 장바구니에 담기 → 결제 → 결제. LangGraph는 이러한 작업 흐름을 효율적으로 관리하는 데 도움이 됩니다.

핵심 개념

1. 상태

상태는 작업 실행의 체크포인트와 같습니다.

from typing import TypedDict, List

class ShoppingState(TypedDict):
    # Current state
    current_step: str
    # Cart items
    cart_items: List[str]
    # Total amount
    total_amount: float
    # User input
    user_input: str

class ShoppingGraph(StateGraph):
    def __init__(self):
        super().__init__()

        # Define states
        self.add_node("browse", self.browse_products)
        self.add_node("add_to_cart", self.add_to_cart)
        self.add_node("checkout", self.checkout)
        self.add_node("payment", self.payment)

2. 상태 전환

상태 전환은 작업 흐름의 "로드맵"을 정의합니다.

class ShoppingController:
    def define_transitions(self):
        # Add transition rules
        self.graph.add_edge("browse", "add_to_cart")
        self.graph.add_edge("add_to_cart", "browse")
        self.graph.add_edge("add_to_cart", "checkout")
        self.graph.add_edge("checkout", "payment")

    def should_move_to_cart(self, state: ShoppingState) -> bool:
        """Determine if we should transition to cart state"""
        return "add to cart" in state["user_input"].lower()

3. 상태 지속성

시스템 안정성을 보장하려면 상태 정보를 유지해야 합니다.

class StateManager:
    def __init__(self):
        self.redis_client = redis.Redis()

    def save_state(self, session_id: str, state: dict):
        """Save state to Redis"""
        self.redis_client.set(
            f"shopping_state:{session_id}",
            json.dumps(state),
            ex=3600  # 1 hour expiration
        )

    def load_state(self, session_id: str) -> dict:
        """Load state from Redis"""
        state_data = self.redis_client.get(f"shopping_state:{session_id}")
        return json.loads(state_data) if state_data else None

4. 오류 복구 메커니즘

어떤 단계든 실패할 수 있으며 이러한 상황을 적절하게 처리해야 합니다.

class ErrorHandler:
    def __init__(self):
        self.max_retries = 3

    async def with_retry(self, func, state: dict):
        """Function execution with retry mechanism"""
        retries = 0
        while retries < self.max_retries:
            try:
                return await func(state)
            except Exception as e:
                retries += 1
                if retries == self.max_retries:
                    return self.handle_final_error(e, state)
                await self.handle_retry(e, state, retries)

    def handle_final_error(self, error, state: dict):
        """Handle final error"""
        # Save error state
        state["error"] = str(error)
        # Rollback to last stable state
        return self.rollback_to_last_stable_state(state)

실제 사례: 지능형 고객 서비스 시스템

지능형 고객 서비스 시스템의 실제 사례를 살펴보겠습니다.

from langgraph.graph import StateGraph, State

class CustomerServiceState(TypedDict):
    conversation_history: List[str]
    current_intent: str
    user_info: dict
    resolved: bool

class CustomerServiceGraph(StateGraph):
    def __init__(self):
        super().__init__()

        # Initialize states
        self.add_node("greeting", self.greet_customer)
        self.add_node("understand_intent", self.analyze_intent)
        self.add_node("handle_query", self.process_query)
        self.add_node("confirm_resolution", self.check_resolution)

    async def greet_customer(self, state: State):
        """Greet customer"""
        response = await self.llm.generate(
            prompt=f"""
            Conversation history: {state['conversation_history']}
            Task: Generate appropriate greeting
            Requirements:
            1. Maintain professional friendliness
            2. Acknowledge returning customers
            3. Ask how to help
            """
        )
        state['conversation_history'].append(f"Assistant: {response}")
        return state

    async def analyze_intent(self, state: State):
        """Understand user intent"""
        response = await self.llm.generate(
            prompt=f"""
            Conversation history: {state['conversation_history']}
            Task: Analyze user intent
            Output format:
            {{
                "intent": "refund/inquiry/complaint/other",
                "confidence": 0.95,
                "details": "specific description"
            }}
            """
        )
        state['current_intent'] = json.loads(response)
        return state

용법

# Initialize system
graph = CustomerServiceGraph()
state_manager = StateManager()
error_handler = ErrorHandler()

async def handle_customer_query(user_id: str, message: str):
    # Load or create state
    state = state_manager.load_state(user_id) or {
        "conversation_history": [],
        "current_intent": None,
        "user_info": {},
        "resolved": False
    }

    # Add user message
    state["conversation_history"].append(f"User: {message}")

    # Execute state machine flow
    try:
        result = await graph.run(state)
        # Save state
        state_manager.save_state(user_id, result)
        return result["conversation_history"][-1]
    except Exception as e:
        return await error_handler.with_retry(
            graph.run,
            state
        )

모범 사례

1. 국가 디자인 원칙

   • 상태를 단순하고 명확하게 유지
   • 필요한 정보만 저장
   • 연재 요구 사항 고려

2. 전환 논리 최적화

   • 조건 전환 사용
   • 무한 루프 방지
   • 최대 걸음 수 제한 설정

3. 오류 처리 전략

   • 우아한 저하 구현
   • 상세정보 기록
   • 롤백 메커니즘 제공

4. 성능 최적화

   • 비동기 작업 사용
   • 상태 캐싱 구현
   • 상태 크기 제어

일반적인 함정과 해결책

1. 국가 폭발

   • 문제: 상태가 너무 많아 유지 관리가 어렵습니다
   • 해결책: 유사한 상태를 병합하고 새 상태를 만드는 대신 상태 조합을 사용하세요

2. 교착상태

   • 문제: 순환 상태 전환으로 인해 작업이 중단됩니다
   • 해결책: 시간 초과 메커니즘 및 강제 종료 조건 추가

3. 국가 일관성

   • 문제: 분산 환경의 일관성 없는 상태
   • 해결책: 분산 잠금 및 트랜잭션 메커니즘 사용

요약

LangGraph 상태 머신은 복잡한 AI Agent 작업 흐름을 관리하기 위한 강력한 솔루션을 제공합니다.

• 명확한 업무 흐름 관리
• 신뢰할 수 있는 상태 지속성
• 종합적인 오류 처리
• 유연한 확장성

위 내용은 LangGraph 상태 머신: 프로덕션에서 복잡한 에이전트 작업 흐름 관리의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!