切換語言為:簡體

OpenAI API結構化輸出解析與流式響應功能詳解

  • 爱糖宝
  • 2024-09-13
  • 2096
  • 0
  • 0

隨著人工智慧技術的不斷髮展,OpenAI的API當中又除了很多有意思的功能。這些 API 不僅可以生成自然語言文字,還能解析結構化資料、進行實時互動和自動化函式呼叫。在本文中,我們將深入探討 OpenAI API 的結構化輸出解析助手和流式響應功能,幫助開發者更高效地利用這些工具。

結構化輸出

OpenAI API 支援透過 response_format 請求引數從模型中提取 JSON 響應。爲了更方便地處理這些響應,OpenAI SDK 提供了一個 client.beta.chat.completions.parse() 方法。這是一個包裝在 client.chat.completions.create() 之上的方法,返回一個 ParsedChatCompletion 物件。

使用 Zod 模式自動解析響應內容

藉助 Zod 模式,可以自動解析和驗證 JSON 響應內容。以下是一個示例:

import { zodResponseFormat } from 'openai/helpers/zod';
import OpenAI from 'openai/index';
import { z } from 'zod';

const Step = z.object({
  explanation: z.string(),
  output: z.string(),
});

const MathResponse = z.object({
  steps: z.array(Step),
  final_answer: z.string(),
});

const client = new OpenAI();

const completion = await client.beta.chat.completions.parse({
  model: 'gpt-4o-2024-08-06',
  messages: [
    { role: 'system', content: '你是一個有幫助的數學老師。' },
    { role: 'user', content: '求解 8x + 31 = 2' },
  ],
  response_format: zodResponseFormat(MathResponse, 'math_response'),
});

const message = completion.choices[0]?.message;
if (message?.parsed) {
  console.log(message.parsed.steps);
  console.log(`答案: ${message.parsed.final_answer}`);
}

在這個示例中,我們定義了一個數學問題的響應格式,並利用 Zod 模式進行解析和驗證。

自動解析函式工具呼叫

如果你使用 zodFunctionTool() 輔助方法,並將工具模式標記為 "strict": true,.parse() 方法將自動解析函式工具呼叫。以下是一個示例:

import { zodFunction } from 'openai/helpers/zod';
import OpenAI from 'openai/index';
import { z } from 'zod';

const Table = z.enum(['orders', 'customers', 'products']);
const Column = z.enum(['id', 'status', 'expected_delivery_date', 'delivered_at', 'shipped_at', 'ordered_at', 'canceled_at']);
const Operator = z.enum(['=', '>', '<', '<=', '>=', '!=']);
const OrderBy = z.enum(['asc', 'desc']);
const DynamicValue = z.object({ column_name: z.string() });
const Condition = z.object({
  column: z.string(),
  operator: Operator,
  value: z.union([z.string(), z.number(), DynamicValue]),
});
const Query = z.object({
  table_name: Table,
  columns: z.array(Column),
  conditions: z.array(Condition),
  order_by: OrderBy,
});

const client = new OpenAI();
const completion = await client.beta.chat.completions.parse({
  model: 'gpt-4o-2024-08-06',
  messages: [
    {
      role: 'system',
      content: '你是一個有幫助的助手。今天是2024年8月6日。你幫助使用者查詢他們需要的資料,透過呼叫查詢函式。',
    },
    {
      role: 'user',
      content: '查詢我去年11月所有已完成但未按時交付的訂單',
    },
  ],
  tools: [zodFunction({ name: 'query', parameters: Query })],
});

const toolCall = completion.choices[0]?.message.tool_calls?.[0];
if (toolCall) {
  const args = toolCall.function.parsed_arguments as z.infer<typeof Query>;
  console.log(args);
  console.log(args.table_name);
}

在這個示例中,我們定義了一個查詢函式的引數格式,並利用 Zod 模式進行解析和驗證。

流式輸出

OpenAI 支援在與 Chat 或 Assistant API 互動時流式響應,提供了更高效的實時互動體驗。

Assistant 流式 API

Assistant 流式 API 允許開發者訂閱事件型別並接收累積的響應。以下是一個示例:

const run = openai.beta.threads.runs
  .stream(thread.id, {
    assistant_id: assistant.id,
  })
  .on('textCreated', (text) => process.stdout.write('\n助手 > '))
  .on('textDelta', (textDelta, snapshot) => process.stdout.write(textDelta.value))
  .on('toolCallCreated', (toolCall) => process.stdout.write(`\n助手 > ${toolCall.type}\n\n`))
  .on('toolCallDelta', (toolCallDelta, snapshot) => {
    if (toolCallDelta.type === 'code_interpreter') {
      if (toolCallDelta.code_interpreter.input) {
        process.stdout.write(toolCallDelta.code_interpreter.input);
      }
      if (toolCallDelta.code_interpreter.outputs) {
        process.stdout.write('\n輸出 >\n');
        toolCallDelta.code_interpreter.outputs.forEach((output) => {
          if (output.type === 'logs') {
            process.stdout.write(`\n${output.logs}\n`);
          }
        });
      }
    }
  });

