Innova IA en Producción: Monitorabilidad y Observabilidad de Sistemas Multi-Agente

gen_ai.system: "openai" | "anthropic" | "google"
gen_ai.request.model: "gpt-4-turbo" | "claude-opus-4-6"
gen_ai.request.max_tokens: 4096
gen_ai.usage.input_tokens: 1240
gen_ai.usage.output_tokens: 856
gen_ai.response.finish_reason: "stop" | "length" | "tool_use"
gen_ai.tool.name: "search_database"
gen_ai.tool.call_count: 1

┌─────────────────────────────────────────────────────────┐
│                    GKE Cluster                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │Orchestr- │  │  Worker  │  │ Evaluator│              │
│  │  ator    │──│   Pool   │──│   Agent  │              │
│  └────┬─────┘  └────┬─────┘  └──────────┘              │
│       │             │                                    │
│       └─────────────┼─────────────────────────────┐     │
│                     │ OpenTelemetry Exporter      │     │
│                     │ (OTLP gRPC)                 │     │
│                     └──────────────┬──────────────┘     │
└───────────────────────────────────┼────────────────────┘
                                    │
                                    │ gRPC (Secure)
                                    │
┌───────────────────────────────────┼────────────────────┐
│        Google Cloud Operations     │                   │
│  ┌──────────────┐  ┌──────────────┐                    │
│  │Cloud Trace   │  │Cloud Metrics │                    │
│  │(Tracing)     │  │(Monitoring)  │                    │
│  └──────────────┘  └──────────────┘                    │
│  ┌──────────────┐  ┌──────────────┐                    │
│  │Cloud Logging │  │Error Reporting│                   │
│  └──────────────┘  └──────────────┘                    │
└─────────────────────────────────────────────────────────┘

// pkg/telemetry/opentelemetry.go
package telemetry

import (
	"context"
	"fmt"
	"os"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/propagation"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/trace"
)

type TelemetryConfig struct {
	ServiceName    string
	ServiceVersion string
	Environment    string
}

func NewOpenTelemetry(ctx context.Context, cfg TelemetryConfig) (trace.Tracer, error) {
	// Configurar exporter para GCP Cloud Trace via OTLP
	projectID := os.Getenv("GOOGLE_CLOUD_PROJECT")
	exporter, err := otlptracegrpc.New(ctx,
		otlptracegrpc.WithEndpoint(fmt.Sprintf("otlp-gcp.cloudtrace.googleapis.com:443")),
		otlptracegrpc.WithHeaders(map[string]string{
			"x-google-project-id": projectID,
		}),
		otlptracegrpc.WithInsecure(false),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create OTLP exporter: %w", err)
	}

	// Configurar tracer provider con GCP resource attributes
	tp := sdktrace.NewTracerProvider(
		sdktrace.WithBatcher(exporter),
		sdktrace.WithResource(
			resource.NewWithAttributes(
				semconv.SchemaURL,
				semconv.ServiceNameKey.String(cfg.ServiceName),
				semconv.ServiceVersionKey.String(cfg.ServiceVersion),
				semconv.DeploymentEnvironmentKey.String(cfg.Environment),
				semconv.CloudProviderKey.String("gcp"),
				semconv.CloudPlatformKey.String("gcp_kubernetes_engine"),
			),
		),
	)

	// Configurar propagación automática de context
	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	))

	// Set global tracer provider
	otel.SetTracerProvider(tp)

	return tp.Tracer(cfg.ServiceName), nil
}

// pkg/agent/instrumented.go
package agent

import (
	"context"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
)

type AgentInstrumentation struct {
	tracer trace.Tracer
}

type AgentContext struct {
	AgentID    string
	AgentType  string
	Goal       string
	InputTokens int
}

type ToolInvocation struct {
	ToolName string
	Args     map[string]interface{}
	Duration time.Duration
	Success  bool
	Error    error
}

func NewAgentInstrumentation(tracer trace.Tracer) *AgentInstrumentation {
	return &AgentInstrumentation{tracer: tracer}
}

