LangGraph 워크플로우 템플릿 (v21)
개요
LangGraph는 복잡한 AI 에이전트 워크플로우를 구축하기 위한 강력한 프레임워크입니다. 이 가이드는 실제 개발 문제를 해결하는 실용적인 템플릿들을 제공합니다. 각 템플릿은 $3-$7 범위의 간단한 도구로 구성되어 있으며, Python 개발자를 위한 최적화된 솔루션을 제공합니다.
1. LangGraph 아키텍처 개요
LangGraph는 노드(Node), 엣지(Edge), 상태(State), 체크포인트(Checkpoint)의 4가지 핵심 구성 요소로 구성됩니다:
- 노드: 워크플로우의 작업 단위 (예: 검색, 생성, 검증)
- 엣지: 노드 간의 실행 흐름 (예: 성공 → 다음 노드, 실패 → 재시도)
- 상태: 워크플로우 내 모든 데이터를 저장 (예: 사용자 질문, 중간 결과)
- 체크포인트: 실패 시 복구를 위한 상태 저장 (예: 재시작 가능)
from typing import TypedDict
from langgraph.graph import StateGraph, END
class State(TypedDict):
question: str
answer: str
history: list
workflow = StateGraph(State)
2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)
문제: LLM의 생성 결과가 사실과 다를 수 있어 신뢰성 있는 답변이 필요함
해결: 검색 → 생성 → 검증의 3단계 워크플로우
import operator
from typing import Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
class RAGState(TypedDict):
question: str
documents: List[str]
answer: str
validation: bool
# 검색 노드
def retrieve(state):
question = state["question"]
# 실제 검색 로직 (예: FAISS, ElasticSearch)
documents = ["문서 1 내용", "문서 2 내용"]
return {"documents": documents}
# 생성 노드
def generate(state):
docs = state["documents"]
question = state["question"]
# 실제 LLM 생성 로직
answer = f"질문: {question}\n문서: {', '.join(docs)}"
return {"answer": answer}
# 검증 노드
def validate(state):
answer = state["answer"]
question = state["question"]
# 간단한 검증 (실제 구현은 LLM 기반)
is_valid = "문서" in answer and len(answer) > 20
return {"validation": is_valid}
# 워크플로우 구성
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_node("validate", validate)
# 엣지 연결
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
# 조건부 엣지
def check_validation(state):
return "validate" if state["validation"] else "generate"
workflow.add_conditional_edges(
"validate",
check_validation,
{
"generate": "generate",
"validate": END
}
)
app = workflow.compile()
3. 템플릿 2: 멀티-도구 에이전트 (계획 → 실행 → 관찰 → 결정)
문제: 단일 도구만 사용하는 에이전트는 복잡한 작업 수행이 어려움
해결: 계획 단계에서 도구를 선택하고 실행, 관찰 후 결정하는 워크플로우
from langchain.tools import Tool
from langchain_core.messages import ToolMessage
import json
class ToolAgentState(TypedDict):
question: str
plan: List[str]
execution_results: List[str]
final_answer: str
# 도구 정의
def search_tool(query):
return f"검색 결과: {query}"
def calculator_tool(expression):
return f"계산 결과: {eval(expression)}"
tools = [
Tool(name="search", func=search_tool, description="웹 검색"),
Tool(name="calculator", func=calculator_tool, description="수학 계산")
]
# 계획 노드
def plan(state):
question = state["question"]
# LLM을 사용한 계획 생성
plan_steps = ["search", "calculator"] # 실제 구현에서는 LLM 사용
return {"plan": plan_steps}
# 실행 노드
def execute(state):
plan_steps = state["plan"]
results = []
for step in plan_steps:
if step == "search":
result = search_tool("Python 개발")
elif step == "calculator":
result = calculator_tool("2+3*4")
results.append(result)
return {"execution_results": results}
# 관찰 노드
def observe(state):
results = state["execution_results"]
# 결과 분석 (예: 정확성, 관련성 판단)
return {"execution_results": results}
# 결정 노드
def decide(state):
# 최종 답변 생성
return {"final_answer": "최종 답변 생성됨"}
# 워크플로우 구성
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan)
workflow.add_node("execute", execute)
workflow.add_node("observe", observe)
workflow.add_node("decide", decide)
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
workflow.add_edge("decide", END)
app = workflow.compile()
4. 템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)
문제: LLM이 실수를 할 경우 인간의 개입이 필요함
해결: 인간 검토 단계를 통한 신뢰성 확보
from typing import Optional
import asyncio
class HumanInLoopState(TypedDict):
question: str
answer: str
review_status: str # "pending", "approved", "rejected"
human_feedback: Optional[str]
# 생성 노드
def generate_answer(state):
question = state["question"]
# LLM 생성
answer = f"질문에 대한 답변: {question}"
return {"answer": answer}
# 일시정지 노드
def pause_for_review(state):
# 실제 구현에서는 외부 시스템(예: Slack, Discord)을 통한 통신
print(f"답변 생성됨: {state['answer']}")
print("사용자에게 검토 요청 중...")
return {"review_status": "pending"}
# 검토 노드
def review_answer(state):
# 인간 입력 대기
feedback = input("검토 결과 (approve/reject): ")
status = "approved" if feedback.lower() == "approve" else "rejected"
return {
"review_status": status,
"human_feedback": feedback
}
# 재생성 노드
def regenerate_answer(state):
question = state["question"]
# 재생성 로직
answer = f"재생성된 답변: {question}"
return {"answer": answer}
# 워크플로우 구성
workflow = StateGraph(HumanInLoopState)
workflow.add_node("generate", generate_answer)
workflow.add_node("pause", pause_for_review)
workflow.add_node("review", review_answer)
workflow.add_node("regenerate", regenerate_answer)
workflow.add_edge("generate", "pause")
workflow.add_edge("pause", "review")
def check_review(state):
return "regenerate" if state["review_status"] == "rejected" else "end"
workflow.add_conditional_edges(
"review",
check_review,
{
"regenerate": "regenerate",
"end": END
}
)
workflow.add_edge("regenerate", END)
app = workflow.compile()
5. 템플릿 5: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)
문제: 여러 데이터 소스에서 동시에 처리하여 효율성 향상
해결: 병렬 실행 후 결과 집계
python
from concurrent.futures import ThreadPoolExecutor
import asyncio
class ParallelAgentState(TypedDict):
question: str
sources: List[str]
results: List[str]
aggregated_result: str
# 병렬 처리 노드
def parallel_process(state):
sources = state["sources"]
question = state["question"]
def process_source(source):
# 각 소스별 처리 로직
return f"{source} 처리 결과: {question}"
# 병렬 실행
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(process_source, sources))
return {"results": results}
# 집계 노드
def aggregate_results(state
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)