LangGraph 워크플로우 템플릿 (v39)
LangGraph 아키텍처 개요
LangGraph는 상태 기반 워크플로우를 구현하기 위한 프레임워크로, 다음과 같은 핵심 구성 요소로 작동합니다:
Nodes (노드): 워크플로우의 각 단계. 각 노드는 특정 작업을 수행하고 상태를 업데이트합니다.
Edges (엣지): 노드 간의 전이 조건을 정의합니다.
State (상태): 워크플로우의 현재 상태를 유지합니다. 노드가 상태를 읽고 변경할 수 있습니다.
Checkpointing (체크포인팅): 상태를 저장하고 복원할 수 있어 중단 후 복구가 가능합니다.
from langgraph.graph import StateGraph
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
# 기본 워크플로우 구성
workflow = StateGraph(AgentState)
템플릿 1: 단순 RAG 에이전트
문서 검색 후 생성하고 검증하는 단순한 RAG 워크플로우입니다:
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
class RAGState(TypedDict):
query: str
documents: list
response: str
validated: bool
def retrieve(state: RAGState):
# 문서 검색 로직
documents = vector_store.similarity_search(state["query"])
return {"documents": documents}
def generate(state: RAGState):
# 생성 로직
prompt = PromptTemplate.from_template(
"Context: {context}\n\nQuestion: {query}\n\nAnswer:"
)
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke([
("system", "You are a helpful assistant."),
("user", prompt.format(
context="\n".join([doc.page_content for doc in state["documents"]]),
query=state["query"]
))
])
return {"response": response.content}
def validate(state: RAGState):
# 응답 검증
# 간단한 예: 응답이 질문과 관련된 내용인지 확인
return {"validated": True}
# 워크플로우 생성
rag_workflow = StateGraph(RAGState)
rag_workflow.add_node("retrieve", retrieve)
rag_workflow.add_node("generate", generate)
rag_workflow.add_node("validate", validate)
rag_workflow.add_edge("retrieve", "generate")
rag_workflow.add_edge("generate", "validate")
rag_workflow.set_entry_point("retrieve")
rag_workflow.set_finish_point("validate")
rag_app = rag_workflow.compile()
템플릿 2: 다중 도구 에이전트
계획 → 실행 → 관찰 → 결정의 반복적인 루프:
from typing import List
import json
class ToolAgentState(TypedDict):
task: str
plan: List[str]
execution_history: List[dict]
final_answer: str
def plan(state: ToolAgentState):
# 작업 계획 생성
llm = ChatOpenAI(model="gpt-4")
plan_prompt = PromptTemplate.from_template(
"Break down the task '{task}' into specific steps. Format as JSON array of strings."
)
response = llm.invoke([
("system", "You are a task planning assistant."),
("user", plan_prompt.format(task=state["task"]))
])
try:
steps = json.loads(response.content)
return {"plan": steps}
except:
return {"plan": [state["task"]]}
def execute(state: ToolAgentState):
# 도구 실행
execution_results = []
for step in state["plan"]:
# 예시: 각 단계에 대한 도구 실행 로직
result = {"step": step, "status": "completed", "output": f"Result for {step}"}
execution_results.append(result)
return {"execution_history": execution_results}
def observe(state: ToolAgentState):
# 실행 결과 관찰
# 예: 성공/실패 여부 확인
failed_steps = [
step for step in state["execution_history"]
if step.get("status") == "failed"
]
return {"execution_history": state["execution_history"]}
def decide(state: ToolAgentState):
# 결정 로직
if not state["execution_history"]:
return {"final_answer": "No execution history available"}
all_success = all(step.get("status") == "completed"
for step in state["execution_history"])
if all_success:
return {"final_answer": "All tasks completed successfully"}
else:
return {"final_answer": "Some tasks failed"}
# 워크플로우 구성
tool_workflow = StateGraph(ToolAgentState)
tool_workflow.add_node("plan", plan)
tool_workflow.add_node("execute", execute)
tool_workflow.add_node("observe", observe)
tool_workflow.add_node("decide", decide)
tool_workflow.add_edge("plan", "execute")
tool_workflow.add_edge("execute", "observe")
tool_workflow.add_edge("observe", "decide")
tool_workflow.set_entry_point("plan")
tool_workflow.set_finish_point("decide")
tool_app = tool_workflow.compile()
템플릿 3: 인간-중개 워크플로우
사용자 검토 후 진행하는 인간 인터페이스:
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState
import asyncio
class HumanReviewState(MessagesState):
review_status: str
review_comment: str
def pause_for_review(state: HumanReviewState):
# 사용자 검토를 위한 일시정지
return {"review_status": "pending"}
def review_process(state: HumanReviewState):
# 검토 프로세스
return {"review_status": "approved"} # 또는 "rejected"
def continue_with_review(state: HumanReviewState):
# 검토 후 진행
if state["review_status"] == "rejected":
return {"messages": [{"role": "assistant", "content": "Process rejected by human review"}]}
return {"messages": [{"role": "assistant", "content": "Process continues after review"}]}
# 인간 검토 워크플로우
review_workflow = StateGraph(HumanReviewState)
review_workflow.add_node("pause", pause_for_review)
review_workflow.add_node("review", review_process)
review_workflow.add_node("continue", continue_with_review)
review_workflow.add_edge("pause", "review")
review_workflow.add_edge("review", "continue")
review_workflow.set_entry_point("pause")
review_workflow.set_finish_point("continue")
review_app = review_workflow.compile(checkpointer=MemorySaver())
템플릿 5: 병렬 실행 에이전트
파이프라인에서 병렬 처리 후 집계:
from concurrent.futures import ThreadPoolExecutor
import time
class ParallelAgentState(TypedDict):
data: List[str]
processed_results: List[dict]
aggregated_result: dict
def fan_out(state: ParallelAgentState):
# 데이터를 병렬로 분할
return {"processed_results": []}
def process_item(item: dict):
# 각 항목 처리
time.sleep(0.1) # 시뮬레이션
return {
"id": item["id"],
"processed": True,
"result": f"Processed {item['data']}"
}
def aggregate_results(state: ParallelAgentState):
# 결과 집계
results = state["processed_results"]
total = len(results)
successful = sum(1 for r in results if r.get("processed"))
return {
"aggregated_result": {
"total_items": total,
"successful": successful,
"failed": total - successful
}
}
# 병렬 처리 워크플로우
parallel_workflow = StateGraph(ParallelAgentState)
parallel_workflow.add_node("fan_out", fan_out)
parallel_workflow.add_node("aggregate", aggregate_results)
parallel_workflow.add_edge("fan_out", "aggregate")
parallel_workflow.set_entry_point("fan_out")
parallel_workflow.set_finish_point("aggregate")
parallel_app = parallel_workflow.compile()
상태 관리 패턴
1. 상태 초기화
def initialize_state():
return {
"messages": [],
"timestamp": time.time(),
"version": "v39",
"metadata": {}
}
# 상태 초기화 및 체크포인트
initial_state = initialize_state()
2. 상태 병합
def merge_state(current_state, new_state):
merged = current_state.copy()
for key, value in new_state.items():
if key in merged and isinstance(merged[key], list):
merged[key].extend(value)
else:
merged[key] = value
return merged
3. 상태 유효성 검사
python
def validate_state(state):
required_fields = ["messages", "timestamp"]
for field in required_fields:
if field not in state:
raise ValueError(f"Missing required field: {field}")
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)