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

Python 개발자를 위한 LangChain/LangGraph 워크플로우 템플릿

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반의 워크플로우 시스템으로, 다음과 같은 핵심 구성 요소로 작동합니다:

• 노드 (Nodes): 각 단계의 작업을 정의하는 함수
• 엣지 (Edges): 노드 간의 전이 조건
• 상태 (State): 모든 노드에서 공유되는 데이터 구조
• 체크포인팅 (Checkpointing): 상태를 저장하고 복구하는 기능

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    user_input: str

# 기본 워크플로우 구성
workflow = StateGraph(AgentState)

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

실제 문제: 문서 검색 후 생성된 답변의 정확성 검증 필요

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# RAG 워크플로우
def retrieve(state):
    # 문서 검색 로직
    vector_store = VectorStore()
    query = state["user_input"]
    retrieved_docs = vector_store.search(query, k=3)
    return {"retrieved_docs": retrieved_docs}

def generate(state):
    # 생성 로직
    prompt = PromptTemplate.from_template(
        "다음 문서를 기반으로 질문에 답하세요: {context}\n질문: {query}"
    )
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()

    context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
    answer = chain.invoke({"context": context, "query": state["user_input"]})

    return {"generated_answer": answer}

def validate(state):
    # 검증 로직
    prompt = PromptTemplate.from_template(
        "다음 답변이 질문에 답하고 있는지 검증하세요:\n질문: {query}\n답변: {answer}\n정답 여부: (예/아니오)"
    )
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()

    verdict = chain.invoke({
        "query": state["user_input"], 
        "answer": state["generated_answer"]
    })

    return {"validation": verdict, "valid": "예" in verdict}

# 워크플로우 정의
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_node("validate", validate)

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

# 검증 결과에 따라 분기
def route_validation(state):
    if state["valid"]:
        return END
    else:
        return "retrieve"  # 재검색

workflow.add_conditional_edges(
    "validate",
    route_validation,
    {
        "retrieve": "retrieve",
        END: END
    }
)

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

실제 문제: 여러 도구를 사용해야 하는 복잡한 작업 처리

from langchain.tools import Tool
from langchain_openai import OpenAI

# 도구 정의
tools = [
    Tool(
        name="search",
        func=lambda query: f"검색 결과: {query}",
        description="웹 검색을 위한 도구"
    ),
    Tool(
        name="calculator",
        func=lambda expression: f"계산 결과: {eval(expression)}",
        description="수학 계산을 위한 도구"
    )
]

class ToolAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    plan: str
    execution_result: str
    observation: str

def plan(state):
    # 작업 계획 생성
    prompt = PromptTemplate.from_template(
        "다음 작업을 위한 계획을 세우세요:\n{task}\n도구 사용 계획:"
    )
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    plan = chain.invoke({"task": state["user_input"]})
    return {"plan": plan}

def execute(state):
    # 도구 실행
    tools_dict = {tool.name: tool for tool in tools}
    # 간단한 실행 로직 (실제 구현 시 도구 선택 로직 포함)
    tool_result = tools_dict["search"].run(state["user_input"])
    return {"execution_result": tool_result}

def observe(state):
    # 실행 결과 관찰
    return {"observation": f"실행 결과: {state['execution_result']}"}

def decide(state):
    # 결정 로직
    prompt = PromptTemplate.from_template(
        "다음 작업을 고려하세요:\n{task}\n관찰 결과: {observation}\n결정:"
    )
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    decision = chain.invoke({
        "task": state["user_input"],
        "observation": state["observation"]
    })
    return {"decision": decision}

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

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)

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

실제 문제: 인간의 판단이 필요한 작업에서 자동화와 인간 통제의 균형

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

class HumanInLoopState(TypedDict):
    messages: Annotated[list, operator.add]
    task: str
    human_review: str
    status: str  # "pending", "approved", "rejected"
    timestamp: str

def task_creation(state):
    # 작업 생성
    task = f"사용자 요청: {state['user_input']}"
    return {"task": task, "status": "pending", "timestamp": datetime.now().isoformat()}

def review_request(state):
    # 검토 요청
    return {"messages": state["messages"] + [{"role": "assistant", "content": 
        f"작업을 검토해 주세요:\n{state['task']}\n[승인/거부]"}]}

def human_review(state):
    # 인간 검토
    review = state["human_review"]  # 사용자 입력
    status = "approved" if "승인" in review else "rejected"
    return {"status": status, "messages": state["messages"] + [{"role": "user", "content": review}]}

def handle_rejection(state):
    # 거부 시 재작업
    return {"messages": state["messages"] + [{"role": "assistant", "content": 
        "작업이 거부되었습니다. 새로운 요청을 주세요"}]}

# 워크플로우 정의
workflow = StateGraph(HumanInLoopState)
workflow.add_node("create_task", task_creation)
workflow.add_node("request_review", review_request)
workflow.add_node("human_review", human_review)
workflow.add_node("handle_rejection", handle_rejection)

workflow.set_entry_point("create_task")
workflow.add_edge("create_task", "request_review")
workflow.add_edge("request_review", "human_review")

def route_review(state):
    if state["status"] == "approved":
        return "END"
    else:
        return "handle_rejection"

workflow.add_conditional_edges(
    "human_review",
    route_review,
    {
        "handle_rejection": "handle_rejection",
        "END": END
    }
)

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

실제 문제: 여러 데이터 소스를 동시에 처리하고 결과를 통합하는 작업

python
import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelProcessingState(TypedDict):
    messages: Annotated[list, operator.add]
    data_sources: list
    processed_results: list
    aggregated_result: str

def fan_out(state):
    # 데이터 소스 분산
    sources = state["data_sources"]
    return {"processed_results": []}

async def process_async(data_source):
    # 비동기 처리 로직
    await asyncio.sleep(1)  # 실제 API 호출
    return f"처리된 {data_source}"

def process_parallel(state):
    # 병렬 처리
    with ThreadPoolExecutor(max

---

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