LangGraph 워크플로우 템플릿 (v16)
LangGraph 아키텍처 개요
LangGraph는 LangChain과 함께 사용되는 상태 기반 워크플로우 프레임워크로, 복잡한 AI 에이전트를 구축하는 데 강력한 도구를 제공합니다. 핵심 구성 요소는 다음과 같습니다:
노드 (Nodes)
- 실행할 작업 단위 (LLM 호출, 도구 실행, 상태 업데이트 등)
- 각 노드는 입력 상태를 받아 출력 상태를 반환
엣지 (Edges)
- 노드 간의 흐름을 정의
- 조건부 엣지로 복잡한 로직 구현 가능
상태 (State)
- 워크플로우 전체에 공유되는 상태 객체
- 각 노드는 상태를 읽고 수정할 수 있음
체크포인팅 (Checkpointing)
- 워크플로우 상태를 저장하고 복구 가능
- 실패 시 재시작 및 복구 기능 제공
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import Message
import operator
class GraphState(TypedDict):
messages: Annotated[list[Message], operator.add]
# 기본 워크플로우 생성
workflow = StateGraph(GraphState)
템플릿 1: 단순 RAG 에이전트 (검색 → 생성 → 검증)
복잡한 RAG 시스템을 단순화하여 개발 속도를 높이는 템플릿입니다.
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
class RAGAgent:
def __init__(self, vectorstore, llm):
self.vectorstore = vectorstore
self.llm = llm
def retrieve(self, state):
# 검색 단계
question = state["messages"][-1].content
docs = self.vectorstore.similarity_search(question, k=3)
return {"retrieved_docs": docs}
def generate(self, state):
# 생성 단계
context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
template = PromptTemplate.from_template("""
질문: {question}
컨텍스트: {context}
위의 컨텍스트를 기반으로 질문에 답변하세요.
""")
prompt = template.format(
question=state["messages"][-1].content,
context=context
)
response = self.llm.invoke(prompt)
return {"generated_response": response}
def validate(self, state):
# 검증 단계
# 간단한 검증 로직 (실제 사용 시 더 복잡한 검증 포함)
response = state["generated_response"]
if len(response.content) < 20:
return {"validation": "FAILED", "reason": "답변이 너무 짧습니다"}
return {"validation": "PASSED"}
# 워크플로우 구성
def create_rag_workflow():
agent = RAGAgent(vectorstore=Chroma(), llm=ChatOpenAI())
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", RunnableLambda(agent.retrieve))
workflow.add_node("generate", RunnableLambda(agent.generate))
workflow.add_node("validate", RunnableLambda(agent.validate))
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
return workflow.compile()
# 사용 예시
rag_workflow = create_rag_workflow()
result = rag_workflow.invoke({
"messages": [Message(content="Python으로 웹 애플리케이션을 만드는 방법은?")]
})
템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)
다중 도구 사용을 위한 순차적 워크플로우로, 복잡한 작업을 분해하고 실행합니다.
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI
class MultiToolAgent:
def __init__(self, tools):
self.tools = tools
self.llm = ChatOpenAI()
def plan(self, state):
# 작업 계획 생성
prompt = """
다음 작업을 수행할 계획을 세우세요:
{task}
가능한 도구들:
{tools}
각 작업에 대해 하나의 도구만 선택하고 계획을 생성하세요.
"""
tools_info = "\n".join([f"- {tool.name}: {tool.description}" for tool in self.tools])
plan = self.llm.invoke(prompt.format(
task=state["messages"][-1].content,
tools=tools_info
))
return {"plan": plan.content}
def execute(self, state):
# 도구 실행
plan = state["plan"]
# 간단한 도구 선택 로직 (실제 구현 시 더 복잡한 로직 필요)
tool_name = "search_tool" # 예시
tool = next((t for t in self.tools if t.name == tool_name), None)
if tool:
result = tool.invoke({"query": state["messages"][-1].content})
return {"execution_result": result}
return {"execution_result": "도구를 찾을 수 없습니다"}
def observe(self, state):
# 실행 결과 관찰
result = state["execution_result"]
return {"observation": f"실행 결과: {result}"}
def decide(self, state):
# 결정 단계
observation = state["observation"]
# 판단 로직 (실제 구현 시 더 복잡한 의사결정 포함)
if "실패" in observation:
return {"decision": "RETRY"}
return {"decision": "COMPLETE"}
def create_multitool_workflow():
# 도구 정의
search_tool = Tool(
name="search_tool",
description="웹 검색을 위한 도구",
func=lambda x: f"검색 결과: {x}"
)
calculator_tool = Tool(
name="calculator_tool",
description="수학 계산을 위한 도구",
func=lambda x: f"계산 결과: {eval(x)}"
)
tools = [search_tool, calculator_tool]
agent = MultiToolAgent(tools)
workflow = StateGraph(GraphState)
workflow.add_node("plan", RunnableLambda(agent.plan))
workflow.add_node("execute", RunnableLambda(agent.execute))
workflow.add_node("observe", RunnableLambda(agent.observe))
workflow.add_node("decide", RunnableLambda(agent.decide))
workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
# 조건부 엣지
def router(state):
if state["decision"] == "RETRY":
return "plan"
return END
workflow.add_conditional_edges("decide", router)
return workflow.compile()
# 사용 예시
multi_tool_workflow = create_multitool_workflow()
result = multi_tool_workflow.invoke({
"messages": [Message(content="123 + 456을 계산해주세요")]
})
템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)
사람의 판단이 필요한 경우에 사용되는 인간-중개 워크플로우입니다.
python
import time
from typing import Literal
from langchain_core.runnables import RunnableLambda
class HumanInLoopAgent:
def __init__(self, llm):
self.llm = llm
def generate(self, state):
# 생성 단계
prompt = """
다음 요청에 대한 답변을 생성하세요:
{request}
답변을 생성한 후 다음 작업을 진행하세요:
1. 답변을 검토하세요
2. 검토 결과를 다음 형식으로 출력하세요:
{"status": "review_required", "reviewer": "human", "decision": "continue"}
"""
response = self.llm.invoke(prompt.format(
request=state["messages"][-1].content
))
return {"generated_response": response, "status": "review_required"}
def review(self, state):
# 검토 단계 (인간이 직접 확인)
print("=== 검토 대기 ===")
print(f"생성된 답변: {state['generated_response'].content}")
print("검토 여부 (y/n): ", end="")
decision = input().lower()
return {"review_status": "approved" if decision == 'y' else "rejected"}
def process_decision(self, state):
# 결정 처리
if state["review_status"] == "rejected":
# 재시작 또는 다른 처리
return {"final_decision": "retry"}
return {"final_decision": "completed"}
def
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)
Top comments (0)