API OpenAI и Anthropic

🎯 Цель

После прочтения этой главы:

• Знаете работу с OpenAI и Anthropic API
• Используете Streaming responses, function calling, vision API
• Знаете уменьшение расходов до 90% через prompt caching
• Добавляете retry, rate limit, error handling в production

Что нужно изучить

• OpenAI SDK — Python client
• Anthropic SDK — Python client
• Chat completions — основной API
• Streaming — real-time response
• Function calling / Tool use — structured actions
• Vision — работа с изображениями
• Embeddings — для semantic search
• Prompt caching(Anthropic) — уменьшение стоимости на 90%
• Batching — async parallel calls
• Стратегии Rate limitingи retry
• Token tracking и observability

Библиотеки

pip install openai anthropic
pip install instructor              # structured output
pip install tenacity                # retry logic
pip install backoff                 # exponential backoff

Примеры кода

OpenAI — basic chat

from openai import OpenAI

client = OpenAI(api_key="sk-...")  # или os.getenv("OPENAI_API_KEY")

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Ты вспомогательный assistant."},
        {"role": "user", "content": "Привет! Что такое list comprehension в Python?"},
    ],
    temperature=0.7,
    max_tokens=500,
)

print(response.choices[0].message.content)
print(f"Tokens: in={response.usage.prompt_tokens}, out={response.usage.completion_tokens}")

Anthropic — basic message

from anthropic import Anthropic

client = Anthropic(api_key="sk-ant-...")

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system="Ты вспомогательный assistant.",
    messages=[
        {"role": "user", "content": "Что такое list comprehension в Python?"},
    ],
)

print(response.content[0].text)
print(f"Tokens: in={response.usage.input_tokens}, out={response.usage.output_tokens}")

Streaming — real-time

OpenAI streaming

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Напишите длинный рассказ"}],
    stream=True,
)

for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)

Anthropic streaming

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Напишите длинный рассказ"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Function Calling / Tool Use

OpenAI function calling

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Возвращает погоду для указанного города",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "Название города"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
    },
}]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Какая погода в Ташкенте?"}],
    tools=tools,
)

# Выполнить tool call
tool_call = response.choices[0].message.tool_calls[0]
if tool_call.function.name == "get_weather":
    args = json.loads(tool_call.function.arguments)
    weather = get_weather(args["city"], args.get("unit", "celsius"))
    
    # Вернуть результат обратно в LLM
    response2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "Какая погода в Ташкенте?"},
            response.choices[0].message,
            {"role": "tool", "tool_call_id": tool_call.id, "content": str(weather)},
        ],
        tools=tools,
    )
    print(response2.choices[0].message.content)

Anthropic tool use

tools = [{
    "name": "get_weather",
    "description": "Возвращает погоду для указанного города",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
        },
        "required": ["city"],
    },
}]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Какая погода в Ташкенте?"}],
)

# Выполнить tool use
for block in response.content:
    if block.type == "tool_use":
        if block.name == "get_weather":
            result = get_weather(**block.input)
            # Вернуть результат
            response2 = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                tools=tools,
                messages=[
                    {"role": "user", "content": "Какая погода в Ташкенте?"},
                    {"role": "assistant", "content": response.content},
                    {"role": "user", "content": [{
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result),
                    }]},
                ],
            )

Vision API

OpenAI vision

import base64

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Что вы видите на этом изображении?"},
            {
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{encode_image('photo.jpg')}"},
            },
        ],
    }],
)

Anthropic vision

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Что вы видите на этом изображении?"},
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/jpeg",
                    "data": encode_image("photo.jpg"),
                },
            },
        ],
    }],
)

Prompt Caching (Anthropic) — 90% дешевле!

# Большой system prompt кешируется, не оплачивается повторно
LARGE_SYSTEM = open("docs.md").read()  # 50K token docs

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": LARGE_SYSTEM,
            "cache_control": {"type": "ephemeral"},  # CACHE!
        },
    ],
    messages=[{"role": "user", "content": "Вопрос про справочник..."}],
)

# Первый раз: full price + cache write (1.25x)
# В следующие 5 минут: 0.1x price (90% cheaper!)

Embeddings

OpenAI embeddings

response = client.embeddings.create(
    model="text-embedding-3-small",  # 1536-dim, $0.02 / 1M tokens
    input=["Salom dunyo", "Machine learning"],
)

embeddings = [d.embedding for d in response.data]
# Shape: [(1536,), (1536,)]

Anthropic embeddings? — нет

У Anthropic нет своего embeddings API. Варианты:

• OpenAI text-embedding-3-small
• Voyage AI (рекомендует Anthropic)
• Cohere embeddings
• Sentence Transformers (local)

Retry + Rate Limiting

from tenacity import retry, stop_after_attempt, wait_exponential
from openai import RateLimitError, APIError

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=lambda e: isinstance(e, (RateLimitError, APIError)),
)
async def call_llm_with_retry(messages: list, model: str = "gpt-4o-mini"):
    response = await async_client.chat.completions.create(
        model=model,
        messages=messages,
    )
    return response.choices[0].message.content

Async batching

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def process_one(text: str):
    response = await async_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Summarize: {text}"}],
    )
    return response.choices[0].message.content

