Introduction

A basic Large Language Model application may send one prompt to one model and return the generated response. Production applications are usually more complex. They may validate input, retrieve authorized knowledge, call tools, generate structured output, request human approval, retry failed operations, preserve state, and record complete execution evidence.

Orchestration coordinates these components and controls the order in which they run. It determines what data each component receives, how intermediate results are stored, when the workflow branches or loops, and how failures are handled.

LangChain provides reusable abstractions for prompts, models, structured outputs, retrievers, tools, middleware, and agents. LangGraph provides lower-level orchestration for stateful workflows represented as graphs.

Learning Objectives

Prerequisites

python -m pip install \
  langchain \
  langgraph \
  langchain-openai \
  pydantic \
  python-dotenv

What Is LLM Orchestration?

LLM orchestration is the coordination of prompts, models, data, tools, validation, state, policies, and human decisions within one controlled application workflow.

What Is LangChain?

LangChain provides composable interfaces for connecting models with prompts, structured output, retrievers, tools, middleware, and agent workflows.

What Is LangGraph?

LangGraph represents an application as a graph consisting of state, nodes, and edges. It is useful when a workflow needs explicit branching, loops, persistence, interrupts, streaming, or human-in-the-loop behavior.

LangChain vs LangGraph

Chain vs Agent vs Graph

Configure a Chat Model

import os

from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model=(
        os.environ[
            "APPROVED_CHAT_MODEL"
        ]
    ),
    api_key=(
        os.environ[
            "APPROVED_MODEL_API_KEY"
        ]
    ),
    temperature=0
)

Create a Prompt Template

from langchain_core.prompts import ChatPromptTemplate


classification_prompt = (
    ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "Classify the supplied support request. "
                "Use only the approved categories: "
                "Bug, Billing, Access, How-To, or Other."
            ),
            (
                "human",
                "{support_request}"
            )
        ]
    )
)

Request Structured Output

from typing import Literal

from pydantic import BaseModel
from pydantic import Field


class SupportClassification(
    BaseModel
):
    category: Literal[
        "Bug",
        "Billing",
        "Access",
        "How-To",
        "Other"
    ]

    confidence: float = Field(
        ge=0.0,
        le=1.0
    )

    reason: str = Field(
        min_length=1
    )


structured_model = (
    model.with_structured_output(
        SupportClassification
    )
)

Compose a LangChain Runnable

classification_chain = (
    classification_prompt
    |
    structured_model
)


classification_result = (
    classification_chain.invoke(
        {
            "support_request": (
                "The application closes "
                "after I upload a document."
            )
        }
    )
)


print(
    classification_result.model_dump()
)

Add Deterministic Preprocessing

from langchain_core.runnables import RunnableLambda


def validate_support_request(
    input_data
):
    support_request = str(
        input_data.get(
            "support_request",
            ""
        )
    ).strip()

    if not support_request:

        raise ValueError(
            "A support request is required."
        )

    if len(
        support_request
    ) > 4000:

        raise ValueError(
            "The support request exceeds "
            "the approved length."
        )

    return {
        "support_request": support_request
    }


validated_classification_chain = (
    RunnableLambda(
        validate_support_request
    )
    |
    classification_prompt
    |
    structured_model
)

What Is a Tool?

A tool is a controlled function or operation that an application or model can request. Every tool should have a clear purpose, typed inputs, bounded permissions, validated outputs, and defined failure behavior.

Define a LangChain Tool

from langchain_core.tools import tool


@tool
def calculate_order_total(
    item_prices: list[
        float
    ]
) -> dict:
    """Calculate the sum of non-negative item prices."""

    if not item_prices:

        raise ValueError(
            "At least one item price is required."
        )

    if any(
        item_price < 0
        for item_price in item_prices
    ):
        raise ValueError(
            "Item prices cannot be negative."
        )

    total = sum(
        item_prices
    )

    return {
        "item_count": len(
            item_prices
        ),
        "total": round(
            total,
            2
        )
    }

Create a Tool-Calling Agent

from langchain.agents import create_agent


agent = create_agent(
    model=model,
    tools=[
        calculate_order_total
    ],
    system_prompt=(
        "You are a calculation assistant. "
        "Use the provided tool for order totals. "
        "Do not estimate totals manually."
    )
)


agent_result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": (
                    "Calculate the total for "
                    "12.50, 7.25, and 3.00."
                )
            }
        ]
    }
)

Retrieval-Augmented Chain

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables import RunnablePassthrough


