Skip to content

Learn Agentic AI by Building One — A Hands-On Guide - Lesson 6: The Agent Loop — Think → Act → Observe → Repeat

7 minute read
Content level: Intermediate
0

Introduction to how Agentic AI works behind the scene

Lesson 5 handled ONE tool call. But real questions need MULTIPLE steps:

"Compare urgent vs low severity resolution times"
  → Step 1: get_avg_resolution(severity="Urgent")
  → Step 2: get_avg_resolution(severity="Low")
  → Step 3: Compare and answer

This is the AGENT LOOP — the core pattern of agentic AI:

┌──────────────────────────────────────────┐
│                                          │
│   User question                          │
│        ↓                                 │
│   ┌─→ THINK: What do I need to do?       │
│   │     ↓                                │
│   │   ACT: Call a tool                   │
│   │     ↓                                │
│   │   OBSERVE: See the result            │
│   │     ↓                                │
│   │   Need more info? ──YES──→ loop back │
│   │     ↓ NO                             │
│   │   RESPOND: Give final answer         │
│   └──────────────────────────────────────┘

This loop is what makes an "agent" different from a simple chatbot. A chatbot responds once. An agent keeps working until the task is done.

The Setup

Same tools from Lesson 5, with a slightly larger dataset:

import json
import boto3

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

CASES_DB = [
    {"id": 1, "subject": "EC2 Instance Rebooted", "severity": "Urgent", "service": "EC2", "days_to_resolve": 10},
    {"id": 2, "subject": "RDS Connection Timeout", "severity": "Urgent", "service": "RDS", "days_to_resolve": 68},
    {"id": 3, "subject": "S3 Permission Denied", "severity": "High", "service": "S3", "days_to_resolve": 1},
    {"id": 4, "subject": "Lambda Memory Limit", "severity": "Low", "service": "Lambda", "days_to_resolve": 12},
    {"id": 5, "subject": "EFS Throughput Issue", "severity": "High", "service": "EFS", "days_to_resolve": 20},
    {"id": 6, "subject": "EKS Pods Failing", "severity": "Normal", "service": "EKS", "days_to_resolve": 11},
    {"id": 7, "subject": "CloudFront Quota", "severity": "Low", "service": "CloudFront", "days_to_resolve": 11},
    {"id": 8, "subject": "Aurora Failover RCA", "severity": "Low", "service": "RDS", "days_to_resolve": 35},
]


def get_case_count(severity: str = None) -> str:
    if severity:
        count = sum(1 for c in CASES_DB if c["severity"].lower() == severity.lower())
        return f"{count} cases with severity '{severity}'"
    return f"{len(CASES_DB)} total cases"


def get_avg_resolution(service: str = None, severity: str = None) -> str:
    cases = CASES_DB
    if service:
        cases = [c for c in cases if c["service"].lower() == service.lower()]
    if severity:
        cases = [c for c in cases if c["severity"].lower() == severity.lower()]
    if not cases:
        return "No cases found for the given filter"
    avg = sum(c["days_to_resolve"] for c in cases) / len(cases)
    parts = [f"service={service}" if service else None, f"severity={severity}" if severity else None]
    label = " for " + ", ".join(p for p in parts if p) if any(parts) else ""
    return f"{avg:.1f} days average resolution{label}"


def list_cases(severity: str = None) -> str:
    cases = CASES_DB
    if severity:
        cases = [c for c in cases if c["severity"].lower() == severity.lower()]
    return json.dumps([{"subject": c["subject"], "severity": c["severity"], "service": c["service"], "days": c["days_to_resolve"]} for c in cases])


TOOL_FUNCTIONS = {
    "get_case_count": get_case_count,
    "get_avg_resolution": get_avg_resolution,
    "list_cases": list_cases,
}

TOOLS = [
    {"name": "get_case_count", "description": "Count cases, optionally by severity",
     "input_schema": {"type": "object", "properties": {"severity": {"type": "string", "enum": ["Urgent","High","Normal","Low"]}}, "required": []}},
    {"name": "get_avg_resolution", "description": "Average resolution time in days. Can filter by service name and/or severity level.",
     "input_schema": {"type": "object", "properties": {"service": {"type": "string", "description": "AWS service name like EC2, RDS, S3"}, "severity": {"type": "string", "description": "Severity level", "enum": ["Urgent","High","Normal","Low"]}}, "required": []}},
    {"name": "list_cases", "description": "List cases with details, optionally by severity",
     "input_schema": {"type": "object", "properties": {"severity": {"type": "string"}}, "required": []}},
]

The Agent Loop

This is the key pattern. Instead of handling one tool call, we loop until the AI says it's done:

