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

LangGraph 아키텍처 개요

LangGraph는 LangChain의 고급 워크플로우 시스템으로, 상태 기반 그래프 기반 AI 에이전트를 구축하는 데 최적화되어 있습니다. 핵심 구성 요소는 다음과 같습니다:

• 노드 (Nodes): 각각의 작업 단계. 입력 상태를 받아 출력 상태를 반환합니다.
• 엣지 (Edges): 노드 간의 흐름을 정의합니다.
• 상태 (State): 에이전트의 현재 상태를 나타내는 Pydantic 모델입니다.
• 체크포인팅 (Checkpointing): 상태를 저장하고 복원할 수 있도록 합니다.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
from langchain_core.messages import BaseMessage

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

# 기본 그래프 구조
graph = StateGraph(AgentState)

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

이 템플릿은 문서 검색 및 생성을 수행하는 기본적인 RAG 에이전트입니다.

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda
import operator

# RAG 상태 정의
class RagState(TypedDict):
    query: str
    context: str
    answer: str
    validation: bool

# 검색 노드
def retrieve(state: RagState):
    # 문서 검색 로직 (예: FAISS, Pinecone)
    vectorstore = ... # Vector store 초기화
    docs = vectorstore.similarity_search(state["query"], k=3)
    context = "\n\n".join([doc.page_content for doc in docs])
    return {"context": context}

# 생성 노드
def generate(state: RagState):
    prompt = PromptTemplate.from_template(
        "질문: {query}\n\n문맥: {context}\n\n답변:"
    )
    model = ChatOpenAI(model="gpt-4")
    answer = model.invoke(prompt.format(query=state["query"], context=state["context"]))
    return {"answer": answer.content}

# 검증 노드
def validate(state: RagState):
    # 생성된 답변의 검증 로직
    prompt = PromptTemplate.from_template(
        "다음 답변이 질문에 대한 적절한 답변인지 평가하세요:\n\n질문: {query}\n\n답변: {answer}\n\n평가: (예/아니요)"
    )
    model = ChatOpenAI(model="gpt-4")
    validation = model.invoke(prompt.format(query=state["query"], answer=state["answer"]))
    return {"validation": validation.content.startswith("예")}

# RAG 그래프 생성
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")

# 조건부 엣지
def route_validation(state: RagState):
    if state["validation"]:
        return "end"
    else:
        return "retrieve"  # 다시 검색

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

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

이 템플릿은 다중 도구를 사용하여 작업을 수행하는 에이전트입니다.

from typing import List
from langchain_core.tools import Tool

class ToolAgentState(TypedDict):
    messages: list[BaseMessage]
    plan: List[str]
    execution_log: List[str]
    final_answer: str

# 도구 정의
def web_search(query: str) -> str:
    # 웹 검색 로직
    return f"검색 결과: {query}"

def calculator(expression: str) -> str:
    # 계산기 로직
    return f"계산 결과: {eval(expression)}"

tools = [
    Tool(name="web_search", func=web_search, description="웹에서 정보 검색"),
    Tool(name="calculator", func=calculator, description="수학 계산")
]

# 계획 노드
def plan(state: ToolAgentState):
    # 작업 계획 생성
    prompt = PromptTemplate.from_template(
        "다음 작업을 수행하기 위한 계획을 세우세요:\n\n{messages}"
    )
    model = ChatOpenAI(model="gpt-4")
    plan = model.invoke(prompt.format(messages=state["messages"]))
    return {"plan": [task.strip() for task in plan.content.split('\n') if task.strip()]}

# 실행 노드
def execute(state: ToolAgentState):
    # 계획된 작업 실행
    execution_log = []
    for task in state["plan"]:
        if "search" in task.lower():
            result = web_search(task)
        elif "calculate" in task.lower():
            result = calculator(task)
        execution_log.append(f"{task}: {result}")
    return {"execution_log": execution_log}

# 관찰 노드
def observe(state: ToolAgentState):
    # 실행 결과 분석
    prompt = PromptTemplate.from_template(
        "다음 실행 결과를 분석하세요:\n\n{execution_log}"
    )
    model = ChatOpenAI(model="gpt-4")
    analysis = model.invoke(prompt.format(execution_log=state["execution_log"]))
    return {"final_answer": analysis.content}

# 멀티-도구 에이전트 그래프
tool_agent = StateGraph(ToolAgentState)
tool_agent.add_node("plan", plan)
tool_agent.add_node("execute", execute)
tool_agent.add_node("observe", observe)

tool_agent.set_entry_point("plan")
tool_agent.add_edge("plan", "execute")
tool_agent.add_edge("execute", "observe")

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

이 템플릿은 인간의 입력을 포함한 결정 프로세스를 구현합니다.

from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage, HumanMessage
from typing import Optional

class HumanInLoopState(TypedDict):
    messages: list[BaseMessage]
    user_review: Optional[str]
    final_decision: Optional[str]

# 일시정지 노드
def pause_for_review(state: HumanInLoopState):
    # 사용자 검토 요청
    human_message = HumanMessage(content="다음 작업을 검토해주세요:")
    return {"messages": [human_message]}

# 검토 노드
def review(state: HumanInLoopState):
    # 사용자 검토 처리
    return {"user_review": state["messages"][-1].content}

# 결정 노드
def decide(state: HumanInLoopState):
    # 사용자 피드백 기반 결정
    prompt = PromptTemplate.from_template(
        "사용자 피드백: {user_review}\n\n결정하세요:"
    )
    model = ChatOpenAI(model="gpt-4")
    decision = model.invoke(prompt.format(user_review=state["user_review"]))
    return {"final_decision": decision.content}

# 인간-중개 워크플로우
human_loop = StateGraph(HumanInLoopState)
human_loop.add_node("pause", pause_for_review)
human_loop.add_node("review", review)
human_loop.add_node("decide", decide)

human_loop.set_entry_point("pause")
human_loop.add_edge("pause", "review")
human_loop.add_edge("review", "decide")

템플릿 5: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)

이 템플릿은 여러 작업을 병렬로 처리하고 결과를 집계합니다.

python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(TypedDict):
    tasks: list[str]
    results: list[dict]
    aggregated_result: str

# 병렬 처리 노드
def fan_out(state: ParallelAgentState):
    # 작업 분할
    tasks = state["tasks"]
    return {"tasks": tasks}

# 처리 노드 (병렬 실행)
def process_task(state: ParallelAgentState):
    # 각 작업을 병렬로 처리
    def process_single_task(task):
        # 작업 처리 로직
        return {"task": task, "result": f"처리 완료: {task}"}

    # 병렬 실행
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(process_single_task, task) for task in state["tasks"]]
        results = [future.result() for future in futures]

    return {"results": results}

# 집계 노드
def aggregate_results(state: ParallelAgentState):
    # 결과 집계
    aggregated = "\n".join([f"{r['task']}: {r['result']}" for r in state

---

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