IA Agentica en Produccion: De Prototipo a Servicio Confiable

npm install openai zod bullmq ioredis @opentelemetry/api @opentelemetry/sdk-node

interface AgentState {
  messages: Message[];
  steps: AgentStep[];
  startTime: number;
  totalTokens: number;
}
 
interface AgentStep {
  thought: string;
  action?: { tool: string; input: Record<string, unknown> };
  observation?: string;
  error?: string;
  durationMs: number;
}
 
interface AgentConfig {
  maxSteps: number;        // Hard cap on reasoning cycles
  maxTokens: number;       // Budget ceiling per run
  timeoutMs: number;       // Wall-clock deadline
  tools: ToolDefinition[];
}
 
async function runReActAgent(
  prompt: string,
  config: AgentConfig
): Promise<AgentResult> {
  const state: AgentState = {
    messages: [{ role: 'user', content: prompt }],
    steps: [],
    startTime: Date.now(),
    totalTokens: 0,
  };
 
  for (let step = 0; step < config.maxSteps; step++) {
    // Guard: wall-clock timeout
    const elapsed = Date.now() - state.startTime;
    if (elapsed > config.timeoutMs) {
      return finalize(state, 'timeout',
        `Agent timed out after ${elapsed}ms (limit: ${config.timeoutMs}ms)`);
    }
 
    // Guard: token budget
    if (state.totalTokens > config.maxTokens) {
      return finalize(state, 'budget_exceeded',
        `Token budget exceeded: ${state.totalTokens}/${config.maxTokens}`);
    }
 
    const stepStart = Date.now();
 
    // Reason: ask the LLM what to do next
    const response = await callLLM(state.messages, config.tools);
    state.totalTokens += response.usage.totalTokens;
 
    // Check if the agent wants to respond (no tool call)
    if (!response.toolCall) {
      state.steps.push({
        thought: response.content,
        durationMs: Date.now() - stepStart,
      });
      return finalize(state, 'complete', response.content);
    }
 
    // Act: execute the tool
    const { tool, input } = response.toolCall;
    let observation: string;
    let error: string | undefined;
 
    try {
      observation = await executeTool(tool, input, config);
    } catch (err) {
      error = err instanceof Error ? err.message : 'Unknown tool error';
      observation = `Tool "${tool}" failed: ${error}`;
    }
 
    // Record the step
    state.steps.push({
      thought: response.content,
      action: { tool, input },
      observation,
      error,
      durationMs: Date.now() - stepStart,
    });
 
    // Observe: feed the result back to the LLM
    state.messages.push(
      { role: 'assistant', content: response.content, toolCall: response.toolCall },
      { role: 'tool', content: observation, toolCallId: response.toolCall.id }
    );
  }
 
  // Exhausted all steps without completing
  return finalize(state, 'max_steps_exceeded',
    `Agent reached ${config.maxSteps} steps without completing`);
}

interface ExecutionPlan {
  goal: string;
  steps: PlanStep[];
  estimatedCost: number;
  estimatedDurationMs: number;
}
 
interface PlanStep {
  id: string;
  description: string;
  tool: string;
  dependencies: string[];  // IDs of steps that must complete first
  fallback?: string;       // Alternative tool if primary fails
}
 
async function planAndExecute(
  prompt: string,
  config: AgentConfig
): Promise<AgentResult> {
  // Phase 1: Generate plan
  const plan = await generatePlan(prompt, config.tools);
 
  // Phase 1b: Validate — reject plans that exceed budget or use unknown tools
  const validation = validatePlan(plan, config);
  if (!validation.valid) {
    return { status: 'plan_rejected', reason: validation.errors };
  }
 
  // Phase 2: Execute with dependency ordering
  const completed = new Map<string, StepResult>();
  const sortedSteps = topologicalSort(plan.steps);
 
  for (const step of sortedSteps) {
    // Wait for dependencies
    const depResults = step.dependencies.map(id => completed.get(id));
    if (depResults.some(r => r?.status === 'failed' && !step.fallback)) {
      completed.set(step.id, {
        status: 'skipped',
        reason: 'Dependency failed'
      });
      continue;
    }
 
    // Execute step with retry
    const result = await executeStepWithRetry(step, depResults, config);
    completed.set(step.id, result);
 
    // If step failed and has a fallback, try the alternative
    if (result.status === 'failed' && step.fallback) {
      const fallbackResult = await executeStepWithRetry(
        { ...step, tool: step.fallback }, depResults, config
      );
      completed.set(step.id, fallbackResult);
    }
  }
 
  return synthesizeResults(plan, completed);
}

