隨著人工智慧技術的不斷髮展，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 的響應。
開始流
有三種建立流的輔助方法：
openai.beta.threads.runs.stream();openai.beta.threads.createAndRunStream();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。