// TraceAgentExecution crea el trace principal de ejecución de un agente
func (ai *AgentInstrumentation) TraceAgentExecution(
	ctx context.Context,
	agentCtx AgentContext,
	fn func(context.Context) ([]byte, error),
) ([]byte, error) {
	spanName := fmt.Sprintf("agent.%s.execute", agentCtx.AgentType)
	ctx, span := ai.tracer.Start(ctx, spanName,
		trace.WithAttributes(
			attribute.String("agent.id", agentCtx.AgentID),
			attribute.String("agent.type", agentCtx.AgentType),
			attribute.String("agent.goal", agentCtx.Goal),
			attribute.Int("agent.input_tokens", agentCtx.InputTokens),
		),
	)
	defer span.End()

	startTime := time.Now()

	// Ejecutar la lógica del agente
	result, err := fn(ctx)

	duration := time.Since(startTime)

	// Registrar métricas de ejecución
	span.SetAttributes(
		attribute.Int("agent.output_length", len(result)),
		attribute.Int64("agent.duration_ms", duration.Milliseconds()),
	)

	if err != nil {
		span.SetStatus(codes.Error, err.Error())
		span.SetAttributes(
			attribute.String("agent.error_type", classifyError(err)),
		)
		return nil, err
	}

	span.SetStatus(codes.Ok, "agent executed successfully")
	return result, nil
}

// TraceLLMCall instrumenta cada llamada al modelo LLM
func (ai *AgentInstrumentation) TraceLLMCall(
	ctx context.Context,
	model string,
	systemPrompt string,
	userPrompt string,
	fn func(context.Context) (*LLMResponse, error),
) (*LLMResponse, error) {
	ctx, span := ai.tracer.Start(ctx, "llm.inference",
		trace.WithAttributes(
			attribute.String("gen_ai.system", extractProvider(model)),
			attribute.String("gen_ai.request.model", model),
			attribute.Int("gen_ai.prompt.length", len(systemPrompt)+len(userPrompt)),
			// NO loggear el prompt completo en producción si contiene PII
		),
	)
	defer span.End()

	startTime := time.Now()
	response, err := fn(ctx)
	duration := time.Since(startTime)

	if err != nil {
		span.SetStatus(codes.Error, err.Error())
		return nil, err
	}

	// Registrar atributos de uso de tokens
	span.SetAttributes(
		attribute.Int("gen_ai.usage.input_tokens", response.InputTokens),
		attribute.Int("gen_ai.usage.output_tokens", response.OutputTokens),
		attribute.Int("gen_ai.usage.total_tokens", response.InputTokens+response.OutputTokens),
		attribute.String("gen_ai.response.finish_reason", response.FinishReason),
		attribute.Int64("llm.duration_ms", duration.Milliseconds()),
		attribute.Float64("llm.tokens_per_second", 
			float64(response.InputTokens+response.OutputTokens)/duration.Seconds(),
		),
	)

	// Emitir evento para tracking de decisiones
	if response.FinishReason == "tool_use" {
		span.AddEvent("llm.tool_use_requested",
			trace.WithAttributes(
				attribute.String("tool.name", response.ToolName),
			),
		)
	}

	return response, nil
}

// TraceToolCall instrumenta cada invocación de herramienta
func (ai *AgentInstrumentation) TraceToolCall(
	ctx context.Context,
	tool ToolInvocation,
	fn func(context.Context) (interface{}, error),
) (interface{}, error) {
	spanName := fmt.Sprintf("tool.%s.execute", tool.ToolName)
	ctx, span := ai.tracer.Start(ctx, spanName,
		trace.WithAttributes(
			attribute.String("tool.name", tool.ToolName),
			attribute.String("tool.args", formatArgs(tool.Args)),
		),
	)
	defer span.End()

	startTime := time.Now()
	result, err := fn(ctx)
	duration := time.Since(startTime)

	span.SetAttributes(
		attribute.Int64("tool.duration_ms", duration.Milliseconds()),
		attribute.Bool("tool.success", tool.Success),
	)

	if err != nil {
		span.SetStatus(codes.Error, err.Error())
		span.AddEvent("tool.execution_failed",
			trace.WithAttributes(
				attribute.String("tool.error", err.Error()),
			),
		)
		return nil, err
	}

	span.AddEvent("tool.execution_succeeded",
		trace.WithAttributes(
			attribute.Int("tool.result_size", estimateSize(result)),
		),
	)

	return result, nil
}

