把财务数据接入 AI 知识图谱的工程实践

如果不用手动翻找，而是直接向 AI 提问，查询发票、未付账单和财务记录，会怎么样？

这就是我最近为 Cognee Community 提交的一项开源贡献背后的想法：为 QuickBooks Online 开发一个数据连接器。

🔗 Pull Request：cognee-community #402

📌 Issue：cognee #4802

QuickBooks 中存储着有价值的业务信息，但 AI 系统需要一种方式来摄取并理解这些数据。

我开发的连接器，可以将 QuickBooks 财务记录接入 Cognee 的 AI 记忆工作流。

这个连接器支持：

📄 摄取发票、供应商账单和贷项通知单。

🔐 OAuth 2.0 身份认证和 token 刷新。

🔄 通过增量同步获取新增和更新的记录。

🗑️ 检测删除操作，帮助移除过时的信息。

🛡️ 处理重试，并提供防止意外删除的保护措施。

🧪 自动化测试，以及使用 DuckDB 进行离线流水线验证。

基本工作流程如下：

QuickBooks → Python Connector → DLT → Cognee → Knowledge Graph

连接器获取财务记录，将其格式化为可读的 Markdown 文档，再送入 Cognee 的数据摄取工作流。

例如，系统不必直接处理原始的发票 JSON，而是可以处理客户名称、发票总额、付款状态和明细项目等信息。

这样一来，这些信息就更便于后续的 AI 搜索和知识提取。

🔍 最大的挑战：保持数据同步

获取记录只是构建可靠连接器的一部分。

发票发生变化时怎么办？如果有人删除了一条记录呢？如果 API 请求失败，没有返回任何数据，又该怎么办？

我使用 Metadata.LastUpdatedTime 实现了增量同步，通过 QuickBooks Change Data Capture（CDC）检测删除操作，还实现了一套兜底的数据核对机制。

我还添加了一项安全检查，防止某次同步扫描结果为空时，意外地将之前已知的记录标记为待删除。

提交的测试套件记录了 12 项通过的测试，覆盖数据格式化、同步、删除处理和重试行为。

这次贡献让我认识到，构建数据连接器，远不只是调用 API。

你需要考虑数据一致性、身份认证、同步状态、故障处理，以及边界情况的测试。

打磨这些细节，让我积累了 Python、REST API、DLT、数据流水线、测试和 AI 知识图谱方面的实践经验。

🙌 探索这个项目

我的 Pull Request：cognee-community #402

原始 Issue：cognee #4802

Cognee：topoteretes/cognee

Cognee Community：topoteretes/cognee-community

感谢 Cognee 的维护者和社区，为大家提供学习和参与开源 AI 基础设施建设的机会。

这个 Pull Request 目前仍处于开放状态，等待审查，也欢迎其他开发者提出反馈。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。