From “Prompt and Pray” to Production‑Ready AI Agents Welcome back to Build Log. I’m Nick Creighton, and in this episode we ripped the band‑aid off the llm.call()‑while‑loop approach that’s been haunting developers since AutoGPT hit the headlines. If you’ve ever watched an agent spin its wheels, burn through credits, and then spectacularly crash, you know the feeling of “I built an AI, but it can’t survive a day in production.” In this companion blog post I’m going to translate the podcast’s high‑energy talk into a concrete, step‑by‑step playbook. You’ll get the why behind each failure mode, the what you need to build a resilient framework, and the how—complete with code snippets, tooling recommendations, and monitoring tips. By the time you finish reading, you should be able to replace that naïve loop with a production‑grade agent that runs reliably, cost‑effectively, and with observability you can actually trust. ### Why the Classic “Prompt‑and‑Pray” Loop Breaks Let’s start with the anatomy of the broken pattern: - Single prompt → LLM call → parse → repeat. The loop never knows the difference between “finished” and “stuck”. - No persistent state. Every iteration throws away context, so the agent can’t remember that it already tried a certain tool or that a particular API returned an error. - No error handling. If a tool returns malformed JSON or an unexpected HTTP status, the loop treats it as a successful answer and continues spiralling. - No observability. You have no logs, no metrics, no way to tell why the agent halted or blew up your budget. The result? Infinite loops, hallucinated API responses, and a sudden “out‑of‑credits” error that looks like a random LLM hallucination. In production that translates to downtime, angry users, and an accountant crying over the credit‑card bill. ### Core Pillars of a Reliable Agent Framework Building a production‑ready agent is less about fancy prompting and more about engineering discipline. Below are the five pillars you must reinforce before you let any LLM run unsupervised. 1. Persistent State Management Think of an agent as a state machine. Each step should read from and write to a durable store (Redis, DynamoDB, or even a simple SQLite file) so that you can: - Resume after a crash without re‑executing the same actions. - Detect loops by tracking a hash of recent tool calls. - Audit the decision path after the fact. Example (Python + Redis): import redis, json, uuid r = redis.Redis(host='localhost', port=6379, db=0) def get_session(session_id): raw = r.get(session_id) return json.loads(raw) if raw else {"history": [], "variables": {}} def save_session(session_id, data): r.set(session_id, json.dumps(data)) 2. Robust Error Recovery Every external call (LLM, HTTP API, database) needs a try / except block that can either: - Retry with exponential back‑off, or - Emit a structured error back to the LLM so it can “think” about the failure, or - Escalate to a human operator when the error is unrecoverable. Wrap LLM calls in a helper that normalises errors: def safe_llm_call(prompt, max_retries=3): for attempt in range(max_retries): try: resp = llm.call(prompt) if not resp or "error" in resp.lower(): raise ValueError("Empty or error response") return resp except Exception as e: if attempt == max_retries - 1: raise wait = 2 ** attempt time.sleep(wait) 3. Observability & Telemetry Instrument every logical step: - Metrics: calls per minute, error rates, token usage, loop iterations. - Traces: use OpenTelemetry or a simple request‑ID that propagates through all function calls. - Logs: JSON‑structured logs that include session ID, tool name, LLM prompt, and raw response. Example using structlog: import structlog log = structlog.get_logger() def call_tool(name, args, session_id): log.info("tool_start", tool=name, args=args, session=session_id) try: result = tool_registryname log.info("tool_success", tool=name, result=result, session=session_id) return result except Exception as exc: log.error("tool_failure", tool=name, error=str(exc), session=session_id) raise 4. Cost & Rate Controls Production agents must never run away with your API budget. Implement: - Token caps per session. Abort once a pre‑defined budget is exceeded. - Rate limiting. Use a token bucket algorithm to cap calls per second per user. - Cost alerts. Push a Slack/webhook notification when spend > $X in 24 h. Simple token‑budget guard: MAX_TOKENS = 25_000 def check_budget(used): if used >= MAX_TOKENS: raise RuntimeError("Token budget exhausted") 5. Declarative Tool Contracts Instead of letting the LLM guess JSON schemas on the fly, define strict contracts for every tool: # contracts/tool_schemas.py TOOL_SCHEMAS = { "search_web": { "type": "object", "properties": { "query": {"type": "string"}, "max_results": {"type": "integer", "minimum": 1, "maximum": 10} }, "required": ["query"] }, ... } Validate the LLM’s output against the schema before invoking the real tool. If validation fails, send a corrective prompt back to the model. ### Actionable Blueprint: Building Your First Production Agent Below is a minimal yet production‑ready skeleton in Python. Feel free to swap in your favourite LLM SDK; the pattern stays the same. - Define the tool registry and schemas. - Set up state persistence. - Wrap LLM calls with error handling and token budgeting. - Loop with explicit termination conditions. import json, time, uuid from typing import Dict, Any from redis import Redis from jsonschema import validate, ValidationError # 1️⃣ Tool definitions ------------------------------------------------- def search_web(query: str, max_results: int = 3) -> list: # placeholder – replace with real search client return [{"title": f"Result {i+1}", "url": f"https://example.com/{i}"} for i in range(max_results)] TOOL_REGISTRY = { "search_web": search_web, # add more tools here } # 2️⃣ State persistence ------------------------------------------------ redis_client = Redis() def load_state(session_id: str) -> Dict[str, Any]: raw = redis_client.get(session_id) return json.loads(raw) if raw else {"history": [], "tokens_used": 0, "loop_count": 0} def save_state(session_id: str, state: Dict[str, Any]): redis_client.set(session_id, json.dumps(state)) # 3️⃣ LLM helper ------------------------------------------------------- MAX_TOKENS = 20_000 def llm_prompt(prompt: str) -> str: # Stub – inject your LLM SDK call here # Must return raw string response raise NotImplementedError def safe_llm(prompt: str, state: dict) -> str: resp = llm_prompt(prompt) # Very naive token estimate – replace with exact count if available tokens = len(resp.split()) state["tokens_used"] += tokens check_budget(state["tokens_used"]) return resp def check_budget(used: int): if used > MAX_TOKENS: raise RuntimeError("Budget exceeded") # 4️⃣ Core loop --------------------------------------------------------- def run_agent(initial_goal: str): session_id = str(uuid.uuid4()) state = load_state(session_id) while True: # termination guard – avoid runaway loops if state["loop_count"] >= 20: raise RuntimeError("Loop limit reached") state["loop_count"] += 1 # Build prompt with context system_prompt = f""" You are an AI agent tasked with: {initial_goal} You have access to the following tools: {', '.join(TOOL_REGISTRY.keys())} When you need to call a tool, output JSON matching the tool schema. If you have reached a final answer, respond with {"final": true, "answer": "..."}. Never repeat the same tool call with identical arguments. """ user_prompt = "\n".join(state["history"][-5:]) # recent context only full_prompt = system_prompt + "\n" + user_prompt # 4️⃣ LLM call with safety net try: llm_raw = safe_llm(full_prompt, state) except Exception as e: # log and break out – you could also retry or fallback print(f"LLM failure: {e}") break # Record raw response state["history"].append(llm_raw) # -------------------------------------------------------------- # 5️⃣ Parse response – decide if it's a tool call or final answer # -------------------------------------------------------------- try: parsed = json.loads(llm_raw) except json.JSONDecodeError: # LLM gave non‑JSON; ask it to correct state["history"].append( "Please respond with valid JSON according to the schema." ) continue # Final answer? if parsed.get("final"): print("✅ Final answer:", parsed.get("answer")) break # Tool call handling tool_name = parsed.get("tool") args = parsed.get("args", {}) if tool_name not in TOOL_REGISTRY: state["history"].append(f"Unknown tool '{tool_name}'.") continue # Validate args against schema schema = TOOL_SCHEMAS.get(tool_name) try: validate(instance=args, schema=schema) except ValidationError as ve: state["history"].append( f"Argument validation failed: {ve.message}. Please correct." ) continue # Execute tool try: result = TOOL_REGISTRYtool_name # Feed result back into LLM on next turn state["history"].append(json.dumps({"tool_result": result})) except Exception as exc: state["history"].append( f"Tool execution error: {str(exc)}. Try a different approach." ) continue # Persist state after each iteration save_state(session_id, state) # Clean up (optional) redis_client.delete(session_id) # Example run ------------------------------------------------------------- if name == "main": run_agent("Find the latest open‑source Rust web framework and summarize its core features.") This skeleton hits every pillar: persistent state (Redis), budget checks, schema validation, explicit loop termination, and a clear hand‑off between LLM output and tool execution. ### Tooling & Libraries Worth Your Time - LangChain 2.x – now offers Runnable abstractions that separate prompt generation, LLM call, and tool invocation. Use it if you need a higher‑level DSL. - OpenTelemetry – integrates with most Python loggers and gives you distributed traces across LLM calls, DB queries, and external APIs. - RedisJSON – lets you store and query structured session data without serialising/deserialising on every hit. - jsonschema – lightweight, fast, and perfect for runtime contract enforcement. - Prometheus + Grafana – for real‑time dashboards of token usage, error rates, and loop counts. ### Monitoring in the Wild Observability is useless unless you act on it. Set up alerts for these signals: - Token‑budget breach. Trigger a Slack webhook that also tags the responsible engineer. - Loop‑limit hits. Indicates a logic gap in your prompt or tool contract. - Tool failure spikes. Could be an upstream API outage; auto‑fallback to a cached response. - Latency outliers. If an LLM call exceeds X seconds, you may be hitting rate limits. A sample Prometheus rule for token
This article continues on our podcast...
Top comments (0)