基于Ollama和LangChain在本地大型语言模型上运行的QuickBooks财务处理智能体。

这是一款可接入 QuickBooks Online 的财务助手，它能获取企业的实时账目数据，让企业主能够与人工智能就实际数字进行交流：真实的发票、真实的现金流、真实的应收账款账龄，而非演示数据。该模型完全在笔记本电脑上运行，通过 Ollama 提供的 NVIDIA GeForce RTX 4070 及 Qwen 3.5 9B 模型来完成所有推理工作，因此在开发过程中无需调用 OpenAI、Anthropic，也不需要云服务或支付任何人工智能使用费用。

接下来的内容将介绍实际代码、架构选择以及该助手的输出结果，而非理论部分。

它解决的问题

使用 QuickBooks 的小型企业主虽然将所有财务数据集中存储在一处，但要将其转化为有价值的洞察，要么需要会计技能，要么需要付费的专业帮助。标准报表只能说明已经发生的情况，却无法指出下一步该做什么。

其目的是让用户输入关于自己财务状况的简单英文问题，从而获得基于数据的具体回复：发票编号、客户姓名以及真实的资金状况，而非泛泛的建议。

系统架构

该系统由三层构成：

• 前端：Next.js 14，搭配 TypeScript 和 shadcn/ui
• 后端：FastAPI、PostgreSQL 以及 SQLAlchemy
• AI部分：LangChain ReAct 智能体、通过 Ollama 调用的 Qwen 3.5 9B 模型，以及一组 QuickBooks 工具

最重要的决策是选择在 LangChain 的智能体框架上构建 AI 层。这一设计使得模型能够在回答之前自行决定需要获取哪些数据，因此回复才会具体而非笼统。

在笔记本电脑 GPU 上运行

整个 AI 层运行在消费级笔记本电脑的 GPU 上，即拥有 8 GB VRAM 的 RTX 4070。具体性能数据如下。

响应时间大约为一分钟。对于财务分析而言，这个速度已经足够，因为提出实质性问题的人期望得到详尽的答复，而非机器人的闲聊。在实际应用中，将 LangChain 抽象层指向云推理服务，响应时间可降至五秒以内，且应用程序代码无需做任何修改。

有三个原因使得在本地开发成为最佳选择：

• 无需任何成本。每一次提示词调整、调试过程以及边缘情况测试都可以在不产生费用且不受速率限制的情况下进行。
• 能够保障数据隐私。QuickBooks 的数据从未离开过本地机器，因此没有任何发票金额、客户名称或现金流数据会传送到第三方 AI 服务提供商那里。
• 能够证明可行性。如果基于消费级 GPU 的 90 亿参数模型就能产生如此高质量的输出，那么构建本地 AI 应用所需的硬件标准其实比许多开发者想象的要低。

由于 LangChain 处于中间位置，开发环境和生产环境共享同一个代码库。更换推理后端仅需两行代码：

工具、内存、智能体逻辑以及提示词均保持不变，模型只是实现细节而已。

智能体

AI 层使用 LangChain 的 ReAct 智能体框架。该智能体拥有一组工具，这些工具实际上是用于获取实时 QuickBooks 数据的普通 Python 函数，智能体会根据用户的问题选择调用相应的函数。以下是 agent.py 中的完整配置：

from langchain_ollama import ChatOllama
from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain_core.runnables import RunnableConfig
from quickbooks_client import fetch_invoices, fetch_profit_loss, \
                             fetch_cash_flow_summary, fetch_ar_aging

llm = ChatOllama(
    model='qwen3.5:9b',
    base_url='http://127.0.0.1:11434'
)

@tool
def get_invoices(config: RunnableConfig) -> str:
    """Fetch all invoices from QuickBooks including amounts and balances."""
    user_id = config.get('configurable', {}).get('user_id')
    return json.dumps(fetch_invoices(user_id=user_id), default=str)

@tool
def get_ar_aging(config: RunnableConfig) -> str:
    """Get AR aging report with buckets and DSO calculation."""
    user_id = config.get('configurable', {}).get('user_id')
    return json.dumps(fetch_ar_aging(user_id=user_id), default=str)

