Разработка AI-ассистента для ERP: аналитика, прогнозы, рекомендации

Разработка AI-ассистента для ERP: аналитика, прогнозы, рекомендации

ERP-системы накапливают тонны данных о продажах, закупе, производстве. Руководитель хочет знать: "Какой цех съедает больше бюджета?" или "Когда закончится сырьё X?". Без SQL-знаний или BI-аналитика получить ответ — задача на часы. Мы строим AI-ассистента, который понимает естественный язык, сам строит SQL, интерпретирует цифры и даёт рекомендации. Экономия времени управленческого персонала — до 80%, а средняя экономия бюджета на аналитику составляет $3,200 в месяц.

Проблемы, которые мы решаем

Ad-hoc-запросы застревают в очереди к аналитику. В компании с 15 руководителями средний аналитик тратит 25 часов в неделю на однотипные "посмотри выручку по клиентам". AI-ассистент берёт 70% таких запросов на себя, освобождая аналитика для глубоких исследований.

Прогнозирование превращается в гадание. Без автоматизации менеджеры строят прогнозы в Excel на основе интуиции. Ассистент подтягивает исторические тренды, сезонность и внешние датасеты, выполняя цепь запросов к ERP через агентный loop.

Алерты пропускаются. Критичные отклонения — просрочка дебиторки, перерасход бюджета — часто замечают слишком поздно. Ежедневный health-check, который мы встраиваем, сканирует KPI и шлёт алерты в Telegram или Slack.

Архитектура ERP-ассистента

Класс ERPAssistant использует LLM (например, Claude Sonnet 4-5) с инструментами: execute_query для безопасных SELECT и get_kpi_metrics для предрасчитанных метрик. Модель сама решает, какие SQL необходим, выполняет их через read-only коннект и возвращает интерпретацию с рекомендациями.

from anthropic import Anthropic import psycopg2 import json from typing import Any from pydantic import BaseModel client = Anthropic() class ERPQueryResult(BaseModel): sql: str data: list[dict] interpretation: str recommendations: list[str] alerts: list[str] class ERPAssistant: def __init__(self, db_connection_string: str, erp_schema: dict): self.conn = psycopg2.connect(db_connection_string) self.schema = erp_schema # Описание таблиц и связей ERP self.tools = [ { "name": "execute_query", "description": "Выполнить SQL запрос к ERP базе данных", "input_schema": { "type": "object", "properties": { "sql": {"type": "string", "description": "SELECT запрос"}, "description": {"type": "string", "description": "Что запрос делает"} }, "required": ["sql"] } }, { "name": "get_kpi_metrics", "description": "Получить предрасчитанные KPI метрики", "input_schema": { "type": "object", "properties": { "metric_type": { "type": "string", "enum": ["revenue", "inventory", "expenses", "headcount", "orders"] }, "period": {"type": "string", "description": "Период: week/month/quarter/year"} }, "required": ["metric_type", "period"] } }, ] def execute_query(self, sql: str) -> list[dict]: """Безопасное выполнение только SELECT запросов""" if not sql.strip().upper().startswith("SELECT"): raise ValueError("Только SELECT запросы разрешены") with self.conn.cursor() as cur: cur.execute(sql) columns = [d[0] for d in cur.description] rows = cur.fetchall() return [dict(zip(columns, row)) for row in rows[:100]] def get_kpi_metrics(self, metric_type: str, period: str) -> dict: """Возвращает предрасчитанные метрики""" # В реальной системе — запросы к таблицам ERP # Здесь упрощённый пример period_sql = { "week": "AND date >= CURRENT_DATE - INTERVAL '7 days'", "month": "AND date >= DATE_TRUNC('month', CURRENT_DATE)", "quarter": "AND date >= DATE_TRUNC('quarter', CURRENT_DATE)", "year": "AND date >= DATE_TRUNC('year', CURRENT_DATE)", } date_filter = period_sql.get(period, period_sql["month"]) if metric_type == "revenue": sql = f"SELECT SUM(amount) as total, COUNT(*) as orders FROM sales WHERE 1=1 {date_filter}" return self.execute_query(sql)[0] if self.execute_query(sql) else {} # ... остальные метрики return {} def dispatch_tool(self, tool_name: str, tool_input: dict) -> Any: if tool_name == "execute_query": return self.execute_query(tool_input["sql"]) elif tool_name == "get_kpi_metrics": return self.get_kpi_metrics(tool_input["metric_type"], tool_input["period"]) raise ValueError(f"Unknown tool: {tool_name}") def answer(self, question: str, user_role: str = "manager") -> ERPQueryResult: """Отвечает на аналитический вопрос""" messages = [{ "role": "user", "content": f"""Вопрос: {question} Роль пользователя: {user_role} Схема ERP базы: {json.dumps(self.schema, ensure_ascii=False, indent=2)[:2000]} Используй инструменты для получения данных, затем: 1. Проинтерпретируй результаты 2. Дай рекомендации если применимо 3. Укажи алерты если данные требуют внимания""" }] sql_queries = [] all_data = [] while True: response = client.messages.create( model="claude-sonnet-4-5", max_tokens=4096, system=f"""Ты — бизнес-аналитик по ERP системе компании. Анализируй данные точно, приводи конкретные цифры. Давай практические рекомендации руководству.""", tools=self.tools, messages=messages, ) if response.stop_reason == "end_turn": final_text = next( (b.text for b in response.content if hasattr(b, "text")), "" ) return ERPQueryResult( sql="; ".join(sql_queries), data=all_data[:10], interpretation=final_text, recommendations=self._extract_list(final_text, "Рекомендации"), alerts=self._extract_list(final_text, "Алерты"), ) tool_results = [] for block in response.content: if block.type == "tool_use": result = self.dispatch_tool(block.name, block.input) if isinstance(result, list): all_data.extend(result) if block.name == "execute_query": sql_queries.append(block.input.get("sql", "")) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result, ensure_ascii=False, default=str), }) messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": tool_results}) def _extract_list(self, text: str, section: str) -> list[str]: """Извлекает пункты из секции текста""" import re if section not in text: return [] section_text = text.split(section)[1].split("\n\n")[0] return [ line.strip("- •*").strip() for line in section_text.splitlines() if line.strip() and line.strip().startswith(("-", "•", "*", "1", "2")) ] 

