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

#ai #llm #developers #tutorial

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

실제로 개발자가 실제로 사용할 수 있는 LangGraph 패턴 가이드

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반 워크플로우 엔진으로, 노드(Node), 엣지(Edge), 상태(State), 체크포인트(Checkpointing)의 네 가지 핵심 구성 요소로 작동합니다.

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

# 상태 정의
class GraphState(TypedDict):
    question: str
    answer: str
    steps: list

# 노드 정의
def retrieve(state):
    # 문서 검색 로직
    return {"documents": ["doc1", "doc2"]}

def generate(state):
    # 답변 생성 로직
    return {"answer": "생성된 답변"}

# 워크플로우 정의
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)

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

문제: 간단한 질문 답변 시스템에서 검색, 생성, 검증의 세 단계를 포함하는 RAG 워크플로우

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

class RAGState(TypedDict):
    question: str
    documents: List[str]
    answer: str
    validation: bool
    retries: int

def retrieve_node(state: RAGState):
    # 검색 로직
    documents = search_documents(state["question"])
    return {"documents": documents, "retries": 0}

def generate_node(state: RAGState):
    # 생성 로직
    prompt = f"질문: {state['question']}\n문서: {state['documents']}"
    answer = call_llm(prompt)
    return {"answer": answer}

def validate_node(state: RAGState):
    # 검증 로직
    validation = evaluate_answer(state["question"], state["answer"])
    return {"validation": validation}

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

def route_validation(state: RAGState):
    if state["validation"]:
        return END
    elif state["retries"] < 2:
        return "retrieve"
    else:
        return "generate"

workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
workflow.add_conditional_edges("validate", route_validation)

# 실행
app = workflow.compile()
result = app.invoke({"question": "Python 프로그래밍 언어는?"})

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

문제: 여러 도구를 순차적으로 사용하여 작업을 완료해야 하는 복잡한 에이전트

from typing import Dict, Any
import json

class ToolAgentState(TypedDict):
    task: str
    plan: list
    executed_tools: list
    observations: list
    final_result: str

class ToolRegistry:
    @staticmethod
    def calculate(expression: str):
        try:
            return {"result": eval(expression), "status": "success"}
        except:
            return {"result": None, "status": "error"}

    @staticmethod
    def search(query: str):
        return {"results": [f"검색 결과: {query}"], "status": "success"}

def plan_node(state: ToolAgentState):
    # 작업 계획 생성
    plan = [
        {"tool": "search", "params": {"query": state["task"]}},
        {"tool": "calculate", "params": {"expression": "2+2"}}
    ]
    return {"plan": plan, "executed_tools": [], "observations": []}

def execute_node(state: ToolAgentState):
    # 도구 실행
    current_step = state["plan"][len(state["executed_tools"])]
    tool_name = current_step["tool"]
    params = current_step["params"]

    tool_registry = ToolRegistry()
    if hasattr(tool_registry, tool_name):
        result = getattr(tool_registry, tool_name)(**params)
        observation = {"tool": tool_name, "result": result}

        return {
            "executed_tools": state["executed_tools"] + [current_step],
            "observations": state["observations"] + [observation]
        }

def observe_node(state: ToolAgentState):
    # 관찰 및 결정
    if len(state["executed_tools"]) == len(state["plan"]):
        final_result = "모든 작업 완료"
        return {"final_result": final_result}
    return {"final_result": None}

# 워크플로우 정의
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("observe", observe_node)

def route_plan(state: ToolAgentState):
    if len(state["executed_tools"]) < len(state["plan"]):
        return "execute"
    return "observe"

workflow.add_edge("plan", "execute")
workflow.add_conditional_edges("execute", route_plan)
workflow.add_edge("observe", END)

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

문제: 인간의 개입이 필요한 결정을 할 때, 사용자 검토 및 승인을 포함하는 워크플로우

import asyncio
from typing import Optional

class HumanInLoopState(TypedDict):
    task: str
    intermediate_results: list
    human_review: Optional[bool]
    human_comment: Optional[str]
    final_decision: Optional[bool]

def generate_intermediate_results(state: HumanInLoopState):
    # 중간 결과 생성
    results = [
        {"type": "analysis", "content": "분석 내용"},
        {"type": "recommendation", "content": "추천 내용"}
    ]
    return {"intermediate_results": results}

def human_review_node(state: HumanInLoopState):
    # 인간 검토 (블로킹)
    print("중간 결과:")
    for result in state["intermediate_results"]:
        print(f"- {result['type']}: {result['content']}")

    # 실제 애플리케이션에서는 이 부분이 사용자 인터페이스와 연결됨
    human_input = input("승인하시겠습니까? (y/n): ")
    approved = human_input.lower() == 'y'

    return {
        "human_review": approved,
        "human_comment": "사용자 검토 완료"
    }

def make_final_decision(state: HumanInLoopState):
    # 최종 결정
    if state["human_review"] is True:
        return {"final_decision": True, "task": "완료됨"}
    return {"final_decision": False, "task": "거부됨"}

# 워크플로우 정의
workflow = StateGraph(HumanInLoopState)
workflow.add_node("generate", generate_intermediate_results)
workflow.add_node("review", human_review_node)
workflow.add_node("decide", make_final_decision)

workflow.add_edge("generate", "review")
workflow.add_edge("review", "decide")
workflow.add_edge("decide", END)

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

문제: 동시에 여러 작업을 처리하고 결과를 집계하는 병렬 워크플로우

from concurrent.futures import ThreadPoolExecutor
import time

class ParallelAgentState(TypedDict):
    tasks: list
    results: list
    aggregated_result: str

def fan_out_node(state: ParallelAgentState):
    # 작업 분산
    return {"tasks": state["tasks"]}

def process_task(state: ParallelAgentState, task: str):
    # 단일 작업 처리
    time.sleep(1)  # 시뮬레이션
    return {"task": task, "result": f"처리된 {task}"}

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

# 병렬 처리 함수
async def parallel_execute(tasks: list):
    results = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(process_task, task) for task in tasks]
        for future in futures:
            results.append(future.result())
    return results

# 워크플로우 정의
workflow = StateGraph(ParallelAgentState)
workflow.add_node("fan_out", fan_out_node)
workflow.add_node("aggregate", aggregate_node)

def route_fan_out(state: ParallelAgentState):
    return "fan_out"

workflow.add_edge("fan_out", "aggregate")
workflow.add_edge("aggregate", END)