Structured Output

from openai import AsyncOpenAI
from pydantic import BaseModel
import json

client = AsyncOpenAI()

class SentimentResult(BaseModel):
    sentiment: str           # positive / negative / neutral
    confidence: float        # 0.0 - 1.0
    keywords: list[str]
    explanation: str

async def analyze_sentiment_json_mode(text: str) -> SentimentResult:
    """
    JSON mode: response_format={"type": "json_object"}
    Гарантирует валидный JSON, но не точную схему.
    Нужно описать схему в системном промпте.
    """
    response = await client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},  # ← JSON mode
        messages=[
            {
                "role": "system",
                "content": (
                    "Анализируй тональность текста. "
                    "Отвечай JSON со строго следующими полями: "
                    "sentiment (positive/negative/neutral), "
                    "confidence (число 0.0-1.0), "
                    "keywords (массив слов), "
                    "explanation (строка)."
                )
            },
            {"role": "user", "content": text},
        ],
        temperature=0,
    )
    raw = response.choices[0].message.content
    return SentimentResult.model_validate_json(raw)

from openai import AsyncOpenAI
from pydantic import BaseModel, Field
from typing import Literal

client = AsyncOpenAI()

class ExtractedEntity(BaseModel):
    text: str
    type: Literal["person", "organization", "location", "date", "money"]
    context: str

class NERResult(BaseModel):
    entities: list[ExtractedEntity]
    summary: str

async def extract_entities(text: str) -> NERResult:
    """
    Structured Outputs: parse() метод + Pydantic модель.
    Гарантирует 100% соответствие схеме через constrained decoding.
    """
    response = await client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",        # Structured Outputs: gpt-4o-2024-08-06+
        messages=[
            {
                "role": "system",
                "content": "Извлекай именованные сущности из текста."
            },
            {"role": "user", "content": text},
        ],
        response_format=NERResult,        # ← передаём Pydantic модель напрямую
        temperature=0,
    )

    # .parsed — уже провалидированная Pydantic-модель
    return response.choices[0].message.parsed

result = await extract_entities(
    "Илон Маск основал SpaceX в 2002 году в Хоторне, Калифорния. "
    "Компания привлекла 100 миллионов долларов на старте."
)
for e in result.entities:
    print(f"[{e.type}] {e.text}")

import anthropic
import json
from pydantic import BaseModel, Field
from typing import Any

client = anthropic.AsyncAnthropic()

def pydantic_to_tool(
    model: type[BaseModel],
    tool_name: str,
    description: str,
) -> dict:
    """Конвертируем Pydantic-модель в Anthropic tool schema."""
    schema = model.model_json_schema()
    # Убираем title на верхнем уровне (не нужен)
    schema.pop("title", None)
    return {
        "name": tool_name,
        "description": description,
        "input_schema": schema,
    }

async def structured_extract(
    text: str,
    output_model: type[BaseModel],
    task_description: str,
    tool_name: str = "extract_data",
) -> BaseModel:
    """
    Используем tool use как structured output.
    Модель вынуждена вызвать инструмент с валидными параметрами.
    """
    tool = pydantic_to_tool(output_model, tool_name, task_description)

    response = await client.messages.create(
        model="claude-opus-4-6",
        tools=[tool],
        tool_choice={"type": "tool", "name": tool_name},  # принудительный вызов
        messages=[{"role": "user", "content": text}],
        max_tokens=1024,
        temperature=0,
    )

    # Извлекаем tool_use блок
    tool_block = next(
        (b for b in response.content if b.type == "tool_use"),
        None,
    )
    if tool_block is None:
        raise ValueError("Model did not call the tool")

    return output_model.model_validate(tool_block.input)


# ── Пример 1: извлечение резюме ──
class ResumeData(BaseModel):
    name: str
    email: str | None = None
    phone: str | None = None
    skills: list[str] = Field(default_factory=list)
    years_experience: int | None = None
    current_position: str | None = None

