LangGraph 워크플로우 템플릿 (v26)
개요
LangGraph는 LangChain과 함께 사용되는 강력한 워크플로우 엔진으로, 복잡한 AI 에이전트를 구축하는 데 필수적입니다. 이번 가이드에서는 실전에서 사용 가능한 LangGraph 템플릿 4개를 제공하며, 각 템플릿은 실제 개발자들이 겪는 문제를 해결합니다.
1. LangGraph 아키텍처 개요
핵심 구성 요소:
- Nodes: 워크플로우의 각 단계 (함수 또는 클래스)
- Edges: 노드 간의 연결 (조건부 또는 정적)
- State: 워크플로우 상태 (사용자 입력, 중간 결과 등)
- Checkpointing: 상태 저장 및 복구 기능
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 상태 정의
class GraphState(TypedDict):
question: str
context: str
answer: str
steps: list
# 그래프 생성
workflow = StateGraph(GraphState)
2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)
이 템플릿은 전통적인 RAG (Retrieval-Augmented Generation) 패턴을 구현하여 문서 기반 답변 생성을 수행합니다.
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
class SimpleRAGAgent:
def __init__(self, vectorstore, llm):
self.vectorstore = vectorstore
self.llm = llm
self.prompt = PromptTemplate.from_template("""
질문: {question}
문서 내용: {context}
위 문서를 기반으로 질문에 답변해주세요.
답변:
""")
def retrieve(self, state):
"""문서 검색"""
question = state["question"]
docs = self.vectorstore.similarity_search(question, k=3)
context = "\n".join([doc.page_content for doc in docs])
return {"context": context}
def generate(self, state):
"""답변 생성"""
prompt = self.prompt.format(
question=state["question"],
context=state["context"]
)
response = self.llm.invoke(prompt)
return {"answer": response.content}
def validate(self, state):
"""답변 검증"""
# 간단한 검증 로직
if len(state["answer"]) < 10:
return {"answer": "답변이 너무 짧습니다. 다시 시도해주세요."}
return {}
# 워크플로우 정의
def create_rag_workflow(agent):
workflow = StateGraph(GraphState)
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_edge("validate", END)
return workflow.compile()
# 사용 예시
# rag_agent = SimpleRAGAgent(vectorstore, llm)
# rag_workflow = create_rag_workflow(rag_agent)
# result = rag_workflow.invoke({"question": "최근 AI 트렌드는?"})
3. 템플릿 2: 멀티-도구 에이전트 (계획 → 실행 → 관찰 → 결정)
이 템플릿은 다양한 도구를 활용한 계획-실행-결정 워크플로우를 구현합니다.
from typing import Union, Dict, Any
from langchain_core.tools import Tool
import json
class MultiToolAgent:
def __init__(self):
self.tools = {
"search": Tool(
name="search",
func=lambda q: f"검색 결과: {q}",
description="웹 검색을 위한 도구"
),
"calculator": Tool(
name="calculator",
func=lambda expr: eval(expr),
description="수학 계산 도구"
)
}
def plan(self, state):
"""계획 생성"""
# 간단한 규칙 기반 계획
question = state["question"]
if "계산" in question:
return {"plan": ["calculator"]}
else:
return {"plan": ["search"]}
def execute(self, state):
"""도구 실행"""
plan = state["plan"]
results = []
for tool_name in plan:
tool = self.tools[tool_name]
result = tool.invoke(state["question"])
results.append(f"{tool_name}: {result}")
return {"execution_results": results}
def observe(self, state):
"""실행 결과 관찰"""
return {"observation": "실행 완료"}
def decide(self, state):
"""결정"""
results = state["execution_results"]
return {"answer": f"결과: {' '.join(results)}"}
# 워크플로우 정의
def create_multi_tool_workflow(agent):
workflow = StateGraph(GraphState)
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_edge("decide", END)
return workflow.compile()
4. 템플릿 3: 인간-중개 워크플로우 (중지 → 검토 → 계속)
이 템플릿은 인간의 개입이 필요한 상황에서 사용되며, 사용자의 승인을 기다립니다.
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage
class HumanInLoopAgent:
def __init__(self, llm):
self.llm = llm
def generate(self, state):
"""초안 생성"""
# 답변 생성 로직
prompt = f"질문: {state['question']}\n답변: "
response = self.llm.invoke(prompt)
return {"draft_answer": response.content}
def review(self, state):
"""검토 요청"""
# 중단 포인트 - 사용자 승인 대기
return {"status": "pending_review"}
def approve(self, state):
"""승인 처리"""
# 사용자 승인 후 진행
return {"approved": True}
def finalize(self, state):
"""최종 답변 생성"""
return {"answer": state["draft_answer"]}
# 인간-중개 워크플로우
def create_human_loop_workflow(agent):
workflow = StateGraph(GraphState)
workflow.add_node("generate", agent.generate)
workflow.add_node("review", agent.review)
workflow.add_node("approve", agent.approve)
workflow.add_node("finalize", agent.finalize)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "review")
# 조건부 엣지
def should_approve(state):
if state.get("status") == "pending_review":
return "approve"
return "finalize"
workflow.add_conditional_edges(
"review",
should_approve,
{"approve": "approve", "finalize": "finalize"}
)
workflow.add_edge("approve", "finalize")
workflow.add_edge("finalize", END)
return workflow.compile()
5. 템플릿 5: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)
이 템플릿은 여러 작업을 병렬로 실행하여 성능 향상을 실현합니다.
python
from concurrent.futures import ThreadPoolExecutor
import asyncio
class ParallelAgent:
def __init__(self, llm):
self.llm = llm
def fan_out(self, state):
"""팬아웃: 작업 분할"""
# 질문을 여러 하위 질문으로 분할
question = state["question"]
sub_questions = [
f"{question}의 역사",
f"{question}의 기술적 특징",
f"{question}의 미래 전망"
]
return {"sub_questions": sub_questions}
def process_parallel(self, state):
"""병렬 처리"""
sub_questions = state["sub_questions"]
# 병렬 실행
results = []
with ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(self._generate_response, q)
for q in sub_questions
]
for future in futures:
results.append(future.result())
return {"sub_answers": results}
def _generate_response(self, question):
"""개별 답변 생성"""
prompt = f"질문: {question}\n답변: "
response = self.llm.invoke(prompt)
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)