AI Integration

AI Integration

Enterprise AI/ML model integration patterns for vision, audio, embeddings, and RAG systems. This skill covers API integration, prompt engineering, and production deployment.

Purpose

Integrate AI capabilities into applications effectively:

• Implement vision and image understanding
• Add audio transcription and processing
• Build semantic search with embeddings
• Create RAG (Retrieval Augmented Generation) systems
• Handle rate limiting and error recovery
• Optimize costs and latency

Features

1. Vision API Integration

import Anthropic from '@anthropic-ai/sdk';
import OpenAI from 'openai';

const anthropic = new Anthropic();
const openai = new OpenAI();

// Analyze image with Claude
async function analyzeImageWithClaude(
  imageUrl: string | Buffer,
  prompt: string
): Promise<string> {
  const imageSource = typeof imageUrl === 'string'
    ? { type: 'url' as const, url: imageUrl }
    : {
        type: 'base64' as const,
        media_type: 'image/jpeg' as const,
        data: imageUrl.toString('base64'),
      };

  const response = await anthropic.messages.create({
    model: 'claude-sonnet-4-20250514',
    max_tokens: 1024,
    messages: [{
      role: 'user',
      content: [
        {
          type: 'image',
          source: imageSource,
        },
        {
          type: 'text',
          text: prompt,
        },
      ],
    }],
  });

  return response.content[0].type === 'text'
    ? response.content[0].text
    : '';
}

// Extract structured data from image
interface ProductInfo {
  name: string;
  description: string;
  price?: string;
  category?: string;
  features: string[];
}

async function extractProductFromImage(imageBuffer: Buffer): Promise<ProductInfo> {
  const prompt = `Analyze this product image and extract:
1. Product name
2. Description (2-3 sentences)
3. Price (if visible)
4. Category
5. Key features (list)

Return as JSON only, no explanation.`;

  const response = await analyzeImageWithClaude(imageBuffer, prompt);

  try {
    return JSON.parse(response);
  } catch {
    throw new Error('Failed to parse product information');
  }
}

// OCR with GPT-4 Vision
async function extractTextFromImage(imageUrl: string): Promise<string> {
  const response = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [{
      role: 'user',
      content: [
        {
          type: 'image_url',
          image_url: { url: imageUrl, detail: 'high' },
        },
        {
          type: 'text',
          text: 'Extract all text from this image. Preserve the original formatting and structure as much as possible.',
        },
      ],
    }],
    max_tokens: 4096,
  });

  return response.choices[0].message.content || '';
}

// Batch image processing
async function batchAnalyzeImages(
  images: Array<{ id: string; url: string }>,
  prompt: string,
  concurrency: number = 3
): Promise<Map<string, string>> {
  const results = new Map<string, string>();
  const queue = new PQueue({ concurrency });

  await Promise.all(
    images.map(image =>
      queue.add(async () => {
        try {
          const result = await analyzeImageWithClaude(image.url, prompt);
          results.set(image.id, result);
        } catch (error) {
          results.set(image.id, `Error: ${error.message}`);
        }
      })
    )
  );

  return results;
}

2. Audio Processing

import { Readable } from 'stream';

// Transcribe audio with Whisper
async function transcribeAudio(
  audioFile: Buffer | string,
  options: {
    language?: string;
    prompt?: string;
    responseFormat?: 'json' | 'text' | 'srt' | 'vtt';
    timestamps?: boolean;
  } = {}
): Promise<TranscriptionResult> {
  const {
    language,
    prompt,
    responseFormat = 'json',
    timestamps = false,
  } = options;

  const file = typeof audioFile === 'string'
    ? fs.createReadStream(audioFile)
    : Readable.from(audioFile);

  const response = await openai.audio.transcriptions.create({
    file,
    model: 'whisper-1',
    language,
    prompt,
    response_format: timestamps ? 'verbose_json' : responseFormat,
  });

  if (timestamps && typeof response !== 'string') {
    return {
      text: response.text,
      segments: response.segments?.map(seg => ({
        start: seg.start,
        end: seg.end,
        text: seg.text,
      })),
      language: response.language,
    };
  }

  return { text: typeof response === 'string' ? response : response.text };
}

