ChatGPT 报销自动化：从零搭建企业级费用管理机器人

1次阅读

没有评论

共计 3684 个字符，预计需要花费 10 分钟才能阅读完成。

在企业日常运营中，员工使用 ChatGPT 产生的费用报销是一个常见但繁琐的流程。传统报销方式存在以下高频问题：

多账户费用归集：员工可能使用个人或不同部门的 OpenAI 账户，导致费用分散难以统一管理。
跨境结算汇率差：OpenAI 的结算货币通常是美元，与企业本地货币存在汇率波动风险。
人工审核效率低：财务人员需要手动核对每笔交易的合理性，耗时且容易出错。

根据某中型企业（500 人规模）的调研数据，每月处理 ChatGPT 报销的工时约为 40 人 / 小时，平均每笔报销需要 3 天完成审批，错误率高达 15%。

系统采用四层架构设计：

费用抓取层：从邮箱、API 等渠道获取原始账单数据
语义解析层：提取关键信息（金额、时间、用途等）
审批路由层：根据规则触发对应审批流程
财务系统对接层：生成会计凭证并同步到 ERP

flowchart LR
    A[费用抓取] --> B[语义解析]
    B --> C[审批路由]
    C --> D[财务系统]

正则表达式：开发快但维护成本高，难以应对账单格式变化
GPT-3.5 微调：准确率高但需要标注数据，适合复杂场景
现成 NLP 服务：开箱即用但可能有数据出境合规风险

Celery：适合实时性要求高的场景，需要维护消息队列
Airflow：擅长复杂依赖的任务编排，学习曲线较陡
自建轮询：实现简单但缺乏失败重试等企业级功能

from typing import Optional
import openai
from tenacity import retry, stop_after_attempt, wait_exponential

class OpenAIClient:
    """封装带自动重试的 API 调用"""

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def get_usage(self, api_key: str, start_date: str, end_date: str) -> Optional[dict]:
        try:
            openai.api_key = api_key
            # 实际调用替换为 OpenAI 的 usage 接口
            return await openai.Usage.retrieve(start_date=start_date, end_date=end_date)
        except openai.error.AuthenticationError:
            # 记录无效 API_KEY 但不重试
            return None

import PyPDF2
from email.parser import BytesParser

def parse_pdf(file_path: str) -> dict:
    """解析 PDF 格式账单"""
    with open(file_path, 'rb') as f:
        reader = PyPDF2.PdfReader(f)
        text = ' '.join(page.extract_text() for page in reader.pages)
    return {
        'source': 'pdf',
        'content': text
    }

def parse_email(raw_email: bytes) -> dict:
    """解析邮件原始内容"""
    msg = BytesParser().parsebytes(raw_email)
    return {
        'source': 'email',
        'subject': msg['subject'],
        'body': msg.get_payload()}

import httpx
from pydantic import BaseModel

class DingTalkApproval(BaseModel):
    """钉钉审批流模型"""
    process_code: str
    originator_userid: str
    form_component_values: list

async def trigger_approval(
    client: httpx.AsyncClient,
    access_token: str,
    approval: DingTalkApproval
) -> str:
    """触发审批流程并返回实例 ID"""
    resp = await client.post(
        'https://oapi.dingtalk.com/topapi/processinstance/create',
        json=approval.dict(),
        params={'access_token': access_token}
    )
    resp.raise_for_status()
    return resp.json()['process_instance_id']

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_openai_client():
    """测试 API 重试逻辑"""
    mock_client = AsyncMock()
    mock_client.get_usage.side_effect = [Exception('Timeout'),
        {'data': 'test'}
    ]

    result = await mock_client.get_usage('fake_key', '2023-01-01', '2023-01-31')
    assert result == {'data': 'test'}
    assert mock_client.get_usage.await_count == 2

使用短期有效的 JWT 令牌，通过 refresh_token 自动续期
日志存储符合 PCI-DSS 标准：
敏感字段 (如 API_KEY) 脱敏
日志加密存储
保留至少 180 天

月末批量处理方案：

使用 aiohttp 实现异步 IO
按部门分片处理
失败任务自动进入重试队列

import asyncio
from aiohttp import ClientSession

async def batch_process(api_keys: list[str]):
    """并发处理多个账户"""
    async with ClientSession() as session:
        tasks = [fetch_account_usage(session, key)
            for key in api_keys
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

常见陷阱：

OpenAI 账单使用 UTC 时间
企业财务系统使用本地时区(如 UTC+8)
月末跨时区可能导致日期错位

解决方案：

from datetime import datetime, timedelta
import pytz

def convert_timezone(
    utc_time: str,
    target_tz: str = 'Asia/Shanghai'
) -> str:
    """UTC 时间转本地时间"""
    utc = datetime.strptime(utc_time, '%Y-%m-%dT%H:%M:%SZ')
    return utc.replace(tzinfo=pytz.utc).astimezone(pytz.timezone(target_tz)
    ).strftime('%Y-%m-%d %H:%M:%S')

当网络超时导致状态不明确时：

先查询是否已存在相同审批
使用唯一 IDempotency-Key
记录操作状态到数据库

扩展方案：

抽象 Provider 接口
配置化账单解析规则
统一货币结算

class AIProvider(ABC):
    @abstractmethod
    async def get_usage(self) -> list[UsageRecord]:
        pass

class UsageRecord(BaseModel):
    """标准化用量记录"""
    amount: Decimal
    currency: str
    usage_date: date
    service_type: str

技术实现要点：

存储原始对话元数据
建立审计日志关联链
实现基于时间的快照查询

CREATE TABLE audit_logs (
    id BIGSERIAL PRIMARY KEY,
    operation VARCHAR(50) NOT NULL,
    parameters JSONB NOT NULL,
    user_id VARCHAR(100) NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    trace_id UUID NOT NULL  -- 全链路追踪
);

通过本文介绍的自动化方案，某客户实际实现了：