#2 LangGraph 심화: 상태 관리와 대화형 에이전트 구축

LangGraph의 상태 관리 메커니즘

이전 포스트에서 LangGraph의 기본 개념과 특징에 대해 살펴보았습니다. 이번에는 LangGraph의 핵심 기능인 상태 관리 메커니즘과 대화형 에이전트 구축 방법에 대해 더 깊이 있게 알아보겠습니다.

1. 상태 관리의 중요성

LLM 애플리케이션에서 상태 관리는 매우 중요합니다. 특히 다음과 같은 이유로 복잡한 시스템에서는 효율적인 상태 관리가 필수적입니다:

• 컨텍스트 유지: 대화형 AI가 이전 대화 내용을 기억해야 함
• 중간 계산 결과 저장: 워크플로우 중간에 생성된 데이터를 후속 단계에서 활용
• 롤백 및 복구: 오류 발생 시 이전 상태로 되돌릴 수 있는 기능
• 병렬 처리: 여러 작업을 동시에 실행하면서 일관된 상태 유지

2. LangGraph의 상태 모델

LangGraph는 불변(Immutable) 상태 모델을 사용합니다. 각 노드가 실행될 때마다 상태의 복사본이 만들어지며, 이 복사본에 변경사항이 적용됩니다. 이는 다음과 같은 이점을 제공합니다:

from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph

# 상태 정의
class ConversationState(TypedDict):
    messages: List[dict]
    context: dict
    current_step: str
    memory: dict

# 그래프 초기화 (상태 타입 명시)
graph = StateGraph(ConversationState)

상태 업데이트 패턴

def update_memory_node(state: ConversationState) -> ConversationState:
    # 상태 복사본 생성 (불변성 유지)
    new_state = state.copy()
    
    # 메모리 업데이트
    current_message = state["messages"][-1]
    new_state["memory"]["last_topic"] = extract_topic(current_message)
    new_state["memory"]["interaction_count"] = state["memory"].get("interaction_count", 0) + 1
    
    return new_state

부분 상태 업데이트

LangGraph는 전체 상태가 아닌 특정 키만 업데이트할 수 있는 기능도 제공합니다:

def process_user_input(state: ConversationState) -> dict:
    # 메시지만 업데이트
    return {"messages": state["messages"] + [{"role": "user", "content": get_user_input()}]}

# 노드 추가 시 업데이트할 키 지정
graph.add_node("process_input", process_user_input, keys=["messages"])

3. 메모리 관리 전략

LangGraph에서 효율적인 메모리 관리를 위한 몇 가지 패턴을 살펴보겠습니다:

컨텍스트 윈도우

대화가 길어지면 모든 메시지를 유지하는 것은 비효율적입니다. 컨텍스트 윈도우를 사용하여 최근 N개의 메시지만 유지할 수 있습니다:

def manage_context_window(state: ConversationState) -> dict:
    messages = state["messages"]
    if len(messages) > MAX_CONTEXT_LENGTH:
        # 중요 메시지(시스템 프롬프트 등)와 최근 메시지만 유지
        important_messages = [msg for msg in messages if msg.get("role") == "system"]
        recent_messages = messages[-MAX_RECENT_MESSAGES:]
        return {"messages": important_messages + recent_messages}
    return {"messages": messages}

graph.add_node("context_manager", manage_context_window)

선택적 메모리 요약

대화 내용이 너무 길어질 경우 LLM을 활용하여 중요 내용을 요약할 수 있습니다:

def summarize_conversation(state: ConversationState) -> dict:
    messages = state["messages"]
    if len(" ".join([m["content"] for m in messages])) > 4000:
        # 대화 내용 요약
        summary = llm.predict(f"다음 대화를 요약해주세요: {messages}")
        return {
            "memory": {**state["memory"], "conversation_summary": summary},
            "messages": [{"role": "system", "content": f"이전 대화 요약: {summary}"}] + messages[-5:]
        }
    return {}

대화형 에이전트 구축

LangGraph를 활용한 복잡한 대화형 에이전트 구축 방법에 대해 알아보겠습니다.

1. 대화형 에이전트의 기본 구조

대화형 에이전트는 다음과 같은 기본 구조를 가집니다:

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  사용자 입력  │────▶│  상태 업데이트 │────▶│ 도구 선택/실행│
└─────────────┘     └─────────────┘     └──────┬──────┘
                                               │
┌─────────────┐     ┌─────────────┐     ┌─────▼──────┐
│  응답 생성   │◀────│  메모리 업데이트 │◀────│   의사 결정  │
└─────────────┘     └─────────────┘     └─────────────┘