enum ErrorCategory {
  TRANSIENT = 'transient',     // Retry with backoff
  PERMANENT = 'permanent',     // Fail fast, don't retry
  RATE_LIMIT = 'rate_limit',   // Retry with longer backoff
  CONTEXT = 'context',         // LLM misunderstood — rephrase
  BUDGET = 'budget',           // Resource limit hit — escalate
}
 
function classifyError(error: unknown): ErrorCategory {
  if (error instanceof Error) {
    const msg = error.message.toLowerCase();
    const status = (error as any).status || (error as any).statusCode;
 
    // HTTP status-based classification
    if (status === 429) return ErrorCategory.RATE_LIMIT;
    if (status === 408 || status === 502 || status === 503 || status === 504) {
      return ErrorCategory.TRANSIENT;
    }
    if (status === 400 || status === 404 || status === 422) {
      return ErrorCategory.PERMANENT;
    }
    if (status === 402) return ErrorCategory.BUDGET;
 
    // Message-based classification for LLM-specific errors
    if (msg.includes('timeout') || msg.includes('econnreset')) {
      return ErrorCategory.TRANSIENT;
    }
    if (msg.includes('context length') || msg.includes('token limit')) {
      return ErrorCategory.CONTEXT;
    }
    if (msg.includes('invalid') || msg.includes('not found')) {
      return ErrorCategory.PERMANENT;
    }
  }
 
  return ErrorCategory.TRANSIENT; // Default to retryable
}

interface RetryConfig {
  maxRetries: number;
  baseDelayMs: number;
  maxDelayMs: number;
  jitterFactor: number;     // 0-1, how much randomness
}
 
const RETRY_CONFIGS: Record<ErrorCategory, RetryConfig | null> = {
  [ErrorCategory.TRANSIENT]: {
    maxRetries: 3,
    baseDelayMs: 500,
    maxDelayMs: 10_000,
    jitterFactor: 0.5,
  },
  [ErrorCategory.RATE_LIMIT]: {
    maxRetries: 5,
    baseDelayMs: 2_000,
    maxDelayMs: 60_000,
    jitterFactor: 1.0,       // Full jitter to spread load
  },
  [ErrorCategory.PERMANENT]: null,  // Never retry
  [ErrorCategory.CONTEXT]: null,    // Rephrase, don't retry same call
  [ErrorCategory.BUDGET]: null,     // Escalate, don't retry
};
 
function calculateDelay(attempt: number, config: RetryConfig): number {
  // Exponential: 500, 1000, 2000, 4000...
  const exponential = config.baseDelayMs * Math.pow(2, attempt);
  const capped = Math.min(exponential, config.maxDelayMs);
 
  // Full jitter: uniform random between 0 and capped value
  const jitter = capped * config.jitterFactor * Math.random();
  const base = capped * (1 - config.jitterFactor);
 
  return base + jitter;
}
 
async function executeWithRetry<T>(
  fn: () => Promise<T>,
  label: string
): Promise<T> {
  let lastError: Error | undefined;
 
  for (let attempt = 0; ; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error instanceof Error ? error : new Error(String(error));
      const category = classifyError(error);
      const retryConfig = RETRY_CONFIGS[category];
 
      // Non-retryable or exhausted retries
      if (!retryConfig || attempt >= retryConfig.maxRetries) {
        throw new AgentError(
          `${label} failed after ${attempt + 1} attempts: ${lastError.message}`,
          { category, attempts: attempt + 1, originalError: lastError }
        );
      }
 
      const delay = calculateDelay(attempt, retryConfig);
      console.warn(
        `[retry] ${label} attempt ${attempt + 1}/${retryConfig.maxRetries}` +
        ` (${category}), waiting ${Math.round(delay)}ms`
      );
      await sleep(delay);
    }
  }
}

