langgraph vs temporal workflow: which orchestration framework for AI agents?

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import os

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

tools = [add, multiply]
model = ChatOpenAI(model="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"])
model_with_tools = model.bind_tools(tools)  # Native tool-calling binding

from typing import TypedDict, Annotated
from operator import add as op_add

class AgentState(TypedDict):
    messages: Annotated[list, op_add]

def agent_node(state: AgentState):
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def process_tool_calls(state: AgentState):
    # langgraph's native tool-calling loop handles this automatically
    return state

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

# langgraph-specific: checkpoint to memory for state persistence
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# Invoke with thread_id for deterministic replay
result = graph.invoke(
    {"messages": [{"role": "user", "content": "What is 5 + 3 times 2?"}]},
    config={"configurable": {"thread_id": "user-123"}}  # State is saved and resumable
)
print(result)

from temporalio import activity, workflow, Client, WorkflowHandle
from temporalio.client import Client
from temporalio.worker import Worker
import asyncio
import os

@activity.defn
async def add(a: int, b: int) -> int:
    """Add two numbers (activity = durable task)."""
    return a + b

@activity.defn
async def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

@workflow.defn
class MathWorkflow:
    @workflow.run
    async def run(self, question: str) -> int:
        # temporal guarantees this workflow is durable and deterministic
        result1 = await workflow.execute_activity(
            add,
            5, 3,
            start_to_close_timeout=300  # Automatic retry on timeout
        )
        # Every step is logged to event history; can replay deterministically
        result2 = await workflow.execute_activity(
            multiply,
            result1, 2,
            start_to_close_timeout=300
        )
        return result2

async def main():
    client = await Client.connect("localhost:7233")  # Temporal server (mandatory)
    handle: WorkflowHandle = await client.start_workflow(
        MathWorkflow.run,
        "What is 5 + 3 times 2?",
        id="math-workflow-1"
    )
    result = await handle.result()  # Blocks until workflow completes or fails
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())