LangGraph란 무엇인가

LangGraph는 LangChain 팀이 개발한 AI 에이전트 프레임워크로, 유향 그래프(Directed Graph)를 사용하여 복잡한 LLM 워크플로우를 모델링합니다. 2025년 말 v1.0 정식 출시 이후 Klarna, Uber, Replit, Elastic 등 글로벌 기업들이 프로덕션 환경에서 활용하고 있으며, 2026년 현재 프로덕션 AI 에이전트를 구축하기 위한 가장 널리 채택된 Python 프레임워크로 자리잡았습니다.

기존의 단순 체인 방식과 달리, LangGraph는 상태(State)를 중심으로 노드(Node)와 엣지(Edge)를 연결하여 조건부 분기, 루프, 병렬 처리 등 복잡한 제어 흐름을 명시적으로 표현할 수 있습니다. 특히 Human-in-the-Loop 패턴과 Checkpointer를 통한 상태 영속화가 기본 지원되어, 장기 실행 워크플로우에 최적화되어 있습니다.

설치 및 환경 구성

LangGraph를 시작하려면 다음 패키지들을 설치합니다.

pip install langgraph==0.3.34 langchain-openai==0.3.12 langchain-community==0.3.19 tavily-python==0.5.0 python-dotenv==1.1.0

환경 변수를 설정합니다. OpenAI API 키와 Tavily(웹 검색) API 키가 필요합니다.

# .env 파일
OPENAI_API_KEY=sk-your-openai-key
TAVILY_API_KEY=tvly-your-tavily-key

핵심 개념: State, Node, Edge

LangGraph의 3대 핵심 요소를 이해해야 합니다.

State(상태)는 그래프를 흐르는 타입 지정된 딕셔너리입니다. 각 노드는 상태를 읽고 업데이트를 반환합니다. 에이전트의 모든 추론 단계를 연결하는 공유 메모리라고 생각하면 됩니다.

Node(노드)는 실제 작업을 수행하는 함수입니다. LLM 호출, 도구 실행, 데이터 변환 등 각각의 작업 단위를 노드로 정의합니다.

Edge(엣지)는 노드 간 연결을 정의합니다. 일반 엣지(항상 다음 노드로 이동)와 조건부 엣지(상태에 따라 분기)가 있습니다.

실전 예제: 리서치 에이전트 만들기

웹 검색으로 주제를 조사하고, 분석한 뒤, 보고서를 작성하는 리서치 에이전트를 단계별로 구축해 보겠습니다.

1단계: 상태 정의

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    research_topic: str
    search_queries: list[str]
    sources: list[dict]
    analysis: str
    final_report: str
    iteration: int
    max_iterations: int

messages 필드에 add_messages 리듀서를 적용하면, 새 메시지가 기존 리스트에 자동으로 추가됩니다. 이 패턴으로 대화 이력이 자연스럽게 누적됩니다.

2단계: LLM과 도구 설정

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from tavily import TavilyClient

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

3단계: 노드 함수 구현

각 노드는 상태를 입력받아 업데이트할 필드만 딕셔너리로 반환합니다.

from langchain_core.messages import HumanMessage, SystemMessage

def generate_queries(state: ResearchState) -> dict:
    topic = state["research_topic"]
    existing = ""
    if state.get("analysis"):
        existing = f"\n기존 분석:\n{state['analysis']}\n새로운 쿼리를 생성하세요."

    response = llm.invoke([
        SystemMessage(content=(
            "리서치 어시스턴트입니다. 주제에 대해 3개의 구체적인 "
            "검색 쿼리를 생성하세요. 쿼리만 한 줄씩 반환하세요." + existing
        )),
        HumanMessage(content=f"주제: {topic}"),
    ])
    queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return {
        "search_queries": state.get("search_queries", []) + queries,
        "messages": [response],
    }

def search_web(state: ResearchState) -> dict:
    queries = state.get("search_queries", [])[-3:]
    all_results = state.get("sources", [])
    for query in queries:
        try:
            response = tavily.search(query=query, max_results=3)
            for r in response.get("results", []):
                if not any(s["url"] == r["url"] for s in all_results):
                    all_results.append({
                        "title": r.get("title", ""),
                        "url": r.get("url", ""),
                        "content": r.get("content", ""),
                    })
        except Exception as e:
            print(f"검색 실패: {e}")
    return {"sources": all_results}

def analyze_results(state: ResearchState) -> dict:
    sources = state.get("sources", [])
    source_text = ""
    for i, s in enumerate(sources, 1):
        source_text += f"\n[{i}] {s['title']}\n{s['content']}\n"
    response = llm.invoke([
        SystemMessage(content=(
            "리서치 분석가입니다. 검색 결과를 분석하여 "
            "1) 핵심 발견 2) 부족한 정보 3) 신뢰도를 제공하세요."
        )),
        HumanMessage(content=f"주제: {state['research_topic']}\n자료:{source_text}"),
    ])
    return {
        "analysis": response.content,
        "iteration": state.get("iteration", 0) + 1,
        "messages": [response],
    }

