LangGraph 워크플로우 템플릿 (v18)
개요
LangGraph는 LangChain과 함께 사용되는 강력한 워크플로우 엔진으로, AI 에이전트의 복잡한 상호작용을 구조화하는 데 최적화되어 있습니다. 이 가이드에서는 실제 프로젝트에서 사용할 수 있는 4가지 핵심 템플릿을 제공합니다. 각 템플릿은 실제 개발자들이 직면하는 문제를 해결하기 위해 설계되었습니다.
1. LangGraph 아키텍처 개요
LangGraph는 다음과 같은 핵심 구성 요소로 구성됩니다:
- Nodes: 각각의 작업 단위 (함수 또는 클래스)
- Edges: 노드 간의 연결 및 전이 조건
- State: 워크플로우의 상태 관리
- Checkpointing: 상태 저장 및 복구 기능
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class GraphState(TypedDict):
messages: Annotated[list, operator.add]
# 기본 워크플로우 구조
workflow = StateGraph(GraphState)
2. 템플릿 1: 단순 RAG 에이전트 (검색 → 생성 → 검증)
이 템플릿은 문서 기반 질문 응답 시스템을 위한 최적화된 구조입니다.
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
class RAGState(TypedDict):
query: str
retrieved_docs: list
generated_answer: str
validation_result: bool
def retrieve_docs(state: RAGState):
# 검색 로직
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
vectorstore = Chroma(
persist_directory="./vectorstore",
embedding_function=OpenAIEmbeddings()
)
docs = vectorstore.similarity_search(state["query"], k=3)
return {"retrieved_docs": docs}
def generate_answer(state: RAGState):
# 답변 생성
template = """주어진 문서를 기반으로 다음 질문에 답변하세요:
{context}
질문: {query}
"""
prompt = PromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4")
chain = prompt | llm | StrOutputParser()
context = "\n\n".join([doc.page_content for doc in state["retrieved_docs"]])
answer = chain.invoke({
"context": context,
"query": state["query"]
})
return {"generated_answer": answer}
def validate_answer(state: RAGState):
# 답변 검증
validation_prompt = PromptTemplate.from_template(
"다음 답변이 질문에 대해 적절한지 판단하세요. 답변이 정확하고 문서에 기반한 것이면 true, 그렇지 않으면 false를 반환하세요:\n\n"
"질문: {query}\n"
"답변: {answer}\n"
"결과:"
)
llm = ChatOpenAI(model="gpt-4")
chain = validation_prompt | llm | StrOutputParser()
result = chain.invoke({
"query": state["query"],
"answer": state["generated_answer"]
})
return {"validation_result": result.lower() == "true"}
# 워크플로우 정의
def create_rag_workflow():
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_docs)
workflow.add_node("generate", generate_answer)
workflow.add_node("validate", validate_answer)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
# 검증 결과에 따라 분기
workflow.add_conditional_edges(
"validate",
lambda state: "retry" if not state["validation_result"] else "end",
{
"retry": "retrieve",
"end": END
}
)
return workflow.compile()
3. 템플릿 2: 멀티-도구 에이전트 (계획 → 실행 → 관찰 → 결정)
이 템플릿은 복잡한 작업을 계획하고 여러 도구를 활용하여 실행하는 에이전트를 구현합니다.
from typing import List, Dict, Any
from langchain.tools import Tool
class ToolAgentState(TypedDict):
task: str
plan: List[str]
execution_results: List[Dict[str, Any]]
current_step: int
final_result: str
# 도구 정의
tools = [
Tool(
name="web_search",
func=lambda query: f"웹 검색 결과: {query}",
description="웹에서 정보를 검색합니다"
),
Tool(
name="calculator",
func=lambda expression: eval(expression),
description="수학 계산을 수행합니다"
),
Tool(
name="data_analysis",
func=lambda data: f"분석 결과: {data}",
description="데이터 분석을 수행합니다"
)
]
def plan_task(state: ToolAgentState):
# 작업 계획 생성
planning_prompt = PromptTemplate.from_template(
"다음 작업을 단계별로 계획하세요:\n"
"작업: {task}\n"
"계획:"
)
llm = ChatOpenAI(model="gpt-4")
chain = planning_prompt | llm | StrOutputParser()
plan = chain.invoke({"task": state["task"]})
plan_steps = plan.split("\n")
return {"plan": plan_steps, "current_step": 0}
def execute_step(state: ToolAgentState):
# 단계 실행
if state["current_step"] >= len(state["plan"]):
return {"execution_results": [], "current_step": len(state["plan"])}
step = state["plan"][state["current_step"]]
tool_name = step.split(":")[0].strip()
# 도구 실행
result = None
for tool in tools:
if tool.name == tool_name:
result = tool.func(step)
break
return {
"execution_results": [{"step": step, "result": result}],
"current_step": state["current_step"] + 1
}
def observe_and_decide(state: ToolAgentState):
# 실행 결과 관찰 및 결정
if not state["execution_results"]:
return {"final_result": "실행 실패"}
# 결과를 기반으로 다음 행동 결정
decision_prompt = PromptTemplate.from_template(
"다음 실행 결과를 분석하고 다음 행동을 결정하세요:\n"
"결과: {results}\n"
"결론:"
)
llm = ChatOpenAI(model="gpt-4")
chain = decision_prompt | llm | StrOutputParser()
results_str = "\n".join([f"{r['step']}: {r['result']}" for r in state["execution_results"]])
decision = chain.invoke({"results": results_str})
return {"final_result": decision}
# 워크플로우 정의
def create_tool_agent_workflow():
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan_task)
workflow.add_node("execute", execute_step)
workflow.add_node("observe", observe_and_decide)
workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
return workflow.compile()
4. 템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)
이 템플릿은 인간의 개입이 필요한 작업을 처리하는 워크플로우를 제공합니다.
python
import asyncio
from typing import Optional
class HumanLoopState(TypedDict):
task: str
generated_output: str
user_review: Optional[str]
is_approved: Optional[bool]
retry_count: int
def generate_output(state: HumanLoopState):
# 초기 출력 생성
prompt = PromptTemplate.from_template(
"다음 작업을 수행하세요:\n"
"작업: {task}\n"
"출력:"
)
llm = ChatOpenAI(model="gpt-4")
chain = prompt | llm | StrOutputParser()
output = chain.invoke({"task": state["task"]})
return {"generated_output": output, "user_review": None, "is_approved": None}
def human_review(state: HumanLoopState):
# 인간 검토 (모의)
print(f"생성된 출력: {state['generated_output']}")
print("사용자 검토를 위한 입력을 기다리는 중...")
# 실제 구현에서는 이 부분에 인간 입력을 받는 로직을 추가
# 예: REST API 호출 또는 웹 UI 대기
return {"user_review": "검토 완료", "is_approved": True}
def
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)