• Использование AI-агентов чаще всего упоминается в связи с автоматизацией клиентского сервиса, анализом данных, маркетингом, продажами и разработкой программного обеспечения. Но на нашем вебинаре мы увидим, что и в инженерных IT-задачах им можно найти полезное применение. Последовательно научимся использовать возможности LLM: от подсказок в чате до полной автоматизации развертывания, управления и тестирования систем.

• Использовать AI-агентов для анализа и управления конфигурацией Linux-систем
• Интегрировать AI-агентов в GitLab CI/CD
• Написать своего первого AI-агента на Python
• Реализовать MVP с доступом к голосовому AI-ассистенту через Asterisk

• Ютуб:
• Вк:
• Рутуб:

• Тэги: LLM, AI Agent, Harness, Yandex AI,

• Голосовые сервисы помогут голосовой почте! (Занятие СЕМ 03.12.2019)

• Программирование на Python. Уровень 1. Базовый курс
• Виртуальная среда Python

• Yandex AI
• Промпт: самый простой пример чат бота на python simple_chat.py с использованием yandex ai
• Промпт: Привет, меня зовут Вчеслав, я инженер
• Промпт: Как меня зовут?
• Python приложение/модуль request_to_llm.py (обсудить role system/user/assistant, context)
• Использование OpenAI API openai_request_to_llm.py
• Промпт: Как называется твоя модель?

• Asterisk. Уровень 3. Интеграция с внешними системами
• Упражнение 6.4. Asterisk REST интерфейс до ARI Stasis

