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

지속 가능한 AI 에이전트 개발을 위한 실전 가이드

---

개요

LangGraph는 LangChain과 함께 사용되는 그래프 기반 에이전트 개발 프레임워크로, 복잡한 AI 워크플로우를 구현할 수 있습니다. 이 가이드는 Python 개발자들이 빠르게 실전 에이전트를 구축할 수 있도록 설계된 5개의 핵심 템플릿을 제공합니다.

---

1. LangGraph 아키텍처 개요

LangGraph의 핵심 구성 요소는 다음과 같습니다:

• Nodes: 워크플로우의 각 단계 (함수 또는 클래스)
• Edges: 노드 간의 연결 관계
• State: 에이전트 상태를 저장하는 데이터 구조
• Checkpointing: 워크플로우 상태 저장 및 복원 기능

기본적인 구조는 다음과 같습니다:

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

class GraphState(TypedDict):
    messages: list
    user_input: str

graph = StateGraph(GraphState)

---

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

이 템플릿은 간단한 RAG 시스템을 구현합니다. 문서 검색 후 생성하고, 생성된 내용을 검증하는 워크플로우입니다.

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

# 상태 정의
class RagState(TypedDict):
    query: str
    documents: list
    generated_answer: str
    validation_result: bool

# 노드 정의
def retrieve(state):
    # 문서 검색 로직 (예: VectorStore 검색)
    documents = vector_store.similarity_search(state["query"])
    return {"documents": documents}

def generate(state):
    # 문서 기반 생성
    prompt = PromptTemplate.from_template(
        "다음 문서를 기반으로 질문에 답하세요:\n\n{documents}\n\n질문: {query}"
    )
    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model | StrOutputParser()
    answer = chain.invoke({
        "documents": "\n\n".join([doc.page_content for doc in state["documents"]]),
        "query": state["query"]
    })
    return {"generated_answer": answer}

def validate(state):
    # 생성된 답변 검증
    prompt = PromptTemplate.from_template(
        "다음 답변이 질문에 대해 정확한지 검증하세요:\n\n질문: {query}\n\n답변: {answer}\n\n정확한지 여부 (true/false):"
    )
    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model | StrOutputParser()
    result = chain.invoke({
        "query": state["query"],
        "answer": state["generated_answer"]
    })
    return {"validation_result": result.lower() == "true"}

# 그래프 생성
workflow = StateGraph(RagState)

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")

# 검증 결과에 따라 결정
workflow.add_conditional_edges(
    "validate",
    lambda x: "retry" if not x["validation_result"] else "end",
    {
        "retry": "retrieve",
        "end": END
    }
)

---

3. 템플릿 2: 멀티 툴 에이전트 (계획 → 실행 → 관찰 → 결정)

이 템플릿은 여러 툴을 사용하여 작업을 수행하는 에이전트입니다. 계획을 세우고, 실행하며, 관찰하고, 결정하는 순서로 워크플로우가 진행됩니다.

from langchain.tools import Tool
from langchain_openai import ChatOpenAI

# 툴 정의
def run_shell_command(command):
    import subprocess
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

def write_file(filename, content):
    with open(filename, 'w') as f:
        f.write(content)
    return f"파일 {filename}이(가) 저장되었습니다."

tools = [
    Tool(
        name="shell",
        func=run_shell_command,
        description="셸 명령어 실행"
    ),
    Tool(
        name="write_file",
        func=write_file,
        description="파일 생성"
    )
]

# 상태 정의
class ToolAgentState(TypedDict):
    task: str
    plan: str
    execution_result: str
    observation: str
    decision: str

def plan(state):
    # 작업 계획
    model = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template("다음 작업을 수행하기 위한 계획을 세우세요: {task}")
    chain = prompt | model | StrOutputParser()
    plan = chain.invoke({"task": state["task"]})
    return {"plan": plan}

def execute(state):
    # 계획 실행
    # 실제 툴 실행 로직
    model = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template("다음 계획을 실행하세요: {plan}")
    chain = prompt | model | StrOutputParser()
    result = chain.invoke({"plan": state["plan"]})
    return {"execution_result": result}

def observe(state):
    # 실행 결과 관찰
    # 예: 실행 결과를 분석하고 정리
    return {"observation": f"실행 완료: {state['execution_result']}"}

def decide(state):
    # 실행 결과에 따른 결정
    # 예: 성공 여부에 따라 후속 작업 결정
    return {"decision": "success" if "success" in state["observation"].lower() else "fail"}

# 그래프 생성
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_conditional_edges(
    "decide",
    lambda x: "retry" if x["decision"] == "fail" else "end",
    {
        "retry": "plan",
        "end": END
    }
)

---

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

이 템플릿은 인간의 개입이 필요한 작업을 처리하는 워크플로우입니다. 작업이 완료되면 인간이 검토하고 결정합니다.

from langgraph.checkpoint.memory import MemorySaver
from typing import Optional

class HumanInLoopState(TypedDict):
    task: str
    output: str
    human_review: Optional[str]
    needs_review: bool

def generate_output(state):
    model = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template("다음 작업을 수행하세요: {task}")
    chain = prompt | model | StrOutputParser()
    output = chain.invoke({"task": state["task"]})
    return {"output": output, "needs_review": True}

def human_review(state):
    # 인간 검토 (실제로는 외부 시스템과 연결)
    # 예: 웹 UI 또는 콘솔에서 입력 대기
    review = input(f"다음 결과를 검토하세요:\n{state['output']}\n(y/n): ")
    return {"human_review": review, "needs_review": False}

def approve_or_reject(state):
    return {"needs_review": False} if state["human_review"] == "y" else {"needs_review": True}

# 그래프 생성
workflow = StateGraph(HumanInLoopState)
workflow.add_node("generate", generate_output)
workflow.add_node("review", human_review)
workflow.add_node("approve", approve_or_reject)

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

workflow.add_conditional_edges(
    "review",
    lambda x: "approve" if x["human_review"] == "y" else "generate",
    {
        "approve": "approve",
        "generate": "generate"
    }
)

---

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

이 템플릿은 여러 작업을 동시에 처리하고 결과를 집계합니다.

python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(TypedDict):
    tasks: list
    results: dict
    final_aggregate: str

def fan_out(state):
    # 여러 작업 생성
    tasks = [


---

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

Comments

[xulingfeng]

Solid read. The comparison between LangGraph 워크플로우 템플릿 (v11) approaches is useful — most articles only cover one side. Having the trade-offs side by side helps a lot.