Skip to content

Python与AI API实操入门

零基础 · 从申请API Key到调用大模型API · 全程手把手带你写出第一段AI代码

本篇定位

这篇教程面向从未写过代码的测试人员。不讲理论,只讲动手。每一步都有完整代码,复制粘贴就能运行。学完本篇你将能够:用Python调用OpenAI / Claude / 豆包等大模型API,处理流式响应,使用结构化输出,并在本地运行模型。

本篇知识架构

你的Python脚本 ↓ 调用 openai 库 httpx 库 ↓ HTTP请求 OpenAI API Claude API 豆包 API Ollama(本地) ↓ 推理 GPT-5 Claude Opus 4.6 Gemini 3 Pro Doubao 1.6 Qwen 3 / Llama 4

第1章:申请API Key

1.1 三大平台API Key申请

OpenAI

  1. 访问 platform.openai.com
  2. 注册 / 登录账号
  3. 进入 API Keys 页面
  4. 点击 "Create new secret key"
  5. 立即复制保存(只显示一次!)
  6. Key格式:sk-proj-xxxxxxxx

Anthropic(Claude)

  1. 访问 console.anthropic.com
  2. 注册 / 登录账号
  3. 进入 API Keys 页面
  4. 点击 "Create Key"
  5. Key格式:sk-ant-xxxxxxxx

豆包(火山引擎)

  1. 访问 console.volcengine.com
  2. 开通"火山方舟"服务
  3. 创建API Key
  4. 创建"推理接入点"(Endpoint),记下Endpoint ID
  5. 豆包使用OpenAI兼容接口,所以也用 openai 库调用

1.2 用 .env 文件管理密钥

安全警告

永远不要把API Key直接写在代码里!如果代码上传到GitHub,你的Key会被泄露,可能产生大额费用。正确做法:把Key放在 .env 文件里,并把 .env 加入 .gitignore

# 第1步:在项目根目录创建 .env 文件
# 文件内容如下(把 xxx 替换成你的真实Key)

OPENAI_API_KEY=sk-proj-xxx
ANTHROPIC_API_KEY=sk-ant-xxx
VOLCENGINE_API_KEY=xxx
VOLCENGINE_ENDPOINT_ID=ep-xxx
# 第2步:在Python代码中读取 .env
from dotenv import load_dotenv
import os

load_dotenv()  # 读取 .env 文件

openai_key = os.getenv("OPENAI_API_KEY")
print(f"Key已加载: {openai_key[:10]}...")  # 只打印前10位,保护隐私
# 第3步:创建 .gitignore 文件,防止 .env 被提交
# .gitignore 内容:
.env
.venv/
__pycache__/

1.3 各平台价格对比

平台模型输入价格 ($/1M tokens)输出价格 ($/1M tokens)特点
OpenAIGPT-5$5.00$15.002026 综合能力第一梯队
OpenAIGPT-5-mini$0.25$1.00性价比之选,日常测试首选
OpenAIo4 / o3-pro$15 / $20$60 / $80推理模型,复杂数学/代码强
AnthropicClaude Opus 4.6$15.00$75.002026-03 发布,500K 长文本、推理强
AnthropicClaude Sonnet 4.6$3.00$15.00性价比款,质量逼近 Opus
AnthropicClaude Haiku 4$0.50$2.50低延迟,跑批首选
GoogleGemini 3 Pro$2.00$8.00原生多模态 + 2M 上下文 + Deep Think
GoogleGemini 3 Flash$0.15$0.60极致便宜,1M 上下文
DeepSeekDeepSeek-V3.5 / R2$0.27 / $0.55$1.10 / $2.19国产开源标杆,R2 是推理专用
火山引擎Doubao 1.6 Pro¥3/1M tokens¥9/1M tokens国内部署,中文优化好
月之暗面Kimi K2$0.6$2.52M 超长上下文
本地Ollama + Qwen3 / Llama 4免费免费完全离线,无隐私顾虑

省钱建议

日常开发调试用 gpt-5-mini / gemini-3-flash / claude-haiku-4 或本地 Ollama 模型,正式评测再用 gpt-5 / claude-opus-4-6 / gemini-3-pro。费用可以控制在每月几美元到十几美元之间。

1.4 环境准备

本教程假设你已经安装了 Python 3.10 或以上版本。首次使用前,先安装下面这些库(它们会在后续章节中用到):

