Python 调用大模型 API：工程实战总纲

import os
from openai import OpenAI

# 官方 OpenAI：默认读 OPENAI_API_KEY 与 OPENAI_BASE_URL 环境变量
client = OpenAI()

resp = client.chat.completions.create(
    model="gpt-4o-mini",  # 示意，具体型号以所用厂商文档为准
    messages=[{"role": "user", "content": "用一句话解释什么是嵌入"}],
)
print(resp.choices[0].message.content)
print(resp.usage.total_tokens)  # 响应自带用量，成本记账靠它

# 兼容端点：只换 base_url 与 key，其余代码一行不改
client = OpenAI(
    base_url="https://api.example-llm.com/v1",
    api_key=os.environ["EXAMPLE_LLM_KEY"],
)

import anthropic

client = anthropic.Anthropic()  # 默认读 ANTHROPIC_API_KEY 环境变量
msg = client.messages.create(
    model="claude-sonnet-4-5",  # 示意，可用型号以 Anthropic 文档为准
    max_tokens=1024,
    messages=[{"role": "user", "content": "用一句话解释什么是嵌入"}],
)
print(msg.content[0].text)
print(msg.usage.input_tokens, msg.usage.output_tokens)

pip install python-dotenv

# .env 文件（必须已加入 .gitignore）
LLM_API_KEY=sk-your-key-here
LLM_BASE_URL=https://api.example-llm.com/v1

import os
from dotenv import load_dotenv

load_dotenv()  # 把 .env 读进环境变量；已存在的环境变量不会被覆盖

key = os.environ["LLM_API_KEY"]  # 缺失时立刻 KeyError——这是优点，不是缺点

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "讲一个 200 字的故事"}],
    stream=True,  # 服务端逐块推送，客户端边收边处理
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)  # flush 让终端也「逐字」出，而不是攒在缓冲区

from pydantic import BaseModel, ValidationError
from openai import OpenAI

client = OpenAI()

class Invoice(BaseModel):
    vendor: str        # 开票方
    amount: float      # 金额
    currency: str      # 币种

def extract_invoice(text: str, max_retries: int = 2) -> Invoice:
    messages = [
        {"role": "system", "content": "从文本抽取发票信息，严格输出 JSON。"},
        {"role": "user", "content": text},
    ]
    for _ in range(max_retries + 1):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={"type": "json_object"},  # 强制 JSON 模式，能力以厂商文档为准
        )
        raw = resp.choices[0].message.content
        try:
            return Invoice.model_validate_json(raw)   # 校验 + 解析一步完成
        except ValidationError as e:
            # 关键一步：把校验错误喂回去，让模型照着错误修
            messages.append({"role": "assistant", "content": raw})
            messages.append({"role": "user", "content": f"输出不符合 schema，错误如下，请修正后重新输出完整 JSON：\n{e}"})
    raise RuntimeError("结构化抽取重试次数用尽")

from openai import APIConnectionError, InternalServerError, RateLimitError
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
from pybreaker import CircuitBreaker

RETRYABLE = (RateLimitError, InternalServerError, APIConnectionError)

llm_breaker = CircuitBreaker(fail_max=5, reset_timeout=60)  # 连续 5 次失败后熔断 60 秒

@llm_breaker
@retry(
    retry=retry_if_exception_type(RETRYABLE),   # 只重试值得重试的异常
    wait=wait_exponential(multiplier=1, max=30),  # 指数退避：1s、2s、4s……封顶 30s
    stop=stop_after_attempt(5),
)
def ask_llm(messages: list[dict]):
    return client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        timeout=30.0,  # 三件套之一：显式超时，绝不依赖 SDK 默认值
    )

from openai import OpenAI

client = OpenAI()

# 1) 把一批请求写入 JSONL 文件并上传
file = client.files.create(file=open("batch_tasks.jsonl", "rb"), purpose="batch")
# 2) 提交批处理任务，24 小时窗口内完成
batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
# 3) 轮询 batch.status，完成后下载结果文件

from collections import defaultdict

ledger = defaultdict(lambda: {"input": 0, "output": 0})

def chat_metered(messages: list[dict], *, tag: str) -> str:
    """统一调用入口：所有 LLM 调用走这里，成本画像按 tag 逐日累积。"""
    resp = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
    ledger[tag]["input"] += resp.usage.prompt_tokens
    ledger[tag]["output"] += resp.usage.completion_tokens
    return resp.choices[0].message.content

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class ChatIn(BaseModel):
    question: str

@app.post("/api/chat")
def chat(body: ChatIn):
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": body.question}],
        stream=True,
    )

    def sse():
        for chunk in stream:                      # 消费 LLM 的流
            delta = chunk.choices[0].delta.content
            if delta:
                yield "data: " + json.dumps({"text": delta}, ensure_ascii=False) + "\n\n"
        yield "data: [DONE]\n\n"                  # 自定义收尾事件

    return StreamingResponse(sse(), media_type="text/event-stream")