Build an AI Prompt Chain Executor

import { Redis } from "ioredis";
import { pool } from "../db";
import { randomBytes } from "node:crypto";
const redis = new Redis(process.env.REDIS_URL!);

interface ChainStep { id: string; name: string; type: "llm" | "transform" | "condition" | "parallel"; config: any; dependsOn: string[]; }
interface ChainDefinition { name: string; steps: ChainStep[]; }
interface ChainExecution { id: string; chain: string; status: "running" | "completed" | "failed"; steps: StepResult[]; totalCost: number; totalTokens: number; totalLatency: number; startedAt: string; }
interface StepResult { stepId: string; name: string; status: "pending" | "running" | "completed" | "failed" | "skipped"; input: any; output: any; cost: number; tokens: number; latencyMs: number; error: string | null; }

// Execute a prompt chain
export async function executeChain(definition: ChainDefinition, input: any): Promise<ChainExecution> {
  const id = `chain-${randomBytes(8).toString("hex")}`;
  const execution: ChainExecution = {
    id, chain: definition.name, status: "running",
    steps: definition.steps.map((s) => ({ stepId: s.id, name: s.name, status: "pending", input: null, output: null, cost: 0, tokens: 0, latencyMs: 0, error: null })),
    totalCost: 0, totalTokens: 0, totalLatency: 0, startedAt: new Date().toISOString(),
  };

  const context: Record<string, any> = { input, ...input };

  // Execute steps in dependency order
  const completed = new Set<string>();
  let hasFailure = false;

  while (completed.size < definition.steps.length && !hasFailure) {
    // Find steps ready to execute (all dependencies completed)
    const ready = definition.steps.filter((s) => !completed.has(s.id) && s.dependsOn.every((d) => completed.has(d)));

    if (ready.length === 0) break;

    // Execute ready steps (parallel if multiple)
    await Promise.all(ready.map(async (step) => {
      const stepResult = execution.steps.find((s) => s.stepId === step.id)!;
      stepResult.status = "running";
      stepResult.input = resolveInput(step.config.input, context);

      const start = Date.now();
      try {
        switch (step.type) {
          case "llm": {
            const result = await executeLLMStep(step.config, stepResult.input);
            stepResult.output = result.output;
            stepResult.cost = result.cost;
            stepResult.tokens = result.tokens;
            break;
          }
          case "transform": {
            stepResult.output = executeTransform(step.config.transform, stepResult.input);
            break;
          }
          case "condition": {
            const condResult = evaluateCondition(step.config.condition, context);
            stepResult.output = condResult;
            if (!condResult) {
              // Skip dependent steps
              const toSkip = definition.steps.filter((s) => s.dependsOn.includes(step.id));
              for (const skip of toSkip) {
                const skipResult = execution.steps.find((s) => s.stepId === skip.id)!;
                skipResult.status = "skipped";
                completed.add(skip.id);
              }
            }
            break;
          }
          case "parallel": {
            // Execute sub-steps in parallel
            const subResults = await Promise.all(
              (step.config.steps || []).map((sub: any) => executeLLMStep(sub, stepResult.input))
            );
            stepResult.output = subResults.map((r: any) => r.output);
            stepResult.cost = subResults.reduce((s: number, r: any) => s + r.cost, 0);
            stepResult.tokens = subResults.reduce((s: number, r: any) => s + r.tokens, 0);
            break;
          }
        }

        stepResult.status = "completed";
        stepResult.latencyMs = Date.now() - start;
        context[step.id] = stepResult.output;
      } catch (error: any) {
        stepResult.status = "failed";
        stepResult.error = error.message;
        stepResult.latencyMs = Date.now() - start;

        if (step.config.required !== false) hasFailure = true;
      }

      completed.add(step.id);
    }));
  }

  execution.status = hasFailure ? "failed" : "completed";
  execution.totalCost = execution.steps.reduce((s, r) => s + r.cost, 0);
  execution.totalTokens = execution.steps.reduce((s, r) => s + r.tokens, 0);
  execution.totalLatency = execution.steps.reduce((s, r) => s + r.latencyMs, 0);

  await pool.query(
    `INSERT INTO chain_executions (id, chain_name, status, steps, total_cost, total_tokens, total_latency_ms, started_at) VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())`,
    [id, definition.name, execution.status, JSON.stringify(execution.steps), execution.totalCost, execution.totalTokens, execution.totalLatency]
  );

  return execution;
}

async function executeLLMStep(config: any, input: any): Promise<{ output: any; cost: number; tokens: number }> {
  // In production: call LLM API
  const prompt = typeof config.prompt === "function" ? config.prompt(input) : interpolate(config.prompt, input);
  // Simplified
  const outputTokens = 500;
  const inputTokens = Math.ceil(prompt.length / 4);
  const cost = (inputTokens * 0.000003 + outputTokens * 0.000015);
  return { output: `Response for: ${prompt.slice(0, 100)}`, cost, tokens: inputTokens + outputTokens };
}

function executeTransform(transform: string, input: any): any {
  switch (transform) {
    case "json_parse": return typeof input === "string" ? JSON.parse(input) : input;
    case "first_item": return Array.isArray(input) ? input[0] : input;
    case "flatten": return Array.isArray(input) ? input.flat() : input;
    case "to_string": return JSON.stringify(input);
    default: return input;
  }
}

function evaluateCondition(condition: string, context: any): boolean {
  try {
    const fn = new Function("ctx", `return ${condition}`);
    return !!fn(context);
  } catch { return false; }
}

function resolveInput(inputConfig: any, context: Record<string, any>): any {
  if (typeof inputConfig === "string") return context[inputConfig] || inputConfig;
  if (typeof inputConfig === "object") {
    const resolved: any = {};
    for (const [key, value] of Object.entries(inputConfig)) {
      resolved[key] = typeof value === "string" && value.startsWith("$") ? context[value.slice(1)] : value;
    }
    return resolved;
  }
  return inputConfig;
}

function interpolate(template: string, data: any): string {
  return template.replace(/\{\{(\w+)\}\}/g, (_, key) => data[key] !== undefined ? String(data[key]) : `{{${key}}}`);
}

// Get chain execution history
export async function getChainHistory(chainName: string, limit: number = 20): Promise<ChainExecution[]> {
  const { rows } = await pool.query(
    "SELECT * FROM chain_executions WHERE chain_name = $1 ORDER BY started_at DESC LIMIT $2",
    [chainName, limit]
  );
  return rows.map((r: any) => ({ ...r, steps: JSON.parse(r.steps) }));
}

// Cost analytics
export async function getCostAnalytics(): Promise<Record<string, { executions: number; avgCost: number; avgLatency: number; failRate: number }>> {
  const { rows } = await pool.query(
    `SELECT chain_name, COUNT(*) as executions, AVG(total_cost) as avg_cost, AVG(total_latency_ms) as avg_latency,
       COUNT(*) FILTER (WHERE status = 'failed') as failures
     FROM chain_executions WHERE started_at > NOW() - INTERVAL '7 days' GROUP BY chain_name`
  );
  return Object.fromEntries(rows.map((r: any) => [r.chain_name, {
    executions: parseInt(r.executions),
    avgCost: Math.round(parseFloat(r.avg_cost) * 10000) / 10000,
    avgLatency: Math.round(parseFloat(r.avg_latency)),
    failRate: parseInt(r.executions) > 0 ? Math.round((parseInt(r.failures) / parseInt(r.executions)) * 100) : 0,
  }]));
}