L3.1. Python для AI-инженерии

Учебный проект Research Assistant, наборы данных для практикумов и эталонные решения выдаются ученикам курса. На сайте открыты теория, практикумы и задания.

Проект, а не скрипт

На L2 был один файл. AI-приложение — это пакет с зависимостями, конфигом, тестами и командой запуска. Минимальная структура:

filter-assistant/
  pyproject.toml          # зависимости и метаданные пакета
  src/filter_assistant/
    __init__.py
    config.py             # настройки из окружения
    llm.py                # клиент модели
  tests/
  .env.example            # пример переменных БЕЗ значений секретов
  .gitignore              # .env, .venv/, __pycache__/

pyproject.toml — единое место для зависимостей. Версии закрепляйте: AI-библиотеки меняются быстро, и «вчера работало» ломается от минорного обновления.

[project]
name = "filter-assistant"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
  "anthropic>=1.0,<2",
  "pydantic>=2.7,<3",
  "pydantic-settings>=2.3,<3",
]

[project.optional-dependencies]
dev = ["pytest>=8", "ruff"]

uv venv && source .venv/bin/activate       # Windows: .venv\Scripts\activate
uv pip install -e ".[dev]"                  # -e: правки в src видны без переустановки

Аннотации типов

Типы — это документация, которую проверяет инструмент. В AI-коде они особенно полезны: ответ модели — это dict неизвестной формы, и без типов ошибка всплывёт в трёх функциях от места, где её можно было поймать.

from dataclasses import dataclass

@dataclass(frozen=True)
class Chunk:
    doc_id: str
    text: str
    position: int

def top_k(chunks: list[Chunk], scores: list[float], k: int = 4) -> list[Chunk]:
    order = sorted(range(len(chunks)), key=lambda i: scores[i], reverse=True)
    return [chunks[i] for i in order[:k]]

X | None — значение может отсутствовать; list[str], dict[str, int] — коллекции; Literal["vector", "bm25"] — только эти строки.

Конфиг — это данные

Настройки (модель, провайдер, пути, лимиты) не зашивают в код. Их читают из переменных окружения и .env, валидируя типы. Так устроен референс:

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="FA_", env_file=".env", extra="ignore")
    llm_provider: str = "echo"          # echo | anthropic | openai_compat
    model: str = "claude-opus-5-5"
    request_timeout: float = 60.0
    max_concurrency: int = 4

settings = Settings()                   # FA_MODEL=... в окружении переопределит значение

Ключ API в Settings не кладут: SDK сам читает ANTHROPIC_API_KEY. Чем меньше мест, где лежит секрет, тем меньше мест, откуда он утечёт.

Ошибки: свои исключения и границы

Внутри пакета ловите только то, что умеете обработать. На границе (CLI, HTTP-эндпоинт) превращайте ошибки в понятный ответ.

class ExtractionError(Exception):
    """Модель не вернула валидные данные после всех попыток."""

def cli_main(path: str) -> int:
    try:
        result = process(path)
    except FileNotFoundError:
        print(f"нет файла: {path}"); return 2
    except ExtractionError as exc:
        print(f"не удалось извлечь поля: {exc}"); return 1
    print(result.model_dump_json(indent=2)); return 0

Async: зачем он в AI-коде

Вызов модели занимает секунды, и почти всё это время программа ждёт сеть. 100 документов по 3 секунды последовательно — 5 минут. Параллельно, по 8 запросов одновременно, — меньше минуты. asyncio позволяет ждать много ответов сразу в одном потоке.

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()
limit = asyncio.Semaphore(8)            # не больше 8 запросов одновременно — иначе 429

async def summarize(text: str) -> str:
    async with limit:
        resp = await client.messages.create(
            model="claude-opus-5-5",
            max_tokens=16000,
            output_config={"effort": "low"},     # простая задача — меньше рассуждений, дешевле
            messages=[{"role": "user", "content": f"Перескажи в 2 предложениях:\n{text}"}],
        )
        return next(b.text for b in resp.content if b.type == "text")

async def main(texts: list[str]) -> list[str | BaseException]:
    return await asyncio.gather(*(summarize(t) for t in texts), return_exceptions=True)

results = asyncio.run(main(["...", "..."]))

Три правила:

• Ограничивайте параллельность (Semaphore). Без ограничения вы упрётесь в лимиты провайдера и получите шквал ошибок 429.
• return_exceptions=True — сбой одного документа не отменяет остальные; ошибки разбираются после.
• Не смешивайте синхронные блокирующие вызовы (time.sleep, requests.get) с async-кодом: они останавливают весь цикл событий.

Практикум: async-клиент к двум API

Скрипт получает список из 20 городов и параллельно (не больше 5 одновременно) запрашивает: публичное API погоды без ключа и LLM — «одна фраза, что взять с собой в такую погоду». Требования: типы на всех функциях, настройки в Settings, таймауты на оба API, сбой одного города не роняет остальные, в конце — время выполнения и сравнение с последовательной версией.

Проверка: параллельная версия быстрее последовательной минимум в 3 раза; при отключённой сети скрипт печатает понятную ошибку по каждому городу и завершается.