Разработка AI-ассистента для ERP: аналитика, прогнозы, рекомендации
ERP-системы содержат огромный массив данных о бизнесе, но работа с ними требует SQL-знаний или специальных навыков BI-инструментов. AI-ассистент для ERP позволяет руководителям получать аналитику на естественном языке: "Какой отдел превысил бюджет в этом квартале?" или "Когда закончатся запасы компонента А при текущем потреблении?"
Архитектура ERP-ассистента
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 на полставки.
Типичные запросы:
- "Когда закончатся материалы для производства X при текущем темпе?"
- "Какой цех превышает плановые расходы?"
- "Топ-5 клиентов по выручке за квартал с динамикой"
Результат:
- Ad-hoc запросы к аналитику: 25/неделю → 7/неделю (-72%)
- Время получения ответа на управленческий вопрос: 1–2 часа → 30 секунд
- Ежедневный автоматический алерт на критические метрики
Сроки
- Базовый Text-to-SQL для ERP: 1 неделя
- Агентный loop с несколькими запросами: 1 неделя
- Интеграция с 1С HTTP-сервисом: 1 неделя
- Алерт-система + автоотчёты: 1 неделя