为 AI Agent 设计数据层:从事务系统到 MCP 与语义模型

2026-08-29 47 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

企业把大语言模型接入业务系统后,真正棘手的问题通常不是“模型会不会回答”,而是“模型能否在安全、准确、可控成本的前提下访问正确的数据”。Fabiane Nardon 分享了 TOTVS 面向企业级 AI Agent 的数据层思路:让确定性的事务逻辑继续由数据库和服务负责,把 LLM 放在理解意图、选择工具和组织结果的位置上。

这套方法的核心,是同时处理三种约束:事务系统要求精确和安全,LLM 天生具有非确定性,Agent 又会迅速消耗上下文窗口和推理预算。数据网格、低延迟数据库、语义本体以及动态 MCP 工具选择,正好对应了这些约束。

让确定性逻辑留在正确的位置

在订单、库存、账务、权限等场景中,数据层不能把最终判断交给模型。LLM 可以识别用户意图,例如“找出缺货但已经承诺交付的订单”,却不应该自行计算库存余额、拼接任意 SQL,或绕过既有权限系统。

更稳妥的分工是:

  • LLM 负责理解:解析自然语言、识别实体、判断用户想完成的任务。
  • 语义层负责映射:把“缺货”“承诺交付”“客户”映射到统一的业务概念和字段。
  • 工具或服务负责执行:通过受控 API、查询服务或 MCP 工具访问事务数据。
  • 数据库负责保证事实:执行事务、一致性校验、权限过滤和聚合计算。

这样做的价值不只是减少幻觉。它还能保留审计链路:系统可以记录 Agent 选择了哪个工具、传入了哪些结构化参数、工具返回了什么结果,以及最终答案引用了哪些事实。

数据网格与低延迟访问

企业数据通常分布在 ERP、CRM、订单、供应链和财务系统中。把所有数据复制到一个“给 AI 使用的总库”看起来简单,却会带来同步延迟、数据所有权模糊和权限失效等问题。

数据网格思路强调由领域团队拥有数据产品,并通过明确的契约对外提供数据。对 Agent 来说,重要的不是知道每张底层表,而是获得稳定、可描述、带权限边界的领域能力,例如:

  • orders.search_commitment_risk
  • inventory.get_available_stock
  • customer.get_credit_status

低延迟架构也需要分层处理。实时库存和订单状态可以从面向查询优化的数据库或缓存读取;较慢的分析结果则可以预先计算;不适合实时访问的历史数据,应通过异步任务或数据产品提供,而不是让 Agent 在一次对话中等待多套系统串行查询。

这里有一个关键判断:Agent 的上下文不应等同于数据库返回的全部行。数据服务应该返回结构化摘要、分页游标、置信信息和数据时间戳,让模型只看到完成当前任务所需的事实。

语义模型和动态 MCP 工具选择

如果每个工具都暴露几十个字段和复杂的参数说明,模型在选择工具时就会消耗大量 token。更严重的是,工具描述之间可能出现同义词和边界冲突,例如“可用库存”“现有库存”“物理库存”在不同系统里含义并不相同。

语义本体可以把这些概念统一起来。它至少应描述:

  • 业务实体:订单、产品、仓库、客户。
  • 关系:订单包含产品,产品存放在仓库,客户拥有订单。
  • 指标定义:可用库存是否扣除了预留量,销售额是否包含税费。
  • 数据来源和新鲜度:字段来自哪个领域系统,允许多长时间的延迟。
  • 访问策略:哪些角色可以读取或修改这些信息。

MCP 可以作为模型与工具之间的标准化边界。实践中不必把所有工具一次性放进上下文,而是可以根据用户意图、用户权限和当前任务动态筛选候选工具。模型先看到短小的工具目录,确定任务类型后,系统再注入对应领域的详细 schema。

这种两阶段选择能同时降低 token 成本和误调用概率:目录负责“找方向”,详细工具负责“执行动作”。不过,动态选择器本身必须是确定性的安全组件,至少要在服务端再次检查权限、参数范围和工具白名单,不能把安全性寄托在模型遵守提示词上。

一个可改造的动态工具选择示例

下面的 Python 示例只使用标准库,演示一个简化的工具目录筛选器。它不是完整 MCP Server,而是可以放在 Agent 编排层前面的最小原型:先按领域和权限筛选,再限制注入上下文的工具数量。

运行前可以直接保存为 tool_selector.py,然后执行 python tool_selector.py。真实项目中,应把 TOOLS 换成从 MCP 注册中心或配置服务读取的数据,并在服务端执行独立的授权校验。

from dataclasses import dataclass
from typing import Iterable


@dataclass(frozen=True)
class Tool:
    name: str
    domain: str
    description: str
    roles: frozenset[str]


TOOLS = (
    Tool(
        name='inventory.get_available_stock',
        domain='inventory',
        description='Return available stock after reservations for a product and warehouse.',
        roles=frozenset({'sales', 'planner'}),
    ),
    Tool(
        name='orders.search_commitment_risk',
        domain='orders',
        description='Find orders whose promised date is at risk because of stock shortage.',
        roles=frozenset({'sales', 'planner', 'manager'}),
    ),
    Tool(
        name='finance.get_customer_credit',
        domain='finance',
        description='Return customer credit status and approved limit.',
        roles=frozenset({'manager', 'finance'}),
    ),
)


def select_tools(
    intent: str,
    role: str,
    domains: Iterable[str],
    limit: int = 2,
) -> list[Tool]:
    allowed_domains = set(domains)
    words = set(intent.lower().replace(',', ' ').split())

    candidates = []
    for tool in TOOLS:
        if tool.domain not in allowed_domains or role not in tool.roles:
            continue
        score = sum(word in tool.description.lower() for word in words)
        candidates.append((score, tool))

    candidates.sort(key=lambda item: (-item[0], item[1].name))
    return [tool for score, tool in candidates[:limit] if score > 0]


def build_context(tools: list[Tool]) -> str:
    return '\n'.join(
        f'- {tool.name}: {tool.description}'
        for tool in tools
    )


if __name__ == '__main__':
    intent = 'Which promised orders are at risk because of stock shortage?'
    selected = select_tools(intent, role='sales', domains={'orders', 'inventory'})
    print(build_context(selected))

这个原型仍有几个生产化要求:工具描述应来自版本化契约;授权应在每次工具调用时重新验证;查询结果应带上租户、时间戳和数据来源;对写操作则应增加幂等键、人工确认和审计记录。对于高风险操作,可以让 Agent 只生成结构化意图,由确定性的业务服务完成最终执行。

落地时检查四个边界

可以把以下清单作为数据层改造的起点:

  1. 事实边界:哪些结果必须由数据库或领域服务计算,模型只能解释而不能推断?
  2. 权限边界:工具目录、工具参数和工具返回结果是否都执行了租户与角色过滤?
  3. 上下文边界:每个任务实际需要哪些字段,是否可以用摘要、分页和按需 schema 替代全量数据?
  4. 延迟边界:哪些数据必须实时,哪些数据允许秒级或分钟级延迟,哪些任务应该异步完成?

采用 MCP 并不会自动解决企业数据治理问题,语义模型也不能替代领域建模。更现实的路线是从一个高价值、低风险的只读流程开始,先建立统一概念、稳定工具契约和可观测调用链,再逐步扩展到跨领域检索以及需要确认的写操作。

对于事务型企业系统,优秀的 Agent 数据层不是把更多数据塞进上下文,而是让模型在恰当的时刻看到更少、更准确、可验证的数据。确定性系统守住事实和安全,语义层负责消除业务歧义,MCP 则把这些能力以可组合的工具形式交给 Agent。


相关推荐