ИИ агент для распознования и анализа записи звонков

07.08.2026 | коды из категории: Нейросети

не настроен просто шаблон

~/mango-dispatcher$ cat agent.py

#!/usr/bin/env python3
import requests
import json
import re
import sqlite3
import os
from datetime import datetime

# ===================== НАСТРОЙКИ =====================
YANDEX_API_KEY = "11111111111111111"          # Из консоли Yandex Cloud
YANDEX_FOLDER_ID = "2222222222222"       # Из консоли Yandex Cloud
DB_PATH = "calls.db"                     # База данных SQLite

# ===================== БАЗА ДАННЫХ =====================
def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute('''
        CREATE TABLE IF NOT EXISTS calls (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            call_id TEXT,
            caller_number TEXT,
            called_number TEXT,
            duration INTEGER,
            record_url TEXT,
            transcript TEXT,
            manager_rating INTEGER,
            verdict TEXT,
            verdict_reason TEXT,
            emotional_tone TEXT,
            created_at TEXT
        )
    ''')
    conn.commit()
    conn.close()
    print("✅ База данных инициализирована")

def save_to_db(call_data, transcript, result):
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute('''
        INSERT OR REPLACE INTO calls
        (call_id, caller_number, called_number, duration, record_url,
         transcript, manager_rating, verdict, verdict_reason, emotional_tone, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    ''', (
        call_data.get('callId'),
        call_data.get('callerNumber'),
        call_data.get('calledNumber'),
        call_data.get('duration'),
        call_data.get('recordUrl'),
        transcript,
        result.get('manager_rating'),
        result.get('verdict'),
        result.get('verdict_reason'),
        result.get('emotional_tone'),
        datetime.now().isoformat()
    ))
    conn.commit()
    conn.close()
    print(f"💾 Сохранено в БД: звонок {call_data.get('callId')}")

# ===================== РАСПОЗНАВАНИЕ =====================
def download_audio(record_url):
    try:
        resp = requests.get(record_url, timeout=60)
        if resp.status_code == 200:
            print(f"✅ Аудио скачано: {len(resp.content)} байт")
            return resp.content
        else:
            print(f"❌ Ошибка скачивания: {resp.status_code}")
            return None
    except Exception as e:
        print(f"❌ Ошибка скачивания: {e}")
        return None

def transcribe_audio(audio_content):
    url = "https://stt.api.cloud.yandex.net/speech/v1/stt:recognize"
    headers = {
        "Authorization": f"Api-Key {YANDEX_API_KEY}",
        "Content-Type": "audio/ogg;codecs=opus"
    }
    try:
        resp = requests.post(url, headers=headers, data=audio_content, timeout=120)
        if resp.status_code == 200:
            result = resp.json().get("result", "")
            print(f"✅ Распознано: {len(result)} символов")
            return result
        else:
            print(f"❌ Ошибка распознавания: {resp.status_code}")
            return None
    except Exception as e:
        print(f"❌ Ошибка распознавания: {e}")
        return None

# ===================== АНАЛИЗ =====================
def analyze_gpt(transcript):
    url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"

    system_prompt = """
    Ты — независимый эксперт по оценке качества сервиса.
    Проанализируй диалог между менеджером и клиентом.

    Оцени поведение менеджера:
    - Профессионализм
    - Эмоциональный фон
    - Вовлеченность

    Верни ответ строго в формате JSON:
    {
        "manager_rating": число от 1 до 5,
        "verdict": "НОРМ" или "ТОКСИЧЕН",
        "verdict_reason": "краткая причина",
        "emotional_tone": "positive" или "neutral" или "negative"
    }
    """

    payload = {
        "modelUri": f"gpt://{YANDEX_FOLDER_ID}/yandexgpt-lite",
        "completionOptions": {
            "stream": False,
            "temperature": 0.3,
            "maxTokens": 1000
        },
        "messages": [
            {"role": "system", "text": system_prompt},
            {"role": "user", "text": f"Диалог:\n\n{transcript}"}
        ]
    }

    headers = {
        "Authorization": f"Api-Key {YANDEX_API_KEY}",
        "Content-Type": "application/json"
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=60)
        if resp.status_code == 200:
            text = resp.json().get("result", {}).get("alternatives", [{}])[0].get("message", {}).get("text", "")
            match = re.search(r'\{.*\}', text, re.DOTALL)
            if match:
                return json.loads(match.group())
        return None
    except Exception as e:
        print(f"❌ Ошибка GPT: {e}")
        return None

# ===================== ГЛАВНАЯ ФУНКЦИЯ =====================
def process_call(call_data):
    call_id = call_data.get('callId')
    record_url = call_data.get('recordUrl')

    print(f"\n🤖 АГЕНТ: Обрабатываю звонок {call_id}")

    # 1. Скачиваем
    audio = download_audio(record_url)
    if not audio:
        return None

    # 2. Распознаём
    transcript = transcribe_audio(audio)
    if not transcript:
        return None

    # 3. Анализируем
    result = analyze_gpt(transcript)
    if not result:
        return None

    # 4. Сохраняем
    save_to_db(call_data, transcript, result)

    print(f"✅ АГЕНТ: Звонок {call_id} обработан! Вердикт: {result.get('verdict')}")
    return result

# ===================== ТЕСТ =====================
if __name__ == "__main__":
    init_db()
    print("🤖 Агент готов к работе")

Оцените статью

Всего: 0

Оставить комментарий

← Назад к списку

Категории

ИИ агент для распознования и анализа записи звонков

не настроен просто шаблон

Категории:

Категории

Комментарии

Оставить комментарий

Категории

ИИ агент для распознования и анализа записи звонков

не настроен просто шаблон

Категории:

Категории

Комментарии

Оставить комментарий

Обратная связь