@ creem io/摄入

用量摄取工具包 克里姆语 缓冲、安全, 以可插接的战略层 自动计量(LLM呼叫,流出,定时处决).

npm install @creem_io/ingestion

为什么不直接给API打电话?

你可以.. creem.events.ingestEvents() 工作时 此软件包存在于 发送保证您的请求路径需求 :

• 永远不要挡道,永远不要扔. enqueue() 内存和返回中的缓冲 马上 交付失败通过 onError 钩子,不是 用户请求中的例外.
• 通过建筑进行安全重试. 每个事件都会有 event_id 在队列中 时间; 重试重新发送相同的 ID 和 Creem 的 (store, event_id) 脱下 他们倒闭了——重审的批次永远不能双开.
• 击打. 当批量填充( 100, 服务器盖) 时或当一个 间隔, 每起事件不会有一个 HTTP 呼叫 .
• 倒毒驱出. 克里姆验证批量的全无。 当 服务器拒绝一个无效事件, 管道会驱逐该事件 (通过 onError并交付剩下的——一个糟糕的事件无法承受 99个好东西被打倒了.
• 服务器无意识. await flush() 处理器返回前( 或手) 给 waitUntil) (中文(简体) ). 闲置的冲浪计时器是 unref"d,所以它从来没有持有 节点进程打开 .

缓冲摄入

import { Ingestion } from "@creem_io/ingestion";

const ingestion = Ingestion({
  apiKey: process.env.CREEM_API_KEY!,
  onError: (error) => console.error("[creem]", error.code, error.message),
  onWarnings: (warnings) => console.warn("[creem]", warnings),
});

// In your request handler — synchronous, cannot fail the request:
ingestion.enqueue({
  name: "tokens_used",
  externalCustomerId: userId, // or { customerId: "cust_..." }
  properties: { tokens: 512, model: "gpt-4o" },
});

// Serverless teardown / graceful shutdown:
await ingestion.flush();

每次活动都给客户取名 一个 方式 : customerId (克里姆语) 编号)或 externalCustomerId (你的身份证,登记为 external_id 编辑 客户API.

直接摄入

当您需要回复( 并准备自行处理错误) :

const result = await ingestion.ingest([
  { name: "tokens_used", customerId: "cust_abc", properties: { tokens: 512 } },
]);
// { accepted: 1, eventIds: ["..."], warnings?: [...] }

无事件 eventId 在发送前先弄一个,所以重试 呼叫.. 或者通过一个 client 用 SDK 的后置配置 —— 重放 同一身份证,不能两次开票.

warnings 你告诉我吧,当一个事件被接受的时候 没有计数器可以听到事件名称, 只有存档的计数匹配, 或者 时间戳在最后的计费期内.

战略

策略将您已经使用的客户端包裹并自动计量 。 这个 链将策略绑定在一个事件名称上; client(customer) 返回新鲜 每个客户包裹客户(战略是无国籍的——一例服务) 同时期客户安全:

const metered = ingestion
  .strategy(new SomeStrategy(/* ... */))
  .cost((emission) => Number(emission.properties.tokens) * 0.000002)
  .ingest("llm-usage");

const client = metered.client({ externalCustomerId: userId });
// use `client` normally — every call enqueues a usage event

排放事件包含战略的原始测量值 properties附加时 sDK 拥有的保留密钥 : strategy (捕获机制)和 _cost (当 .cost() 解析器随附). 从键开始 _ 为: 为 SDK 写入值保留.

解析器在电话里运行 如果抛出, 事件就会被放弃 并通过 onError 作为 resolver_failed - 它从不扔进 打电话给你 计量.

维塞尔 AI SDK

一次一次包一个语言模型 generateText 编号 : streamText 电话是 计数器——代号计数器、模型和供应商,没有按呼号编码。 已导入 从 /ai-sdk 子路径,所以其类型仅在使用时加载(AI SDK 6+) 编号 : LanguageModelV3+ 键V4:

import { openai } from "@ai-sdk/openai";
import { generateText } from "ai";
import { AiSdkStrategy } from "@creem_io/ingestion/ai-sdk";

const metered = ingestion
  .strategy(new AiSdkStrategy(openai("gpt-4o")))
  .ingest("llm-usage");

const model = metered.client({ externalCustomerId: userId });
const result = await generateText({ model, prompt });
// → event: { input_tokens, output_tokens, cached_input_tokens,
//            total_tokens, model, vendor, _llm, strategy }

水流从水流中流出 finish 因为它经过.. 消耗一流就足够了 没有什么是缓冲或双读的 A 类 流出在其前已取消 finish 部分账单什么也没有.

定时执行

最便宜的策略:工作单位时间,计费期限. 仅 成功处决——投掷处决传播其错误 没有帐单 :

import { DeltaTimeStrategy } from "@creem_io/ingestion";

const metered = ingestion
  .strategy(new DeltaTimeStrategy())
  .ingest("compute-time");

const timed = metered.client({ customerId });
const output = await timed(() => renderVideo(input));
// → event: { delta_time_ms, strategy }

进一步的战略(原始 LLM 提供者客户端、流/字节、, OpenTeleometry) 基于相同的 IngestionStrategy 导出的基准类 在这里.

配置