LangGraph 워크플로우 템플릿 (v14)
1. LangGraph 아키텍처 개요
LangGraph는 LangChain과 함께 사용되는 그래프 기반 워크플로우 엔진입니다. 주요 구성 요소는 다음과 같습니다:
Nodes (노드): 작업 단위로, 각 노드는 특정 기능을 수행합니다. 예: retrieve, generate, validate
Edges (엣지): 노드 간의 연결선으로, 실행 흐름을 정의합니다. 예: continue 또는 end
State (상태): 워크플로우 상태를 관리하는 데이터 구조입니다. 노드 간 공유됩니다.
Checkpointing (체크포인팅): 상태를 지속적으로 저장하여 중단된 위치에서 재시작할 수 있도록 합니다.
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class GraphState(TypedDict):
question: str
answer: str
documents: list
# 워크플로우 생성
workflow = StateGraph(GraphState)
2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)
이 템플릿은 문서 검색 및 생성의 기본 흐름을 제공합니다.
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
class SimpleRAGAgent:
def __init__(self, vectorstore, llm):
self.vectorstore = vectorstore
self.llm = llm
self.retriever = vectorstore.as_retriever()
def retrieve(self, state):
documents = self.retriever.invoke(state["question"])
return {"documents": documents}
def generate(self, state):
# 문서와 질문을 기반으로 답변 생성
context = "\n".join([doc.page_content for doc in state["documents"]])
prompt = PromptTemplate.from_template("""
질문: {question}
문서: {context}
위 문서를 기반으로 질문에 답하세요.
""")
chain = prompt | self.llm | StrOutputParser()
answer = chain.invoke({"question": state["question"], "context": context})
return {"answer": answer}
def validate(self, state):
# 답변 검증 (간단한 키워드 검사)
if "error" in state["answer"].lower():
return {"answer": "답변 생성에 실패했습니다. 다시 시도해주세요."}
return {"answer": state["answer"]}
# 사용 예시
agent = SimpleRAGAgent(vectorstore, llm)
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", agent.retrieve)
workflow.add_node("generate", agent.generate)
workflow.add_node("validate", agent.validate)
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
workflow.add_edge("validate", END)
graph = workflow.compile()
3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)
이 템플릿은 여러 도구를 사용하여 문제를 해결하는 에이전트를 구현합니다.
from langchain.tools import Tool
from langchain_core.tools import tool
import json
class MultiToolAgent:
def __init__(self):
self.tools = [
Tool(name="weather", func=self.get_weather, description="날씨 정보 조회"),
Tool(name="calculator", func=self.calculate, description="수학 계산")
]
@tool
def get_weather(self, location: str):
"""날씨 정보 조회"""
# 실제 API 호출 로직
return f"{location}의 날씨는 맑음입니다"
@tool
def calculate(self, expression: str):
"""수학 계산"""
try:
result = eval(expression)
return f"결과: {result}"
except:
return "계산 오류"
def plan(self, state):
# 문제 해결을 위한 계획 수립
planning_prompt = f"""
주어진 질문: {state["question"]}
사용 가능한 도구: {self.tools}
계획을 세우세요.
"""
# 간단한 규칙 기반 계획
if "날씨" in state["question"]:
plan = ["weather"]
elif "계산" in state["question"]:
plan = ["calculator"]
else:
plan = ["calculator"] # 기본 계획
return {"plan": plan}
def execute(self, state):
# 계획에 따라 도구 실행
results = []
for tool_name in state["plan"]:
for tool in self.tools:
if tool.name == tool_name:
result = tool.invoke(state["question"])
results.append(result)
return {"execution_results": results}
def observe(self, state):
# 실행 결과 관찰 및 분석
return {"observation": f"실행 완료: {len(state['execution_results'])}개의 작업"}
def decide(self, state):
# 결정: 다음 단계 결정
if len(state["execution_results"]) > 0:
return {"next": "complete"}
else:
return {"next": "retry"}
# 사용 예시
agent = MultiToolAgent()
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.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
# 조건부 엣지 추가
def should_continue(state):
if state["next"] == "complete":
return "end"
else:
return "plan"
workflow.add_conditional_edges("decide", should_continue)
workflow.add_edge("end", END)
graph = workflow.compile()
4. 템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)
이 템플릿은 인간의 개입이 필요한 경우에 사용됩니다.
from typing import Optional
import time
class HumanInLoopAgent:
def __init__(self, llm):
self.llm = llm
def pause_for_human_review(self, state):
"""사용자 검토를 위한 일시정지"""
# 실제 사용자 입력 대기 (시뮬레이션)
print(f"사용자 검토 필요: {state['question']}")
print("사용자 입력을 기다리는 중...")
time.sleep(1) # 실제 구현에서는 blocking I/O 필요
return {"review_status": "pending"}
def review(self, state):
"""사용자 검토"""
# 실제 검토 로직 (예: API 호출)
print("사용자 검토 완료")
return {"review_status": "approved"}
def continue_workflow(self, state):
"""워크플로우 계속 진행"""
# 검토 후 진행 로직
return {"status": "continue"}
# 사용 예시
agent = HumanInLoopAgent(llm)
workflow = StateGraph(GraphState)
workflow.add_node("generate", agent.generate)
workflow.add_node("pause", agent.pause_for_human_review)
workflow.add_node("review", agent.review)
workflow.add_node("continue", agent.continue_workflow)
workflow.add_edge("generate", "pause")
workflow.add_edge("pause", "review")
workflow.add_edge("review", "continue")
workflow.add_edge("continue", END)
graph = workflow.compile()
5. 템플릿 5: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)
이 템플릿은 여러 작업을 병렬로 처리하고 결과를 집계합니다.
python
from concurrent.futures import ThreadPoolExecutor
import asyncio
class ParallelExecutionAgent:
def __init__(self, llm, num_workers=3):
self.llm = llm
self.num_workers = num_workers
def fan_out(self, state):
"""팬아웃: 작업 분할"""
# 문제 분할
tasks = [
{"id": 1, "question": f"질문 1: {state['question']}"},
{"id": 2, "question": f"질문 2: {state['question']}"},
{"id": 3, "question": f"질문 3: {state['question']}"}
]
return {"tasks": tasks, "task_count": len(tasks)}
def process(self, task):
"""개별 작업 처리"""
# 각 작업 병렬 처리
prompt = PromptTemplate.from_template("질문: {question}")
chain = prompt | self.llm | StrOutputParser()
result = chain.invoke({"question": task["question"]})
return {"task_id": task["id"], "result": result}
async def parallel_process
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)