LLM Agents and Tool Use

A chatbot is a knowledgeable advisor — they can tell you what to do but you have to do it yourself. An agent is a capable employee — you tell them what you want and they figure out the steps, use the right tools, and hand you the result. The difference is agency: the ability to act in the world, not just generate text about acting.

The key constraint: agents are only as reliable as the LLM driving them. Every tool call is an LLM decision — and LLMs make mistakes. Production agents need guardrails, confirmation steps for irreversible actions, and human-in-the-loop for high-stakes decisions.

import os, json
from groq import Groq

client = Groq(api_key=os.environ.get('GROQ_API_KEY'))

# ── Define tools as JSON schemas ──────────────────────────────────────
TOOLS = [
    {
        'type': 'function',
        'function': {
            'name': 'get_transaction',
            'description': 'Look up a Stripe transaction by ID. Returns status, amount, merchant, and settlement date.',
            'parameters': {
                'type': 'object',
                'properties': {
                    'transaction_id': {
                        'type': 'string',
                        'description': 'The Stripe transaction ID (e.g. TXN123456)',
                    },
                },
                'required': ['transaction_id'],
            },
        },
    },
    {
        'type': 'function',
        'function': {
            'name': 'get_settlement_status',
            'description': 'Check settlement status for a merchant. Returns pending and completed settlement amounts.',
            'parameters': {
                'type': 'object',
                'properties': {
                    'merchant_id': {
                        'type': 'string',
                        'description': 'The merchant ID',
                    },
                    'date_range': {
                        'type': 'string',
                        'description': 'Date range: "today", "last_7_days", "last_30_days"',
                        'enum': ['today', 'last_7_days', 'last_30_days'],
                    },
                },
                'required': ['merchant_id'],
            },
        },
    },
    {
        'type': 'function',
        'function': {
            'name': 'calculate_fee',
            'description': 'Calculate Stripe processing fee for a transaction amount.',
            'parameters': {
                'type': 'object',
                'properties': {
                    'amount': {'type': 'number', 'description': 'Amount in USD'},
                    'payment_type': {
                        'type': 'string',
                        'enum': ['domestic', 'international'],
                    },
                },
                'required': ['amount', 'payment_type'],
            },
        },
    },
]

# ── Real tool implementations ─────────────────────────────────────────
def get_transaction(transaction_id: str) -> dict:
    """Simulated transaction lookup."""
    mock_db = {
        'TXN123': {'status': 'settled', 'amount': 5000, 'merchant': 'DoorDash', 'date': '2026-03-28'},
        'TXN456': {'status': 'pending', 'amount': 12500, 'merchant': 'Amazon', 'date': None},
        'TXN789': {'status': 'failed',  'amount': 2499, 'merchant': 'Uber Eats', 'date': None},
    }
    return mock_db.get(transaction_id, {'error': 'Transaction not found'})

def get_settlement_status(merchant_id: str, date_range: str = 'last_7_days') -> dict:
    return {
        'merchant_id':  merchant_id,
        'date_range':   date_range,
        'pending_inr':  45000,
        'settled_inr':  230000,
        'next_settlement': '2026-03-30',
    }

def calculate_fee(amount: float, payment_type: str) -> dict:
    rate = 0.03 if payment_type == 'international' else 0.02
    return {'amount': amount, 'fee': amount * rate, 'rate_pct': rate * 100}

TOOL_FUNCTIONS = {
    'get_transaction':      get_transaction,
    'get_settlement_status': get_settlement_status,
    'calculate_fee':        calculate_fee,
}

