LangGraph 워크플로우 템플릿 (v48)
개요: LangGraph 아키텍처와 실전 패턴
LangGraph는 LangChain의 고급 워크플로우 엔진으로, 복잡한 AI 에이전트를 구현하는 데 강력한 도구를 제공합니다. 이 템플릿 가이드는 실제 개발자가 자주 겪는 문제를 해결하기 위한 5가지 핵심 패턴을 제공합니다.
기본 구성 요소
- Nodes: 워크플로우의 각 단계
- Edges: 노드 간의 실행 흐름
- State: 모든 노드가 공유하는 상태 객체
- Checkpointing: 실행 중단/재개를 위한 상태 저장 기능
핵심 설계 원칙
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
# 추가 상태 필드
템플릿 1: 간단한 RAG 에이전트
문제: 문서 검색 및 생성을 통합한 RAG 워크플로우에서 검증 과정이 필요
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
# RAG 에이전트 구현
class RAGAgent:
def __init__(self, vector_store_path: str):
self.vector_store = Chroma(persist_directory=vector_store_path)
self.llm = ChatOpenAI(model="gpt-4o-mini")
self.retriever = self.vector_store.as_retriever()
def retrieve(self, state: dict):
"""문서 검색"""
query = state["messages"][-1].content
docs = self.retriever.invoke(query)
return {"retrieved_docs": docs}
def generate(self, state: dict):
"""문서 기반 생성"""
context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
prompt = ChatPromptTemplate.from_messages([
("system", "다음 컨텍스트를 기반으로 답변을 생성하세요: {context}"),
("user", "{query}")
])
chain = prompt | self.llm | StrOutputParser()
response = chain.invoke({
"context": context,
"query": state["messages"][-1].content
})
return {"generated_response": response}
def validate(self, state: dict):
"""생성된 응답 검증"""
# 간단한 검증 규칙
response = state["generated_response"]
if len(response) < 10:
raise ValueError("응답이 너무 짧습니다")
return {"valid": True}
# 워크플로우 정의
def build_rag_workflow():
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", RAGAgent.retrieve)
workflow.add_node("generate", RAGAgent.generate)
workflow.add_node("validate", RAGAgent.validate)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
workflow.add_edge("validate", END)
return workflow.compile()
템플릿 2: 다중 도구 사용 에이전트
문제: 여러 도구를 활용한 계획-실행-관찰-결정 워크플로우
from langchain.tools import Tool
from langchain_openai import OpenAI
import json
class MultiToolAgent:
def __init__(self):
self.tools = [
Tool(
name="search",
func=lambda q: f"검색 결과: {q}",
description="웹 검색을 수행합니다"
),
Tool(
name="calculator",
func=lambda e: f"계산 결과: {eval(e)}",
description="수학 계산을 수행합니다"
)
]
self.llm = ChatOpenAI(model="gpt-4o-mini")
def plan(self, state: dict):
"""작업 계획 생성"""
prompt = ChatPromptTemplate.from_messages([
("system", "다음 작업을 위한 계획을 생성하세요: {task}"),
("user", "도구 목록: {tools}")
])
plan = self.llm.invoke(prompt.format(
task=state["messages"][-1].content,
tools=[tool.name for tool in self.tools]
))
return {"plan": plan.content}
def execute(self, state: dict):
"""계획 실행"""
plan = state["plan"]
# 간단한 실행 로직 (실제 구현에서는 도구 선택 로직 포함)
return {"execution_result": "실행 완료"}
def observe(self, state: dict):
"""실행 결과 관찰"""
# 실제 관찰 로직
return {"observation": "관찰 완료"}
def decide(self, state: dict):
"""결정 로직"""
# 실행 결과에 따라 결정
return {"decision": "다음 단계로 진행"}
# 워크플로우 구성
def build_multitool_workflow():
agent = MultiToolAgent()
workflow = StateGraph(AgentState)
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()
템플릿 3: 인간-중개 워크플로우
문제: 인간의 검토와 승인을 포함한 결정 워크플로우
import asyncio
from datetime import datetime
class HumanInLoopAgent:
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o-mini")
self.waiting_for_human = False
def generate(self, state: dict):
"""생성 작업"""
prompt = ChatPromptTemplate.from_messages([
("system", "다음 내용을 기반으로 생성하세요: {input}"),
("user", "결정이 필요한 경우 인간 검토가 필요합니다")
])
response = self.llm.invoke(prompt.format(input=state["messages"][-1].content))
return {"generated": response.content, "requires_human_review": True}
def human_review(self, state: dict):
"""인간 검토"""
# 실제 구현에서는 API 또는 대화형 인터페이스 필요
# 여기서는 가상의 검토 로직
return {"review_status": "approved", "reviewed_at": datetime.now()}
def continue_workflow(self, state: dict):
"""워크플로우 재개"""
return {"workflow_continued": True}
# 인간-중개 워크플로우
def build_human_in_loop_workflow():
agent = HumanInLoopAgent()
workflow = StateGraph(AgentState)
workflow.add_node("generate", agent.generate)
workflow.add_node("human_review", agent.human_review)
workflow.add_node("continue", agent.continue_workflow)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "human_review")
workflow.add_edge("human_review", "continue")
workflow.add_edge("continue", END)
return workflow.compile()
템플릿 5: 병렬 실행 에이전트
문제: 여러 작업을 동시에 처리하고 결과를 집계하는 병렬 처리 워크플로우
python
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List
class ParallelAgent:
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o-mini")
def fan_out(self, state: dict):
"""작업 분산"""
# 여러 작업으로 분할
tasks = ["분석1", "분석2", "분석3"]
return {"tasks": tasks, "task_results": []}
async def process_task(self, task: str):
"""개별 작업 처리"""
prompt = ChatPromptTemplate.from_messages([
("system", f"{task} 작업을 처리하세요"),
("user", "결과를 요약하여 반환하세요")
])
response = self.llm.invoke(prompt.format(task=task))
return f"{task}_결과: {response.content}"
def aggregate(self, state: dict):
"""결과 집계"""
# 병렬 처리 결과를 집계
results = state["task_results"]
summary = " ".join(results)
return {"aggregated_summary": summary}
# 병렬
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)