Workflows

To build production-grade agentic systems, you need a precise orchestration layer. This section provides the exact steps and code patterns required to deploy workflows using Python (LangGraph) and TypeScript (Vercel AI SDK).

Python: Implementation with LangGraph

LangGraph is ideal for complex, stateful agents that require fine-grained control over loops and human intervention.

Mental Mode

LangGraph is the industry standard for building stateful, multi-agent applications in Python. It models workflows as a State Machine.

• State: A shared object that is passed between nodes (usually a TypedDict or a Pydantic model)
• Nodes: Python function that take the state as input and perform work (through an LLM or API call for example) and return an updated state
• Edges: Define the transition from one node to another.
• Conditional Edges: The routing logic that looks at state and decides the next path to take.

To handle state and the stochastic behaviour of LLMs, the following come in handy:

1. Retries: Define a RetryPolicy at the node level to automatically retry with exponential backoff (in the case the output in malformed or the function fails)
2. Validation: Use Pydantic to validate the state at every transition. This ensures that even if an LLM "hallucinates" a field, the workflow crashes early and safely rather than propagate bad data.
3. Human-in-the-loop: You can configure the graph to interrupt_before a specific node (e.g., publish_to_social_media). The state is stored, allowing the human to review it before the workflow resumes.

Setup Steps

1. Environment: Create a .env file and add OPENAI_API_KEY=your_key_here.
2. Installation: pip install langgraph langchain-openai python-dotenv.
3. Define State: Use TypedDict to define the “schema” of your agent’s memory.
4. Initialize Model: Use ChatOpenAI and bind any necessary tools.
5. Define Nodes: Create Python functions that take state and return an updated state.
6. Construct Graph: Use StateGraph, add nodes, and define edges.
7. Compile & Run: Compile the graph (with checkpointers for persistence) and invoke it.

Code Example: Customer Support Router with Human Review

import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

# 1. Define the State
class AgentState(TypedDict):
    query: str
    classification: str
    response: str
    needs_review: bool

# 2. Initialize the Model
model = ChatOpenAI(model="gpt-4o", temperature=0)

# 3. Define Nodes
def classifier_node(state: AgentState):
    # LLM decides if this is a billing or tech issue
    prompt = f"Classify this query:{state['query']}. Options: billing, tech."
    res = model.predict(prompt)
    return {"classification": res.strip().lower()}

def billing_node(state: AgentState):
    return {"response": "Handling billing inquiry...", "needs_review": True}

def tech_node(state: AgentState):
    return {"response": "Handling tech support...", "needs_review": False}

# 4. Routing Logic
def route_query(state: AgentState) -> Literal["billing", "tech"]:
    return state["classification"]

# 5. Build the Graph
workflow = StateGraph(AgentState)

workflow.add_node("classifier", classifier_node)
workflow.add_node("billing", billing_node)
workflow.add_node("tech", tech_node)

workflow.set_entry_point("classifier")
workflow.add_conditional_edges("classifier", route_query)
workflow.add_edge("billing", END)
workflow.add_edge("tech", END)

# 6. Persistence & Human Review Gate
memory = MemorySaver()
# We interrupt execution BEFORE 'billing' to allow human review
app = workflow.compile(checkpointer=memory, interrupt_before=["billing"])

# 7. Execution
config = {"configurable": {"thread_id": "user_1"}}
initial_state = {"query": "I was overcharged on my last invoice."}
app.invoke(initial_state, config)

# Execution pauses here if classification is 'billing'
# Human can resume by calling app.invoke(None, config)

TypeScript: Implementation with Vercel AI SDK

Vercel AI SDK Workflows are built for durability and are excellent for web applications where long-running tasks need to survive serverless timeouts.

The key thing to remember is that workflows in the AI SDK are async functions where each discrete step is wrapped in a step() function. This provides caching and ensures that if the server crashes or times out, the workflow can resume from the last successful step without re-running expensive LLM calls.

To handle stochastic behaviour, we can use Zod for validation. Use generateObject within a step to ensure the LLM output matches your exact schema.

	...
	schema: z.object({ summary: z.string(), tags: z.array(z.string()) }),
	...

Here too, we can build in Human-in-the-loop. For tasks requiring approval, you can use the "Wait for Event" pattern. The workflow suspends its execution and waits for an external webhook or a manual "resume" signal from the UI.

Setup Steps

1. Environment: Add OPENAI_API_KEY to your .env.local.
2. Installation: npm install ai @ai-sdk/openai @vercel/workflow zod.
3. Initialize Provider: Create an openai instance.
4. Define Workflow: Use the workflow() wrapper.
5. Use Steps: Wrap every async call in step() to ensure durability and caching.
6. Validate Output: Use generateObject with a zod schema inside a step.
7. Handle Branching: Use standard TS if/else logic between steps.

Code Example: Validated Data Analysis Pipeline

import { workflow, step } from '@vercel/workflow';
import { openai } from '@ai-sdk/openai';
import { generateObject } from 'ai';
import { z } from 'zod';

export const analysisWorkflow = workflow(async (input: { data: string }) => {
  // 1. Validated Extraction Step
  const extraction = await step('extract-entities', async () => {
    return await generateObject({
      model: openai('gpt-4o'),
      schema: z.object({
        sentiment: z.enum(['positive', 'negative', 'neutral']),
        urgency: z.number().min(1).max(5),
        entities: z.array(z.string()),
      }),
      prompt: `Analyze this data:${input.data}`,
    });
  });

  // 2. Branching Logic based on LLM output
  if (extraction.object.urgency > 4) {
    await step('escalate-issue', async () => {
      // Logic to send Slack notification or PagerDuty
      return { status: 'escalated' };
    });
  }

  // 3. Transformation Step
  const summary = await step('generate-summary', async () => {
    // This step is durable; if it fails, it will retry based on config
    const res = await openai('gpt-4o').generateText({
      prompt: `Summarize these entities:${extraction.object.entities.join(', ')}`,
    });
    return res.text;
  });

  return { summary, status: 'complete' };
});

Designing for Finality: The Developer’s Checklist

When moving from a script to a production workflow, verify these three points:

1. API Key Management: Always use process.env or os.getenv. Never hardcode keys. For multi-tenant apps, pass keys through a secure vault at runtime.
2. The “Dry Run” Node: Before calling a destructive tool (e.g., delete_user), add a node that generates a “Plan of Action” and waits for a boolean approved flag in the state.
3. Traceability: Wrap your graph/workflow in an observability provider (LangSmith for Python, Langfuse for TS). This allows you to debug the exact JSON that was passed between nodes when a failure occurs.