2. 고급 대화 에이전트 예제

다음은 고객 지원을 위한 대화형 에이전트의 구현 예시입니다:

from langgraph.graph import StateGraph
from typing import TypedDict, List, Dict, Any
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langchain.llms import OpenAI

# 상태 정의
class AgentState(TypedDict):
    messages: List[Dict[str, str]]
    tools_results: Dict[str, Any]
    current_tool: str
    memory: Dict[str, Any]
    final_response: str

# 시스템 프롬프트
SYSTEM_PROMPT = """
당신은 고객 지원 전문가입니다. 다음 도구를 사용하여 고객 문의에 응답하세요:
1. 제품 검색: 제품 정보를 조회합니다.
2. 주문 조회: 고객의 주문 상태를 확인합니다.
3. FAQ 조회: 자주 묻는 질문에서 관련 정보를 찾습니다.
4. 티켓 생성: 고객 문제를 해결할 수 없는 경우 지원 티켓을 생성합니다.
"""

# 노드 함수 정의
def initialize_conversation() -> AgentState:
    """새 대화 시작"""
    return {
        "messages": [{"role": "system", "content": SYSTEM_PROMPT}],
        "tools_results": {},
        "current_tool": "",
        "memory": {"interaction_count": 0, "user_info": {}},
        "final_response": ""
    }

def process_user_message(state: AgentState) -> AgentState:
    """사용자 메시지 처리"""
    new_state = state.copy()
    new_state["memory"]["interaction_count"] += 1
    
    # 사용자 정보 추출 (이름, 이메일 등)
    last_message = new_state["messages"][-1]["content"]
    extracted_info = extract_user_info(last_message)
    if extracted_info:
        new_state["memory"]["user_info"].update(extracted_info)
    
    return new_state

def decide_next_action(state: AgentState) -> Dict[str, str]:
    """다음 작업 결정: 도구 사용 또는 직접 응답"""
    messages_for_llm = [
        SystemMessage(content=SYSTEM_PROMPT),
        *[HumanMessage(content=m["content"]) if m["role"] == "user" 
          else AIMessage(content=m["content"]) 
          for m in state["messages"] if m["role"] != "system"]
    ]
    
    # 기존 도구 실행 결과 추가
    if state["tools_results"]:
        tool_results_str = "\n".join([f"{k}: {v}" for k, v in state["tools_results"].items()])
        messages_for_llm.append(SystemMessage(content=f"도구 실행 결과:\n{tool_results_str}"))
    
    # LLM에 의사 결정 요청
    llm = OpenAI(model_name="gpt-4o")
    decision = llm.predict_messages(
        messages_for_llm,
        "다음 중 하나를 선택하세요: 제품검색, 주문조회, FAQ조회, 티켓생성, 직접응답"
    )
    
    action = decision.content.strip()
    
    if action in ["제품검색", "주문조회", "FAQ조회", "티켓생성"]:
        return {"current_tool": action}
    else:
        return {"current_tool": "직접응답"}

def execute_tool(state: AgentState) -> Dict:
    """선택된 도구 실행"""
    tool = state["current_tool"]
    last_message = state["messages"][-1]["content"]
    
    results = {}
    if tool == "제품검색":
        product_name = extract_product_name(last_message)
        results = search_product_database(product_name)
    elif tool == "주문조회":
        order_id = extract_order_id(last_message)
        results = get_order_status(order_id)
    elif tool == "FAQ조회":
        query = last_message
        results = search_faq_database(query)
    elif tool == "티켓생성":
        user_info = state["memory"]["user_info"]
        results = create_support_ticket(user_info, last_message)
    
    return {"tools_results": {**state["tools_results"], tool: results}}

def generate_response(state: AgentState) -> Dict:
    """최종 응답 생성"""
    messages_for_llm = [
        SystemMessage(content=SYSTEM_PROMPT),
        *[HumanMessage(content=m["content"]) if m["role"] == "user" 
          else AIMessage(content=m["content"]) 
          for m in state["messages"] if m["role"] != "system"]
    ]
    
    # 도구 실행 결과 추가
    if state["tools_results"]:
        tool_results_str = "\n".join([f"{k}: {v}" for k, v in state["tools_results"].items()])
        messages_for_llm.append(SystemMessage(content=f"다음 정보를 바탕으로 응답하세요: {tool_results_str}"))
    
    # 사용자 정보 추가 (있는 경우)
    if state["memory"]["user_info"]:
        user_info_str = ", ".join([f"{k}: {v}" for k, v in state["memory"]["user_info"].items()])
        messages_for_llm.append(SystemMessage(content=f"사용자 정보: {user_info_str}"))
    
    # LLM으로 응답 생성
    llm = OpenAI(model_name="gpt-4o")
    response = llm.predict_messages(messages_for_llm)
    
    return {
        "final_response": response.content,
        "messages": state["messages"] + [{"role": "assistant", "content": response.content}]
    }