enum CircuitState {
  CLOSED = 'closed',       // Normal: requests flow through
  OPEN = 'open',           // Tripped: fail immediately
  HALF_OPEN = 'half_open', // Testing: allow one probe request
}
 
class CircuitBreaker {
  private state = CircuitState.CLOSED;
  private failureCount = 0;
  private lastFailureTime = 0;
  private successCount = 0;
 
  constructor(
    private readonly name: string,
    private readonly config: {
      failureThreshold: number;  // Failures before opening
      resetTimeoutMs: number;    // How long to stay open
      halfOpenSuccesses: number; // Successes needed to close
    }
  ) {}
 
  async execute<T>(fn: () => Promise<T>, fallback?: () => Promise<T>): Promise<T> {
    // Check if circuit should transition from open to half-open
    if (this.state === CircuitState.OPEN) {
      const elapsed = Date.now() - this.lastFailureTime;
      if (elapsed >= this.config.resetTimeoutMs) {
        this.state = CircuitState.HALF_OPEN;
        this.successCount = 0;
      } else if (fallback) {
        return fallback();
      } else {
        throw new CircuitOpenError(
          `Circuit "${this.name}" is open. Retry after ` +
          `${this.config.resetTimeoutMs - elapsed}ms`
        );
      }
    }
 
    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
 
      if (fallback && this.state === CircuitState.OPEN) {
        return fallback();
      }
      throw error;
    }
  }
 
  private onSuccess(): void {
    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.config.halfOpenSuccesses) {
        this.state = CircuitState.CLOSED;
        this.failureCount = 0;
      }
    } else {
      this.failureCount = 0;
    }
  }
 
  private onFailure(): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();
 
    if (this.failureCount >= this.config.failureThreshold) {
      this.state = CircuitState.OPEN;
    }
  }
 
  getState(): { state: CircuitState; failures: number } {
    return { state: this.state, failures: this.failureCount };
  }
}

interface DegradationLayer {
  name: string;
  isAvailable: () => Promise<boolean>;
  execute: (input: AgentInput) => Promise<AgentOutput>;
}
 
class DegradationPipeline {
  private layers: DegradationLayer[];
 
  constructor(layers: DegradationLayer[]) {
    // Ordered from most capable to most resilient
    this.layers = layers;
  }
 
  async execute(input: AgentInput): Promise<AgentOutput & { layer: string }> {
    for (const layer of this.layers) {
      try {
        const available = await Promise.race([
          layer.isAvailable(),
          sleep(2000).then(() => false), // Health check timeout
        ]);
 
        if (!available) continue;
 
        const result = await layer.execute(input);
        return { ...result, layer: layer.name };
      } catch {
        continue; // Fall through to next layer
      }
    }
 
    // All layers exhausted — return the hardcoded safety net
    return {
      response: "I'm currently experiencing technical difficulties. " +
                "Please try again in a few minutes or contact support.",
      layer: 'safety_net',
      confidence: 0,
    };
  }
}
 
// Example: customer support agent with four degradation layers
const supportAgent = new DegradationPipeline([
  {
    name: 'full_agent',
    isAvailable: async () => {
      const llm = await checkLLMHealth();
      const kb = await checkKnowledgeBaseHealth();
      const crm = await checkCRMHealth();
      return llm && kb && crm;
    },
    execute: async (input) => {
      // Full capability: LLM + knowledge base + CRM tools
      return runReActAgent(input.query, fullConfig);
    },
  },
  {
    name: 'no_crm',
    isAvailable: async () => {
      const llm = await checkLLMHealth();
      const kb = await checkKnowledgeBaseHealth();
      return llm && kb;
    },
    execute: async (input) => {
      // LLM + knowledge base only — no customer-specific data
      const result = await runReActAgent(input.query, kbOnlyConfig);
      result.response += "\n\nNote: I couldn't access your account details. " +
        "For order-specific questions, please contact support.";
      return result;
    },
  },
  {
    name: 'cached_responses',
    isAvailable: async () => true, // Cache is always "available"
    execute: async (input) => {
      // Semantic search over cached responses for common queries
      const cached = await searchResponseCache(input.query);
      if (cached && cached.similarity > 0.85) {
        return { response: cached.response, confidence: cached.similarity };
      }
      throw new Error('No suitable cached response');
    },
  },
  {
    name: 'static_fallback',
    isAvailable: async () => true,
    execute: async (input) => {
      // Rule-based response for known intents
      const intent = classifyIntentLocally(input.query);
      const staticResponse = STATIC_RESPONSES[intent] || STATIC_RESPONSES.default;
      return { response: staticResponse, confidence: 0.3 };
    },
  },
]);

