来源:https://www.laoyingai.com/
使用Python调用ChatGPT API主要依赖OpenAI官方库,以下是完整的调用方法和代码示例。
1. 安装与配置
首先安装官方Python库:
bash
pip install openai
获取API密钥后,推荐使用环境变量管理(创建.env文件或直接设置环境变量):
python
import os from openai import OpenAI # 从环境变量读取密钥(推荐) client = OpenAI( api_key=os.environ.get("OPENAI_API_KEY") )安全提醒:切勿将API密钥硬编码在代码中。
2. 基础调用方法
方法一:Chat Completions API(当前标准)
python
from openai import OpenAI client = OpenAI() completion = client.chat.completions.create( model="gpt-4o", # 或 "gpt-4-turbo", "gpt-3.5-turbo" messages=[ {"role": "system", "content": "你是一个专业的编程助手。"}, {"role": "user", "content": "如何在Python中实现快速排序?"} ], temperature=0.7 # 控制随机性,0-1之间 ) print(completion.choices[0].message.content)方法二:Responses API(较新的主要API)
python
response = client.responses.create( model="gpt-4o", instructions="你是一个友好的客服助手", input="请问你们支持7天无理由退货吗?" ) print(response.output_text)
参数说明
model:模型名称,如
gpt-4o、gpt-4-turbo、gpt-3.5-turbomessages:对话历史,每个消息包含
role(system/user/assistant)和contenttemperature:0-1之间,值越低输出越确定,越高越有创造性
max_tokens:限制响应最大长度
3. 多轮对话管理
多轮对话需要维护完整的消息历史:
python
def run_conversation(): messages = [ {"role": "system", "content": "你是一个善于倾听的助手。"} ] while True: user_input = input("用户: ") if user_input.lower() == 'quit': break # 添加用户消息 messages.append({"role": "user", "content": user_input}) # 获取响应 completion = client.chat.completions.create( model="gpt-4o", messages=messages ) assistant_reply = completion.choices[0].message.content print(f"助手: {assistant_reply}") # 添加助手回复到历史 messages.append({"role": "assistant", "content": assistant_reply})关键点:每次调用都需要发送完整的历史记录,否则模型会丢失上下文。
4. 流式响应(Streaming)
适用于聊天界面等需要实时输出的场景:
python
# 同步流式 stream = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "讲一个关于独角兽的短故事"}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="", flush=True)python
# 异步流式 from openai import AsyncOpenAI import asyncio async def stream_response(): client = AsyncOpenAI() stream = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "你好"}], stream=True ) async for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="") asyncio.run(stream_response())5. 异步调用
提升并发性能,适用于高吞吐场景:
python
import asyncio from openai import AsyncOpenAI async def main(): client = AsyncOpenAI() response = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "解释什么是异步编程"}] ) print(response.choices[0].message.content) asyncio.run(main())6. 函数调用(Function Calling)
让模型调用自定义函数获取实时数据:
python
import json from openai import OpenAI # 定义工具函数 def get_weather(latitude: float, longitude: float) -> str: """获取指定坐标的天气""" # 实际应调用真实API return f"{latitude}, {longitude} 位置的温度为15°C" # 定义工具描述 tools = [{ "type": "function", "function": { "name": "get_weather", "description": "获取指定经纬度的当前温度", "parameters": { "type": "object", "properties": { "latitude": {"type": "number"}, "longitude": {"type": "number"} }, "required": ["latitude", "longitude"] } } }] client = OpenAI() messages = [{"role": "user", "content": "纽约现在的天气怎么样?"}] completion = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools ) # 处理工具调用 if completion.choices[0].finish_reason == "tool_calls": tool_call = completion.choices[0].message.tool_calls[0] arguments = json.loads(tool_call.function.arguments) # 执行函数 result = get_weather(**arguments) # 将结果返回给模型 messages.append(completion.choices[0].message) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) # 获取最终回复 final_response = client.chat.completions.create( model="gpt-4o", messages=messages ) print(final_response.choices[0].message.content)7. 错误处理与重试
python
from openai import OpenAI, APIError, APIConnectionError, RateLimitError import time def call_with_retry(max_retries=3): for attempt in range(max_retries): try: completion = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Hello"}] ) return completion.choices[0].message.content except RateLimitError: wait_time = min(2 ** attempt + 0.5, 10) print(f"速率限制,等待 {wait_time} 秒...") time.sleep(wait_time) except APIConnectionError: print("连接错误,重试中...") time.sleep(1) except APIError as e: print(f"API错误: {e}") break raise Exception("所有重试尝试失败")8. 结构化输出(JSON模式)
python
completion = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "从文本中提取姓名和城市,以JSON格式返回。"}, {"role": "user", "content": "张三住在北京,是一名工程师。"} ], response_format={"type": "json_object"} ) print(completion.choices[0].message.content) # 输出:{"name": "张三", "city": "北京"}注意事项
成本控制:注意
max_tokens限制,响应token比输入token成本更高速率限制:免费层每分钟3次调用,生产环境需升级计划
API版本:确保使用最新SDK(v1.0+),旧版
openai.Completion已废弃
来源 https://www.laoyingai.com/