Часть 1: JSON

JSON — универсальный язык общения в мире API. LLM возвращает JSON, tool calling принимает JSON, состояние агента сохраняется как JSON. Нужно уметь уверенно работать с ним в обе стороны.

Стандартная библиотека json

Базовый json — dumps и loads
python 
import json

# Python объект → JSON строка
data = {
    "model": "gpt-4o",
    "messages": [
        {"role": "user", "content": "Привет!"}
    ],
    "temperature": 0.7,
    "stream": False,
}

json_str = json.dumps(data)
print(json_str)
# {"model": "gpt-4o", "messages": [{"role": "user", "content": "Привет!"}], ...}

# Красивый вывод с отступами — для отладки и логов
pretty = json.dumps(data, indent=2, ensure_ascii=False)
print(pretty)
# {
#   "model": "gpt-4o",
#   "messages": [
#     {
#       "role": "user",
#       "content": "Привет!"
#     }
#   ],
#   ...
# }

# JSON строка → Python объект
parsed = json.loads(json_str)
print(type(parsed))           # <class 'dict'>
print(parsed["model"])        # "gpt-4o"
print(parsed["messages"][0])  # {"role": "user", "content": "Привет!"}

Чтение и запись JSON-файлов

Агенты часто сохраняют состояние, историю диалогов или кэш в файл:

Работа с JSON-файлами
python 
import json
from pathlib import Path

# Запись в файл
state = {
    "session_id": "abc-123",
    "messages": [{"role": "user", "content": "Привет"}],
    "step": 3,
}

state_file = Path("agent_state.json")

with open(state_file, "w", encoding="utf-8") as f:
    json.dump(state, f, indent=2, ensure_ascii=False)

# Чтение из файла
with open(state_file, "r", encoding="utf-8") as f:
    loaded = json.load(f)

print(loaded["session_id"])  # "abc-123"

# Краткий паттерн через Path (Python 3.10+)
state_file.write_text(
    json.dumps(state, indent=2, ensure_ascii=False),
    encoding="utf-8"
)

loaded = json.loads(state_file.read_text(encoding="utf-8"))

Даты, datetime и кастомные типы

Стандартный json не умеет сериализовать datetime, UUID и другие типы — нужен кастомный энкодер:

Кастомный JSONEncoder для datetime и UUID
python 
import json
from datetime import datetime
from uuid import UUID, uuid4

# ❌ Стандартный json упадёт
data = {"id": uuid4(), "created_at": datetime.now(), "value": 42}
# json.dumps(data)  # TypeError: Object of type UUID is not JSON serializable

# ✅ Кастомный encoder
class AgentJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()         # "2025-03-22T10:00:00"
        if isinstance(obj, UUID):
            return str(obj)                # "a1b2c3d4-..."
        if hasattr(obj, "model_dump"):     # Pydantic модели!
            return obj.model_dump()
        return super().default(obj)

result = json.dumps(data, cls=AgentJSONEncoder, ensure_ascii=False)
print(result)
# {"id": "a1b2c3d4-...", "created_at": "2025-03-22T10:00:00", "value": 42}

# Совет: если используешь Pydantic — просто вызывай model.model_dump_json()
# Он умеет обрабатывать datetime и UUID автоматически

Извлечение JSON из текста LLM

LLM часто оборачивает JSON в markdown-блоки или добавляет пояснения. Нужно надёжно его извлечь:

Устойчивый парсер JSON из ответа LLM
python 
import json
import re
from typing import Any

def extract_json(text: str) -> Any | None:
    """
    Извлекает JSON из текста LLM.
    Обрабатывает: ```json ... ```, ``` ... ```, голый JSON.
    """
    # Приоритет 1: блок ```json ... ```
    match = re.search(r"```json\s*(.*?)\s*```", text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1))
        except json.JSONDecodeError:
            pass

    # Приоритет 2: любой ``` ... ```
    match = re.search(r"```\s*(.*?)\s*```", text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1))
        except json.JSONDecodeError:
            pass

    # Приоритет 3: ищем первый { ... } или [ ... ] в тексте
    match = re.search(r"(\{.*\}|\[.*\])", text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1))
        except json.JSONDecodeError:
            pass

    # Приоритет 4: весь текст как JSON
    try:
        return json.loads(text.strip())
    except json.JSONDecodeError:
        return None

# Тест с разными форматами ответов LLM
responses = [
    # Вариант 1: блок с языком
    'Вот результат анализа:\n```json\n{"action": "search", "query": "python"}\n```',

    # Вариант 2: голый JSON
    '{"action": "search", "query": "python"}',

    # Вариант 3: JSON внутри текста
    'Агент решил выполнить: {"action": "search", "query": "python"} для поиска.',

    # Вариант 4: невалидный
    "Не знаю как ответить.",
]

