AI Digital Accountant Development

AI Accountant — Digital Worker for Financial Accounting

AI Accountant automates routine accounting operations: processing primary documents (invoices, acts, waybills), data reconciliation, journal entry preparation, report generation, accounts receivable monitoring. Does not replace chief accountant but removes 60–70% of operational workload.

Primary Document Processing

import anthropic
import base64
from pathlib import Path
from pydantic import BaseModel
from typing import Optional, Literal

client = anthropic.Anthropic()

class InvoiceData(BaseModel):
    document_type: Literal["invoice", "act", "waybill", "upd"]
    vendor_name: str
    vendor_inn: Optional[str]
    vendor_kpp: Optional[str]
    document_number: str
    document_date: str
    amount_without_vat: float
    vat_rate: Optional[float]        # 20, 10, 0, None (no VAT)
    vat_amount: Optional[float]
    total_amount: float
    items: list[dict]                 # Document lines
    payment_purpose: Optional[str]   # Payment description
    contract_reference: Optional[str]

def extract_document_data(file_path: str) -> InvoiceData:
    """Extract data from scanned/PDF document via Claude Vision"""

    with open(file_path, "rb") as f:
        file_content = base64.standard_b64encode(f.read()).decode("utf-8")

    ext = Path(file_path).suffix.lower()
    media_type = "application/pdf" if ext == ".pdf" else "image/jpeg"

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "document" if ext == ".pdf" else "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": file_content,
                    },
                },
                {
                    "type": "text",
                    "text": """Extract all financial document details.
For invoices: vendor ID, tax ID, document number, date, amounts with/without VAT.
Return JSON per InvoiceData schema. Use null for missing fields.""",
                },
            ],
        }],
    )

    return InvoiceData.model_validate_json(response.content[0].text)

Verification and Contract Matching

class DocumentVerifier:

    async def verify_invoice(
        self,
        invoice: InvoiceData,
        contract_id: str,
    ) -> dict:
        """Verifies invoice against contract and reference data"""

        # Parallel verification
        contract_task = contracts_db.get(contract_id)
        vendor_task = vendor_db.get_by_inn(invoice.vendor_inn)
        previous_invoices_task = invoices_db.get_for_contract(contract_id)

        contract, vendor, previous = await asyncio.gather(
            contract_task, vendor_task, previous_invoices_task
        )

        issues = []

        # Vendor ID verification
        if vendor and vendor.inn != invoice.vendor_inn:
            issues.append(f"Vendor ID mismatch: document={invoice.vendor_inn}, reference={vendor.inn}")

        # Amount verification against contract
        total_paid = sum(p.amount for p in previous if p.status == "approved")
        if total_paid + invoice.total_amount > contract.max_amount * 1.05:  # 5% tolerance
            issues.append(f"Contract amount exceeded: paid {total_paid}, new invoice {invoice.total_amount}, limit {contract.max_amount}")

        # VAT verification
        if invoice.vat_amount and invoice.vat_rate:
            expected_vat = invoice.amount_without_vat * invoice.vat_rate / 100
            if abs(expected_vat - invoice.vat_amount) > 1:  # 1 rub tolerance
                issues.append(f"VAT calculation error: expected {expected_vat:.2f}, document shows {invoice.vat_amount:.2f}")

        # Duplicate check
        duplicate = next(
            (p for p in previous if p.document_number == invoice.document_number and p.vendor_inn == invoice.vendor_inn),
            None,
        )
        if duplicate:
            issues.append(f"Duplicate document: #{invoice.document_number} already processed {duplicate.processed_date}")

        return {
            "valid": len(issues) == 0,
            "issues": issues,
            "requires_manual_review": len(issues) > 0,
            "vendor_verified": vendor is not None,
        }

Journal Entry Generation

