Prompt Chaining | Agentic Workflow Design Pattern

Agentic workflows combine Large Language Models (LLMs) and programmatic tools along deterministic, structured execution paths. As the foundational design pattern in agentic engineering, Prompt Chaining decomposes a complex objective into sequential, interdependent subtasks where each model call consumes the structured output of its predecessor.

1. Introduction to Agentic Workflows & Deterministic Control

In AI engineering, a workflow is defined as an orchestrated system where Large Language Models, heuristic logic, and external tools interact through predefined execution graphs. Unlike unstructured or fully autonomous agents that make unbounded decisions in a loop, an agentic workflow maintains strict developer oversight over execution paths.

Core Tenets of Orchestrated Workflows

• Traceability: At any point during runtime, the system knows the exact node being executed and the next transition in the state machine.
• Predictability: Complex, multi-stage reasoning tasks are broken down into discrete steps with verified pre-conditions and post-conditions.
• Cost & Latency Management: Programmatic boundaries prevent infinite tool-calling loops and redundant inference calls.

2. The Prompt Chaining Pattern

The Prompt Chaining design pattern solves tasks by sequentially linking multiple LLM calls. Each node in the chain performs a specialized transformation, and its output directly seeds the input context for the subsequent node.

Key Characteristics

1. Unidirectional Control Flow: Information moves forward through discrete, specialized stages rather than relying on a single monolithic prompt.
2. Context Specialization: Instead of overloading a single model context window with complex system instructions, each step uses a dedicated prompt tailored to an isolated task (e.g., classification, extraction, translation, or synthesis).
3. Data Dependency: Node $N$ cannot execute until Node $N-1$ produces a validated intermediate artifact.

3. Programmatic Checks and Conditional Gates

A common failure mode in naive LLM pipelines is cascading hallucination—when an error in an early stage compounds and causes subsequent stages to produce invalid or costly outputs.

Prompt Chaining mitigates this by embedding Programmatic Gates (also referred to as Grader Nodes or Conditional Routers).

Functional Role of a Gate

• Validation: Verifies whether the output of a prior LLM call meets schema requirements, confidence thresholds, or domain constraints.
• Early Termination: Halts processing when an input is out-of-scope, malicious, or malformed, conserving API costs and compute time.
• Conditional Branching: Dynamically directs the execution pointer to alternative subgraphs (e.g., fallback prompts, escalation paths, or specialized downstream nodes).

4. End-to-End System Architecture

The reference implementation constructs a domain-scoped Q&A workflow using LangGraph and LangChain. The system only answers machine learning queries; non-ML queries are filtered out early by a classification gate.

5. State Management & Shared Schema in LangGraph

LangGraph models workflows as state machines. A centralized Shared State dictionary flows between all nodes. Each node reads from the state, performs its computation, and writes updates back to the state.

from typing import TypedDict, Optional, Any
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END

# 1. Define the Shared State Schema
class WorkflowState(TypedDict):
    query: str
    model: Optional[ChatOpenAI]
    from_ml_topic: Optional[bool]
    ai_answer: Optional[str]

# 2. Define Structured Output Schema for Classification
class GraderOutput(BaseModel):
    from_machine_learning_topic: bool = Field(
        description="True if the query is related to Machine Learning, False otherwise."
    )

Note: Initializing the LLM instance once and persisting it in the WorkflowState avoids re-instantiating client connections or re-parsing configurations during multi-node traversals.

6. Implementing Nodes and Structured Classifiers

Node 1: Model Initialization (build_model)

Instantiates the foundation model (gpt-4o-mini) and binds it to the execution state.

def build_model(state: WorkflowState) -> dict:
    """Initializes and persists the model in the shared graph state."""
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    return {"model": model}

Node 2: Topic Classification (get_query_topic)

Enforces deterministic schema parsing using Pydantic-based structured outputs to classify whether the incoming query is within the Machine Learning domain.

def get_query_topic(state: WorkflowState) -> dict:
    """Classifies if the query falls within the Machine Learning domain."""
    model = state["model"]
    query = state["query"]

    # Bind the structured output schema to the model
    structured_grader = model.with_structured_output(GraderOutput)

    classifier_prompt = PromptTemplate(
        template="""You are an expert classifier that determines if a user's query 
is related to Machine Learning topics.
Given the user query, return true if it is related to machine learning, otherwise return false.

User Query:
{query}""",
        input_variables=["query"]
    )

    classifier_chain = classifier_prompt | structured_grader
    result: GraderOutput = classifier_chain.invoke({"query": query})

    return {"from_ml_topic": result.from_machine_learning_topic}