for resp in responses:
    result = extract_json(resp)
    print(f"Результат: {result}")
✅ Лучший способ — Structured Outputs

Если провайдер поддерживает Structured Outputs (OpenAI) или tool calling (OpenAI, Anthropic) — используй их вместо ручного парсинга. Модель гарантированно вернёт валидный JSON по схеме. Ручной парсинг нужен только как fallback или со старыми моделями.

Слияние и обновление JSON-объектов

Частый паттерн в агентах — объединение конфигов или обновление состояния:

Слияние словарей — паттерны для конфигов
python 
# Базовый конфиг агента
base_config = {
    "model": "gpt-4o-mini",
    "temperature": 0.7,
    "max_tokens": 2000,
    "tools_enabled": True,
}

# Переопределения для конкретной задачи
overrides = {
    "model": "gpt-4o",   # Заменяем модель
    "temperature": 0.2,  # Снижаем температуру для точности
}

# Поверхностное слияние (shallow merge)
config = {**base_config, **overrides}
print(config["model"])       # "gpt-4o"  ← из overrides
print(config["max_tokens"])  # 2000      ← из base_config

# Глубокое слияние вложенных dict
def deep_merge(base: dict, override: dict) -> dict:
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result

nested_base = {"llm": {"model": "gpt-4o-mini", "temperature": 0.7}, "rag": {"top_k": 5}}
nested_override = {"llm": {"model": "gpt-4o"}}  # Только модель

merged = deep_merge(nested_base, nested_override)
print(merged)
# {"llm": {"model": "gpt-4o", "temperature": 0.7}, "rag": {"top_k": 5}}
# temperature сохранилась!

Часть 2: Переменные окружения

Переменные окружения — стандартный способ передачи конфигурации в приложение. API-ключи, URL баз данных, флаги режима работы — всё это должно жить в окружении, а не в коде.

os.environ — стандартный доступ

os.environ — базовая работа с окружением
python 
import os

# Чтение переменной — KeyError если не установлена
api_key = os.environ["OPENAI_API_KEY"]

# Безопасное чтение с дефолтным значением
model = os.environ.get("AGENT_MODEL", "gpt-4o-mini")

# Проверка наличия
if "ANTHROPIC_API_KEY" not in os.environ:
    print("Anthropic API ключ не настроен")

# Установка переменной (только для текущего процесса!)
os.environ["DEBUG"] = "true"

# Все переменные окружения — os.environ это dict-like объект
for key, value in os.environ.items():
    if "KEY" in key:  # Осторожно — не логируй секреты!
        print(f"{key}=***")  # Маскируй значения!
🚫 Никогда не хардкодь секреты

Код живёт в Git. Git — это история. Даже если ты удалишь ключ в следующем коммите, он останется в истории и может быть скомпрометирован. GitHub активно сканирует публичные репозитории на API-ключи и автоматически отзывает найденные.

❌ НИКОГДА ТАК:

client = OpenAI(api_key="sk-proj-AbCd1234...")  # Хардкод в коде!

✅ ПРАВИЛЬНО:

client = OpenAI()  # Читает OPENAI_API_KEY из окружения автоматически

.env файлы и python-dotenv

Хранить переменные окружения в файле .env — стандарт для локальной разработки. python-dotenv загружает их в os.environ при старте:

Структура .env файла

.env — конфигурация проекта
bash 
# .env
# ⚠️ Этот файл НЕ должен попадать в Git!

# --- LLM API ключи ---
OPENAI_API_KEY=sk-proj-...
ANTHROPIC_API_KEY=sk-ant-api03-...
GOOGLE_API_KEY=AIzaSy...

# --- Инструменты агента ---
TAVILY_API_KEY=tvly-...
SERP_API_KEY=...

# --- Базы данных ---
POSTGRES_URL=postgresql://user:password@localhost:5432/agent_db
REDIS_URL=redis://localhost:6379/0
QDRANT_URL=http://localhost:6333

# --- Настройки агента ---
AGENT_MODEL=gpt-4o
AGENT_MAX_ITERATIONS=15
AGENT_TEMPERATURE=0.7
AGENT_MEMORY_ENABLED=true

# --- Режим работы ---
DEBUG=false
LOG_LEVEL=INFO
ENVIRONMENT=development  # development | staging | production
.gitignore — .env никогда не в Git
bash 
# .gitignore — добавь в любой AI-проект
.env
.env.local
.env.*.local