class AccountingEntryGenerator:

    ACCOUNT_MAPPING = {
        "materials": {"debit": "10.01", "credit": "60.01"},
        "services": {"debit": "26",     "credit": "60.01"},
        "goods": {"debit": "41.01",     "credit": "60.01"},
        "vat_input": {"debit": "19.03", "credit": "60.01"},
    }

    async def generate_entries(
        self,
        invoice: InvoiceData,
        cost_center: str,
    ) -> list[dict]:
        """Generates journal entries"""

        # LLM classifies expense type
        classification = await client.messages.create(
            model="claude-opus-4-5",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": f"""Classify expense for accounting entries.
Vendor: {invoice.vendor_name}
Items in document: {[item['name'] for item in invoice.items[:5]]}

Return JSON: {{"expense_type": "materials|services|goods|fixed_assets", "account": "26|44|10|08", "vat_deductible": true|false}}"""
            }],
        )

        classification_data = json.loads(classification.content[0].text)
        entries = []

        # Main entry
        account_map = self.ACCOUNT_MAPPING.get(classification_data["expense_type"], self.ACCOUNT_MAPPING["services"])
        entries.append({
            "debit": account_map["debit"],
            "credit": account_map["credit"],
            "amount": invoice.amount_without_vat,
            "description": f"{invoice.vendor_name} / {invoice.document_number} from {invoice.document_date}",
            "cost_center": cost_center,
            "analytic": invoice.vendor_inn,
        })

        # VAT entry
        if invoice.vat_amount and classification_data.get("vat_deductible"):
            entries.append({
                "debit": "19.03",
                "credit": "60.01",
                "amount": invoice.vat_amount,
                "description": f"VAT / {invoice.vendor_name} / {invoice.document_number}",
                "cost_center": cost_center,
            })

        return entries

    async def post_to_1c(self, entries: list[dict]) -> str:
        """Uploads entries via API"""
        xml_data = self.format_1c_xml(entries)
        result = await onec_api.post_document(xml_data)
        return result["document_id"]

Accounts Receivable Monitoring

class ReceivablesMonitor:

    async def daily_ar_check(self) -> dict:
        """Daily overdue receivables check"""

        overdue_invoices = await invoices_db.get_overdue()

        report = {
            "total_overdue": sum(i.amount for i in overdue_invoices),
            "by_aging_bucket": self.group_by_aging(overdue_invoices),
            "actions_taken": [],
        }

        for invoice in overdue_invoices:
            days_overdue = invoice.days_overdue

            if days_overdue <= 7:
                # Friendly reminder
                await self.send_reminder(invoice, tone="friendly")
                report["actions_taken"].append(f"Reminder: {invoice.customer_name}")

            elif days_overdue <= 30:
                # Formal notice
                await self.send_reminder(invoice, tone="formal")
                await crm.create_task(customer_id=invoice.customer_id, title="AR Follow-up")
                report["actions_taken"].append(f"Formal notice: {invoice.customer_name}")

            elif days_overdue > 30:
                # Escalate to CFO
                await self.escalate_to_cfo(invoice)
                report["actions_taken"].append(f"ESCALATION: {invoice.customer_name}, {days_overdue} days overdue")

        return report

    async def generate_ar_reminder(self, invoice: dict, tone: str) -> str:
        response = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": f"Write payment reminder. Tone: {tone}. Professional business style. 2-3 paragraphs."
            }, {
                "role": "user",
                "content": f"Invoice #{invoice['number']} from {invoice['date']} for {invoice['amount']:,.0f}, overdue {invoice['days_overdue']} days, customer: {invoice['customer_name']}"
            }],
        )
        return response.choices[0].message.content

Case Study: Manufacturing Company, 300 Documents/Day

Situation: 4 accountants processed 300 documents daily (invoices, acts, waybills). 40% of time — manual data entry and reconciliation.

AI Accountant handled:

• Scans and PDFs via OCR + Claude Vision
• Automatic vendor verification
• Contract matching
• Journal entry generation (XML upload)
• Weekly AR report

Results:

• Documents processed without accountant: 68%
• Data entry errors: -94%
• Average document processing time: 12 min → 40 sec
• Accountants focus on: complex documents, tax issues, audit

Compliance Note: Integration with vendor registration database, tax verification systems, and compliance with financial document standards required.

Timeline

• OCR + document extraction: 2–3 weeks
• Verification logic and references: 1–2 weeks
• Entry generator + expense classifier: 2 weeks
• System integration: 2–3 weeks
• AR monitoring and communications: 1–2 weeks
• Total: 8–12 weeks