pip3 install "openai>=1.55" httpx python-dotenv tiktoken pydantic
库名用途
openai (>=1.55)OpenAI 官方 SDK,也兼容豆包、DeepSeek 等 OpenAI 兼容接口
httpx更现代的 HTTP 客户端,支持异步和流式
python-dotenv.env 文件读取环境变量,保护 API Key
tiktoken计算 Token 数量,用于费用估算
pydantic数据校验和结构化输出解析

建议在项目目录下创建虚拟环境(python3 -m venv .venv)后安装,避免污染全局环境。后续代码默认你在已激活的环境中运行。

第2章:第一次调用Chat API

2.1 完整代码示例

下面是一段可以直接运行的完整代码。每一行都有注释,确保你理解每一步在做什么。

# ============================================
# first_call.py —— 你的第一次大模型API调用
# ============================================

# 第1步:导入需要的库
from openai import OpenAI        # OpenAI官方SDK
from dotenv import load_dotenv   # 读取.env文件
import os                        # 操作系统相关功能

# 第2步:加载环境变量(从.env文件读取API Key)
load_dotenv()

# 第3步:创建客户端(用API Key初始化连接)
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY")
)

# 第4步:发送请求
response = client.chat.completions.create(
    model="gpt-5-mini",         # 使用的模型(先用便宜的练手)
    messages=[                   # 对话消息列表
        {
            "role": "system",    # 系统角色:设定AI的行为
            "content": "你是一个专业的软件测试工程师。"
        },
        {
            "role": "user",      # 用户角色:你的问题
            "content": "请用一句话解释什么是回归测试。"
        }
    ],
    temperature=0.7,             # 随机性(0=确定性,1=创造性)
    max_tokens=200               # 最多生成多少个token
)

# 第5步:提取并打印结果
answer = response.choices[0].message.content
print(f"模型回答: {answer}")

# 第6步:查看Token用量
usage = response.usage
print(f"\n--- Token 用量 ---")
print(f"输入Token: {usage.prompt_tokens}")
print(f"输出Token: {usage.completion_tokens}")
print(f"总计Token: {usage.total_tokens}")

运行方法

在终端中执行:python3 first_call.py。如果一切正常,你会看到模型的回答和Token用量。

2.2 请求结构解析

上面代码中 client.chat.completions.create() 的参数含义:

参数类型必填说明
modelstring使用哪个模型
messageslist对话消息数组,每条包含 role 和 content
temperaturefloat0~2,越高越随机。测试一致性时设0
max_tokensint限制输出长度,防止费用失控
top_pfloat核采样,和temperature二选一调即可
nint一次生成几个回答,默认1
streambool是否启用流式返回(下一章详解)

2.3 响应结构解析

# response 对象的结构长这样:
{
    "id": "chatcmpl-abc123",
    "object": "chat.completion",
    "model": "gpt-5-mini",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "回归测试是在修改代码后重新运行已有测试用例,确保改动没有引入新缺陷。"
            },
            "finish_reason": "stop"
        }
    ],
    "usage": {
        "prompt_tokens": 28,
        "completion_tokens": 35,
        "total_tokens": 63
    }
}
字段含义测试关注点
choices[0].message.content模型的回答文本核心检查对象
choices[0].finish_reason停止原因:stop / length / content_filter如果是 length 说明被截断了
usage.prompt_tokens输入消耗的Token数用于费用计算
usage.completion_tokens输出消耗的Token数用于费用计算
model实际使用的模型名确认是否和请求一致

第3章:处理流式返回(SSE)

3.1 为什么需要流式

普通调用(非流式)需要等模型把所有内容生成完毕后,才一次性返回给你。如果模型回答很长,你可能要等好几秒才能看到任何内容。 流式调用(SSE)让模型一边生成一边返回,每生成一个词就推送过来。这就是ChatGPT界面上"逐字打出"的效果。

非流式:等待3秒 → 一次性返回全部文本 **体验:**用户盯着空白等很久 **适合:**自动化脚本、批量测试 流式(SSE):0.3秒收到第一个词 → 逐字输出 **体验:**用户立刻看到内容在打出 **适合:**对话产品、实时交互

3.2 流式调用完整代码

# ============================================
# stream_call.py —— 流式调用示例
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os
import time

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 记录开始时间
start_time = time.time()
first_token_time = None

# 发送流式请求(关键:stream=True)
stream = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[
        {"role": "user", "content": "用3个要点解释什么是性能测试。"}
    ],
    stream=True    # 启用流式
)

