Batch Processing with Agents: A Practical Quick Start Guide

Batch Processing with Agents: A Practical Quick Start Guide

In the rapidly evolving space of artificial intelligence and automation, the ability to process large datasets efficiently is paramount. While individual agent interactions are powerful, many real-world applications demand the coordinated execution of agents across a multitude of inputs. This is where batch processing with agents shines, offering a scalable and solid approach to automating complex tasks. This guide will provide a practical quick start, complete with examples, to help you integrate this powerful paradigm into your workflows.

What is Batch Processing with Agents?

At its core, batch processing with agents involves submitting a collection of tasks or data points to a system where each item is processed independently or semi-independently by an intelligent agent. Instead of a user-agent-user interaction loop for a single task, you define a set of inputs, specify the agent’s behavior, and then let the system run through all inputs, typically in parallel or a highly optimized sequence. The outputs are then collected, aggregated, or further processed.

Key Benefits:

• Scalability: Easily handle millions of data points without manual intervention for each item.
• Efficiency: Optimize resource utilization by processing items concurrently.
• Consistency: Ensure uniform application of logic and rules across all inputs.
• Automation: Free up human resources from repetitive, high-volume tasks.
• Cost-Effectiveness: Often cheaper than real-time, on-demand processing for non-urgent tasks.

When to Use Batch Processing with Agents?

Consider batch processing with agents for scenarios like:

• Document Classification: Categorizing thousands of incoming emails, invoices, or legal documents.
• Data Enrichment: Adding context, sentiment scores, or entity recognition to large datasets.
• Content Generation: Creating multiple product descriptions, social media posts, or article summaries based on various inputs.
• Image Tagging/Analysis: Applying descriptive tags or identifying objects in a large collection of images.
• Code Review/Refactoring Suggestions: Analyzing multiple code files for potential improvements.
• Customer Support Ticket Routing: Automatically assigning tickets to the right department based on their content.

The Core Components of a Batch Agent System

Before exploring examples, let’s understand the essential components:

1. Inputs (The Batch): A collection of data points, often in a structured format (CSV, JSONL, database table, list of URLs).
2. Agent Definition: The core logic, persona, and tools of your agent. This defines what the agent does with each input.
3. Execution Engine: The mechanism that orchestrates the processing. This could be a simple loop, a multiprocessing library, a distributed task queue (e.g., Celery, Apache Kafka), or a cloud-based serverless function orchestrator (e.g., AWS Step Functions, Google Cloud Workflows).
4. Output Collection: A method to gather and store the results of each agent’s execution.
5. Error Handling & Monitoring: Strategies to deal with failures, retry mechanisms, and observability into the batch’s progress.

Quick Start: Practical Examples with Python

We’ll use Python as our language of choice due to its rich ecosystem for AI, data processing, and concurrency. For our agent, we’ll simulate an LLM-powered agent using a simple function, but in a real-world scenario, this would be an actual API call to OpenAI, Anthropic, or a local LLM.

Example 1: Simple Document Summarization (Local Batch)

Let’s say you have a list of articles and you want an agent each one.

import json
import time

# --- 1. Agent Definition (Simulated LLM Agent) ---
# In a real scenario, this would involve an actual LLM API call
def summarize_document_agent(document_text: str) -> str:
 """Simulates an AI agent summarizing a document."""
 print(f"Processing document (first 30 chars): '{document_text[:30]}...' ")
 # Simulate LLM processing time
 time.sleep(0.5) 
 summary = f"SUMMARY: The document discusses topics related to {document_text.split()[2]} and {document_text.split()[-2]}. It is a concise overview."
 return summary

# --- 2. Inputs (The Batch) ---
articles = [
 "The quick brown fox jumps over the lazy dog. This is a classic sentence for testing typography and keyboard layouts. It contains every letter of the alphabet.",
 "Artificial intelligence is transforming industries globally. From healthcare to finance, AI is enhancing efficiency, driving innovation, and creating new opportunities.",
 "Quantum computing represents a major change in computation. useing principles of quantum mechanics, it promises to solve problems intractable for classical computers.",
 "The history of the internet is a fascinating journey from ARPANET to the World Wide Web. It has reshaped communication, commerce, and access to information."
]

# --- 3. Execution Engine (Simple Loop) ---
results = []
for i, article in enumerate(articles):
 print(f"\n--- Processing Article {i+1}/{len(articles)} ---")
 summary = summarize_document_agent(article)
 results.append({"original_text": article, "summary": summary})

# --- 4. Output Collection ---
print("\n--- Batch Processing Complete ---")
for i, result in enumerate(results):
 print(f"Article {i+1} Summary: {result['summary']}")