@tool
def draft_collection_notice(customer_name: str, amount: float,
                            days_overdue: int) -> str:
    """Drafts a reminder for an overdue invoice based on severity."""
    if days_overdue > 60:
        tone = 'Firm & Urgent'
        msg = f'Final reminder: your balance of ${amount} is significantly overdue.'
    elif days_overdue > 30:
        tone = 'Standard Follow-up'
        msg = f'Checking on your ${amount} payment, now 30+ days past due.'
    else:
        tone = 'Friendly Reminder'
        msg = f'Quick note: your payment of ${amount} was due recently.'
    return json.dumps({'tone': tone, 'draft': msg})

tools = [get_invoices, get_profit_loss, get_cash_flow_summary,
         get_ar_aging, draft_collection_email, draft_collection_notice]

agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt=(
        'You are an AI CFO assistant for small businesses. '
        'Analyze financial data and give clear, actionable advice. '
        'Always highlight cash flow risks and overdue invoices. '
        'Be concise and specific with numbers.'
        '\n\n--- COLLECTIONS SPECIALIZATION ---'
        '\nWhen asked about unpaid invoices or specific customers:'
        '\n1. Use get_ar_aging to identify debt seniority.'
        '\n2. Use draft_collection_notice to generate professional follow-ups.'
        '\n3. Explain the impact on liquidity, not just the invoice.'
    )
)

有两点值得注意。首先，user_id会通过LangChain的RunnableConfig传递到每一次工具调用中，这正是该系统实现多租户架构的关键：每个用户的智能体仅能访问其自己的QuickBooks数据。其次，智能体在决定调用哪个工具时实际上会读取工具的文档字符串，因此这些文档应作为给模型的指令来编写，而非面向人类的描述。

对话历史

每个会话都会保留自己的历史记录，这样后续问题就可以参考之前的对话内容：

conversation_history: dict[str, list] = {}

def ask(question: str, session_id: str = 'default',
        user_id: str = None) -> str:
    if session_id not in conversation_history:
        conversation_history[session_id] = []

    conversation_history[session_id].append(
        {'role': 'user', 'content': question}
    )

    config = {'configurable': {
        'user_id': user_id,
        'session_id': session_id
    }}

    response = agent.invoke(
        {'messages': conversation_history[session_id]},
        config=config
    )

    answer = response['messages'][-1].content
    conversation_history[session_id].append(
        {'role': 'assistant', 'content': answer}
    )
    return answer

用户可以询问“我的DSO是多少？”，然后再问“是哪些客户导致了这个数值？”，客服人员能够理解这里的“那个”指的是什么。历史记录以session_id为键，因此不同用户之间的对话是相互独立的。

聊天端点

聊天功能背后的FastAPI端点非常简单。关键在于user_id来自经过验证的JWT，而非请求体，这样就能防止任何人通过修改参数来获取其他用户的数据。

@app.post('/ask')
async def ask_agent(
    request: AskRequest,
    current_user: User = Depends(get_current_user)
):
    # Always use the verified user from JWT, never trust the request body
    user_id = str(current_user.id)

    answer = ask(
        question=request.question,
        session_id=request.session_id,
        user_id=user_id
    )

    return {
        'question': request.question,
        'answer': answer,
        'session_id': request.session_id
    }

令牌刷新

QuickBooks 客户端会自动刷新 OAuth 令牌。这些令牌在60分钟后过期，因此客户端会在每次请求前检查令牌的有效期，并在需要时进行更新：

def get_qb_client(user_id: str = None):
    db = SessionLocal()
    qb_token = db.query(QBToken).filter(
        QBToken.user_id == uuid.UUID(user_id)
    ).first()

    if not qb_token:
        raise Exception(f'No QB token found for user {user_id}')

    auth_client = AuthClient(
        client_id=os.getenv('QB_CLIENT_ID'),
        client_secret=os.getenv('QB_CLIENT_SECRET'),
        redirect_uri=os.getenv('QB_REDIRECT_URI'),
        environment=os.getenv('QB_ENVIRONMENT'),
        access_token=qb_token.access_token,
        refresh_token=qb_token.refresh_token,
    )

    # Only refresh if token is older than 55 minutes
    token_age = datetime.utcnow() - qb_token.updated_at
    if token_age > timedelta(minutes=55):
        auth_client.refresh()
        qb_token.access_token  = auth_client.access_token
        qb_token.refresh_token = auth_client.refresh_token
        qb_token.updated_at    = datetime.utcnow()
        db.commit()

    return QuickBooks(
        auth_client=auth_client,
        refresh_token=auth_client.refresh_token,
        company_id=qb_token.realm_id,
        minorversion=75
    )

