Такое вот бывает если открыть порты бд на сервере)

Такая вот оказывается есть годнота по базе LLM

Небольшое открытие про rate limits

Недавно понял, что всегда реализовывал rate-limit-ы и прочие ограничения у внешних сервисов вручную - счётчики , sleep-ы, retry-логика и т.п.
И только сейчас заметил, что для Python существует целый класс библиотек, решающих это из коробки.

Более того, есть shared rate limiters — которые используют общую БД или Redis для синхронизации лимитов между несколькими инстансами сервиса.

Наиболее интересные варианты, которые нашёл:
limits - самая популярная и довольно простая
asyncio-limiter - асинхронная и довольно низкоуровневая, если хочется что то сложное дописать поверх
redis-rate-limiters - понравилась больше всего, две довольно понятных концепции работы с токенами + атомарность


еще немного удивился что не нашел принципиально других лимитеров - кажется на поверхности лежит идея делать рейт-лимиты не для python-методов апи, а для доменов/урлов, например сделать либу-надстройку над httpx, которая позволяет клиенту httpx передать конфиг с рейт-лимитами по урлам, чтобы в коде дальше вообще про это не думать

Новая рубрика #нейрохрючево

Буду делиться самыми смешными ai-gen штуками, которые встречаю в работе.

Выпуск первый.
LLM генерила тесты и забыла поставить autocommit=True. Чтобы пофиксить — полезла в бизнес-логику:

if hasattr(self.repo._session, 'commit'):
    await self.repo._session.commit()

#нейрохрючево
системник

### No hasattr()
# ❌
if hasattr(obj, "model"):
    model = obj.model

КОДИК:

    def from_litellm(cls, response: Union[ModelResponse, ResponsesAPIResponse]) -> 'LLMResponse':
        if isinstance(response, ResponsesAPIResponse):
            model = response.model
            content = ""
            structured_data = None
            
            # Extract content from output
            if response.output:
                for item in response.output:
                    if hasattr(item, 'type') and item.type == 'message':
                        # item is usually a ChatCompletionAssistantMessageParam-like object
                        # or a dict if it was dumped.
                        if hasattr(item, 'content'):
                            if isinstance(item.content, list):
                                for part in item.content:
                                    if hasattr(part, 'type') and part.type == 'output_text':
                                        content += part.text
                                    elif isinstance(part, dict) and part.get('type') == 'output_text':
                                        content += part.get('text', '')
                            elif isinstance(item.content, str):
                                content += item.content
                        elif isinstance(item, dict) and 'content' in item:
                            if isinstance(item['content'], list):
                                for part in item['content']:
                                    if part.get('type') == 'output_text':
                                        content += part.get('text', '')
                            elif isinstance(item['content'], str):
                                content += item['content']

Снял видос для ШВМ
https://www.youtube.com/watch?v=OZkXxuRuY1k&t=4s

За последние пол года активно работал с голосовыми агентами и открыл для себя pipecat
что то очень похожее на fastapi для api - очень удобный оркестратор для voice-pipelin-а
плюсом куча коннекторов и возможность кастомно докидывать любые из кубиков


общий пайплайн для voice-ассистентов обычно выглядит так:

микрофон → STT → LLM → TTS → динамик


в pipecat для этого есть удобные абстракции:

Frame - пакет данных: контент или event
FrameProcessor - блок работы с Frame
Pipeline - последовательность FrameProcessor
Transport - связь агента с браузером, мобильным клиентом/keycloak/внешними коннекторами
PipelineTask - life-cicle пайплайна
PipelineRunner - сам запуск


по коду в итоге все очень легко:

завели pipeline - основная сущность, в нем все степы:

pipeline = Pipeline(
    [
        transport.input(),
        stt_service,
        user_aggregator,
        llm_service,
        tts_service,
        transport.output(),
        assistant_aggregator,
    ]
)

создали из него task:

task = PipelineTask(
    pipeline,
    params=PipelineParams(
        enable_metrics=True,
        enable_usage_metrics=True,
    ),
)

в таск накидываем event-handlerov - on_start/on_join/on_error и тп:

@task.event_handler("on_pipeline_started")
async def on_pipeline_started(task, frame):
    ...

и просто запускаем все вместе:

runner = PipelineRunner(handle_sigint=handle_sigint)
await runner.run(task)

по сути главное - выбрать и реализовать(а чаще - просто подключить) step-ы Pipeline-а, продумать логику on_event и по сути все!
и это только базовое использование sdk - pipecat еще умеет в cloud, у него есть свой VAD и еще много чего

Три недели как тыкаю max-подписку chatgpt - начинал с sol с high/ultra ризонинга, теперь добрался до Terra и даже до Luna - и для большинства задачи вполне хватает их с средним ризонингом

дальше пошел смотреть бенчи/цены/траты по токенам - и ничего особо мне это не дало, какую модельку то выбрать для маленького бага? а для сложного маленького бага? а для простого бага но в котором нужно проследить очень длинный трейс?

по этому решил забилить свой небольшой бенч - дать реальные задачи разным моделям с разным ризонингом и сравнить - качество и цену

Перед началом эксперимента мне казалось, что результат будет примерно одинаковым или чуть лучше у маленьких моделей с высоким уровнем рассуждений, но на моих задачах получилось довольно неожиданно.

Кейс 1: внести небольшие правки в код. Качество одинаковое, но Terra закончила задачу за 54 секунды, а Luna за 134 секунды. Terra использовала 111k токенов, а Luna 211k.

Кейс 2: большая задача по изменению бизнес-логики в среднем сервисе. Terra High 6,83/10, Sol Low 9,50/10. Sol Low оказался качественнее, быстрее и потратил в 2,5 раза меньше токенов.

Кейс 3: большая задача. Написать большой сервис, включая Bluetooth-логику и математику. По качеству результаты оказались довольно близкими, но Luna даже не стала запускать рантайм. Luna High 5,70/10, Sol Medium 6,63/10.

Какой тут можно сделать вывод?
Похоже, что небольшие модели имеют смысл только с low/medium уровнем рассуждений. Всё-таки лучше взять модель побольше, но с меньшим уровнем рассуждений. Она работает быстрее и дешевле.

Также есть вопросы к передовым моделям, например, к Soul 5.6 Ultra. Возникает ощущение, что они чуть лучше показывают себя на бенчмарках, а в реальных задачах не дают настолько большого прироста пользы. Но это уже тема отдельного поста.

Наконец-то ИИ всех заменил!
…Но пока только по бенчам.
Пробую себя в видеоконтенте - собрал краткие новости уходящей недели про ИИ.
https://youtu.be/6-5AEvhlSZc

pydantic выпустил свой фреймворк для агентов - обещан рилтайм, ембеддинги, все типизированное(пример langchain) и ваще легко просто молодежно
постараюсь сделать обзорчик в ближайшее время - по quickstart пока нравится

https://pydantic.dev/docs/ai/overview/

Открыл для себя написание статей через надиктовывание в чатгпт - почитал бенчи/подписки и собрал все вместе в новой статье на хабре - го читать!

https://habr.com/ru/articles/1081060/