透過這個示例,我們可以看到如何建立一個執行並訂閱多個事件,以便實時處理來自 Assistant 的響應。

開始流

有三種建立流的輔助方法:

  1. openai.beta.threads.runs.stream();

  2. openai.beta.threads.createAndRunStream();

  3. openai.beta.threads.runs.submitToolOutputsStream();

這些方法分別用於啟動並流式響應已填充訊息的現有執行、向執行緒新增訊息並啟動執行、以及向等待輸出的執行提交工具輸出並啟動流。

自動化函式呼叫

openai.chat.completions.runTools() 方法返回一個 Runner,用於使用聊天完成自動化函式呼叫。以下是一個示例:

client.chat.completions.runTools({
  model: 'gpt-3.5-turbo',
  messages: [{ role: 'user', content: '這周的天氣如何?' }],
  tools: [
    {
      type: 'function',
      function: {
        function: getWeather as (args: { location: string; time: Date }) => any,
        parse: parseFunction as (args: strings) => { location: string; time: Date },
        parameters: {
          type: 'object',
          properties: {
            location: { type: 'string' },
            time: { type: 'string', format: 'date-time' },
          },
        },
      },
    },
  ],
});

在這個示例中,我們定義了一個獲取天氣的函式,並透過 runTools 方法自動化函式呼叫。

綜合示例

程式碼

import { zodResponseFormat, zodFunction } from 'openai/helpers/zod';
import OpenAI from 'openai/index';
import { z } from 'zod';

// 定義 OpenAIHelper 類,用於封裝 OpenAI API 的各種功能
class OpenAIHelper {
  private client: OpenAI;

  constructor(apiKey: string) {
    // 初始化 OpenAI 客戶端
    this.client = new OpenAI(apiKey);
  }

  /**
   * 解析結構化響應
   * @param messages - 與模型互動的訊息陣列
   * @param responseSchema - 用於驗證響應的 Zod 模式
   * @returns 解析後的響應資料
   */
  async parseStructuredResponse(messages: any[], responseSchema: z.ZodType<any>) {
    const completion = await this.client.beta.chat.completions.parse({
      model: 'gpt-4o-2024-08-06',
      messages,
      response_format: zodResponseFormat(responseSchema, 'structured_response'),
    });

    const message = completion.choices[0]?.message;
    if (message?.parsed) {
      return message.parsed;
    }
    return null;
  }

  /**
   * 解析數學問題的響應
   * @param messages - 與模型互動的訊息陣列
   * @returns 解析後的數學問題響應資料
   */
  async parseMathResponse(messages: any[]) {
    // 定義每一步的結構
    const Step = z.object({
      explanation: z.string(), // 每一步的解釋
      output: z.string(), // 每一步的輸出
    });

    // 定義數學問題響應的結構
    const MathResponse = z.object({
      steps: z.array(Step), // 包含多個步驟
      final_answer: z.string(), // 最終答案
    });

    // 解析結構化響應
    return this.parseStructuredResponse(messages, MathResponse);
  }

  /**
   * 解析查詢函式的響應
   * @param messages - 與模型互動的訊息陣列
   * @returns 解析後的查詢函式響應資料
   */
  async parseQueryFunction(messages: any[]) {
    // 定義查詢相關的列舉和物件
    const Table = z.enum(['orders', 'customers', 'products']);
    const Column = z.enum(['id', 'status', 'expected_delivery_date', 'delivered_at', 'shipped_at', 'ordered_at', 'canceled_at']);
    const Operator = z.enum(['=', '>', '<', '<=', '>=', '!=']);
    const OrderBy = z.enum(['asc', 'desc']);
    const DynamicValue = z.object({ column_name: z.string() });
    const Condition = z.object({
      column: z.string(),
      operator: Operator,
      value: z.union([z.string(), z.number(), DynamicValue]),
    });
    const Query = z.object({
      table_name: Table,
      columns: z.array(Column),
      conditions: z.array(Condition),
      order_by: OrderBy,
    });

    // 解析查詢函式的響應
    const completion = await this.client.beta.chat.completions.parse({
      model: 'gpt-4o-2024-08-06',
      messages,
      tools: [zodFunction({ name: 'query', parameters: Query })],
    });

    const toolCall = completion.choices[0]?.message.tool_calls?.[0];
    if (toolCall) {
      const args = toolCall.function.parsed_arguments as z.infer<typeof Query>;
      return args;
    }
    return null;
  }

