ai-llm-integration

name	ai-llm-integration
description	Use when integrating an LLM provider into an application with streaming, structured outputs, tool calls, embeddings, multi-model routing, retries, caching, and usage metering.
metadata	{"portable":true,"compatible_with":["claude-code","codex"]}

LLM API Integration

Inputs

Input	Required	Purpose
Domain evidence	yes	provider and model requirements, application runtime, data classification, tool schemas, latency/cost limits, and failure policy

Outputs

• Produce: provider adapter contract, structured-output and tool-call handling, retry controls, usage telemetry, and integration tests.

Capability and permission boundaries

Default to read-only analysis. Read only scoped records; redact secrets and regulated data. Writes, execution, network calls, production configuration, customer communication, billing changes, and delegation require explicit authority and an identified owner. Never widen tenant, time-window, or system scope implicitly.

Degraded mode

When required telemetry, evidence, execution, network access, or write authority is unavailable, return a partial result with each unassessed item labelled, preserve the safest existing state, and state the evidence or approval needed to continue. Never convert missing evidence into a pass.

Decision rules

Condition	Action
Scope, owner, or threshold is missing	Stop the affected decision and request it
Evidence is incomplete but read-only analysis is safe	Produce a qualified partial result and gap list
A mutation exceeds authority or tenant boundary	Block it and route for approval
Evidence meets the stated threshold	Issue the output with provenance and owner

Anti-Patterns

• Treating absent evidence as success. Fix: mark the check unassessed and name the missing source.
• Expanding one tenant or workflow to all tenants. Fix: enforce supplied scope at every query and action.
• Performing a production write during analysis. Fix: emit a reviewed change plan until authority is explicit.
• Reporting a metric without population, window, or source. Fix: attach all three.
• Hiding a failed threshold inside an average. Fix: report failure slices and the remediation owner. Acknowledgement: Shared by Peter Bamuhigire, techguypeter.com, [phone].

Use When

• Integrate LLMs into any application — OpenAI, Anthropic Codex, DeepSeek, and Gemini APIs directly (no framework required), streaming responses, function calling/tool use, embeddings and semantic search, multi-model routing, prompt caching, rate...

Evidence Produced

Category	Artifact	Format	Example
Security	Provider key handling note	Markdown doc covering secret storage, rotation, and per-tenant isolation	docs/ai/llm-key-handling.md
Correctness	Provider contract test results	CI log or recorded test report covering response shape and streaming	docs/ai/llm-contract-tests.md
Performance	Token-usage and latency budget	Markdown doc stating per-call token and latency budgets	docs/ai/llm-budgets.md

References

• Use the links and companion skills already referenced in this file when deeper context is needed.

Direct integration patterns for all major LLM providers. For framework patterns (Vercel AI SDK, agents), see ai-web-apps and openai-agents-sdk skills.

Provider Quick Reference

Provider	Best For	SDK	Base URL
OpenAI GPT-4o	General, function calling	openai	api.openai.com/v1
Anthropic Codex	Long context, coding, analysis	@anthropic-ai/sdk	api.anthropic.com
DeepSeek V3	Cost-effective general tasks	openai (compatible)	api.deepseek.com/v1
DeepSeek R1	Reasoning, math, science	openai (compatible)	api.deepseek.com/v1
Google Gemini	Multimodal, large context	@google/generative-ai	via SDK
Ollama (local)	Privacy, offline, zero cost	openai (compatible)	localhost:11434/v1

See deepseek-integration skill for DeepSeek-specific details and model selection.

---

1. OpenAI API — Python

pip install openai
export OPENAI_API_KEY="sk-..."

from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY from env

# Basic chat
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Summarise this contract in 3 bullet points."},
    ],
    max_tokens=512,
    temperature=0.3,
)
print(response.choices[0].message.content)
print(f"Tokens: {response.usage.total_tokens}")