# curl http://localhost:8088/ari/asterisk/ping?api_key=asterisk:asterisk
{"ping":"pong" ...

• Промпт: “Asterisk ARI самый простой пример на Python 3.12 приложения Stasis(my-first-app), выводит на экран нажатые цифры, параметры подключения аналогичны curl …” ari_print.py
• Промпт: “Asterisk 20, ARI самый простой пример на Python 3.12 приложения Stasis(my-first-app), повторяет фразы пользователя, параметры подключения аналогичны curl …” ari_echo.py
• Python приложение ari_echo_talk_detect.py

• Yandex SpeechKit (обсудить)
• Python модуль speech_and_text.py
• Python модуль wav_and_ogg.py

• Итог: Python приложение ari_chat.py или ari_chat_talk_detect.py

1. Если звоним в первый раз (контекста еще нет) - представиться (пересказать системный промпт?)

превращаем бота в агента (agent_rand_hello.py):

1. Добавить возможность отправить список дел на почту
2. Добавить возможность агенту позвонить и напомнить о деле

• Linux. Уровень 2. Администрирование сетевых сервисов
• Администрирование сервисов и сетей Linux

• Система управления версиями Git
• Установить etckeeper на gate

gate# git -C /etc/ status

gate# git -C /etc/ log

• Проверить наличие, или установить Сервис DHCP

gate# git -C /etc/ diff

gate# git -C /etc/ add -v .

gate# git -C /etc/ commit -m "dhcp confgured"

• Импортируем в LAN Windows (LTSC) или Linux клиент
• Устанавливаем на него VSCode (отключаем Copilot)

• Visual Studio Code Remote - SSH на gate
• Сервис Git (VSCode Open Folder)

• Continue
• Промпт: кратко формат файла resolv.conf

gate:/etc# code resolv.conf

code ex1.bash
#!/ ...

for i ...

• Cline
• Plan Промпт: какой диапазон адресов выдает dhcp сервер?
• Act Промпт: расширь используемый диапазон до 119-го адреса
• Статистика DHCP сервера

• Cтатьи в разделе Нейросети
• OpenCode(может понадобиться Переменные окружения)
• Plan Промпт: какие сервисы установлены в этой системе? (Plan · Big Pickle OpenCode Zen)
• Build Промпт: установи squid, создай минимальный файл конфигурации, разреши доступ всем к сайтам из файла /etc/squid/permit_domains.txt, не создавай сертификаты, задай уточняющие вопросы
• Результат - шаг 2 из статьи Самый простой пример CI/CD
• Утилита curl

• DevOps. Уровень 1. Инфраструктура как код, основные инструменты
• DevOps1. Основные инструменты

• Выполняем шаги 4 и 5 из статьи Самый простой пример CI/CD
• Запускаем OpenCode из под пользователя gitlab-runner на gate
• Может понадобиться

gate# apt install xxd

• Выполняем шаг 6 из статьи Самый простой пример CI/CD
• Добавляем Пример CI/CD с использованием ИИ

import os
import requests

# Вставьте сюда ваши данные
YANDEX_API_KEY = "..."               # API-ключ сервисного аккаунта
FOLDER_ID = "b1gg4j6ru4esprat3uu7"                          # ID каталога в Yandex Cloud

URL = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"

def get_ai_response(user_message: str) -> str:
    headers = {
        "Authorization": f"Api-Key {YANDEX_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "modelUri": f"gpt://{FOLDER_ID}/yandexgpt/latest",
        "completionOptions": {
            "stream": False,
            "temperature": 0.7,
            "maxTokens": "200"
        },
        "messages": [
            {"role": "user", "text": user_message}
        ]
    }

    resp = requests.post(URL, json=payload, headers=headers, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    # Yandex GPT возвращает список вариантов; берём первый
    return data["result"]["alternatives"][0]["message"]["text"]

if __name__ == "__main__":
    print("Чат-бот с Yandex GPT (введите 'выход' для завершения)")
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ("выход", "quit", "exit"):
            break
        try:
            ai_reply = get_ai_response(user_input)
            print("Бот:", ai_reply)
        except Exception as e:
            print("Ошибка:", e)

# pip install aiohttp asyncio

import asyncio
import json
import aiohttp

# Параметры подключения (из вашего curl-примера)
ARI_HOST = "localhost"
ARI_PORT = 8088
ARI_USER = "asterisk"
ARI_PASS = "asterisk"
APP_NAME = "my-first-app"

WS_URL = f"ws://{ARI_HOST}:{ARI_PORT}/ari/events?app={APP_NAME}&api_key={ARI_USER}:{ARI_PASS}"
REST_URL = f"http://{ARI_HOST}:{ARI_PORT}/ari"

async def main():
    async with aiohttp.ClientSession() as session:
        # Проверяем, что Asterisk жив (как ваш curl-ping)
        async with session.get(
            f"{REST_URL}/asterisk/ping?api_key={ARI_USER}:{ARI_PASS}"
        ) as resp:
            if resp.status != 200:
                print(f"Не удалось подключиться к ARI: HTTP {resp.status}")
                return
            print("Asterisk ARI доступен, подключаемся к WebSocket…")

        # Подписка на события приложения my-first-app
        async with session.ws_connect(WS_URL) as ws:
            print(f"WebSocket открыт. Приложение «{APP_NAME}» слушает события.")
            print("Позвоните на экстеншен с Stasis(my-first-app) и нажимайте клавиши.\n")

            async for msg in ws:
                if msg.type != aiohttp.WSMsgType.TEXT:
                    continue

                event = json.loads(msg.data)
                etype = event.get("type")

                if etype == "StasisStart":
                    channel = event["channel"]
                    print(f"[Звонок поступил] {channel['name']} (id: {channel['id']})")

                elif etype == "ChannelDtmfReceived":
                    digit = event.get("digit", "?")
                    channel = event.get("channel", {})
                    name = channel.get("name", "?")
                    duration = event.get("duration_ms", 0)
                    print(f"  Нажато: «{digit}»  (канал: {name}, длительность: {duration} мс)")

                elif etype == "StasisEnd":
                    channel = event["channel"]
                    print(f"[Звонок завершён] {channel['name']}")

                # Можно uncomment для отладки — увидите все события:
                # else:
                #     print(f"  (событие: {etype})")


if __name__ == "__main__":
    asyncio.run(main())

root# 
mkdir -p /var/spool/asterisk/recording/
chown -R asterisk:asterisk /var/spool/asterisk/recording/
chmod -R 775 /var/spool/asterisk/recording/

pip install websocket-client requests

import json
import requests
import websocket

# --- Настройки подключения (аналогичны curl из вопроса) ---
ARI_URL = "http://localhost:8088/ari"
WS_URL  = "ws://localhost:8088/ari/events?app=my-first-app&api_key=asterisk:asterisk"
AUTH    = ("asterisk", "asterisk")

# Состояние каждого канала: "recording" или "playing"
channels: dict[str, str] = {}


def start_recording(cid: str):
    """Запустить запись речи звонящего (макс 5 сек, пауза 2 сек — стоп)."""
    channels[cid] = "recording"
    requests.post(f"{ARI_URL}/channels/{cid}/record", auth=AUTH, params={
        "name": f"echo-{cid}",
        "format": "wav",
        "maxDurationSeconds": 5,
        "maxSilenceSeconds": 2,
        "beep": "true",
    })


def on_message(ws, message):
    event = json.loads(message)
    etype = event.get("type")

    if etype == "StasisStart":
        cid = event["channel"]["id"]
        print(f"→ Звонок от {event['channel']['name']}")
        requests.post(f"{ARI_URL}/channels/{cid}/answer", auth=AUTH)
        start_recording(cid)

    elif etype == "RecordingFinished":
        cid = event["recording"]["target_uri"].removeprefix("channel:")
        if cid in channels:
            print("  Запись готова, воспроизвожу…")
            channels[cid] = "playing"
            requests.post(f"{ARI_URL}/channels/{cid}/play", auth=AUTH, params={
                "media": f"recording:echo-{cid}",
            })

    elif etype == "PlaybackFinished":
        cid = event["playback"]["target_uri"].removeprefix("channel:")
        if cid in channels and channels[cid] == "playing":
            print("  Воспроизведение окончено, снова запись…")
            start_recording(cid)

    elif etype == "StasisEnd":
        cid = event["channel"]["id"]
        channels.pop(cid, None)
        print(f"← Звонок завершён: {event['channel']['name']}")


if __name__ == "__main__":
    print("Echo-приложение 'my-first-app' запущено. Ожидание звонков…")
    ws = websocket.WebSocketApp(WS_URL, on_message=on_message)
    ws.run_forever()