def format_documents(
    documents
):
    formatted_sections = []

    for document in documents:

        source_id = document.metadata.get(
            "source_id",
            "unknown-source"
        )

        formatted_sections.append(
            f"Source: {source_id}\n"
            f"{document.page_content}"
        )

    return "\n\n".join(
        formatted_sections
    )


rag_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer using only the supplied context. "
            "If the context is insufficient, state that "
            "the answer cannot be established."
        ),
        (
            "human",
            "Question:\n{question}\n\n"
            "Context:\n{context}"
        )
    ]
)


rag_chain = (
    {
        "question": RunnablePassthrough(),
        "context": (
            retriever
            |
            RunnableLambda(
                format_documents
            )
        )
    }
    |
    rag_prompt
    |
    model
    |
    StrOutputParser()
)

Treat Retrieved Content as Untrusted

LangGraph State

State is the shared data structure passed through the graph. Each node reads required state fields and returns updates.

from typing import TypedDict


class WorkflowState(
    TypedDict,
    total=False
):
    user_request: str

    request_category: str

    retrieved_context: list[
        str
    ]

    draft_response: str

    validation_errors: list[
        str
    ]

    retry_count: int

    final_response: str

LangGraph Nodes

A node is a Python callable representing one workflow operation. It receives the current state and returns a state update.

def normalize_request(
    state: WorkflowState
) -> dict:
    user_request = str(
        state.get(
            "user_request",
            ""
        )
    ).strip()

    if not user_request:

        raise ValueError(
            "A user request is required."
        )

    return {
        "user_request": user_request,
        "retry_count": 0,
        "validation_errors": []
    }


def classify_request(
    state: WorkflowState
) -> dict:
    request_text = state[
        "user_request"
    ].lower()

    if "policy" in request_text:

        category = "knowledge"

    elif "calculate" in request_text:

        category = "calculation"

    else:

        category = "general"

    return {
        "request_category": category
    }

Edges and Control Flow

Edges connect nodes. A normal edge always follows one path. A conditional edge selects the next node according to routing logic.

Build a Basic LangGraph

from langgraph.graph import END
from langgraph.graph import START
from langgraph.graph import StateGraph


workflow_builder = StateGraph(
    WorkflowState
)


workflow_builder.add_node(
    "normalize_request",
    normalize_request
)


workflow_builder.add_node(
    "classify_request",
    classify_request
)


workflow_builder.add_edge(
    START,
    "normalize_request"
)


workflow_builder.add_edge(
    "normalize_request",
    "classify_request"
)


workflow_builder.add_edge(
    "classify_request",
    END
)


workflow = workflow_builder.compile()


workflow_result = workflow.invoke(
    {
        "user_request": (
            "Explain the approved policy."
        )
    }
)


print(
    workflow_result
)

Conditional Routing

def route_request(
    state: WorkflowState
) -> str:
    category = state[
        "request_category"
    ]

    allowed_routes = {
        "knowledge",
        "calculation",
        "general"
    }

    if category not in allowed_routes:

        raise ValueError(
            "Unsupported request category."
        )

    return category


def retrieve_knowledge(
    state: WorkflowState
) -> dict:
    return {
        "retrieved_context": [
            "Replace with authorized "
            "retrieval results."
        ]
    }


def perform_calculation(
    state: WorkflowState
) -> dict:
    return {
        "draft_response": (
            "Replace with a validated "
            "calculation result."
        )
    }


def answer_general_request(
    state: WorkflowState
) -> dict:
    return {
        "draft_response": (
            "Replace with the configured "
            "general-response chain."
        )
    }

workflow_builder.add_node(
    "retrieve_knowledge",
    retrieve_knowledge
)


workflow_builder.add_node(
    "perform_calculation",
    perform_calculation
)


workflow_builder.add_node(
    "answer_general_request",
    answer_general_request
)


workflow_builder.add_conditional_edges(
    "classify_request",
    route_request,
    {
        "knowledge": "retrieve_knowledge",
        "calculation": "perform_calculation",
        "general": "answer_general_request"
    }
)

Controlled Loops

Graphs can loop when an output requires correction, another tool call, additional evidence, or human input. Every loop must have a clear termination condition.

MAXIMUM_RETRIES = 2


def validate_response(
    state: WorkflowState
) -> dict:
    draft_response = str(
        state.get(
            "draft_response",
            ""
        )
    ).strip()

    validation_errors = []

    if not draft_response:

        validation_errors.append(
            "Response is empty."
        )

    if len(
        draft_response
    ) > 5000:

        validation_errors.append(
            "Response exceeds the approved length."
        )

    return {
        "validation_errors": (
            validation_errors
        )
    }


