Batching processes accumulated records in groups instead of handling each one as it arrives. It can reduce repeated-call overhead or support scheduled analysis of a selected cohort. section 5.1.1 section 11.3.3
Financial data workflows use batching in several ways:
- Merge daily or weekly banking data for risk modeling, with some workflows combining near-real-time updates and a daily batch. section 5.1.1
- Analyze a daily cohort of high-churn-risk users, then aggregate recommendations for operational review. section 14.3.5
- Send news texts to an embedding API in groups of 100 rather than as thousands of individual requests. section 11.3.3
Listing 11.7 Batch embedding with the OpenAI API
from openai import OpenAI
from tqdm import tqdm
df = pd.read_parquet('news_ready_for_embedding.parquet')
client = OpenAI(api_key="your-api-key")
print("="*80)
print("STEP 2: TEXT EMBEDDING")
print("="*80)
print(f"Total texts: {len(df)}")
texts = df['text_for_embedding'].tolist()
embeddings = []
batch_size = 100
for i in tqdm(range(0, len(texts), batch_size), desc="Embedding"):
batch = texts[i:i+batch_size]
response = client.embeddings.create(
model="text-embedding-3-small",
input=batch
)
embeddings.extend([item.embedding for item in response.data])
df['embedding'] = embeddings
df.to_parquet('news_embedded.parquet')
print(f"\nSaved: news_embedded.parquet")
if embeddings:
print(f"Embedding dimension: {len(embeddings[0])}")
print(f"Sample embedding (first 5 dims): {embeddings[0][:5]}")
else:
print("No embeddings were generated (empty input).")
See a Python example that batches news texts for embedding, collects vectors, and saves the results to Parquet. section 11.3.3
Listing 14.13 Batch analysis function
def analyze_batch(agent, users_df, max_users=10):
"""Analyze multiple users and aggregate results"""
results = []
for idx, (_, user) in enumerate(users_df.head(max_users).iterrows()):
print(f" [{idx+1}/{max_users}] {user['user_id'][:8]}... ", end="")
try:
result = agent.analyze_user(user['user_id'], user['churn_score'])
results.append(result)
action = result.get('recommended_action', 'N/A')
print(f"Done - {action}")
except Exception as e:
print(f"Error: {e}")
results.append({"user_id": user['user_id'], "status": "error"})
return results
batch_results = analyze_batch(
agent,
high_risk_users.nlargest(10, 'churn_score'),
max_users=10
)
See a cohort-analysis function that processes up to ten high-risk users, records errors, and returns results. section 14.3.5
