Agentic pipelines

TL;DR

• "Call the LLM in a loop" is not a production architecture. You need retry logic, observability, state persistence, cost budgets, and audit trails baked into the infrastructure layer.
• Three tool categories cover the space: LangGraph (complex stateful Python-first agent logic with conditional edges), n8n (low-code business process automation with AI nodes), Temporal (durable execution that survives crashes, deploys, and multi-day runs).
• The deterministic plan pattern is the 2025-2026 industry consensus for reliable production agents. The LLM generates a step-by-step plan once; deterministic code executes it. No LLM calls during execution reduces hallucination compounding.
• A DAG (directed acyclic graph) is easier to debug, retry, and reason about than a looping graph. Add cyclic branching only when you have proven you need it.
• Every agentic pipeline needs the same baseline: input validation, step-level logging, retry with backoff, cost budget, timeout per step, output validation. These are non-negotiable in production.

The problem it solves

Your agent works in your terminal. It calls the LLM, uses a tool, calls the LLM again, and produces a result. You're excited.

You run it in production. At step 7 of 12, a transient network error kills the container. The $200 of compute is gone and you restart from step 1. With 50 concurrent runs per day, that isn't a bug you can ignore.

Without pipeline infrastructure, you have no resume capability, no cost tracking, no visibility into which steps fail most often, and no way to catch the runs that silently produce wrong output. That's the problem agentic pipeline frameworks solve.

What is it?

An agentic pipeline is the infrastructure layer that wraps AI steps with the production requirements that any serious software system needs: state management, error handling, observability, retries, and audit trails.

Three tools dominate this space. Each occupies a different position on the trade-off between LLM-native workflow complexity and operational complexity.

How it works

LangGraph: Python-first stateful graph

LangGraph models your workflow as a directed graph where each node is a Python function and edges define control flow. The key features are add_conditional_edges (route to different nodes based on runtime state) and a built-in checkpointer that serializes graph state to a database after each step. I've found this checkpointer invaluable in production: when a long-running workflow crashes, LangGraph reloads from the last checkpoint and continues rather than starting over.

Best for: complex stateful agent logic with multi-level conditional branching, human-in-the-loop interrupt gates, and multi-agent coordination. Used in production by Klarna, LinkedIn, and Elastic. Choose LangGraph when your workflow logic is conditional, Python-first, and complex enough to justify the learning curve.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    step_count: int
    result: str | None

def call_llm(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response], "step_count": state["step_count"] + 1}

def call_tool(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    result = execute_tool(last_message.tool_calls[0])
    return {"messages": [result]}

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tool"
    return END

graph = StateGraph(AgentState)
graph.add_node("call_llm", call_llm)
graph.add_node("call_tool", call_tool)
graph.set_entry_point("call_llm")
graph.add_conditional_edges("call_llm", should_continue)
graph.add_edge("call_tool", "call_llm")
app = graph.compile(checkpointer=memory_saver)

n8n: Low-code with AI nodes

n8n is a visual workflow automation platform with 400+ pre-built integrations and built-in AI nodes (call OpenAI, classify with Claude, extract with Gemini). The visual editor means non-engineers can author, maintain, and debug workflows without writing code. It's self-hostable and production-grade for workflows that fit its model.

Best for: business process automation where AI is one step in a larger integration workflow. Not suited for complex conditional agent logic or deeply recursive graphs. Choose n8n when the workflow is integration-focused, the logic is linear or lightly branched, and the team includes non-engineers who will maintain it long-term.

Temporal: Durable execution that survives crashes

Temporal is not AI-native. It's a durable execution engine where your workflow code is ordinary Python (or Go, Java, TypeScript). Temporal wraps every activity (LLM call, API call, database write) with exactly-once semantics and automatic retry. If your workflow server crashes at step 15 of 20, Temporal replays all activities up to step 15 from its event log and resumes from step 16.

Best for: multi-day workflows, exactly-once guarantees, and coordination across multiple services. The setup cost is high (you need to run the Temporal server), but the reliability guarantees are unmatched. Choose Temporal when workflows must survive crashes, deploys, and days-long execution windows.

import asyncio
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.common import RetryPolicy

@activity.defn
async def summarize_document(doc_url: str) -> str:
    """Non-deterministic activity: fetch and LLM-summarize a document."""
    content = await fetch_url(doc_url)
    return await llm_client.summarize(content)

@workflow.defn
class DocumentAnalysisWorkflow:
    @workflow.run
    async def run(self, doc_urls: list[str]) -> dict:
        results = []
        for url in doc_urls:
            # Each activity auto-retries; state persists across crashes
            summary = await workflow.execute_activity(
                summarize_document,
                url,
                schedule_to_close_timeout=timedelta(minutes=5),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            results.append({"url": url, "summary": summary})
        return {"summaries": results, "total": len(results)}

The sequence below shows how Temporal handles a mid-workflow crash without data loss or double-execution.