// Real-time transcription with streaming
async function* streamTranscription(
  audioStream: ReadableStream
): AsyncGenerator<string> {
  // For real-time, use Deepgram or AssemblyAI
  const deepgram = new Deepgram(process.env.DEEPGRAM_API_KEY!);

  const connection = await deepgram.transcription.live({
    model: 'nova-2',
    language: 'en',
    smart_format: true,
    interim_results: true,
  });

  connection.on('transcriptReceived', (message) => {
    const transcript = message.channel?.alternatives?.[0]?.transcript;
    if (transcript) {
      yield transcript;
    }
  });

  // Pipe audio to connection
  const reader = audioStream.getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    connection.send(value);
  }

  connection.close();
}

// Generate speech from text
async function generateSpeech(
  text: string,
  options: {
    voice?: 'alloy' | 'echo' | 'fable' | 'onyx' | 'nova' | 'shimmer';
    model?: 'tts-1' | 'tts-1-hd';
    speed?: number;
  } = {}
): Promise<Buffer> {
  const { voice = 'alloy', model = 'tts-1', speed = 1 } = options;

  const response = await openai.audio.speech.create({
    model,
    voice,
    input: text,
    speed,
  });

  return Buffer.from(await response.arrayBuffer());
}

3. Embeddings & Vector Search

import { Pinecone } from '@pinecone-database/pinecone';

const pinecone = new Pinecone();

// Generate embeddings
async function generateEmbeddings(
  texts: string[],
  model: string = 'text-embedding-3-small'
): Promise<number[][]> {
  const response = await openai.embeddings.create({
    model,
    input: texts,
  });

  return response.data.map(d => d.embedding);
}

// Index documents
interface Document {
  id: string;
  content: string;
  metadata?: Record<string, any>;
}

async function indexDocuments(
  documents: Document[],
  indexName: string,
  namespace: string = 'default'
): Promise<void> {
  const index = pinecone.index(indexName);

  // Process in batches
  const batchSize = 100;
  for (let i = 0; i < documents.length; i += batchSize) {
    const batch = documents.slice(i, i + batchSize);

    const embeddings = await generateEmbeddings(
      batch.map(d => d.content)
    );

    const vectors = batch.map((doc, j) => ({
      id: doc.id,
      values: embeddings[j],
      metadata: {
        content: doc.content.substring(0, 1000), // Store truncated content
        ...doc.metadata,
      },
    }));

    await index.namespace(namespace).upsert(vectors);
  }
}

// Semantic search
interface SearchResult {
  id: string;
  score: number;
  content: string;
  metadata?: Record<string, any>;
}

async function semanticSearch(
  query: string,
  indexName: string,
  options: {
    namespace?: string;
    topK?: number;
    filter?: Record<string, any>;
    minScore?: number;
  } = {}
): Promise<SearchResult[]> {
  const {
    namespace = 'default',
    topK = 10,
    filter,
    minScore = 0.7,
  } = options;

  const [queryEmbedding] = await generateEmbeddings([query]);

  const index = pinecone.index(indexName);
  const results = await index.namespace(namespace).query({
    vector: queryEmbedding,
    topK,
    filter,
    includeMetadata: true,
  });

  return results.matches
    ?.filter(m => m.score && m.score >= minScore)
    .map(match => ({
      id: match.id,
      score: match.score || 0,
      content: match.metadata?.content as string || '',
      metadata: match.metadata,
    })) || [];
}

4. RAG Implementation

interface RAGConfig {
  indexName: string;
  namespace?: string;
  topK?: number;
  model?: string;
  systemPrompt?: string;
}

class RAGSystem {
  private config: RAGConfig;

  constructor(config: RAGConfig) {
    this.config = {
      namespace: 'default',
      topK: 5,
      model: 'claude-sonnet-4-20250514',
      systemPrompt: 'You are a helpful assistant. Answer based on the provided context.',
      ...config,
    };
  }