# 逐chunk接收并打印
full_response = ""
for chunk in stream:
    # 每个chunk的结构:chunk.choices[0].delta.content
    content = chunk.choices[0].delta.content

    if content is not None:
        # 记录第一个Token的时间(TTFT)
        if first_token_time is None:
            first_token_time = time.time()

        print(content, end="", flush=True)  # 逐字打印,不换行
        full_response += content

# 计算耗时
end_time = time.time()
ttft = first_token_time - start_time if first_token_time else 0
total_time = end_time - start_time

print(f"\n\n--- 性能指标 ---")
print(f"TTFT(首Token延迟): {ttft:.3f} 秒")
print(f"总耗时: {total_time:.3f} 秒")
print(f"总字数: {len(full_response)} 字")

3.3 TTFT —— 第一个Token延迟

TTFT(Time To First Token)是衡量大模型响应速度最重要的指标之一。它衡量的是从发出请求到收到第一个词的时间。

发送请求 → 等待(TTFT) → 收到第一个Token → 持续接收 → 接收完毕

指标含义合格标准(参考)
TTFT首Token延迟对话场景 < 1秒
TPS每秒生成Token数> 30 tokens/s 体验较好
Total Latency完整响应总耗时视场景而定

第4章:Structured Output(结构化输出)

4.1 什么是结构化输出

普通调用中模型返回的是自由文本——你说"帮我提取信息",它可能返回一段话、一个列表、或者各种格式。结构化输出让模型强制返回符合你定义的JSON格式,方便程序自动解析。

4.2 方式一:response_format = json_object

# ============================================
# structured_output.py —— 结构化输出示例
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os, json

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[
        {
            "role": "system",
            "content": "你是测试用例生成器。请以JSON格式返回,包含字段:test_name, steps(数组), expected_result。"
        },
        {
            "role": "user",
            "content": "为'用户登录'功能生成一个测试用例。"
        }
    ],
    response_format={"type": "json_object"}   # 强制返回JSON
)

# 解析返回的JSON
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, ensure_ascii=False, indent=2))

4.3 方式二:Pydantic模型 + 解析

Pydantic可以定义精确的数据结构,自动校验类型和必填字段。这在测试中非常有用——你可以直接验证模型输出是否符合Schema。

# ============================================
# pydantic_output.py —— 使用Pydantic定义输出结构
# ============================================
from openai import OpenAI
from pydantic import BaseModel
from dotenv import load_dotenv
import os

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 定义期望的输出结构
class TestCase(BaseModel):
    test_name: str                 # 用例名称
    precondition: str              # 前置条件
    steps: list[str]               # 测试步骤
    expected_result: str           # 预期结果
    priority: str                  # 优先级

# 使用 parse 方法让模型直接输出结构化数据
response = client.beta.chat.completions.parse(
    model="gpt-5-mini",
    messages=[
        {
            "role": "system",
            "content": "你是测试用例生成器。"
        },
        {
            "role": "user",
            "content": "为'用户修改密码'功能生成一个测试用例。"
        }
    ],
    response_format=TestCase
)

# 结果已经是 TestCase 对象,可以直接用
test_case = response.choices[0].message.parsed
print(f"用例名称: {test_case.test_name}")
print(f"前置条件: {test_case.precondition}")
print(f"测试步骤:")
for i, step in enumerate(test_case.steps):
    print(f"  {i+1}. {step}")
print(f"预期结果: {test_case.expected_result}")
print(f"优先级: {test_case.priority}")

4.4 测试场景:验证输出是否符合Schema

# 场景:批量验证模型输出的JSON是否每次都符合Schema
from pydantic import BaseModel, ValidationError
import json

class ExpectedOutput(BaseModel):
    answer: str
    confidence: float       # 必须是浮点数
    sources: list[str]      # 必须是字符串列表

# 模拟多次模型输出
model_outputs = [
    '{"answer": "北京", "confidence": 0.95, "sources": ["百科"]}',
    '{"answer": "上海", "confidence": "高", "sources": ["百科"]}',     # confidence不是数字
    '{"answer": "广州"}',                                               # 缺少字段
]

for i, output in enumerate(model_outputs):
    try:
        data = json.loads(output)
        validated = ExpectedOutput(**data)
        print(f"  ✅ 输出 {i+1}: Schema验证通过")
    except ValidationError as e:
        print(f"  ❌ 输出 {i+1}: Schema验证失败")
        print(f"     错误: {e.errors()[0]['msg']}")
    except json.JSONDecodeError:
        print(f"  ❌ 输出 {i+1}: 不是合法JSON")

