Python 调用 OpenAI 兼容 API 批量处理文本教程：摘要、分类、标题生成

1Q748

270人浏览 · 2026-05-14 19:00:00

1Q748 · 2026-05-14 19:00:00 发布

摘要

很多 AI 应用最开始并不复杂：读取一批文本，调用模型，输出摘要、分类或标题。

如果你使用的是 OpenAI 兼容 API，Python 里通常只需要配置：

api_key base_url model

本文用一个完整示例演示如何用 Python 调用 OpenAI 兼容 API 批量处理文本，包括：

单条文本摘要
多条文本批量处理
JSONL 结果保存
常见报错排查

示例中的接口地址可以替换为你自己的 OpenAI 兼容服务。文末会用 51relay 作为示例入口。

一、准备环境

安装依赖：

pip install openai

准备三个环境变量：

export RELAY_BASE_URL="https://example.com/v1" export RELAY_API_KEY="sk-xxxxxxxxxxxxxxxx" export RELAY_MODEL="your-model"

说明：

RELAY_BASE_URL：OpenAI 兼容接口地址，一般以 /v1 结尾
RELAY_API_KEY：API Key
RELAY_MODEL：模型名

不要把真实 API Key 写进代码仓库。

二、先测试接口是否可用

在写 Python 代码前，建议先用 curl 查询模型列表：

curl -sS "$RELAY_BASE_URL/models" \ -H "Authorization: Bearer $RELAY_API_KEY"

如果能返回模型列表，说明 Base URL 和 API Key 基本可用。

再测试最小请求：

curl -sS "$RELAY_BASE_URL/chat/completions" \ -H "Authorization: Bearer $RELAY_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "your-model", "messages": [ { "role": "user", "content": "请回复一句：接口已接通" } ] }'

curl 能通，再继续写 Python 脚本。

三、Python 单条文本摘要

新建 summary_one.py：

import os from openai import OpenAI client = OpenAI( api_key=os.environ["RELAY_API_KEY"], base_url=os.environ["RELAY_BASE_URL"] ) text = """ OpenAI 兼容 API 的接入重点是 Base URL、API Key 和模型名。接入前建议先用 curl 查询模型列表，再用最小请求验证接口是否可用。 """ resp = client.chat.completions.create( model=os.environ["RELAY_MODEL"], messages=[ { "role": "user", "content": f"请把下面内容总结成 3 条要点：\n\n{text}" } ] ) print(resp.choices[0].message.content)

运行：

python summary_one.py

如果能输出摘要，说明 Python 调用链路已经跑通。

四、批量处理文本

准备一个 input.txt：

OpenAI 兼容 API 配置时，Base URL 通常要带 /v1。 Claude Code 配置时，ANTHROPIC_BASE_URL 通常填基础域名。 API Key 不建议写在前端或提交到 Git 仓库。 model not found 通常是模型名写错或账号没有权限。

新建 batch_summary.py：

import os import json from pathlib import Path from openai import OpenAI client = OpenAI( api_key=os.environ["RELAY_API_KEY"], base_url=os.environ["RELAY_BASE_URL"] ) model = os.environ["RELAY_MODEL"] items = [ line.strip() for line in Path("input.txt").read_text(encoding="utf-8").splitlines() if line.strip() ] def summarize(text): resp = client.chat.completions.create( model=model, messages=[ { "role": "user", "content": f"请把下面内容改写成一句适合技术文章的小标题：{text}" } ] ) return resp.choices[0].message.content.strip() with Path("output.jsonl").open("w", encoding="utf-8") as f: for index, item in enumerate(items, start=1): result = summarize(item) row = { "index": index, "input": item, "output": result } f.write(json.dumps(row, ensure_ascii=False) + "\n") print(row)

运行：

python batch_summary.py

输出文件 output.jsonl 每行是一条结果：

{"index":1,"input":"OpenAI 兼容 API 配置时，Base URL 通常要带 /v1。","output":"OpenAI 兼容 API 的 Base URL 配置要点"}

五、批量分类示例

除了摘要，也可以做分类。

def classify(text): prompt = f""" 请判断下面内容属于哪一类：分类只能从下面选择： - 配置问题 - 认证问题 - 模型问题 - 安全问题 - 其他只返回分类名称，不要解释。内容： {text} """ resp = client.chat.completions.create( model=model, messages=[ { "role": "user", "content": prompt } ] ) return resp.choices[0].message.content.strip()

可以把 summarize(item) 换成 classify(item)。

六、增加错误处理

批量任务不能因为一条失败就全部中断。

可以这样写：

with Path("output.jsonl").open("w", encoding="utf-8") as f: for index, item in enumerate(items, start=1): try: result = summarize(item) row = { "index": index, "input": item, "output": result, "status": "success" } except Exception as e: row = { "index": index, "input": item, "error": str(e), "status": "failed" } f.write(json.dumps(row, ensure_ascii=False) + "\n") print(row)

这样即使某条失败，也能继续处理后面的内容。