LangGraph 워크플로우 템플릿 (v23)
소개
LangGraph는 LangChain과 함께 사용되는 고급 AI 에이전트 구축 프레임워크입니다. 이 가이드에서는 실제 개발자가 문제를 해결할 수 있는 4가지 핵심 워크플로우 템플릿을 제공합니다. 각 템플릿은 실제 프로덕션 환경에서 사용될 수 있는 최적화된 구조를 갖추고 있으며, Python 코드와 실제 사용 예시를 포함합니다.
1. LangGraph 아키텍처 개요
LangGraph는 그래프 기반의 워크플로우 시스템으로, 다음 구성 요소로 구성됩니다:
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
# 상태 정의
class State(TypedDict):
messages: list
current_step: str
tool_calls: dict
# 노드 정의
def retrieve_node(state):
# 검색 로직
return {"messages": [f"Retrieved: {state['query']}"]}
def generate_node(state):
# 생성 로직
return {"messages": [f"Generated: {state['query']}"]}
# 그래프 생성
workflow = StateGraph(State)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
# 엣지 정의
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)
# 체크포인트 저장
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)
이 템플릿은 검색-생성-검증 파이프라인을 구현합니다. 실시간 데이터 검색과 결과 검증 기능을 포함합니다.
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
import json
class RAGState(TypedDict):
query: str
retrieved_docs: list
generated_answer: str
validation_result: dict
current_step: Literal["retrieve", "generate", "validate"]
class RAGAgent:
def __init__(self, vectorstore, llm):
self.vectorstore = vectorstore
self.llm = llm
def retrieve(self, state):
docs = self.vectorstore.similarity_search(state["query"], k=3)
return {
"retrieved_docs": [doc.page_content for doc in docs],
"current_step": "generate"
}
def generate(self, state):
context = "\n".join(state["retrieved_docs"])
prompt = PromptTemplate.from_template("""
주어진 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트: {context}
질문: {query}
답변:
""")
chain = prompt | self.llm
answer = chain.invoke({
"context": context,
"query": state["query"]
})
return {
"generated_answer": answer.content,
"current_step": "validate"
}
def validate(self, state):
# 간단한 검증 로직
validation = {
"is_valid": len(state["generated_answer"]) > 10,
"confidence": 0.85,
"feedback": "검증 완료"
}
return {
"validation_result": validation,
"current_step": "end"
}
def create_graph(self):
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", self.retrieve)
workflow.add_node("generate", self.generate)
workflow.add_node("validate", self.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 = RAGAgent(vectorstore, ChatOpenAI(model="gpt-4"))
# app = rag_agent.create_graph()
# result = app.invoke({"query": "Python 데이터 분석 방법"})
3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)
이 템플릿은 작업 계획, 실행, 관찰 및 결정을 수행하는 다중 도구 에이전트입니다. 복잡한 작업을 단계별로 분해하고 실행합니다.
from typing import TypedDict, Literal, Any
from langgraph.graph import StateGraph, END
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
class ToolAgentState(TypedDict):
task: str
plan: list
execution_log: list
current_step: Literal["plan", "execute", "observe", "decide"]
results: dict
class ToolAgent:
def __init__(self, tools, llm):
self.tools = tools
self.llm = llm
def plan(self, state):
# 작업 계획 생성
plan_prompt = """
다음 작업을 수행하는 데 필요한 단계를 계획하세요:
{task}
가능한 도구 목록:
{tools}
계획을 JSON 형식으로 반환하세요.
"""
plan = self.llm.invoke(plan_prompt.format(
task=state["task"],
tools=[tool.name for tool in self.tools]
))
plan_data = json.loads(plan.content)
return {
"plan": plan_data["steps"],
"current_step": "execute"
}
def execute(self, state):
# 각 단계 실행
step = state["plan"][0] if state["plan"] else None
if not step:
return {"current_step": "decide"}
# 도구 실행
tool_result = self.execute_tool(step["tool"], step["input"])
return {
"execution_log": state["execution_log"] + [tool_result],
"plan": state["plan"][1:] if len(state["plan"]) > 1 else [],
"current_step": "observe"
}
def execute_tool(self, tool_name, input_data):
# 특정 도구 실행
tool = next((t for t in self.tools if t.name == tool_name), None)
if not tool:
return {"error": f"도구 {tool_name}를 찾을 수 없습니다"}
try:
result = tool.run(input_data)
return {"tool": tool_name, "input": input_data, "result": result}
except Exception as e:
return {"tool": tool_name, "error": str(e)}
def observe(self, state):
# 실행 결과 관찰
if state["execution_log"]:
last_result = state["execution_log"][-1]
if "error" in last_result:
# 오류 처리
return {"current_step": "decide"}
return {"current_step": "execute"}
def decide(self, state):
# 최종 결정
if not state["plan"]:
return {"current_step": "end"}
return {"current_step": "plan"}
# 사용 예시
tools = [
Tool(name="search", func=lambda x: f"검색 결과: {x}", description="검색 도구"),
Tool(name="calculate", func=lambda x: f"계산 결과: {x}", description="계산 도구")
]
# tool_agent = ToolAgent(tools, ChatOpenAI(model="gpt-4"))
# app = tool_agent.create_graph()
4. 템플릿 3: 인간-중개 워크플로우 (중단 → 검토 → 계속)
이 템플릿은 인간 검토를 포함하는 워크플로우를 구현합니다. 중요한 결정이나 검증 작업에서 인간의 개입이 필요한 경우 사용됩니다.
python
from typing import TypedDict, Literal, Any
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class HumanInLoopState(TypedDict):
task: str
generated_output: str
human_review: dict
current_step: Literal["generate", "pause", "review", "continue"]
is_approved: bool
class HumanInLoopAgent:
def __init__(self, llm):
self.llm = llm
def generate(self, state):
# 작업 생성
prompt = f"다음 작업을 수행하세요:\n{state['task']}"
result = self.llm.invoke(prompt)
return {
"generated_output": result.content,
"current_step": "pause"
}
def pause(self, state):
# 인간 검토를 위한 중단
return {
"current_step": "review"
}
def review(self, state):
# 인간 검토
# 실제 구현에서는 외부 시스템 또는 UI를 통해 인간이 검토
# 여기서는 예시로 설정
human_decision = input("
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)