FME与AI：FME的OpenAIChatGPTConnector每处理一个feature就发一次API请求，一个含几千条记录的图层跑下来token账单直接失控——到底该在Sampler之后把多条记录聚合为一次请求，还是该按记录数/字符数设一个动态分批阈值，聚合粒度变了之后输出质量又会怎么变？

一、问题的本质：不是"请求太多"，而是"调用语义错配"

很多团队第一次被FME的ChatGPT Connector账单"教育"，是在一个看似普通的地址标准化或POI分类任务上。图层里有四千多条记录，工作流跑完，后台账单显示请求数四千多次、输入token接近百万级。直觉反应是"请求太多，合并一下就好了"，但真正的问题更深一层：Connector的调用语义是"逐feature"，而业务语义往往是"逐批次"或"逐类别"。当调用单元与业务单元不一致时，成本与质量会同时失真。

Safe Software官方文档对OpenAI ChatGPT Connector的描述是：它在转换器中为每个输入feature构造一次请求，把feature属性填入提示词模板，返回结果写回属性。这意味着请求次数与feature数量线性相关。本文评述：这不是设计缺陷，而是"通用连接器"的合理默认——它无法预知你的业务是否允许合并。真正需要工程判断的，是把"调用单元"从feature重构成batch，而这个重构点放在哪里，决定了后续所有成本与质量曲线。

一个容易被忽略的事实是：大模型的计费单位是token而非请求。OpenAI的定价页（2024—2025年多次调整）显示，输入token与输出token单价不同，且存在"每次请求的固定系统提示开销"。也就是说，逐feature调用的真实成本 = 请求数 ×（系统提示token + 单条内容token）+ 输出token。系统提示往往有几百token，当单条记录内容只有几十token时，系统提示的重复开销可能占总输入成本的70%以上。这才是账单失控的主因，而不是"记录本身太长"。

关键判断：如果你的单条记录内容token数远小于系统提示token数，那么"合并请求"的收益几乎全部来自系统提示的复用，而不是内容本身。这决定了聚合策略的第一优先级是"复用系统提示"，而非"压缩内容"。

为了把讨论落到可操作的层面，本文先给出一个贯穿全文的分析主线：聚合（Aggregate）→ 分批（Batch）→ 校验（Validate）。聚合解决"系统提示复用"，分批解决"上下文窗口与质量衰减"，校验解决"聚合带来的错误传播"。三者不是三选一，而是一条流水线上的三个工位。后文所有章节都围绕这条主线展开，避免陷入"聚合好还是分批好"的伪二元争论。

二、Token成本的解剖学：输入、输出与隐藏的重复开销

2.1 成本公式的拆解

把一次FME工作流的LLM成本写成可计算的形式，是治理的第一步。设图层有 N 条feature，逐feature调用时：

总成本 ≈ N × (T_sys + T_item) × P_in + N × T_out × P_out
其中：T_sys = 系统提示token，T_item = 单条内容token
      T_out = 单条输出token，P_in/P_out = 输入/输出单价

聚合为每批 B 条后：

总成本 ≈ ceil(N/B) × (T_sys + B × T_item) × P_in + ceil(N/B) × B × T_out × P_out
      = N × T_item × P_in + ceil(N/B) × T_sys × P_in + N × T_out × P_out

两式相减，聚合带来的节省为 (N − ceil(N/B)) × T_sys × P_in。本文评述：这个式子揭示了一个常被忽视的结论——节省只与 T_sys 和批数有关，与 T_item 无关。所以"把提示词写短"对逐feature调用的帮助有限，而"提高批大小B"的边际收益在B增大时迅速递减（因为 ceil(N/B) 下降变慢）。这为后文"动态阈值"提供了理论依据：批大小不是越大越好，存在一个由质量约束决定的拐点。

2.2 隐藏开销：重试、超时与输出膨胀

真实账单里还有三类"看不见"的成本。第一是重试：网络抖动或限流导致的失败重发，会让实际请求数高于feature数。OpenAI的速率限制文档指出，超过RPM/TPM阈值会返回429，客户端若不做指数退避，重试风暴会放大成本。第二是超时：长输出被截断后重跑，等于付了两次钱。第三是输出膨胀：要求模型"详细解释"时，输出token可能远超输入，而输出单价通常更高。

笔者在实践中见过一个典型案例（模拟数据，用于说明量级）：某地址清洗任务，单条内容约40 token，系统提示约320 token，输出约60 token。逐feature跑5000条，输入token约180万，其中系统提示占约160万。改为每批25条后，请求数降到200次，系统提示开销降到约6.4万token，输入总成本下降约85%。这个量级与公式预测一致，说明成本治理的杠杆点确实在系统提示复用。

