LLM Strategy

pnpm add @polar-sh/ingestion ai @ai-sdk/openai

import { Ingestion } from "@polar-sh/ingestion";
import { LLMStrategy } from "@polar-sh/ingestion/strategies/LLM";
import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";

// Setup the LLM Ingestion Strategy
const llmIngestion = Ingestion({ accessToken: process.env.POLAR_ACCESS_TOKEN })
  .strategy(new LLMStrategy(openai("gpt-4o")))
  .cost((ctx) => ({ amount: 123, currency: "usd" })) // Optional: Set the cost of the LLM usage
  .ingest("openai-usage");

export async function POST(req: Request) {
  const { prompt }: { prompt: string } = await req.json();

  // Get the wrapped LLM model with ingestion capabilities
  // Pass Customer Id to properly annotate the ingestion events with a specific customer
  const model = llmIngestion.client({
    customerId: "xxx",
  });

  const { text } = await generateText({
    model,
    system: "You are a helpful assistant.",
    prompt,
  });

  return Response.json({ text });
}

{
  "customerId": "123",
  "name": "openai-usage",
  "metadata": {
    "inputTokens": 100,
    "outputTokens": 200,
    "cachedInputTokens": 10,
    "totalTokens": 300,
    "model": "gpt-4o",
    "provider": "openai.responses",
    "strategy": "LLM",
    "_cost": {
      "amount": 123, // Amount is expected to be in cents. $1.23 should be represented as 123
      "currency": "usd"
    },
    "_llm": {
      ... //
    }
  }
}

import os
from polar_sdk.ingestion import Ingestion

ingestion = Ingestion(os.getenv("POLAR_ACCESS_TOKEN"))

ingestion.ingest({
    "name": "my-event",
    "external_customer_id": "CUSTOMER_ID",
    "metadata": {
        "usage": 13.37,
    }
})

import os
from polar_sdk.ingestion import Ingestion
from polar_sdk.ingestion.strategies import PydanticAIStrategy
from pydantic import BaseModel
from pydantic_ai import Agent


ingestion = Ingestion(os.getenv("POLAR_ACCESS_TOKEN"))
strategy = ingestion.strategy(PydanticAIStrategy, "ai_usage")


class MyModel(BaseModel):
    city: str
    country: str


agent = Agent("gpt-4.1-nano", output_type=MyModel)

if __name__ == '__main__':
    result = agent.run_sync("The windy city in the US of A.")
    print(result.output)
    strategy.ingest("CUSTOMER_ID", result)

{
  "name": "ai_usage",
  "external_customer_id": "CUSTOMER_ID",
  "metadata": {
    "requests": 1,
    "total_tokens": 78,
    "request_tokens": 58,
    "response_tokens": 20
  }
}