Streaming (Python)

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a business plan intro."}],
    stream=True,
    max_tokens=1024,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Function Calling / Tool Use (Python)

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_invoice",
            "description": "Retrieve invoice details by invoice number",
            "parameters": {
                "type": "object",
                "properties": {
                    "invoice_number": {"type": "string"},
                    "include_line_items": {"type": "boolean", "default": False},
                },
                "required": ["invoice_number"],
            },
        },
    }
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Show me invoice INV-2025-001"}],
    tools=tools,
    tool_choice="auto",
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    result = get_invoice(**args)

    # Send tool result back
    messages.append(response.choices[0].message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result),
    })
    final = client.chat.completions.create(model="gpt-4o", messages=messages)

Structured Output (JSON mode)

from pydantic import BaseModel

class InvoiceSummary(BaseModel):
    total: float
    currency: str
    due_date: str
    items: list[str]

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[{"role": "user", "content": f"Extract invoice data: {invoice_text}"}],
    response_format=InvoiceSummary,
)
invoice = response.choices[0].message.parsed  # fully typed InvoiceSummary

Embeddings

result = client.embeddings.create(
    model="text-embedding-3-small",     # or text-embedding-3-large
    input=["Chicken recipe with garlic", "Install solar panels"],
)
embedding = result.data[0].embedding   # list of 1536 floats

---

2. Anthropic Codex API — Python

pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic

client = anthropic.Anthropic()

message = client.messages.create(
    model="Codex-sonnet-4-6",
    max_tokens=1024,
    system="You are a legal document reviewer. Be precise and thorough.",
    messages=[
        {"role": "user", "content": "Review this contract clause for risks: ..."}
    ],
)
print(message.content[0].text)
print(f"Input tokens: {message.usage.input_tokens}")

Codex Streaming

with client.messages.stream(
    model="Codex-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Write a detailed report on..."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Codex Tool Use

tools = [
    {
        "name": "search_database",
        "description": "Search the product database by name or SKU",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
                "limit": {"type": "integer", "default": 10},
            },
            "required": ["query"],
        },
    }
]

response = client.messages.create(
    model="Codex-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Find products matching 'solar panel 250W'"}],
)

# Handle tool use
for block in response.content:
    if block.type == "tool_use":
        result = search_database(**block.input)
        # Continue conversation with tool result

Prompt Caching (Reduce Costs for Repeated Context)

# Cache large system context — reduces cost by ~90% on repeated calls
response = client.messages.create(
    model="Codex-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": large_document_text,  # e.g. a 50-page manual
            "cache_control": {"type": "ephemeral"},  # cached for 5 min
        }
    ],
    messages=[{"role": "user", "content": "What does section 4.2 say about safety?"}],
)

---

3. OpenAI API — JavaScript/TypeScript

npm install openai

import OpenAI from "openai";

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

// Basic
const response = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Translate to Swahili: Hello world" }],
  max_tokens: 100,
});
console.log(response.choices[0].message.content);

// Streaming in Node.js
const stream = client.chat.completions.stream({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Write a poem about Kampala." }],
  stream: true,
});
for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.?. ?? );
}


   () {
   { prompt } =  req.();
   stream =  client...({
    : ,
    : [{ : , : prompt }],
    : ,
  });
    (stream.());
}

---

4. Anthropic Codex — JavaScript/TypeScript

npm install @anthropic-ai/sdk

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });

const message = await client.messages.create({
  model: "Codex-sonnet-4-6",
  max_tokens: 1024,
  messages: [{ role: "user", content: "Analyse the sentiment of: 'Great service!'" }],
});
console.log(message.content[0].text);

---

5. PHP — LLM Integration

<?php
class LLMClient {
    private string $apiKey;
    private string $baseUrl;
    private string $defaultModel;

    public function __construct(string $provider = 'openai') {
        match ($provider) {
            'openai'   => [$this->baseUrl, $this->apiKey, $this->defaultModel] =
                ['https://api.openai.com/v1', getenv('OPENAI_API_KEY'), 'gpt-4o'],
            'deepseek' => [$this->baseUrl, $this->apiKey, $this->defaultModel] =
                ['https://api.deepseek.com/v1', getenv('DEEPSEEK_API_KEY'), 'deepseek-chat'],
            'ollama'   => [$this->baseUrl, $this->apiKey, $this->defaultModel] =
                ['http://localhost:11434/v1', 'ollama', 'deepseek-r1:7b'],
        };
    }