注：上表为模拟数据，参数取自公开定价页的典型量级，仅用于说明成本结构，不代表任何具体项目实测。

三、两条路线之争：Sampler后聚合 vs 动态分批阈值

3.1 路线A：Sampler之后聚合为单次请求

做法是：用Sampler抽取需要处理的子集，再用Aggregator（或ListBuilder）把多条记录拼成一个JSON数组或分隔文本，一次性送入Connector。优点是实现简单、请求数最少、系统提示只付一次。缺点也很直接：受上下文窗口限制，几千条记录不可能一次塞进去；即便塞得下，模型对长输入的"中间遗忘"（lost in the middle）现象会显著拉低质量。

关于长上下文的质量衰减，Liu等人在2023年的研究《Lost in the Middle: How Language Models Use Long Contexts》中给出了系统性证据：当关键信息位于长上下文的中部时，模型检索与利用该信息的准确率明显低于位于开头或结尾时。本文评述：这项研究对FME用户的意义在于——"一次塞满"不仅可能超窗，即便不超窗，也会让位于批次中部的记录被系统性忽略。因此路线A只适合"总量小、且每条记录都需要被同等重视"的场景，不适合大批量。

3.2 路线B：按记录数/字符数设动态分批阈值

做法是：不追求"一次搞定"，而是设定批大小上限（按条数或按字符/token数），把记录切成若干批，每批一次请求。批大小可以是固定值，也可以根据内容长度动态调整——短记录多塞几条，长记录少塞几条，保证每批的token量落在安全区间。

动态阈值的核心是"以token预算为约束，而非以条数为约束"。因为同样25条记录，地址字段可能只有20字符，也可能有200字符，token量差一个数量级。按字符数分批（例如每批不超过8000字符）比按条数分批更稳健。本文评述：按条数分批是"工程直觉"，按token/字符分批是"成本直觉"，后者更接近真实账单。

3.3 两条路线的对比与选择矩阵

选择矩阵的结论是：绝大多数FME批处理场景应选路线B，路线A只在"需要跨记录推理"时才有价值。例如"给一批POI统一打主题标签"可能需要全局视角，而"逐条地址标准化"完全不需要。判断标准很简单：如果任务对每条记录的处理是独立的，就分批；如果任务依赖记录间关系，才考虑聚合。这个判断比任何参数调优都重要。

四、聚合粒度如何改变输出质量：机制与量化

4.1 三种质量衰减机制

聚合粒度变大，质量不会线性下降，而是通过三种机制非线性衰减。第一是注意力稀释：批次越大，单条记录在上下文中的"相对权重"越低，模型对细节的敏感度下降。第二是输出对齐困难：要求模型对B条记录逐条输出结构化结果时，B越大，漏条、错位、合并的概率越高。第三是错误传播：一条记录触发模型"跑题"，可能污染整批输出。

这三类机制在学术上都有对应研究。关于结构化输出的可靠性，OpenAI在其结构化输出（Structured Outputs）文档中强调，使用JSON Schema约束能显著降低格式错误率，但并未承诺"条数无关的稳定性"。本文评述：Schema约束解决的是"格式对不对"，不解决"条数全不全"，后者必须靠分批大小和校验来兜底。这是很多团队踩坑的地方——以为开了JSON模式就万事大吉。

4.2 一个可复现的质量-粒度曲线

要回答"聚合粒度变了输出质量怎么变"，不能靠感觉，要设计一个可复现的评估。笔者的建议是构造一个"金标准子集"：从图层中人工标注200—300条，覆盖典型与边界情况，然后在不同批大小下跑同一提示词，统计三个指标——逐条召回率（输出条数/输入条数）、字段准确率（与金标准一致的比例）、格式合规率（可被JSON解析的比例）。

注：上表为整合模拟数据，趋势参考长上下文与结构化输出的公开研究结论，具体数值随模型版本、提示词与任务类型变化，不可直接套用。

这张表最重要的信息不是具体数字，而是曲线的形状：成本随B增大快速下降然后趋平，质量随B增大缓慢下降然后加速崩塌。两条曲线的"剪刀差"区域就是最优工作区，通常在B=15—30之间（视任务而定）。本文评述：寻找这个区域的方法不是调参玄学，而是做一次小规模A/B扫描——固定提示词，只变B，跑金标准子集，画出你自己的两条曲线。这一步花半天，能省下几个月的账单焦虑。

4.3 聚合粒度对"跨记录推理"的反向影响

