LLM API 基础调用：OpenAI、Anthropic、Google 全面对比与 FastAPI 服务端封装

import os
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

async def chat_completion(
    messages: list[dict],
    model: str = "gpt-4o",
    temperature: float = 0.7,
    max_tokens: int = 4096,
) -> str:
    response = await client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    return response.choices[0].message.content

# 使用示例
import asyncio

async def main():
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain async/await in Python in 3 sentences."},
    ]
    result = await chat_completion(messages)
    print(result)

asyncio.run(main())

async def chat_with_tools():
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string"},
                        "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                    },
                    "required": ["city"],
                },
            },
        }
    ]

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
        tools=tools,
        tool_choice="auto",
    )

    message = response.choices[0].message
    if message.tool_calls:
        for call in message.tool_calls:
            print(f"Function: {call.function.name}")
            print(f"Arguments: {call.function.arguments}")
            # 执行函数并返回结果...

from pydantic import BaseModel

class ExtractedInfo(BaseModel):
    name: str
    age: int
    hobbies: list[str]

async def structured_extract(text: str) -> ExtractedInfo:
    response = await client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "Extract structured info from text."},
            {"role": "user", "content": text},
        ],
        response_format=ExtractedInfo,
    )
    return response.choices[0].message.parsed

import os
from anthropic import AsyncAnthropic

anthropic_client = AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

async def claude_chat(
    messages: list[dict],
    model: str = "claude-3-5-sonnet-20241022",
    max_tokens: int = 4096,
) -> str:
    response = await anthropic_client.messages.create(
        model=model,
        max_tokens=max_tokens,
        messages=messages,
    )
    return response.content[0].text

# Claude 使用不同消息格式：不含 system 在 messages 中
async def main():
    response = await anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system="You are a Python expert.",  # system 是独立参数
        messages=[
            {"role": "user", "content": "Explain Python GIL."}
        ],
    )
    print(response.content[0].text)

async def claude_with_thinking():
    response = await anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        thinking={"type": "enabled", "budget_tokens": 2000},
        messages=[{"role": "user", "content": "Solve this math problem step by step..."}],
    )
    for block in response.content:
        if block.type == "thinking":
            print(f"[Thinking] {block.thinking}")
        elif block.type == "text":
            print(f"[Answer] {block.text}")

import google.generativeai as genai

genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))

async def gemini_chat(
    prompt: str,
    model: str = "gemini-1.5-pro-latest",
) -> str:
    model_obj = genai.GenerativeModel(model)
    response = await model_obj.generate_content_async(prompt)
    return response.text

# 多模态输入（图像+文本）
async def gemini_multimodal(image_path: str, prompt: str):
    model_obj = genai.GenerativeModel("gemini-1.5-pro-latest")
    image = PIL.Image.open(image_path)
    response = await model_obj.generate_content_async([prompt, image])
    return response.text

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """使用 tiktoken 精确计算 token 数量。"""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str = "gpt-4o",
) -> float:
    """估算 API 调用成本（美元）。"""
    prices = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
        "gemini-1.5-pro": {"input": 1.25, "output": 5.00},
    }
    p = prices.get(model, prices["gpt-4o"])
    # 价格单位是 /1M tokens
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

# 预算管理上下文管理器
from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class BudgetTracker:
    budget_usd: float = 10.0
    spent_usd: float = 0.0
    total_input_tokens: int = 0
    total_output_tokens: int = 0

    def can_afford(self, estimated_cost: float) -> bool:
        return self.spent_usd + estimated_cost <= self.budget_usd

    def record(self, input_tokens: int, output_tokens: int, model: str):
        cost = estimate_cost(input_tokens, output_tokens, model)
        self.spent_usd += cost
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens

async def chat_with_tracking(
    messages: list[dict],
    tracker: BudgetTracker,
    model: str = "gpt-4o",
) -> str:
    input_tokens = sum(count_tokens(m["content"], model) for m in messages)
    estimated_output = 2048
    est_cost = estimate_cost(input_tokens, estimated_output, model)

    if not tracker.can_afford(est_cost):
        raise RuntimeError(f"Budget exceeded: ${tracker.spent_usd:.4f} / ${tracker.budget_usd}")

    response = await client.chat.completions.create(
        model=model,
        messages=messages,
    )

    usage = response.usage
    tracker.record(usage.prompt_tokens, usage.completion_tokens, model)
    return response.choices[0].message.content

async def stream_chat(messages: list[dict], model: str = "gpt-4o"):
    """流式输出，适合长文本生成与实时 UI 更新。"""
    stream = await client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        stream_options={"include_usage": True},
    )

    full_text = ""
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            text = chunk.choices[0].delta.content
            full_text += text
            print(text, end="", flush=True)
        # 最终 chunk 包含 usage
        if chunk.usage:
            print(f"\n\nTokens: {chunk.usage.total_tokens}")
    return full_text

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

app = FastAPI()

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o"
    stream: bool = True

@app.post("/chat")
async def chat_endpoint(req: ChatRequest):
    if not req.stream:
        result = await chat_completion(req.messages, req.model)
        return {"content": result}

    async def event_generator():
        stream = await client.chat.completions.create(
            model=req.model,
            messages=req.messages,
            stream=True,
        )
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                data = json.dumps({"text": chunk.choices[0].delta.content})
                yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"

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

import asyncio
import random
from openai import RateLimitError, APIError, APITimeoutError

