Molayo

© 2026 Molayo

Dev.to헤드라인2026. 05. 25. 23:58

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

요약

LangGraph를 활용하여 복잡한 AI 에이전트를 구축하기 위한 워크플로우 템플릿 가이드입니다. Nodes, Edges, State, Checkpointing 등 핵심 구성 요소를 설명하고 RAG 에이전트 구현 예시를 제공합니다.

핵심 포인트

  • LangGraph의 4가지 핵심 구성 요소(Nodes, Edges, State, Checkpointing) 설명
  • 검색, 생성, 검증 단계를 포함하는 RAG 에이전트 워크플로우 템플릿 제공
  • TypedDict를 활용한 상태 관리 및 노드 간 데이터 흐름 구현 방법 안내

python
# LangGraph 워크플로우 템플릿 (v33)
# Python 개발자를 위한 LangChain/LangGraph 기반 AI 에이전트 템플릿 가이드

"""
LangGraph는 LangChain의 고급 워크플로우 시스템으로, 복잡한 AI 에이전트를 구축할 수 있는 강력한 도구입니다.
이 가이드는 실제 개발 환경에서 활용할 수 있는 4가지 핵심 템플릿을 제공합니다.
"""

# 1. LangGraph 아키텍처 개요
"""
LangGraph는 다음과 같은 핵심 구성 요소로 구성됩니다:

- Nodes: 각각의 작업 단위 (함수 또는 클래스)
- Edges: 노드 간의 연결 (조건부 또는 정적 경로)
- State: 노드 간 공유되는 상태 (dict 형태)
- Checkpointing: 상태 저장 및 복원 기능

이러한 구성 요소를 잘 활용하면 복잡한 에이전트 워크플로우를 구축할 수 있습니다.
"""

# 2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.vectorstores import VectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import PromptTemplate

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

# 노드 함수들
def retrieve(state: RAGState):
    # 벡터 스토어에서 관련 문서 검색
    vector_store = VectorStore.from_documents(
        documents=state["question"],
        embedding=OpenAIEmbeddings()
    )
    retrieved_docs = vector_store.similarity_search(state["question"], k=3)
    context = "\n".join([doc.page_content for doc in retrieved_docs])
    return {"context": context}

def generate(state: RAGState):
    # LLM을 사용하여 답변 생성
    prompt = PromptTemplate.from_template("""
    주어진 문맥을 바탕으로 질문에 답하세요:

    문맥: {context}

    질문: {question}

    답변:
    """)

    llm = ChatOpenAI(model="gpt-4-turbo")
    answer = llm.invoke([
        ("system", "정확하고 간결한 답변을 제공하세요."),
        ("user", prompt.format(context=state["context"], question=state["question"]))
    ])

    return {"answer": answer.content}

def validate(state: RAGState):
    # 생성된 답변의 신뢰성 검증
    llm = ChatOpenAI(model="gpt-4-turbo")
    validation_prompt = PromptTemplate.from_template("""
    아래 답변이 주어진 문맥과 일치하는지 검증하세요.

    문맥: {context}
    질문: {question}
    답변: {answer}

    판단: 답변이 정확하고 신뢰할 수 있는지 판단하세요.
    """)

    validation_result = llm.invoke([
        ("user", validation_prompt.format(
            context=state["context"],
            question=state["question"],
            answer=state["answer"]
        ))
    ])

    return {"validation": "정확" in validation_result.content or "신뢰할 수" in validation_result.content}

# 워크플로우 그래프 구성
rag_graph = StateGraph(RAGState)
rag_graph.add_node("retrieve", retrieve)
rag_graph.add_node("generate", generate)
rag_graph.add_node("validate", validate)
rag_graph.set_entry_point("retrieve")
rag_graph.add_edge("retrieve", "generate")
rag_graph.add_edge("generate", "validate")
rag_graph.add_edge("validate", END)

# 실행
rag_workflow = rag_graph.compile()

# 사용 예시:
# result = rag_workflow.invoke({
#     "question": "LangGraph는 무엇인가요?"
# })

# 3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)
class ToolAgentState(TypedDict):
    input: str
    plan: str
    execution: str
    observation: str
    decision: str
    tools: list