# ── Agent loop with function calling ─────────────────────────────────
def run_agent(user_message: str, max_turns: int = 5) -> str:
    messages = [{'role': 'user', 'content': user_message}]
    print(f"User: {user_message}
")

    for turn in range(max_turns):
        response = client.chat.completions.create(
            model='openai/gpt-oss-120b',
            messages=messages,
            tools=TOOLS,
            tool_choice='auto',   # LLM decides when to call tools
            temperature=0,
            max_tokens=500,
        )
        msg = response.choices[0].message

        # ── No tool call — final answer ───────────────────────────────
        if not msg.tool_calls:
            print(f"Agent: {msg.content}")
            return msg.content

        # ── Execute each tool call ────────────────────────────────────
        messages.append({'role': 'assistant', 'content': msg.content,
                          'tool_calls': [tc.model_dump() for tc in msg.tool_calls]})

        for tool_call in msg.tool_calls:
            fn_name   = tool_call.function.name
            fn_args   = json.loads(tool_call.function.arguments)

            print(f"Tool call: {fn_name}({fn_args})")

            if fn_name in TOOL_FUNCTIONS:
                result = TOOL_FUNCTIONS[fn_name](**fn_args)
            else:
                result = {'error': f'Unknown tool: {fn_name}'}

            print(f"Result: {result}
")

            messages.append({
                'role':         'tool',
                'tool_call_id': tool_call.id,
                'content':      json.dumps(result),
            })

    return "Max turns reached."

# ── Test the agent ────────────────────────────────────────────────────
run_agent("What is the status of transaction TXN456?")
print("=" * 60)
run_agent("I'm merchant MID789. How much will I pay in fees for a $50,000 international payment, and what's my settlement status?")

import os, json
from groq import Groq
from collections import deque

client = Groq(api_key=os.environ.get('GROQ_API_KEY'))

# ── Short-term memory — sliding window to stay within context limit ───
class ConversationMemory:
    def __init__(self, max_messages: int = 20, system_prompt: str = ''):
        self.system_prompt = system_prompt
        self.history       = deque(maxlen=max_messages)  # auto-drops oldest
        self.summary       = ''   # compressed summary of dropped messages

    def add(self, role: str, content: str):
        self.history.append({'role': role, 'content': content})

    def get_messages(self) -> list[dict]:
        msgs = []
        if self.system_prompt:
            msgs.append({'role': 'system', 'content': self.system_prompt})
        if self.summary:
            msgs.append({'role': 'system',
                          'content': f'Summary of earlier conversation: {self.summary}'})
        msgs.extend(list(self.history))
        return msgs

    def compress(self):
        """Summarise history when approaching context limit."""
        if len(self.history) < 15:
            return
        old_messages = list(self.history)[:10]
        summary_prompt = (
            'Summarise this conversation in 3 sentences, '
            'keeping all important facts:

' +
            '
'.join(f"{m['role']}: {m['content']}" for m in old_messages)
        )
        response = client.chat.completions.create(
            model='openai/gpt-oss-120b',
            messages=[{'role': 'user', 'content': summary_prompt}],
            max_tokens=200, temperature=0,
        )
        self.summary = response.choices[0].message.content
        # Remove the messages we summarised
        for _ in range(10):
            self.history.popleft()

# ── Long-term memory — persist key facts across sessions ──────────────
class LongTermMemory:
    def __init__(self):
        self.facts = {}   # In production: Redis/Postgres

    def remember(self, user_id: str, key: str, value: str):
        if user_id not in self.facts:
            self.facts[user_id] = {}
        self.facts[user_id][key] = value

    def recall(self, user_id: str) -> str:
        if user_id not in self.facts:
            return ''
        facts = self.facts.get(user_id, {})
        return '
'.join(f'- {k}: {v}' for k, v in facts.items())

    def extract_and_store(self, user_id: str, conversation: str):
        """Ask LLM to extract memorable facts from conversation."""
        prompt = f"""Extract key facts worth remembering about this user from the conversation.
Return as JSON: {{"facts": [{{"key": "...", "value": "..."}}]}}
Only extract genuinely useful facts (preferences, identity, recurring issues).

Conversation:
{conversation}"""
        response = client.chat.completions.create(
            model='openai/gpt-oss-120b',
            messages=[{'role': 'user', 'content': prompt}],
            max_tokens=300, temperature=0,
        )
        try:
            data = json.loads(response.choices[0].message.content)
            for fact in data.get('facts', []):
                self.remember(user_id, fact['key'], fact['value'])
        except Exception:
            pass

# ── Demo: agent with memory ───────────────────────────────────────────
ltm  = LongTermMemory()
conv = ConversationMemory(
    max_messages=10,
    system_prompt='You are a Stripe support agent with memory of past interactions.',
)

# Simulate session 1
user_id = 'merchant_MID001'
turns = [
    "Hi, I'm Sarah. I run an online boutique clothing store on my website.",
    "I process about 50 orders a day averaging $200 each.",
    "My biggest problem is international payment failures.",
]

print("Session 1:")
for user_msg in turns:
    conv.add('user', user_msg)
    response = client.chat.completions.create(
        model='openai/gpt-oss-120b',
        messages=conv.get_messages(),
        max_tokens=100, temperature=0.3,
    )
    reply = response.choices[0].message.content
    conv.add('assistant', reply)
    print(f"  User:  {user_msg}")
    print(f"  Agent: {reply[:80]}...
")

# Extract and store facts from this session
full_conv = '
'.join(f"{m['role']}: {m['content']}" for m in list(conv.history))
ltm.extract_and_store(user_id, full_conv)

print("Stored facts about this merchant:")
print(ltm.recall(user_id))

import os, json
from groq import Groq
from dataclasses import dataclass
from typing import Callable

client = Groq(api_key=os.environ.get('GROQ_API_KEY'))

# ── Specialist agent factory ──────────────────────────────────────────
@dataclass
class SpecialistAgent:
    name:        str
    description: str
    system:      str
    tools:       list[dict]
    tool_fns:    dict[str, Callable]

    def run(self, task: str) -> str:
        messages = [
            {'role': 'system', 'content': self.system},
            {'role': 'user',   'content': task},
        ]
        for _ in range(5):
            response = client.chat.completions.create(
                model='openai/gpt-oss-120b',
                messages=messages,
                tools=self.tools if self.tools else None,
                tool_choice='auto' if self.tools else None,
                temperature=0, max_tokens=400,
            )
            msg = response.choices[0].message
            if not msg.tool_calls:
                return msg.content or ''

            messages.append({
                'role': 'assistant', 'content': msg.content,
                'tool_calls': [tc.model_dump() for tc in msg.tool_calls],
            })
            for tc in msg.tool_calls:
                fn   = tc.function.name
                args = json.loads(tc.function.arguments)
                res  = self.tool_fns.get(fn, lambda **k: {'error': 'unknown'})(**args)
                messages.append({
                    'role': 'tool', 'tool_call_id': tc.id,
                    'content': json.dumps(res),
                })
        return 'Max steps reached.'

# ── Define specialist agents ──────────────────────────────────────────
transaction_agent = SpecialistAgent(
    name='Transaction Agent',
    description='Looks up transaction details and dispute status',
    system='You are a transaction lookup specialist. Use tools to find transaction data.',
    tools=[{
        'type': 'function',
        'function': {
            'name': 'get_transaction',
            'description': 'Get transaction details by ID',
            'parameters': {
                'type': 'object',
                'properties': {'txn_id': {'type': 'string'}},
                'required': ['txn_id'],
            },
        },
    }],
    tool_fns={
        'get_transaction': lambda txn_id: {
            'id': txn_id, 'amount': 5000, 'status': 'disputed',
            'merchant': 'DoorDash', 'customer': 'James Miller',
            'dispute_deadline': '2026-04-01',
        },
    },
)

compliance_agent = SpecialistAgent(
    name='Compliance Agent',
    description='Checks deadlines and evidence requirements',
    system='You are a compliance specialist. Assess dispute requirements and deadlines.',
    tools=[],
    tool_fns={},
)

communication_agent = SpecialistAgent(
    name='Communication Agent',
    description='Drafts customer and merchant communications',
    system='You are a communication specialist. Draft clear, professional messages.',
    tools=[],
    tool_fns={},
)

# ── Orchestrator ──────────────────────────────────────────────────────
def orchestrator(request: str) -> str:
    """
    Breaks a complex request into sub-tasks,
    delegates to specialists, and synthesises the final response.
    """
    plan_prompt = f"""You are an orchestrator for a Stripe dispute resolution system.

Available specialists:
- Transaction Agent: looks up transaction details and dispute status
- Compliance Agent: checks deadlines and evidence requirements
- Communication Agent: drafts responses and emails

For this request, create a plan as JSON:
{{"steps": [{{"agent": "...", "task": "..."}}]}}

Request: {request}"""

    plan_response = client.chat.completions.create(
        model='openai/gpt-oss-120b',
        messages=[{'role': 'user', 'content': plan_prompt}],
        temperature=0, max_tokens=300,
    )
    plan_text = plan_response.choices[0].message.content
    try:
        import re
        json_match = re.search(r'{.*}', plan_text, re.DOTALL)
        plan = json.loads(json_match.group()) if json_match else {'steps': []}
    except Exception:
        plan = {'steps': []}

    agents = {
        'Transaction Agent':   transaction_agent,
        'Compliance Agent':    compliance_agent,
        'Communication Agent': communication_agent,
    }

    results = {}
    print(f"
Orchestrator plan: {len(plan.get('steps', []))} steps")
    for step in plan.get('steps', []):
        agent_name = step.get('agent', '')
        task       = step.get('task', '')
        if agent_name in agents:
            print(f"  → {agent_name}: {task[:60]}...")
            results[agent_name] = agents[agent_name].run(task)

    # Synthesise final answer
    synthesis_prompt = (
        f"Original request: {request}

"
        + '

'.join(f"{name} result:
{result}"
                       for name, result in results.items())
        + "

Synthesise a clear, complete response:"
    )
    final = client.chat.completions.create(
        model='openai/gpt-oss-120b',
        messages=[{'role': 'user', 'content': synthesis_prompt}],
        temperature=0.2, max_tokens=400,
    )
    return final.choices[0].message.content

result = orchestrator(
    "Transaction TXN999 is disputed. Check the details, "
    "verify the deadline, and draft a response to the customer."
)
print(f"
Final response:
{result}")

import os, json
from groq import Groq
from typing import Callable, Any
import logging

client = Groq(api_key=os.environ.get('GROQ_API_KEY'))
logger = logging.getLogger(__name__)

# ── Layer 1: Tool-level validation ────────────────────────────────────
def safe_tool(fn: Callable, max_calls: int = 3, dry_run: bool = False):
    """Wrapper that adds call counting, dry-run mode, and logging."""
    call_count = [0]

    def wrapper(**kwargs):
        call_count[0] += 1
        if call_count[0] > max_calls:
            return {'error': f'Tool call limit ({max_calls}) exceeded. Stopping.'}

        logger.info(f"Tool call {call_count[0]}/{max_calls}: {fn.__name__}({kwargs})")

        if dry_run:
            return {'dry_run': True, 'would_call': fn.__name__, 'with': kwargs}

        try:
            result = fn(**kwargs)
            logger.info(f"Tool result: {result}")
            return result
        except Exception as e:
            logger.error(f"Tool error: {e}")
            return {'error': str(e)}

    return wrapper

# ── Layer 2: Irreversible action confirmation ─────────────────────────
IRREVERSIBLE_TOOLS = {'send_email', 'delete_record', 'process_refund', 'send_sms'}

def requires_confirmation(tool_name: str, args: dict) -> bool:
    """Return True if this action requires human confirmation."""
    if tool_name in IRREVERSIBLE_TOOLS:
        return True
    # High-value transactions always need confirmation
    if tool_name == 'process_refund' and args.get('amount', 0) > 10_000:
        return True
    return False

def confirm_action(tool_name: str, args: dict) -> bool:
    """In production: show UI confirmation. Here: auto-approve for demo."""
    print(f"  ⚠ Confirmation required: {tool_name}({args})")
    # In production: return based on user UI input
    # For high-stakes: require manager approval
    return True   # auto-approve in demo

# ── Layer 3: Output validation ────────────────────────────────────────
def validate_agent_output(output: str, expected_format: str = None) -> dict:
    """Validate agent output before acting on it."""
    if not output or len(output.strip()) < 10:
        return {'valid': False, 'reason': 'Output too short or empty'}

    # Check for hallucination signals
    hallucination_phrases = [
        'I made up', 'I invented', 'fictional', 'hypothetical example',
    ]
    for phrase in hallucination_phrases:
        if phrase.lower() in output.lower():
            return {'valid': False, 'reason': f'Possible hallucination: "{phrase}"'}

    if expected_format == 'json':
        try:
            json.loads(output)
        except Exception:
            return {'valid': False, 'reason': 'Expected JSON but got invalid JSON'}

    return {'valid': True}

# ── Safe agent with all three layers ─────────────────────────────────
class SafeAgent:
    def __init__(self, tools: list[dict], tool_fns: dict,
                 dry_run: bool = False, max_tool_calls: int = 10):
        self.tools    = tools
        self.tool_fns = {
            name: safe_tool(fn, max_calls=3, dry_run=dry_run)
            for name, fn in tool_fns.items()
        }
        self.total_calls = 0
        self.max_calls   = max_tool_calls

    def run(self, task: str, system: str = '') -> str:
        messages = []
        if system:
            messages.append({'role': 'system', 'content': system})
        messages.append({'role': 'user', 'content': task})

        for turn in range(8):
            if self.total_calls >= self.max_calls:
                return f'Safety limit: max {self.max_calls} tool calls reached.'

            response = client.chat.completions.create(
                model='openai/gpt-oss-120b',
                messages=messages,
                tools=self.tools,
                tool_choice='auto',
                temperature=0, max_tokens=500,
            )
            msg = response.choices[0].message
            if not msg.tool_calls:
                result = msg.content or ''
                validation = validate_agent_output(result)
                if not validation['valid']:
                    print(f"  ⚠ Output validation failed: {validation['reason']}")
                return result

            messages.append({
                'role': 'assistant', 'content': msg.content,
                'tool_calls': [tc.model_dump() for tc in msg.tool_calls],
            })

            for tc in msg.tool_calls:
                fn_name = tc.function.name
                fn_args = json.loads(tc.function.arguments)
                self.total_calls += 1

                # Confirmation for irreversible actions
                if requires_confirmation(fn_name, fn_args):
                    if not confirm_action(fn_name, fn_args):
                        result = {'cancelled': True, 'reason': 'User rejected action'}
                        messages.append({
                            'role': 'tool', 'tool_call_id': tc.id,
                            'content': json.dumps(result),
                        })
                        continue

                fn     = self.tool_fns.get(fn_name, lambda **k: {'error': 'unknown'})
                result = fn(**fn_args)
                messages.append({
                    'role': 'tool', 'tool_call_id': tc.id,
                    'content': json.dumps(result),
                })

        return 'Max turns reached.'

print("SafeAgent: production-ready agent with guardrails")
print("Layers: tool-level validation + confirmation + output validation")

import time
import logging
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

# ── Per-run cost and loop monitor — the layer that sits around the agent ──
# In production this wraps every agent run, regardless of which agent or task.
# Groq/OpenAI-compatible pricing example: $0.15 per 1M input tokens, $0.75 per 1M output tokens
INPUT_COST_PER_TOKEN  = 0.15 / 1_000_000
OUTPUT_COST_PER_TOKEN = 0.75 / 1_000_000

@dataclass
class RunMonitor:
    run_id:          str
    max_usd:         float = 0.50        # hard spend ceiling per run
    max_seconds:     float = 45.0        # wall-clock ceiling per run
    started_at:      float = field(default_factory=time.time)
    spent_usd:       float = 0.0
    tool_call_log:   list  = field(default_factory=list)   # (tool_name, args_hash) pairs

    def record_llm_call(self, input_tokens: int, output_tokens: int):
        cost = input_tokens * INPUT_COST_PER_TOKEN + output_tokens * OUTPUT_COST_PER_TOKEN
        self.spent_usd += cost
        if self.spent_usd > self.max_usd:
            raise RuntimeError(
                f"Run {self.run_id}: spend budget exceeded "
                f"(${self.spent_usd:.4f} > ${self.max_usd:.2f}) — forcing early stop"
            )
        if time.time() - self.started_at > self.max_seconds:
            raise RuntimeError(f"Run {self.run_id}: wall-clock budget exceeded — forcing early stop")

    def record_tool_call(self, tool_name: str, args: dict):
        signature = (tool_name, tuple(sorted(args.items())))
        repeat_count = self.tool_call_log.count(signature)
        self.tool_call_log.append(signature)
        if repeat_count >= 2:
            # Same tool, same arguments, three times in one run — almost always a stuck loop
            raise RuntimeError(
                f"Run {self.run_id}: {tool_name}({args}) repeated {repeat_count + 1} times — "
                f"likely stuck, killing run and alerting on-call"
            )
        logger.info(f"Run {self.run_id}: tool call #{len(self.tool_call_log)} — {tool_name}({args})  "
                    f"spend so far: ${self.spent_usd:.4f}")

# ── Usage inside the agent loop ────────────────────────────────────────
monitor = RunMonitor(run_id='run_8f21ac', max_usd=0.50, max_seconds=45.0)

def guarded_tool_call(monitor, tool_name, args, tool_fn):
    monitor.record_tool_call(tool_name, args)   # raises if this looks like a stuck loop
    return tool_fn(**args)

def guarded_llm_call(monitor, response):
    usage = response.usage   # Groq/OpenAI-compatible responses include token usage
    monitor.record_llm_call(usage.prompt_tokens, usage.completion_tokens)
    return response

print("RunMonitor: wraps every LLM call and tool call in an agent run")
print(f"  Hard spend ceiling: ${monitor.max_usd} per run")
print(f"  Wall-clock ceiling: {monitor.max_seconds}s per run")
print("  Repeated identical tool call (3x) → run killed, on-call alerted")

A single request-response turn where the model calls one function and returns an answer — what is the weather in Austin, followed by one weather-API lookup — is tool use, not agency. What makes something an agent is a loop: the model observes a result, decides what to do next based on that result, and keeps deciding across multiple steps without a human choosing each step in advance. The Stripe dispute example in this module — look up the transaction, then decide whether to check the deadline, then decide whether to draft an email — is an agent because each decision depends on the previous tool's output. One tool call bolted onto a chatbot is not.

Every additional tool is an additional way for the LLM to pick wrong. With three tools (get_transaction, get_settlement_status, calculate_fee) the model rarely confuses which one to call. With thirty overlapping tools — several that could plausibly answer the same question — tool selection itself becomes a failure mode, and hallucinated arguments get more likely because the model is juggling more schemas at once. Production agents are usually more reliable with a small, sharply-scoped toolset than a large general-purpose one; if an agent needs thirty capabilities, that is often a sign it should be several specialist agents behind an orchestrator, not one agent with thirty tools.

ReAct and function calling solve different problems. Function calling is the wire format — how a tool call is represented and returned (structured JSON instead of text you regex out). ReAct is the reasoning pattern — the model explicitly reasons through what it knows, what it still needs, and which tool gets it there, before acting, then observes the result and reasons again before the next action. You can, and in production usually do, implement ReAct-style reasoning using native function calling as the execution mechanism. Dropping the reasoning step entirely — jumping straight to tool calls with no visible intermediate reasoning — is what actually causes agents to call plausible-sounding tools that do not fit the actual task.

A workflow (or chain) has its control flow fixed by the developer ahead of time: step one always runs, then step two, then step three, regardless of what step one returns — the orchestrator code decides the sequence. An agent has its control flow decided by the LLM at runtime: the multi-agent orchestrator in this module writes a plan and can route to a different specialist, skip a step, or loop back depending on what a previous specialist found. Workflows are more predictable and cheaper to run; agents are more flexible but strictly less predictable, because the same input can legitimately take a different path through the system on different runs. Choosing between them is an engineering trade-off, not a matter of which sounds more impressive.

Nothing about the agent loop guarantees this. Left unmanaged, an LLM that gets an error or an empty result from a tool frequently retries the identical call rather than changing strategy — this module's errors section covers exactly this failure mode. Self-correction only happens because you engineer it: tracking (tool_name, args) pairs already tried, injecting an explicit message when a repeat is detected, returning structured error types the model can reason about instead of opaque failures, and capping max_turns so a stuck agent stops instead of burning cost indefinitely. Robust agent behaviour is a designed property of the system around the LLM, not an emergent property of the LLM itself.

A simple function-calling pipeline executes one, or a fixed sequence of, tool calls per request and returns an answer — the developer decides how many calls happen and in what order. An agent runs a loop: after each tool result, the LLM itself decides whether it has enough information to answer or needs to call another tool, and if so, which one — that decision is made by the model at each step, not hardcoded by the developer. The practical consequence is that an agent's number of steps and exact path are not known in advance, which is exactly why production agents need max_turns caps, loop detection, and cost budgets that a fixed pipeline does not need.

Two dominant failure modes: infinite or repeated tool calls, where the LLM retries the same call after getting an error or null result instead of changing approach; and hallucinated arguments, where the model invents plausible-sounding field names or values that are not in the tool's schema. In production you catch the first by hashing the tool name and arguments for every call in the current run and flagging repeats back to the model with the prior result attached, plus a hard max_turns ceiling as a backstop. You catch the second by making schemas strict — required fields listed explicitly, additionalProperties set to false, enum constraints on categorical fields — and validating every tool call against the schema before executing it, feeding validation errors back to the model rather than letting a malformed call reach a real system.

ReAct interleaves reasoning and acting: the model explicitly reasons about what it knows and what it needs, takes an action (a tool call), observes the result, then reasons again before the next action — as opposed to jumping straight from a question to a tool call with no visible intermediate reasoning. It is still relevant because it addresses a different layer than native function calling. Function calling is just the transport — how a tool call is structured and returned. ReAct is about forcing the model to think before each action, which measurably reduces wrong-tool selection and hallucinated arguments on multi-step tasks. In practice you implement ReAct-style reasoning on top of native function calling — they are complementary, not competing.

Several layers, applied together: a hard max_turns and max_tool_calls ceiling so a stuck agent stops instead of looping indefinitely; per-tool call-count limits so one unreliable tool doesn't dominate the budget; classification of every tool as reversible or irreversible, with irreversible actions — send email, process a refund, delete a record — routed through an explicit human confirmation step rather than executed automatically; a dry_run mode during development so tools log intended actions without executing them; and output validation before acting on the agent's final answer, checking for signals like hallucination phrases or malformed structured output. None of these are optional extras — an agent that can take real-world actions without all of them is not production-ready regardless of how good the underlying LLM is.

Ask whether the sequence of steps can be known ahead of time. If the task always follows the same steps regardless of intermediate results — fetch data, transform it, write it somewhere — a deterministic workflow is more reliable, cheaper, and easier to debug than an agent, because its control flow doesn't depend on an LLM's per-step judgment. Reach for an agent when the right next step genuinely depends on what a previous step returned in a way you can't enumerate in advance — like the multi-agent dispute system in this module, where whether compliance needs to flag risk depends on what the transaction lookup actually found. The rule of thumb: use the least autonomous architecture that gets the job done, since every increment of autonomy trades predictability and cost control for flexibility.