Build AI agents faster with LangGraph workflow templates in Python

Python developers can now accelerate agentic AI development using prebuilt LangGraph templates designed for real-world use cases. These modular workflows handle everything from retrieval-augmented generation to parallel execution and human oversight, eliminating months of boilerplate code while maintaining full customization flexibility.

Understanding LangGraph's core architecture

LangGraph provides a state-driven framework for orchestrating AI agents through four foundational components. Each workflow begins with nodes representing discrete processing steps, which connect via edges to define execution order. The state object tracks all shared data across the graph, while checkpointing enables fault tolerance by saving intermediate states for recovery.

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

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]
    context: str

workflow = StateGraph(GraphState)

This pattern proves particularly valuable when building systems requiring memory, such as chatbots or document analysis pipelines where context must persist across multiple steps.

Template 1: Streamlining RAG workflows

Most production LLM applications rely on retrieval-augmented generation (RAG) to ground responses in verified sources. This template implements a three-stage pipeline where queries first retrieve relevant documents, then generate answers based on retrieved context, and finally validate output quality.

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

class RAGAgent:
    def __init__(self, llm):
        self.llm = llm
        self.retriever = self._setup_retriever()
        self.prompt = PromptTemplate.from_template(
            "Use the provided context to answer the question:\n\n{context}\n\nQuestion: {question}"
        )

    def retrieve(self, state):
        question = state["messages"][-1].content
        docs = self.retriever.invoke(question)
        context = "\n\n".join([doc.page_content for doc in docs])
        return {"context": context}

    def generate(self, state):
        chain = self.prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "question": state["messages"][-1].content,
            "context": state["context"]
        })
        return {"messages": [result]}

The validation stage double-checks response completeness, ensuring generated answers contain sufficient information before final delivery to users.

Template 2: Coordinating multi-tool agents

Complex tasks often require orchestrating multiple specialized tools. This workflow demonstrates how to design agents that first plan a sequence of actions, execute each tool call, observe results, and finally decide whether to continue or stop.

from langchain.tools import Tool
from langchain_core.messages import ToolMessage

class MultiToolAgent:
    def __init__(self):
        self.tools = [
            Tool(
                name="weather",
                func=lambda location: f"Weather in {location} is sunny",
                description="Fetch weather information"
            ),
            Tool(
                name="calculator",
                func=lambda expression: f"Result: {eval(expression)}",
                description="Perform mathematical calculations"
            )
        ]

The planner creates an execution roadmap, the executor runs each tool in sequence, the observer analyzes outputs, and the decider determines whether to proceed or abort based on accumulated results.

Template 3: Human-in-the-loop decision making

Some workflows require human approval for critical decisions. This template introduces pause nodes that halt execution until a user reviews and provides feedback, then resumes the process automatically.

class HumanInLoopAgent:
    def pause_for_review(self, state):
        return {"status": "pending_review"}

    def review(self, state):
        if self.user_feedback:
            return {"messages": [self.user_feedback]}
        return {"messages": ["User review required"]}

This approach proves essential for applications handling sensitive data or requiring regulatory compliance, where automated decisions can't replace human oversight.

Template 4: Parallel processing for data volumes

When dealing with large datasets, processing chunks concurrently dramatically reduces latency. This template splits input into parallel branches, processes each chunk independently, then aggregates results in a final step.

from concurrent.futures import ThreadPoolExecutor

class ParallelAgent:
    def fan_out(self, state):
        data_chunks = [{"chunk_id": i, "data": f"chunk_{i}"} for i in range(4)]
        return {"chunks": data_chunks}

    def parallel_process(self, state):
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = [
                executor.submit(self.process_chunk, chunk)
                for chunk in state["chunks"]
            ]
            results = [future.result() for future in futures]
        return {"processed_results": results}

The fan-out phase divides work, parallel execution handles the heavy lifting, and aggregation produces a unified output from distributed results.

State management and checkpointing

Maintaining workflow consistency across restarts requires robust state handling. This template demonstrates how to implement checkpoints using memory storage, enabling recovery from crashes or service interruptions.

from langgraph.checkpoint.memory import MemorySaver

class StateManager:
    def __init__(self):
        self.checkpointer = MemorySaver()

    def update_state(self, thread_id, new_state):
        self.checkpointer.put(thread_id, new_state)

    def rollback(self, thread_id, version):
        # Restore state to previous version
        pass

This mechanism ensures agents can resume from the exact point of failure without losing critical context or partial results.

Real-time streaming for responsive agents

Modern AI applications demand responsive user experiences. This template adds streaming capabilities to monitor agent execution in real time, useful for debugging or building interactive interfaces.

from langchain_core.callbacks import BaseCallbackHandler

class StreamingHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token, **kwargs):
        print(f"Token: {token}", end="", flush=True)

Implementing streaming handlers transforms static responses into dynamic conversations where users see answers materialize word by word, significantly improving perceived responsiveness.

Next steps for AI developers

These LangGraph templates provide production-ready starting points that eliminate months of development time while offering full customization flexibility. Whether building internal tools, customer-facing assistants, or complex data pipelines, the modular architecture scales from simple prototypes to enterprise-grade systems. The community continues to expand these patterns, so staying updated with the latest releases ensures access to cutting-edge agentic AI capabilities.