Beyond Prompting: The ReAct Loop & Function Calling
Master the Thought-Action-Observation lifecycle, parse tool calls safely, and implement human-in-the-loop checkpoints.
- •Scrub through live Thought-Action-Observation traces with our Agentic Workspace widget
- •Enforce strict schema validation and error feedback loops to cure malformed JSON calls
- •Implement deterministic recursion guards and token expenditure circuit breakers
- •Design human-in-the-loop approval gates for high-impact infrastructure actions
The ReAct Cognitive Cycle & Tool Scrubber
Raw LLMs hallucinate because they lack real-time environment access. The ReAct (Reason + Act) architecture grounds the model: it generates an internal reasoning trace ('Thought'), executes external environment tools ('Action'), inspects the structured output ('Observation'), and repeats until the goal is solved. Test the interactive scrubber below to trace RPC calls, token usage, and human approval gates.
ReAct Agent Cognitive Cycle & Approval Flow
ReAct Cognitive Loop & RPC Trace Scrubber
User Prompt & Instruction Framing
Production Agent State Engine Implementation Suite
Inspect the production-ready implementation of a hardened ReAct agent. It includes strict Pydantic tool registration, automated error-reflection loops that coach the LLM when JSON schemas fail, and pause/resume human approval gates.
import asyncio
import json
from typing import List, Dict, Any, Optional
class ReActAgent:
"""Robust ReAct runtime with circuit breakers, token budgets, and tool dispatch."""
def __init__(self, llm_client, tool_registry, max_steps: int = 8, token_budget: int = 4000):
self.client = llm_client
self.registry = tool_registry
self.max_steps = max_steps
self.token_budget = token_budget
async def run(self, user_query: str) -> Dict[str, Any]:
messages = [
{"role": "system", "content": "You are a precise autonomous engineering agent. Use tools to verify all factual claims before concluding."},
{"role": "user", "content": user_query}
]
total_tokens = 0
for step in range(self.max_steps):
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=self.registry.get_schemas(),
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
total_tokens += response.usage.total_tokens
if total_tokens > self.token_budget:
raise TimeoutError("Token expenditure budget exceeded.")
# Termination condition: LLM decided it has enough evidence
if not msg.tool_calls:
return {"status": "SUCCESS", "answer": msg.content, "steps": step + 1, "tokens": total_tokens}
# Execute tool invocations
for call in msg.tool_calls:
result = await self.registry.dispatch(call.function.name, call.function.arguments)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result)
})
return {"status": "EXHAUSTED", "error": "Maximum reasoning steps reached without convergence."}Never execute arbitrary agent-generated commands or database mutations without a human approval checkpoint or a sandboxed ephemeral runtime.
Run this lesson's Triton kernels and multi-file code directly in an ephemeral pre-configured PyTorch 2.5 container.