AI Assistant for ERP: Analytics, Forecasts, Recommendations

AI Assistant for ERP: Analytics, Forecasts, Recommendations

ERP systems accumulate tons of data on sales, procurement, and production. A manager wants to know: "Which shop floor is eating the most budget?" or "When will raw material X run out?" Without SQL knowledge or a BI analyst, getting an answer takes hours. We build an AI assistant that understands natural language, constructs SQL itself, interprets numbers, and gives recommendations. Time savings for management personnel — up to 80%, with average budget savings on analytics of $3,200 per month.

Problems We Solve

Ad-hoc queries get stuck in the analyst queue. In a company with 15 managers, the average analyst spends 25 hours a week on repetitive "check revenue by customer" requests. The AI assistant handles 70% of such requests, freeing the analyst for deep research.

Forecasting turns into guesswork. Without automation, managers build forecasts in Excel based on intuition. The assistant pulls historical trends, seasonality, and external datasets, executing a chain of queries to ERP through an agent loop.

Alerts are missed. Critical deviations — overdue receivables, budget overruns — are often noticed too late. The daily health check we embed scans KPIs and sends alerts to Telegram or Slack.

Architecture of the ERP Assistant

The ERPAssistant class uses an LLM (e.g., Claude Sonnet 4-5) with tools: execute_query for safe SELECTs and get_kpi_metrics for pre-calculated metrics. The model decides which SQL is needed, executes it through a read-only connection, and returns an interpretation with recommendations.

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 table and relationship description

        self.tools = [
            {
                "name": "execute_query",
                "description": "Execute an SQL query against the ERP database",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "SELECT query"},
                        "description": {"type": "string", "description": "What the query does"}
                    },
                    "required": ["sql"]
                }
            },
            {
                "name": "get_kpi_metrics",
                "description": "Get pre-calculated KPI metrics",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "metric_type": {
                            "type": "string",
                            "enum": ["revenue", "inventory", "expenses", "headcount", "orders"]
                        },
                        "period": {"type": "string", "description": "Period: week/month/quarter/year"}
                    },
                    "required": ["metric_type", "period"]
                }
            },
        ]

    def execute_query(self, sql: str) -> list[dict]:
        """Safe execution of SELECT queries only"""
        if not sql.strip().upper().startswith("SELECT"):
            raise ValueError("Only SELECT queries are allowed")

        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:
        """Returns pre-calculated metrics"""
        # In a real system — queries to ERP tables
        # Simplified example below
        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 {}
        # ... other metrics
        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:
        """Answer an analytical question"""

        messages = [{
            "role": "user",
            "content": f"""Question: {question}
User role: {user_role}

ERP database schema:
{json.dumps(self.schema, ensure_ascii=False, indent=2)[:2000]}

Use tools to get data, then:
1. Interpret the results
2. Give recommendations if applicable
3. Highlight alerts if data requires attention"""
        }]

        sql_queries = []
        all_data = []

        while True:
            response = client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=4096,
                system=f"""You are a business analyst for the company's ERP system.
Analyze data accurately, cite specific numbers.
Give practical recommendations to management.""",
                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, "Recommendations"),
                    alerts=self._extract_list(final_text, "Alerts"),
                )

            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]:
        """Extract items from a text section"""
        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"))
        ]

Integration with 1C via API

import requests

class OneCIntegration:
    """Integration with 1C:Enterprise via HTTP service"""

    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]:
        """Get sales report from 1C"""
        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]:
        """Get item balances"""
        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:
        """Get budget execution"""
        response = requests.get(
            f"{self.base_url}/hs/api/v1/budget/{period}",
            auth=self.auth,
        )
        return response.json()

Automatic Reports and Alerts

import asyncio
from datetime import datetime

class ERPAlertSystem:
    """Automatically detects anomalies and sends alerts"""

    def __init__(self, assistant: ERPAssistant):
        self.assistant = assistant

    async def daily_health_check(self) -> list[str]:
        """Daily audit of key metrics"""
        checks = [
            "Are there items with critically low stock (less than a week)?",
            "Are any department budgets exceeded this month?",
            "Is there overdue receivables older than 30 days?",
            "Which indicators differ significantly from last month?",
        ]

        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:
        """Generate an executive report for management"""
        result = self.assistant.answer(
            f"Prepare an executive report for {period}: key metrics, trends, risks, recommendations",
            user_role="ceo"
        )
        return result.interpretation

Practical Case: Manufacturing Company

Our client — a factory using 1C:ERP, with 15 managers and one part-time BI analyst. The analyst was drowning in ad-hoc queries. We deployed the assistant in two weeks. We integrated with 1C via HTTP service and set up an alert system for critical deviations.

Typical questions the assistant now solves in seconds:

• "When will materials for production X run out at the current pace?"
• "Which shop floor exceeds planned expenses?"
• "Top 5 customers by revenue for the quarter with dynamics"

Results after six months:

• Ad-hoc queries to the analyst dropped from 25 to 7 per week (72% reduction)
• Time to get a management answer fell from 1–2 hours to 30 seconds — 50x faster
• Daily alert on critical metrics prevented two cash gaps, saving the company $12,000

Why an AI Assistant for ERP is More Than Just Text-to-SQL

Simple Text-to-SQL models make mistakes in JOINs and context. Our agent loop with multiple queries allows data refinement: first get total amount, then break down by warehouse, then request trend. The LLM chooses the next action based on tools and history. This reduces hallucination and boosts accuracy to 90% on complex queries — 40% higher than standard solutions without an agent loop.

How to Ensure Query Security?

We apply three protection levels. First — parsing: the code blocks any query that does not start with SELECT. Second — a read-only database user with no INSERT/UPDATE/DELETE rights. Third — logging all generated SQL to an audit table. If the model attempts DDL, the parser raises an exception before execution.

How to Implement an AI Assistant in 3 Steps?

1. Data schema analysis — gain access to the ERP schema, identify key KPIs.
2. Assistant configuration — set up prompts, tools, and integration with 1C or another ERP.
3. Testing and launch — verify accuracy on 20 typical queries, enable the alert system.

"According to the factory data, time spent on routine reports decreased by 72%."

Timelines

What's Included in the Work

• Documentation: database schema description, prompts, and constraints.
• Access: read-only user, logging, monitoring dashboard.
• Training: a session for managers and the analyst.
• Support: 2 weeks post-production, fixing inaccuracies in generation.

Pricing is determined individually — depends on ERP schema complexity, number of agents, and need for fine-tuning. Contact us for a consultation and a turnkey project estimate. Order implementation now — our engineers have years of experience integrating with 1C and other ERPs.