// pkg/orchestrator/instrumented.go
package orchestrator

import (
	"context"
	"fmt"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
)

type OrchestratorInstrumentation struct {
	tracer trace.Tracer
	agentInstrumentation *agent.AgentInstrumentation
}

type Task struct {
	ID          string
	Type        string
	Description string
	Priority    int
}

type OrchestrationResult struct {
	Success       bool
	FinalOutput   []byte
	AgentTraceIDs []string
	TokenUsage    int
	ExecutionTime time.Duration
	ValidationPasses bool
}

func NewOrchestratorInstrumentation(
	tracer trace.Tracer,
	agentInstr *agent.AgentInstrumentation,
) *OrchestratorInstrumentation {
	return &OrchestratorInstrumentation{
		tracer: tracer,
		agentInstrumentation: agentInstr,
	}
}

// TraceOrchestration crea el trace raíz de toda la ejecución
func (oi *OrchestratorInstrumentation) TraceOrchestration(
	ctx context.Context,
	task Task,
	fn func(context.Context) (*OrchestrationResult, error),
) (*OrchestrationResult, error) {
	spanName := fmt.Sprintf("orchestration.%s", task.Type)
	ctx, rootSpan := oi.tracer.Start(ctx, spanName,
		trace.WithAttributes(
			attribute.String("orchestration.task_id", task.ID),
			attribute.String("orchestration.task_type", task.Type),
			attribute.String("orchestration.description", task.Description),
			attribute.Int("orchestration.priority", task.Priority),
		),
	)
	defer rootSpan.End()

	startTime := time.Now()

	// Ejecutar la orquestación
	result, err := fn(ctx)

	duration := time.Since(startTime)

	// Métricas de alto nivel
	rootSpan.SetAttributes(
		attribute.Bool("orchestration.success", result.Success),
		attribute.Int("orchestration.total_token_usage", result.TokenUsage),
		attribute.Int64("orchestration.duration_ms", duration.Milliseconds()),
		attribute.Bool("orchestration.validation_passed", result.ValidationPasses),
		attribute.Int("orchestration.agents_invoked", len(result.AgentTraceIDs)),
	)

	// Validación de calidad del resultado
	if err != nil {
		rootSpan.SetStatus(codes.Error, err.Error())
		return nil, err
	}

	if !result.ValidationPasses {
		rootSpan.SetStatus(codes.Error, "validation failed")
		rootSpan.AddEvent("orchestration.validation_failed",
			trace.WithAttributes(
				attribute.String("validation.reason", "output did not meet quality criteria"),
			),
		)
		return result, fmt.Errorf("validation failed")
	}

	rootSpan.SetStatus(codes.Ok, "orchestration completed successfully")
	return result, nil
}

// pkg/telemetry/metrics.go
package telemetry