需要指出一个反向效应：对于依赖记录间关系的任务（如去重、聚类、一致性校验），聚合粒度太小反而有害。因为模型看不到足够多的邻居，无法判断"这两条是不是同一个实体"。这类任务的批大小下限由"关系半径"决定，而非成本。本文评述：成本优化必须服从任务语义，不能一刀切。工程上应把任务分为"独立型"与"关系型"两类，前者用成本最优批大小，后者用语义最优批大小，分别配置。

五、工程落地：可复用的分批器与提示词模板

5.1 在FME中实现动态分批的三种方式

方式一：纯转换器组合。用Counter生成序号，用Expression计算累计字符数，用AttributeCreator打上"批次号"（累计字符数整除阈值），再用Aggregator按批次号聚合。这种方式无需写代码，适合不想维护脚本的团队，但对"变长记录"的切分不够精确。

方式二：PythonCaller。在Python中实现"贪心装箱"：遍历feature，累加token估算值，超过预算就切批。这种方式最灵活，能精确控制每批token量，也方便接入tiktoken做token计数。缺点是需要在FME环境里管理Python依赖。

方式三：混合式。用转换器做粗切（按条数上限），用PythonCaller做细调（按token预算二次切分）。本文评述：对大多数团队，方式二是性价比最高的选择，因为token预算是成本治理的核心约束，而只有代码能精确执行这个约束。

5.2 PythonCaller分批器模板

import fme
import fmeobjects