resume_text = """
Иван Петров
Senior Python Developer | ivan@example.com | +7 916 123-45-67
10 лет опыта в backend-разработке.
Стек: Python, FastAPI, PostgreSQL, Redis, Docker, Kubernetes.
"""

resume = await structured_extract(
    text=resume_text,
    output_model=ResumeData,
    task_description="Извлеки контактные данные и навыки из резюме.",
)
print(resume.model_dump())


# ── Пример 2: классификация документа ──
from typing import Literal

class DocumentClassification(BaseModel):
    category: Literal["contract", "invoice", "report", "letter", "other"]
    subcategory: str | None = None
    language: Literal["ru", "en", "other"]
    urgency: Literal["low", "medium", "high", "critical"]
    summary: str = Field(description="Краткое содержание в 1-2 предложениях")
    action_required: bool

doc_text = "СЧЁТ-ФАКТУРА №1234 от 15.03.2025. Оплатить до 22.03.2025. Сумма: 150 000 руб."

classification = await structured_extract(
    text=doc_text,
    output_model=DocumentClassification,
    task_description="Классифицируй документ и определи необходимые действия.",
)
print(classification.model_dump())

from pydantic import BaseModel, Field
from typing import Literal
from datetime import date

# ── Плохая схема — мало подсказок ──
class BadSchema(BaseModel):
    s: str
    v: float
    t: str
    items: list

# ── Хорошая схема — богатые description ──
class OrderItem(BaseModel):
    product_name: str = Field(description="Название товара как в тексте")
    quantity: int     = Field(description="Количество единиц, целое число")
    unit_price: float = Field(description="Цена за единицу в рублях без НДС")

class OrderData(BaseModel):
    order_number: str = Field(
        description="Номер заказа или счёта, строка вида 'ORD-1234' или '№1234'"
    )
    order_date: str = Field(
        description="Дата заказа в формате YYYY-MM-DD"
    )
    customer_name: str = Field(
        description="Полное имя клиента или название компании"
    )
    items: list[OrderItem] = Field(
        description="Список позиций заказа"
    )
    total_amount: float = Field(
        description="Итоговая сумма в рублях"
    )
    currency: Literal["RUB", "USD", "EUR"] = Field(
        default="RUB",
        description="Валюта. Если не указана — RUB"
    )
    notes: str | None = Field(
        default=None,
        description="Дополнительные примечания или None если нет"
    )


# ── Советы по проектированию схем ──

# 1. Используй Literal для перечислений вместо str
# Плохо:  status: str  → модель может написать "Active", "active", "активный"
# Хорошо: status: Literal["active", "inactive", "pending"]

# 2. Явные форматы для дат и чисел
# Плохо:  date: str               → "15 марта", "15.03.25", "2025-03-15"
# Хорошо: date: str = Field(description="Дата в формате YYYY-MM-DD")

# 3. None вместо пустой строки для отсутствующих данных
# Плохо:  email: str  → будет "" или "не указан"
# Хорошо: email: str | None = None

# 4. Вложенные объекты лучше плоских для сложных структур
class AddressBad(BaseModel):
    city: str; street: str; building: str  # плоско, нет контекста

class Address(BaseModel):
    city: str    = Field(description="Название города")
    street: str  = Field(description="Улица с приставкой: ул., пр., бул.")
    building: str = Field(description="Номер дома, может включать корпус: '15к2'")

class PersonGood(BaseModel):
    name: str
    address: Address   # вложенный объект — модель понимает структуру

import anthropic
import json
from pydantic import BaseModel, ValidationError

client = anthropic.AsyncAnthropic()

