LangGraph 워크플로우 템플릿 (v50)
실용적인 LangChain/LangGraph 워크플로우 템플릿 가이드
Python 개발자들을 위한 AI 에이전트 구축 솔루션
1. LangGraph 아키텍처 개요
LangGraph는 LangChain과 함께 사용되는 상태 기반 워크플로우 프레임워크입니다. 다음과 같은 핵심 구성 요소로 작동합니다:
- 노드 (Nodes): 워크플로우의 각 단계
- 엣지 (Edges): 노드 간의 전이 조건
- 상태 (State): 모든 노드 간 공유되는 데이터 구조
- 체크포인트 (Checkpointing): 실패 시 복구를 위한 상태 저장
from typing import Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
# 상태 정의
class GraphState(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
user_input: str
context: str
# 워크플로우 생성
workflow = StateGraph(GraphState)
2. 템플릿 1: 단순 RAG 에이전트 (검색 → 생성 → 검증)
문제: RAG 시스템에서 검색, 생성, 검증 단계를 잘 통합하는 것이 어렵습니다.
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnablePassthrough
# 검색 노드
def retrieve_node(state):
question = state["user_input"]
vectorstore = Chroma(persist_directory="chroma_db")
retriever = vectorstore.as_retriever()
docs = retriever.invoke(question)
return {"context": "\n".join([doc.page_content for doc in docs])}
# 생성 노드
def generate_node(state):
prompt = PromptTemplate.from_template(
"Context: {context}\n\nQuestion: {user_input}\n\nAnswer:"
)
llm = ChatOpenAI(model="gpt-4")
chain = prompt | llm
response = chain.invoke(state)
return {"messages": [response]}
# 검증 노드
def validate_node(state):
# 간단한 검증 로직 (실제 구현은 더 복잡)
if len(state["messages"][0].content) < 10:
return {"messages": [state["messages"][0]]} # 재시도
return {"messages": [state["messages"][0]]}
# 워크플로우 구성
workflow = StateGraph(GraphState)
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")
workflow.add_edge("validate", END)
# 시작 노드
workflow.set_entry_point("retrieve")
3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)
문제: 여러 도구를 순차적으로 실행하고 상태를 관리하는 것이 복잡합니다.
from typing import List
from langchain_core.tools import Tool
from langchain_core.pydantic_v1 import BaseModel, Field
# 도구 정의
class SearchInput(BaseModel):
query: str = Field(description="검색어")
class CalculatorInput(BaseModel):
expression: str = Field(description="수학 표현식")
# 검색 도구
@tool("search")
def search_tool(query: str) -> str:
return f"검색 결과: {query}"
# 계산 도구
@tool("calculator")
def calculator_tool(expression: str) -> str:
try:
result = eval(expression)
return f"계산 결과: {result}"
except:
return "계산 오류"
# 계획 노드
def plan_node(state):
# 사용자 요청을 기반으로 작업 계획 생성
plan = [
{"action": "search", "input": state["user_input"]},
{"action": "calculator", "input": "2+2"}
]
return {"plan": plan, "current_action": 0}
# 실행 노드
def execute_node(state):
plan = state["plan"]
current = state["current_action"]
if current >= len(plan):
return {"messages": [f"모든 작업 완료"]}
action = plan[current]
tool_map = {"search": search_tool, "calculator": calculator_tool}
# 도구 실행
tool = tool_map[action["action"]]
result = tool.invoke(action["input"])
return {
"messages": [result],
"current_action": current + 1
}
# 관찰 노드
def observe_node(state):
# 실행 결과를 분석하여 다음 단계 결정
return {"messages": [f"관찰 결과: {state['messages'][-1]}"]}
# 결정 노드
def decide_node(state):
if state["current_action"] >= len(state["plan"]):
return "end"
return "execute"
# 워크플로우 구성
workflow = StateGraph(GraphState)
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):
if state["current_action"] >= len(state["plan"]):
return "end"
return "execute"
workflow.add_conditional_edges("decide", route_decision)
workflow.set_entry_point("plan")
4. 템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)
문제: 인간의 판단이 필요한 AI 작업에서 인간과의 협업 흐름을 구현해야 합니다.
from typing import Literal
from langchain_core.messages import HumanMessage
# 일시정지 노드
def pause_node(state):
# 사용자 검토를 위한 메시지 생성
review_message = f"작업 완료: {state['messages'][-1].content}\n\n계속하시겠습니까? (y/n)"
return {"messages": [HumanMessage(content=review_message)]}
# 검토 노드
def review_node(state):
# 사용자 응답 처리
user_response = state.get("user_response", "").lower()
if user_response == "n":
return {"messages": [HumanMessage(content="작업 취소")]}
return {"messages": [HumanMessage(content="작업 재개")]}
# 계속 노드
def continue_node(state):
return {"messages": [HumanMessage(content="계속 진행")]}
# 워크플로우 구성
workflow = StateGraph(GraphState)
workflow.add_node("start", lambda s: s)
workflow.add_node("work", lambda s: {"messages": [f"작업 중: {s['user_input']}"]})
workflow.add_node("pause", pause_node)
workflow.add_node("review", review_node)
workflow.add_node("continue", continue_node)
# 엣지 정의
workflow.add_edge("start", "work")
workflow.add_edge("work", "pause")
workflow.add_edge("pause", "review")
def route_review(state):
user_response = state.get("user_response", "").lower()
return "continue" if user_response == "y" else "end"
workflow.add_conditional_edges("review", route_review)
workflow.add_edge("continue", END)
workflow.set_entry_point("start")
5. 템플릿 5: 병렬 실행 에이전트 (분산 → 처리 → 집계)
문제: 여러 작업을 동시에 처리하고 결과를 통합하는 병렬 워크플로우가 필요합니다.
python
from concurrent.futures import ThreadPoolExecutor
import asyncio
# 분산 노드
def fan_out_node(state):
# 여러 작업을 생성
tasks = [
{"task": "process_data", "input": "데이터1"},
{"task": "process_data", "input": "데이터2"},
{"task": "process_data", "input": "데이터3"}
]
return {"tasks": tasks, "results": []}
# 처리 노드
def process_node(state):
task = state["current_task"]
# 실제 처리 로직
result = f"처리 결과: {task['input']}"
return {
"results": state["results"] + [result],
"current_task": None
}
# 집계 노드
def aggregate_node(state):
# 모든 결과를 집계
aggregated = " | ".join(state["results"])
return {"messages": [f"집계 결과: {aggregated}"]}
# 병렬 처리 워크플로우
workflow
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)