async def process_batch(texts: list[str], max_concurrent: int = 10):
    sem = asyncio.Semaphore(max_concurrent)
    
    async def bounded(text):
        async with sem:
            return await process_one(text)
    
    return await asyncio.gather(*[bounded(t) for t in texts])

# 100 текстов с 10 concurrent
results = asyncio.run(process_batch(texts, max_concurrent=10))

Cost tracking middleware

import logging
from contextlib import contextmanager

logger = logging.getLogger("llm_costs")

PRICES = {
    "gpt-4o-mini": (0.15, 0.60),
    "claude-sonnet-4-6": (3.00, 15.00),
    "claude-haiku-4-5": (0.80, 4.00),
}

@contextmanager
def track_llm_call(model: str, user_id: int = None):
    """Usage: with track_llm_call("gpt-4o-mini"): ..."""
    response_holder = {}
    
    def hook(response):
        response_holder["response"] = response
    
    yield hook
    
    response = response_holder.get("response")
    if response and hasattr(response, "usage"):
        u = response.usage
        in_price, out_price = PRICES[model]
        cost = (u.prompt_tokens * in_price + u.completion_tokens * out_price) / 1_000_000
        
        logger.info(f"model={model} in={u.prompt_tokens} out={u.completion_tokens} "
                    f"cost=${cost:.6f} user={user_id}")

Интеграция с backend

Streaming chat endpoint в FastAPI (SSE)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    message: str
    session_id: str

async def stream_chat(messages: list):
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        stream=True,
    )
    
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            text = chunk.choices[0].delta.content
            yield f"data: {json.dumps({'text': text})}\n\n"
    
    yield "data: [DONE]\n\n"

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    history = await get_history(req.session_id)
    messages = history + [{"role": "user", "content": req.message}]
    
    return StreamingResponse(
        stream_chat(messages),
        media_type="text/event-stream",
    )

WebSocket chat

from fastapi import WebSocket

@app.websocket("/ws/chat")
async def chat_ws(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            data = await websocket.receive_json()
            messages = data["messages"]
            
            async with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=messages,
            ) as stream:
                async for text in stream.text_stream:
                    await websocket.send_json({"type": "delta", "text": text})
                
                await websocket.send_json({"type": "done"})
    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})
        await websocket.close()

Multi-provider abstraction

from abc import ABC, abstractmethod

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

class OpenAIProvider(LLMProvider):
    def __init__(self, model="gpt-4o-mini"):
        self.client = AsyncOpenAI()
        self.model = model
    
    async def chat(self, messages, **kwargs):
        response = await self.client.chat.completions.create(
            model=self.model, messages=messages, **kwargs)
        return response.choices[0].message.content

class AnthropicProvider(LLMProvider):
    def __init__(self, model="claude-sonnet-4-6"):
        from anthropic import AsyncAnthropic
        self.client = AsyncAnthropic()
        self.model = model
    
    async def chat(self, messages, **kwargs):
        # Отделить system message
        system = next((m["content"] for m in messages if m["role"] == "system"), None)
        msgs = [m for m in messages if m["role"] != "system"]
        
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=kwargs.pop("max_tokens", 1024),
            system=system,
            messages=msgs,
            **kwargs,
        )
        return response.content[0].text

# Usage
provider = OpenAIProvider("gpt-4o-mini")
# или
provider = AnthropicProvider("claude-haiku-4-5")

response = await provider.chat([{"role": "user", "content": "Привет"}])

Ресурсы

• OpenAI docs — platform.openai.com/docs
• Anthropic docs — docs.anthropic.com
• OpenAI Cookbook — cookbook.openai.com
• Anthropic Cookbook — GitHub
• LiteLLM — universal LLM wrapper: litellm.ai
• OpenRouter — один API для многих моделей: openrouter.ai

🏋️ Упражнения

🟢 Easy

1. “Hello World” с OpenAI и Anthropic API — 5 вопросов-ответов.
2. Получите streaming response, выводите каждый char отдельно.
3. Embedding для similarity между 2 предложениями.

🟡 Medium

1. Function calling: agent с 3 tool — weather, calculator, search.
2. Vision: загрузите изображение, извлеките structured data (Instructor + vision).
3. Prompt caching: 10 вопросов с большим system prompt — посмотрите разницу цены.

🔴 Hard

1. Multi-provider chat: OpenAI/Anthropic/Google — один abstraction, auto-fallback.
2. Cost-aware router: автоматический выбор подходящей модели по сложности input и размеру контекста.
3. Streaming chatbot: FastAPI + WebSocket + Postgres history + Redis caching.

Capstone

notebooks/month-05/03_llm_apis.ipynb:

• Полное знакомство с 3 provider (OpenAI, Anthropic, OpenRouter)
• Multi-turn chatbot со streaming
• Function calling — 5 tool
• Vision — image classification
• Cost tracking dashboard

✅ Чек-лист

• Знаю OpenAI и Anthropic API
• Использую Streaming responses
• Function calling / tool use
• Работа с Vision API
• Расчёт и сохранение Embeddings
• Prompt caching (Anthropic)
• Async batching
• Retry и rate limit handling
• Cost tracking и observability

Переходим к LangChain и LlamaIndex.