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

소개

이 가이드는 Python 개발자들이 LangChain/LangGraph를 사용하여 AI 에이전트를 구축할 때 활용할 수 있는 실질적인 워크플로우 템플릿들을 제공합니다. 각 템플릿은 실제 문제를 해결하도록 설계되었으며, $3-$7 범위의 가격으로도 충분히 가치 있는 실용성을 가지고 있습니다.

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반의 워크플로우 시스템으로, 다음과 같은 핵심 구성 요소로 이루어져 있습니다:

• Nodes: 워크플로우의 각 단계
• Edges: 노드 간의 전이 조건
• State: 모든 노드가 접근할 수 있는 공유 상태
• Checkpointing: 상태를 저장하고 복구하는 기능

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

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

# 그래프 생성
workflow = StateGraph(State)

2. 템플릿 1: 간단한 RAG 에이전트 (Retrieve → Generate → Validate)

이 템플릿은 문서 검색 후 생성 후 유효성 검사를 수행하는 기본적인 RAG 워크플로우입니다.

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

class RAGState(TypedDict):
    query: str
    retrieved_docs: list
    generated_answer: str
    validation_result: bool

def retrieve_node(state: RAGState):
    # 문서 검색 로직 (예: FAISS 또는 Pinecone)
    vector_store = FAISS.load_local("vector_store", OpenAIEmbeddings())
    docs = vector_store.similarity_search(state["query"])
    return {"retrieved_docs": docs}

def generate_node(state: RAGState):
    # 검색된 문서를 기반으로 답변 생성
    prompt = PromptTemplate.from_template(
        "문서: {docs}\n질문: {query}\n답변:"
    )
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    answer = chain.invoke({
        "docs": "\n".join([doc.page_content for doc in state["retrieved_docs"]]),
        "query": state["query"]
    })
    return {"generated_answer": answer}

def validate_node(state: RAGState):
    # 답변 유효성 검사 (예: LLM 기반)
    validation_prompt = PromptTemplate.from_template(
        "질문: {query}\n답변: {answer}\n정확한 답변인가? (예/아니오)"
    )
    validation_chain = validation_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    result = validation_chain.invoke({
        "query": state["query"],
        "answer": state["generated_answer"]
    })
    return {"validation_result": result.strip().lower() == "예"}

# 그래프 구성
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("validate", validate_node)

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

# 유효성 검사 결과에 따라 전이
workflow.add_conditional_edges(
    "validate",
    lambda state: "valid" if state["validation_result"] else "invalid",
    {
        "valid": "generate",
        "invalid": "retrieve"
    }
)

# 컴파일 및 실행
app = workflow.compile()
result = app.invoke({"query": "Python의 상속 개념은 무엇인가?"})

3. 템플릿 2: 멀티-도구 에이전트 (Plan → Execute → Observe → Decide)

이 템플릿은 도구 사용 계획을 세우고 실행, 관찰 후 결정을 내리는 에이전트입니다.

from langchain.tools import Tool
from langchain_core.messages import HumanMessage

class ToolAgentState(TypedDict):
    plan: str
    execution_results: list
    observation: str
    decision: str
    tools: list

def plan_node(state: ToolAgentState):
    # 실행 계획 생성
    planning_prompt = PromptTemplate.from_template(
        "사용자의 요청: {request}\n도구 목록: {tools}\n계획 생성:"
    )
    plan_chain = planning_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    plan = plan_chain.invoke({
        "request": state["request"],
        "tools": [tool.name for tool in state["tools"]]
    })
    return {"plan": plan}

def execute_node(state: ToolAgentState):
    # 계획에 따라 도구 실행
    results = []
    for tool in state["tools"]:
        if tool.name in state["plan"]:
            try:
                result = tool.run(state["request"])
                results.append({"tool": tool.name, "result": result})
            except Exception as e:
                results.append({"tool": tool.name, "error": str(e)})
    return {"execution_results": results}

def observe_node(state: ToolAgentState):
    # 실행 결과 관찰
    observation_prompt = PromptTemplate.from_template(
        "계획: {plan}\n실행 결과: {results}\n관찰:"
    )
    obs_chain = observation_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    observation = obs_chain.invoke({
        "plan": state["plan"],
        "results": str(state["execution_results"])
    })
    return {"observation": observation}

def decide_node(state: ToolAgentState):
    # 결정 생성
    decision_prompt = PromptTemplate.from_template(
        "관찰 내용: {observation}\n최종 결정:"
    )
    dec_chain = decision_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    decision = dec_chain.invoke({"observation": state["observation"]})
    return {"decision": decision}

# 도구 정의
tools = [
    Tool(name="search_web", func=lambda x: f"웹 검색 결과: {x}"),
    Tool(name="calculate", func=lambda x: f"계산 결과: {eval(x)}"),
    Tool(name="lookup_info", func=lambda x: f"정보 조회: {x}")
]

# 그래프 구성
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("observe", observe_node)
workflow.add_node("decide", decide_node)

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

app = workflow.compile()
result = app.invoke({
    "request": "3+5*2의 결과는?",
    "tools": tools
})

4. 템플릿 3: 인간-인터페이스 워크플로우 (Pause → Review → Continue)

이 템플릿은 인간의 검토를 포함한 워크플로우입니다.

python
import asyncio
from datetime import datetime

class HumanLoopState(TypedDict):
    task: str
    generated_output: str
    human_review: str
    final_output: str
    review_status: str  # pending, approved, rejected

def generate_node(state: HumanLoopState):
    # 작업 생성
    prompt = PromptTemplate.from_template("작업 설명: {task}\n생성된 내용:")
    chain = prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    output = chain.invoke({"task": state["task"]})
    return {"generated_output": output}

def human_review_node(state: HumanLoopState):
    # 인간 검토 (실제로는 외부 시스템 또는 사용자 입력 필요)
    # 여기서는 가상의 검토 결과를 반환
    review_prompt = PromptTemplate.from_template(
        "작업: {task}\n생성된 내용: {output}\n검토 요청:"
    )
    review_chain = review_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
    review = review_chain.invoke({
        "task": state["task"],
        "output": state["generated_output"]
    })

    # 검토 상태를 설정 (실제 구현에서는 사용자 입력 기다림)
    status = "approved"  # 또는 "rejected"
    return {
        "human_review": review,
        "review_status": status
    }

def finalize_node(state: HumanLoopState):
    # 최종 결과 생성
    if state["review_status"] == "approved":
        return {"final_output": state["generated_output"]}
    else:
        # 재생성 로직
        return {"final_output": "재작성된 내용"}

# 그래프 구성
workflow = StateGraph(HumanLoopState)
workflow.add_node("generate", generate_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("finalize", finalize_node)

workflow.set_entry_point("generate")
workflow.add_edge("generate

---

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