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

#ai #llm #developers #tutorial

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

개발자를 위한 경량 LangGraph 워크플로우 템플릿

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반 워크플로우를 구현하기 위한 경량 프레임워크입니다. 핵심 구성 요소는 다음과 같습니다:

Nodes: 워크플로우의 각 단계 (함수 또는 클래스)
Edges: 노드 간 연결 및 조건
State: 모든 노드가 공유하는 상태 객체
Checkpointing: 상태 저장 및 복원

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

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

# 기본 구조
graph = StateGraph(State)
graph.add_node("node1", node_function)
graph.add_edge("node1", END)

2. 템플릿 1: 간단한 RAG 에이전트 (Retrieve → Generate → Validate)

문제: 문서 검색 후 생성된 답변을 검증해야 할 때

import operator
from typing import Annotated, List, Dict, Any
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnableLambda
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class RAGState(TypedDict):
    messages: Annotated[list, operator.add]
    context: str
    answer: str

def retrieve(state: RAGState):
    # 문서 검색 로직
    vector_store = YourVectorStore()  # 사용자 정의
    query = state["messages"][-1].content
    context = vector_store.search(query, k=3)
    return {"context": "\n".join(context)}

def generate(state: RAGState):
    # LLM 답변 생성
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"""주어진 문맥을 기반으로 질문에 답하세요:
    문맥: {state['context']}
    질문: {state['messages'][-1].content}"""

    response = llm.invoke(prompt)
    return {"answer": response.content}

def validate(state: RAGState):
    # 답변 검증
    if len(state["answer"]) < 10:  # 너무 짧은 답변
        return {"messages": [AIMessage(content="답변이 너무 짧습니다. 다시 시도해주세요.")]}
    return {"messages": [AIMessage(content=state["answer"])]}

# 그래프 구성
rag_graph = StateGraph(RAGState)
rag_graph.add_node("retrieve", retrieve)
rag_graph.add_node("generate", generate)
rag_graph.add_node("validate", validate)

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

# 실행
runnable = rag_graph.compile()
result = runnable.invoke({"messages": [HumanMessage(content="Python asyncio는 어떻게 사용하나요?")]})

3. 템플릿 2: 멀티-도구 에이전트 (Plan → Execute → Observe → Decide)

문제: 여러 도구를 순차적으로 사용해야 하는 복잡한 작업

from langchain.tools import Tool
from langchain_core.tools import tool

# 도구 정의
@tool
def search_web(query: str) -> str:
    """웹 검색 도구"""
    return f"검색 결과: {query}에 대한 정보"

@tool
def execute_shell(command: str) -> str:
    """명령어 실행"""
    import subprocess
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

@tool
def file_editor(filename: str, content: str) -> str:
    """파일 편집"""
    with open(filename, 'w') as f:
        f.write(content)
    return f"파일 {filename}이(가) 저장되었습니다."

# 상태
class MultiToolState(TypedDict):
    messages: Annotated[list, operator.add]
    plan: List[str]
    current_tool: str
    tool_result: str

def plan(state: MultiToolState):
    # 작업 계획 생성
    plan = ["search_web", "execute_shell", "file_editor"]
    return {"plan": plan}

def execute(state: MultiToolState):
    # 현재 도구 실행
    tool_name = state["plan"][0]
    # 실제 도구 호출 로직 (예: langchain tools 사용)
    tool = getattr(sys.modules[__name__], tool_name)
    # 간단한 예시
    return {"current_tool": tool_name, "tool_result": f"{tool_name} 실행 완료"}

def observe(state: MultiToolState):
    # 실행 결과 검토
    return {"messages": [AIMessage(content=f"도구 {state['current_tool']} 실행 결과: {state['tool_result']}")]}

def decide(state: MultiToolState):
    # 다음 단계 결정
    if len(state["plan"]) > 1:
        return {"plan": state["plan"][1:]}  # 다음 도구로 이동
    return {"messages": [AIMessage(content="모든 작업 완료")]}

# 그래프 구성
multi_tool_graph = StateGraph(MultiToolState)
multi_tool_graph.add_node("plan", plan)
multi_tool_graph.add_node("execute", execute)
multi_tool_graph.add_node("observe", observe)
multi_tool_graph.add_node("decide", decide)

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

runnable = multi_tool_graph.compile()

4. 템플릿 3: 인간-중개 워크플로우 (Pause → Review → Continue)

문제: 중요한 결정을 인간이 검토해야 하는 상황

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

class HumanInLoopState(TypedDict):
    messages: Annotated[list, operator.add]
    decision: str  # 'approve', 'reject', 'review'
    review_comments: str

def generate_suggestion(state: HumanInLoopState):
    # 제안 생성
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"다음 작업에 대한 제안을 생성하세요: {state['messages'][-1].content}"
    suggestion = llm.invoke(prompt).content
    return {"messages": [AIMessage(content=suggestion)]}

def human_review(state: HumanInLoopState):
    # 인간 검토 지점
    return {"decision": "review", "messages": [AIMessage(content="검토 요청됨")]}

def process_decision(state: HumanInLoopState):
    # 결정 처리
    if state["decision"] == "approve":
        return {"messages": [AIMessage(content="결정 승인")]}
    elif state["decision"] == "reject":
        return {"messages": [AIMessage(content="결정 거부")]}
    else:
        return {"messages": [AIMessage(content="검토 필요")]}

# 인간 검토가 필요한 경우
human_loop_graph = StateGraph(HumanInLoopState)
human_loop_graph.add_node("generate", generate_suggestion)
human_loop_graph.add_node("review", human_review)
human_loop_graph.add_node("process", process_decision)

human_loop_graph.set_entry_point("generate")
human_loop_graph.add_edge("generate", "review")
human_loop_graph.add_edge("review", "process")
human_loop_graph.add_edge("process", END)

runnable = human_loop_graph.compile(checkpointer=MemorySaver())

5. 템플릿 5: 병렬 실행 에이전트 (Fan-out → Process → Aggregate)

문제: 동시에 여러 작업을 수행하고 결과를 집계해야 할 때


python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelState(TypedDict):
    messages: Annotated[list, operator.add]
    results: List[Dict]
    tasks: List[str]

def fan_out(state: ParallelState):
    # 작업 분할
    tasks = ["task1", "task2", "task3"]  # 실제 작업 목록
    return {"tasks": tasks, "results": []}

def process_task(state: ParallelState, task: str):
    # 개별 작업 처리
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"작업 '{task}'에 대한 결과를 생성하세요"
    result = llm.invoke(prompt).content
    return {"results": [{"task": task, "result": result}]}

async def parallel_processing(state: ParallelState):
    # 병렬 처리
    tasks = state["tasks"]

    async def process_single(task):
        # 비동기 처리 로직
        llm = ChatOpenAI(model="gpt-4o-mini")
        prompt = f"작업 '{task}'에 대한 결과를 생성하세요"
        result = await llm.ainvoke(prompt)
        return {"task": task

---

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