LLM API Integration: Интеграция с LLM API

Введение

Интеграция с LLM API — ключевой навык для разработки приложений с AI. Рассмотрим основные API, SDK и лучшие практики.

---

OpenAI API

Базовая интеграция

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# Chat Completions
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of France?"}
    ],
    temperature=0.7,
    max_tokens=500
)

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

Streaming responses

from openai import OpenAI

client = OpenAI()

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True
)

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

Async API

import asyncio
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()
    
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    
    print(response.choices[0].message.content)

asyncio.run(main())

---

Anthropic Claude API

Базовое использование

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1000,
    temperature=0,
    system="You are a helpful assistant specialized in technical writing.",
    messages=[
        {"role": "user", "content": "Explain how transformers work"}
    ]
)

print(message.content[0].text)

Streaming с Claude

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1000,
    messages=[{"role": "user", "content": "Write a detailed explanation"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    
    message = stream.get_final_message()

Tool use с Claude

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, e.g. 'Tokyo, Japan'"
                }
            },
            "required": ["location"]
        }
    },
    {
        "name": "calculate",
        "description": "Perform a calculation",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "Mathematical expression"
                }
            },
            "required": ["expression"]
        }
    }
]

message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1000,
    tools=tools,
    messages=[{"role": "user", "content": "What's the weather in Paris and what's 25 * 4?"}]
)

# Handle tool use
for content in message.content:
    if content.type == "tool_use":
        if content.name == "get_weather":
            weather = get_weather(content.input["location"])
            print(f"Weather: {weather}")
        elif content.name == "calculate":
            result = eval(content.input["expression"])
            print(f"Result: {result}")

---

Google Gemini API

Базовое использование

import google.generativeai as genai

genai.configure(api_key="your-api-key")

model = genai.GenerativeModel("gemini-2.0-flash")

response = model.generate_content("Explain quantum computing")
print(response.text)

Multi-modal input

import google.generativeai as genai
from PIL import Image

genai.configure(api_key="your-api-key")
model = genai.GenerativeModel("gemini-2.0-flash")

image = Image.open("diagram.png")

response = model.generate_content([
    "Explain this diagram",
    image
])

print(response.text)

Streaming с Gemini

import google.generativeai as genai

genai.configure(api_key="your-api-key")
model = genai.GenerativeModel("gemini-2.0-flash")

response = model.generate_content(
    "Write a detailed essay about AI",
    stream=True
)

for chunk in response:
    print(chunk.text, end="")

---

Meta Llama API

Использование через Fireworks AI

import openai

# Используем OpenAI SDK с Llama через Fireworks
client = openai.OpenAI(
    api_key="fireworks-api-key",
    base_url="https://api.fireworks.ai/inference/v1"
)

response = client.chat.completions.create(
    model="accounts/fireworks/models/llama-v3-70b-instruct",
    messages=[{"role": "user", "content": "Hello!"}],
    temperature=0.7,
    max_tokens=1000
)

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

Использование через Together AI

import openai

client = openai.OpenAI(
    api_key="together-api-key",
    base_url="https://api.together.xyz/v1"
)

response = client.chat.completions.create(
    model="meta-llama/Llama-3-70b-chat-hf",
    messages=[{"role": "user", "content": "Write code for a REST API"}]
)

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

---

Multi-provider abstraction

LiteLLM — унифицированный API

import litellm

# Единый API для всех провайдеров
response = litellm.completion(
    model="gpt-4",  # или "claude-sonnet-4", "gemini-2.0-flash"
    messages=[{"role": "user", "content": "Hello!"}]
)

# Автоматический fallback
try:
    response = litellm.completion(
        model="claude-sonnet-4",
        messages=[{"role": "user", "content": "Hello!"}]
    )
except litellm.exceptions.ContextWindowExceededError:
    # Fallback на другую модель
    response = litellm.completion(
        model="claude-haiku-3",
        messages=[{"role": "user", "content": "Hello!"}]
    )

Прокси сервер LiteLLM

# Запуск LiteLLM proxy
litellm --model ollama/llama3 --port 4000

# Или с несколькими моделями
litellm \
  --model gpt-4 claude-sonnet-4 gemini-2.0-flash \
  --port 4000 \
  --api_key "your-key"

# Использование с прокси
import openai

