Implementing State-Based AI Workflows with LangGraph Templates

LangGraph 워크플로우 템플릿 (v39)

LangGraph provides a framework for building stateful, multi-actor applications. It utilizes Nodes, Edges, and Checkpointing to maintain and restore workflow states.

Why This Matters

Traditional LLM chains often lack the ability to handle complex cycles or recover from mid-process failures. By utilizing a state-based architecture with checkpointing, engineers can transition from linear sequences to iterative loops that allow for validation and human intervention, reducing the risk of unrecoverable execution errors in production agents.

Key Insights

• State Management: Using TypedDict with Annotated operators allows for specific state update behaviors, such as appending messages via operator.add.
• Cyclic Execution: The Multi-Tool Agent implements a Plan → Execute → Observe → Decide loop to handle complex tasks iteratively.
• Human-in-the-Loop: The HumanReviewState template enables process pausing via MemorySaver checkpointers for manual approval or rejection.
• Parallel Processing: The ParallelAgentState pattern utilizes fan_out and aggregation nodes to process data items concurrently before finalizing results.

Working Examples

Basic LangGraph workflow configuration with a shared state.

from langgraph.graph import StateGraph
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

workflow = StateGraph(AgentState)

Simple RAG agent workflow involving retrieval, generation, and validation.

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()

Practical Applications

References:

• https://dev.to/matias_yoon_738a24cb1190f/langgraph-weokeupeulrou-tempeulris-v39-293
• https://gumroad.com/l/auto ($5)