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

#ai #llm #developers #tutorial

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

Product ID: LangGraph-Templates-v47

Price: $5

Target: Python developers building AI agents with LangChain/LangGraph

Problem: Lack of reusable, production-ready LangGraph templates for common AI agent patterns

Overview

LangGraph is the foundation for building autonomous AI agents with fine-grained control over execution flow, memory, and decision-making. This guide provides 5 reusable templates that solve common agent architecture challenges, from simple RAG workflows to human-in-the-loop decision systems.

Each template includes:

Complete working code
Architecture explanations
Error handling patterns
Deployment considerations
Memory and performance optimization tips

1. LangGraph Architecture Overview

Core Components

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage

# 1. State Definition
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

# 2. Node Functions
def retrieve_node(state):
    # Your retrieval logic here
    return {"messages": [HumanMessage(content="Retrieved data")]} 

def generate_node(state):
    # Your generation logic here
    return {"messages": [HumanMessage(content="Generated response")]}

# 3. Graph Construction
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_edge("retrieve", "generate")
workflow.set_entry_point("retrieve")

Key Concepts:

Nodes: Individual processing steps
Edges: Flow control between nodes
State: Shared memory across execution
Checkpointing: Persistent state management for recovery

2. Template 1: Simple RAG Agent (Retrieve → Generate → Validate)

Use Case: Document QA with validation and feedback loops

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnablePassthrough
from typing import List
import operator

class RAGState(TypedDict):
    question: str
    retrieved_docs: List[str]
    generated_answer: str
    validation_result: bool

def retrieve_docs(state):
    # Simulate vector store lookup
    vectorstore = FAISS.load_local("vectorstore", embeddings)
    docs = vectorstore.similarity_search(state["question"])
    return {"retrieved_docs": [doc.page_content for doc in docs]}

def generate_answer(state):
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "Answer the question based on context:\n{context}\nQuestion: {question}"
    )
    chain = prompt | llm
    answer = chain.invoke({
        "context": "\n".join(state["retrieved_docs"]),
        "question": state["question"]
    })
    return {"generated_answer": answer.content}

def validate_answer(state):
    # Simple validation logic
    if len(state["generated_answer"]) < 20:
        return {"validation_result": False}
    return {"validation_result": True}

# Build the workflow
rag_workflow = StateGraph(RAGState)
rag_workflow.add_node("retrieve", retrieve_docs)
rag_workflow.add_node("generate", generate_answer)
rag_workflow.add_node("validate", validate_answer)
rag_workflow.set_entry_point("retrieve")

# Add conditional edges
def validate_condition(state):
    if not state["validation_result"]:
        return "generate"
    return END

rag_workflow.add_edge("retrieve", "generate")
rag_workflow.add_edge("generate", "validate")
rag_workflow.add_conditional_edges("validate", validate_condition, {
    "generate": "generate",
    END: END
})

rag_app = rag_workflow.compile()

3. Template 2: Multi-tool Agent (Plan → Execute → Observe → Decide)

Use Case: Complex problem-solving with tool selection and execution

from langchain.tools import Tool
from langchain_openai import ChatOpenAI

# Tool definitions
tools = [
    Tool(
        name="web_search",
        func=lambda q: f"Search results for: {q}",
        description="Search the web for information"
    ),
    Tool(
        name="calculate",
        func=lambda x, y: f"Result: {int(x) * int(y)}",
        description="Perform calculations"
    )
]

class ToolAgentState(TypedDict):
    plan: str
    tools_used: List[str]
    observations: List[str]
    decision: str

def plan_node(state):
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "Plan how to solve: {question}\nAvailable tools: {tools}"
    )
    plan = llm.invoke(prompt.format(
        question=state["question"],
        tools=", ".join([t.name for t in tools])
    ))
    return {"plan": plan.content}

def execute_node(state):
    # Execute tools based on plan
    tool_results = []
    for tool in tools:
        if tool.name in state["plan"]:
            result = tool.func("query")
            tool_results.append(result)
    return {"tools_used": [t.name for t in tools if t.name in state["plan"]], 
            "observations": tool_results}

def observe_node(state):
    # Process observations
    return {"observations": state["observations"]}

def decide_node(state):
    # Make decision based on observations
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "Based on observations: {observations}\nMake a decision:"
    )
    decision = llm.invoke(prompt.format(observations=state["observations"]))
    return {"decision": decision.content}

# Workflow
tool_workflow = StateGraph(ToolAgentState)
tool_workflow.add_node("plan", plan_node)
tool_workflow.add_node("execute", execute_node)
tool_workflow.add_node("observe", observe_node)
tool_workflow.add_node("decide", decide_node)

tool_workflow.set_entry_point("plan")
tool_workflow.add_edge("plan", "execute")
tool_workflow.add_edge("execute", "observe")
tool_workflow.add_edge("observe", "decide")
tool_workflow.add_edge("decide", END)

tool_app = tool_workflow.compile()

4. Template 3: Human-in-the-loop Workflow (Pause → Review → Continue)

Use Case: Critical decision-making requiring human approval

class HumanLoopState(TypedDict):
    request: str
    analysis: str
    human_review: str
    approval: bool
    final_output: str

def analyze_request(state):
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "Analyze request: {request}\nProvide analysis:"
    )
    analysis = llm.invoke(prompt.format(request=state["request"]))
    return {"analysis": analysis.content}

def human_review_node(state):
    # This is a placeholder - in practice, this would interface with a UI
    print(f"Human Review Required: {state['analysis']}")
    # Simulate human approval
    approval = input("Approve? (y/n): ").lower() == 'y'
    return {"approval": approval}

def generate_output(state):
    if not state["approval"]:
        return {"final_output": "Request rejected by human"}
    llm = ChatOpenAI(model="gpt-4")
    prompt = PromptTemplate.from_template(
        "Generate final response to: {request}\nBased on analysis: {analysis}"
    )
    output = llm.invoke(prompt.format(
        request=state["request"],
        analysis=state["analysis"]
    ))
    return {"final_output": output.content}

# Workflow
human_workflow = StateGraph(HumanLoopState)
human_workflow.add_node("analyze", analyze_request)
human_workflow.add_node("review", human_review_node)
human_workflow.add_node("generate", generate_output)

human_workflow.set_entry_point("analyze")
human_workflow.add_edge("analyze", "review")
human_workflow.add_edge("review", "generate")
human_workflow.add_edge("generate", END)

human_app = human_workflow.compile()

5. Template 4: Parallel Execution Agent (Fan-out → Process → Aggregate)

Use Case: Data processing across multiple parallel workers


python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelState(TypedDict):
    input_data: List[str]
    processed_results: List[str]
    aggregated_result: str

def fan_out_node(state):
    # Split data into chunks for parallel processing
    chunks = [state["input_data"][i:i+2] for i in range(0, len(state["input_data"]), 2)]
    return {"processed_results": chunks}

def process_chunk(state):
    # Simulate processing each chunk
    chunk = state["processed_results"][-1]  # Last chunk
    results = []
    for item in chunk:
        results.append(f"Processed: {item}")
    return {"processed_results": state["processed_results"][:-1] + [results]}

def aggregate_results(state):
    # Combine all results
    all_results = []
    for chunk_result in state["processed_results"]:
        all_results.extend(chunk_result)
    return {"aggregated_result": " | ".join(all_results)}

# Workflow
parallel_workflow = StateGraph(ParallelState)
parallel_workflow.add_node("fan_out", fan_out_node)
parallel_workflow.add_node("process", process_chunk)
parallel_workflow.add_node("aggregate", aggregate_results)

parallel_workflow.set_entry_point("

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)