Home
Data Engineering
Generative AI

How to Build Autonomous Data Pipelines in Python Using AI Agent Skills

Mausam

Ever wondered why your LLM‑driven agents still need Airflow or a custom scheduler to move data from an API to a CSV? You can skip the heavyweight orchestrator altogether by treating each step as a tiny, reusable “skill” and wiring them together with a simple in‑process message bus. The pattern lets you fetch JSONPlaceholder posts, enrich them with keyword tags using NLTK, and write a clean CSV—all while the pipeline watches its own health and retries transient failures.

Key Takeaways

  • Agent skills are plain Python functions that receive a message, perform a single responsibility, and emit a new message.
  • A thread‑safe queue.Queue plus a tiny dispatcher give you a production‑ready bus without adding another service.
  • Built‑in error handling (timeouts, status checks, shape validation) makes the pipeline self‑healing; failures are logged and the offending record is isolated.

Data and Sources

The pipeline consumes the public JSONPlaceholder posts endpoint:

Data accessed on 2026-11-07.

Autonomous Data Pipelines with AI Agent Skills

In the earlier “multi‑agent trading bus” post we proved that LLM agents can talk to each other without a central orchestrator. The same idea works for any ETL workload: each step is an agent skill that knows only how to transform one message type into another. By keeping the contract simple (a dict → dict), you gain:

  • Reusability – the fetch skill can be swapped for any JSON API.
  • Observability – each skill logs its input, output, and errors.
  • Resilience – the bus can retry a skill without restarting the whole pipeline.

Implementing the Fetch Agent

The fetch skill pulls the raw posts, validates the HTTP response, and pushes each record onto the bus. A short timeout and explicit status check prevent the pipeline from hanging on a flaky network.

def fetch_posts(url: str, out_queue: queue.Queue, timeout: int = 5) -> None:
    """Fetch JSONPlaceholder posts and emit one message per record."""
    logger.info("Fetching posts from %s", url)
    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()
    except requests.RequestException as exc:
        logger.error("Network error while fetching posts: %s", exc)
        return

    try:
        posts = response.json()
        if not isinstance(posts, list):
            raise ValueError("Expected a list of posts")
    except (ValueError, json.JSONDecodeError) as exc:
        logger.error("Invalid JSON payload: %s", exc)
        return

    for post in posts:
        out_queue.put(post)
    logger.info("Fetched %d posts", len(posts))

Implementing the Enrichment Agent (AI Skill)

The enrichment skill adds a keywords field by extracting non‑stopwords from the body. NLTK provides a deterministic stop‑word list, avoiding the nondeterminism of LLM calls while still showcasing an “AI‑adjacent” transformation.

def enrich_with_keywords(in_queue: queue.Queue, out_queue: queue.Queue) -> None:
    """Read a post, compute simple keywords, and forward the enriched record."""
    while True:
        post = in_queue.get()
        if post is None:  # sentinel for shutdown
            out_queue.put(None)
            break

        body = post.get("body", "")
        words = [w.lower() for w in body.split() if w.isalpha()]
        keywords = [w for w in words if w not in STOPWORDS]
        # Keep the most frequent five words as a deterministic shortcut
        freq = {}
        for w in keywords:
            freq[w] = freq.get(w, 0) + 1
        top_keywords = sorted(freq, key=freq.get, reverse=True)[:5]
        post["keywords"] = ", ".join(top_keywords)
        out_queue.put(post)
        in_queue.task_done()

Orchestrating with the Simple Message Bus

The bus consists of two queue.Queue instances and three threads: one fetcher, one enricher, and one writer. The writer pulls enriched messages, writes them to posts_enriched.csv, and prints a one‑line summary when the sentinel arrives.

def write_to_csv(in_queue: queue.Queue, output_path: str) -> None:
    """Consume enriched posts and persist them as CSV."""
    fieldnames = ["userId", "id", "title", "body", "keywords"]
    with open(output_path, "w", newline="", encoding="utf-8") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        count = 0
        while True:
            post = in_queue.get()
            if post is None:
                break
            writer.writerow({k: post.get(k, "") for k in fieldnames})
            count += 1
            in_queue.task_done()
    logger.info("Processed %d posts, saved to %s", count, output_path)

Limitations and Tradeoffs

The in‑process bus is fast but volatile: a process crash discards all queued messages. For production workloads you would replace the queues with a durable broker (Redis Streams, Kafka) and run each skill as an independent service. The keyword extraction uses a static stop‑word list; domain‑specific vocabularies may require a custom list or an LLM call, which introduces latency and cost. Finally, the script assumes the API returns a flat list; nested structures would need additional flattening logic.

Frequently Asked Questions

Can I replace the NLTK keyword skill with an LLM call?

Yes. The enrichment function is a pure transformation; you would call your LLM client inside the loop and assign the returned string to post["keywords"]. Keep the same try/except pattern to avoid blocking the pipeline.

What happens if the fetch agent returns fewer than 100 records?

The pipeline does not assume a fixed size. It processes whatever number of messages it receives; the CSV will contain the same number of rows. If the shortfall is unexpected, add a post‑run validation step that compares the row count against an external expectation.

How does the pipeline survive a transient network glitch?