interface ModelConfig {
  provider: 'openai' | 'anthropic' | 'google';
  model: string;
  maxTokens: number;
  costPer1kTokens: number;
}
 
const MODEL_CHAIN: ModelConfig[] = [
  { provider: 'anthropic', model: 'claude-sonnet-4-20250514', maxTokens: 8192, costPer1kTokens: 0.003 },
  { provider: 'openai', model: 'gpt-4o', maxTokens: 4096, costPer1kTokens: 0.005 },
  { provider: 'google', model: 'gemini-2.0-flash', maxTokens: 8192, costPer1kTokens: 0.001 },
];
 
async function callLLMWithFallback(
  messages: Message[],
  tools: ToolDefinition[]
): Promise<LLMResponse> {
  const errors: Array<{ model: string; error: string }> = [];
 
  for (const config of MODEL_CHAIN) {
    const breaker = getCircuitBreaker(`llm:${config.provider}`);
 
    try {
      return await breaker.execute(
        () => callProvider(config, messages, tools),
      );
    } catch (error) {
      const msg = error instanceof Error ? error.message : 'Unknown';
      errors.push({ model: `${config.provider}/${config.model}`, error: msg });
    }
  }
 
  throw new AllModelsFailedError(
    `All ${MODEL_CHAIN.length} models failed`,
    errors
  );
}

import { trace, SpanKind, SpanStatusCode, context } from '@opentelemetry/api';
 
const tracer = trace.getTracer('agent-service', '1.0.0');
 
async function tracedAgentRun(
  prompt: string,
  config: AgentConfig
): Promise<AgentResult> {
  return tracer.startActiveSpan('agent.run', {
    kind: SpanKind.SERVER,
    attributes: {
      'gen_ai.system': 'custom',
      'gen_ai.request.model': config.model,
      'agent.max_steps': config.maxSteps,
      'agent.timeout_ms': config.timeoutMs,
    },
  }, async (runSpan) => {
    try {
      const result = await runReActAgent(prompt, config);
 
      runSpan.setAttributes({
        'agent.status': result.status,
        'agent.steps_taken': result.steps.length,
        'agent.total_tokens': result.totalTokens,
        'agent.duration_ms': result.durationMs,
      });
      runSpan.setStatus({ code: SpanStatusCode.OK });
 
      return result;
    } catch (error) {
      runSpan.setStatus({
        code: SpanStatusCode.ERROR,
        message: error instanceof Error ? error.message : 'Unknown',
      });
      runSpan.recordException(error as Error);
      throw error;
    } finally {
      runSpan.end();
    }
  });
}
 
async function tracedToolCall(
  toolName: string,
  input: Record<string, unknown>
): Promise<string> {
  return tracer.startActiveSpan(`tool.${toolName}`, {
    kind: SpanKind.CLIENT,
    attributes: {
      'tool.name': toolName,
      'tool.input_keys': Object.keys(input).join(','),
    },
  }, async (toolSpan) => {
    const start = Date.now();
    try {
      const result = await executeTool(toolName, input);
      toolSpan.setAttributes({
        'tool.duration_ms': Date.now() - start,
        'tool.output_length': result.length,
        'tool.success': true,
      });
      return result;
    } catch (error) {
      toolSpan.setAttributes({
        'tool.duration_ms': Date.now() - start,
        'tool.success': false,
        'tool.error_category': classifyError(error),
      });
      throw error;
    } finally {
      toolSpan.end();
    }
  });
}

interface AgentAlert {
  name: string;
  condition: string;
  severity: 'warning' | 'critical';
  runbook: string;
}
 
