08. Agentic Workflows & State Machines (LangGraph)

Android Architecture Mental Model: Linear AI pipelines resemble classic imperative procedures: Step A → Step B → Step C. Real-world tasks fail when Step B fails or requires iterative refinement. Agentic workflows (like LangGraph) are identical to an MVI (Model-View-Intent) unidirectional state machine driven by a Reducer. State is immutable. Nodes act as pure transformer functions (or side-effect dispatchers). Edges evaluate conditions to determine whether to transition to the next state, loop back to self-correct, or pause execution waiting for external user input.

1. The ReAct (Reason + Act) Graph Cycle

The core engine of an autonomous agent is the cyclic loop: evaluating the current state, deciding whether to invoke tools, mutating the state with tool outputs, and looping until completion conditions are met.

flowchart TD START([Graph Entry]) --> AGENT[Agent Reasoning Node: LLM Evaluation] AGENT --> COND{Conditional Edge:
Tool Required?} COND -- Yes: Invoke Tool --> TOOLS[Tool Execution Node] TOOLS -->|Reducer: Append Tool Message| AGENT COND -- No: Task Completed --> END_NODE([Graph Exit: Return State])

2. Core Components of LangGraph

Component	System Role	Android / MVI Analogy
State Schema	A typed dictionary/dataclass representing the single source of truth across the graph lifecycle.	`UiState` / `ViewState` immutable data class.
Nodes	Python functions or Kotlin suspend routines that receive the current state and return partial state updates.	State Reducers / Interactors / Use Cases.
Edges & Conditional Edges	Routing functions that inspect state keys to determine the next destination node.	Navigation Graphs or sealed class branch matching (`when(event)`).
Checkpointers	Persistence engines (Postgres, Redis, SQLite) that snapshot graph state at every step for replay and recovery.	`SavedStateHandle` or Room DB transaction logs.

3. Human-in-the-Loop (HITL) Interruption Pattern

High-stakes operations (such as issuing financial refunds, deleting database tables, or sending emails) should not run autonomously. LangGraph introduces interruption boundaries that suspend graph execution, yield control to the caller, and resume when external authorization is received.

sequenceDiagram autonumber participant Engine as LangGraph Runner participant DB as Postgres Checkpointer participant User as Admin Reviewer (Android UI) Engine->>Engine: Node: Propose Refund Transaction ($500) Engine->>DB: Checkpoint State (Status: PENDING_APPROVAL) Engine-->>User: Interrupt! Yield state to UI for review Note over User: Admin reviews details in UI
and taps 'Approve' User->>Engine: Resume Graph with Approval Payload Engine->>DB: Hydrate State from Checkpoint Engine->>Engine: Node: Execute Financial Transfer API Engine-->>User: Emitted State: Transaction Complete

4. Dual-Stack Implementations: State Machine Agent Loop

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# 1. Define State with Reducer (add_messages appends instead of overwriting)
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iteration_count: int

# 2. Define Worker Nodes
def reasoner_node(state: AgentState) -> dict:
    msgs = state["messages"]
    count = state.get("iteration_count", 0) + 1
    # Simulated model decision
    if count < 2:
        return {"messages": [AIMessage(content="CALL_TOOL: verify_inventory")], "iteration_count": count}
    return {"messages": [AIMessage(content="Task complete: All items verified.")], "iteration_count": count}

def action_node(state: AgentState) -> dict:
    return {"messages": [HumanMessage(content="TOOL_OUTPUT: Inventory count is 42 units.")]}

# 3. Define Conditional Routing
def route_next(state: AgentState) -> str:
    last_msg = state["messages"][-1].content
    if "CALL_TOOL" in last_msg:
        return "action_node"
    return END

# 4. Build and Compile Graph
workflow = StateGraph(AgentState)
workflow.add_node("reasoner_node", reasoner_node)
workflow.add_node("action_node", action_node)

workflow.add_edge(START, "reasoner_node")
workflow.add_conditional_edges("reasoner_node", route_next)
workflow.add_edge("action_node", "reasoner_node")

app = workflow.compile()

# Execute graph
initial_state = {"messages": [HumanMessage(content="Verify stock for SKU-901")], "iteration_count": 0}
result = app.invoke(initial_state)

for m in result["messages"]:
    print(f"[{m.type}]: {m.content}")

import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update

// 1. Immutable State Contract
data class AgentState(
    val messages: List<String> = emptyList(),
    val iterationCount: Int = 0,
    val isComplete: Boolean = false
)

// 2. Sealed Intents / Node Transitions
sealed interface AgentIntent {
    data class Reason(val prompt: String) : AgentIntent
    data class ExecuteTool(val toolName: String) : AgentIntent
}

class AgentStateMachine {
    private val _state = MutableStateFlow(AgentState())
    val state = _state.asStateFlow()

    // 3. Reducer Cycle
    suspend fun process(intent: AgentIntent) {
        when (intent) {
            is AgentIntent.Reason -> {
                val currentCount = _state.value.iterationCount + 1
                if (currentCount < 2) {
                    _state.update {
                        it.copy(
                            messages = it.messages + "CALL_TOOL: verify_inventory",
                            iterationCount = currentCount
                        )
                    }
                    // Conditional Edge: Transition to Tool
                    process(AgentIntent.ExecuteTool("verify_inventory"))
                } else {
                    _state.update {
                        it.copy(
                            messages = it.messages + "Task complete: All items verified.",
                            iterationCount = currentCount,
                            isComplete = true
                        )
                    }
                }
            }
            is AgentIntent.ExecuteTool -> {
                // Execute Local Routine and route back to Reasoner
                val toolResult = "TOOL_OUTPUT: Inventory count is 42 units."
                _state.update { it.copy(messages = it.messages + toolResult) }
                process(AgentIntent.Reason(toolResult))
            }
        }
    }
}

5. Progressive Glossary

Term	Technical Definition	Android / Systems Analogy
State Reducer	A pure function that takes the current state and a partial update to calculate the next immutable state.	MVI Reducer pattern or Redux `reduce(state, action)`.
Cyclic Graph	A directed execution flow that allows loops and backtracking rather than strictly moving forward (DAG).	A finite state machine handling screen state retries and pagination loops.
Checkpointing	Serializing the entire execution snapshot to persistent storage at graph node transitions.	Android `onSaveInstanceState()` / `SavedStateHandle` snapshotting.
HITL (Human-in-the-Loop)	Suspending graph execution at critical nodes until an external human approval payload is received.	Presenting an Android Runtime Permission Dialog or Biometric prompt before continuing.

Sources & Reference Standards

LangChain / LangGraph Architecture Documentation: StateGraph and Multi-Agent Orchestration
Yao et al.: ReAct: Synergizing Reasoning and Acting in Language Models (ICLR)
Google Research: Building Stateful Multi-Agent Systems on Cloud Run

← Module 07: Tool Calling Proceed to Module 09: FastAPI & Cloud Run →