# Но шаблон конфига — коммитим, чтобы коллеги знали что настраивать
# .env.example — ДА
# .env — НЕТ
.env.example — шаблон без секретов
bash 
# .env.example — коммитим в Git как документацию
# Скопируй в .env и заполни реальными значениями

OPENAI_API_KEY=               # Получить на platform.openai.com
ANTHROPIC_API_KEY=            # Получить на console.anthropic.com
TAVILY_API_KEY=               # Получить на tavily.com

AGENT_MODEL=gpt-4o-mini
AGENT_MAX_ITERATIONS=10
DEBUG=false

Загрузка .env в Python

python-dotenv — загрузка переменных окружения
python 
from dotenv import load_dotenv
import os

# Загружаем .env — вызывать один раз при старте приложения
load_dotenv()

# Теперь переменные доступны через os.environ
api_key = os.environ["OPENAI_API_KEY"]
model = os.environ.get("AGENT_MODEL", "gpt-4o-mini")

# Явно указать путь к файлу (если .env не в текущей директории)
load_dotenv(".env.production")
load_dotenv("/path/to/project/.env")

# Найти .env автоматически по дереву директорий
from dotenv import find_dotenv
load_dotenv(find_dotenv())  # Ищет .env вверх по дереву от текущей директории

# override=True — переменные из .env перезаписывают системные
load_dotenv(override=True)

# Проверить, что всё загрузилось
from dotenv import dotenv_values
config = dotenv_values(".env")   # Возвращает dict, не трогая os.environ
print(config.keys())

Приведение типов из окружения

Все переменные окружения — строки. Нужно явно приводить типы:

Приведение типов переменных окружения
python 
import os
from dotenv import load_dotenv

load_dotenv()

# ❌ Частая ошибка: строка вместо числа
max_iter = os.environ.get("AGENT_MAX_ITERATIONS", "10")
# for _ in range(max_iter):  # TypeError: 'str' object cannot be interpreted as integer

# ✅ Явное приведение типов
max_iter = int(os.environ.get("AGENT_MAX_ITERATIONS", "10"))
temperature = float(os.environ.get("AGENT_TEMPERATURE", "0.7"))

# ✅ Bool — нужно быть аккуратным
raw_debug = os.environ.get("DEBUG", "false")
# bool("false") == True ← строка не пустая!
debug = raw_debug.lower() in ("true", "1", "yes", "on")  # Правильный способ

# Удобная функция для bool переменных
def get_bool_env(key: str, default: bool = False) -> bool:
    val = os.environ.get(key, str(default)).strip().lower()
    return val in ("true", "1", "yes", "on")

debug = get_bool_env("DEBUG")
memory_enabled = get_bool_env("AGENT_MEMORY_ENABLED", default=True)

# ✅ Список из ENV (разделитель запятая)
raw_tools = os.environ.get("ENABLED_TOOLS", "web_search,calculator")
enabled_tools = [t.strip() for t in raw_tools.split(",") if t.strip()]
print(enabled_tools)  # ["web_search", "calculator"]

Часть 3: pydantic-settings — правильный подход

Ручное приведение типов из os.environ быстро становится громоздким. pydantic-settings автоматизирует это: читает переменные окружения и .env, приводит типы, валидирует значения.

ℹ️ Связь с предыдущим уроком

В уроке по Pydantic мы уже видели BaseSettings в конце. Здесь разберём его детально — это лучший способ конфигурировать AI-агентов в реальных проектах.

pip install pydantic-settings
bash 
pip install pydantic-settings
config.py — полная конфигурация агента
python 
# config.py
from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal
from functools import lru_cache

class Settings(BaseSettings):
    """Полная конфигурация AI-агента из переменных окружения."""

    model_config = SettingsConfigDict(
        env_file=".env",               # Загружать из .env автоматически
        env_file_encoding="utf-8",
        case_sensitive=False,          # OPENAI_API_KEY == openai_api_key
        extra="ignore",                # Игнорировать неизвестные переменные
    )

    # --- API ключи (SecretStr скрывает значение в логах) ---
    openai_api_key: SecretStr = Field(description="OpenAI API key")
    anthropic_api_key: SecretStr | None = Field(default=None)
    tavily_api_key: SecretStr | None = Field(default=None)

    # --- Настройки LLM ---
    agent_model: str = Field(default="gpt-4o-mini")
    agent_temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    agent_max_tokens: int = Field(default=4096, gt=0)
    agent_max_iterations: int = Field(default=10, ge=1, le=100)

    # --- Память и хранилище ---
    redis_url: str = Field(default="redis://localhost:6379/0")
    vector_db_url: str = Field(default="http://localhost:6333")
    vector_db_collection: str = Field(default="agent_memory")

    # --- Режим работы ---
    environment: Literal["development", "staging", "production"] = "development"
    debug: bool = False
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"

    # --- Валидация ---
    @field_validator("agent_model")
    @classmethod
    def validate_model(cls, v: str) -> str:
        known_prefixes = ("gpt-", "claude-", "gemini-", "mistral-", "llama")
        if not any(v.startswith(p) for p in known_prefixes):
            raise ValueError(f"Неизвестная модель: {v}")
        return v

    # --- Удобные методы ---
    def openai_key(self) -> str:
        return self.openai_api_key.get_secret_value()

    def anthropic_key(self) -> str | None:
        return self.anthropic_api_key.get_secret_value() if self.anthropic_api_key else None

    def is_production(self) -> bool:
        return self.environment == "production"

