Part 2: Durable Agent Execution with LangGraph + Temporal

Part 2 of 5 in the series “Building Production-Grade AI Agents: From Prototype to Scale”

In Part 1, we identified the core problem: AI agents fail mid-execution and lose all progress. In this part, we solve it by combining LangGraph’s graph-based agent logic with Temporal’s durable execution guarantees.

The result: agents that finish what they start, no matter what goes wrong.

The Core Insight

LangGraph defines what agents think. Temporal guarantees that they finish thinking.

LangGraph = a directed graph of reasoning steps
Temporal  = a system that ensures every step completes (or retries until it does)

The integration pattern is simple: each LangGraph graph runs inside a Temporal activity. Temporal handles retries, timeouts, and crash recovery. LangGraph handles the agent logic.

LangGraph: Graph-Based Agent Logic

If you’re coming from simple LLM chains, LangGraph’s key advantage is explicit control flow. You define nodes (what the agent does at each step) and edges (how it decides what to do next).

from langgraph.graph import StateGraph, END
from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    retrieved_docs: list[dict]
    analysis: str
    quality_score: float
    iteration: int

async def retrieve_node(state: ResearchState) -> ResearchState:
    """Search for relevant documents."""
    docs = await retriever.search(state["query"], top_k=10)
    return {**state, "retrieved_docs": docs}

async def analyze_node(state: ResearchState) -> ResearchState:
    """Analyze retrieved documents with LLM."""
    prompt = f"Analyze these documents for: {state['query']}\n\n"
    prompt += "\n---\n".join(d["content"] for d in state["retrieved_docs"])

    analysis = await llm.ainvoke(prompt)
    return {**state, "analysis": analysis}

async def quality_check_node(state: ResearchState) -> ResearchState:
    """Score the analysis quality."""
    score = await evaluate_quality(state["analysis"], state["query"])
    return {**state, "quality_score": score}

async def refine_query_node(state: ResearchState) -> ResearchState:
    """Generate a better query based on what we found."""
    refined = await llm.ainvoke(
        f"The query '{state['query']}' got low-quality results. "
        f"Generate a better search query."
    )
    return {
        **state, 
        "query": refined, 
        "iteration": state["iteration"] + 1,
    }

def should_continue(state: ResearchState) -> str:
    """Conditional edge: continue or stop."""
    if state["quality_score"] >= 0.8:
        return "done"
    if state["iteration"] >= 3:
        return "done"  # max retries
    return "retry"

def build_research_graph():
    graph = StateGraph(ResearchState)
    graph.add_node("retrieve", retrieve_node)
    graph.add_node("analyze", analyze_node)
    graph.add_node("quality_check", quality_check_node)
    graph.add_node("refine_query", refine_query_node)
    graph.set_entry_point("retrieve")
    graph.add_edge("retrieve", "analyze")
    graph.add_edge("analyze", "quality_check")
    graph.add_conditional_edges(
p        "quality_check",
        should_continue,
        {"done": END, "retry": "refine_query"},
    )
    graph.add_edge("refine_query", "retrieve")
    return graph.compile()

This gives you:

• Loops: Quality too low? Refine and retry.
• Conditionals: Branch based on state, not just the last LLM output.
• Explicit state: Every node receives and returns typed state — no hidden globals.

But it doesn’t give you durability. If the process dies between “analyze” and “quality_check,” all work is lost.

Temporal: Making It Durable

Temporal is a workflow engine. It records every step as an event. If a worker dies, another worker picks up the workflow and replays it from the last completed step.

Wrapping LangGraph as a Temporal Activity

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

@activity.defn
async def run_research_agent(query: str) -> dict:
    """
    A complete LangGraph agent, wrapped as one Temporal activity.

    If this fails (LLM timeout, OOM, server crash):
    - Temporal retries automatically
    - Up to 3 attempts with exponential backoff
    - Heartbeat ensures Temporal knows we're alive
    """
    graph = build_research_graph()

    initial_state = {
        "query": query,
        "retrieved_docs": [],
        "analysis": "",
        "quality_score": 0.0,
        "iteration": 0,
    }

    result = await graph.ainvoke(initial_state)

    # Heartbeat: tell Temporal we're still working
    activity.heartbeat(f"completed with score={result['quality_score']}")

    return {
        "analysis": result["analysis"],
        "quality_score": result["quality_score"],
        "iterations": result["iteration"],
    }

@activity.defn
async def run_synthesis_agent(findings: list[dict]) -> str:
    """Synthesize multiple research findings into a report."""
    graph = build_synthesis_graph()
    result = await graph.ainvoke({"findings": findings})
    return result["report"]

The Temporal Workflow

@workflow.defn
class ResearchPipeline:
    """
    Durable research pipeline:
    1. Research (with retries)
    2. Human approval (waits indefinitely, survives restarts)
    3. Synthesis (with retries)
    """

