We've spent a lot of time building efficient systems for ingesting dynamic feeds, taming I/O bottlenecks with asyncio, and even setting up intelligent agents that can select tools based on context. But the real challenge often begins when you move beyond the "happy path" and deploy these agents into production. How do you ensure that a critical piece of content is processed exactly once, even if your agent crashes or the LLM API temporarily fails? How do you manage concurrent external API calls without hitting rate limits, and how do you recover gracefully when things inevitably go wrong? If you're a developer or data scientist looking to move your Generative AI agents from clever prototypes to robust, production-grade systems that handle real-world volatility, this post is for you. I'll walk you through building an agent that is not just smart, but truly resilient and reliable, using the Discord Engineering Blog RSS feed as our live, dynamic data source.
Key Takeaways
- Idempotency in agent processing is crucial and can be reliably achieved with a simple state ledger (like SQLite) to track processing status.
- Asynchronous tools, integrated with retry mechanisms and circuit breakers, are fundamental for agents interacting with external, potentially unreliable services like LLM APIs.
asyncio.Semaphoreis an indispensable primitive for orchestrating concurrent agent tasks, allowing fine-grained control over resource utilization and preventing rate limits.- A well-designed agent execution loop incorporates feed ingestion, state checks, tool execution, and state updates into a single, fault-tolerant asynchronous flow.
- Self-correction is embedded through stateful retries and explicit status tracking, enabling agents to pick up exactly where they left off after interruptions.
The Problem
The core problem I faced was moving beyond a fire-and-forget approach for processing dynamic content. Our previous explorations focused on getting data *into* the system quickly and efficiently. But what happens when that data needs to be analyzed by a Generative AI agent, and that analysis might take time, involve external API calls, and potentially fail? If we just re-process everything on every run, we waste resources and risk duplicate actions. If we don't handle failures, we lose valuable insights. The goal here is to build an agent that is aware of its own work, can recover from failures, and ensures that each unique piece of content is analyzed exactly once, even when operating on a constantly updating feed.
Data and Sources
For this demonstration, we'll be analyzing entries from the Discord Engineering Blog RSS feed. This provides a live, dynamic stream of content, simulating a real-world scenario where new items appear unpredictably.
- Discord Engineering Blog RSS Feed: https://discord.com/blog/rss.xml
feedparserlibrary documentation: https://pythonhosted.org/feedparser/httpxlibrary documentation: https://www.python-httpx.org/- Python's
sqlite3documentation: https://docs.python.org/3/library/sqlite3.html
Data accessed on 2026-10-27
Step 1 — Re-engaging the Feed: Concurrent & Selective Ingestion
The first step is to reliably fetch the RSS feed and identify new entries. While we've covered concurrent ingestion before, here the focus is on *selective* ingestion, preparing the data for idempotent processing. We'll use httpx for asynchronous HTTP requests and feedparser to parse the RSS XML.
The sub-problem is twofold: efficiently fetching the feed without blocking, and extracting a unique identifier for each entry. The solution involves an asynchronous HTTP client and leveraging feedparser's ability to parse XML from a string, not just a URL. This gives us more control over the network call.
import httpx
import feedparser
import asyncio
import sqlite3
import time
import json
from typing import List, Dict, Any, Optional
DISCORD_RSS_FEED_URL = "https://discord.com/blog/rss.xml"
DB_NAME = "agent_ledger.db"
MAX_LLM_CONCURRENCY = 3 # Limit concurrent LLM calls
LLM_MOCK_DELAY = 5 # seconds to simulate LLM processing time
async def fetch_feed(url: str) -> List[Dict[str, Any]]:
"""Fetches and parses an RSS feed asynchronously."""
async with httpx.AsyncClient(timeout=10) as client:
try:
response = await client.get(url)
response.raise_for_status() # Raise an exception for bad status codes
feed = feedparser.parse(response.text)
entries = []
for entry in feed.entries:
# Use link as a primary identifier, fall back to title if link missing
entry_id = entry.link if hasattr(entry, 'link') else entry.title
entries.append({
"id": entry_id,
"title": entry.title,
"link": entry.link,
"published": entry.published,
"summary": entry.summary
})
print(f"Fetched {len(entries)} entries from {url}")
return entries
except httpx.HTTPStatusError as e:
print(f"HTTP error fetching feed {url}: {e.response.status_code} - {e.response.text}")
return []
except httpx.RequestError as e:
print(f"Network error fetching feed {url}: {e}")
return []
except Exception as e:
print(f"Error parsing feed {url}: {e}")
return []
Here, fetch_feed uses httpx.AsyncClient to get the XML content. Once fetched, feedparser.parse() processes it. Crucially, each entry gets a unique id, typically its link, which will be vital for tracking its processing state. Error handling for network and HTTP status issues is already built in, making the ingestion more robust.
Step 2 — Crafting Asynchronous Tools for Agent Intelligence
Our agent needs tools to perform its work. For this post, we'll simulate a Generative AI tool that "analyzes" content. The sub-problem is to create an asynchronous tool that mimics external API calls, including potential failures and retries, without blocking our main event loop. The solution involves an async function with a simple backoff retry mechanism.
async def mock_llm_analysis_tool(content: str, retries: int = 3, delay: int = 2) -> Optional[str]:
"""
Mocks an asynchronous LLM analysis tool with retries.
Simulates network latency and occasional failures.
"""
for attempt in range(retries):
try:
print(f" [LLM Tool] Analyzing content (attempt {attempt + 1}/{retries})...")
await asyncio.sleep(LLM_MOCK_DELAY) # Simulate LLM API call latency
# Simulate occasional LLM failure for resilience testing
if hash(content) % 7 == 0: # Roughly 1/7 chance of failure
raise ConnectionError("Simulated LLM API transient error")
analysis_result = f"Analyzed content summary for: {content[:50]}..."
print(f" [LLM Tool] Analysis complete for {content[:20]}...")
return analysis_result
except ConnectionError as e:
print(f" [LLM Tool] Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
await asyncio.sleep(delay)
delay *= 2 # Exponential backoff
except Exception as e:
print(f" [LLM Tool] Unexpected error during analysis: {e}")
return None
print(f" [LLM Tool] Failed to analyze content after {retries} attempts.")
return None
The mock_llm_analysis_tool function uses asyncio.sleep to simulate the latency of an LLM API call. I've also introduced a probabilistic failure (hash(content) % 7 == 0) to demonstrate the retry logic. This is critical for real-world scenarios where external services are never 100% reliable. The exponential backoff helps prevent overwhelming a struggling service.
Step 3 — State Management for Idempotent Processing: The SQLite Ledger
Idempotency is paramount: we want to process each unique content item exactly once. The sub-problem is to maintain state across agent runs, tracking which items have been processed, are pending, or have failed. The solution is a simple SQLite database acting as a ledger.
I chose SQLite because it's lightweight, embedded, and perfect for local state management without needing a separate database server. It allows us to persist processing status, making our agent robust to restarts and failures.
def init_db():
"""Initializes the SQLite database for tracking processed items."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS processed_items (
item_id TEXT PRIMARY KEY,
status TEXT NOT NULL,
analysis_result TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
def get_item_status(item_id: str) -> Optional[str]:
"""Retrieves the processing status of an item."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("SELECT status FROM processed_items WHERE item_id = ?", (item_id,))
result = cursor.fetchone()
conn.close()
return result[0] if result else None
def update_item_status(item_id: str, status: str, analysis_result: Optional[str] = None):
"""Updates the processing status of an item."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.
