LangGraph 워크플로우 템플릿 (v42)

LangGraph 아키텍처 개요

LangGraph는 고급 AI 에이전트를 구축하기 위한 강력한 프레임워크로, 노드(Node), 엣지(Edge), 상태(State), 체크포인팅(Checkpointing)으로 구성됩니다.

노드(Node): 각 단계의 작업을 정의합니다. 예를 들어 '리트리브', '생성', '검증' 등입니다.

엣지(Edge): 노드 간의 흐름을 정의합니다. 조건부 엣지를 사용하여 복잡한 논리를 구현할 수 있습니다.

상태(State): 에이전트의 현재 상태를 관리합니다. 메시지 목록, 변수, 상태 기록 등을 포함합니다.

체크포인팅(Checkpointing): 에이전트의 진행 상황을 저장하여 재시작 시 복구할 수 있도록 합니다.

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

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]

# 기본 구성
workflow = StateGraph(GraphState)

템플릿 1: 간단한 RAG 에이전트 (리트리브 → 생성 → 검증)

이 템플릿은 문서 검색 및 생성을 위한 기본적인 RAG 워크플로우입니다.

from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langgraph.graph import StateGraph, END

class RAGState(TypedDict):
    query: str
    context: list
    answer: str
    validation: bool

def retrieve_node(state):
    # 검색 엔진에서 관련 문서 가져오기
    query = state["query"]
    # 실제 구현에서는 VectorStore나 검색 엔진을 사용
    context = [
        f"문서 1 내용: {query} 관련 정보",
        f"문서 2 내용: {query} 관련 설명"
    ]
    return {"context": context}

def generate_node(state):
    # LLM을 사용하여 답변 생성
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "질문: {query}\n문맥: {context}\n\n답변을 생성하세요:"
    )
    chain = prompt | llm
    answer = chain.invoke({
        "query": state["query"],
        "context": "\n".join(state["context"])
    })
    return {"answer": answer.content}

def validate_node(state):
    # 답변 검증 (간단한 예시)
    return {"validation": True}

# 워크플로우 구성
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("validate", validate_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
workflow.add_edge("validate", END)

app = workflow.compile()

템플릿 2: 멀티-도구 에이전트 (계획 → 실행 → 관찰 → 결정)

이 템플릿은 여러 도구를 사용하여 복잡한 작업을 수행하는 에이전트입니다.

import json
from typing import List, Dict
from langgraph.graph import StateGraph, END

class ToolAgentState(TypedDict):
    task: str
    plan: List[str]
    execution_log: List[Dict]
    observations: List[str]
    final_decision: str

def plan_node(state):
    # 작업 계획 생성
    plan = [
        "도구 1 실행",
        "도구 2 실행",
        "도구 3 실행"
    ]
    return {"plan": plan}

def execute_node(state):
    # 각 도구 실행
    execution_results = []
    for tool in state["plan"]:
        result = {"tool": tool, "status": "completed", "data": f"{tool} 실행 결과"}
        execution_results.append(result)
    return {"execution_log": execution_results}

def observe_node(state):
    # 실행 결과 관찰
    observations = []
    for log in state["execution_log"]:
        observations.append(f"{log['tool']}: {log['data']}")
    return {"observations": observations}

def decide_node(state):
    # 최종 결정
    final_decision = f"모든 작업 완료: {len(state['observations'])}개 관찰"
    return {"final_decision": final_decision}

# 워크플로우 구성
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("observe", observe_node)
workflow.add_node("decide", decide_node)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
workflow.add_edge("decide", END)

app = workflow.compile()

템플릿 3: 인간-인-라인 워크플로우 (일시정지 → 검토 → 계속)

이 템플릿은 인간 검토가 필요한 작업을 위한 인간-인-라인 프로세스입니다.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from typing import Literal

class HumanInLoopState(TypedDict):
    task: str
    result: str
    human_review: str
    approval_status: Literal["approved", "rejected", "pending"]

def execute_task(state):
    # 작업 실행
    result = f"작업 '{state['task']}' 완료"
    return {"result": result}

def human_review_node(state):
    # 인간 검토 요청 (실제 구현에서는 외부 API 호출)
    return {"approval_status": "pending"}

def handle_approval(state):
    # 승인 또는 거절 처리
    if state["human_review"] == "approve":
        return {"approval_status": "approved"}
    else:
        return {"approval_status": "rejected"}

# 체크포인팅 추가
memory = MemorySaver()
workflow = StateGraph(HumanInLoopState)

workflow.add_node("execute", execute_task)
workflow.add_node("review", human_review_node)
workflow.add_node("handle_approval", handle_approval)

workflow.set_entry_point("execute")
workflow.add_edge("execute", "review")

# 조건부 엣지: 인간 검토 결과에 따라 다르게 처리
def route_review(state):
    if state["approval_status"] == "approved":
        return "handle_approval"
    else:
        return "review"

workflow.add_conditional_edges(
    "review",
    route_review,
    {
        "handle_approval": "handle_approval",
        "review": "review"
    }
)

workflow.add_edge("handle_approval", END)
app = workflow.compile(checkpointer=memory)

템플릿 4: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)

이 템플릿은 여러 작업을 병렬로 처리하고 결과를 집계하는 워크플로우입니다.

from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict
import asyncio

class ParallelAgentState(TypedDict):
    tasks: List[str]
    results: List[Dict]
    aggregated_result: str

def fanout_node(state):
    # 작업 분산
    tasks = ["task_1", "task_2", "task_3"]
    return {"tasks": tasks}

def process_task(state):
    # 각 작업 처리
    def process_single_task(task):
        # 실제 작업 수행
        return {"task": task, "result": f"{task} 처리 완료"}

    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(process_single_task, task) for task in state["tasks"]]
        results = [future.result() for future in futures]

    return {"results": results}

def aggregate_node(state):
    # 결과 집계
    aggregated = "집계 결과: " + ", ".join([r["task"] for r in state["results"]])
    return {"aggregated_result": aggregated}

# 병렬 워크플로우 구성
workflow = StateGraph(ParallelAgentState)
workflow.add_node("fanout", fanout_node)
workflow.add_node("process", process_task)
workflow.add_node("aggregate", aggregate_node)

workflow.set_entry_point("fanout")
workflow.add_edge("fanout", "process")
workflow.add_edge("process", "aggregate")
workflow.add_edge("aggregate", END)

app = workflow.compile()

상태 관리 패턴

1. 상태 병합 패턴

from typing import Annotated, List
import operator

class MessageState(TypedDict):
    messages: Annotated[List[dict], operator.add]

# 메시지 추가 시 병합
def add_message(state, message):
    return {"messages": [message]}

2. 상태 초기화 패턴

python
def initialize_state(initial_state):
    return {
        "messages": [],
        "metadata": {},
        "timestamp": time.time()


---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)