def __init__(self):
        self._approved = False
        self._status = "starting"
    @workflow.run
    async def run(self, topic: str) -> dict:
        self._status = "researching"
        # Step 1: Research (retries on failure)
        findings = await workflow.execute_activity(
            run_research_agent,
            topic,
            start_to_close_timeout=timedelta(minutes=5),
            heartbeat_timeout=timedelta(seconds=60),
            retry_policy=RetryPolicy(
                maximum_attempts=3,
                backoff_coefficient=2.0,
                maximum_interval=timedelta(seconds=30),
            ),
        )
        self._status = "awaiting_approval"
        # Step 2: Wait for human (durable - survives crashes/deploys)
        await workflow.wait_condition(lambda: self._approved)
        self._status = "synthesizing"
        # Step 3: Synthesize
        report = await workflow.execute_activity(
            run_synthesis_agent,
            [findings],
            start_to_close_timeout=timedelta(minutes=3),
            retry_policy=RetryPolicy(maximum_attempts=2),
        )
        self._status = "complete"
        return {"report": report, "findings": findings}
    @workflow.signal
    async def approve(self):
        """External signal: human approved the findings."""
        self._approved = True
    @workflow.query
    def get_status(self) -> str:
        """Query current status without affecting workflow."""
        return self._status

What Temporal Gives You: Crash Recovery

Here’s the sequence when things go wrong:

Crash Recovery using Temporal

Key point: In this basic version, the retry re-runs the entire LangGraph from scratch. In Part 3, we’ll add Redis checkpointing so retries resume from the last successful node — not from the beginning.

What Temporal Gives You: Human-in-the-Loop

This is where Temporal truly shines. A workflow can wait indefinitely for a human signal — and it costs nothing. No blocked threads, no polling, no state to manage.

Human in the loop (Agentic Ai)

What happens during those 3 days:

• The workflow exists only as events in Temporal’s database
• No threads blocked, no memory consumed
• Servers can restart, redeploy — doesn’t matter
• When the signal arrives, a worker picks up the workflow and replays it

What Temporal Gives You: Multi-Agent Fan-Out

Run multiple LangGraph agents in parallel as child workflows or concurrent activities:

@workflow.defn
class MultiAgentResearch:
    @workflow.run
    async def run(self, topic: str) -> dict:
        # Fan-out: research + critique in parallel
        research_task = workflow.execute_activity(
            run_research_agent, topic,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )
        critic_task = workflow.execute_activity(
            run_critic_agent, topic,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )
        # Fan-in: wait for both
        research, critique = await asyncio.gather(
            research_task, critic_task
        )
        # Synthesize using both perspectives
        report = await workflow.execute_activity(
            run_synthesis_agent,
            {"research": research, "critique": critique},
            start_to_close_timeout=timedelta(minutes=3),
        )
        return report

Running the Temporal Worker

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

async def main():
    # Connect to Temporal server
    client = await Client.connect("localhost:7233")
    # Start worker (processes activities + workflows)
    worker = Worker(
        client,
        task_queue="agent-tasks",
        workflows=[ResearchPipeline, MultiAgentResearch],
        activities=[run_research_agent, run_synthesis_agent, run_critic_agent],
    )
    await worker.run()

# Start a workflow from your API
async def start_research(topic: str) -> str:
    client = await Client.connect("localhost:7233")
    handle = await client.start_workflow(
        ResearchPipeline.run,
        topic,
        id=f"research-{uuid4()}",
        task_queue="agent-tasks",
    )
    return handle.id  # Return workflow ID for status checks

# Send approval signal
async def approve_research(workflow_id: str):
    client = await Client.connect("localhost:7233")
    handle = client.get_workflow_handle(workflow_id)
    await handle.signal(ResearchPipeline.approve)

# Query status
async def get_status(workflow_id: str) -> str:
    client = await Client.connect("localhost:7233")
    handle = client.get_workflow_handle(workflow_id)
    return await handle.query(ResearchPipeline.get_status)

Practical Tips

1. Set Heartbeat Timeouts

If your LangGraph agent takes >30 seconds (it will), use heartbeats so Temporal doesn’t kill it prematurely:

@activity.defn
async def run_long_agent(query: str) -> dict:
    graph = build_graph()

    # Heartbeat during long operations
    for i, step in enumerate(graph.stream({"query": query})):
        activity.heartbeat(f"step {i}: {step.get('node', 'unknown')}")

    return step

2. Make Activities Idempotent

Temporal may run an activity twice (crash after completion but before acknowledgment). Design for it:

@activity.defn
async def run_research_agent(query: str, idempotency_key: str) -> dict:
    # Check if we already have results for this key
    cached = await redis.get(f"result:{idempotency_key}")
    if cached:
        return json.loads(cached)

    # Run the agent
    result = await graph.ainvoke({"query": query})

    # Cache result (idempotent on retry)
    await redis.setex(f"result:{idempotency_key}", 3600, json.dumps(result))
    return result

3. Separate Task Queues by Cost

Put expensive GPU-heavy activities on dedicated workers:

# Cheap activities (orchestration, caching)
worker_light = Worker(client, task_queue="agent-light", ...)

# Expensive activities (LLM calls, embedding)
worker_heavy = Worker(client, task_queue="agent-heavy", ...)

4. Use Workflow Timeouts as Cost Guards

Prevent runaway agents from burning your API budget:

handle = await client.start_workflow(
    ResearchPipeline.run,
    topic,
    id=f"research-{uuid4()}",
    task_queue="agent-tasks",
    execution_timeout=timedelta(minutes=30),  # Hard cap
)

What’s Missing

This setup re-runs the entire LangGraph from scratch on retry. If your graph has 8 nodes and it fails at node 7, you waste the work from nodes 1–6.

In the next part, we’ll add Redis checkpointing — so retries resume from the last successful node, not from the beginning. Combined with Postgres for durable history, you get a complete two-tier memory system.

Next: [Part 3 — Agent Memory: The Redis + Postgres Pattern → ](https://medium.com/p/d79fd779acc5) Previous: ← Part 1 — Why Your AI Agent Architecture Won’t Scale