diff --git a/.changeset/smart-frogs-help.md b/.changeset/smart-frogs-help.md new file mode 100644 index 000000000..914b9b7b7 --- /dev/null +++ b/.changeset/smart-frogs-help.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/openai": patch +--- + +Adding additional assistant tasks diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 3e3b7ca89..2027f880d 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -60,6 +60,7 @@ const EnvironmentSchema = z.object({ REDIS_PORT: z.coerce.number().optional(), REDIS_USERNAME: z.string().optional(), REDIS_PASSWORD: z.string().optional(), + REDIS_TLS_DISABLED: z.string().optional(), DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10), DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1), diff --git a/apps/webapp/app/services/runExecutionRateLimiter.server.ts b/apps/webapp/app/services/runExecutionRateLimiter.server.ts index b482c5c50..cf63c2c97 100644 --- a/apps/webapp/app/services/runExecutionRateLimiter.server.ts +++ b/apps/webapp/app/services/runExecutionRateLimiter.server.ts @@ -397,7 +397,7 @@ function getRateLimiter() { username: env.REDIS_USERNAME, password: env.REDIS_PASSWORD, enableAutoPipelining: true, - tls: {} + ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }) }, defaultConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT, }); diff --git a/docs/integrations/apis/openai/assistants.mdx b/docs/integrations/apis/openai/assistants.mdx index 73cdb48d5..7630b2db6 100644 --- a/docs/integrations/apis/openai/assistants.mdx +++ b/docs/integrations/apis/openai/assistants.mdx @@ -33,6 +33,54 @@ const assistant = await io.openai.beta.assistants.create("create-assistant", { }); ``` +### `update()` + +Update an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/modifyAssistant) + +```ts example.ts +const file = await io.openai.files.createAndWaitForProcessing("upload-file", { + purpose: "assistants", + file: fs.createReadStream("./fixtures/mydata.csv"), +}); + +const assistantId = "asst_abc123"; + +const assistant = await io.openai.beta.assistants.update("update-assistant", assistantId, { + file_ids: [file.id], // add a file to the assistant +}); +``` + +### `list()` + +List assistants. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/listAssistants) + +```ts example.ts +const assistants = await io.openai.beta.assistants.list("list"); + +// with pagination +const assistants = await io.openai.beta.assistants.list("list", { + limit: 10, + order: "desc", + after: "asst_abc123", +}); +``` + +### `retrieve()` + +Retrieve an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/getAssistant) + +```ts example.ts +const assistant = await io.openai.beta.assistants.retrieve("get-assistant", "asst_abc123"); +``` + +### `del()` + +Delete an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/deleteAssistant) + +```ts example.ts +const deletedAssistant = await io.openai.beta.assistants.del("delete-assistant", "asst_abc123"); +``` + ## Threads Create threads that assistants can interact with. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/threads/createThread) @@ -132,12 +180,26 @@ Create messages within threads. [Official OpenAI docs](https://platform.openai.c ### `list()` -List all messages in a thread. +List messages in a thread. ```ts example.ts const messages = await io.openai.beta.threads.messages.list("list-messages", "thread_abc123"); +// with pagination +const messages = await io.openai.beta.threads.messages.list("list-messages", "thread_abc123", { + limit: 10, + order: "desc", + after: "message_abc123", +}); ``` +If you want to list all messages in a thread, you can use the `listAll()` helper: + +```ts example.ts +const messages = await io.openai.beta.threads.messages.listAll("list-messages", "thread_abc123"); +``` + +This will automatically paginate through all messages in the thread and return them as a single array. + ### `create()` Create a message. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/messages/createMessage) diff --git a/integrations/openai/src/assistants.ts b/integrations/openai/src/assistants.ts index c0a25cd15..1beeb058c 100644 --- a/integrations/openai/src/assistants.ts +++ b/integrations/openai/src/assistants.ts @@ -2,13 +2,13 @@ import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk"; import { OpenAIRunTask } from "./index"; import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types"; import OpenAI from "openai"; -import { createTaskOutputProperties, handleOpenAIError } from "./taskUtils"; +import { createTaskOutputProperties, handleOpenAIError, isRequestOptions } from "./taskUtils"; export class Assistants { constructor( private runTask: OpenAIRunTask, private options: OpenAIIntegrationOptions - ) {} + ) { } async create( key: IntegrationTaskKey, @@ -54,4 +54,173 @@ export class Assistants { handleOpenAIError ); } + + async update( + key: IntegrationTaskKey, + id: string, + params: Prettify, + options: OpenAIRequestOptions = {} + ): Promise { + return this.runTask( + key, + async (client, task) => { + const { data, response } = await client.beta.assistants + .update(id, params, { + idempotencyKey: task.idempotencyKey, + ...options, + }) + .withResponse(); + + const outputProperties = createTaskOutputProperties(undefined, response.headers); + + task.outputProperties = [ + ...(outputProperties ?? []), + { + label: "assistantId", + text: data.id, + }, + ]; + + return data; + }, + { + name: "Update Assistant", + params, + properties: [ + ...(params.model ? [{ label: "model", text: params.model }] : []), + ...(params.name ? [{ label: "name", text: params.name }] : []), + ...(params.file_ids && params.file_ids.length > 0 + ? [{ label: "files", text: params.file_ids.join(", ") }] + : []), + ], + }, + handleOpenAIError + ); + } + + list( + key: IntegrationTaskKey, + params?: Prettify, + options?: OpenAIRequestOptions, + ): Promise; + list( + key: IntegrationTaskKey, + options?: OpenAIRequestOptions, + ): Promise; + async list( + key: IntegrationTaskKey, + params: Prettify | OpenAIRequestOptions = {}, + options: OpenAIRequestOptions | undefined = undefined + ): Promise { + return this.runTask( + key, + async (client, task) => { + if (isRequestOptions(params)) { + const { data, response } = await client.beta.assistants + .list({ + idempotencyKey: task.idempotencyKey, + ...params, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return data.data; + } + + const { data, response } = await client.beta.assistants + .list(params, { + idempotencyKey: task.idempotencyKey, + ...options, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return data.data; + + }, + { + name: "List Assistants", + params, + properties: !isRequestOptions(params) ? [ + ...(params.before ? [{ label: "before", text: params.before }] : []), + ...(params.order ? [{ label: "order", text: params.order }] : []), + ...(params.after ? [{ label: "after", text: params.after }] : []), + ...(params.limit ? [{ label: "limit", text: String(params.limit) }] : []), + ] : [], + }, + handleOpenAIError + ); + } + + async del( + key: IntegrationTaskKey, + id: string, + options: OpenAIRequestOptions = {} + ): Promise { + return this.runTask( + key, + async (client, task) => { + const { data, response } = await client.beta.assistants + .del(id, { + idempotencyKey: task.idempotencyKey, + ...options, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return data; + }, + { + name: "Delete Assistant", + params: { + id, + }, + properties: [ + { + label: "assistantId", + text: id, + }, + ], + }, + handleOpenAIError + ); + } + + async retrieve( + key: IntegrationTaskKey, + id: string, + options: OpenAIRequestOptions = {} + ): Promise { + return this.runTask( + key, + async (client, task) => { + const { data, response } = await client.beta.assistants + .retrieve(id, { + idempotencyKey: task.idempotencyKey, + ...options, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return data; + }, + { + name: "Retrieve Assistant", + params: { + id, + }, + properties: [ + { + label: "assistantId", + text: id, + }, + ], + }, + handleOpenAIError + ); + } } diff --git a/integrations/openai/src/taskUtils.ts b/integrations/openai/src/taskUtils.ts index 88b50a465..634140039 100644 --- a/integrations/openai/src/taskUtils.ts +++ b/integrations/openai/src/taskUtils.ts @@ -60,11 +60,11 @@ function createTaskUsageProperties( }, ...("completion_tokens" in usage ? [ - { - label: "Completion Usage", - text: String(usage.completion_tokens), - }, - ] + { + label: "Completion Usage", + text: String(usage.completion_tokens), + }, + ] : []), ]; } @@ -83,35 +83,35 @@ function createTaskRateLimitProperties(headers: Headers | undefined) { return [ ...(remainingRequests ? [ - { - label: "Remaining Requests", - text: remainingRequests ?? "Unknown", - }, - ] + { + label: "Remaining Requests", + text: remainingRequests ?? "Unknown", + }, + ] : []), ...(resetRequests ? [ - { - label: "Reset Requests", - text: resetRequests ?? "Unknown", - }, - ] + { + label: "Reset Requests", + text: resetRequests ?? "Unknown", + }, + ] : []), ...(remainingTokens ? [ - { - label: "Remaining Tokens", - text: remainingTokens ?? "Unknown", - }, - ] + { + label: "Remaining Tokens", + text: remainingTokens ?? "Unknown", + }, + ] : []), ...(resetTokens ? [ - { - label: "Reset Tokens", - text: resetTokens ?? "Unknown", - }, - ] + { + label: "Reset Tokens", + text: resetTokens ?? "Unknown", + }, + ] : []), ]; } @@ -282,3 +282,32 @@ export const backgroundTaskRetries: FetchRetryOptions = { randomize: true, }, }; + +type KeysEnum = { [P in keyof Required]: true }; + +const requestOptionsKeys: KeysEnum = { + method: true, + path: true, + query: true, + headers: true, + idempotencyKey: true, +}; + +export const isRequestOptions = (obj: unknown): obj is OpenAIRequestOptions => { + return ( + typeof obj === 'object' && + obj !== null && + !isEmptyObj(obj) && + Object.keys(obj).every((k) => hasOwn(requestOptionsKeys, k)) + ); +}; + +function isEmptyObj(obj: Object | null | undefined): boolean { + if (!obj) return true; + for (const _k in obj) return false; + return true; +} + +function hasOwn(obj: Object, key: string): boolean { + return Object.prototype.hasOwnProperty.call(obj, key); +} \ No newline at end of file diff --git a/integrations/openai/src/threads.ts b/integrations/openai/src/threads.ts index 537fd2c59..a22e7d635 100644 --- a/integrations/openai/src/threads.ts +++ b/integrations/openai/src/threads.ts @@ -7,6 +7,7 @@ import { createBackgroundFetchUrl, createTaskOutputProperties, handleOpenAIError, + isRequestOptions, } from "./taskUtils"; import { RunSubmitToolOutputsParams } from "openai/resources/beta/threads/runs/runs"; import { ThreadUpdateParams } from "openai/resources/beta/threads/threads"; @@ -15,7 +16,7 @@ export class Threads { constructor( private runTask: OpenAIRunTask, private options: OpenAIIntegrationOptions - ) {} + ) { } /** * Create a thread and run it in one task. @@ -261,7 +262,7 @@ class Runs { constructor( private runTask: OpenAIRunTask, private options: OpenAIIntegrationOptions - ) {} + ) { } /** * Creates a run and waits for it to complete by polling in the background. @@ -551,15 +552,70 @@ class Messages { constructor( private runTask: OpenAIRunTask, private options: OpenAIIntegrationOptions - ) {} + ) { } + + /** + * Returns messages for a given thread. + */ + list( + key: IntegrationTaskKey, + threadId: string, + params?: Prettify, + options?: OpenAIRequestOptions + ): Promise + list( + key: IntegrationTaskKey, + threadId: string, + options?: OpenAIRequestOptions + ): Promise + async list( + key: IntegrationTaskKey, + threadId: string, + params: Prettify | OpenAIRequestOptions = {}, + options: OpenAIRequestOptions | undefined = undefined + ): Promise { + return this.runTask( + key, + async (client, task, io) => { + if (isRequestOptions(params)) { + const { data: page, response } = await client.beta.threads.messages + .list(threadId, { + idempotencyKey: task.idempotencyKey, + ...params, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return page.data; + } + + const { data: page, response } = await client.beta.threads.messages + .list(threadId, params, { + idempotencyKey: task.idempotencyKey, + ...options, + }) + .withResponse(); + + task.outputProperties = createTaskOutputProperties(undefined, response.headers); + + return page.data; + }, + { + name: "List Messages", + properties: [{ label: "threadId", text: threadId }], + }, + handleOpenAIError + ); + } /** * Returns all messages for a given thread. */ - async list( + async listAll( key: IntegrationTaskKey, threadId: string, - options: OpenAIRequestOptions = {} + options: OpenAIRequestOptions = {}, ): Promise { return this.runTask( key, @@ -573,8 +629,8 @@ class Messages { const allMessages = []; - for await (const fineTuningJob of page) { - allMessages.push(fineTuningJob); + for await (const message of page) { + allMessages.push(message); } task.outputProperties = createTaskOutputProperties(undefined, response.headers); @@ -582,7 +638,7 @@ class Messages { return allMessages; }, { - name: "List Messages", + name: "List All Messages", properties: [{ label: "threadId", text: threadId }], }, handleOpenAIError diff --git a/integrations/openai/src/types.ts b/integrations/openai/src/types.ts index fabc24d71..2a060d873 100644 --- a/integrations/openai/src/types.ts +++ b/integrations/openai/src/types.ts @@ -34,4 +34,4 @@ export type OpenAIRequestOptions = { path?: string; headers?: OpenAIHeaders; idempotencyKey?: string; -}; +}; \ No newline at end of file diff --git a/references/job-catalog/src/openai.ts b/references/job-catalog/src/openai.ts index f8a53f9c2..c2b2c1405 100644 --- a/references/job-catalog/src/openai.ts +++ b/references/job-catalog/src/openai.ts @@ -228,6 +228,37 @@ client.defineJob({ }, }); +client.defineJob({ + id: "openai-manage-assistant", + name: "OpenAI GPT Manage Assistant", + version: "0.0.1", + trigger: invokeTrigger({ + schema: z.object({ + assistantId: z.string().optional(), + }), + }), + integrations: { + openai, + }, + run: async (payload, io, ctx) => { + const assistants = await io.openai.beta.assistants.list("list", { + limit: 10 + }); + + if (payload.assistantId) { + await io.openai.beta.assistants.retrieve("retrieve", payload.assistantId); + await io.openai.beta.assistants.update("update", payload.assistantId, { + name: "Updated name", + }); + await io.openai.beta.assistants.del("delete", payload.assistantId); + } + + for (const assistant of assistants) { + await io.openai.beta.assistants.del(`delete ${assistant.id}`, assistant.id); + } + }, +}); + client.defineJob({ id: "openai-use-assistant", name: "OpenAI GPT Use Assistant",