async def extract_with_retry(
    text: str,
    output_model: type[BaseModel],
    system: str,
    max_retries: int = 3,
) -> BaseModel:
    """
    Structured extraction с retry.
    При ValidationError — показываем модели ошибку и просим исправить.
    """
    messages = [{"role": "user", "content": text}]

    for attempt in range(max_retries):
        response = await client.messages.create(
            model="claude-opus-4-6",
            system=system,
            messages=messages + [
                {"role": "assistant", "content": "{"}
            ],
            max_tokens=1024,
            temperature=0,
        )

        raw = "{" + response.content[0].text
        raw_json = extract_json(raw)

        if raw_json is None:
            # Нет JSON — просим переделать
            messages.append({"role": "assistant", "content": response.content[0].text})
            messages.append({
                "role": "user",
                "content": "Твой ответ не содержит валидного JSON. Ответь только JSON-объектом."
            })
            continue

        try:
            return output_model.model_validate_json(raw_json)

        except ValidationError as e:
            # Показываем ошибку валидации и просим исправить
            error_details = str(e)
            messages.append({"role": "assistant", "content": raw_json})
            messages.append({
                "role": "user",
                "content": (
                    f"JSON невалиден по схеме. Ошибки:\n{error_details}\n\n"
                    f"Исправь JSON. Схема:\n"
                    f"{json.dumps(output_model.model_json_schema(), ensure_ascii=False)}"
                )
            })

    raise RuntimeError(f"Failed to extract valid {output_model.__name__} after {max_retries} attempts")


# ── Более мощный вариант: используем instructor ──
# pip install instructor

import instructor
from openai import AsyncOpenAI

# instructor автоматически оборачивает клиент и добавляет retry+validation
oai = instructor.from_openai(AsyncOpenAI())
ant = instructor.from_anthropic(anthropic.AsyncAnthropic())

class ProductInfo(BaseModel):
    name: str
    price_rub: float
    in_stock: bool
    category: str

# OpenAI через instructor — автоматический retry при ValidationError
product_oai = await oai.chat.completions.create(
    model="gpt-4o",
    response_model=ProductInfo,          # ← вместо response_format
    messages=[{
        "role": "user",
        "content": "iPhone 15 Pro 256GB, цена 99 990 руб., есть в наличии, категория: смартфоны"
    }],
    max_retries=3,                       # instructor повторяет при ошибке
)

# Anthropic через instructor
product_ant = await ant.messages.create(
    model="claude-opus-4-6",
    response_model=ProductInfo,
    messages=[{
        "role": "user",
        "content": "Samsung Galaxy S24 Ultra, 89 990р, нет в наличии, электроника"
    }],
    max_tokens=512,
    max_retries=3,
)

print(product_oai.model_dump())
print(product_ant.model_dump())

import instructor
import anthropic
from pydantic import BaseModel

ant = instructor.from_anthropic(anthropic.AsyncAnthropic())

class ArticleAnalysis(BaseModel):
    title: str
    main_topic: str
    key_points: list[str]
    sentiment: str
    word_count_estimate: int
    tags: list[str]

async def stream_analysis(article_text: str):
    """Стримим структурированный анализ — поля появляются по мере генерации."""

    # create_partial возвращает async iterator частично заполненных объектов
    async for partial in ant.messages.create_partial(
        model="claude-opus-4-6",
        response_model=ArticleAnalysis,
        messages=[{
            "role": "user",
            "content": f"Проанализируй статью:\n\n{article_text}"
        }],
        max_tokens=1024,
    ):
        # partial — неполная модель, незаполненные поля = None
        if partial.title:
            print(f"\rЗаголовок: {partial.title}", end="")
        if partial.main_topic:
            print(f"\nТема: {partial.main_topic}", end="")
        if partial.key_points:
            print(f"\nПунктов: {len(partial.key_points)}", end="")

    print("\nГотово!")
    return partial   # последний элемент — полная модель

from pydantic import BaseModel, Field
from typing import Literal, Annotated, Union
import anthropic, instructor

ant = instructor.from_anthropic(anthropic.AsyncAnthropic())

# Разные типы событий агента
class ToolCallEvent(BaseModel):
    event_type: Literal["tool_call"] = "tool_call"
    tool_name: str
    arguments: dict
    reasoning: str = Field(description="Почему вызываем этот инструмент")

