ClickHouse for Enterprise AI: Architecture & Integration

import clickhouse_connect

# Connect to production ClickHouse instance
client = clickhouse_connect.get_client(host='localhost', port=8123, username='default', password='')

# Step 1: Create MergeTree table with vector array column
create_table_sql = """
CREATE TABLE IF NOT EXISTS ai_telemetry_spans
(
  span_id UUID,
  service_name String,
  prompt_text String,
  embedding Array(Float32),
  latency_ms UInt32,
  token_cost Float64,
  created_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
ORDER BY (service_name, created_at)
"""
client.command(create_table_sql)
print("ClickHouse ai_telemetry_spans table created.")

# Step 2: Query top 5 most semantically similar spans using built-in cosineDistance
def search_similar_telemetry_spans(query_vector: list[float], top_k: int = 5):
  search_sql = """
  SELECT 
      span_id,
      service_name,
      prompt_text,
      cosineDistance(embedding, {query_vector:Array(Float32)}) AS distance
  FROM ai_telemetry_spans
  WHERE created_at >= now() - INTERVAL 7 DAY
  ORDER BY distance ASC
  LIMIT {top_k:UInt32}
  """
  result = client.query(search_sql, parameters={"query_vector": query_vector, "top_k": top_k})
  return result.result_rows

if __name__ == "__main__":
  dummy_vec = [0.019] * 1536
  matches = search_similar_telemetry_spans(dummy_vec, top_k=5)
  print(f"Retrieved {len(matches)} vector similarity matches from ClickHouse.")