# Синглтон — создаём один раз при старте приложения
@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()


# Использование в любом модуле:
# from config import get_settings
# settings = get_settings()
# print(settings.agent_model)  # "gpt-4o-mini"

Несколько окружений: dev / staging / prod

Разные .env файлы для разных окружений
python 
import os
from pydantic_settings import BaseSettings, SettingsConfigDict

# Структура файлов:
# .env                 ← база (не коммитим)
# .env.development     ← локальная разработка (не коммитим)
# .env.production      ← прод настройки (не коммитим!)
# .env.example         ← шаблон (коммитим)

env = os.environ.get("ENVIRONMENT", "development")

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        # Порядок приоритетов: env vars > .env.{environment} > .env
        env_file=[".env", f".env.{env}"],
        env_file_encoding="utf-8",
    )
    agent_model: str = "gpt-4o-mini"
    debug: bool = False

settings = Settings()

# Запуск с нужным окружением:
# ENVIRONMENT=production python agent.py
# ENVIRONMENT=staging python agent.py

Структура проекта: как организовать конфиги

Рекомендуемая структура для AI-агента:

my-agent/
├── .env               ← секреты, не в Git
├── .env.example       ← шаблон, в Git
├── .gitignore         ← .env в игноре
│
├── config.py          ← Settings(BaseSettings)
├── agent.py           ← from config import get_settings
├── tools.py           ← from config import get_settings
│
├── data/
│   ├── agent_state.json   ← сохранённое состояние
│   └── memory.json        ← долгосрочная память
│
└── tests/
    └── .env.test          ← тестовые переменные

Валидация конфига при старте

Лучше узнать о проблеме конфигурации сразу, а не когда агент уже начал работу:

Проверка конфигурации при запуске
python 
# main.py — точка входа агента
import sys
from pydantic import ValidationError
from config import get_settings

def check_config() -> None:
    """Проверяем конфиг при старте. Падаем явно с понятным сообщением."""
    try:
        settings = get_settings()
    except ValidationError as e:
        print("❌ Ошибка конфигурации:")
        for err in e.errors():
            field = err["loc"][0]
            msg = err["msg"]
            print(f"   {field}: {msg}")
        print("\nПроверь .env файл (шаблон: .env.example)")
        sys.exit(1)

    # Проверяем доступность опциональных сервисов
    warnings = []
    if not settings.anthropic_api_key:
        warnings.append("ANTHROPIC_API_KEY не задан — Claude недоступен")
    if not settings.tavily_api_key:
        warnings.append("TAVILY_API_KEY не задан — веб-поиск недоступен")

    for w in warnings:
        print(f"⚠️  {w}")

    print(f"✅ Конфиг загружен: model={settings.agent_model}, env={settings.environment}")

if __name__ == "__main__":
    check_config()
    # ... запуск агента

JSON для сохранения состояния агента

Агентам нужна персистентность: сохранять историю диалогов, промежуточные результаты, долгосрочную память. JSON + файловая система — простейшее решение для старта:

Простое хранилище состояния агента
python 
import json
from pathlib import Path
from datetime import datetime
from typing import Any

