如何在C#中实现OpenAI Batch API完整批量处理流程

现有代码的核心问题 #

1. 文件上传未遵循OpenAI API要求：必须使用multipart/form-data格式，并指定purpose=batch参数，原代码直接传递StreamContent会导致接口拒绝请求。
2. HttpClient实例重复创建：每次实例化客户端会耗尽连接池，引发性能问题，应复用实例。
3. 缺少关键流程方法：未实现批处理状态查询、结果文件获取及下载功能，无法完成完整的批处理流程。

完整修复后的代码 #

using System;
using System.IO;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading.Tasks;
using Newtonsoft.Json.Linq;

public class OpenAIBatchClient
{
    // 复用HttpClient实例，避免连接池耗尽
    private static readonly HttpClient _httpClient = new HttpClient();
    private readonly string _openAiApiKey;

    public OpenAIBatchClient(string openAiApiKey)
    {
        _openAiApiKey = openAiApiKey;
        // 仅在首次使用时设置授权头
        if (!_httpClient.DefaultRequestHeaders.Contains("Authorization"))
        {
            _httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _openAiApiKey);
        }
    }

    // 上传.jsonl文件到OpenAI，返回文件ID
    public async Task<string> UploadFileAsync(string filePath)
    {
        using var multipartContent = new MultipartFormDataContent();
        using var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read);
        using var fileContent = new StreamContent(fileStream);

        // 设置文件内容类型
        fileContent.Headers.ContentType = new MediaTypeHeaderValue("application/jsonl");
        // 添加文件和purpose参数（必须为batch）
        multipartContent.Add(fileContent, "file", Path.GetFileName(filePath));
        multipartContent.Add(new StringContent("batch"), "purpose");

        var response = await _httpClient.PostAsync("https://api.openai.com/v1/files", multipartContent);
        if (response.IsSuccessStatusCode)
        {
            var responseBody = await response.Content.ReadAsStringAsync();
            return JObject.Parse(responseBody)["id"].ToString();
        }
        else
        {
            throw new Exception($"文件上传失败: {response.StatusCode} - {await response.Content.ReadAsStringAsync()}");
        }
    }

    // 创建批处理任务，返回批处理ID
    public async Task<string> CreateBatchAsync(string inputFileId)
    {
        var batchRequest = new
        {
            input_file_id = inputFileId,
            endpoint = "/v1/chat/completions",
            completion_window = "24h",
            metadata = new { description = "nightly eval job" }
        };

        var content = new StringContent(
            JObject.FromObject(batchRequest).ToString(),
            System.Text.Encoding.UTF8,
            "application/json"
        );

        var response = await _httpClient.PostAsync("https://api.openai.com/v1/batches", content);
        if (response.IsSuccessStatusCode)
        {
            var responseBody = await response.Content.ReadAsStringAsync();
            return JObject.Parse(responseBody)["id"].ToString();
        }
        else
        {
            throw new Exception("批处理任务创建失败: " + await response.Content.ReadAsStringAsync());
        }
    }

    // 查询批处理任务状态
    public async Task<string> GetBatchStatusAsync(string batchId)
    {
        var response = await _httpClient.GetAsync($"https://api.openai.com/v1/batches/{batchId}");
        if (response.IsSuccessStatusCode)
        {
            var responseBody = await response.Content.ReadAsStringAsync();
            return JObject.Parse(responseBody)["status"].ToString();
        }
        else
        {
            throw new Exception($"查询批处理状态失败: {response.StatusCode} - {await response.Content.ReadAsStringAsync()}");
        }
    }

    // 获取批处理完成后的结果文件ID
    public async Task<string> GetBatchResultFileIdAsync(string batchId)
    {
        var response = await _httpClient.GetAsync($"https://api.openai.com/v1/batches/{batchId}");
        if (response.IsSuccessStatusCode)
        {
            var responseBody = await response.Content.ReadAsStringAsync();
            var json = JObject.Parse(responseBody);
            // 任务完成后才会有output_file_id
            if (json["status"].ToString() == "completed")
            {
                return json["output_file_id"].ToString();
            }
            throw new Exception("批处理任务未完成，无法获取结果文件");
        }
        else
        {
            throw new Exception($"获取结果文件ID失败: {response.StatusCode} - {await response.Content.ReadAsStringAsync()}");
        }
    }

    // 下载结果文件到本地
    public async Task DownloadResultFileAsync(string fileId, string savePath)
    {
        var response = await _httpClient.GetAsync($"https://api.openai.com/v1/files/{fileId}/content");
        if (response.IsSuccessStatusCode)
        {
            using var stream = await response.Content.ReadAsStreamAsync();
            using var fileStream = new FileStream(savePath, FileMode.Create, FileAccess.Write);
            await stream.CopyToAsync(fileStream);
        }
        else
        {
            throw new Exception($"下载结果文件失败: {response.StatusCode} - {await response.Content.ReadAsStringAsync()}");
        }
    }
}

使用示例 #

class Program
{
    static async Task Main(string[] args)
    {
        var apiKey = "你的OpenAI API密钥";
        var client = new OpenAIBatchClient(apiKey);
        var inputFilePath = @"D:\input_batch.jsonl";
        var outputSavePath = @"D:\batch_results.jsonl";

        try
        {
            // 1. 上传输入文件
            string fileId = await client.UploadFileAsync(inputFilePath);
            Console.WriteLine($"文件上传成功，ID: {fileId}");

            // 2. 创建批处理任务
            string batchId = await client.CreateBatchAsync(fileId);
            Console.WriteLine("批处理任务创建成功，ID: " + batchId);

            // 3. 轮询等待任务完成（实际生产中建议用定时任务或Webhook）
            string status;
            do
            {
                status = await client.GetBatchStatusAsync(batchId);
                Console.WriteLine($"当前批处理状态: {status}");
                await Task.Delay(60000); // 每分钟查询一次
            } while (status != "completed" && status != "failed");

            if (status == "failed")
            {
                Console.WriteLine("批处理任务执行失败");
                return;
            }

            // 4. 获取结果文件ID并下载
            string resultFileId = await client.GetBatchResultFileIdAsync(batchId);
            await client.DownloadResultFileAsync(resultFileId, outputSavePath);
            Console.WriteLine("结果文件已下载到: " + outputSavePath);
        }
        catch (Exception ex)
        {
            Console.WriteLine("错误: " + ex.Message);
        }
    }
}

关键注意事项 #

• 输入的.jsonl文件必须符合OpenAI Batch API格式要求，每条记录需包含custom_id和method/url/body字段。
• 生产环境中不建议使用轮询，推荐配置OpenAI的Webhook来接收任务完成通知。
• 确保API密钥有足够的权限访问Batch API，且账户余额充足。

内容的提问来源于stack exchange，提问作者TANVI