  async query(question: string): Promise<RAGResponse> {
    // Step 1: Retrieve relevant documents
    const context = await semanticSearch(question, this.config.indexName, {
      namespace: this.config.namespace,
      topK: this.config.topK,
    });

    if (context.length === 0) {
      return {
        answer: "I couldn't find relevant information to answer your question.",
        sources: [],
        confidence: 0,
      };
    }

    // Step 2: Build context string
    const contextText = context
      .map((doc, i) => `[${i + 1}] ${doc.content}`)
      .join('\n\n');

    // Step 3: Generate answer
    const response = await anthropic.messages.create({
      model: this.config.model!,
      max_tokens: 2048,
      system: `${this.config.systemPrompt}

Use the following context to answer the user's question. If the answer is not in the context, say so.

Context:
${contextText}`,
      messages: [{
        role: 'user',
        content: question,
      }],
    });

    const answer = response.content[0].type === 'text'
      ? response.content[0].text
      : '';

    return {
      answer,
      sources: context.map(c => ({
        id: c.id,
        content: c.content,
        score: c.score,
      })),
      confidence: Math.max(...context.map(c => c.score)),
    };
  }

  // Hybrid search (keyword + semantic)
  async hybridQuery(
    question: string,
    keywords?: string[]
  ): Promise<RAGResponse> {
    // Semantic search
    const semanticResults = await semanticSearch(question, this.config.indexName, {
      namespace: this.config.namespace,
      topK: this.config.topK! * 2,
    });

    // Keyword filter (if provided)
    let results = semanticResults;
    if (keywords && keywords.length > 0) {
      results = semanticResults.filter(r =>
        keywords.some(k =>
          r.content.toLowerCase().includes(k.toLowerCase())
        )
      );
    }

    // Rerank and take top K
    const topResults = results.slice(0, this.config.topK);

    // Generate answer using top results
    return this.generateAnswer(question, topResults);
  }

  private async generateAnswer(
    question: string,
    context: SearchResult[]
  ): Promise<RAGResponse> {
    // ... same generation logic
  }
}

// Usage
const rag = new RAGSystem({
  indexName: 'knowledge-base',
  topK: 5,
  systemPrompt: 'You are a customer support agent. Be helpful and concise.',
});

const response = await rag.query('How do I reset my password?');

5. Structured Output

import { z } from 'zod';
import { zodResponseFormat } from 'openai/helpers/zod';

// Define schema
const SentimentSchema = z.object({
  sentiment: z.enum(['positive', 'negative', 'neutral']),
  confidence: z.number().min(0).max(1),
  topics: z.array(z.string()),
  summary: z.string(),
});

type SentimentAnalysis = z.infer<typeof SentimentSchema>;

// Get structured output
async function analyzeSentiment(text: string): Promise<SentimentAnalysis> {
  const response = await openai.beta.chat.completions.parse({
    model: 'gpt-4o',
    messages: [{
      role: 'system',
      content: 'Analyze the sentiment of the provided text.',
    }, {
      role: 'user',
      content: text,
    }],
    response_format: zodResponseFormat(SentimentSchema, 'sentiment_analysis'),
  });

  return response.choices[0].message.parsed!;
}

// Claude tool use for structured output
async function extractEntities(text: string): Promise<{
  people: string[];
  organizations: string[];
  locations: string[];
  dates: string[];
}> {
  const response = await anthropic.messages.create({
    model: 'claude-sonnet-4-20250514',
    max_tokens: 1024,
    tools: [{
      name: 'extract_entities',
      description: 'Extract named entities from text',
      input_schema: {
        type: 'object',
        properties: {
          people: {
            type: 'array',
            items: { type: 'string' },
            description: 'Names of people mentioned',
          },
          organizations: {
            type: 'array',
            items: { type: 'string' },
            description: 'Organization names',
          },
          locations: {
            type: 'array',
            items: { type: 'string' },
            description: 'Location names',
          },
          dates: {
            type: 'array',
            items: { type: 'string' },
            description: 'Dates mentioned',
          },
        },
        required: ['people', 'organizations', 'locations', 'dates'],
      },
    }],
    tool_choice: { type: 'tool', name: 'extract_entities' },
    messages: [{
      role: 'user',
      content: `Extract entities from: ${text}`,
    }],
  });

  const toolUse = response.content.find(c => c.type === 'tool_use');
  return toolUse?.input as any;
}