async def robust_chat_completion(
    messages: list[dict],
    model: str = "gpt-4o",
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
) -> str:
    """带指数退避和抖动的健壮 API 调用。"""
    for attempt in range(max_retries + 1):
        try:
            response = await client.chat.completions.create(
                model=model,
                messages=messages,
            )
            return response.choices[0].message.content

        except RateLimitError as e:
            if attempt == max_retries:
                raise
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            retry_after = e.response.headers.get("retry-after")
            if retry_after:
                delay = float(retry_after)
            print(f"Rate limited, retrying in {delay:.1f}s...")
            await asyncio.sleep(delay)

        except (APITimeoutError, APIError) as e:
            if attempt == max_retries:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            print(f"API error ({e}), retrying in {delay:.1f}s...")
            await asyncio.sleep(delay)

import base64

async def vision_chat(image_path: str, prompt: str):
    with open(image_path, "rb") as f:
        image_b64 = base64.b64encode(f.read()).decode()

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{image_b64}",
                        "detail": "high",
                    },
                },
            ],
        }
    ]
    return await chat_completion(messages, model="gpt-4o")

async def audio_chat(audio_path: str, prompt: str):
    with open(audio_path, "rb") as f:
        audio_b64 = base64.b64encode(f.read()).decode()

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "input_audio", "input_audio": {"data": audio_b64, "format": "mp3"}},
            ],
        }
    ]
    return await chat_completion(messages, model="gpt-4o-audio-preview")

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import Literal, AsyncGenerator
import asyncio
import json
import os
import time

app = FastAPI(title="LLM API Gateway")

# 并发控制
SEM = asyncio.Semaphore(10)

class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str

class ChatCompletionRequest(BaseModel):
    messages: list[ChatMessage]
    model: Literal["gpt-4o", "gpt-4o-mini", "claude-3-5-sonnet"] = "gpt-4o"
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=4096, ge=1, le=8192)
    stream: bool = False
    response_format: dict | None = None

    @property
    def total_input_tokens(self) -> int:
        return sum(count_tokens(m.content, self.model) for m in self.messages)

class ChatCompletionResponse(BaseModel):
    content: str
    model: str
    usage: dict
    latency_ms: float

async def get_llm_response(
    req: ChatCompletionRequest,
) -> tuple[str, dict]:
    """路由到对应提供商。"""
    async with SEM:
        if req.model.startswith("gpt"):
            response = await client.chat.completions.create(
                model=req.model,
                messages=[m.model_dump() for m in req.messages],
                temperature=req.temperature,
                max_tokens=req.max_tokens,
                response_format=req.response_format,
            )
            content = response.choices[0].message.content
            usage = {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens,
            }
        elif req.model.startswith("claude"):
            response = await anthropic_client.messages.create(
                model=req.model,
                max_tokens=req.max_tokens,
                messages=[{"role": m.role, "content": m.content} for m in req.messages if m.role != "system"],
                system=next((m.content for m in req.messages if m.role == "system"), None),
                temperature=req.temperature,
            )
            content = response.content[0].text
            usage = {
                "prompt_tokens": response.usage.input_tokens,
                "completion_tokens": response.usage.output_tokens,
                "total_tokens": response.usage.input_tokens + response.usage.output_tokens,
            }
        else:
            raise HTTPException(400, f"Unsupported model: {req.model}")
    return content, usage

@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def chat_completions(req: ChatCompletionRequest):
    if req.total_input_tokens > 100_000:
        raise HTTPException(400, "Input too large (max 100K tokens)")

    start = time.perf_counter()

    if req.stream:
        async def stream_gen() -> AsyncGenerator[str, None]:
            if req.model.startswith("gpt"):
                stream = await client.chat.completions.create(
                    model=req.model,
                    messages=[m.model_dump() for m in req.messages],
                    stream=True,
                    temperature=req.temperature,
                    max_tokens=req.max_tokens,
                )
                async for chunk in stream:
                    if chunk.choices and chunk.choices[0].delta.content:
                        data = json.dumps({"text": chunk.choices[0].delta.content})
                        yield f"data: {data}\n\n"
            yield "data: [DONE]\n\n"

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

    content, usage = await get_llm_response(req)
    latency = (time.perf_counter() - start) * 1000

    return ChatCompletionResponse(
        content=content,
        model=req.model,
        usage=usage,
        latency_ms=round(latency, 2),
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from abc import ABC, abstractmethod
from typing import AsyncIterator

class LLMProvider(ABC):
    @abstractmethod
    async def chat(self, messages: list[dict], **kwargs) -> str: ...

    @abstractmethod
    async def stream_chat(self, messages: list[dict], **kwargs) -> AsyncIterator[str]: ...

    @abstractmethod
    async def embed(self, texts: list[str]) -> list[list[float]]: ...

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)

    async def chat(self, messages, model="gpt-4o", **kwargs):
        resp = await self.client.chat.completions.create(model=model, messages=messages, **kwargs)
        return resp.choices[0].message.content

    async def stream_chat(self, messages, model="gpt-4o", **kwargs):
        stream = await self.client.chat.completions.create(model=model, messages=messages, stream=True, **kwargs)
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

    async def embed(self, texts: list[str], model="text-embedding-3-small"):
        resp = await self.client.embeddings.create(model=model, input=texts)
        return [d.embedding for d in resp.data]

class UnifiedLLMClient:
    """统一多提供商客户端，支持自动故障转移。"""
    def __init__(self, providers: list[LLMProvider]):
        self.providers = providers

    async def chat_with_fallback(self, messages: list[dict], **kwargs) -> str:
        for i, provider in enumerate(self.providers):
            try:
                return await provider.chat(messages, **kwargs)
            except Exception as e:
                if i == len(self.providers) - 1:
                    raise
                print(f"Provider {i} failed: {e}, trying next...")