def write_report(state: ResearchState) -> dict:
    response = llm.invoke([
        SystemMessage(content=(
            "리서치 보고서 작성자입니다. 분석과 자료를 바탕으로 "
            "요약, 핵심 발견, 결론, 출처를 포함한 보고서를 작성하세요."
        )),
        HumanMessage(content=(
            f"주제: {state['research_topic']}\n분석:\n{state['analysis']}"
        )),
    ])
    return {"final_report": response.content, "messages": [response]}

4단계: 조건부 라우팅

분석 결과에 따라 추가 조사를 할지, 보고서를 작성할지 결정하는 라우팅 함수입니다.

def should_continue(state: ResearchState) -> str:
    iteration = state.get("iteration", 0)
    max_iter = state.get("max_iterations", 3)
    analysis = state.get("analysis", "").lower()

    if iteration >= max_iter:
        return "write_report"
    if "low" in analysis and "confidence" in analysis:
        return "generate_queries"
    if "부족" in analysis or "추가 조사" in analysis:
        return "generate_queries"
    return "write_report"

5단계: 그래프 조립 및 실행

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(ResearchState)

workflow.add_node("generate_queries", generate_queries)
workflow.add_node("search_web", search_web)
workflow.add_node("analyze_results", analyze_results)
workflow.add_node("write_report", write_report)

workflow.add_edge(START, "generate_queries")
workflow.add_edge("generate_queries", "search_web")
workflow.add_edge("search_web", "analyze_results")

workflow.add_conditional_edges(
    "analyze_results",
    should_continue,
    {"generate_queries": "generate_queries", "write_report": "write_report"},
)
workflow.add_edge("write_report", END)

memory = MemorySaver()
agent = workflow.compile(checkpointer=memory)

initial_state = {
    "research_topic": "2026년 AI 에이전트 프로덕션 활용 사례",
    "messages": [], "search_queries": [], "sources": [],
    "analysis": "", "final_report": "",
    "iteration": 0, "max_iterations": 3,
}

config = {"configurable": {"thread_id": "session-001"}}

for event in agent.stream(initial_state, config=config):
    for node_name, output in event.items():
        print(f"Node: {node_name}")
        if node_name == "write_report":
            print(output["final_report"])

다른 프레임워크와의 비교

LangGraph 외에도 AutoGen(Microsoft)과 CrewAI가 대표적인 에이전트 프레임워크입니다.

AutoGen은 에이전트 간 대화를 통한 문제 해결에 최적화되어 있습니다. AssistantAgent와 UserProxyAgent 구조로 자연스러운 에이전트 간 상호작용이 가능하며, 코드 실행 환경이 내장되어 있습니다. 다만 v0.4에서 API가 크게 변경되었고, Microsoft Agent Framework로의 전환이 진행 중입니다.

CrewAI는 역할 기반 에이전트 팀 운영에 초점을 맞추고 있습니다. role, goal, backstory로 에이전트를 직관적으로 정의할 수 있어 빠른 프로토타이핑에 유리합니다. 그러나 복잡한 조건부 로직이나 프로덕션 수준의 상태 관리에는 한계가 있습니다.

LangGraph는 명시적 제어 흐름과 상태 영속화가 강점입니다. 학습 곡선이 다소 높지만, 프로덕션 안정성이 가장 뛰어나며 LangSmith 통합으로 관찰성(Observability)도 우수합니다.

프로덕션 배포 시 고려사항

LangGraph 에이전트를 프로덕션에 배포할 때 반드시 고려해야 할 사항들입니다.

무한 루프 방지: recursion_limit을 설정하여 에이전트가 무한 루프에 빠지는 것을 방지합니다.

result = app.invoke(input, {"configurable": {"recursion_limit": 25}})

컨텍스트 윈도우 관리: 대화가 길어지면 토큰 한도를 초과할 수 있으므로, trim_messages로 최근 메시지만 유지합니다.

from langchain_text_splitters import trim_messages
trimmed = trim_messages(state["messages"], max_tokens=8000, strategy="last")

API 장애 대비: tenacity 라이브러리로 재시도 로직을 적용합니다.

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def resilient_node(state) -> dict:
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception:
        return {"messages": [fallback_response]}

상태 영속화: 메모리 체크포인터 대신 PostgresSaver나 SqliteSaver를 사용하여 서버 재시작 후에도 상태를 복원할 수 있습니다.

비용 관리: 토큰 사용량을 실시간으로 모니터링하고, 레이트 리미팅을 적용하여 예상치 못한 비용 증가를 방지합니다.

정리

LangGraph는 AI 에이전트 개발의 핵심 프레임워크로, 그래프 기반 상태 관리와 조건부 라우팅으로 복잡한 워크플로우를 명시적으로 제어할 수 있습니다. 초기 학습 비용이 있지만, 프로덕션 환경에서의 안정성과 확장성은 다른 프레임워크를 앞서고 있습니다. CrewAI로 빠르게 프로토타입을 만들고, 검증된 워크플로우를 LangGraph로 재구현하는 전략을 권장합니다.