6. Production Patterns

// Rate limiting and retry
import Bottleneck from 'bottleneck';

const limiter = new Bottleneck({
  reservoir: 100, // Initial tokens
  reservoirRefreshAmount: 100,
  reservoirRefreshInterval: 60 * 1000, // Per minute
  maxConcurrent: 10,
});

async function withRateLimit<T>(fn: () => Promise<T>): Promise<T> {
  return limiter.schedule(fn);
}

// Retry with exponential backoff
async function withRetry<T>(
  fn: () => Promise<T>,
  maxRetries: number = 3,
  baseDelay: number = 1000
): Promise<T> {
  let lastError: Error;

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error as Error;

      // Check if retryable
      if (error.status === 429 || error.status >= 500) {
        const delay = baseDelay * Math.pow(2, attempt);
        await new Promise(r => setTimeout(r, delay));
        continue;
      }

      throw error;
    }
  }

  throw lastError!;
}

// Caching layer
const cache = new Map<string, { data: any; expires: number }>();

async function cachedEmbedding(
  text: string,
  ttl: number = 3600000 // 1 hour
): Promise<number[]> {
  const key = `embedding:${hashString(text)}`;
  const cached = cache.get(key);

  if (cached && cached.expires > Date.now()) {
    return cached.data;
  }

  const [embedding] = await generateEmbeddings([text]);
  cache.set(key, { data: embedding, expires: Date.now() + ttl });

  return embedding;
}

// Cost tracking
class CostTracker {
  private costs: Map<string, number> = new Map();

  track(model: string, inputTokens: number, outputTokens: number): void {
    const pricing = MODEL_PRICING[model] || { input: 0, output: 0 };
    const cost = (inputTokens * pricing.input + outputTokens * pricing.output) / 1000;

    const current = this.costs.get(model) || 0;
    this.costs.set(model, current + cost);
  }

  getReport(): Record<string, number> {
    return Object.fromEntries(this.costs);
  }

  getTotalCost(): number {
    return Array.from(this.costs.values()).reduce((a, b) => a + b, 0);
  }
}

Use Cases

1. Document Q&A System

// Build document Q&A
async function buildDocumentQA(documents: string[]): Promise<RAGSystem> {
  // Chunk documents
  const chunks = documents.flatMap((doc, docIndex) =>
    chunkText(doc, 500, 50).map((chunk, chunkIndex) => ({
      id: `doc-${docIndex}-chunk-${chunkIndex}`,
      content: chunk,
      metadata: { documentIndex: docIndex },
    }))
  );

  // Index chunks
  await indexDocuments(chunks, 'document-qa');

  // Return RAG system
  return new RAGSystem({
    indexName: 'document-qa',
    topK: 5,
    systemPrompt: 'Answer questions based on the provided documents.',
  });
}

2. Content Moderation

// Moderate content with AI
async function moderateContent(content: string): Promise<ModerationResult> {
  const response = await openai.moderations.create({ input: content });
  const result = response.results[0];

  return {
    flagged: result.flagged,
    categories: Object.entries(result.categories)
      .filter(([_, flagged]) => flagged)
      .map(([category]) => category),
    scores: result.category_scores,
  };
}

Best Practices

Do's

• Implement rate limiting - Respect API limits
• Cache embeddings - Avoid redundant API calls
• Handle errors gracefully - Implement retry logic
• Monitor costs - Track token usage
• Use streaming - For better UX with long responses
• Chunk appropriately - Balance context vs. relevance

Don'ts

• Don't expose API keys in frontend code
• Don't skip input validation
• Don't ignore rate limit errors
• Don't cache sensitive data inappropriately
• Don't use overly large context windows
• Don't forget fallback strategies

Related Skills

• api-architecture - API design patterns
• caching-strategies - Caching for AI responses
• backend-development - Integration patterns

Reference Resources

• OpenAI API Reference
• Anthropic API Reference
• Pinecone Documentation
• LangChain Documentation

Converted and distributed by TomeVault — claim your Tome and manage your conversions.