在55分钟而非60分钟时进行刷新，可以为避免时间偏差预留缓冲。刷新操作在服务器端完成，令牌存储在PostgreSQL中，而从不在前端。

智能体生成的内容

这才是关键部分。以下是针对“我的DSO是多少，状况是否良好？”这一问题，在QuickBooks沙箱数据上运行后的实际输出：

这里没有任何预设内容。智能体调用了get_ar_aging函数，获取了实时的数据并对其进行了处理。回复中列出了相关客户名称，估算了延迟对现金流的影响，并建议起草催款通知。它能做到这一点是因为存在draft_collection_notice工具，且智能体能够判断何时适用该工具。

为何这不是一个简单的RAG机器人

它与普通检索型聊天机器人的区别在于，该智能体能够自行选择使用的工具。当被问及“我最大的现金流风险是什么？”时，它并不会尝试运行所有相关功能，而是先对问题进行分析，再挑选出相关的工具。ReAct循环的运作方式如下：

针对这个问题，它会跳过get_invoices和get_profit_loss，因为这些功能与问题无关。这种选择性使得回答更加精准，同时避免上下文窗口被不必要的数据填满。

控制面板

该控制面板使用的是与智能体相同的QuickBooks工具。所有数据都是实时的，用户与QuickBooks之间没有缓存层，因此账龄图表以及逾期最严重的发票列表会在每次页面加载时自动更新。

AR账龄分析页面超越了QuickBooks的常规报表功能。它能实时计算DSO值，为每个区间分配风险等级，并列出每张发票的确切逾期天数。这些数据均来自代理程序调用的同一個fetch_ar_aging函数，从而确保数据来源的统一性。

值得特别提及的设计选择

• 只读型OAuth授权。所申请的权限不包含写入权限，因此该应用从结构上就无法修改QuickBooks中的数据。
• 使用JWT中的user_id。/ask接口仅信任经过验证的令牌，而不考虑客户端发送的其他任何内容。
• 不会存储任何财务相关数据。只有在需要时才会从QuickBooks获取数据，使用后立即丢弃。数据库中仅保存用户信息、令牌及调度安排。
• 开发过程中使用本地大语言模型。在开发和测试阶段，没有任何财务数据被传输到外部AI服务提供商。

与优先采用云服务的方式相比，本地化部署的实际优势在于：

其成本体现为延迟，4070显卡的延迟约为60秒，而云API的延迟则低于5秒。对于那些需要得到实质性答案的用户而言，这在开发和测试阶段尚可接受，而在生产环境中转为云模式也极为简单。

所需时间

原本需要数周才能完成的工作现在几天内就能搞定。以下是全程使用AI编码辅助工具的独立开发者的大致时间安排：

核心系统在三到四天后开始运行，之后还需一天时间进行优化和测试。最耗时的三个方面是：QuickBooks API中的各种问题（其报表布局极不直观）、对LangChain智能体的调优（工具的文档说明比预期更重要），以及为实现多租户架构而在每次工具调用中正确传递user_id。速度的提升主要得益于FastAPI内置的验证功能、LangChain的提供者抽象机制，以及用于处理样板代码、错误调试和架构问题的AI编程辅助功能。

需要改进的地方

如果重新开发，会有几处做法会不同：

• 从一开始就将各种工具与智能体连接起来。之前是先完成数据层，之后才将其连接到智能体，因此接口不匹配的问题直到后期才显现出来。

