🔄 LangGraph ReAct Agent#

LangGraph ReAct Agent sits right beside the Strands Hello World demo to prove a key architectural insight: the ReAct (Reason + Act) agent pattern is universal across AI frameworks. Here, the same Amazon Bedrock model (amazon.nova-lite-v1:0) is driven by LangGraph's create_react_agent() instead of AWS Strands.

Equipped with a simple @tool def greet(name: str) -> str tool, the agent is asked to "Greet Alice and Bob". It reasons that two greetings are needed, invokes greet("Alice") and greet("Bob"), observes the friendly outputs, and synthesizes a final response while the service extracts all intermediate tool messages from the graph state for display in the UI.

1. The ReAct Loop (Reason → Act → Observe)

Whether implemented in LangGraph, AWS Strands, or raw OpenAI code, a ReAct agent executes the same cyclic state graph:

1. Reason (Agent Node): The LLM inspects the conversation state and decides whether it needs external tool actions or can answer immediately.
2. Act (Tools Node): If the LLM emits tool_calls, the graph routes to the tool executor node, which runs greet(name) for each call.
3. Observe (State Edge): Each tool output is appended to the state's messages list as a ToolMessage, and control loops back to the Agent Node.

2. Side-by-Side Framework Comparison: Strands vs. LangGraph

• Model Setup: Strands uses BedrockModel(model_id=...); LangChain/LangGraph uses init_chat_model(model_id, model_provider="bedrock_converse").
• Tool Decorator: Strands uses from strands import tool; LangGraph uses from langchain_core.tools import tool.
• Invocation: Strands calls agent(prompt); LangGraph calls agent.invoke({"messages": [("user", prompt)]}).

3. Extracting Tool Outputs from Graph State

LangGraph returns the complete state dictionary {"messages": [...]}. Filtering for getattr(m, "type", None) == "tool" isolates the exact strings returned by greet(), while messages[-1].content holds the assistant's final reply.

1. Core AI Components in 02_hello_world_langgraph.py

