Исаев Василий - заметки про it, python и ai

За последние пол года активно работал с голосовыми агентами и открыл для себя 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 и еще много чего

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

### 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']

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

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

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

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

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

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

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

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


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