const PRODUCTION_ALERTS: AgentAlert[] = [
  {
    name: 'success_rate_drop',
    condition: 'success_rate < 0.95 over 5 minutes',
    severity: 'critical',
    runbook: 'Check LLM provider status, review recent tool errors, ' +
             'verify knowledge base connectivity',
  },
  {
    name: 'p95_latency_spike',
    condition: 'p95_duration_ms > 15000 over 5 minutes',
    severity: 'warning',
    runbook: 'Check LLM response times, look for tool timeout patterns, ' +
             'verify no infinite loop in agent steps',
  },
  {
    name: 'token_cost_anomaly',
    condition: 'hourly_token_cost > 2x rolling_7d_average',
    severity: 'warning',
    runbook: 'Check for prompt injection attempts, review agent step counts, ' +
             'look for context window overflow patterns',
  },
  {
    name: 'circuit_breaker_open',
    condition: 'any circuit breaker state == open',
    severity: 'critical',
    runbook: 'Identify which dependency tripped, check downstream service health, ' +
             'verify degradation pipeline is activating',
  },
  {
    name: 'eval_score_regression',
    condition: 'rolling_eval_score < baseline - 0.5 over 1 hour',
    severity: 'warning',
    runbook: 'Compare recent prompts to baseline, check for model version change, ' +
             'review knowledge base freshness',
  },
];

import { Queue, Worker, Job } from 'bullmq';
import IORedis from 'ioredis';
 
const connection = new IORedis({
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT || '6379'),
  maxRetriesPerRequest: null,
});
 
// Producer: accepts requests and enqueues them
const agentQueue = new Queue('agent-runs', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 86400 },    // Keep completed jobs for 24h
    removeOnFail: { age: 604800 },       // Keep failed jobs for 7 days
  },
});
 
async function enqueueAgentRun(request: AgentRequest): Promise<string> {
  const job = await agentQueue.add('run', request, {
    priority: request.priority || 5,      // Lower = higher priority
    timeout: request.timeoutMs || 120_000,
  });
  return job.id!;
}
 
// Worker: processes agent runs from the queue
const worker = new Worker('agent-runs', async (job: Job<AgentRequest>) => {
  const { prompt, config, conversationId } = job.data;
 
  // Load conversation state from external store (not process memory)
  const history = await loadConversationHistory(conversationId);
 
  const result = await tracedAgentRun(prompt, {
    ...config,
    messages: history,
  });
 
  // Persist updated state
  await saveConversationHistory(conversationId, result.messages);
 
  // Publish result for the waiting client
  await publishResult(job.id!, result);
 
  return result;
}, {
  connection,
  concurrency: parseInt(process.env.WORKER_CONCURRENCY || '10'),
  limiter: {
    max: 50,           // Max 50 jobs per duration window
    duration: 60_000,  // Per minute — prevents LLM rate limits
  },
});
 
worker.on('failed', (job, err) => {
  console.error(`Agent run ${job?.id} failed:`, err.message);
  // Emit metric for alerting
  metrics.increment('agent.run.failed', {
    error_category: classifyError(err)
  });
});

interface ConversationStore {
  // Hot path: Redis (active conversations)
  getActive(conversationId: string): Promise<ConversationState | null>;
  setActive(conversationId: string, state: ConversationState, ttlSeconds?: number): Promise<void>;
 
  // Cold path: database (historical)
  archive(conversationId: string): Promise<void>;
  getArchived(conversationId: string): Promise<ConversationState | null>;
}
 
class RedisConversationStore implements ConversationStore {
  constructor(private redis: IORedis) {}
 
  async getActive(id: string): Promise<ConversationState | null> {
    const data = await this.redis.get(`conv:${id}`);
    return data ? JSON.parse(data) : null;
  }
 
  async setActive(
    id: string,
    state: ConversationState,
    ttlSeconds = 3600
  ): Promise<void> {
    await this.redis.setex(
      `conv:${id}`,
      ttlSeconds,
      JSON.stringify(state)
    );
  }
 
  async archive(id: string): Promise<void> {
    const state = await this.getActive(id);
    if (state) {
      await db.conversations.insertOne({ ...state, archivedAt: new Date() });
      await this.redis.del(`conv:${id}`);
    }
  }
 
  async getArchived(id: string): Promise<ConversationState | null> {
    return db.conversations.findOne({ conversationId: id });
  }
}