1. 项目概述：ERP财务辅助Agent开发全流程
最近在做一个挺有意思的项目——基于DeepSeek API的ERP财务辅助Agent开发。这个项目的核心目标是用AI技术来简化企业财务流程中的重复性工作，比如自动对账、凭证生成这些让人头疼的活。作为一个在财务系统领域摸爬滚打多年的开发者，我深知传统ERP系统在智能化方面的短板，这次就想试试用AI Agent来补上这块短板。
这个Demo版本采用了本地模拟+API调用的混合架构，既能快速验证核心功能，又为后续对接真实ERP系统留好了接口。整个开发过程从环境搭建到Docker部署大概花了两周时间，期间踩了不少坑，也积累了一些实用的经验。项目代码已经完整开源，你完全可以从零开始复现整个开发流程。
特别说明：虽然项目使用了DeepSeek API，但核心架构设计是通用的，你可以轻松替换成其他大模型API。我在代码里专门做了抽象层处理，这点后面会详细讲到。
2. 技术选型与核心架构
2.1 技术栈组成
这个项目用到了几个关键组件：
- DeepSeek API：作为核心的AI能力提供方，处理自然语言理解和财务逻辑推理
- FastAPI：构建RESTful接口，方便后续与企业现有系统集成
- SQLite：本地轻量级数据库，存储财务规则和操作记录
- Docker：容器化部署，确保环境一致性
选择这套技术组合主要考虑了几个因素：
- 开发效率：FastAPI的异步特性非常适合AI应用的高并发场景
- 成本控制：SQLite无需额外部署数据库服务，适合Demo阶段
- 扩展性：Docker打包后可以无缝迁移到生产环境
2.2 系统架构设计
整个系统采用分层设计，从上到下依次是：
- 接口层：处理HTTP请求和响应，包括身份验证和参数校验
- 业务逻辑层：核心的财务规则引擎和AI调用封装
- 数据访问层：操作本地数据库和模拟的ERP数据
- 基础设施层：日志、监控等支撑组件
python复制# 典型的调用链示例
@app.post("/voucher/generate")
async def generate_voucher(request: VoucherRequest):
# 参数校验
validate_request(request)
# 调用业务逻辑
voucher_data = await voucher_service.generate(
request.transaction_data,
request.accounting_standards
)
# 记录操作日志
audit_logger.log(request.user_id, "voucher_generate")
return {"data": voucher_data}
这种分层设计最大的好处是职责清晰，后续如果要替换某个组件（比如从SQLite迁移到MySQL）或者增加新功能，影响范围可以控制在单个层内。
3. 核心功能实现细节
3.1 财务规则引擎
财务处理最核心的就是要符合会计准则，我在项目里实现了一个轻量级的规则引擎。这个引擎主要做三件事：
- 科目匹配：根据交易描述自动匹配会计科目
- 借贷平衡：确保每笔凭证的借贷方金额相等
- 辅助核算：处理涉及部门、项目等维度的核算需求
规则配置采用JSON格式，方便非技术人员维护：
json复制{
"rule_name": "office_supplies_purchase",
"description": "办公用品采购",
"match_keywords": ["办公用品", "文具", "耗材"],
"debit_account": "6602-办公费",
"credit_account": "1002-银行存款",
"required_fields": ["department", "purpose"]
}
实际开发中发现，单纯依赖关键词匹配准确率只有70%左右。后来加入了金额区间判断（比如办公用品单笔通常不超过1万元）和上下文关联（比如同一供应商连续交易）后，准确率提升到了92%。
3.2 AI集成方案
DeepSeek API的集成我做了两层封装：
- 基础封装层：处理认证、重试、限流等通用逻辑
python复制class AIClient:
def __init__(self, api_key):
self.session = ClientSession()
self.api_key = api_key
self.retry_policy = ExponentialBackoff(retries=3)
async def chat_completion(self, messages):
headers = {"Authorization": f"Bearer {self.api_key}"}
async with self.retry_policy:
async with self.session.post(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
json={"messages": messages}
) as resp:
return await resp.json()
- 领域适配层：将财务专业需求转化为AI能理解的prompt
python复制def build_voucher_prompt(transaction):
return [
{"role": "system", "content": "你是一名资深会计，需要根据以下交易信息生成记账凭证..."},
{"role": "user", "content": f"""
交易时间：{transaction.date}
交易对方：{transaction.counterparty}
金额：{transaction.amount}
摘要：{transaction.description}
请按照{transaction.standard}准则处理
"""}
]
这种设计有个很大的好处——当需要切换AI提供商时，只需要修改基础封装层的实现，业务代码完全不用动。我在项目里专门写了测试用例来验证这点的有效性。
4. 本地模拟环境搭建
4.1 模拟数据生成
为了方便测试，我开发了一个数据模拟器，可以生成各种类型的财务交易数据：
- 供应商付款
- 客户收款
- 费用报销
- 固定资产采购
模拟器支持配置多种参数：
python复制class TransactionGenerator:
def __init__(self):
self.companies = ["阿里", "腾讯", "百度", "京东"]
self.departments = ["财务部", "技术部", "市场部", "人事部"]
self.expense_types = ["差旅", "招待", "办公", "采购"]
def generate_payment(self, amount_range=(1000, 50000)):
amount = random.randint(*amount_range)
return {
"date": datetime.now().strftime("%Y-%m-%d"),
"counterparty": random.choice(self.companies),
"amount": amount,
"description": f"支付{random.choice(['货款', '服务费', '保证金'])}",
"type": "payment"
}
重要提示：模拟数据虽然方便测试，但要注意不要和真实财务数据混淆。我在代码里特意加了明显的测试标记，数据库也用了单独的test模式。
4.2 接口模拟方案
对于暂时没有权限访问的真实ERP接口，我用FastAPI模拟了以下几个关键端点：
/api/transactions获取交易列表/api/vouchers凭证查询与创建/api/ledgers科目余额查询
模拟接口不仅返回预设的数据，还会记录调用日志，方便调试：
python复制@app.post("/api/vouchers")
async def create_voucher(voucher: dict):
logger.info(f"Voucher created: {voucher}")
# 模拟处理延迟
await asyncio.sleep(0.5)
return {"success": True, "voucher_id": str(uuid.uuid4())}
5. 部署与运维方案
5.1 Docker化部署
项目提供了完整的Docker支持，包括：
- 应用容器（基于python:3.10-slim）
- Redis缓存（可选）
- Nginx反向代理（可选）
docker-compose.yml关键配置：
yaml复制version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DEEPSEEK_API_KEY=${API_KEY}
volumes:
- ./logs:/app/logs
redis:
image: redis:alpine
ports:
- "6379:6379"
部署时发现一个常见问题：直接挂载代码目录在Windows上会有文件权限问题。解决方案是：
- 在Dockerfile里明确设置工作目录权限
- 或者使用docker volume代替直接挂载
5.2 监控与日志
考虑到财务系统的特殊性，我加强了日志记录：
- 所有凭证生成操作记录完整上下文
- AI调用请求和响应记录（脱敏后）
- 数据库变更审计日志
日志采用JSON格式，方便接入ELK等系统：
json复制{
"timestamp": "2023-08-20T14:32:15Z",
"level": "INFO",
"service": "voucher-service",
"operation": "generate_voucher",
"user": "demo_user",
"transaction_id": "txn_12345",
"metadata": {
"amount": 5000.00,
"accounts": ["1001", "6602"]
}
}
6. 典型问题排查实录
6.1 科目匹配错误
现象：采购固定资产被错误识别为费用支出
排查过程：
- 检查关键词配置，发现"设备采购"确实同时匹配了固定资产和办公费
- 查看AI生成的判断依据，发现模型过度依赖金额大小（5万元以下判为费用）
- 检查交易上下文，发现缺少供应商类型信息
解决方案：
- 在规则中增加供应商白名单
- 添加金额与科目关联矩阵
- 对模糊匹配的情况要求人工确认
6.2 API限流处理
现象：月末批量处理时频繁出现429错误
优化方案：
- 实现令牌桶算法进行客户端限流
- 增加指数退避重试机制
- 对批量操作采用异步队列处理
关键代码：
python复制class RateLimiter:
def __init__(self, rate=5, period=1):
self.rate = rate
self.period = period
self.tokens = rate
self.last_check = time.time()
async def acquire(self):
now = time.time()
elapsed = now - self.last_check
if elapsed > self.period:
self.tokens = self.rate
self.last_check = now
if self.tokens >= 1:
self.tokens -= 1
return True
await asyncio.sleep(self.period - elapsed)
return await self.acquire()
7. 扩展方向与优化建议
虽然当前版本已经实现了基础功能，但还有不少可以提升的地方：
-
多会计准则支持：
- 目前主要针对中国企业会计准则
- 可以增加IFRS、US GAAP等国际准则的处理逻辑
- 需要扩展科目映射表和规则引擎
-
异常检测能力：
- 基于历史数据训练简单模型
- 检测异常交易（如金额突增、非工作时间交易）
- 与风控系统集成
-
性能优化：
- 对AI调用实现批处理
- 增加缓存层（特别是科目余额查询）
- 考虑使用gRPC替代部分HTTP调用
-
安全增强：
- 增加操作二次确认机制
- 完善审计日志
- 敏感数据加密存储
这个项目最让我惊喜的是AI在财务规则理解上的表现。比如有笔交易描述是"支付深圳腾讯科技有限公司季度技术服务费"，模型不仅能准确识别出要计入"技术服务费"科目，还自动添加了"腾讯科技"作为辅助核算项。这种语义理解能力是传统规则引擎很难实现的。