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

LangGraph는 현대 AI 에이전트 개발을 위한 강력한 프레임워크로, 상태 기반 워크플로우를 구현하는 데 최적화되어 있습니다. 이 가이드에서는 실제 개발자들이 자주 직면하는 문제를 해결하는 5가지 핵심 템플릿을 제공합니다.

1. LangGraph 아키텍처 개요

LangGraph는 다음과 같은 핵심 구성 요소로 구성됩니다:

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]
    # 기타 상태 필드

# 노드 정의
def retrieve_node(state):
    # 문서 검색 로직
    pass

def generate_node(state):
    # 생성 로직
    pass

# 엣지 정의
def route_to_next(state):
    # 다음 노드 결정 로직
    pass

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

# 체크포인팅 설정
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

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

이 템플릿은 문서 검색과 생성을 결합한 기본적인 RAG 워크플로우입니다.

from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
import operator
from typing import Annotated, List
from langgraph.graph import StateGraph, END

class RAGState(TypedDict):
    query: str
    retrieved_docs: List[str]
    generated_response: str
    validation_result: bool

# 검색 노드
def retrieve_node(state: RAGState):
    # Chroma에서 문서 검색
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=OpenAIEmbeddings()
    )

    docs = vectorstore.similarity_search(state["query"], k=3)
    return {"retrieved_docs": [doc.page_content for doc in docs]}

# 생성 노드
def generate_node(state: RAGState):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant that answers questions based on provided context."),
        ("human", """Context: {context}
Question: {question}
Answer:""")
    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({
        "context": "\n".join(state["retrieved_docs"]),
        "question": state["query"]
    })

    return {"generated_response": response.content}

# 검증 노드
def validate_node(state: RAGState):
    # 생성된 응답 검증
    if len(state["generated_response"]) < 10:
        return {"validation_result": False}

    # 간단한 키워드 기반 검증
    keywords = ["answer", "question", "context", "response"]
    valid = any(keyword in state["generated_response"].lower() for keyword in keywords)
    return {"validation_result": valid}

# 워크플로우 정의
def create_rag_workflow():
    workflow = StateGraph(RAGState)

    workflow.add_node("retrieve", retrieve_node)
    workflow.add_node("generate", generate_node)
    workflow.add_node("validate", validate_node)

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

    # 검증 결과에 따라 처리
    def route_validation(state: RAGState):
        if not state["validation_result"]:
            return "retry"
        return "end"

    workflow.add_conditional_edges(
        "validate",
        route_validation,
        {
            "retry": "retrieve",
            "end": END
        }
    )

    workflow.set_entry_point("retrieve")
    return workflow.compile()

# 사용 예시
# app = create_rag_workflow()
# result = app.invoke({"query": "Python 디자인 패턴에 대해 설명해주세요"})

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

이 템플릿은 여러 도구를 활용할 수 있는 복잡한 에이전트를 위한 워크플로우입니다.

from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_core.messages import ToolMessage

class ToolAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    plan: List[str]
    current_tool: str
    tool_result: str

# 계획 노드
def plan_node(state: ToolAgentState):
    # 도구 사용 계획 생성
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a task planner. Analyze the following request and create a plan of actions."),
        ("human", "Request: {request}\nPlan:")

    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({"request": state["messages"][-1].content})

    return {"plan": response.content.split('\n')}

# 실행 노드
def execute_node(state: ToolAgentState):
    # 도구 실행
    tool_name = state["plan"][0]  # 첫 번째 계획 항목

    # 실제 도구 실행 로직 (예: 파일 시스템 조작)
    if tool_name == "file_search":
        # 파일 검색 로직
        result = "Found files matching criteria"
    elif tool_name == "code_analysis":
        # 코드 분석 로직
        result = "Code analysis complete"
    else:
        result = "Tool not recognized"

    return {"current_tool": tool_name, "tool_result": result}

# 관찰 노드
def observe_node(state: ToolAgentState):
    # 도구 결과 관찰
    messages = state["messages"]
    tool_message = ToolMessage(
        content=state["tool_result"],
        tool_call_id=state["current_tool"]
    )

    return {"messages": [tool_message]}

# 결정 노드
def decide_node(state: ToolAgentState):
    # 다음 단계 결정
    if len(state["plan"]) > 1:
        return "continue"
    return "end"

# 워크플로우 정의
def create_tool_workflow():
    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.add_edge("plan", "execute")
    workflow.add_edge("execute", "observe")
    workflow.add_edge("observe", "decide")

    def route_decision(state: ToolAgentState):
        if state["plan"] and len(state["plan"]) > 1:
            return "plan"
        return "end"

    workflow.add_conditional_edges(
        "decide",
        route_decision,
        {
            "plan": "plan",
            "end": END
        }
    )

    workflow.set_entry_point("plan")
    return workflow.compile()

# 사용 예시
# app = create_tool_workflow()
# result = app.invoke({"messages": [HumanMessage(content="파일 시스템에서 모든 Python 파일을 검색해주세요")]})

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

이 템플릿은 인간의 개입이 필요한 작업을 처리하기 위한 워크플로우입니다.


python
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import AIMessage

class HumanInLoopState(TypedDict):
    messages: Annotated[list, operator.add]
    action_needed: bool
    user_feedback: str
    processed_result: str

# 작업 생성 노드
def create_task_node(state: HumanInLoopState):
    # 작업 생성 및 인간 개입 필요성 판단
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Generate task based on user request and determine if human review is needed."),
        ("human", "Request: {request}\nGenerate task:")
    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({"request": state["messages"][-1].content})

    # 간단한 규칙으로 인간 개입 필요성 판단
    needs_review = "review" in response.content.lower() or "approve" in response.content.lower()

    return {
        "messages": [AIMessage(content=response.content)],
        "action_needed":

---

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

推荐订阅源