Human-in-the-loop approvals inside a LangGraph + Temporal agent
If you deploy LLM-based agents to execute real-money financial transactions, you quickly learn that pure autonomy is a liability. In our quantitative trading desk, we built an LLM-driven execution agent to parse alternative data sources, draft order parameters, and execute block trades. Letting an LLM directly hit our broker APIs with $100,000 orders is out of the question. We needed a reliable human-in-the-loop (HITL) approval gate.
My team first tried solving this using LangGraph’s native memory checkpointers and thread-interrupt features. It worked beautifully on a local Jupyter notebook. But the moment we pushed it to our Kubernetes cluster, we hit a wall. In production, a human approval can take five minutes or five hours. In that window, pods get evicted, deployments trigger restarts, and stateless HTTP gateways timeout.
This post outlines how we solved this by wrapping a LangGraph state machine inside a Temporal workflow, combining the cognitive flexibility of agentic state graphs with the bulletproof durability of event-driven workflows.
The architectural mismatch: LangGraph vs. Production reality
LangGraph is designed for managing complex, cyclic LLM steps. It structures agent memory, tool calls, and conditional branches using state graphs. To implement approvals, LangGraph offers interrupts—breakpoints that halt execution before or after a specific node runs, saving state to a checkpointer (like PostgreSQL or Redis) and waiting for a client to resume the thread.
Here is where the architecture breaks down in production:
- State serialization overhead: LangGraph checkpointers save the entire state dict, including serialized system messages, tool call histories, and model configurations. If your schema changes during an active deployment, deserializing older, long-pending threads will crash your workers.
- Lack of active orchestration: LangGraph is a library, not an execution engine. It does not handle queueing, retrying failed external API calls, rate limiting broker connections, or managing execution timeouts.
- HTTP connection lifecycle: Holding an open connection from a client UI (like a Slack bot or an internal admin panel) waiting for an agent to finish a 10-minute task is impossible. You need a truly asynchronous, decoupled system where the agent halts, frees up compute resources, and wakes up only when an external event occurs.
The solution: Temporal as the orchestrator, LangGraph as the brain
Instead of using LangGraph to manage the lifecycle of our long-lived approvals, we externalized state tracking to Temporal.
Temporal is a durable execution platform. It guarantees that code runs to completion, regardless of network partitions, hardware crashes, or long-running delays. By putting LangGraph inside a Temporal Workflow, we can run agentic planning steps as discrete Temporal Activities, pause execution indefinitely without consuming CPU cycles, and resume work safely when a human sends an explicit approval Signal.
flowchart LR A["Client Trigger"] --> B["Temporal Workflow"] B --> C["LangGraph Run"] C -->|"Needs Approval"| D["Temporal Signal Wait"] E["Slack UI Button"] -->|"Approve Signal"| D D --> F["Resume LangGraph"] F --> G["Execute Trade"]
The implementation
Let’s build a complete, production-grade implementation of this pattern. We will define:
1. A LangGraph agent that analyzes a trading opportunity and decides if it needs manual sign-off based on a risk threshold.
2. A Temporal Workflow that runs the agent, detects the approval request, yields execution to wait for an external Signal, and completes the trade post-approval.
1. Setting up the LangGraph agent state and graph
First, we define our state and the graph nodes. We will use a simplified model of a trading agent. If the requested trade value exceeds $50,000, our agent sets an approval_required flag in its state and halts.
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
import operator
class TradeState(TypedDict):
ticker: str
action: Literal[“BUY”, “SELL”]
quantity: int
price: float
analysis: str
approval_required: bool
approved: bool
execution_status: str
def analyze_market(state: TradeState) -> dict:
# Simulating LLM analysis step
ticker = state[“ticker”]
action = state[“action”]
total_value = state[“quantity”] * state[“price”]
analysis = f”Analyzing {action} order for {ticker}. Total estimated value: ${total_value:,.2f}.”
approval_required = total_value > 50000.0
return {
“analysis”: analysis,
“approval_required”: approval_required,
“execution_status”: “PENDING_DECISION”
}
def process_execution(state: TradeState) -> dict:
if state[“approval_required”] and not state[“approved”]:
return {“execution_status”: “BLOCKED_BY_RISK”}
# In a real system, this would call the broker API
return {“execution_status”: “EXECUTED”}
# Constructing the LangGraph
builder = StateGraph(TradeState)
builder.add_node(“analyze_market”, analyze_market)
builder.add_node(“process_execution”, process_execution)
# Define routing logic
def route_approval(state: TradeState) -> Literal[“process_execution”, END]:
if state[“approval_required”] and not state[“approved”]:
# Exit the current graph execution early so the orchestrator can take over
return END
return “process_execution”
builder.add_edge(START, “analyze_market”)
builder.add_conditional_edges(
“analyze_market”,
route_approval,
{
“process_execution”: “process_execution”,
END: END
}
)
builder.add_edge(“process_execution”, END)
trading_agent = builder.compile()
2. The Temporal workflow definition
Now we write the Temporal Workflow. In Temporal, the workflow manages the state sequence and guarantees durability, while Activities perform non-deterministic operations like database writes, API calls, or LLM requests.
We define a workflow that runs the LangGraph agent inside an Activity. If the agent returns a state where approval_required is true, the workflow pauses using workflow.wait_condition and waits for an external approve_trade Signal.
from dataclasses import dataclass
@dataclass
class TradeParams:
ticker: str
action: str
quantity: int
price: float
python
# workflows.py
from datetime import timedelta
from temporalio import workflow, activity
from agent_graph import trading_agent, TradeState
from temporal_shared import TradeParams
# Import activities safely at runtime to prevent circular import loops
with workflow.unsafe.imports_passed_through():
from activities import execute_agent_step
@workflow.defn
class TradeExecutionWorkflow:
def __init__(self) -> None:
self._approved: bool = False
self._signal_received: bool = False
@workflow.signal
def approve_trade(self, decision: bool) -> None:
self._approved = decision
self._signal_received = True
@workflow.run
async def run(self, params: TradeParams) -> dict:
# Initialize the LangGraph state dict
current_state: TradeState = {
“ticker”: params.ticker,
“action”: params.action,
“quantity”: params.quantity,
“price”: params.price,
“analysis”: “”,
“approval_required”: False,
“approved”: False,
“execution_status”: “INIT”
}
# Run step 1: LLM planning and risk evaluation
current_state = await workflow.execute_activity(
execute_agent_step,
current_state,
start_to_close_timeout=timedelta(seconds=60),
)
# Check if the agent flagged this trade for human approval
if current_state[“approval_required”] and not current_state[“approved”]:
# Pause execution until the approval signal is received
# Temporal yields all compute resources here; no polling, no active threads.
await workflow.wait_condition(lambda: self._signal_received)
if not self._approved:
current_state[“execution_status”] = “REJECTED_BY_HUMAN”
return current_state
# Feed the approval signal state back into the agent context
current_state[“approved”] = True
# Run step 2: Execute trade in the market
current_state = await workflow.execute_activity(
execute_agent_step,
current_state,
start_to_close_timeout=timedelta(seconds=60),
)
return current_state
3. Implementing the Temporal activities
Temporal Activities execute the non-deterministic components of our code. Here, we run our LangGraph compiled graph with the state dict passed from the workflow.
from temporalio import activity
from agent_graph import trading_agent, TradeState
@activity.defn
async def execute_agent_step(state: TradeState) -> TradeState:
activity.heartbeat(“Executing LangGraph node processing step”)
# Run the graph synchronously. It resumes from whichever state is passed in.
result = await trading_agent.ainvoke(state)
# LangGraph returns the updated state dictionary
return TradeState(
ticker=result[“ticker”],
action=result[“action”],
quantity=result[“quantity”],
price=result[“price”],
analysis=result[“analysis”],
approval_required=result[“approval_required”],
approved=result[“approved”],
execution_status=result[“execution_status”]
)
4. Running the application and handling signals
To run this system, we need a worker process that registers our workflows and activities, and a client trigger script to kick off executions.
Here is the worker implementation:
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from workflows import TradeExecutionWorkflow
from activities import execute_agent_step
async def main():
# Connect to local Temporal server (default port 7233)
client = await Client.connect(“localhost:7233”)
worker = Worker(
client,
task_queue=“trading-task-queue”,
workflows=[TradeExecutionWorkflow],
activities=[execute_agent_step],
)
print(“Worker started. Listening for tasks on ‘trading-task-queue’…”)
await worker.run()
if __name__ == “__main__”:
asyncio.run(main())
And here is the trigger script that simulates starting a high-value trade, pausing for approval, and then sending the resume Signal:
import asyncio
from temporalio.client import Client
from temporal_shared import TradeParams
async def main():
client = await Client.connect(“localhost:7233”)
# Order over $50k ($250 * 300 = $75,000) – will trigger the HITL path
trade = TradeParams(
ticker=“MSFT”,
action=“BUY”,
quantity=300,
price=250.00
)
workflow_id = “trade-workflow-msft-001”
print(f”Starting trade workflow: {workflow_id}…”)
handle = await client.start_workflow(
“TradeExecutionWorkflow”,
trade,
id=workflow_id,
task_queue=“trading-task-queue”,
)
# Allow the workflow to execute its initial analysis step
await asyncio.sleep(3)
# Query workflow description or status to check state
desc = await handle.describe()
print(f”Workflow Status: {desc.status}”)
print(“\n— Human Operator Panel —“)
print(“Simulating review window. Operator receives Slack alert.”)
print(“Executing decision: APPROVED”)
# Send the approval Signal to the running Temporal Workflow
await handle.signal(“approve_trade”, True)
# Wait for the workflow to complete and print final state
result = await handle.result()
print(“\n— Final Workflow State —“)
print(f”Analysis: {result[‘analysis’]}”)
print(f”Approval Required: {result[‘approval_required’]}”)
print(f”Approved: {result[‘approved’]}”)
print(f”Execution Status: {result[‘execution_status’]}”)
if __name__ == “__main__”:
asyncio.run(main())
Running the system and analyzing output
To run this pipeline:
1. Fire up a local Temporal cluster (using the Temporal CLI: temporal server start-dev).
2. Run python worker.py in one terminal window.
3. Run python client.py in another terminal window.
Worker started. Listening for tasks on ‘trading-task-queue’…
When you execute client.py, you will see the workflow kick off, hit the approval constraint, wait, and then resume to completion when you signal it:
Starting trade workflow: trade-workflow-msft-001…
Workflow Status: RUNNING
— Human Operator Panel —
Simulating review window. Operator receives Slack alert.
Executing decision: APPROVED
— Final Workflow State —
Analysis: Analyzing BUY order for MSFT. Total estimated value: $75,000.00.
Approval Required: True
Approved: True
Execution Status: EXECUTED
Critical lessons and pitfalls
Deploying this hybrid pattern taught us several hard truths about distributed agent state.
1. The Pydantic / Serialization Trap
Temporal uses custom JSON payloads to pass parameters between workflows, activities, and signals. LangGraph states frequently contain rich Python types (e.g., AIMessage, HumanMessage, or custom Pydantic schemas).
Never pass LangGraph’s raw memory state objects across Temporal borders. If you try to pass an active LangGraph state dictionary with langchain_core message instances, the default Temporal Python SDK serializer will fail with a deserialization crash or drop critical fields. Keep the state definitions passing through your Temporal activities restricted to primitive dictionaries, strings, numbers, and native Python floats.
2. Handling Signal-Before-Run race conditions
If your frontend application fires the approval signal immediately upon seeing a user interaction, but your Temporal worker is heavily loaded and hasn’t yet entered the workflow.wait_condition logic block, Temporal will still buffer the incoming Signal.
However, you must configure your workflow class constructor to initialize safety flags correctly:
self._approved = False
self._signal_received = False
Because Temporal persists signals to the workflow execution history, if a signal is received before the first activity completes, the _signal_received boolean is flipped. When the workflow execution finally reaches the wait_condition, it resolves instantly without freezing.
3. Debugging non-deterministic errors
Temporal guarantees replay reliability by running the Workflow code again from the beginning when a worker crashes. During replay, any activity execution returns its historical result rather than executing again.
If you instantiate your LangGraph builders or invoke model definitions directly inside the Workflow code (instead of inside a Temporal Activity), your code will trigger a NonDeterminismError. Always keep LangGraph invocations restricted to Activities to isolate non-deterministic LLM behavior and graph compiles from the deterministic replay path of the Temporal orchestrator.