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

개요

이 가이드는 Python 개발자들을 위한 LangGraph 기반 AI 에이전트 워크플로우 템플릿을 제공합니다. 3-7달러 범위의 실용적인 도구로, 로컬 AI 솔루션과 에이전트 기반 워크플로우를 구축할 때 발생하는 핵심 문제들을 해결합니다.

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반의 워크플로우 시스템으로, 다음 구성 요소로 이루어집니다:

핵심 구성 요소:

• Nodes: 각 작업 단계 (함수 또는 클래스)
• Edges: 노드 간의 연결과 조건
• State: 워크플로우 상태 (데이터 구조)
• Checkpointing: 상태 저장 및 복원 기능

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

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

# 상태 기반 그래프 생성
workflow = StateGraph(GraphState)
workflow.add_node("process", process_node)
workflow.add_edge("process", "validate")
workflow.add_conditional_edges(
    "validate",
    route_message,
    {"continue": "process", "end": END}
)

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

이 템플릿은 로컬 RAG 시스템에서 가장 일반적인 패턴을 구현합니다. 특히 긴 문맥 처리를 위한 최적화를 포함합니다.

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
import operator

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

    def retrieve(self, state):
        query = state["query"]
        # 상하위 3개 문서 검색 (성능 최적화)
        docs = self.vectorstore.similarity_search(query, k=3)
        context = "\n\n".join([doc.page_content for doc in docs])
        return {"context": context, "documents": docs}

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

        질문: {query}
        """)

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

    def validate(self, state):
        # 답변 검증 로직
        if len(state["answer"]) < 10:
            return {"validation": "fail", "retry": True}
        return {"validation": "pass", "retry": False}

# 워크플로우 구성
def create_rag_workflow(agent):
    workflow = StateGraph(RAGState)
    workflow.add_node("retrieve", agent.retrieve)
    workflow.add_node("generate", agent.generate)
    workflow.add_node("validate", agent.validate)

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

    workflow.add_conditional_edges(
        "validate",
        lambda x: "retry" if x["retry"] else "end",
        {"retry": "retrieve", "end": END}
    )
    return workflow.compile()

# 사용 예시
rag_agent = RAGAgent(vectorstore, llm)
rag_workflow = create_rag_workflow(rag_agent)
result = rag_workflow.invoke({"query": "AI 기술의 미래는?"})

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

복잡한 작업을 위한 계획-실행-관찰 패턴을 구현하여, 다양한 도구를 효율적으로 활용합니다.

from langchain.tools import Tool
from langchain_core.runnables import RunnableLambda

class MultiToolAgent:
    def __init__(self, tools):
        self.tools = tools
        self.tool_map = {tool.name: tool for tool in tools}

    def plan(self, state):
        # 작업 계획 생성
        prompt = PromptTemplate.from_template("""
        다음 작업을 수행할 도구를 선택하세요:
        {tasks}

        사용 가능한 도구: {available_tools}
        """)

        # 도구 선택 및 계획 생성
        plan = self.llm.invoke({
            "tasks": state["tasks"],
            "available_tools": [t.name for t in self.tools]
        })
        return {"plan": plan, "execution_plan": self.parse_plan(plan)}

    def execute(self, state):
        # 계획된 작업 실행
        results = []
        for task in state["execution_plan"]:
            tool = self.tool_map.get(task["tool"])
            if tool:
                try:
                    result = tool.run(task["input"])
                    results.append({"tool": task["tool"], "result": result})
                except Exception as e:
                    results.append({"tool": task["tool"], "error": str(e)})
        return {"execution_results": results}

    def observe(self, state):
        # 실행 결과 관찰 및 피드백 수집
        return {"feedback": self.analyze_results(state["execution_results"])}

    def decide(self, state):
        # 다음 단계 결정
        if any(r.get("error") for r in state["execution_results"]):
            return {"next_step": "retry"}
        return {"next_step": "complete"}

# 도구 정의 예시
calculator = Tool(
    name="calculator",
    func=lambda x: eval(x),
    description="수학 계산을 위한 도구"
)

search_tool = Tool(
    name="search",
    func=lambda x: f"검색 결과: {x}",
    description="웹 검색 도구"
)

# 워크플로우 구성
def create_multitool_workflow(agent):
    workflow = StateGraph(MultiToolState)
    workflow.add_node("plan", agent.plan)
    workflow.add_node("execute", agent.execute)
    workflow.add_node("observe", agent.observe)
    workflow.add_node("decide", agent.decide)

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

    workflow.add_conditional_edges(
        "decide",
        lambda x: x["next_step"],
        {"retry": "plan", "complete": END}
    )
    return workflow.compile()

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

사람의 개입이 필요한 작업을 처리하는 데 유용한 템플릿입니다.

python
import asyncio
from datetime import datetime
from enum import Enum

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

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

    def pause_for_review(self, state):
        # 사용자 검토를 위한 일시정지
        review_request = {
            "task": state["current_task"],
            "context": state["context"],
            "timestamp": datetime.now().isoformat()
        }

        # 데이터베이스에 요청 저장 (실제로는 DB 연결)
        self.save_pending_review(review_request)

        return {"status": "waiting_for_approval", "review_id": str(id(review_request))}

    def review(self, state):
        # 인간 검토 처리
        if state["approval_status"] == ApprovalStatus.APPROVED:
            return {"status": "approved", "next_step": "continue"}
        elif state["approval_status"] == ApprovalStatus.REJECTED:
            return {"status": "rejected", "next_step": "retry"}
        return {"status": "pending"}

    def continue_workflow(self, state):
        # 검토 이후 계속 진행
        return {"status": "completed", "result": self.process_final_result(state)}

# 인간 검토 처리 예시
async def human_review_handler(review_id):
    """실제 인간 검토 처리"""
    # 웹훅 또는 메시지 큐를 통한 처리
    await asyncio.sleep(2)  # 시뮬레이션
    return ApprovalStatus.APPROVED

# 워크플로우 구성
def create_human_in_loop_workflow(agent):
    workflow = StateGraph(HumanInLoopState)
    workflow.add_node("process", agent.process)
    workflow.add_node("pause_for_review", agent.pause_for_review)
    workflow.add_node("review", agent.review)
    workflow.add_node("continue_workflow", agent.continue_workflow)

    workflow.set_entry_point("process")
    workflow.add_edge("process", "pause_for_review")
    workflow.add_edge

---

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