在30亿参数的模型上测试提示词功能。90亿参数的Qwen虽然输出质量更高，但在当前GPU上运行速度很慢，因此若能使用小型模型快速进行提示词实验，再用90亿参数的模型做最终验证，就能节省大量时间。

从一开始就支持流式输出。屏幕上没有任何内容，只能等待一分钟的情况实在令人难受，而流式输出功能本应早在第一阶段就实现。

测试QuickBooks相关工具。只有真实数据中才会出现边缘情况，若能用沙箱数据进行小型测试，就能避免多次漫长的调试过程。

已知局限

任何客观的说明都应指出尚未完成的部分。以下是当前版本限制其扩展规模的因素，以及解决这些问题的方法。

内存中的对话历史

对话历史存储在RAM中的普通Python字典中：

这种方案适用于单个开发服务器，但在生产环境中存在三个问题：每次重启后历史记录都会丢失；无法在多个服务器实例间运行，因此无法实现水平扩展；此外会话之间没有任何数据保留，关闭浏览器就会结束当前对话。

在生产环境中，解决方案是将历史记录存储到Redis中，以实现快速读写并设置过期时间，或者存储到PostgreSQL中以实现完整的数据持久化。LangChain的消息历史抽象层支持这两种存储方式，因此切换起来十分方便。

刻意不使用RAG

该系统不使用任何检索技术：没有嵌入模型，没有向量数据库，也没有分块处理流程。这是有意为之。财务数据处于持续变化中，发票会被支付、新的发票会出现，余额也会变动，因此昨天生成的嵌入模型今天可能就已经过时。每次请求都从QuickBooks获取实时数据，这样才能确保得到的结果反映企业的当前状况。

RAG更适合处理合同、政策文件和历史报告这类静态或变化缓慢的资料。而对于实时交易数据，直接调用工具在准确性方面优于基于嵌入模型的检索方式。

其成本体现在延迟和token使用量上，因为直接获取数据速度更慢，且生成的上下文内容比向量存储中的简洁摘要更多。对于每家企业每天只有数十到几百张发票的情况，这种成本是可接受的。在规模更大的情况下，采用混合方案更为合理，即用RAG处理历史数据，用实时工具处理当前数据。

大量发票处理

目前所有发票都会完整地传递给模型。30到50张发票时效果尚可，但一旦达到数千张，就会超出令牌限制，导致质量下降。解决办法是在模型接收数据之前进行过滤：仅获取与问题相关的发票，先对大型数据集进行总结，或者采用两阶段处理方式，在开始推理之前通过快速检索缩小数据范围。

暂不支持流式处理

服务器会一次性返回完整答案，因此本地GPU用户需要等待大约一分钟才能看到结果。FastAPI提供了StreamingResponse功能，Ollama可通过其API实现令牌流式传输，LangChain的智能体也支持流式回调。虽然将这些功能连接起来很简单，但为确保核心功能正常运行，这一工作被推迟了，它将是产品发布前的优先事项。

会话无法持续

session_id是在聊天页面加载时于前端生成的。打开新标签页、刷新页面或关闭浏览器后重新返回都会生成新的session_id，历史记录也会重置，即便服务器从未重启过。财务助手若能利用持久化存储功能，比如记住某客户正在使用分期付款计划，或是上周曾讨论过某张发票，将会受益匪浅。要实现这一点，就需要将会话信息存储在数据库中，并允许用户为过往对话命名、继续查看或删除。

要点总结

LangChain的抽象层、Ollama的本地推理能力以及QuickBooks的数据访问功能相结合，产生的效果远超单一功能的简单叠加。该智能体不仅能够检索数据，还能对数据进行分析、筛选出关键信息，并根据每位用户的财务状况给出回复。

采用本地化方案并非妥协，而是一种具备真正隐私保护、成本优势及迭代效率的合理发展策略。像RTX 4070这样的消费级GPU就能够运行90亿参数的模型。对于任何计划开发人工智能应用的人来说，更重要的问题不是该选择哪家云服务提供商，而是能否在开发阶段就在本地运行高性能模型，这样就能以零推理成本完成产品验证，而关于云服务提供商的决策则可以等到有实际使用数据后再做。在RTX 4070上，答案是肯定的。