# Optionally save to JSON
with open("summaries.json", "w") as f:
 json.dump(results, f, indent=2)
print("Results saved to summaries.json")

This basic example demonstrates the fundamental flow: define an agent, prepare inputs, iterate, and collect outputs. However, for larger batches, a simple loop is inefficient.

Example 2: Parallel Processing with multiprocessing

To speed things up, especially for CPU-bound tasks or when dealing with I/O-bound tasks that can be parallelized (like multiple API calls), we can use Python’s multiprocessing module.

import json
import time
from multiprocessing import Pool

# --- 1. Agent Definition (Same as before) ---
def summarize_document_agent(document_text: str) -> str:
 print(f"Processing document (first 30 chars): '{document_text[:30]}...' ")
 time.sleep(0.5) # Simulate LLM processing time
 summary = f"SUMMARY: The document discusses topics related to {document_text.split()[2]} and {document_text.split()[-2]}. It is a concise overview."
 return summary

# --- 2. Inputs (Same as before) ---
articles = [
 "The quick brown fox jumps over the lazy dog. This is a classic sentence for testing typography and keyboard layouts. It contains every letter of the alphabet.",
 "Artificial intelligence is transforming industries globally. From healthcare to finance, AI is enhancing efficiency, driving innovation, and creating new opportunities.",
 "Quantum computing represents a major change in computation. useing principles of quantum mechanics, it promises to solve problems intractable for classical computers.",
 "The history of the internet is a fascinating journey from ARPANET to the World Wide Web. It has reshaped communication, commerce, and access to information.",
 "Machine learning, a subset of AI, focuses on algorithms that allow systems to learn from data. Supervised, unsupervised, and reinforcement learning are key paradigms.",
 "Natural Language Processing (NLP) enables computers to understand, interpret, and generate human language. It's crucial for chatbots, translation, and sentiment analysis."
]

# --- 3. Execution Engine (Multiprocessing Pool) ---
print("\n--- Starting Parallel Batch Processing ---")
start_time = time.time()

# Use a Pool to distribute tasks across CPU cores
# The number of processes can be adjusted, often min(CPU_COUNT, len(articles)) or a fixed number
with Pool(processes=4) as pool: 
 # The map function applies summarize_document_agent to each item in 'articles'
 # It blocks until all results are ready
 summaries = pool.map(summarize_document_agent, articles)

results = []
for i, article in enumerate(articles):
 results.append({"original_text": article, "summary": summaries[i]})

end_time = time.time()
print(f"\n--- Parallel Batch Processing Complete in {end_time - start_time:.2f} seconds ---")

# --- 4. Output Collection ---
for i, result in enumerate(results):
 print(f"Article {i+1} Summary: {result['summary']}")

with open("parallel_summaries.json", "w") as f:
 json.dump(results, f, indent=2)
print("Results saved to parallel_summaries.json")

You’ll notice a significant speed improvement with multiprocessing.Pool, especially as the number of articles grows. This approach is effective for local execution where your agent’s task is CPU-bound or involves waiting for external resources (like API calls) that can be initiated in parallel.

Example 3: Integrating with an Actual LLM (Conceptual)

Let’s refine our agent to use a real LLM. For this, we’ll use a placeholder for an API call, assuming you have an API key set up (e.g., OPENAI_API_KEY).

# This is conceptual. Replace with actual API integration.
import os
# from openai import OpenAI # Uncomment if using OpenAI
import json
import time
from multiprocessing import Pool

# client = OpenAI() # Uncomment if using OpenAI

# --- 1. Agent Definition (Real LLM Agent Structure) ---
def real_llm_summarize_agent(document_text: str) -> str:
 """Agent that calls an actual LLM API for summarization."""
 print(f"Calling LLM for document (first 30 chars): '{document_text[:30]}...' ")
 try:
 # Simulate API call with a delay
 time.sleep(1.0) 

 # --- REAL LLM API CALL EXAMPLE (Uncomment and fill in details) ---
 # response = client.chat.completions.create(
 # model="gpt-3.5-turbo",
 # messages=[
 # {"role": "system", "content": "You are a helpful assistant that summarizes text concisely."},
 # {"role": "user", "content": f"Summarize the following document: {document_text}"}
 # ],
 # temperature=0.7,
 # max_tokens=150
 # )
 # summary = response.choices[0].message.content.strip()
 # --- END REAL LLM API CALL EXAMPLE ---

 # Placeholder summary if not using actual API
 summary = f"[LLM Summary] This document is primarily about {document_text.split()[2]}. It provides an overview of its key points."

 return summary
 except Exception as e:
 print(f"Error summarizing document: {e}")
 return f"ERROR: Could not summarize document due to {e}"

