Agentic Workflows & Pipelines

1. Workflow Engines for Agent Orchestration

Agent workflows need more than simple sequential execution. They require conditional branching (take path A if the research is sufficient, path B if more data is needed), loops (keep revising until the reviewer approves), parallel execution (run multiple researchers simultaneously), and error handling (retry on failure, compensate on unrecoverable errors). Two dominant approaches have emerged: LangGraph's built-in state machine model and external workflow engines like Temporal.

2. Conditional Branching and Loops

The most common workflow pattern in agent systems is the conditional loop: execute a step, evaluate the result, and either proceed to the next stage or loop back for refinement. In LangGraph, this is implemented through conditional edges that inspect the current state and return the name of the next node to execute.

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

class WorkflowState(TypedDict):
    messages: Annotated[list, add_messages]
    draft: str
    review_feedback: str
    revision_count: int
    quality_score: float
    approved: bool

def writer_node(state: WorkflowState) -> dict:
    """Generate or revise the draft based on feedback."""
    if state["revision_count"] == 0:
        prompt = "Write an initial draft based on the research."
    else:
        prompt = (
            f"Revise the draft based on this feedback: "
            f"{state['review_feedback']}"
        )
    draft = writer_llm.invoke(prompt)
    return {
        "draft": draft.content,
        "revision_count": state["revision_count"] + 1
    }

def reviewer_node(state: WorkflowState) -> dict:
    """Review the draft and assign a quality score."""
    review = reviewer_llm.invoke(
        f"Review this draft critically. Score 0 to 1.\n\n{state['draft']}"
    )
    score = extract_score(review.content)
    return {
        "review_feedback": review.content,
        "quality_score": score,
        "approved": score >= 0.8
    }

def should_revise(state: WorkflowState) -> Literal["revise", "publish"]:
    """Loop back for revision or proceed to publish."""
    if state["approved"]:
        return "publish"
    if state["revision_count"] >= 3:  # Safety limit
        return "publish"  # Publish best effort
    return "revise"

graph = StateGraph(WorkflowState)
graph.add_node("write", writer_node)
graph.add_node("review", reviewer_node)
graph.add_node("publish", publish_node)

graph.set_entry_point("write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", should_revise, {
    "revise": "write",   # Loop back
    "publish": "publish"  # Proceed
})
graph.add_edge("publish", END)

3. Error Handling, Retries, and Compensation

Production workflows must handle three categories of failures: transient errors (API rate limits, network timeouts) that succeed on retry, permanent errors (invalid input, unsupported operation) that require alternative approaches, and partial failures where some steps succeed before a later step fails. Compensation logic undoes the effects of previously completed steps when a workflow cannot proceed.

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientAgentNode:
    """Wraps an agent node with retry and fallback logic."""

    def __init__(self, primary_llm, fallback_llm, max_retries=3):
        self.primary = primary_llm
        self.fallback = fallback_llm
        self.max_retries = max_retries

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30)
    )
    async def _call_with_retry(self, llm, messages):
        return await llm.ainvoke(messages)

    async def __call__(self, state: dict) -> dict:
        messages = state["messages"]
        try:
            # Try primary model with retries
            result = await self._call_with_retry(self.primary, messages)
        except Exception as e:
            # Fall back to secondary model
            try:
                result = await self._call_with_retry(
                    self.fallback, messages
                )
            except Exception as fallback_err:
                # Both models failed; return error state
                return {
                    "error": f"All models failed: {fallback_err}",
                    "needs_human": True
                }
        return {"messages": [result]}

4. Checkpointing and Streaming

Long-running agent workflows can take minutes or even hours. Checkpointing saves the workflow state after each step so that execution can resume from the last successful point after a crash or restart. Streaming provides real-time visibility into the workflow's progress, letting users see intermediate results and intervene if needed.

from langgraph.checkpoint.postgres import PostgresSaver

# Production-grade checkpointing with PostgreSQL
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agents"
)
app = graph.compile(checkpointer=checkpointer)

# Stream execution to see real-time progress
config = {"configurable": {"thread_id": "research-task-101"}}

async for event in app.astream_events(
    {"messages": [("user", "Research and write about fusion energy")]},
    config=config,
    version="v2"
):
    if event["event"] == "on_chain_start":
        print(f"Starting: {event['name']}")
    elif event["event"] == "on_chain_end":
        print(f"Completed: {event['name']}")
    elif event["event"] == "on_chat_model_stream":
        # Stream individual tokens for real-time display
        print(event["data"]["chunk"].content, end="")

5. Human-in-the-Loop Gates

For high-stakes workflows, human review gates pause execution at critical decision points and wait for human approval before proceeding. LangGraph supports this natively through its interrupt_before parameter, which halts execution before specified nodes and waits for explicit resumption.

# Compile with interrupt points for human review
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["publish"]  # Pause before publishing
)

# Run until the interrupt point
config = {"configurable": {"thread_id": "article-draft-42"}}
result = app.invoke(initial_state, config)

# At this point, execution is paused before "publish"
# A human reviews the draft (could be hours later)
current_state = app.get_state(config)
draft = current_state.values["draft"]
print("Draft for review:", draft[:500])

# Human approves: resume execution
app.update_state(config, {"human_approved": True})
final_result = app.invoke(None, config)  # Resume from checkpoint

# Human rejects: update with feedback and re-run from writer
# app.update_state(config, {"review_feedback": "Needs more data"})
# final_result = app.invoke(None, config)

6. Lab: Multi-Agent Research System

This lab builds a complete multi-agent research pipeline with four stages: a Planner that decomposes the research topic into subtasks, a Researcher that gathers information for each subtask, a Writer that synthesizes the research into a coherent document, and a Reviewer that provides quality feedback. The system includes a revision loop between the Writer and Reviewer, a human approval gate before final output, and checkpoint-based recovery.

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

class ResearchPipelineState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str
    subtasks: list[str]
    research_results: list[dict]
    draft: str
    review_feedback: str
    quality_score: float
    revision_count: int
    human_approved: bool

def planner(state: ResearchPipelineState) -> dict:
    """Decompose the topic into 3-5 research subtasks."""
    plan = planner_llm.invoke(
        f"Break this topic into 3-5 specific research subtasks: "
        f"{state['topic']}"
    )
    subtasks = parse_subtasks(plan.content)
    return {"subtasks": subtasks}

async def parallel_research(state: ResearchPipelineState) -> dict:
    """Research all subtasks in parallel."""
    tasks = [
        research_agent.ainvoke(subtask)
        for subtask in state["subtasks"]
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Filter out failed tasks, keep successful results
    valid = [r for r in results if not isinstance(r, Exception)]
    return {"research_results": valid}

# Assemble the full pipeline
graph = StateGraph(ResearchPipelineState)
graph.add_node("plan", planner)
graph.add_node("research", parallel_research)
graph.add_node("write", writer_node)
graph.add_node("review", reviewer_node)
graph.add_node("publish", publish_node)

graph.set_entry_point("plan")
graph.add_edge("plan", "research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", should_revise, {
    "revise": "write",
    "publish": "publish"
})
graph.add_edge("publish", END)

# Compile with checkpointing and human-in-the-loop
checkpointer = PostgresSaver.from_conn_string(DB_URL)
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["publish"]
)