随着人工智能技术的不断发展，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。