第5章:Tool Calling(工具调用)

5.1 什么是Tool Calling

大模型本身只能"说话",不能"做事"。Tool Calling让模型可以调用你事先定义好的函数。比如模型自己不知道今天的天气,但它可以决定调用你写好的 get_weather 函数来获取。

用户提问 → 模型判断需要调工具 → 返回工具名+参数 → 你执行工具 → 把结果喂回模型 → 模型组织最终回答

5.2 完整代码示例

# ============================================
# tool_calling.py —— 工具调用完整流程
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os, json

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ---- 第1步:定义工具(告诉模型有哪些工具可用)----
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_test_result",
            "description": "查询指定测试用例的执行结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "case_id": {
                        "type": "string",
                        "description": "测试用例ID,如 TC-001"
                    }
                },
                "required": ["case_id"]
            }
        }
    }
]

# ---- 第2步:模拟工具的实际实现 ----
def get_test_result(case_id):
    """模拟查询测试结果的函数"""
    fake_db = {
        "TC-001": {"status": "passed", "duration": "2.3s", "last_run": "2026-04-15"},
        "TC-002": {"status": "failed", "duration": "5.1s", "last_run": "2026-04-14"},
    }
    return fake_db.get(case_id, {"error": f"用例 {case_id} 不存在"})

# ---- 第3步:发送请求,让模型决定是否调用工具 ----
messages = [
    {"role": "user", "content": "帮我查一下TC-001这个测试用例的执行结果"}
]

response = client.chat.completions.create(
    model="gpt-5-mini",
    messages=messages,
    tools=tools
)

# ---- 第4步:检查模型是否决定调用工具 ----
choice = response.choices[0]

if choice.finish_reason == "tool_calls":
    tool_call = choice.message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)

    print(f"模型决定调用工具: {func_name}")
    print(f"参数: {func_args}")

    # ---- 第5步:执行工具并获取结果 ----
    result = get_test_result(func_args["case_id"])
    print(f"工具返回: {result}")

    # ---- 第6步:把工具结果喂回模型 ----
    messages.append(choice.message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result, ensure_ascii=False)
    })

    final_response = client.chat.completions.create(
        model="gpt-5-mini",
        messages=messages
    )

    print(f"\n模型最终回答: {final_response.choices[0].message.content}")
else:
    print(f"模型直接回答: {choice.message.content}")

5.3 测试视角:如何验证工具选择

从测试角度,Tool Calling需要验证的核心问题:

验证维度检查内容典型测试用例
工具选择模型是否选对了工具明确需要查天气时,是否调了get_weather而不是get_news
参数提取参数是否正确提取"查TC-001的结果"是否提取出 case_id="TC-001"
不该调时不调简单问题不需要工具"你好"不应该触发任何工具调用
结果整合模型是否正确使用工具返回的结果工具返回passed,模型不应该说failed

第6章:本地运行模型(Ollama)

6.1 为什么要本地运行

本地运行模型的好处:免费无网络依赖数据不出本机。非常适合日常练习和敏感数据场景。

6.2 安装Ollama

# macOS / Linux(一条命令搞定)
curl -fsSL https://ollama.com/install.sh | sh

# 验证安装
ollama --version

Windows用户访问 ollama.com 下载安装包。

6.3 下载和运行模型

# 下载 Qwen3(8B参数,约5GB;2026 主流开源中文模型)
ollama pull qwen3

# 也可以选 Llama 4(Meta 2026 开源旗舰,原生多模态)
# ollama pull llama4

# 交互式对话(测试模型是否正常)
ollama run qwen3

# 查看已下载的模型列表
ollama list

6.4 用Python调用本地模型

Ollama完全兼容OpenAI API格式。只需要改两个参数:base_urlmodel

# ============================================
# local_model.py —— 调用本地Ollama模型
# ============================================
from openai import OpenAI

# 关键:指向本地Ollama服务,不需要API Key
client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"    # Ollama不验证Key,随便填
)

response = client.chat.completions.create(
    model="qwen3",
    messages=[
        {"role": "user", "content": "请用一句话解释什么是单元测试。"}
    ],
    temperature=0.7
)

print(response.choices[0].message.content)

6.5 本地 vs 云端对比