def route_after_validation(
    state: WorkflowState
) -> str:
    validation_errors = state.get(
        "validation_errors",
        []
    )

    retry_count = state.get(
        "retry_count",
        0
    )

    if not validation_errors:

        return "complete"

    if retry_count >= MAXIMUM_RETRIES:

        return "fallback"

    return "revise"

Message State and Reducers

Conversational workflows commonly need to accumulate messages rather than replace them after every node.

from langgraph.graph import MessagesState


class AgentState(
    MessagesState
):
    approved: bool

    action_summary: str

Short-Term Persistence

A checkpointer can preserve graph state for a workflow thread. A thread identifier connects repeated invocations to the correct persisted state.

from langgraph.checkpoint.memory import InMemorySaver


checkpointer = InMemorySaver()


persistent_workflow = (
    workflow_builder.compile(
        checkpointer=checkpointer
    )
)


workflow_configuration = {
    "configurable": {
        "thread_id": (
            "generated-workflow-thread-id"
        )
    }
}


result = persistent_workflow.invoke(
    {
        "user_request": (
            "Explain the approved process."
        )
    },
    config=workflow_configuration
)

Short-Term State vs Long-Term Memory

Human-in-the-Loop Approval

A graph can pause before a consequential action and request an external approval decision.

from langgraph.types import interrupt


def request_action_approval(
    state: WorkflowState
) -> dict:
    approval_decision = interrupt(
        {
            "action": (
                "proposed-consequential-action"
            ),
            "summary": state.get(
                "draft_response",
                ""
            ),
            "required_decision": [
                "approve",
                "reject"
            ]
        }
    )

    if approval_decision not in {
        "approve",
        "reject"
    }:
        raise ValueError(
            "Unsupported approval decision."
        )

    return {
        "final_response": (
            "Action approved."
            if approval_decision == "approve"
            else "Action rejected."
        )
    }

Resume an Interrupted Graph

from langgraph.types import Command


resumed_result = (
    persistent_workflow.invoke(
        Command(
            resume="approve"
        ),
        config=workflow_configuration
    )
)

Streaming Graph Updates

for workflow_update in (
    persistent_workflow.stream(
        {
            "user_request": (
                "Explain the approved process."
            )
        },
        config=workflow_configuration,
        stream_mode="updates"
    )
):
    print(
        workflow_update
    )

Retry and Fallback

Timeout Wrapper

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError


def run_with_timeout(
    function,
    timeout_seconds,
    *args,
    **kwargs
):
    if timeout_seconds <= 0:

        raise ValueError(
            "timeout_seconds must be positive."
        )

    with ThreadPoolExecutor(
        max_workers=1
    ) as executor:

        future = executor.submit(
            function,
            *args,
            **kwargs
        )

        try:
            return future.result(
                timeout=timeout_seconds
            )

        except TimeoutError as error:

            future.cancel()

            raise TimeoutError(
                "Operation exceeded the "
                "approved timeout."
            ) from error

Multi-Agent Orchestration

Multi-agent systems divide work among specialized agent roles. They can be useful when roles require distinct tools, context, policies, or evaluation criteria.

Recursion and Iteration Limits

graph_configuration = {
    "configurable": {
        "thread_id": (
            "generated-thread-id"
        )
    },
    "recursion_limit": 20
}


result = workflow.invoke(
    {
        "user_request": (
            "Process the supplied request."
        )
    },
    config=graph_configuration
)

Test Nodes Independently

def test_normalize_request():

    result = normalize_request(
        {
            "user_request": (
                "  Explain the policy.  "
            )
        }
    )

    assert result[
        "user_request"
    ] == "Explain the policy."

    assert result[
        "retry_count"
    ] == 0


def test_normalize_request_rejects_empty_input():

    try:
        normalize_request(
            {
                "user_request": "   "
            }
        )

        assert False

    except ValueError as error:

        assert (
            "required"
            in str(
                error
            ).lower()
        )

Test Routing

def test_knowledge_route():

    route = route_request(
        {
            "request_category": (
                "knowledge"
            )
        }
    )

    assert route == "knowledge"


def test_unsupported_route():

    try:
        route_request(
            {
                "request_category": (
                    "unknown"
                )
            }
        )

        assert False

    except ValueError:

        assert True

End-to-End Graph Testing

Observability

Orchestration Metrics

Security Controls

Prompt Injection Defense

Cost and Performance

Version the Complete Workflow

Rollback

Common Orchestration Mistakes

Troubleshooting

Recommended Orchestration Workflow

Hands-On Practice

Knowledge Check

Orchestration Readiness Checklist

Lesson Summary