def should_use_tool(state: AgentState) -> bool:
    """도구를 사용할지 여부 결정"""
    return state["current_tool"] != "직접응답"

# 그래프 구성
agent_graph = StateGraph(AgentState)

# 노드 추가
agent_graph.add_node("initialize", initialize_conversation)
agent_graph.add_node("process_message", process_user_message)
agent_graph.add_node("decide_action", decide_next_action)
agent_graph.add_node("execute_tool", execute_tool)
agent_graph.add_node("generate_response", generate_response)

# 엣지 추가
agent_graph.add_edge("initialize", "process_message")
agent_graph.add_edge("process_message", "decide_action")
agent_graph.add_conditional_edges(
    "decide_action",
    should_use_tool,
    {
        True: "execute_tool",
        False: "generate_response"
    }
)
agent_graph.add_edge("execute_tool", "generate_response")

# 시작 노드 설정
agent_graph.set_entry_point("initialize")

# 에이전트 컴파일
agent = agent_graph.compile()

3. 대화 에이전트의 핵심 패턴

도구 선택 및 실행

LangGraph를 사용하면 에이전트가 조건에 따라 적절한 도구를 선택하고 실행할 수 있습니다. 이는 add_conditional_edges 메서드를 통해 구현됩니다:

def should_use_tool(state):
    return state["current_tool"] != "직접응답"

graph.add_conditional_edges(
    "decide_action",
    should_use_tool,
    {
        True: "execute_tool",
        False: "generate_response"
    }
)

다중 조건부 분기

더 복잡한 조건부 분기가 필요한 경우, 여러 조건을 정의할 수 있습니다:

def get_next_action(state):
    tool = state["current_tool"]
    if tool == "제품검색":
        return "product_search"
    elif tool == "주문조회":
        return "order_lookup"
    elif tool == "FAQ조회":
        return "faq_search"
    elif tool == "티켓생성":
        return "create_ticket"
    else:
        return "direct_response"

graph.add_conditional_edges(
    "decide_action",
    get_next_action,
    {
        "product_search": "search_products",
        "order_lookup": "lookup_order",
        "faq_search": "search_faq",
        "create_ticket": "create_support_ticket",
        "direct_response": "generate_response"
    }
)

4. 대화 히스토리 관리

효율적인 대화 히스토리 관리를 위한 패턴입니다:

def manage_conversation_history(state: AgentState) -> Dict:
    """대화 히스토리 관리"""
    messages = state["messages"]
    # 시스템 메시지 분리
    system_messages = [m for m in messages if m["role"] == "system"]
    user_assistant_messages = [m for m in messages if m["role"] != "system"]
    
    # 대화가 길어지면 요약
    if len(user_assistant_messages) > 10:
        llm = OpenAI(model_name="gpt-4o")
        
        # 요약할 대화 선택 (오래된 것)
        messages_to_summarize = user_assistant_messages[:-6]
        messages_to_keep = user_assistant_messages[-6:]
        
        # 대화 요약
        messages_text = "\n".join([f"{m['role']}: {m['content']}" for m in messages_to_summarize])
        summary = llm.predict(f"다음 대화를 간결하게 요약해주세요:\n{messages_text}")
        
        # 요약 메시지 생성
        summary_message = {"role": "system", "content": f"이전 대화 요약: {summary}"}
        
        # 업데이트된 메시지 목록
        updated_messages = system_messages + [summary_message] + messages_to_keep
        
        return {"messages": updated_messages}
    
    return {"messages": messages}

 

결론

LangGraph의 상태 관리 메커니즘과 대화형 에이전트 구축 패턴을 통해 복잡한 AI 워크플로우를 효율적으로 구현할 수 있습니다. 상태 관리는 LLM 기반 애플리케이션의 핵심이며, 대화형 에이전트는 사용자와의 자연스러운 상호작용을 가능하게 합니다.

다음 포스트에서는 LangGraph의 분기 및 조건부 실행과 오류 처리 전략에 대해 더 자세히 알아보겠습니다.