	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > ADK with LangGraph for Stateful Workflows

ADK with LangGraph for Stateful Workflows

Author: Venkata Sudhakar

Complex business workflows require multiple sequential decisions, each dependent on the output of the previous step. LangGraph models these workflows as directed graphs where nodes are processing steps and edges are transitions. By using Gemini-powered ADK agents as LangGraph nodes, you get the best of both worlds: LangGraph's explicit state management and ADK's tool-calling capabilities.

Each node in the graph receives the current workflow state as a TypedDict, processes it using a Gemini call, updates the state, and returns it. LangGraph routes execution to the next node based on conditional edges. The StateGraph ensures that state mutations are tracked and the full execution history is available for debugging and audit.

The below example shows ShopMax India using a LangGraph workflow for order return processing, where each step is handled by a dedicated Gemini agent node.

from typing import TypedDict, Literal
import google.genai as genai
from langgraph.graph import StateGraph, END

client = genai.Client(api_key="your-api-key")
MODEL = "gemini-2.0-flash"

# Shared workflow state
class ReturnState(TypedDict):
    order_id: str
    reason: str
    eligibility: str
    refund_amount_rs: int
    resolution: str

def check_eligibility(state: ReturnState) -> ReturnState:
    # Node 1: determine if the return is within policy
    prompt = (
        "Order ID: " + state["order_id"] + "\n"
        "Return reason: " + state["reason"] + "\n"
        "ShopMax India policy: Electronics returnable within 7 days if defective.\n"
        "Is this return eligible? Reply with only: ELIGIBLE or NOT_ELIGIBLE"
    )
    resp = client.models.generate_content(model=MODEL, contents=prompt)
    verdict = resp.text.strip().upper()
    state["eligibility"] = "eligible" if "ELIGIBLE" in verdict else "not_eligible"
    return state

def calculate_refund(state: ReturnState) -> ReturnState:
    # Node 2: calculate refund amount
    prompt = (
        "Order " + state["order_id"] + " returned as defective.\n"
        "Product: Samsung 55-inch TV, original price Rs 72000.\n"
        "Calculate full refund in Rs. Reply with only the number."
    )
    resp = client.models.generate_content(model=MODEL, contents=prompt)
    try:
        state["refund_amount_rs"] = int(resp.text.strip().replace(",", ""))
    except ValueError:
        state["refund_amount_rs"] = 72000
    return state

def reject_return(state: ReturnState) -> ReturnState:
    # Node 3: handle ineligible returns
    state["resolution"] = "Return rejected: outside 7-day window or reason not covered by policy."
    state["refund_amount_rs"] = 0
    return state

def finalize_return(state: ReturnState) -> ReturnState:
    # Node 4: write final resolution
    amt = state["refund_amount_rs"]
    state["resolution"] = (
        "Return approved for order " + state["order_id"] + ". "
        "Refund of Rs " + str(amt) + " credited in 3-5 business days."
    )
    return state

def route_eligibility(state: ReturnState) -> Literal["calculate_refund", "reject_return"]:
    return "calculate_refund" if state["eligibility"] == "eligible" else "reject_return"

# Build the graph
graph = StateGraph(ReturnState)
graph.add_node("check_eligibility", check_eligibility)
graph.add_node("calculate_refund", calculate_refund)
graph.add_node("reject_return", reject_return)
graph.add_node("finalize_return", finalize_return)

graph.set_entry_point("check_eligibility")
graph.add_conditional_edges("check_eligibility", route_eligibility)
graph.add_edge("calculate_refund", "finalize_return")
graph.add_edge("reject_return", END)
graph.add_edge("finalize_return", END)

app = graph.compile()

initial_state = ReturnState(
    order_id="ORD-20260301-4421",
    reason="TV screen has dead pixels, product is defective",
    eligibility="",
    refund_amount_rs=0,
    resolution=""
)

result = app.invoke(initial_state)
print("Eligibility:", result["eligibility"])
print("Refund: Rs", result["refund_amount_rs"])
print("Resolution:", result["resolution"])

It gives the following output,

Eligibility: eligible
Refund: Rs 72000
Resolution: Return approved for order ORD-20260301-4421. Refund of Rs 72000 credited in 3-5 business days.

LangGraph makes the workflow structure explicit and auditable - you can visualise the graph, inspect state at each node, and add checkpoints for long-running workflows. For ShopMax India in production, use a PostgreSQL checkpoint store via langgraph-checkpoint-postgres to persist workflow state across restarts. This is critical for multi-day approval workflows where a manager must review before the workflow continues. Add ADK FunctionTools inside each node to connect to your order management system, payment gateway, and notification service for end-to-end automation.

Send your comments, suggestions or queries regarding this site to [email protected].