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

소개

LangGraph는 LangChain과 함께 사용되는 고급 AI 에이전트 구축 프레임워크입니다. 이 가이드에서는 실제 개발자가 문제를 해결할 수 있는 4가지 핵심 워크플로우 템플릿을 제공합니다. 각 템플릿은 실제 프로덕션 환경에서 사용될 수 있는 최적화된 구조를 갖추고 있으며, Python 코드와 실제 사용 예시를 포함합니다.

1. LangGraph 아키텍처 개요

LangGraph는 그래프 기반의 워크플로우 시스템으로, 다음 구성 요소로 구성됩니다:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 상태 정의
class State(TypedDict):
    messages: list
    current_step: str
    tool_calls: dict

# 노드 정의
def retrieve_node(state):
    # 검색 로직
    return {"messages": [f"Retrieved: {state['query']}"]}

def generate_node(state):
    # 생성 로직
    return {"messages": [f"Generated: {state['query']}"]}

# 그래프 생성
workflow = StateGraph(State)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)

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

# 체크포인트 저장
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

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

이 템플릿은 검색-생성-검증 파이프라인을 구현합니다. 실시간 데이터 검색과 결과 검증 기능을 포함합니다.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
import json

class RAGState(TypedDict):
    query: str
    retrieved_docs: list
    generated_answer: str
    validation_result: dict
    current_step: Literal["retrieve", "generate", "validate"]

class RAGAgent:
    def __init__(self, vectorstore, llm):
        self.vectorstore = vectorstore
        self.llm = llm

    def retrieve(self, state):
        docs = self.vectorstore.similarity_search(state["query"], k=3)
        return {
            "retrieved_docs": [doc.page_content for doc in docs],
            "current_step": "generate"
        }

    def generate(self, state):
        context = "\n".join(state["retrieved_docs"])
        prompt = PromptTemplate.from_template("""
        주어진 컨텍스트를 기반으로 질문에 답변하세요.
        컨텍스트: {context}
        질문: {query}
        답변:
        """)

        chain = prompt | self.llm
        answer = chain.invoke({
            "context": context,
            "query": state["query"]
        })

        return {
            "generated_answer": answer.content,
            "current_step": "validate"
        }

    def validate(self, state):
        # 간단한 검증 로직
        validation = {
            "is_valid": len(state["generated_answer"]) > 10,
            "confidence": 0.85,
            "feedback": "검증 완료"
        }
        return {
            "validation_result": validation,
            "current_step": "end"
        }

    def create_graph(self):
        workflow = StateGraph(RAGState)
        workflow.add_node("retrieve", self.retrieve)
        workflow.add_node("generate", self.generate)
        workflow.add_node("validate", self.validate)

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

        return workflow.compile()

# 사용 예시
# rag_agent = RAGAgent(vectorstore, ChatOpenAI(model="gpt-4"))
# app = rag_agent.create_graph()
# result = app.invoke({"query": "Python 데이터 분석 방법"})

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

이 템플릿은 작업 계획, 실행, 관찰 및 결정을 수행하는 다중 도구 에이전트입니다. 복잡한 작업을 단계별로 분해하고 실행합니다.

from typing import TypedDict, Literal, Any
from langgraph.graph import StateGraph, END
from langchain.tools import Tool
from langchain_openai import ChatOpenAI

class ToolAgentState(TypedDict):
    task: str
    plan: list
    execution_log: list
    current_step: Literal["plan", "execute", "observe", "decide"]
    results: dict

class ToolAgent:
    def __init__(self, tools, llm):
        self.tools = tools
        self.llm = llm

    def plan(self, state):
        # 작업 계획 생성
        plan_prompt = """
        다음 작업을 수행하는 데 필요한 단계를 계획하세요:
        {task}

        가능한 도구 목록:
        {tools}

        계획을 JSON 형식으로 반환하세요.
        """

        plan = self.llm.invoke(plan_prompt.format(
            task=state["task"],
            tools=[tool.name for tool in self.tools]
        ))

        plan_data = json.loads(plan.content)
        return {
            "plan": plan_data["steps"],
            "current_step": "execute"
        }

    def execute(self, state):
        # 각 단계 실행
        step = state["plan"][0] if state["plan"] else None
        if not step:
            return {"current_step": "decide"}

        # 도구 실행
        tool_result = self.execute_tool(step["tool"], step["input"])

        return {
            "execution_log": state["execution_log"] + [tool_result],
            "plan": state["plan"][1:] if len(state["plan"]) > 1 else [],
            "current_step": "observe"
        }

    def execute_tool(self, tool_name, input_data):
        # 특정 도구 실행
        tool = next((t for t in self.tools if t.name == tool_name), None)
        if not tool:
            return {"error": f"도구 {tool_name}를 찾을 수 없습니다"}

        try:
            result = tool.run(input_data)
            return {"tool": tool_name, "input": input_data, "result": result}
        except Exception as e:
            return {"tool": tool_name, "error": str(e)}

    def observe(self, state):
        # 실행 결과 관찰
        if state["execution_log"]:
            last_result = state["execution_log"][-1]
            if "error" in last_result:
                # 오류 처리
                return {"current_step": "decide"}

        return {"current_step": "execute"}

    def decide(self, state):
        # 최종 결정
        if not state["plan"]:
            return {"current_step": "end"}
        return {"current_step": "plan"}

# 사용 예시
tools = [
    Tool(name="search", func=lambda x: f"검색 결과: {x}", description="검색 도구"),
    Tool(name="calculate", func=lambda x: f"계산 결과: {x}", description="계산 도구")
]
# tool_agent = ToolAgent(tools, ChatOpenAI(model="gpt-4"))
# app = tool_agent.create_graph()

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

이 템플릿은 인간 검토를 포함하는 워크플로우를 구현합니다. 중요한 결정이나 검증 작업에서 인간의 개입이 필요한 경우 사용됩니다.

python
from typing import TypedDict, Literal, Any
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class HumanInLoopState(TypedDict):
    task: str
    generated_output: str
    human_review: dict
    current_step: Literal["generate", "pause", "review", "continue"]
    is_approved: bool

class HumanInLoopAgent:
    def __init__(self, llm):
        self.llm = llm

    def generate(self, state):
        # 작업 생성
        prompt = f"다음 작업을 수행하세요:\n{state['task']}"
        result = self.llm.invoke(prompt)

        return {
            "generated_output": result.content,
            "current_step": "pause"
        }

    def pause(self, state):
        # 인간 검토를 위한 중단
        return {
            "current_step": "review"
        }

    def review(self, state):
        # 인간 검토
        # 실제 구현에서는 외부 시스템 또는 UI를 통해 인간이 검토
        # 여기서는 예시로 설정
        human_decision = input("

---

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