def agent_run(question: str, max_steps: int = 10) -> str:
    """
    The agent loop: keep calling tools until the AI has enough info to answer.
    This is the fundamental pattern behind every AI agent.
    """

    print(f"🧑 Question: {question}")

    messages = [{"role": "user", "content": question}]
    step = 0

    while step < max_steps:
        step += 1

        # THINK: Send everything to the AI
        response = bedrock.invoke_model(
            modelId="us.anthropic.claude-sonnet-4-20250514-v1:0",
            contentType="application/json",
            accept="application/json",
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1024,
                "system": "You are a support case analyst. Use tools to answer questions. Be concise and data-driven.",
                "tools": TOOLS,
                "messages": messages,
            }),
        )
        result = json.loads(response["body"].read())

        # If AI is done (no more tool calls), return the answer
        if result["stop_reason"] == "end_turn":
            final = next((b["text"] for b in result["content"] if b["type"] == "text"), "")
            print(f"\n✅ Final answer (after {step} step(s)):")
            print(f"   {final}")
            return final

        # ACT: Execute each tool the AI wants to call
        # (AI can request MULTIPLE tools in one step!)
        messages.append({"role": "assistant", "content": result["content"]})

        tool_results = []
        for block in result["content"]:
            if block["type"] == "tool_use":
                name = block["name"]
                args = block["input"]
                print(f"   Step {step} 🔧 {name}({args})")

                # Execute the tool
                output = TOOL_FUNCTIONS[name](**args)
                print(f"          📊 → {output}")

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block["id"],
                    "content": output,
                })

        # OBSERVE: Send results back and loop
        messages.append({"role": "user", "content": tool_results})

    return "Max steps reached without a final answer."

Running It

Simple (1 tool call):

agent_run("How many urgent cases are there?")

Output:

🧑 Question: How many urgent cases are there?

Step 1 🔧 get_case_count({'severity': 'Urgent'})
📊 → 2 cases with severity 'Urgent'

✅ Final answer (after 2 step(s)):
There are 2 urgent cases currently in the system.

Medium (2 tool calls):

agent_run("Compare the number of urgent vs low severity cases")

Output:

🧑 Question: Compare the number of urgent vs low severity cases

Step 1 🔧 get_case_count({'severity': 'Urgent'})
📊 → 2 cases with severity 'Urgent'

Step 1 🔧 get_case_count({'severity': 'Low'})
📊 → 3 cases with severity 'Low'

✅ Final answer (after 2 step(s)):
Urgent vs Low Severity Case Comparison:

  • Urgent cases: 2
  • Low severity cases: 3

There are 50% more low severity cases than urgent cases (3 vs 2). This suggests a relatively balanced distribution with slightly more low-priority issues in the current case load.

Complex (multiple tool calls + reasoning):

agent_run("Which severity level has the worst resolution time? Show me the cases.")

Output:

🧑 Question: Which severity level has the worst resolution time? Show me the cases.

Step 1 🔧 get_avg_resolution({'severity': 'Urgent'})
📊 → 39.0 days average resolution for severity=Urgent

Step 1 🔧 get_avg_resolution({'severity': 'High'})
📊 → 10.5 days average resolution for severity=High

Step 1 🔧 get_avg_resolution({'severity': 'Normal'})
📊 → 11.0 days average resolution for severity=Normal

Step 1 🔧 get_avg_resolution({'severity': 'Low'})
📊 → 19.3 days average resolution for severity=Low

Step 2 🔧 list_cases({'severity': 'Urgent'})
📊 → [{"subject": "EC2 Instance Rebooted", "severity": "Urgent", "service": "EC2", "days": 10}, {"subject": "RDS Connection Timeout", "severity": "Urgent", "service": "RDS", "days": 68}]

✅ Final answer (after 3 step(s)):
Here are the urgent cases:

  1. EC2 Instance Rebooted (EC2) - 10 days
  2. RDS Connection Timeout (RDS) - 68 days

The RDS connection timeout case is driving up the average with 68 days resolution time, while the EC2 case was resolved relatively quickly at 10 days. This suggests there may be a specific issue with the RDS case that needs attention.

Watch how the AI autonomously decides how many steps it needs. For the complex question, it calls multiple tools, compares the results, then fetches the specific cases — all without you telling it the steps.

Key Takeaway

The agent loop is: Think → Act → Observe → Repeat. The AI autonomously decides how many steps it needs. This is what makes it an "agent" — it plans and executes multi-step tasks. A chatbot gives you one response. An agent works through a problem.

Next up: Lesson 7 — A full agent with real-world tools and reasoning →
Previous: Lesson 5 — The big one: giving AI the ability to use TOOLS →