client = openai.OpenAI(
    api_key="any-key",
    base_url="http://localhost:4000/v1"
)

# Работает с любой моделью на прокси
response = client.chat.completions.create(
    model="gpt-4",  # или claude-sonnet-4, gemini-2.0-flash
    messages=[{"role": "user", "content": "Hello!"}]
)

---

Rate limiting и retry

Автоматический retry

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((RateLimitError, Timeout))
)
def call_llm_with_retry(prompt):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

Rate limiting

import time
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=10, period=60)  # 10 calls per minute
def call_llm(prompt):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )

# Batch processing с rate limiting
def batch_process(prompts, batch_size=5):
    results = []
    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i+batch_size]
        
        for prompt in batch:
            result = call_llm(prompt)
            results.append(result)
        
        if i + batch_size < len(prompts):
            time.sleep(60)  # Wait for rate limit reset
    
    return results

---

Cost optimization

Кэширование prompt

import hashlib
import json

prompt_cache = {}

def get_llm_response(prompt, model="gpt-4o"):
    # Hash prompt для кэша
    prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
    
    if prompt_hash in prompt_cache:
        return prompt_cache[prompt_hash]
    
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    
    result = response.choices[0].message.content
    prompt_cache[prompt_hash] = result
    
    return result

Выбор оптимальной модели

def smart_model_selection(task, prompt):
    """Выбор модели на основе типа задачи"""
    
    model_costs = {
        "gpt-4o": 0.000005,      # $5/M tokens
        "gpt-4o-mini": 0.00000015, # $0.15/M tokens
        "claude-sonnet-4": 0.000003,
        "claude-haiku-3": 0.00000025,
        "gemini-2.0-flash": 0.000000075,
    }
    
    # Simple tasks → cheaper model
    if is_simple_task(task, prompt):
        return "gemini-2.0-flash"
    elif is_creative_task(task, prompt):
        return "gpt-4o"
    elif is_code_task(task, prompt):
        return "claude-sonnet-4"
    else:
        return "gpt-4o"

def is_simple_task(task, prompt):
    simple_keywords = ["hello", "what is", "define", "translate"]
    return any(kw in prompt.lower() for kw in simple_keywords)

---

Error handling

Comprehensive error handling

from openai import OpenAI, APIError, RateLimitError, APIConnectionError

def safe_llm_call(prompt, model="gpt-4o", max_retries=3):
    client = OpenAI()
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.7
            )
            return response.choices[0].message.content
            
        except RateLimitError as e:
            wait_time = 2 ** (attempt + 1)
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
            
        except APIConnectionError as e:
            print(f"Connection error: {e}")
            time.sleep(2)
            
        except APIError as e:
            print(f"API error: {e}")
            return None
            
    return "Error: Max retries exceeded"

---

Best practices

1. System prompts

# Good system prompt
system_prompt = """You are an expert Python programmer.
You write clean, well-documented code following PEP 8.
You explain your reasoning before providing code.
You include error handling in all code examples."""

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": "Write a REST API endpoint"}
    ]
)

2. Structured output

from pydantic import BaseModel

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str
    relevance: float

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Search for AI papers"}],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "search_result",
            "schema": SearchResult.model_json_schema()
        }
    }
)

result = SearchResult.model_validate_json(response.choices[0].message.content)

3. Prompt versioning

PROMPTS = {
    "v1": "Explain the following concept",
    "v2": "Explain the following concept in simple terms with examples",
    "v3": "Explain the following concept. Include: 1) Definition 2) Key ideas 3) Examples 4) Common misconceptions"
}

def get_prompt(version: str = "v3") -> str:
    return PROMPTS.get(version, PROMPTS["v3"])

---

SDK comparison

SDK              | Models                    | Features
-----------------|---------------------------|------------------
OpenAI           | GPT-4, GPT-3.5, DALL-E    | Streaming, Tools
Anthropic        | Claude                    | Streaming, Tools, PDF
Google Gemini    | Gemini                    | Multi-modal, Streaming
LiteLLM          | All providers             | Unified API, Fallback
HuggingFace      | Open source models        | Inference API, Serverless

---

Итоги

Интеграция с LLM API требует внимания к rate limiting, cost optimization, error handling и выбору правильной модели для задачи. Используйте abstraction layers как LiteLLM для flexibility, и всегда implement proper retry logic.