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

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반의 워크플로우 엔진으로, 노드(Node), 엣지(Edge), 상태(State), 체크포인트(Checkpointing)로 구성됩니다.

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

# 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

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

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

실제 데이터베이스와 LLM을 통합하는 데 필요한 최소한의 RAG 구조입니다:

import sqlite3
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

class SimpleRAGAgent:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.llm = ChatOpenAI(model="gpt-4o-mini")

    def retrieve(self, state: AgentState) -> AgentState:
        messages = state["messages"]
        query = messages[-1].content

        # 데이터베이스에서 관련 문서 검색
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT content FROM documents 
            WHERE content LIKE ? 
            ORDER BY similarity(?, content) DESC 
            LIMIT 5
        """, (f"%{query}%", query))

        docs = cursor.fetchall()
        conn.close()

        return {"documents": [doc[0] for doc in docs]}

    def generate(self, state: AgentState) -> AgentState:
        # 검색된 문서와 질문으로 생성
        docs = state["documents"]
        messages = state["messages"]

        prompt = f"""
        문서 내용: {docs}
        질문: {messages[-1].content}
        답변:
        """

        response = self.llm.invoke([HumanMessage(content=prompt)])
        return {"messages": [response]}

    def validate(self, state: AgentState) -> AgentState:
        # 생성된 답변 검증
        response = state["messages"][-1].content
        if len(response) < 10:
            return {"messages": [HumanMessage(content="답변이 너무 짧습니다. 다시 시도해주세요.")]}
        return {"next": "continue"}

# 워크플로우 구축
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", SimpleRAGAgent.retrieve)
workflow.add_node("generate", SimpleRAGAgent.generate)
workflow.add_node("validate", SimpleRAGAgent.validate)
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
workflow.add_edge("validate", END)
workflow.set_entry_point("retrieve")

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

다수의 도구를 활용하여 복잡한 작업을 수행하는 에이전트:

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

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

@tool
def calculate_math(expression: str) -> str:
    """수학 계산 도구"""
    try:
        result = eval(expression)
        return f"결과: {result}"
    except:
        return "계산 오류"

@tool
def save_data(key: str, value: str) -> str:
    """데이터 저장 도구"""
    return f"데이터 저장: {key} = {value}"

class MultiToolAgent:
    def __init__(self):
        self.tools = [search_web, calculate_math, save_data]
        self.llm = ChatOpenAI(model="gpt-4o-mini")

    def plan(self, state: AgentState) -> AgentState:
        # 현재 상태에서 필요한 작업 계획
        messages = state["messages"]
        task = messages[-1].content

        prompt = f"""
        작업: {task}
        다음 작업을 계획하세요:
        1. 어떤 도구가 필요한가?
        2. 실행 순서는?
        3. 예상 결과는?
        """

        plan = self.llm.invoke([HumanMessage(content=prompt)])
        return {"plan": plan.content}

    def execute(self, state: AgentState) -> AgentState:
        # 계획된 작업 실행
        plan = state["plan"]
        # 간단한 도구 실행 구현
        return {"execution_results": "실행 완료"}

    def observe(self, state: AgentState) -> AgentState:
        # 실행 결과 관찰
        results = state["execution_results"]
        return {"observation": f"관찰 결과: {results}"}

    def decide(self, state: AgentState) -> AgentState:
        # 다음 단계 결정
        observation = state["observation"]
        if "실패" in observation:
            return {"next": "retry"}
        return {"next": "continue"}

# 워크플로우 구축
workflow = StateGraph(AgentState)
workflow.add_node("plan", MultiToolAgent.plan)
workflow.add_node("execute", MultiToolAgent.execute)
workflow.add_node("observe", MultiToolAgent.observe)
workflow.add_node("decide", MultiToolAgent.decide)
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
workflow.add_edge("decide", END)
workflow.set_entry_point("plan")

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

사람의 검토를 포함한 작업 흐름:

import time
from typing import Literal

class HumanInLoopAgent:
    def __init__(self, model="gpt-4o-mini"):
        self.llm = ChatOpenAI(model=model)
        self.checkpoint = None

    def process(self, state: AgentState) -> AgentState:
        # AI 처리
        messages = state["messages"]
        prompt = f"처리 요청: {messages[-1].content}"
        response = self.llm.invoke([HumanMessage(content=prompt)])
        return {"ai_response": response.content}

    def pause_for_review(self, state: AgentState) -> AgentState:
        # 인간 검토를 위한 일시정지
        ai_response = state["ai_response"]
        return {
            "status": "waiting_for_review",
            "review_request": f"검토 요청:\n{ai_response}"
        }

    def continue_workflow(self, state: AgentState) -> AgentState:
        # 검토 후 계속 진행
        if state["review_approved"]:
            return {"messages": [HumanMessage(content="검토 승인")]}
        return {"messages": [HumanMessage(content="검토 거부")]}

    def retry_with_feedback(self, state: AgentState) -> AgentState:
        # 피드백 기반 재시도
        feedback = state["feedback"]
        return {"messages": [HumanMessage(content=f"피드백 반영: {feedback}")]}

    def wait_for_human_input(self) -> AgentState:
        # 인간 입력 대기
        print("사람의 입력을 기다리는 중...")
        human_input = input("사람 입력: ")
        return {"messages": [HumanMessage(content=human_input)]}

# 인간 검토 워크플로우
workflow = StateGraph(AgentState)
workflow.add_node("process", HumanInLoopAgent.process)
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)
workflow.set_entry_point("process")

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

여러 작업을 동시에 처리 후 결과 집계:


python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgent:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers
        self.llm = ChatOpenAI(model="gpt-4o-mini")

    def fan_out(self, state: AgentState) -> AgentState:
        # 작업 분기
        messages = state["messages"]
        task = messages[-1].content

        # 병렬 처리를 위한 작업 목록 생성
        tasks = [
            f"분석: {task} - 1",
            f"분석: {task} - 2", 
            f"분석: {task} - 3"
        ]
        return {"parallel_tasks":

---

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

推荐订阅源

DEV Community