Update concurrency limits and refactor promise queue handling

docs/src/content/docs/reference/scripts/inline-prompts.mdx

@@ -72,7 +72,7 @@ defTool(
 await Promise.all(env.files => prompt`Summarize the ${file}`)
 ```
 
-Internally, GenAIScript applies a concurrent limit of 5 per model by default. You can change this limit using the `modelConcurrency` option.
+Internally, GenAIScript applies a concurrent limit of 8 per model by default. You can change this limit using the `modelConcurrency` option.
 
 ```js "modelConcurrency"
 script({

packages/core/src/concurrency.ts

@@ -1,6 +1,7 @@
 import pLimit, { LimitFunction } from "p-limit"
 import { runtimeHost } from "./host"
 import { normalizeInt } from "./util"
+import { PROMISE_QUEUE_CONCURRENCY_DEFAULT } from "./constants"
 
 export type ConcurrentLimitFunction = LimitFunction
 
@@ -20,7 +21,23 @@ export function concurrentLimit(
 export class PLimitPromiseQueue implements PromiseQueue {
     private queue: LimitFunction
     constructor(private readonly options: PromiseQueueOptions) {
-        this.queue = pLimit(options?.concurrency ?? 5)
+        this.queue = pLimit(
+            options?.concurrency ?? PROMISE_QUEUE_CONCURRENCY_DEFAULT
+        )
+    }
+
+    async mapAll<T extends unknown, Arguments extends unknown[], ReturnType>(
+        values: T[],
+        fn: (value: T, ...arguments_: Arguments) => Awaitable<ReturnType>,
+        ...arguments_: Arguments
+    ): Promise<ReturnType[]> {
+        return await Promise.all(
+            values.map((value) => this.queue(fn, value, ...arguments_))
+        )
+    }
+
+    async all<T = any>(fns: (() => Awaitable<T>)[]): Promise<T[]> {
+        return await Promise.all(fns.map((fn) => this.queue(fn)))
     }
 
     add<Arguments extends unknown[], ReturnType>(

packages/core/src/constants.ts

@@ -242,4 +242,5 @@ export const OPENAI_RETRY_DEFAULT_DEFAULT = 1000
 export const TEMPLATE_ARG_FILE_MAX_TOKENS = 4000
 export const TEMPLATE_ARG_DATA_SLICE_SAMPLE = 2000
 
-export const CHAT_REQUEST_PER_MODEL_CONCURRENT_LIMIT = 5
+export const CHAT_REQUEST_PER_MODEL_CONCURRENT_LIMIT = 8
+export const PROMISE_QUEUE_CONCURRENCY_DEFAULT = 16

packages/core/src/promptcontext.ts

@@ -225,7 +225,7 @@ export async function createPromptContext(
             await runtimeHost.select(message, options),
         input: async (message) => await runtimeHost.input(message),
         confirm: async (message) => await runtimeHost.confirm(message),
-        pQueue: (options) => new PLimitPromiseQueue(options),
+        promiseQueue: (options) => new PLimitPromiseQueue(options),
     })
 
     const ctx: PromptContext & RunPromptContextNode = {

packages/core/src/types/prompt_template.d.ts

@@ -2284,11 +2284,26 @@ interface PromiseQueue {
      * @param fn
      */
     add<Arguments extends unknown[], ReturnType>(
-        function_: (
-            ...arguments_: Arguments
-        ) => PromiseLike<ReturnType> | ReturnType,
+        function_: (...arguments_: Arguments) => Awaitable<ReturnType>,
         ...arguments_: Arguments
     ): Promise<ReturnType>
+
+    /**
+     * Runs all the functions in the queue with limited concurrency
+     * @param fns
+     */
+    all<T = any>(fns: (() => Awaitable<T>)[]): Promise<T[]>
+
+    /**
+     * Applies a function to all the values in the queue with limited concurrency
+     * @param values 
+     * @param fn 
+     */
+    async mapAll<T extends unknown, Arguments extends unknown[], ReturnType>(
+        values: T[],
+        fn: (value: T, ...arguments_: Arguments) => Awaitable<ReturnType>,
+        ...arguments_: Arguments
+    ): Promise<ReturnType[]>
 }
 
 interface PromiseQueueOptions {
@@ -2303,9 +2318,9 @@ interface PromptHost extends ShellHost {
     container(options?: ContainerOptions): Promise<ContainerHost>
 
     /**
-     * Create a new job promise queue
+     * Create a new promise queue to run async functions with limited concurrency
      */
-    pQueue(options?: PromiseQueueOptions): PromiseQueue
+    promiseQueue(options?: PromiseQueueOptions): PromiseQueue
 }
 
 interface ContainerHost extends ShellHost {