  /**
   * 流式助手響應
   * @param threadId - 執行緒 ID
   * @param assistantId - 助手 ID
   * @param eventHandlers - 事件處理程式
   * @returns 執行例項
   */
  streamAssistantResponses(threadId: string, assistantId: string, eventHandlers: any) {
    const run = this.client.beta.threads.runs
      .stream(threadId, {
        assistant_id: assistantId,
      })
      .on('textCreated', eventHandlers.onTextCreated) // 當新文字建立時觸發
      .on('textDelta', eventHandlers.onTextDelta) // 當文字有增量更新時觸發
      .on('toolCallCreated', eventHandlers.onToolCallCreated) // 當工具呼叫建立時觸發
      .on('toolCallDelta', eventHandlers.onToolCallDelta); // 當工具呼叫有增量更新時觸發

    return run;
  }

  /**
   * 自動化函式呼叫
   * @param model - 模型名稱
   * @param messages - 與模型互動的訊息陣列
   * @param tools - 要執行的工具陣列
   * @returns 函式呼叫結果
   */
  runTools(model: string, messages: any[], tools: any[]) {
    return this.client.chat.completions.runTools({
      model,
      messages,
      tools,
    });
  }
}

使用

// 使用示例
(async () => {
  const helper = new OpenAIHelper('your-api-key');

  // 解析數學問題響應
  const mathMessages = [
    { role: 'system', content: '你是一個有幫助的數學老師。' },
    { role: 'user', content: '求解 8x + 31 = 2' },
  ];
  const mathResponse = await helper.parseMathResponse(mathMessages);
  console.log('Math Response:', mathResponse);

  // 解析查詢函式
  const queryMessages = [
    {
      role: 'system',
      content: '你是一個有幫助的助手。今天是2024年8月6日。你幫助使用者查詢他們需要的資料,透過呼叫查詢函式。',
    },
    {
      role: 'user',
      content: '查詢我去年11月所有已完成但未按時交付的訂單',
    },
  ];
  const queryResponse = await helper.parseQueryFunction(queryMessages);
  console.log('Query Response:', queryResponse);

  // 流式助手響應
  const eventHandlers = {
    onTextCreated: (text: string) => process.stdout.write('\n助手 > '),
    onTextDelta: (textDelta: any, snapshot: any) => process.stdout.write(textDelta.value),
    onToolCallCreated: (toolCall: any) => process.stdout.write(`\n助手 > ${toolCall.type}\n\n`),
    onToolCallDelta: (toolCallDelta: any, snapshot: any) => {
      if (toolCallDelta.type === 'code_interpreter') {
        if (toolCallDelta.code_interpreter.input) {
          process.stdout.write(toolCallDelta.code_interpreter.input);
        }
        if (toolCallDelta.code_interpreter.outputs) {
          process.stdout.write('\n輸出 >\n');
          toolCallDelta.code_interpreter.outputs.forEach((output: any) => {
            if (output.type === 'logs') {
              process.stdout.write(`\n${output.logs}\n`);
            }
          });
        }
      }
    },
  };

  const threadId = 'your-thread-id';
  const assistantId = 'your-assistant-id';
  helper.streamAssistantResponses(threadId, assistantId, eventHandlers);

  // 自動化函式呼叫
  const tools = [
    {
      type: 'function',
      function: {
        function: (args: { location: string; time: Date }) => {
          // 獲取天氣的實際實現
          return getWeather(args);
        },
        parse: (args: string) => {
          // 解析函式引數的實際實現
          return parseFunction(args);
        },
        parameters: {
          type: 'object',
          properties: {
            location: { type: 'string' },
            time: { type: 'string', format: 'date-time' },
          },
        },
      },
    },
  ];

  const toolRunner = helper.runTools('gpt-3.5-turbo', [{ role: 'user', content: '這周的天氣如何?' }], tools);
  console.log('Tool Runner:', toolRunner);
})();

/**
 * 假設的獲取天氣的函式
 * @param args - 包含位置和時間的引數
 * @returns 天氣資訊
 */
function getWeather(args: { location: string; time: Date }) {
  // 這裏是獲取天氣的邏輯
  return `The weather in ${args.location} at ${args.time} is sunny.`;
}

/**
 * 假設的解析函式引數的函式
 * @param args - 字串形式的引數
 * @returns 解析後的引數物件
 */
function parseFunction(args: string) {
  // 這裏是解析函式引數的邏輯
  const parsedArgs = JSON.parse(args);
  return {
    location: parsedArgs.location,
    time: new Date(parsedArgs.time),
  };
}

結論

本文介紹瞭如何利用 OpenAI API 的結構化輸出解析助手和流式響應功能,透過使用 Zod 模式自動解析和驗證 JSON 響應,展示了數學問題和查詢函式的解析示例。此外,還講解了如何實現流式響應,實時處理來自 Assistant 的響應,並透過 runTools 方法自動化函式呼叫,提供了一個綜合示例,幫助開發者更高效地使用 OpenAI API。

0則評論

您的電子郵件等資訊不會被公開,以下所有項目均必填

OK! You can skip this field.