Интеграция с 1С через API

import requests class OneCIntegration: """Интеграция с 1С:Предприятие через HTTP-сервис""" def __init__(self, base_url: str, username: str, password: str): self.base_url = base_url self.auth = (username, password) def get_sales_report(self, date_from: str, date_to: str) -> list[dict]: """Получает отчёт по продажам из 1С""" response = requests.get( f"{self.base_url}/hs/api/v1/sales", auth=self.auth, params={"dateFrom": date_from, "dateTo": date_to} ) return response.json() def get_inventory_status(self) -> list[dict]: """Получает остатки товаров""" response = requests.get( f"{self.base_url}/hs/api/v1/inventory", auth=self.auth, ) return response.json() def get_budget_execution(self, period: str) -> dict: """Получает исполнение бюджета""" response = requests.get( f"{self.base_url}/hs/api/v1/budget/{period}", auth=self.auth, ) return response.json() 

Автоматические отчёты и алерты

import asyncio from datetime import datetime class ERPAlertSystem: """Автоматически обнаруживает аномалии и отправляет алерты""" def __init__(self, assistant: ERPAssistant): self.assistant = assistant async def daily_health_check(self) -> list[str]: """Ежедневный аудит ключевых метрик""" checks = [ "Есть ли товары с критически низким запасом (менее недели)?", "Превышены ли бюджеты каких-либо отделов в этом месяце?", "Есть ли просроченная дебиторская задолженность более 30 дней?", "Какие показатели значительно отличаются от прошлого месяца?", ] alerts = [] for check in checks: result = self.assistant.answer(check) if result.alerts: alerts.extend(result.alerts) return alerts def generate_executive_report(self, period: str = "month") -> str: """Генерирует исполнительный отчёт для руководства""" result = self.assistant.answer( f"Подготовь исполнительный отчёт за {period}: ключевые метрики, тренды, риски, рекомендации", user_role="ceo" ) return result.interpretation 

Практический кейс: производственная компания

Наш клиент — завод с ERP 1С:ERP, 15 руководителей и один BI-аналитик на полставки. Аналитик тонул в ad-hoc-запросах. Мы развернули ассистента за две недели. Внедрили интеграцию с 1С через HTTP-сервис, настроили алерт-систему на критические отклонения.

Типичные вопросы, которые ассистент теперь решает за секунды:

• "Когда закончатся материалы для производства X при текущем темпе?"
• "Какой цех превышает плановые расходы?"
• "Топ-5 клиентов по выручке за квартал с динамикой"

Результат за полгода:

• Ad-hoc-запросы к аналитику сократились с 25 до 7 в неделю (на 72%)
• Время получения ответа на управленческий вопрос упало с 1–2 часов до 30 секунд — в 50 раз быстрее
• Ежедневный алерт на критические метрики предотвратил два кассовых разрыва, сэкономив компании $12,000

Почему AI-ассистент для ERP — не просто Text-to-SQL?

Простые Text-to-SQL модели ошибаются в JOIN и контексте. Наш агентный loop с несколькими запросами позволяет уточнять данные: сначала получить общую сумму, потом развернуть по складам, потом запросить тренд. Выбор следующего действия делает LLM на основе инструментов и истории. Это снижает hallucination и повышает accuracy до 90% на сложных запросах — на 40% больше, чем у стандартных решений без agent loop.

Как обеспечить безопасность запросов?

Применяем три уровня защиты. Первый — парсинг: код блокирует любые запросы, не начинающиеся с SELECT. Второй — read-only пользователь БД без прав на INSERT/UPDATE/DELETE. Третий — логирование всех сгенерированных SQL в таблицу аудита. Если модель пытается выполнить DDL, парсер возвращает исключение до выполнения.

Как внедрить AI-ассистента за 3 шага?

1. Анализ схемы данных — получаем доступ к схеме ERP, определяем ключевые KPI.
2. Конфигурация ассистента — настраиваем промпты, инструменты и интеграцию с 1С или другой ERP.
3. Тестирование и запуск — проверяем точность на 20 типовых запросах, включаем алерт-систему.

"По данным завода, время на рутинные отчёты сократилось на 72%"

Сроки

Что входит в работу

• Документация: описание схемы базы данных, промптов и ограничений.
• Доступы: read-only пользователь, логирование, дашборд мониторинга.
• Обучение: сессия для руководителей и аналитика.
• Поддержка: 2 недели пост-продакшн, исправление неточностей в генерации.

Стоимость рассчитывается индивидуально — зависит от сложности ERP-схемы, количества агентов и необходимости fine-tuning. Свяжитесь с нами, чтобы получить консультацию и оценить проект под ключ. Закажите внедрение сейчас — наши инженеры имеют многолетний опыт интеграции с 1С и другими ERP.