📺
dev.to
Yet another screw-up.
I screw up so often that I actually learned to love the process of screwing up!
Please don’t confuse it with being reckless, though. Think of it as a fast incremental learning process, just like fine-tuning a model.
The Most Recent Screw-Up
At the beginning of this year (2024), I created my first automation using GenAI for prototyping a travel app I used for giving a talk at the AWS User Group Berlin meetup showing case the new Amplify Gen 2 services (* API.
Recently, I needed to automate another process for which GenAI was a good fit. Therefore, I reused the same approach I had before and created the following state machine using AWS Step Functions:
in order to avoid abuse. OpenAI is, of course, no different (especially as it is being explored by so many people around the world right now). Therefore, it greatly restricts your ability to parallelize.
GenAI is generally still very slow considering low latency APIs all around us nowadays, especially if you require more complex models like your model, you will see yourself waiting more than 6 hours for it to complete and eventually failing.
Models like o1 are still
The Fix
As I was, of course, in disbelief there wouldn’t be a better way to achieve what I wanted, I started re-reading the documentation.… to my happy surprise, I see a shiny new this year (2024).
So I wrote the following lambda (in typescript) instead:
CODE
import OpenAI, { toFile } from 'openai';
import { BatchWriteCommand, BatchWriteCommandInput, DynamoDBDocumentClient } from '@aws-sdk/lib-dynamodb';
import { DynamoDBClient, GetItemCommand, QueryCommand } from '@aws-sdk/client-dynamodb';
import { v4 as uuidv4 } from 'uuid';
import fs from 'fs';
import { GenerateTipsEvent } from '../shared/types/tips';
import { FileLike } from 'openai/uploads.mjs';
// using AWS ParamsAndSecretsLayerVersion for secrets caching
const AWS_SECRETS_EXTENTION_SERVER_ENDPOINT = "http://localhost:2773/secretsmanager/get?secretId="
let openai: OpenAI | undefined;
// table name of table used for tracking
// OpenAI batch requests for post processing later on
const BATCH_REQUESTS_TABLE = process.env.BATCH_REQUESTS_TABLE || 'BatchRequests';
// db containing my data used in prompts
const ddbClient = new DynamoDBClient({});
const ddb = DynamoDBDocumentClient.from(ddbClient);
// OpenAI single chat completion request to be grouped in a batch
interface BatchRequestLineItem {
custom_id: string;
method: string;
url: string;
body: {
model: string;
messages: {
role: string;
content: string;
}[];
};
}
// my interface for tracking batch requests
interface BatchRequestInput {
lineItem: BatchRequestLineItem;
// and any other ids/data your use case need to work with later on
// eg. categoryId
}
// lambda entry point
export const handler = async (event: any) => {
try {
const { categoryId, model } = event;
if (!model || !categoryId) {
return {
statusCode: 400,
body: JSON.stringify({ error: 'Missing required parameters' }),
};
}
// as client is reused in warm starts,
// make sure it's initialized
await initOpenAi();
console.log('Generating batch requests');
const batchRequestInputs = await generateBatchRequests(categoryId, model);
if (batchRequestInputs.length === 0) {
return {
statusCode: 404,
body: JSON.stringify({ error: 'No units found' }),
};
}
// in memory files representations used in batch request API
const files = await createBatchFile(batchRequestInputs.map(batchRequestInput => batchRequestInput.lineItem));
for (const file of files) {
console.log('Uploading file ', file.name);
// uploads file to OpenAI
const upload = await openai?.files.create({
purpose: 'batch',
file: file
});
if (!upload) continue;
console.log('File uploaded', JSON.stringify(upload, null, 2));
// creates a new batch in OpenAI for uploaded file
console.log('Creating batch');
const requestedBatch = await openai?.batches.create({
completion_window: '24h',
endpoint: '/v1/chat/completions',
input_file_id: upload.id,
metadata: {
description: `Batch request for generating tips for ${categoryId}`,
categoryId: categoryId,
}
});
console.log('Batch created', JSON.stringify(requestedBatch, null, 2));
if (!requestedBatch) continue;
// stores batch request locally so we can identify
// each batch result and each result row later on post processing
console.log('Storing batch request', batchRequestInputs.length);
storeBatchRequest(batchRequestInputs.map(batchRequestInput => ({
id: requestedBatch.id,
body: JSON.stringify(batchRequestInput.lineItem),
model: model,
type: 'tip',
batchId: upload?.id,
filename: upload?.filename,
bytes: upload?.bytes,
status: 'uploaded',
// and any other data you may need to reference
// your data later (eg: categoryId)
})));
console.log('Batch request stored');
}
return {
statusCode: 200,
body: JSON.stringify({
message: 'Batch request stored',
batchRequestInputsLength: batchRequestInputs.length,
}),
};
} catch (error) {
console.error('Error:', error);
throw error;
}
}
// utility function for slicing dynamodb commands
// in chunks to avoid max sizes limitations
const chunk = <T>(arr: T[], size: number): T[][] => {
return Array.from({ length: Math.ceil(arr.length / size) }, (_, i) =>
arr.slice(i * size, i * size + size)
);
};
async function initOpenAi() {
if (!openai) {
const openAiSecret = JSON.parse(await getSecretValue(process.env.OPEN_AI_SECRET_NAME!))
openai = new OpenAI({
apiKey: openAiSecret.apiKey,
organization: openAiSecret.orgId,
});
}
}
// retrieving cached secret
const getSecretValue = async (secretName: string) => {
const url = `${AWS_SECRETS_EXTENTION_SERVER_ENDPOINT}${secretName}`;
const response = await fetch(url, {
method: "GET",
headers: {
"X-Aws-Parameters-Secrets-Token": process.env.AWS_SESSION_TOKEN!,
},
});
if (!response.ok) {
throw new Error(
`Error occured while requesting secret ${secretName}. Responses status was ${response.status}`
);
}
const secretContent = (await response.json()) as { SecretString: string };
return secretContent.SecretString;
};
// retrieving internal data
const getTopicsForCategory = async (categoryId: string) => {
const result = await ddb.send(new QueryCommand({
TableName: process.env.TOPICS_TABLE,
// all the other boring stuff
}));
return result.Items || [];
};
// retrieving internal data
const getSectionsForTopic = async (topicId: string) => {
const result = await ddb.send(new QueryCommand({
TableName: process.env.SECTIONS_TABLE,
// all the other boring stuff
}));
return result.Items || [];
};
// retrieving internal data
const getUnitsForSection = async (sectionId: string) => {
const result = await ddb.send(new QueryCommand({
TableName: process.env.UNITS_TABLE,
// all the other boring stuff
}));
return result.Items || [];
};
// retrieving internal data
const getCategory = async (categoryId: string) => {
const result = await ddb.send(new GetItemCommand({
TableName: process.env.CATEGORIES_TABLE,
// all the other boring stuff
}));
return result.Item;
};
// create batch requests input params
async function generateBatchRequests(
categoryId: string,
model: string
): Promise<BatchRequestInput[]> {
try {
// Fetch initial data in parallel
const [category, topics] = await Promise.all([
getCategory(categoryId),
getTopicsForCategory(categoryId),
]);
if (!category?.id.S) {
throw new Error(`Category not found: ${categoryId}`);
}
// Process topics sequentially but handle sections in parallel
const batchRequests: BatchRequestInput[] = [];
for (const topic of topics) {
if (!topic.id.S) continue;
const sections = await getSectionsForTopic(topic.id.S);
// Process all sections for this topic in parallel
const sectionPromises = sections.map(async (section) => {
if (!section.id.S) return;
const units = await getUnitsForSection(section.id.S);
// Process all units for this section
return Promise.all(units.map(async (unit) => {
if (!unit.id.S) return;
const event = {
// all your custom data you need to inject in your prompt
};
const customId = uuidv4();
return {
lineItem: {
custom_id: customId,
method: 'POST',
url: '/v1/chat/completions',
body: {
model: model,
messages: [
{ role: 'system', content: getSystemPrompt(event) },
{ role: 'user', content: getUserPrompt(event) }
],
},
},
// other data you may need later on
};
}));
});
// Wait for all sections and their units to be processed
const sectionResults = await Promise.all(sectionPromises);
// Flatten and filter out any undefined values
batchRequests.push(
...sectionResults
.flat()
.filter((request): request is BatchRequestInput => request !== undefined)
);
}
return batchRequests;
} catch (error) {
console.error('Error generating batch requests:', error);
throw error;
}
}
// stores batch request in dynamodb
async function storeBatchRequest(requests: {
id: string;
model: string;
status: string;
type: string;
body: string;
batchId: string | undefined;
filename: string | undefined;
bytes: number | undefined;
customId: string;
// and any other param you need to work on your data later on
}[]) {
const batches = chunk(requests, 25);
for (const batch of batches) {
const batchWriteParams: BatchWriteCommandInput = {
RequestItems: {
[BATCH_REQUESTS_TABLE]: batch.map(batchRequestInput => ({
PutRequest: {
Item: {
...batchRequestInput
}
}
}))
}
};
const command = new BatchWriteCommand(batchWriteParams);
try {
await ddb.send(command);
} catch (error) {
console.error('Error writing batch requests batch:', error);
throw error;
}
}
}
// creates in memory file representations of batch requests of max 200MB
// OpenAI restricts file uploads to 200MB
async function createBatchFile(items: BatchRequestLineItem[], maxSizeMB: number = 200): Promise<FileLike[]> {
const MAX_FILE_SIZE = maxSizeMB * 1024 * 1024; // Convert MB to bytes
const files: FileLike[] = [];
let currentItems: BatchRequestLineItem[] = [];
let currentSize = 0;
let fileIndex = 0;
const createBatchFile = async (items: BatchRequestLineItem[], index: number): Promise<FileLike> => {
// Convert items to JSONL string
const jsonlContent = items
.map(item => JSON.stringify(item))
.join('\n');
// Create file using OpenAI's toFile utility
return await toFile(
new Blob([jsonlContent], { type: 'application/jsonl' }),
`batch_${Date.now()}_${index}.jsonl`
);
};
// Process items and create batches
for (const item of items) {
const line = JSON.stringify(item) + '\n';
const itemSize = Buffer.byteLength(line, 'utf-8');
// Check if adding this item would exceed the size limit
if (currentSize + itemSize > MAX_FILE_SIZE) {
// Create file from current batch
const file = await createBatchFile(currentItems, fileIndex);
files.push(file);
// Reset for next batch
currentItems = [];
currentSize = 0;
fileIndex++;
}
// Add item to current batch
currentItems.push(item);
currentSize += itemSize;
}
// Process remaining items
if (currentItems.length > 0) {
const file = await createBatchFile(currentItems, fileIndex);
files.push(file);
}
return files;
}
const getSystemPrompt = (event: GenerateTipsEvent) => `You are a... rest of prompt`
const getUserPrompt = (event: GenerateTipsEvent) => `Generate... rest of prompt`
This function ran in 4518.0ms to upload one file of 4Mb with 391 prompts of a total of 1,301,368 tokens (in and out).
The processing of those prompts cost about $1, and it took 18 minutes for the batch to complete.
and let’s keep moving things forward!
Vollständiger Original-Bericht
Ausführliche Details, Code-Beispiele & Hersteller-Stellungnahme auf dev.to.
SOCIAL SHARE CARD GENERATOR