# 도구 정의
def plan_action(state: ToolAgentState):
    # 사용자 입력에 따라 실행 계획 수립
    llm = ChatOpenAI(model="gpt-4-turbo")
    plan_prompt = PromptTemplate.from_template("""
    사용자의 요청을 분석하여 실행 계획을 세우세요:

    요청: {input}

    계획:
    """)

    plan = llm.invoke([
        ("user", plan_prompt.format(input=state["input"]))
    ])

    return {"plan": plan.content}

def execute_tool(state: ToolAgentState):
    # 계획된 도구 실행
    llm = ChatOpenAI(model="gpt-4-turbo")
    tool_prompt = PromptTemplate.from_template("""
    다음 도구를 실행하세요:

    계획: {plan}

    실행 결과:
    """)

    execution_result = llm.invoke([
        ("user", tool_prompt.format(plan=state["plan"]))
    ])

    return {"execution": execution_result.content}

def observe_result(state: ToolAgentState):
    # 실행 결과 관찰
    llm = ChatOpenAI(model="gpt-4-turbo")
    observation_prompt = PromptTemplate.from_template("""
    아래 실행 결과를 분석하세요:

    실행: {execution}

    관찰:
    """)

    observation = llm.invoke([
        ("user", observation_prompt.format(execution=state["execution"]))
    ])

    return {"observation": observation.content}

def make_decision(state: ToolAgentState):
    # 관찰 결과를 바탕으로 결정
    llm = ChatOpenAI(model="gpt-4-turbo")
    decision_prompt = PromptTemplate.from_template("""
    다음 정보를 바탕으로 최종 결정을 내리세요:

    실행: {execution}
    관찰: {observation}

    결정:
    """)

    decision = llm.invoke([
        ("user", decision_prompt.format(
            execution=state["execution"],
            observation=state["observation"]
        ))
    ])

    return {"decision": decision.content}

# 워크플로우 구성
tool_agent_graph = StateGraph(ToolAgentState)
tool_agent_graph.add_node("plan", plan_action)
tool_agent_graph.add_node("execute", execute_tool)
tool_agent_graph.add_node("observe", observe_result)
tool_agent_graph.add_node("decide", make_decision)

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

# 실행
tool_agent_workflow = tool_agent_graph.compile()

# 사용 예시:
# result = tool_agent_workflow.invoke({
#     "input": "날씨 정보를 가져와서 요약해줘"
# })

# 4. 템플릿 3: 인간-중개 워크플로우 (중단 → 검토 → 진행)
from typing import Optional
from enum import Enum

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

class HumanInLoopState(TypedDict):
    task: str
    solution: str
    user_feedback: Optional[str]
    approval_status: ApprovalStatus

def generate_solution(state: HumanInLoopState):
    llm = ChatOpenAI(model="gpt-4-turbo")
    prompt = PromptTemplate.from_template("""
    사용자 요청에 대한 해결책을 제시하세요:

    요청: {task}

    해결책:
    """)

    solution = llm.invoke([
        ("user", prompt.format(task=state["task"]))
    ])

    return {"solution": solution.content}

def human_review(state: HumanInLoopState):
    # 인간의 검토를 위한 중단
    # 실제 구현에서는 외부 API 또는 메시지 큐를 통해 human approval을 받음
    # 여기서는 단순 예시로 설정
    return {"approval_status": ApprovalStatus.PENDING}

def process_approval(state: HumanInLoopState):
    # 인간 승인 처리
    if state["approval_status"] == ApprovalStatus.APPROVED:
        return {"solution": f"승인된 해결책: {state['solution']}"}
    elif state["approval_status"] == ApprovalStatus.REJECTED:
        # 재시도 또는 대안 제안
        llm = ChatOpenAI(model="gpt-4-turbo")
        alternative_prompt = PromptTemplate.from_template("""
        아래 해결책이 거부되었습니다. 대안을 제안하세요:

        원본 해결책: {solution}

        대안:
        """)

        alternative = llm.invoke([
            ("user", alternative_prompt.format(solution=state["solution"]))
        ])

        return {"solution": alternative.content}
    else:
        return {"solution": "검토 중"}

# 워크플로우 구성
human_loop_graph = StateGraph(HumanInLoopState)
human

---

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

AI 자동 생성 콘텐츠

본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
0

댓글

0