class BatchSplitter(object):
    def __init__(self):
        self.buf = []
        self.buf_tokens = 0
        self.batch_id = 0
        self.max_tokens = 6000      # 每批token预算（含系统提示预留）
        self.max_items = 40         # 条数硬上限，防止短记录塞太多
        self.sys_reserve = 800      # 系统提示预留

    def _est_tokens(self, text):
        # 粗略估算：中文约1字1token，英文约4字符1token
        # 生产环境建议用 tiktoken 精确计数
        return max(1, len(text) // 2)

    def input(self, feature):
        text = feature.getAttribute('content') or ''
        t = self._est_tokens(text)
        if self.buf and (self.buf_tokens + t + self.sys_reserve > self.max_tokens
                         or len(self.buf) >= self.max_items):
            self._flush()
        self.buf.append(feature)
        self.buf_tokens += t

    def _flush(self):
        for f in self.buf:
            f.setAttribute('batch_id', self.batch_id)
        self.batch_id += 1
        self.buf = []
        self.buf_tokens = 0

    def close(self):
        if self.buf:
            self._flush()

    def process_group(self):
        pass

这段代码的关键设计是"双约束"：token预算与条数上限同时生效。只设token预算，短记录会被塞进过多条，触发4.1节说的输出对齐困难；只设条数上限，长记录会超窗。本文评述：双约束是动态分批的最小可行实现，缺一不可。生产环境还应加入"单条超长记录单独成批"的兜底逻辑，避免一条巨长记录卡死整批。

5.3 提示词模板：让模型"逐条对齐"

分批之后，提示词必须显式要求"逐条输出、条数一致、带id回填"。一个可用的模板骨架如下：

你是地址标准化助手。输入是一个JSON数组，每个元素含 id 与 raw。
要求：
1. 对每个元素独立处理，输出JSON数组，元素顺序与输入一致；
2. 每个输出元素必须包含 id 与 normalized 两个字段；
3. 不得合并、拆分或遗漏任何元素；
4. 无法处理的元素，normalized 填 null，并在 reason 中说明。
输入：{{batch_json}}
输出：

模板里"不得合并、拆分或遗漏"这句话不是客套，而是针对4.1节"输出对齐困难"的直接对策。配合JSON Schema约束，能把格式错误率压到很低。本文评述：提示词工程在批处理场景的核心任务，是把"批"重新拆回"条"——让模型在内部保持逐条视角，而不是把一批当成一个整体来概括。这是聚合粒度与输出质量之间的关键调节阀。

六、评估闭环：用FME自身做质量回归

成本治理不能只看账单，必须和质量绑定。一个可操作的闭环是：每次调整批大小或提示词，都用固定的金标准子集跑一遍，把召回率、准确率、合规率写入一个版本化表格，形成"配置-质量"的回归记录。FME本身就能承担这个评估流水线——用FeatureReader读金标准，用ChangeDetector比对输出，用StatisticsCalculator汇总指标。

关于评估方法学，HELM（Holistic Evaluation of Language Models，Liang等，2022）提出的多维度评估思路值得借鉴：不要只看单一准确率，要同时看鲁棒性、公平性与效率。本文评述：把"效率"（token/条）作为与准确率并列的一等指标，是FME场景下最实用的改造。因为在这个场景里，一个准确率99%但成本10倍的配置，和一个准确率95%但成本1/10的配置，后者往往才是业务要的。

另一个值得引入的概念是"LLM-as-a-judge"（Zheng等，2023，MT-Bench/Chatbot Arena相关工作）。当金标准标注成本高时，可以用一个更强的模型对输出做批量打分，作为粗筛。但要注意其偏差——评委模型可能偏好与自己风格相似的输出。本文评述：LLM评委适合做"趋势监控"，不适合做"最终验收"，最终验收仍应回到人工标注的小样本上。

七、监控、告警与成本护栏

再好的分批策略，也需要护栏防止"某天有人把批大小改成1"这类事故。建议在三个层面设护栏。第一层是工作流层：在Connector前加一个"预估token统计"分支，把每批token数写入日志，超过阈值直接终止。第二层是平台层：利用OpenAI控制台的用量限额（Usage Limits）设置硬上限，超过即拒绝请求。第三层是组织层：为不同项目分配独立API key，按key统计成本，避免互相污染。

关于可观测性，OpenTelemetry的GenAI语义约定（Semantic Conventions for Generative AI）已经定义了LLM调用的标准属性，包括模型名、输入输出token数、请求耗时等。本文评述：把FME的日志接入这套标准，能让成本数据与APM系统打通，比自建一套私有指标更可持续。对于用FME Server/Cloud的团队，还可以结合其作业日志做按工作流的成本归集。

一个实用的经验阈值：当单次工作流的预估成本超过团队日预算的5%时，应强制走"分批+金标准抽检"流程；当单条记录的平均token超过500时，应检查是否有字段被无意拼进了提示词。

八、前沿预判：从"批处理"到"语义路由"

当前的"聚合-分批-校验"仍是静态策略：批大小由人设定，提示词由人写死。前沿方向是"语义路由"——根据每条记录的难度与类型，动态决定它走哪条处理路径。简单记录走小模型或规则，复杂记录走大模型，模糊记录走"大模型+人工复核"。这在学术上对应"LLM路由"（routing）研究，如RouteLLM（Ong等，2024）提出的用学习到的路由器在强弱模型间分配请求，在保持质量的同时显著降本。

本文评述：FME场景天然适合路由，因为空间数据本身带有强烈的"难度信号"——地址完整度、字段缺失率、几何复杂度都可以作为路由特征。把"批大小"升级为"路由策略"，是成本治理的下一个台阶。但要注意，路由会引入新的评估复杂度：你不仅要评估每条路径的质量，还要评估路由本身的准确率。这是一个"用复杂度换成本"的交易，是否划算取决于业务对一致性的要求。

另一个值得关注的方向是"提示词缓存"（prompt caching）。部分模型厂商已推出对重复前缀的缓存计费优惠，这对"系统提示固定、内容变化"的FME场景是直接利好。本文评述：缓存把2.1节公式里的 T_sys 单价打下来，等于变相放大了聚合的收益，但缓存命中依赖前缀完全一致，提示词模板的稳定性因此变成一项成本资产。

九、结论与操作清单

回到最初的问题：该在Sampler之后聚合，还是该设动态分批阈值？答案是——先判断任务类型，再选路线；绝大多数独立型任务应选动态分批，聚合只留给关系型任务。聚合粒度对质量的影响不是线性的，而是"缓慢下降后加速崩塌"，最优工作区在成本曲线趋平、质量曲线尚未陡降的剪刀差区域。

给出一份可直接执行的操作清单：

1. 统计当前工作流的 N、T_sys、T_item、T_out，代入2.1节公式，算出理论成本与实际账单的差距，定位浪费来源。
2. 把任务分为"独立型"与"关系型"，分别设定批大小策略。
3. 用PythonCaller实现双约束分批器（token预算+条数上限），单条超长记录单独成批。
4. 提示词显式要求"逐条输出、条数一致、id回填"，配合JSON Schema。
5. 构造200—300条金标准子集，扫描B=1/10/25/50/100，画出自己的成本-质量曲线。
6. 把评估结果版本化，每次改配置都回归一次。
7. 设置三层成本护栏：工作流预估、平台限额、按key归集。
8. 关注prompt caching与路由策略，作为下一阶段优化方向。

拓展阅读方面，Safe Software官方社区有大量Connector实践讨论，OpenAI的Cookbook提供了结构化输出与批处理的示例代码，LangChain与LlamaIndex的文档中也有关于批处理与路由的工程模式。建议读者结合自己的数据先做一次小规模扫描，再决定批大小，而不是照搬任何文章里的数字——包括本文的模拟数据。

主要参考文献