本文接上期15.1_OpenAI 库 —— 全方位使用指南继续八、文字转语音TTS8.1 基本用法from openai import OpenAI client OpenAI() # 把文字变成语音 response client.audio.speech.create( modeltts-1, # tts-1快或 tts-1-hd音质更好 voicealloy, # 音色alloy/echo/fable/onyx/nova/shimmer input你好欢迎使用语音合成功能。今天天气真不错, speed1.0, # 语速0.25~4.01.0为正常 ) # 保存为音频文件 response.stream_to_file(output.mp3) print(语音已保存为 output.mp3)8.2 不同音色对比from openai import OpenAI client OpenAI() text 欢迎来到Python编程课堂今天我们来学习函数。 # 6种音色 voices [alloy, echo, fable, onyx, nova, shimmer] for voice in voices: response client.audio.speech.create( modeltts-1, voicevoice, inputtext, ) filename fvoice_{voice}.mp3 response.stream_to_file(filename) print(f✅ {voice} → {filename})九、函数调用Function Calling / Tools⭐ 高级功能9.1 什么是函数调用# 问题AI 只能生成文字不能真的执行操作 # 比如你问北京今天天气怎么样AI 不知道实时天气 # 解决函数调用Function Calling # 1. 你告诉 AI你可以调用这些函数 # 2. AI 判断需要调用哪个函数、传什么参数 # 3. 你的代码执行这个函数拿到真实结果 # 4. 把结果告诉 AI # 5. AI 用自然语言总结给你 # 流程 # 用户问天气 → AI说调用get_weather(city北京) # → 你的代码真的调用天气API → 把结果给AI # → AI说北京今天晴25度适合出行9.2 完整示例天气查询from openai import OpenAI import json client OpenAI() # 第1步定义你的函数 def get_weather(city: str) - str: 模拟获取天气实际项目中调用真实天气API # 模拟数据 weather_data { 北京: {temp: 28, condition: 晴, humidity: 45}, 上海: {temp: 31, condition: 多云, humidity: 70}, 广州: {temp: 33, condition: 雷阵雨, humidity: 85}, } data weather_data.get(city, {temp: 25, condition: 未知, humidity: 50}) return json.dumps(data, ensure_asciiFalse) # 第2步描述函数给 AI 听 tools [ { type: function, function: { name: get_weather, # 函数名 description: 获取指定城市的当前天气信息, # 功能描述 parameters: { # 参数描述JSON Schema格式 type: object, properties: { city: { type: string, description: 城市名称如北京、上海 } }, required: [city] } } } ] # 第3步第一次调用AI 决定要调用什么函数 messages [ {role: user, content: 北京今天天气怎么样} ] response client.chat.completions.create( modelgpt-3.5-turbo, messagesmessages, toolstools, # 告诉AI有哪些工具可用 tool_choiceauto, # 让AI自己决定是否调用 ) message response.choices[0].message # 第4步检查 AI 是否要调用函数 if message.tool_calls: # AI 决定调用函数 tool_call message.tool_calls[0] function_name tool_call.function.name # get_weather function_args json.loads(tool_call.function.arguments) # {city: 北京} print(fAI 要调用{function_name}) print(f参数{function_args}) # 第5步你的代码真正执行这个函数 result get_weather(**function_args) # 调用真实函数 print(f函数返回{result}) # 第6步把函数结果告诉 AI messages.append(message) # 加入 AI 的 tool_call 消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: result, # 函数执行结果 }) # 第7步第二次调用AI 用自然语言总结 final_response client.chat.completions.create( modelgpt-3.5-turbo, messagesmessages, toolstools, ) print(f\nAI 最终回复{final_response.choices[0].message.content}) # 北京今天天气晴朗气温28°C湿度45%非常适合户外活动 else: # AI 直接回答了不需要调用函数 print(message.content)9.3 多函数调用from openai import OpenAI import json client OpenAI() # 定义多个函数 def get_weather(city): return json.dumps({city: city, temp: 26, condition: 晴}) def search_flights(from_city, to_city, date): return json.dumps({flights: [ {flight: CA1234, time: 08:30, price: 680}, {flight: MU5678, time: 14:00, price: 520}, ]}) def book_hotel(city, check_in, nights): return json.dumps({hotel: 如家精选, price: 299, rating: 4.5}) # 描述所有函数 tools [ { type: function, function: { name: get_weather, description: 获取城市天气, parameters: { type: object, properties: {city: {type: string}}, required: [city] } } }, { type: function, function: { name: search_flights, description: 搜索两个城市之间的航班, parameters: { type: object, properties: { from_city: {type: string}, to_city: {type: string}, date: {type: string} }, required: [from_city, to_city, date] } } }, { type: function, function: { name: book_hotel, description: 预订酒店, parameters: { type: object, properties: { city: {type: string}, check_in: {type: string}, nights: {type: integer} }, required: [city, check_in, nights] } } }, ] # 用户问题可能触发多个函数 messages [{role: user, content: 我下周三想从北京飞上海帮我看看天气和航班}] response client.chat.completions.create( modelgpt-4, messagesmessages, toolstools, ) # AI 可能一次返回多个 tool_calls for tool_call in response.choices[0].message.tool_calls: print(f调用{tool_call.function.name}) print(f参数{tool_call.function.arguments})十、微调Fine-tuning10.1 什么是微调通俗解释# 比喻 # 基础模型 一个什么都懂一点的通才大学生 # 微调后 一个在你公司实习了3个月的专属助理 # 什么时候需要微调 # ✅ 需要特定格式输出如特定JSON结构 # ✅ 需要特定风格如模仿某位作家 # ✅ 需要领域知识如法律、医疗术语 # ✅ 想减少 prompt 长度降低成本 # 什么时候不需要 # ❌ 简单的问答、翻译、摘要通用模型就够了 # ❌ 数据量太少至少需要几十条高质量数据10.2 准备训练数据import json # 训练数据格式JSONL每行一个JSON # 每行包含一组 messages和 Chat 格式一样 training_data [ { messages: [ {role: system, content: 你是一个电商客服态度友好回答简短。}, {role: user, content: 我的快递到哪了}, {role: assistant, content: 亲请提供您的订单号我帮您查一下物流信息~} ] }, { messages: [ {role: system, content: 你是一个电商客服态度友好回答简短。}, {role: user, content: 这个衣服能退货吗}, {role: assistant, content: 亲收到货7天内支持无理由退货哦在订单页面点击申请退货就行~} ] }, { messages: [ {role: system, content: 你是一个电商客服态度友好回答简短。}, {role: user, content: 你们有优惠券吗}, {role: assistant, content: 亲目前店铺有满200减30的优惠券在首页可以领取哦~} ] }, # ... 至少需要 50~100 条高质量数据 ] # 写入 JSONL 文件 with open(training_data.jsonl, w, encodingutf-8) as f: for item in training_data: f.write(json.dumps(item, ensure_asciiFalse) \n) print(f已生成训练文件共 {len(training_data)} 条数据)10.3 上传文件并创建微调任务from openai import OpenAI import time client OpenAI() # 第1步上传训练文件 file client.files.create( fileopen(training_data.jsonl, rb), purposefine-tune # 指定用途为微调 ) print(f文件上传成功ID{file.id}) # 例如file-abc123xyz # 第2步创建微调任务 job client.fine_tuning.jobs.create( training_filefile.id, modelgpt-3.5-turbo, # 基于哪个模型微调 # hyperparameters{ # 可选超参数 # n_epochs: 3, # 训练轮数 # } ) print(f微调任务已创建ID{job.id}) print(f状态{job.status}) # validating_files # 第3步等待训练完成 while True: job client.fine_tuning.jobs.retrieve(job.id) print(f状态{job.status}) if job.status succeeded: print(f 训练完成新模型{job.fine_tuned_model}) break elif job.status failed: print(f❌ 训练失败{job.error}) break time.sleep(30) # 每30秒检查一次 # 第4步使用微调后的模型 # 假设新模型名为ft:gpt-3.5-turbo:my-org::abc123 response client.chat.completions.create( modeljob.fine_tuned_model, # 使用微调后的模型 messages[ {role: system, content: 你是一个电商客服态度友好回答简短。}, {role: user, content: 发票怎么开} ] ) print(response.choices[0].message.content) # 会用你训练的风格回答十一、助手AssistantsAPI11.1 什么是 Assistants# Chat Completions每次调用都是独立的你自己管理对话历史 # AssistantsOpenAI 帮你管理一切你只需要对话 # Assistants 的超能力 # 1. 持久记忆对话存在 Thread 里不会丢 # 2. 代码解释器AI 可以真的运行 Python 代码 # 3. 文件检索上传文件AI 可以从中找答案 # 4. 函数调用AI 可以调用你定义的函数 # 5. 多步推理AI 可以连续调用多个工具 # 适用场景 # - 客服机器人需要记住用户历史 # - 数据分析助手需要运行代码 # - 文档问答需要检索文件11.2 完整示例带代码解释器的数学助手from openai import OpenAI import time client OpenAI() # 第1步创建助手 assistant client.beta.assistants.create( name数学分析助手, instructions你是一个数学分析助手。 你可以使用代码解释器来进行计算。 回答要清晰展示计算过程。, modelgpt-4, tools[{type: code_interpreter}] # 赋予运行代码的能力 ) print(f助手已创建{assistant.id}) # 第2步创建对话线程 thread client.beta.threads.create() print(f线程已创建{thread.id}) # 第3步发送用户消息 message client.beta.threads.messages.create( thread_idthread.id, roleuser, content计算 1 到 100 的质数之和 ) print(f消息已发送{message.id}) # 第4步运行助手 run client.beta.threads.runs.create( thread_idthread.id, assistant_idassistant.id, ) print(f运行已开始{run.id}) # 第5步等待完成 while run.status in (queued, in_progress, cancelling): time.sleep(1) run client.beta.threads.runs.retrieve( thread_idthread.id, run_idrun.id ) print(f 状态{run.status}) if run.status completed: # 第6步获取回复 messages client.beta.threads.messages.list( thread_idthread.id, orderdesc # 最新的在前 ) # 第一条是最新的 AI 回复 ai_message messages.data[0] for content_block in ai_message.content: if content_block.type text: print(f\nAI 回复\n{content_block.text.value}) elif run.status failed: print(f❌ 运行失败{run.last_error}) # 第7步继续对话同一个线程 client.beta.threads.messages.create( thread_idthread.id, roleuser, content那1到1000的呢 # AI 记得之前的上下文 ) # 再次运行...重复第4~6步11.3 上传文件给助手from openai import OpenAI client OpenAI() # 上传文件 file client.files.create( fileopen(company_report.pdf, rb), purposeassistants # 注意purpose 是 assistants ) # 创建带文件检索能力的助手 assistant client.beta.assistants.create( name报告分析助手, instructions根据上传的文件回答用户问题。, modelgpt-4, tools[{type: file_search}], # 文件检索工具 tool_resources{ file_search: { vector_store_ids: [] # 或创建向量存储 } } )十二、错误处理12.1 常见错误类型from openai import ( OpenAI, APIError, # 通用 API 错误 APIConnectionError, # 网络连接错误 APITimeoutError, # 请求超时 RateLimitError, # 请求频率超限429 AuthenticationError, # 密钥无效401 PermissionDeniedError, # 权限不足403 NotFoundError, # 资源不存在404 BadRequestError, # 请求参数错误400 ) client OpenAI()12.2 完整的错误处理模板from openai import OpenAI, APIError, RateLimitError, APITimeoutError import time client OpenAI(timeout60.0, max_retries2) def safe_chat(user_message, system_prompt你是一个有帮助的助手。): 带完整错误处理的对话函数 try: response client.chat.completions.create( modelgpt-3.5-turbo, messages[ {role: system, content: system_prompt}, {role: user, content: user_message}, ], temperature0.7, max_tokens1000, ) return response.choices[0].message.content except AuthenticationError: print(❌ API Key 无效请检查密钥) return None except RateLimitError: print(⚠️ 请求太频繁等待后重试...) time.sleep(60) return safe_chat(user_message, system_prompt) # 递归重试 except APITimeoutError: print(⚠️ 请求超时请重试) return None except APIError as e: print(f❌ API 错误{e.status_code} - {e.message}) return None except Exception as e: print(f❌ 未知错误{e}) return None # 使用 result safe_chat(你好介绍一下Python) if result: print(result)12.3 带指数退避的重试import time from openai import OpenAI, APIError, RateLimitError client OpenAI() def chat_with_retry(messages, max_retries5): 指数退避重试 for attempt in range(max_retries): try: response client.chat.completions.create( modelgpt-3.5-turbo, messagesmessages, ) return response.choices[0].message.content except RateLimitError: wait_time 2 ** attempt # 1s, 2s, 4s, 8s, 16s print(f限流{wait_time}秒后重试第{attempt1}次) time.sleep(wait_time) except APIError as e: if e.status_code 500: # 服务器错误可以重试 wait_time 2 ** attempt print(f服务器错误{wait_time}秒后重试) time.sleep(wait_time) else: # 客户端错误4xx重试也没用 raise raise Exception(超过最大重试次数)十三、实战案例13.1 智能翻译工具from openai import OpenAI client OpenAI() def translate(text, target_lang英文): 智能翻译 response client.chat.completions.create( modelgpt-3.5-turbo, messages[ { role: system, content: f你是专业翻译。将用户输入翻译成{target_lang}。只输出翻译结果。 }, {role: user, content: text} ], temperature0.3, ) return response.choices[0].message.content # 测试 print(translate(今天天气真好适合出去走走)) # The weather is really nice today, perfect for a walk. print(translate(机器学习, target_lang日文)) # 機械学習13.2 代码审查助手from openai import OpenAI client OpenAI() def review_code(code): AI 代码审查 response client.chat.completions.create( modelgpt-4, messages[ { role: system, content: 你是资深代码审查员。请审查以下代码 1. 找出 bug 和潜在问题 2. 提出性能优化建议 3. 检查代码风格 4. 给出改进后的代码 用中文回答。 }, {role: user, content: f请审查以下代码\npython\n{code}\n} ], temperature0.2, ) return response.choices[0].message.content # 测试 code_to_review def find_max(lst): max lst[0] for i in range(len(lst)): if lst[i] max: max lst[i] return max print(review_code(code_to_review)) # AI 会指出 # 1. 变量名 max 覆盖了内置函数 # 2. 没有处理空列表 # 3. 可以用 for item in lst 替代 range(len(lst)) # 4. 可以直接用内置 max() 函数13.3 文章摘要生成器from openai import OpenAI client OpenAI() def summarize(text, style简洁, max_words100): 生成文章摘要 style_prompts { 简洁: 用最少的话概括核心内容, 详细: 保留关键细节和论据, bullet: 用要点列表形式, 小学生: 用小学生能懂的语言, } response client.chat.completions.create( modelgpt-3.5-turbo, messages[ { role: system, content: f你是摘要专家。{style_prompts.get(style, style)}。不超过{max_words}字。 }, {role: user, content: f请为以下文章写摘要\n\n{text}} ], temperature0.3, ) return response.choices[0].message.content article 这里放一篇长文章... print(summarize(article, stylebullet))13.4 RAG 问答系统检索增强生成from openai import OpenAI import numpy as np client OpenAI() class SimpleRAG: 简易 RAG 系统先检索相关文档再让 AI 回答 def __init__(self, documents): self.documents documents self.embeddings self._build_index() def _build_index(self): 构建文档向量索引 print(正在构建索引...) response client.embeddings.create( modeltext-embedding-3-small, inputself.documents ) return [np.array(item.embedding) for item in response.data] def _search(self, query, top_k3): 语义搜索最相关的文档 query_resp client.embeddings.create( modeltext-embedding-3-small, inputquery ) query_emb np.array(query_resp.data[0].embedding) # 计算相似度 scores [] for i, doc_emb in enumerate(self.embeddings): sim np.dot(query_emb, doc_emb) / ( np.linalg.norm(query_emb) * np.linalg.norm(doc_emb) ) scores.append((sim, i)) scores.sort(reverseTrue) return [self.documents[idx] for _, idx in scores[:top_k]] def ask(self, question): 提问 # 1. 检索相关文档 relevant_docs self._search(question) context \n\n.join(relevant_docs) # 2. 让 AI 基于文档回答 response client.chat.completions.create( modelgpt-3.5-turbo, messages[ { role: system, content: 根据提供的参考资料回答问题。如果资料中没有答案说我不确定。 }, { role: user, content: f参考资料\n{context}\n\n问题{question} } ], temperature0.2, ) return response.choices[0].message.content # 使用 knowledge_base [ Python 3.12 于2023年10月发布引入了更好的错误提示。, Flask 是一个轻量级 Python Web 框架适合小型项目。, Django 是一个全功能 Python Web 框架自带ORM和Admin。, FastAPI 是现代高性能 Python Web 框架支持异步。, Python 的 GIL 限制了多线程的CPU并行性能。, ] rag SimpleRAG(knowledge_base) answer rag.ask(哪个Python Web框架性能最好) print(answer) # 根据资料FastAPI 是现代高性能 Python Web 框架支持异步...13.5 批量处理from openai import OpenAI import time client OpenAI() def batch_process(items, prompt_template, delay0.5): 批量调用 API带限速保护 results [] for i, item in enumerate(items, 1): print(f处理 {i}/{len(items)}{item[:30]}...) try: response client.chat.completions.create( modelgpt-3.5-turbo, messages[ {role: user, content: prompt_template.format(itemitem)} ], temperature0.3, ) results.append(response.choices[0].message.content) except Exception as e: print(f ❌ 失败{e}) results.append(None) time.sleep(delay) # 限速每次请求间隔0.5秒 return results # 例子批量生成商品描述 products [无线蓝牙耳机, 机械键盘, 便携充电宝, 智能手环] template 为以下商品写一句吸引人的广告语不超过20字{item} slogans batch_process(products, template) for product, slogan in zip(products, slogans): print(f {product}{slogan})十四、费用与 Token 计算14.1 价格参考2024年┌────────────────────┬──────────────┬──────────────┐ │ 模型 │ 输入价格 │ 输出价格 │ │ │ (每百万token) │ (每百万token) │ ├────────────────────┼──────────────┼──────────────┤ │ gpt-4o │ $2.50 │ $10.00 │ │ gpt-4o-mini │ $0.15 │ $0.60 │ │ gpt-4-turbo │ $10.00 │ $30.00 │ │ gpt-3.5-turbo │ $0.50 │ $1.50 │ │ dall-e-3 (1024²) │ - │ $0.04/张 │ │ whisper-1 │ $0.006/分钟 │ - │ │ tts-1 │ $15/百万字符 │ - │ └────────────────────┴──────────────┴──────────────┘ 价格可能变动以官网为准14.2 估算 Token 数import tiktoken # OpenAI 官方 tokenizer 库 # pip install tiktoken def count_tokens(text, modelgpt-3.5-turbo): 计算文本的 token 数 encoding tiktoken.encoding_for_model(model) tokens encoding.encode(text) return len(tokens) # 测试 print(count_tokens(Hello world)) # 2 print(count_tokens(你好世界)) # 5 print(count_tokens(Python是一种编程语言)) # 约 8 # 估算费用 text 写一篇关于人工智能的500字文章 input_tokens count_tokens(text) output_tokens 700 # 假设输出约700 token cost (input_tokens * 0.5 output_tokens * 1.5) / 1_000_000 print(f预估费用${cost:.6f}) # 约 $0.001十五、最佳实践总结15.1 代码规范# ✅ 好的实践 # 1. 使用环境变量管理密钥 import os client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 2. 设置超时和重试 client OpenAI(timeout60.0, max_retries3) # 3. 限制 max_tokens 防止失控 response client.chat.completions.create( modelgpt-3.5-turbo, messages[...], max_tokens2000, ) # 4. 检查 finish_reason if response.choices[0].finish_reason length: print(⚠️ 输出被截断了考虑增加 max_tokens) # 5. 用 try/except 包裹所有 API 调用 # 6. 批量操作加延时避免触发限流 # 7. 日志记录每次调用的 token 用量监控成本15.2 安全注意事项# ❌ 绝对不要 # 1. 把 API Key 写在代码里提交到 Git # 2. 把用户输入直接拼进 system promptPrompt 注入攻击 # 3. 在前端代码中暴露 API Key # ✅ 应该 # 1. 用环境变量或密钥管理服务 # 2. 对用户输入做过滤和转义 # 3. API Key 只放在后端服务器 # 4. 设置用量上限在 OpenAI 后台设置15.3 Prompt 工程技巧# 1. 明确角色 你是一位有10年经验的Python架构师 # 2. 给出具体指令不要模糊 # ❌ 帮我写代码 # ✅ 用Python写一个函数输入是列表输出是去重后的排序列表要求时间复杂度O(nlogn) # 3. 给出示例Few-shot 示例输入[3,1,2] → 输出[1,2,3]。现在请处理[5,2,8,1] # 4. 指定输出格式 请以JSON格式输出包含name、age、city三个字段 # 5. 分步骤思考 请一步步思考先分析问题再设计方案最后写代码 # 6. 设定约束 回答不超过100字、不要使用任何外部库、用中文回答十六、API 方法速查表┌─────────────────────────────────────────────────────────────────┐ │ openai 库核心方法速查 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 文本生成 │ │ client.chat.completions.create(model, messages, ...) │ │ │ │ ️ 图像生成 │ │ client.images.generate(model, prompt, size, ...) │ │ client.images.edit(model, image, mask, prompt, ...) │ │ client.images.create_variation(model, image, n, ...) │ │ │ │ 嵌入向量 │ │ client.embeddings.create(model, input) │ │ │ │ 语音转文字 │ │ client.audio.transcriptions.create(model, file, ...) │ │ client.audio.translations.create(model, file) │ │ │ │ 文字转语音 │ │ client.audio.speech.create(model, voice, input, ...) │ │ │ │ 文件管理 │ │ client.files.create(file, purpose) │ │ client.files.list() │ │ client.files.delete(file_id) │ │ │ │ 微调 │ │ client.fine_tuning.jobs.create(training_file, model) │ │ client.fine_tuning.jobs.retrieve(job_id) │ │ client.fine_tuning.jobs.list() │ │ │ │ 助手Assistants │ │ client.beta.assistants.create(name, instructions, model, tools)│ │ client.beta.threads.create() │ │ client.beta.threads.messages.create(thread_id, role, content) │ │ client.beta.threads.runs.create(thread_id, assistant_id) │ │ client.beta.threads.messages.list(thread_id) │ │ │ │ 模型列表 │ │ client.models.list() │ │ │ └─────────────────────────────────────────────────────────────────┘十七、学习路径建议第1天安装 配置 第一次对话chat.completions 第2天理解 messages 结构 system/user/assistant 第3天多轮对话 流式输出 第4天temperature / max_tokens / 参数调优 第5天图像生成DALL·E 第6天嵌入 语义搜索 第7天语音转文字 文字转语音 第8天函数调用Function Calling 第9天错误处理 重试 成本控制 第10天实战项目聊天机器人/RAG/翻译工具 第11天微调Fine-tuning 第12天助手AssistantsAPI十八、一句话总结openai 库 通往 AI 世界的万能钥匙记住核心调用模式from openai import OpenAI client OpenAI() response client.功能模块.方法.create(参数...) result response.提取结果五大核心功能聊天→client.chat.completions.create()画图→client.images.generate()向量化→client.embeddings.create()语音转文字→client.audio.transcriptions.create()文字转语音→client.audio.speech.create()三条铁律 密钥放环境变量绝不硬编码⏱️ 永远设置 timeout永远 try/except 关注 token 消耗控制成本