import (
	"context"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

type AgentMetrics struct {
	meter metric.Meter

	// Métricas de ejecución
	taskCounter           metric.Int64Counter
	taskLatency           metric.Float64Histogram
	tokenUsage            metric.Int64Histogram
	
	// Métricas de calidad
	validationPassRate    metric.Float64Gauge
	outputQualityScore    metric.Float64Histogram
	correctionLoopCount   metric.Int64Histogram
	
	// Métricas de costos
	costPerTask           metric.Float64Histogram
	totalCost             metric.Float64Counter
}

func NewAgentMetrics(meter metric.Meter) (*AgentMetrics, error) {
	m := &AgentMetrics{meter: meter}

	var err error

	// Counter de tareas ejecutadas
	m.taskCounter, err = meter.Int64Counter(
		"agent.tasks.total",
		metric.WithDescription("Total number of agent tasks executed"),
	)
	if err != nil {
		return nil, err
	}

	// Histogram de latencia de tareas
	m.taskLatency, err = meter.Float64Histogram(
		"agent.tasks.latency_ms",
		metric.WithDescription("Task execution latency in milliseconds"),
		metric.WithExplicitBucketBoundaries(10, 50, 100, 250, 500, 1000, 2000, 5000, 10000),
	)
	if err != nil {
		return nil, err
	}

	// Histogram de uso de tokens
	m.tokenUsage, err = meter.Int64Histogram(
		"agent.tokens.usage",
		metric.WithDescription("Token consumption per task"),
		metric.WithExplicitBucketBoundaries(100, 500, 1000, 2000, 4000, 8000, 16000, 32000),
	)
	if err != nil {
		return nil, err
	}

	// Gauge de tasa de validación exitosa
	m.validationPassRate, err = meter.Float64Gauge(
		"agent.validation.pass_rate",
		metric.WithDescription("Rate of tasks passing validation"),
	)
	if err != nil {
		return nil, err
	}

	// Histogram de score de calidad de output
	m.outputQualityScore, err = meter.Float64Histogram(
		"agent.output.quality_score",
		metric.WithDescription("Output quality score (0-100)"),
		metric.WithExplicitBucketBoundaries(0, 20, 40, 60, 80, 90, 95, 99, 100),
	)
	if err != nil {
		return nil, err
	}

	// Histogram de loops de corrección
	m.correctionLoopCount, err = meter.Int64Histogram(
		"agent.correction.loops",
		metric.WithDescription("Number of correction loops per task"),
		metric.WithExplicitBucketBoundaries(0, 1, 2, 3, 4, 5, 10),
	)
	if err != nil {
		return nil, err
	}

	// Histogram de costo por tarea
	m.costPerTask, err = meter.Float64Histogram(
		"agent.cost.per_task",
		metric.WithDescription("Cost in USD per task"),
		metric.WithExplicitBucketBoundaries(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0),
	)
	if err != nil {
		return nil, err
	}

	// Counter de costo total
	m.totalCost, err = meter.Float64Counter(
		"agent.cost.total_usd",
		metric.WithDescription("Total cost in USD"),
	)
	if err != nil {
		return nil, err
	}

	return m, nil
}

func (m *AgentMetrics) RecordTask(
	ctx context.Context,
	taskType string,
	latency time.Duration,
	tokenUsage int,
	validationPassed bool,
	qualityScore float64,
	correctionLoops int,
	costUSD float64,
) {
	commonAttrs := []attribute.KeyValue{
		attribute.String("task.type", taskType),
	}

	// Registrar counter de tareas
	m.taskCounter.Add(ctx, 1, metric.WithAttributes(commonAttrs...))

	// Registrar latencia
	m.taskLatency.Record(ctx, float64(latency.Milliseconds()), 
		metric.WithAttributes(commonAttrs...))

	// Registrar uso de tokens
	m.tokenUsage.Record(ctx, int64(tokenUsage), metric.WithAttributes(commonAttrs...))

	// Registrar quality score
	if qualityScore > 0 {
		m.outputQualityScore.Record(ctx, qualityScore, metric.WithAttributes(commonAttrs...))
	}

	// Registrar loops de corrección
	if correctionLoops > 0 {
		m.correctionLoopCount.Record(ctx, int64(correctionLoops), 
			metric.WithAttributes(commonAttrs...))
	}

	// Registrar costo
	m.costPerTask.Record(ctx, costUSD, metric.WithAttributes(commonAttrs...))
	m.totalCost.Add(ctx, costUSD, metric.WithAttributes(commonAttrs...))
}

# k8s/orchestrator-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-orchestrator
  labels:
    app: agent-orchestrator
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-orchestrator
  template:
    metadata:
      labels:
        app: agent-orchestrator
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
    spec:
      containers:
      - name: orchestrator
        image: gcr.io/my-project/agent-orchestrator:v1.0.0
        ports:
        - containerPort: 8080
          name: http
        - containerPort: 9090
          name: metrics
        env:
        - name: GOOGLE_CLOUD_PROJECT
          valueFrom:
            configMapKeyRef:
              name: app-config
              key: project-id
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "otlp-gcp.cloudtrace.googleapis.com:443"
        - name: OTEL_SERVICE_NAME
          value: "agent-orchestrator"
        - name: OTEL_RESOURCE_ATTRIBUTES
          value: "deployment.environment=production,k8s.pod.name=$(POD_NAME)"
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2000m"
            memory: "2Gi"
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /readyz
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

# alerts/agent-system-alerts.yaml
apiVersion: monitoring.googleapis.com/v1
kind: AlertPolicy
metadata:
  name: agent-system-alerts
  displayName: "Alertas de Sistema Multi-Agente"
  description: "Alertas para monitoreo de sistemas multi-agente en producción"
spec:
  conditions:
  # Alerta: Tasa de validación baja
  - displayName: "Tasa de validación < 80%"
    conditionThreshold:
      filter: >
        resource.type="k8s_container" AND
        metric.type="custom.googleapis.com/agent/validation/pass_rate" AND
        metric.label.task_type="*"
      comparison: COMPARISON_LT
      thresholdValue: 0.8
      duration: 3600s
      aggregations:
      - alignmentPeriod: 300s
        perSeriesAligner: ALIGN_MEAN
        crossSeriesReducer: REDUCE_MEAN
        groupByFields:
        - metric.label.task_type
  
  # Alerta: Latencia p99 > 10s
  - displayName: "Latencia p99 > 10s"
    conditionThreshold:
      filter: >
        resource.type="k8s_container" AND
        metric.type="custom.googleapis.com/agent/tasks/latency_ms" AND
        resource.label.container_name="orchestrator"
      comparison: COMPARISON_GT
      thresholdValue: 10000
      duration: 900s
      aggregations:
      - alignmentPeriod: 300s
        perSeriesAligner: ALIGN_PERCENTILE_99
        crossSeriesReducer: REDUCE_MEAN
        groupByFields:
        - metric.label.task_type
  
  # Alerta: Loops de corrección > 3
  - displayName: "Loops de corrección > 3"
    conditionThreshold:
      filter: >
        resource.type="k8s_container" AND
        metric.type="custom.googleapis.com/agent/correction/loops"
      comparison: COMPARISON_GT
      thresholdValue: 3
      duration: 600s
      aggregations:
      - alignmentPeriod: 300s
        perSeriesAligner: ALIGN_PERCENTILE_95
        crossSeriesReducer: REDUCE_MEAN
  
  # Alerta: Costo por tarea > $1
  - displayName: "Costo por tarea > $1"
    conditionThreshold:
      filter: >
        resource.type="k8s_container" AND
        metric.type="custom.googleapis.com/agent/cost/per_task"
      comparison: COMPARISON_GT
      thresholdValue: 1.0
      duration: 1800s
      aggregations:
      - alignmentPeriod: 600s
        perSeriesAligner: ALIGN_PERCENTILE_95
        crossSeriesReducer: REDUCE_MEAN
        groupByFields:
        - metric.label.task_type
  
  # Alerta: Tasa de errores de herramientas > 5%
  - displayName: "Tasa de errores de herramientas > 5%"
    conditionThreshold:
      filter: >
        resource.type="k8s_container" AND
        metric.type="custom.googleapis.com/tool/error_rate"
      comparison: COMPARISON_GT
      thresholdValue: 0.05
      duration: 900s
      aggregations:
      - alignmentPeriod: 300s
        perSeriesAligner: ALIGN_MEAN
        crossSeriesReducer: REDUCE_MEAN
        groupByFields:
        - metric.label.tool_name
  
  documentation:
    content: |
      Estas alertas están diseñadas para detectar problemas en sistemas multi-agente
      que no serían visibles con métricas tradicionales.
      
      Acciones recomendadas:
      1. Revisar traces en Cloud Trace para identificar el agente específico
      2. Analizar logs estructurados para entender el patrón de razonamiento
      3. Evaluar si se requiere ajuste de prompts o restricciones de safety
      4. Considerar rollback de cambios recientes en configuración de agentes
  
  notificationChannels:
  - name: projects/my-project/notificationChannels/123456789
    displayName: "#sre-alerts"