维度本地(Ollama)云端(OpenAI / Claude)
费用免费按Token收费
隐私数据不出本机数据发送到服务商
速度取决于本机硬件通常更快,有GPU集群
模型能力7B~70B参数,能力有限数千亿参数,能力最强
网络要求无需网络必须联网
适合场景学习、调试、敏感数据正式评测、生产环境

第7章:费用控制与Token计算

7.1 用 tiktoken 计算Token数

在发送请求之前,先用 tiktoken 库预估一下这次请求会消耗多少Token、花多少钱。避免不小心发了一段很长的文本,产生高额费用。

# ============================================
# token_calc.py —— Token计算与费用预估
# ============================================
import tiktoken

# 加载GPT-5使用的tokenizer
encoder = tiktoken.encoding_for_model("gpt-5")

# 计算一段文本的Token数
text = "请帮我分析这个测试报告中的失败用例,并给出改进建议。"
tokens = encoder.encode(text)

print(f"文本: {text}")
print(f"Token数: {len(tokens)}")
print(f"Token列表: {tokens[:10]}...")  # 只看前10个

# ---- 费用预估函数 ----
def estimate_cost(prompt_text, estimated_output_tokens=500, model="gpt-5-mini"):
    """预估一次API调用的费用"""
    enc = tiktoken.encoding_for_model("gpt-5")
    input_tokens = len(enc.encode(prompt_text))

    prices = {
        "gpt-5":      {"input": 5.00, "output": 15.00},
        "gpt-5-mini": {"input": 0.25, "output": 1.00},
    }

    p = prices.get(model, prices["gpt-5-mini"])
    input_cost  = (input_tokens / 1_000_000) * p["input"]
    output_cost = (estimated_output_tokens / 1_000_000) * p["output"]
    total = input_cost + output_cost

    print(f"\n--- 费用预估 ({model}) ---")
    print(f"输入: {input_tokens} tokens → ${input_cost:.6f}")
    print(f"输出: ~{estimated_output_tokens} tokens → ${output_cost:.6f}")
    print(f"总计: ${total:.6f}")
    return total

# 使用
estimate_cost("请帮我写一篇500字的测试报告", estimated_output_tokens=800)
estimate_cost("请帮我写一篇500字的测试报告", estimated_output_tokens=800, model="gpt-5")

7.2 压测时的成本控制策略

策略做法效果
限制max_tokens设置 max_tokens=200防止模型生成过长回答
先用便宜模型开发调试用 gpt-5-mini费用降低90%+
发前预估用tiktoken算费用,超预算跳过避免意外高额消费
设置用量上限在OpenAI后台设定月度预算硬性兜底
本地模型替代调试阶段用Ollama完全免费
缓存结果相同Prompt不重复请求减少无意义调用

7.3 各场景费用估算

场景请求数平均输入Token平均输出Token模型预估费用
日常调试(学习)50次/天100200gpt-5-mini~$0.007/天
批量Prompt测试100条200500gpt-5-mini~$0.03
模型评测500条5001000gpt-5~$5.60
压力测试1000条200500gpt-5~$5.50
长文本评测100条50002000gpt-5~$3.25

费用控制口诀

调试用mini,评测用4o,本地用Ollama,发前算Token。每月设预算,跑前先预估。

第8章:综合练习与答案

学习建议

以下三个练习覆盖了本篇所有核心知识点。建议先自己写,遇到不会的再回翻对应章节,不要直接抄答案。

参考答案说明

练习1、练习2、练习3 下方给出的完整脚本,就是这一章对应的参考答案。建议先自己实现,再对照代码检查:是否正确读取环境变量、是否保留异常处理、是否输出统计结果、是否把结果保存下来。

练习1:批量测试10个Prompt

写一个脚本,读取一组Prompt,逐个发送给模型,检查返回结果是否包含关键词,并保存结果到JSON文件。

# ============================================
# exercise_1.py —— 批量Prompt测试
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os, json, time

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 测试用例:每个包含prompt和期望关键词
test_cases = [
    {"prompt": "中国的首都是哪里?",      "keyword": "北京"},
    {"prompt": "1+1等于几?",             "keyword": "2"},
    {"prompt": "水的化学式是什么?",       "keyword": "H2O"},
    {"prompt": "地球绕太阳一圈多久?",    "keyword": "一年"},
    {"prompt": "Python是什么类型的语言?", "keyword": "编程"},
    {"prompt": "HTTP的全称是什么?",       "keyword": "Hypertext"},
    {"prompt": "JSON的全称是什么?",       "keyword": "JavaScript"},
    {"prompt": "TCP和UDP的区别?",         "keyword": "连接"},
    {"prompt": "什么是API?",              "keyword": "接口"},
    {"prompt": "什么是回归测试?",         "keyword": "测试"},
]

