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

개요: LangGraph 아키텍처 소개

LangGraph는 LangChain과 함께 사용되는 그래프 기반 워크플로우 엔진으로, 복잡한 AI 에이전트를 구축할 때 핵심적인 도구입니다. 기본 구성 요소는 다음과 같습니다:

• Nodes (노드): 각각의 작업 단위 (예: LLM 호출, 도구 실행, 데이터 검증)
• Edges (엣지): 노드 간의 흐름 정의
• State (상태): 워크플로우 내에서 공유되는 데이터 구조
• Checkpointing (체크포인팅): 상태 저장 및 복구 기능

from typing import Annotated, List
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import BaseMessage

# 기본 상태 구조
class AgentState(MessagesState):
    # 추가적인 상태 필드를 정의할 수 있습니다
    pass

# 간단한 노드 정의
def simple_node(state: AgentState):
    return {"messages": [f"Processed: {len(state['messages'])} messages"]}

템플릿 1: 단순 RAG 에이전트 (검색 → 생성 → 검증)

실제 프로덕션 환경에서 가장 흔한 패턴 중 하나입니다. 질문을 받고 문서를 검색한 후 생성하고 검증하는 워크플로우입니다.

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
import os

# RAG 에이전트 상태
class RAGState(MessagesState):
    query: str
    retrieved_docs: List[str]

# 검색 노드
def retrieve_node(state: RAGState):
    # FAISS 인덱스에서 문서 검색
    vectorstore = FAISS.load_local("vectorstore", OpenAIEmbeddings())
    docs = vectorstore.similarity_search(state["query"], k=3)
    return {"retrieved_docs": [doc.page_content for doc in docs]}

# 생성 노드
def generate_node(state: RAGState):
    # 검색된 문서와 질문으로 LLM 생성
    llm = ChatOpenAI(model="gpt-4")
    prompt = f"문서: {state['retrieved_docs']}\n질문: {state['query']}"
    response = llm.invoke(prompt)
    return {"messages": [response]}

# 검증 노드
def validate_node(state: RAGState):
    # 생성된 응답 검증 (예: 관련성 확인)
    if "불충분한 정보" in str(state["messages"][-1]):
        return {"messages": [f"검증 실패: {state['query']}에 대한 충분한 정보 없음"]}
    return {"messages": [f"검증 성공: {state['query']}"]}

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

# 엣지 정의
rag_workflow.add_edge("retrieve", "generate")
rag_workflow.add_edge("generate", "validate")
rag_workflow.set_entry_point("retrieve")
rag_workflow.set_finish_point("validate")

rag_chain = rag_workflow.compile()

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

복잡한 작업을 처리하기 위해 다단계 계획과 실행을 구현한 템플릿입니다.

from typing import Dict, List
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI

class ToolAgentState(MessagesState):
    plan: List[str]
    execution_result: Dict[str, str]

# 계획 노드
def plan_node(state: ToolAgentState):
    llm = ChatOpenAI(model="gpt-4")
    # 현재 상황에 맞춘 계획 생성
    plan_prompt = f"다음 작업을 수행하기 위한 계획을 세우세요: {state['query']}"
    plan = llm.invoke(plan_prompt).content
    return {"plan": plan.split("\n")}

# 실행 노드
def execute_node(state: ToolAgentState):
    # 계획된 단계를 하나씩 실행
    results = {}
    for step in state["plan"]:
        # 각 단계에 대해 실행 (예: 도구 호출)
        if "search" in step:
            results[step] = "검색 결과"
        elif "calculate" in step:
            results[step] = "계산 결과"
    return {"execution_result": results}

# 관찰 노드
def observe_node(state: ToolAgentState):
    # 실행 결과 분석
    analysis = "실행 결과 분석: "
    for key, value in state["execution_result"].items():
        analysis += f"{key}: {value}; "
    return {"messages": [analysis]}

# 결정 노드
def decide_node(state: ToolAgentState):
    # 결과에 따라 다음 행동 결정
    if "error" in str(state["execution_result"]):
        return {"messages": ["실패: 에러가 발생함"]}
    return {"messages": ["성공: 모든 작업 완료"]}

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

tool_workflow.add_edge("plan", "execute")
tool_workflow.add_edge("execute", "observe")
tool_workflow.add_edge("observe", "decide")
tool_workflow.set_entry_point("plan")
tool_workflow.set_finish_point("decide")

tool_chain = tool_workflow.compile()

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

사용자 입력이 필요한 자동화 시나리오입니다. 예를 들어, 보안 검토나 사용자 승인을 위한 워크플로우입니다.

from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import HumanMessage

class HumanInLoopState(MessagesState):
    user_approval: bool
    action_needed: str

# 작업 생성 노드
def create_action_node(state: HumanInLoopState):
    action = "예상되는 작업: 보안 검토 요청"
    return {"action_needed": action}

# 일시정지 노드
def pause_node(state: HumanInLoopState):
    # 사용자에게 작업 요청 (예: 이메일 알림)
    return {"messages": [HumanMessage(content="사용자에게 작업 승인 요청")], "user_approval": False}

# 검토 노드
def review_node(state: HumanInLoopState):
    # 사용자 승인 대기
    return {"messages": [f"검토 중: {state['action_needed']}"]}

# 계속 노드
def continue_node(state: HumanInLoopState):
    if state["user_approval"]:
        return {"messages": [f"승인됨: {state['action_needed']}"]}
    return {"messages": [f"거부됨: {state['action_needed']}"]}

# 인간-중개 워크플로우
human_loop_workflow = StateGraph(HumanInLoopState)
human_loop_workflow.add_node("create_action", create_action_node)
human_loop_workflow.add_node("pause", pause_node)
human_loop_workflow.add_node("review", review_node)
human_loop_workflow.add_node("continue", continue_node)

human_loop_workflow.add_edge("create_action", "pause")
human_loop_workflow.add_edge("pause", "review")
human_loop_workflow.add_edge("review", "continue")
human_loop_workflow.set_entry_point("create_action")
human_loop_workflow.set_finish_point("continue")

human_loop_chain = human_loop_workflow.compile()

템플릿 5: 병렬 실행 에이전트 (분산 → 처리 → 집계)

여러 작업을 병렬로 실행하고 결과를 집계하는 고성능 워크플로우입니다.

python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(MessagesState):
    tasks: List[Dict]
    results: List[Dict]

# 작업 분산 노드
def fan_out_node(state: ParallelAgentState):
    # 병렬 처리를 위한 작업 생성
    tasks = [
        {"id": 1, "type": "api_call", "data": "data1"},
        {"id": 2, "type": "api_call", "data": "data2"},
        {"id": 3, "type": "api_call", "data": "data3"}
    ]
    return {"tasks": tasks}

# 작업 처리 노드
async def process_task(task):
    # 각 작업을 비동기로 처리
    await asyncio.sleep(1)  # 시뮬레이션
    return {"task_id": task["id"], "result": f"processed_{task['data']}"}

# 병렬 처리 노드
def parallel_process

---

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