class ThoughtEvent(BaseModel):
    event_type: Literal["thought"] = "thought"
    content: str
    confidence: float = Field(ge=0.0, le=1.0)

class AnswerEvent(BaseModel):
    event_type: Literal["answer"] = "answer"
    content: str
    sources: list[str] = Field(default_factory=list)
    is_final: bool = True

# Discriminated union — Pydantic выбирает тип по event_type
AgentEvent = Annotated[
    Union[ToolCallEvent, ThoughtEvent, AnswerEvent],
    Field(discriminator="event_type")
]

class AgentStep(BaseModel):
    step_number: int
    event: AgentEvent

async def parse_agent_step(description: str) -> AgentStep:
    return await ant.messages.create(
        model="claude-opus-4-6",
        response_model=AgentStep,
        messages=[{
            "role": "user",
            "content": f"Преобразуй описание в структуру шага агента:\n{description}"
        }],
        max_tokens=512,
    )

step = await parse_agent_step(
    "Шаг 3: агент решает запросить погоду через weather_api для Москвы"
)
print(step.event.event_type)   # "tool_call"
print(step.event.tool_name)    # "weather_api"

import anthropic
import instructor
import logging
import time
from dataclasses import dataclass
from typing import TypeVar, Type
from pydantic import BaseModel

logger = logging.getLogger(__name__)
T = TypeVar("T", bound=BaseModel)

@dataclass
class ExtractionResult:
    data: BaseModel
    attempts: int
    elapsed_sec: float
    model_used: str

class DataExtractor:
    """
    Production-ready экстрактор структурированных данных из текста.
    Поддерживает Anthropic и OpenAI, автоматический retry, логирование.
    """

    def __init__(
        self,
        provider: str = "anthropic",
        model: str = "claude-opus-4-6",
        max_retries: int = 3,
    ):
        self.provider = provider
        self.model = model
        self.max_retries = max_retries

        if provider == "anthropic":
            self._client = instructor.from_anthropic(
                anthropic.AsyncAnthropic()
            )
        elif provider == "openai":
            from openai import AsyncOpenAI
            self._client = instructor.from_openai(AsyncOpenAI())
        else:
            raise ValueError(f"Unknown provider: {provider}")

    async def extract(
        self,
        text: str,
        schema: Type[T],
        system: str = "Извлекай структурированные данные из текста точно по схеме.",
        temperature: float = 0,
    ) -> ExtractionResult:
        t0 = time.monotonic()

        kwargs = dict(
            model=self.model,
            response_model=schema,
            messages=[{"role": "user", "content": text}],
            max_tokens=2048,
            max_retries=self.max_retries,
        )
        if self.provider == "anthropic":
            kwargs["system"] = system  # type: ignore
        else:
            kwargs["messages"] = [
                {"role": "system", "content": system},
                *kwargs["messages"],
            ]

        result = await self._client.messages.create(**kwargs) \
            if self.provider == "anthropic" \
            else await self._client.chat.completions.create(**kwargs)

        elapsed = time.monotonic() - t0
        logger.info("Extracted %s in %.2fs", schema.__name__, elapsed)

        return ExtractionResult(
            data=result,
            attempts=1,         # instructor отслеживает внутри
            elapsed_sec=elapsed,
            model_used=self.model,
        )


# Использование
extractor = DataExtractor(provider="anthropic")

class InvoiceData(BaseModel):
    invoice_number: str
    date: str
    vendor: str
    amount_total: float
    currency: str = "RUB"
    line_items: list[dict] = []

result = await extractor.extract(
    text="Счёт №INV-2025-042 от 20.03.2025. ООО 'Рога и Копыта'. Итого: 45 600 руб.",
    schema=InvoiceData,
)
print(f"Извлечено за {result.elapsed_sec:.2f}с:")
print(result.data.model_dump())