results = []
passed = 0

for i, tc in enumerate(test_cases):
    print(f"[{i+1}/{len(test_cases)}] 测试: {tc['prompt']}")

    try:
        response = client.chat.completions.create(
            model="gpt-5-mini",
            messages=[{"role": "user", "content": tc["prompt"]}],
            max_tokens=200,
            temperature=0
        )
        answer = response.choices[0].message.content
        is_pass = tc["keyword"] in answer

        if is_pass:
            passed += 1
            print(f"  ✅ 通过(包含'{tc['keyword']}')")
        else:
            print(f"  ❌ 失败(未找到'{tc['keyword']}')")
            print(f"     回答: {answer[:80]}...")

        results.append({
            "prompt": tc["prompt"],
            "keyword": tc["keyword"],
            "answer": answer,
            "passed": is_pass
        })

    except Exception as e:
        print(f"  ❌ 异常: {e}")
        results.append({"prompt": tc["prompt"], "error": str(e), "passed": False})

    time.sleep(0.5)  # 避免触发限流

# 保存结果
with open("test_results.json", "w", encoding="utf-8") as f:
    json.dump(results, f, ensure_ascii=False, indent=2)

print(f"\n========== 测试完成 ==========")
print(f"通过: {passed}/{len(test_cases)}")
print(f"通过率: {passed/len(test_cases)*100:.1f}%")
print(f"结果已保存到 test_results.json")

练习2:比较两个模型的回答质量

同一个问题同时发给两个模型,对比结果。

# ============================================
# exercise_2.py —— 模型对比测试
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os, time

load_dotenv()

# 配置两个模型客户端
cloud_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
local_client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

models = [
    {"name": "gpt-5-mini",  "client": cloud_client, "model": "gpt-5-mini"},
    {"name": "qwen3(本地)", "client": local_client, "model": "qwen3"},
]

test_prompt = "请解释什么是冒烟测试,并给出一个具体的例子。"
print(f"测试Prompt: {test_prompt}\n")

for m in models:
    print(f"--- {m['name']} ---")
    start = time.time()

    try:
        resp = m["client"].chat.completions.create(
            model=m["model"],
            messages=[{"role": "user", "content": test_prompt}],
            max_tokens=300,
            temperature=0
        )
        elapsed = time.time() - start
        answer = resp.choices[0].message.content

        print(f"耗时: {elapsed:.2f}s")
        print(f"回答: {answer[:200]}...")
        print(f"输出长度: {len(answer)} 字")
    except Exception as e:
        print(f"错误: {e}")

    print()

练习3:流式响应TTFT计算器

对同一个Prompt重复测试N次,计算TTFT的平均值、最大值、最小值。

# ============================================
# exercise_3.py —— TTFT批量测试
# ============================================
from openai import OpenAI
from dotenv import load_dotenv
import os, time

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def measure_ttft(prompt, model="gpt-5-mini"):
    """测量单次请求的TTFT"""
    start = time.time()
    first_token_time = None

    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=100
    )

    for chunk in stream:
        if chunk.choices[0].delta.content:
            if first_token_time is None:
                first_token_time = time.time()
            # 消费完整个流
    
    end = time.time()
    ttft = (first_token_time - start) if first_token_time else None
    total = end - start
    return {"ttft": ttft, "total": total}

# 运行测试
prompt = "用3句话解释持续集成。"
rounds = 5
results = []

print(f"Prompt: {prompt}")
print(f"测试轮次: {rounds}\n")

for i in range(rounds):
    r = measure_ttft(prompt)
    results.append(r)
    print(f"  第{i+1}轮: TTFT={r['ttft']:.3f}s  总耗时={r['total']:.3f}s")
    time.sleep(1)

# 统计
ttft_values = [r["ttft"] for r in results if r["ttft"]]
print(f"\n--- TTFT统计 ---")
print(f"  最小值: {min(ttft_values):.3f}s")
print(f"  最大值: {max(ttft_values):.3f}s")
print(f"  平均值: {sum(ttft_values)/len(ttft_values):.3f}s")