Методические материалы Лохтурова Вячеслава

This page is read only. You can view the source, but not change it. Ask your administrator if you think this is wrong.
====== Использование AI-агентов ======

===== Реклама =====

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

===== Техническое задание =====

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

===== Запись вебинара =====

  * Рутуб: https://rutube.ru/video/private/d1d43ee640c9ec4166f75fce562b384c/
  * Вк: https://vkvideo.ru/video-2190892_456239505
  * Ютуб: https://youtu.be/ULv9aKzdc20

  * Тэги: Yandex AI, LLM, AI Agent, Harness, VSCode, Cline AI, OpenCode AI, AI-driven DevOps, 

===== С чего начиналось =====

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

===== Сценарий 1: Вы немного программист, пишем свой ИИ чат =====

  * [[https://www.specialist.ru/course/python1-a|Программирование на Python. Уровень 1. Базовый курс]]
  * [[Язык программирования Python#Виртуальная среда Python]]

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

===== Сценарий 2: Вы специалист по Asterisk, добавляем интеграцию с ИИ чатом =====

  * [[https://www.specialist.ru/course/asterisk3|Asterisk. Уровень 3. Интеграция с внешними системами]]
  * [[Asterisk. Интеграция#Упражнение 6.4. Asterisk REST интерфейс]] до ARI Stasis

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

  * **Промпт**: "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 приложение [[Язык программирования Python#ari_echo_talk_detect.py]]

  * [[Yandex SpeechKit]] (обсудить)
  * Python модуль [[Язык программирования Python#speech_and_text.py]]
  * Python модуль [[Язык программирования Python#wav_and_ogg.py]]

  * **Итог**: Python приложение [[Язык программирования Python#ari_chat.py]] или [[Язык программирования Python#ari_chat_talk_detect.py]]

==== ToDo ====

=== Улучшение UX (User Experience) ===
  - Если звоним в первый раз (контекста еще нет) - представиться (пересказать системный промпт?)

=== Добавление функциональности (Feature Management) ===

превращаем бота в агента ([[Язык программирования Python#agent_rand_hello.py]]):

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

===== Сценарий 3: Вы администратор систем Linux, используем ИИ агентов в работе ===== 

  * [[https://www.specialist.ru/course/yun2-b|Linux. Уровень 2. Администрирование сетевых сервисов]]
  * [[Администрирование сервисов и сетей Linux]]

  * [[https://www.specialist.ru/course/git|Система управления версиями Git]]
  * Установить [[Сервис Git#etckeeper]] на gate
<code>
gate# git -C /etc/ status

gate# git -C /etc/ log
</code>

  * Проверить наличие, или установить [[Сервис DHCP]]
<code>
gate# git -C /etc/ diff

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

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

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

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

  * [[Continue]]
  * **Промпт**: кратко формат файла resolv.conf
<code>
gate:/etc# code resolv.conf

code ex1.bash
#!/ ...

for i ...
</code>

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

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

===== Сценарий 4: Вы DevOps инженер, используем ИИ агентов в pipeline ===== 

  * [[https://www.specialist.ru/course/kuber|DevOps. Уровень 1. Инфраструктура как код, основные инструменты]]
  * [[DevOps1. Основные инструменты]]

  * Выполняем шаги 4 и 5 из статьи [[https://habr.com/ru/articles/716454/|Самый простой пример CI/CD]]
  * Запускаем [[OpenCode]] из под пользователя gitlab-runner на gate
  * Может понадобиться
<code>
gate# apt install xxd
</code>
  * Выполняем шаг 6 из статьи [[https://habr.com/ru/articles/716454/|Самый простой пример CI/CD]]
  * Добавляем [[Инструмент GitLab#Пример CI/CD с использованием ИИ]]



===== Черновик =====

==== simple_chat.py ====

<code>
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)
</code>
==== ari_print.py ====
<code>
# pip install aiohttp asyncio
</code><code>
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())
</code>
==== ari_echo.py ====

<code>
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
</code><code>
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()
</code>
Методические материалы Лохтурова Вячеслава

User Tools

Site Tools

Page Tools