• Pydantic Contracts (LangGraphAgentRequest & LangGraphAgentResponse): Document the input payload (prompt: str = "Please greet Alice and Bob.") and output payload (prompt: str, response: str, and tool_calls: list[str]).
• Step ② Model Identifier (model_id): Resolves the target Bedrock model ID (defaulting to "amazon.nova-lite-v1:0").
• @tool def greet(name: str) -> str: Decorated with LangChain's @tool decorator (from langchain.tools import tool). Accepts a person's name: str and returns f"Hello, {name}! Welcome to the world of AI agents.". Its docstring ("Greet someone by name.") and type annotation define the JSON Schema tool specification sent to Bedrock.
• Step ③ Chat Model Initialization (llm = init_chat_model(model_id, model_provider="bedrock_converse")): Instantiates a LangChain chat model using the unified bedrock_converse provider, which translates LangChain messages and tool schemas into Amazon Bedrock Converse API requests.
• Step ④ ReAct Graph Compilation (agent = create_react_agent(model=llm, tools=[greet])): Uses LangGraph's prebuilt factory to compile a cyclic ReAct state graph consisting of an LLM reasoning node and a tool execution node. Unlike Strands' stateful Agent instance, a compiled LangGraph graph is stateless across invocations unless a checkpointer is attached, so each agent.invoke(...) call starts with a fresh message list.
• run_langgraph_agent(prompt: str) -> dict: Executes the compiled ReAct graph and extracts the tool observations:
  • Step ⑤ (response = agent.invoke({"messages": [{"role": "user", "content": prompt}]})): Seeds the graph state with the user message and runs the Reason → Act → Observe loop until the model emits a final response with no further tool calls. Extracts the full conversation history via messages = response["messages"].
  • Step ⑥ (tool_calls = [f"{m.name} -> {m.content}" for m in messages if m.type == "tool"]): Filters the message history for ToolMessage objects (m.type == "tool") to capture each tool execution and its returned greeting string, returning {"prompt": prompt, "response": messages[-1].text, "tool_calls": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_langgraph_agent(prompt) invokes agent.invoke({"messages": [{"role": "user", "content": prompt}]}) on the compiled LangGraph ReAct graph (e.g., for "Please greet Alice and Bob.").
2. On Turn 1, the LangGraph agent node calls Amazon Bedrock via bedrock_converse. Nova Lite inspects the greet tool schema and emits tool calls for greet(name="Alice") and greet(name="Bob").
3. LangGraph's tool node executes greet("Alice") and greet("Bob"), appends two ToolMessage objects (m.type == "tool") to state["messages"], and routes control back to the agent node.
4. On Turn 2, Bedrock reads the tool outputs and produces the final assistant reply (messages[-1].text).
5. Step ⑥ collects the tool observations into tool_calls and returns {"prompt": prompt, "response": messages[-1].text, "tool_calls": tool_calls}.

LangGraph ReAct Hello World Agent Service.

"""LangGraph ReAct Hello World Agent Service.

Demonstrates the LangGraph tool-calling ReAct agentic loop on Amazon Bedrock:
1. Defines a tool `greet(name: str)`.
2. Passes the prompt to a LangGraph ReAct agent backed by Bedrock Converse.
3. Observes tool execution steps (calling greet for each person).
4. Synthesizes the final response.
"""

import os

from dotenv import find_dotenv, load_dotenv
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel

# ① Load AWS credentials and Bedrock configuration from .env
load_dotenv(find_dotenv())


class LangGraphAgentRequest(BaseModel):
    """Input payload for LangGraph Hello World Agent."""

    prompt: str = "Please greet Alice and Bob."


class LangGraphAgentResponse(BaseModel):
    """Output payload from LangGraph Hello World Agent."""

    prompt: str
    response: str
    tool_calls: list[str]


# ② Resolve the target Bedrock model identifier
model_id = os.getenv("BEDROCK_MODEL_ID", "amazon.nova-lite-v1:0")


@tool
def greet(name: str) -> str:
    """Greet someone by name."""
    return f"Hello, {name}! Welcome to the world of AI agents."


# ③ Initialize the LLM via the Amazon Bedrock Converse provider
llm = init_chat_model(model_id, model_provider="bedrock_converse")

# ④ Create a LangGraph ReAct agent bound to the greet tool
agent = create_react_agent(model=llm, tools=[greet])


def run_langgraph_agent(prompt: str = "Please greet Alice and Bob.") -> dict:
    """Run the LangGraph tool-enabled ReAct agent."""
    # ⑤ Invoke the ReAct agent loop (Reason -> Act -> Observe)
    response = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
    messages = response["messages"]

    # ⑥ Every "tool" message is the observation from one greet() call
    tool_calls = [f"{m.name} -> {m.content}" for m in messages if m.type == "tool"]

    return {"prompt": prompt, "response": messages[-1].text, "tool_calls": tool_calls}

Flask Blueprint server and route handler (langgraph_route) in src/python/app.py.

"""Flask server for 05-ai-aws-strands."""

import asyncio
import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


s_hello = _load_service("01_hello_world_agent.py", "s_hello")
s_langgraph = _load_service("02_hello_world_langgraph.py", "s_langgraph")
s_fn_tool = _load_service("03_function_to_tool.py", "s_fn_tool")
s_tip = _load_service("04_tip_calculator.py", "s_tip")
s_sales = _load_service("05_multi_tool_sales.py", "s_sales")
s_inv = _load_service("06_custom_tool_inventory.py", "s_inv")
s_calc = _load_service("07_prebuilt_tools.py", "s_calc")
s_multi_pre = _load_service("08_multi_prebuilt_tools.py", "s_multi_pre")
s_aws = _load_service("09_use_aws.py", "s_aws")
s_cls_tool = _load_service("10_class_based_tools.py", "s_cls_tool")
s_async_tool = _load_service("11_async_tools.py", "s_async_tool")
s_travel = _load_service("12_travel_assistant.py", "s_travel")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/hello-world-agent", methods=["POST"])
def hello_world_route():
    # ① Parse prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Invoke the Strands Hello World Agent in 01_hello_world_agent.py
        res = s_hello.run_hello_world_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/langgraph-agent", methods=["POST"])
def langgraph_route():
    # ① Parse prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Invoke the LangGraph + Bedrock Converse agent in 02_hello_world_langgraph.py
        res = s_langgraph.run_langgraph_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/function-to-tool", methods=["POST"])
def function_to_tool_route():
    # ① Parse question from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the HTTP server health-check tool agent in 03_function_to_tool.py
        res = s_fn_tool.run_function_to_tool_agent(question)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tip-calculator", methods=["POST"])
def tip_calculator_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the multi-parameter tip calculator agent in 04_tip_calculator.py
        res = s_tip.run_tip_calculator_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-tool-sales", methods=["POST"])
def multi_tool_sales_route():
    # ① Parse task from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    task = (data.get("task") or data.get("message") or "").strip()
    if not task:
        return jsonify({"error": "Task cannot be empty"}), 400
    try:
        # ② Run the 3-tool sales pipeline agent in 05_multi_tool_sales.py
        res = s_sales.run_multi_tool_sales_agent(task)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/custom-inventory", methods=["POST"])
def custom_inventory_route():
    # ① Parse question from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the custom inventory lookup agent in 06_custom_tool_inventory.py
        res = s_inv.run_custom_inventory_agent(question)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/prebuilt-calculator", methods=["POST"])
def prebuilt_calculator_route():
    # ① Parse math query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the prebuilt calculator tool agent in 07_prebuilt_tools.py
        res = s_calc.run_prebuilt_calculator_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-prebuilt-tools", methods=["POST"])
def multi_prebuilt_tools_route():
    # ① Parse task from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    task = (data.get("task") or data.get("message") or "").strip()
    if not task:
        return jsonify({"error": "Task cannot be empty"}), 400
    try:
        # ② Run the http_request + calculator agent in 08_multi_prebuilt_tools.py
        res = s_multi_pre.run_multi_prebuilt_tools_agent(task)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/use-aws", methods=["POST"])
def use_aws_route():
    # ① Parse natural-language AWS prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Run the prebuilt use_aws (boto3) agent in 09_use_aws.py
        res = s_aws.run_use_aws_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/class-based-tools", methods=["POST"])
def class_based_tools_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the stateful InventoryTools class agent in 10_class_based_tools.py
        res = s_cls_tool.run_class_based_tools_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/async-tools", methods=["POST"])
def async_tools_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the async parallel warehouse tools agent in 11_async_tools.py
        res = asyncio.run(s_async_tool.run_async_tools_agent(query))
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/travel-assistant", methods=["POST"])
def travel_assistant_route():
    # ① Parse travel query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the 4-tool travel assistant agent in 12_travel_assistant.py
        res = s_travel.run_travel_assistant_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Review these core concepts comparing LangGraph's ReAct agent architecture with AWS Strands on Amazon Bedrock:

1. Framework-Agnostic ReAct Loop: The Reason → Act → Observe cycle is a fundamental AI agent pattern, not a library-specific feature; both LangGraph and AWS Strands drive the exact same Bedrock Converse tool-calling protocol underneath.
2. LangChain's Unified Model Loader (init_chat_model): Calling init_chat_model(model_id, model_provider="bedrock_converse") connects LangChain/LangGraph to Amazon Bedrock's Converse API using the same amazon.nova-lite-v1:0 model ID as Strands.
3. Prebuilt ReAct State Graph (create_react_agent): langgraph.prebuilt.create_react_agent(model=llm, tools=[greet]) compiles a cyclic state graph containing an agent node (LLM call), a conditional router (checking for tool_calls), and a tool execution node.
4. Tool Definition Parity (@tool): Just like Strands, LangChain's @tool decorator inspects the Python function name (greet), parameter type hints (name: str), and docstring ("Greet someone by name.") to generate the JSON Schema sent to the model.
5. Multi-Entity Tool Fan-Out: When prompted with "Please greet Alice and Bob", the model recognizes two target entities and invokes greet("Alice") and greet("Bob") before synthesizing its final reply.
6. Stateless Compiled Graph Invocation: Because create_react_agent is compiled without a persistent checkpointer here, agent can be instantiated once at module level and safely invoked concurrently via agent.invoke({"messages": [...]}).
7. Observing Tool Execution via ToolMessage: In LangGraph, every tool return value is appended to response["messages"] as a message with m.type == "tool", allowing [f"{m.name} -> {m.content}" for m in messages if m.type == "tool"] to capture the full execution trace.
8. Boilerplate Comparison (LangGraph vs. Strands): While LangGraph offers fine-grained graph node and edge customization for complex state machines, AWS Strands reduces standard tool-calling agents to a more concise Agent(model=..., tools=[...])(prompt) call.