Python调用可灵AI接口实现批量自动化脚本教程

想用Python脚本自动化调用可灵AI的接口，实现批量内容生成？这确实是个提升效率的好思路。核心就是通过其提供的HTTP API，来构造请求、传递参数并解析响应。下面，我们就来聊聊几种主流的实现方法，你可以根据任务规模和技术栈来选择。

一、使用 requests 库同步批量调用

对于任务量不是特别大、对并发要求不高，同时又希望逐条控制请求顺序和错误处理的场景，这个方法非常稳妥。思路很直接：循环构造JSON请求体，然后发送POST请求。

首先，得安装依赖库：pip install requests。

接着，准备好从可灵AI平台获取的API Key，在请求头的Authorization字段里填上“Bearer + 你的密钥”。

然后，构建包含prompt（提示词）、model_id（模型ID）、temperature（温度参数）等关键信息的JSON数据体，每一个请求就对应一个生成任务。

具体操作时，用一个for循环遍历你的输入文本列表，每次循环都调用一次requests.post发送请求，并把响应结果里生成的内容提取出来，保存到本地的CSV文件里。为了友好使用，避免触发平台的频率限制，建议在每次请求后加上time.sleep(0.5)这样的短暂停顿。

二、使用 asyncio + aiohttp 异步并发调用

如果需要处理的批量任务非常多，追求高吞吐量以缩短整体运行时间，那么异步并发方案就派上用场了。它利用异步I/O复用，能在单线程内同时维持多个HTTP连接，效率提升显著。

第一步同样是安装依赖：pip install aiohttp。

然后，需要定义一个异步的请求函数，它接收具体的prompt和一个aiohttp的ClientSession实例。在函数内部构造好请求头和JSON载荷后，调用session.post来获取响应。

关键技巧在于，使用asyncio.gather来并发执行所有任务，把准备好的prompts列表和共享的Session传进去就行。别忘了做好异常处理，对每个响应解析result字段，捕获可能的异常并记录下失败任务的索引，这样可以避免因为单个任务出错而导致整个流程中断。

最后，将所有成功的响应结果，按行写入一个JSONL文件，每条结果都是一个独立的对象，后续处理起来很方便。

三、基于 Celery 的分布式任务队列调用

如果面对的是企业级、需要长期运行、并且要求具备重试机制、失败告警和任务状态追踪的复杂批量生成场景，那么Celery这类分布式任务队列就是一个更专业的选择。它将生成任务解耦为生产者-消费者模型，支持横向扩展worker节点来提升处理能力。

首先，需要部署Redis或RabbitMQ作为消息中间件，并配置Celery实例去连接它。

接着，定义一个用@app.task装饰的任务函数，函数内部封装好requests.post的调用逻辑，并设计好异常回退策略。

主程序负责从Excel表格里读取prompt列，然后为每个prompt调用task.delay(prompt)，将任务推送到队列中。

之后，启动多个celery worker进程，它们会自动从队列里拉取任务并执行。生成的结果可以存入PostgreSQL这类数据库中，通常包含task_id、prompt、output、status、created_at等字段，便于管理和查询。

还可以通过Flower这样的监控界面，实时查看所有任务的状态、耗时甚至失败时的错误堆栈信息，运维起来一目了然。

四、使用 requests.Session + 连接复用优化吞吐

这个方法适用于在单机上进行高频次调用的场景。它的核心优势在于通过复用TCP连接，减少了每次请求的握手开销，从而能在不突破平台QPS限制的前提下，尽可能压榨出更高的性能。

操作上，先创建一个全局的requests.Session实例，并为其设置适配器，启用连接池（例如设置pool_connections=10, pool_maxsize=20）。

预先设定好请求的URL模板和默认的请求头，在实际调用时，只需要动态替换prompt和seed这类参数即可。

发送请求时，使用session.post来代替普通的requests.post，这样才能确保底层的TCP连接被复用。

对于返回状态码不是200的响应，要留意检查response.headers.get("X-RateLimit-Remaining")这个值。如果它低于你设定的安全阈值，最好主动sleep一段时间，等待限制重置。

输出结果的处理也可以优化，比如每生成50条结果，就将它们合并打成一个ZIP包，再写入指定的目录，这样能减少小文件的数量，更有利于管理。