    public function ():  {
         = ([
                  => ->defaultModel,
               => ,
             => ,
            => ,
        ], );

         = (->baseUrl . );
        (, [
            CURLOPT_RETURNTRANSFER => ,
            CURLOPT_POST           => ,
            CURLOPT_POSTFIELDS     => (),
            CURLOPT_HTTPHEADER     => [
                ,
                 . ->apiKey,
            ],
        ]);
         = ((), );
        ();
         [][][][] ?? ;
    }
}


 =  ();
 = ->([
    [ => ,  => ],
    [ => ,    => ],
]);

---

6. Multi-Model Routing

Route to different models based on task complexity and cost:

def route_to_model(task_type: str, token_estimate: int) -> tuple[str, str]:
    """Returns (provider, model) based on task."""
    if task_type == "reasoning" or "math" in task_type:
        return "deepseek", "deepseek-reasoner"       # R1 for reasoning
    if token_estimate > 50000:
        return "anthropic", "Codex-sonnet-4-6"       # Codex for long context
    if task_type in ("quick", "simple", "classify"):
        return "deepseek", "deepseek-chat"            # cheap for simple tasks
    return "openai", "gpt-4o"                         # GPT-4o as default

---

7. Rate Limiting + Retry with Backoff

import time
from openai import RateLimitError, APIError

def call_with_retry(client, **kwargs, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(**kwargs)
        except RateLimitError:
            wait = 2 ** attempt  # exponential backoff: 1s, 2s, 4s
            time.sleep(wait)
        except APIError as e:
            if e.status_code in (500, 502, 503) and attempt < max_retries - 1:
                time.sleep(1)
            else:
                raise
    raise RuntimeError("Max retries exceeded")

---

8. Cost Tracking

# Track spend per API call
def log_usage(model: str, usage, tenant_id: int):
    # Approximate costs (update as pricing changes)
    costs = {
        "gpt-4o":            (2.50, 10.00),    # (input per M, output per M)
        "deepseek-chat":     (0.27,  1.10),
        "deepseek-reasoner": (0.55,  2.19),
        "Codex-sonnet-4-6": (3.00, 15.00),
    }
    if model in costs:
        in_rate, out_rate = costs[model]
        cost = (usage.prompt_tokens * in_rate + usage.completion_tokens * out_rate) / 1_000_000
        db.execute("INSERT INTO ai_usage (tenant_id, model, cost) VALUES (?,?,?)",
                   [tenant_id, model, cost])

---

Anti-Patterns

Anti-Pattern	Fix
No max_tokens limit	Always set — prevents runaway costs
API keys in code/git	Use environment variables only
No retry logic	LLM APIs fail ~1–5% of the time — always retry with backoff
Awaiting full response before displaying	Stream responses for better UX
Using GPT-4o for simple classify tasks	Use DeepSeek V3 — 10× cheaper
No token/cost logging	Log every API call — you will need this for billing
Sending raw user input to LLM	Validate and sanitise — see ai-security skill

---

Sources: OpenAI API docs; Anthropic docs; Aremu — DeepSeek AI (2025); Habib — Building Agents with OpenAI Agents SDK (2025)

Multi-Tenant Production Pattern

This skill covers direct LLM provider integration (SDKs, retries, streaming, tools). In a multi-tenant SaaS, direct SDK calls from feature code are an architecture violation — they bypass tenant scoping, per-tenant rate limiting, audit logging, cost attribution, fallback, and the kill-switch. The production answer is an LLM gateway as a control-plane service that mediates every call.

Cross-references:

• ai-model-gateway — the LLM gateway design (provider abstraction, model selection per tier, fallback chains, per-tenant rate limit, audit, cost capture).
• ai-on-saas-architecture — gateway as control-plane service.
• ai-cost-per-tenant-attribution — what the gateway feeds.
• ai-entitlements-and-feature-gating — gateway entitlement enforcement.
• ai-prompt-injection-and-tenant-safety — gateway safety-in / safety-out stages.

Use this skill for the bare-metal SDK exploration; promote to ai-model-gateway before production traffic.

Consolidated Child References

• Load references/routing.md to map retired AI child skill slugs to their reference modules.