LLM Providers

Интеграция с OpenAI и Anthropic для генерации текста.

Возможности

• Единый интерфейс для разных провайдеров
• Streaming ответов
• Function calling
• Подсчёт токенов
• Retry с backoff

OpenAI

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

from kit.providers.llm import OpenAIProvider

provider = OpenAIProvider(api_key="sk-...")

response = await provider.generate(
    prompt="Напиши короткую историю о роботе",
    model="gpt-4o",
    max_tokens=500,
    temperature=0.7
)

print(response.content)
print(f"Tokens: {response.usage.total_tokens}")

Streaming

async for chunk in provider.stream(
    prompt="Объясни квантовую механику",
    model="gpt-4o"
):
    print(chunk, end="", flush=True)

С системным промптом

response = await provider.generate(
    prompt="Как приготовить борщ?",
    system="Ты опытный шеф-повар. Отвечай кратко и по делу.",
    model="gpt-4o-mini"
)

Chat формат

messages = [
    {"role": "system", "content": "Ты полезный ассистент"},
    {"role": "user", "content": "Привет!"},
    {"role": "assistant", "content": "Здравствуйте! Чем могу помочь?"},
    {"role": "user", "content": "Расскажи о Python"}
]

response = await provider.chat(messages, model="gpt-4o")

Function Calling

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        }
    }
]

response = await provider.generate(
    prompt="Какая погода в Москве?",
    model="gpt-4o",
    tools=tools
)

if response.tool_calls:
    for call in response.tool_calls:
        print(f"Function: {call.name}, Args: {call.arguments}")

Anthropic

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

from kit.providers.llm import AnthropicProvider

provider = AnthropicProvider(api_key="sk-ant-...")

response = await provider.generate(
    prompt="Напиши хайку о программировании",
    model="claude-3-5-sonnet-20241022",
    max_tokens=200
)

print(response.content)

Streaming

async for chunk in provider.stream(
    prompt="Расскажи длинную историю",
    model="claude-3-5-sonnet-20241022"
):
    print(chunk, end="")

С изображением

response = await provider.generate(
    prompt="Что на этом изображении?",
    model="claude-3-5-sonnet-20241022",
    images=["path/to/image.jpg"]  # или base64
)

API Reference

LLMResponse

@dataclass
class LLMResponse:
    content: str
    model: str
    usage: TokenUsage
    tool_calls: List[ToolCall] = None
    finish_reason: str = None

@dataclass
class TokenUsage:
    input_tokens: int
    output_tokens: int
    total_tokens: int

OpenAIProvider

class OpenAIProvider:
    def __init__(
        self,
        api_key: str,
        base_url: str = None,
        default_model: str = "gpt-4o",
        timeout: float = 60.0,
        max_retries: int = 3
    )

    async def generate(
        self,
        prompt: str,
        model: str = None,
        system: str = None,
        max_tokens: int = None,
        temperature: float = 0.7,
        tools: List[dict] = None
    ) -> LLMResponse

    async def chat(
        self,
        messages: List[dict],
        model: str = None,
        **kwargs
    ) -> LLMResponse

    async def stream(
        self,
        prompt: str,
        model: str = None,
        **kwargs
    ) -> AsyncGenerator[str, None]

    async def close(self) -> None

AnthropicProvider

class AnthropicProvider:
    def __init__(
        self,
        api_key: str,
        default_model: str = "claude-3-5-sonnet-20241022",
        timeout: float = 60.0,
        max_retries: int = 3
    )

    async def generate(
        self,
        prompt: str,
        model: str = None,
        system: str = None,
        max_tokens: int = 4096,
        temperature: float = 0.7,
        images: List[str] = None
    ) -> LLMResponse

    async def stream(
        self,
        prompt: str,
        model: str = None,
        **kwargs
    ) -> AsyncGenerator[str, None]

    async def close(self) -> None

Примеры из production

Autoshorts — генерация сценария

class ScriptGenerator:
    def __init__(self):
        self.llm = OpenAIProvider(api_key=settings.openai_api_key)

    async def generate(self, topic: str, style: str) -> dict:
        system = """
        Ты сценарист для коротких видео.
        Создавай сценарии из 5-7 сцен.
        Каждая сцена: текст для озвучки + описание визуала.
        """

        prompt = f"Тема: {topic}\nСтиль: {style}"

        response = await self.llm.generate(
            prompt=prompt,
            system=system,
            model="gpt-4o",
            temperature=0.8
        )

        return self.parse_script(response.content)

DedMoroz.ai — персонализированные поздравления

class GreetingGenerator:
    def __init__(self):
        self.llm = AnthropicProvider(api_key=settings.anthropic_api_key)

    async def generate(self, name: str, wishes: str) -> str:
        system = """
        Ты Дед Мороз. Создавай тёплые, добрые поздравления.
        Используй имя получателя. Упоминай его пожелания.
        Длина: 3-4 предложения.
        """

        prompt = f"Имя: {name}\nПожелания: {wishes}"

        response = await self.llm.generate(
            prompt=prompt,
            system=system,
            model="claude-3-5-sonnet-20241022",
            max_tokens=300
        )

        return response.content