class AgentStateStore:
    """Простое JSON-хранилище для состояния агента."""

    def __init__(self, data_dir: str = "data"):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(exist_ok=True)

    def _path(self, session_id: str) -> Path:
        return self.data_dir / f"session_{session_id}.json"

    def save(self, session_id: str, state: dict) -> None:
        """Сохраняет состояние сессии."""
        state["_saved_at"] = datetime.now().isoformat()
        self._path(session_id).write_text(
            json.dumps(state, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )

    def load(self, session_id: str) -> dict | None:
        """Загружает состояние сессии. None если не найдена."""
        path = self._path(session_id)
        if not path.exists():
            return None
        return json.loads(path.read_text(encoding="utf-8"))

    def update(self, session_id: str, updates: dict) -> dict:
        """Загружает состояние, обновляет поля, сохраняет."""
        state = self.load(session_id) or {}
        state.update(updates)
        self.save(session_id, state)
        return state

    def append_message(self, session_id: str, role: str, content: str) -> None:
        """Добавляет сообщение в историю диалога."""
        state = self.load(session_id) or {"messages": []}
        state.setdefault("messages", [])
        state["messages"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
        })
        self.save(session_id, state)

    def list_sessions(self) -> list[str]:
        """Список всех сохранённых сессий."""
        return [
            p.stem.replace("session_", "")
            for p in self.data_dir.glob("session_*.json")
        ]

# Использование
store = AgentStateStore()

# Новая сессия
store.save("abc-123", {"goal": "Исследовать тему asyncio", "steps": []})

# Добавляем сообщения
store.append_message("abc-123", "user", "Расскажи об asyncio")
store.append_message("abc-123", "assistant", "Asyncio — это библиотека...")

# Обновляем состояние
store.update("abc-123", {"current_step": 2, "status": "running"})

# Загружаем
state = store.load("abc-123")
print(f"Цель: {state['goal']}")
print(f"Сообщений: {len(state['messages'])}")
ℹ️ Когда перейти с файлов на БД

JSON-файлы отлично работают для прототипирования и одного пользователя. Когда нужны: одновременные сессии, поиск по истории, надёжность — переходи на Redis (для сессий) или PostgreSQL (для долгосрочного хранения). LangGraph поддерживает оба варианта через checkpointers.

Шпаргалка

JSON + ENV cheatsheet
python 
import json, os
from pathlib import Path
from dotenv import load_dotenv
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import SecretStr

# ── JSON ─────────────────────────────────────────────
# str → dict
data = json.loads('{"key": "value"}')

# dict → str
text = json.dumps(data, indent=2, ensure_ascii=False)

# file → dict
data = json.loads(Path("file.json").read_text(encoding="utf-8"))

# dict → file
Path("file.json").write_text(
    json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8"
)

# Извлечь JSON из текста LLM (```json...``` или голый JSON)
import re
match = re.search(r"```json\s*(.*?)\s*```", llm_text, re.DOTALL)
data = json.loads(match.group(1)) if match else json.loads(llm_text.strip())

# ── ENV ──────────────────────────────────────────────
load_dotenv()                                    # Загрузить .env

api_key = os.environ["OPENAI_API_KEY"]           # KeyError если нет
model = os.environ.get("MODEL", "gpt-4o-mini")  # С дефолтом
count = int(os.environ.get("MAX_ITER", "10"))    # Приведение типа
debug = os.environ.get("DEBUG", "false").lower() in ("true", "1", "yes")

# ── PYDANTIC SETTINGS (лучший способ) ────────────────
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    openai_api_key: SecretStr       # Читает OPENAI_API_KEY
    agent_model: str = "gpt-4o"    # Читает AGENT_MODEL с дефолтом
    debug: bool = False             # Читает DEBUG, "false" → False

from functools import lru_cache

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()

# ── БЕЗОПАСНОСТЬ ─────────────────────────────────────
# .gitignore: .env, .env.local, .env.*.local
# Коммитить: .env.example (без значений)
# Не логировать: SecretStr, api_key, password
# SecretStr: key.get_secret_value() для использования, repr показывает *****

Практическое задание

Задание: конфигурация и хранилище для агента

  1. Создай файл .env.example для AI-агента: API ключи, модель, параметры, флаги режима. Добавь .env в .gitignore
  2. Напиши Settings(BaseSettings) с 5+ полями разных типов (str, int, float, bool, SecretStr). Проверь, что при отсутствии обязательного поля получишь ValidationError с понятным сообщением
  3. Напиши функцию extract_json(text: str), которая надёжно извлекает JSON из разных форматов ответа LLM (с блоком ```json``` и без)
  4. Реализуй класс SimpleStateStore: методы save(), load(), append_message(). Сохрани 3 диалога, загрузи один и распечатай историю
  5. Добавь в main.py проверку конфига при старте с понятными сообщениями об ошибках

Что дальше

🌐

HTTP-клиенты: httpx

Async HTTP-запросы к внешним API в агентах

 ← 📦

Pydantic: типы и валидация

Предыдущая тема: строгая типизация данных

 ← ⚡

Asyncio и async/await

Первая тема модуля: параллельность

 🤖

Модуль 02: Ядро агентов

Tool calling, ReAct, Plan-and-Execute