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

LangGraph 아키텍처 개요

LangGraph는 상태 기반 워크플로우를 구현하는 데 특화된 프레임워크로, 다음과 같은 핵심 구성 요소로 작동합니다:

Nodes (노드): 각각의 처리 단계
Edges (엣지): 노드 간의 전이 조건
State (상태): 워크플로우 전체에서 공유되는 데이터 스토어
Checkpointing (체크포인팅): 상태 저장 및 복구 기능

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

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

# 기본 워크플로우 구성
graph = StateGraph(GraphState)
graph.add_node("node1", node_function)
graph.add_edge("node1", END)

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

유저가 질문을 입력하면, RAG 워크플로우는 문서 검색 → 생성 → 결과 검증 단계로 작동합니다.

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnableLambda
import operator

class RAGState(TypedDict):
    question: str
    context: str
    answer: str
    validation: bool

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

    def retrieve(self, state: RAGState):
        # 문서 검색
        docs = self.vector_store.similarity_search(state["question"])
        context = "\n".join([doc.page_content for doc in docs])
        return {"context": context}

    def generate(self, state: RAGState):
        # 답변 생성
        prompt = f"""
        질문: {state['question']}
        문맥: {state['context']}

        위 정보를 바탕으로 답변을 생성해주세요.
        """
        response = self.llm.invoke([HumanMessage(content=prompt)])
        return {"answer": response.content}

    def validate(self, state: RAGState):
        # 답변 검증
        prompt = f"""
        질문: {state['question']}
        답변: {state['answer']}
        문맥: {state['context']}

        위 답변이 문맥에 기반하여 정확한지 검증해주세요.
        """
        response = self.llm.invoke([HumanMessage(content=prompt)])
        return {"validation": "정확" in response.content}

# 워크플로우 실행
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", RAGAgent(vector_store, llm).retrieve)
workflow.add_node("generate", RAGAgent(vector_store, llm).generate)
workflow.add_node("validate", RAGAgent(vector_store, llm).validate)

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

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

복잡한 작업을 수행하기 위해 여러 도구를 활용하는 에이전트입니다.

from langchain.tools import Tool
from langchain_openai import OpenAI
from typing import List

class ToolAgentState(TypedDict):
    plan: List[str]
    execution_log: List[str]
    results: List[dict]
    decision: str

class ToolAgent:
    def __init__(self, tools: List[Tool]):
        self.tools = tools
        self.llm = OpenAI(temperature=0)

    def plan(self, state: ToolAgentState):
        # 작업 계획 수립
        prompt = f"""
        다음 작업을 수행하기 위한 계획을 세우세요:
        - 파일 읽기/쓰기
        - 시스템 명령 실행
        - 데이터 분석

        상태: {state}

        계획을 세우고, 각 단계에 대해 실행 순서를 명확히 해주세요.
        """
        response = self.llm.invoke(prompt)
        plan = [step.strip() for step in response.split('\n') if step.strip()]
        return {"plan": plan}

    def execute(self, state: ToolAgentState):
        # 도구 실행
        results = []
        for action in state["plan"]:
            result = self._execute_tool(action)
            results.append({"action": action, "result": result})
        return {"results": results, "execution_log": state["plan"]}

    def _execute_tool(self, action: str):
        # 실제 도구 실행 로직
        if "read_file" in action:
            # 파일 읽기
            return "파일 내용"
        elif "system_command" in action:
            # 시스템 명령 실행
            return "명령 실행 결과"
        else:
            return "도구 실행 결과"

    def observe_and_decide(self, state: ToolAgentState):
        # 실행 결과 관찰 및 결정
        prompt = f"""
        다음 실행 결과를 기반으로 결정을 내리세요:
        결과들: {state['results']}

        결정: 실행 성공/실패
        """
        response = self.llm.invoke(prompt)
        return {"decision": response.content}

# 도구 정의
tools = [
    Tool(
        name="file_reader",
        func=lambda x: "file content",
        description="파일 내용 읽기"
    ),
    Tool(
        name="system_command",
        func=lambda x: "command result",
        description="시스템 명령 실행"
    )
]

workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", ToolAgent(tools).plan)
workflow.add_node("execute", ToolAgent(tools).execute)
workflow.add_node("observe", ToolAgent(tools).observe_and_decide)

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

템플릿 3: 인간-함께-작업 워크플로우 (일시정지 → 검토 → 계속)

사용자가 개입할 수 있는 인간-함께-작업 워크플로우를 구현합니다.

from datetime import datetime
from langgraph.graph import StateGraph, END

class HumanInLoopState(TypedDict):
    task: str
    user_input: str
    agent_output: str
    status: str
    timestamp: str

class HumanInLoopAgent:
    def __init__(self):
        self.checkpoint_manager = MemorySaver()

    def process_task(self, state: HumanInLoopState):
        # 작업 처리
        prompt = f"""
        작업: {state['task']}
        결과를 생성하세요.
        """
        # LLM 호출 (예시)
        agent_output = "작업 결과"
        return {"agent_output": agent_output, "status": "pending_review"}

    def pause_for_review(self, state: HumanInLoopState):
        # 검토를 위한 일시정지
        print(f"작업을 검토해야 합니다: {state['agent_output']}")
        return {"status": "awaiting_review"}

    def continue_workflow(self, state: HumanInLoopState):
        # 사용자 검토 후 계속
        return {"status": "completed"}

# 워크플로우 구현
workflow = StateGraph(HumanInLoopState)
workflow.add_node("process", HumanInLoopAgent().process_task)
workflow.add_node("pause", HumanInLoopAgent().pause_for_review)
workflow.add_node("continue", HumanInLoopAgent().continue_workflow)

workflow.add_edge("process", "pause")
workflow.add_edge("pause", "continue")
workflow.add_edge("continue", END)

# 실행 예제
agent = HumanInLoopAgent()

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

여러 작업을 병렬로 처리하고 결과를 집계합니다.

python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(TypedDict):
    tasks: List[dict]
    results: List[dict]
    aggregated_result: dict

class ParallelAgent:
    def __init__(self):
        pass

    def fan_out(self, state: ParallelAgentState):
        # 작업 분산
        tasks = [
            {"id": 1, "type": "data_processing", "input": "data1"},
            {"id": 2, "type": "analysis", "input": "data2"},
            {"id": 3, "type": "reporting", "input": "data3"}
        ]
        return {"tasks": tasks}

    def process_parallel(self, state: ParallelAgentState):
        # 병렬 처리
        def process_task(task):
            # 각 작업별 처리 로직
            if task["type"] == "data_processing":
                return {"id": task["id"], "result": "processing_complete"}
            elif task["type"] == "analysis":
                return {"id": task["id"], "result": "analysis_complete"}
            elif task

---

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