Розробка 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"Невідомий інструмент: {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.