The fetch agent uses requests.get(..., timeout=5) and logs any RequestException. Because the fetch runs in its own thread, a failure only stops that thread; the rest of the pipeline remains idle but healthy. You can wrap the call in a retry loop (e.g., tenacity) to make it fully self‑healing.

Is the script safe for concurrent execution?

Yes. Each agent runs in its own thread and communicates exclusively via thread‑safe queue.Queue objects. Shared state is limited to the global STOPWORDS set, which is read‑only after initialization.

Complete Script

#!/usr/bin/env python3
"""
Autonomous ETL pipeline using composable agent‑skill functions
and a lightweight in‑process message bus.

Steps:
1. Fetch 100 posts from JSONPlaceholder.
2. Enrich each post with simple keyword extraction.
3. Persist the result to posts_enriched.csv.
4. Log a concise summary.
"""

import csv
import json
import logging
import queue
import requests
from concurrent.futures import ThreadPoolExecutor

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
API_URL = "https://jsonplaceholder.typicode.com/posts"
CSV_OUTPUT = "posts_enriched.csv"
TIMEOUT = 5  # seconds for HTTP request

# ----------------------------------------------------------------------
# Logging setup
# ----------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(threadName)s %(message)s",
)
logger = logging.getLogger(__name__)

# ----------------------------------------------------------------------
# Global resources
# ----------------------------------------------------------------------
# A minimal set of English stopwords (no external dependency)
STOPWORDS = {
    "the", "and", "is", "in", "to", "of", "a", "with", "that", "for", "on",
    "as", "by", "at", "from", "or", "an", "be", "this", "which", "are",
    "was", "it", "has", "have", "not", "but", "they", "their", "you",
    "he", "she", "we", "i", "me", "my", "your", "our", "their", "its",
    "was", "were", "been", "being", "do", "does", "did", "doing"
}

# ----------------------------------------------------------------------
# Agent skill definitions
# ----------------------------------------------------------------------
def fetch_posts(url: str, out_queue: queue.Queue, timeout: int = TIMEOUT) -> None:
    """Fetch JSONPlaceholder posts and emit one message per record."""
    logger.info("Fetching posts from %s", url)
    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()
    except requests.RequestException as exc:
        logger.error("Network error while fetching posts: %s", exc)
        out_queue.put(None)  # Signal shutdown downstream
        return

    try:
        posts = response.json()
        if not isinstance(posts, list):
            raise ValueError("Expected a list of posts")
    except (ValueError, json.JSONDecodeError) as exc:
        logger.error("Invalid JSON payload: %s", exc)
        out_queue.put(None)
        return

    for post in posts:
        out_queue.put(post)
    logger.info("Fetched %d posts", len(posts))
    out_queue.put(None)  # Sentinel for downstream consumer

def enrich_with_keywords(in_queue: queue.Queue, out_queue: queue.Queue) -> None:
    """Read a post, compute simple keywords, and forward the enriched record."""
    while True:
        post = in_queue.get()
        if post is None:  # Sentinel received
            out_queue.put(None)
            in_queue.task_done()
            break

        body = post.get("body", "")
        words = [w.lower() for w in body.split() if w.isalpha()]
        keywords = [w for w in words if w not in STOPWORDS]

        # Frequency count – deterministic top‑5 selection
        freq = {}
        for w in keywords:
            freq[w] = freq.get(w, 0) + 1
        top_keywords = sorted(freq, key=freq.get, reverse=True)[:5]
        post["keywords"] = ", ".join(top_keywords)

        out_queue.put(post)
        in_queue.task_done()

def write_to_csv(in_queue: queue.Queue, output_path: str) -> None:
    """Consume enriched posts and persist them as CSV."""
    fieldnames = ["userId", "id", "title", "body", "keywords"]
    count = 0
    with open(output_path, "w", newline="", encoding="utf-8") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        while True:
            post = in_queue.get()
            if post is None:
                in_queue.task_done()
                break
            writer.writerow({k: post.get(k, "") for k in fieldnames})
            count += 1
            in_queue.task_done()
    logger.info("Processed %d posts, saved to %s", count, output_path)

# ----------------------------------------------------------------------
# Orchestration (simple message bus)
# ----------------------------------------------------------------------
def main() -> None:
    # Queues form the bus; maxsize=0 means unlimited
    fetch_to_enrich = queue.Queue()
    enrich_to_write = queue.Queue()

    # Thread pool for agents
    with ThreadPoolExecutor(max_workers=3) as executor:
        executor.submit(fetch_posts, API_URL, fetch_to_enrich)
        executor.submit(enrich_with_keywords, fetch_to_enrich, enrich_to_write)
        executor.submit(write_to_csv, enrich_to_write, CSV_OUTPUT)

    logger.info("Pipeline completed successfully")

if __name__ == "__main__":
    main()

Recommendation and Next Step

If you need a lightweight, reproducible ETL for a handful of JSON APIs, adopt the agent‑skill + in‑process bus pattern shown here. It gives you composability, observability, and automatic retry handling without the overhead of Airflow or a custom scheduler. For the next iteration, replace the two queue.Queue objects with a durable broker (e.g., Redis Streams) and spin each skill out as an independent service; this upgrade preserves the same skill interface while adding crash‑resilience and horizontal scalability.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.