Conditional Edge: The Gate Function (grader_node)

A pure Python function that inspects the shared state and determines the next node.

def grader_node(state: WorkflowState) -> str:
    """Evaluates the classification flag and routes execution."""
    if state.get("from_ml_topic") is True:
        return "continue"
    return "exit"

Node 3: Answer Generation (answer_query)

Executes domain-specific answering logic only after the gating condition has been verified.

def answer_query(state: WorkflowState) -> dict:
    """Generates a concise domain-specific response."""
    model = state["model"]
    query = state["query"]

    answer_prompt = PromptTemplate(
        template="""You are an expert in machine learning. Answer the user's query 
concisely in under 200 words.

Query: {query}""",
        input_variables=["query"]
    )

    answer_chain = answer_prompt | model | StrOutputParser()
    result = answer_chain.invoke({"query": query})

    return {"ai_answer": result}

7. Graph Compilation and Assembly

To assemble the workflow, nodes are registered onto a StateGraph, and their execution flow is wired using static and conditional edges.

def build_graph():
    # Initialize StateGraph with the schema
    workflow = StateGraph(WorkflowState)

    # 1. Register Nodes
    workflow.add_node("build_model", build_model)
    workflow.add_node("get_query_topic", get_query_topic)
    workflow.add_node("answer_query", answer_query)

    # 2. Add Standard Edges
    workflow.add_edge(START, "build_model")
    workflow.add_edge("build_model", "get_query_topic")

    # 3. Add Conditional Edge (Gate)
    workflow.add_conditional_edges(
        "get_query_topic",
        grader_node,
        {
            "continue": "answer_query",
            "exit": END
        }
    )

    # 4. Connect terminal node to END
    workflow.add_edge("answer_query", END)

    # Compile into a runnable application
    return workflow.compile()

8. Workflow Execution & Execution Tracing

def execute_prompt_chain_workflow(user_query: str):
    app = build_graph()
    
    initial_state: WorkflowState = {
        "query": user_query,
        "model": None,
        "from_ml_topic": None,
        "ai_answer": None
    }
    
    final_state = app.invoke(initial_state)
    
    # Defensive State Parsing
    if not final_state.get("from_ml_topic"):
        print("Status: Terminated at Gate.")
        print("Reason: The query is not related to Machine Learning topics.")
    else:
        print("Status: Workflow Completed Successfully.")
        print(f"AI Answer:\n{final_state.get('ai_answer')}")
    
    return final_state

Scenario A: In-Domain Execution Path

• Input Query: "What are the latest advancements in machine learning?"

Scenario B: Out-of-Domain Execution Path

• Input Query: "What is the capital of India?"

9. Defensive Programming & Production Guardrails

When engineering multi-node graphs, missing keys and failed state assumptions can cause unexpected runtime crashes.

Defensive Access Best Practices

1. Explicit Guard Verification: Always evaluate gating keys (such as from_ml_topic) before accessing subsequent downstream keys.
2. Use Safe Accessors: Access dictionary properties with .get("ai_answer") rather than raw bracket notation (state["ai_answer"]) to prevent unhandled KeyError exceptions on early workflow exits.
3. Structured Outputs over Regex: Rely on JSON/Pydantic output parsers (with_structured_output) for classification rather than free-form string matching, preventing edge failures caused by conversational model preambles.

10. Architectural Component Comparison

11. Summary and Key Takeaways

• Agentic Workflows vs. Raw Agents: Workflows orchestrate LLM calls and tool executions through deterministic, predictable code paths rather than open-ended autonomous loops.
• Prompt Chaining Mental Model: Decomposes a multi-step objective into a series of smaller, sequential subtasks where each node's output feeds the next node's input.
• Programmatic Gates: Conditional edges that evaluate intermediate outputs to decide whether to advance the workflow or exit early, preventing cascading errors.
• State-Driven Architecture: LangGraph relies on a centralized SharedState schema that keeps track of variables, model instances, and artifacts across nodes.
• Structured Output Guarantees: Using Pydantic models with with_structured_output ensures routing decisions are made on clean, boolean-typed data rather than ambiguous text strings.
• Defensive Consumption: Production systems must account for early exits by checking precondition flags before accessing downstream state variables to avoid runtime KeyError exceptions.