# --- 2. Inputs (Same as before, maybe add more) ---
articles = [
 "The quick brown fox jumps over the lazy dog. This is a classic sentence for testing typography and keyboard layouts. It contains every letter of the alphabet.",
 "Artificial intelligence is transforming industries globally. From healthcare to finance, AI is enhancing efficiency, driving innovation, and creating new opportunities.",
 "Quantum computing represents a major change in computation. useing principles of quantum mechanics, it promises to solve problems intractable for classical computers.",
 "The history of the internet is a fascinating journey from ARPANET to the World Wide Web. It has reshaped communication, commerce, and access to information.",
 "Machine learning, a subset of AI, focuses on algorithms that allow systems to learn from data. Supervised, unsupervised, and reinforcement learning are key paradigms.",
 "Natural Language Processing (NLP) enables computers to understand, interpret, and generate human language. It's crucial for chatbots, translation, and sentiment analysis.",
 "Computer vision allows machines to 'see' and interpret visual data. Applications range from facial recognition to autonomous vehicles and medical image analysis.",
 "Robotics integrates mechanics, electronics, and computer science to design, build, and operate robots. They are used in manufacturing, exploration, and service industries."
]

# --- 3. Execution Engine (Multiprocessing Pool for API calls) ---
print("\n--- Starting Parallel Batch Processing with Simulated LLM API Calls ---")
start_time = time.time()

# For API calls, the bottleneck is often network I/O, not CPU. 
# A Pool still helps manage concurrent requests.
# Be mindful of API rate limits! You might need a lower 'processes' count or add delays.
with Pool(processes=4) as pool: 
 llm_summaries = pool.map(real_llm_summarize_agent, articles)

results_llm = []
for i, article in enumerate(articles):
 results_llm.append({"original_text": article, "llm_summary": llm_summaries[i]})

end_time = time.time()
print(f"\n--- Parallel LLM Batch Processing Complete in {end_time - start_time:.2f} seconds ---")

# --- 4. Output Collection ---
for i, result in enumerate(results_llm):
 print(f"Article {i+1} LLM Summary: {result['llm_summary']}")

with open("llm_batch_summaries.json", "w") as f:
 json.dump(results_llm, f, indent=2)
print("Results saved to llm_batch_summaries.json")

This conceptual example highlights how to structure your agent function for real LLM integration and demonstrates that the multiprocessing.Pool pattern remains valid. When dealing with actual API calls, be extremely mindful of:

• API Rate Limits: Most LLM providers have limits on how many requests you can make per minute or second. You might need to implement a custom rate limiter or use libraries that handle this (e.g., tenacity for retries with exponential backoff).
• Cost: LLM usage is typically billed per token. Batch processing can quickly incur significant costs, so ensure your prompts are efficient.
• Error Handling: Implement solid try-except blocks to catch API errors, network issues, and invalid responses.

Advanced Considerations and Best Practices

• Input/Output Formats: For large batches, consider using JSONL (JSON Lines) for input and output files, as it’s easier to stream and append than a single large JSON array.
• Distributed Systems: For truly massive batches (millions of items) or highly complex agent workflows, explore distributed task queues like Celery with RabbitMQ/Redis, Apache Kafka, or cloud services like AWS Batch, Google Cloud Dataflow, or Azure Functions/Logic Apps.
• Idempotency: Design your agent tasks to be idempotent where possible. This means running the same task multiple times with the same input yields the same result, which simplifies retries and error recovery.
• State Management: If agents need to maintain state across tasks or depend on previous results, you’ll need a more sophisticated orchestration layer.
• Monitoring and Logging: Implement thorough logging for each agent’s execution, including input, output, duration, and any errors. Use metrics to track progress and identify bottlenecks.
• Batch Sizing: The optimal batch size (number of items processed concurrently) depends on your agent’s task, available resources (CPU, RAM, network bandwidth), and external API rate limits. Experiment to find the sweet spot.
• Checkpointing: For very long-running batches, periodically save the progress. If the process crashes, you can resume from the last checkpoint instead of starting over.
• Security: Ensure sensitive data is handled securely, especially when interacting with external APIs or storing outputs.

Conclusion

Batch processing with agents is a powerful paradigm for scaling AI-driven automation. By defining intelligent agents and orchestrating their execution over large datasets, organizations can unlock unprecedented levels of efficiency, consistency, and cost savings. This quick start guide has provided the foundational knowledge and practical Python examples to get you started. As you venture into more complex scenarios, remember to consider parallelization, solid error handling, and the scalability options offered by modern distributed systems to build truly resilient and high-performing agent-based batch pipelines.

🕒 Last updated: March 26, 2026  ·  Originally published: January 17, 2026