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

개요

LangGraph는 LangChain과 함께 사용할 수 있는 고급 워크플로우 엔진으로, AI 에이전트의 복잡한 실행 흐름을 정의하고 관리할 수 있게 해줍니다. 이 가이드는 실전에서 사용 가능한 4가지 핵심 워크플로우 템플릿을 제공하여, Python 개발자들이 로컬 환경에서 안정적이고 확장 가능한 에이전트 시스템을 구축할 수 있도록 도와줍니다.

1. LangGraph 아키텍처 개요

LangGraph는 다음과 같은 핵심 구성 요소로 이루어져 있습니다:

노드 (Nodes)

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

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

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

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

엣지 (Edges)

# 조건부 엣지
def should_continue(state):
    if len(state["messages"]) > 3:
        return END
    return "generate"

# 그래프 정의
graph = StateGraph(State)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_edge("retrieve", "generate")
graph.add_conditional_edges("generate", should_continue)

상태 (State)

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    tool_calls: list
    error: str
    user_input: str

체크포인트 (Checkpointing)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = StateGraph(State, checkpointer=memory)

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

이 템플릿은 문서 검색, 생성, 검증의 기본적인 RAG 흐름을 구현합니다:

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing import TypedDict
import operator

class RagState(TypedDict):
    query: str
    documents: list
    response: str
    validation: bool

# 검색 노드
def retrieve(state: RagState):
    # 문서 검색 (예: ChromaDB 또는 Elasticsearch)
    documents = ["document1", "document2", "document3"]  # 실제 구현에서는 검색 로직 추가
    return {"documents": documents}

# 생성 노드
def generate(state: RagState):
    llm = ChatOpenAI(model="gpt-4")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("user", "Based on these documents: {documents}, answer the question: {query}")
    ])

    chain = prompt | llm
    response = chain.invoke({
        "documents": state["documents"],
        "query": state["query"]
    })

    return {"response": response.content}

# 검증 노드
def validate(state: RagState):
    # 생성된 응답 검증
    # 예: 정확도, 관련성 등을 기준으로 검증
    validation_result = True  # 실제 로직에서 검증 수행
    return {"validation": validation_result}

# 그래프 정의
def create_rag_workflow():
    workflow = StateGraph(RagState)
    workflow.add_node("retrieve", retrieve)
    workflow.add_node("generate", generate)
    workflow.add_node("validate", validate)

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

    # 검증 결과에 따라 분기
    def route_validation(state):
        if state["validation"]:
            return END
        else:
            return "generate"  # 재시도

    workflow.add_conditional_edges("validate", route_validation)

    return workflow.compile()

# 사용 예시
workflow = create_rag_workflow()
result = workflow.invoke({"query": "What is LangGraph?"})
print(result["response"])

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

복잡한 작업을 여러 툴을 사용하여 처리하는 에이전트를 위한 템플릿입니다:

from langchain.tools import Tool
from langgraph.graph import StateGraph, END
from typing import List

class ToolAgentState(TypedDict):
    task: str
    plan: List[str]
    execution_log: List[str]
    observation: str
    final_decision: str

# 툴 정의
def web_search(query: str):
    return f"Search results for '{query}'"

def calculate(num1: int, num2: int, operation: str):
    if operation == "add":
        return num1 + num2
    elif operation == "multiply":
        return num1 * num2

# 계획 노드
def plan(state: ToolAgentState):
    tasks = [
        "Search information about the task",
        "Calculate required values",
        "Generate final answer"
    ]
    return {"plan": tasks}

# 실행 노드
def execute(state: ToolAgentState):
    # 각 단계를 실행하고 결과 기록
    execution_log = []

    for task in state["plan"]:
        if "search" in task.lower():
            result = web_search(state["task"])
            execution_log.append(f"Search result: {result}")
        elif "calculate" in task.lower():
            result = calculate(10, 20, "add")
            execution_log.append(f"Calculation result: {result}")

    return {"execution_log": execution_log}

# 관찰 노드
def observe(state: ToolAgentState):
    # 실행 결과를 관찰하고 판단
    observation = f"Executed {len(state['execution_log'])} steps successfully"
    return {"observation": observation}

# 결정 노드
def decide(state: ToolAgentState):
    # 최종 결정을 내림
    final_decision = "Task completed successfully"
    return {"final_decision": final_decision}

# 그래프 정의
def create_tool_agent_workflow():
    workflow = StateGraph(ToolAgentState)
    workflow.add_node("plan", plan)
    workflow.add_node("execute", execute)
    workflow.add_node("observe", observe)
    workflow.add_node("decide", decide)

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

    return workflow.compile()

# 사용 예시
workflow = create_tool_agent_workflow()
result = workflow.invoke({"task": "Find information and calculate total"})
print(result["final_decision"])

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

사용자 검토가 필요한 작업을 처리하는 템플릿:

python
from langgraph.graph import StateGraph, END
from typing import Optional

class HumanInLoopState(TypedDict):
    task_description: str
    generated_output: str
    user_review: Optional[str]
    approval: Optional[bool]
    retry_count: int

# 생성 노드
def generate_output(state: HumanInLoopState):
    # AI가 출력 생성
    output = f"Generated output for task: {state['task_description']}"
    return {"generated_output": output}

# 사용자 검토 노드
def review_output(state: HumanInLoopState):
    # 사용자 입력을 기다림 (실제 구현에서는 API 호출 또는 큐 시스템 사용)
    # 여기서는 가상의 사용자 입력을 가정
    user_review = "User approved"  # 실제 구현에서는 외부 입력 처리
    approval = True  # 사용자 승인 여부
    return {"user_review": user_review, "approval": approval}

# 재시도 로직
def retry_if_needed(state: HumanInLoopState):
    if not state["approval"] and state["retry_count"] < 3:
        return "generate_output"  # 재시도
    elif not state["approval"]:
        return "reject"  # 최대 재시도 횟수 초과
    else:
        return END  # 승인됨

# 거부 노드
def reject(state: HumanInLoopState):
    return {"generated_output": "Task rejected due to user disapproval"}

# 그래프 정의
def create_human_in_loop_workflow():
    workflow = StateGraph(HumanInLoopState)
    workflow.add_node("generate_output", generate_output)
    workflow.add_node("review_output", review_output)
    workflow.add_node("reject", reject)



---

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