nusa-embedding-0001) with Cloudflare Vectorize enables ultra-low-latency RAG architectures without maintaining self-hosted vector infrastructure.
Quickstart
import os
import requests
from openai import OpenAI
# Initialize Neosantara client
client = OpenAI(
base_url="https://api.neosantara.xyz/v1",
api_key=os.environ["NEOSANTARA_API_KEY"]
)
CF_ACCOUNT_ID = os.environ["CLOUDFLARE_ACCOUNT_ID"]
CF_API_TOKEN = os.environ["CLOUDFLARE_API_TOKEN"]
INDEX_NAME = "vehicle-manual"
cf_headers = {
"Authorization": f"Bearer {CF_API_TOKEN}",
"Content-Type": "application/json"
}
# 1. Embed user query
query_text = "What is the emergency braking procedure?"
emb_res = client.embeddings.create(
model="nusa-embedding-0001",
input=query_text
)
query_vector = emb_res.data[0].embedding
# 2. Query Cloudflare Vectorize v2 API
query_url = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/vectorize/v2/indexes/{INDEX_NAME}/query"
query_payload = {
"vector": query_vector,
"topK": 1,
"returnValues": False,
"returnMetadata": "all"
}
res = requests.post(query_url, headers=cf_headers, json=query_payload).json()
matched_passage = res["result"]["matches"][0]["metadata"]["text"]
# 3. Grounded generation via chat completion
completion = client.chat.completions.create(
model="deepseek-v4.1-flash",
messages=[
{"role": "system", "content": f"Answer based strictly on this reference:\n\n{matched_passage}"},
{"role": "user", "content": query_text}
]
)
print(completion.choices[0].message.content)
RAG with ChromaDB
Local and in-memory RAG setup using ChromaDB.
Embeddings Endpoint
Specifications for vector dimensions and input parameters.
Implementation Workflow
1
Environment Prerequisites
Install required Python packages:Export necessary credentials:
pip install -U openai requests
export NEOSANTARA_API_KEY="nsk_your_api_key"
export CLOUDFLARE_ACCOUNT_ID="your_cloudflare_account_id"
export CLOUDFLARE_API_TOKEN="your_cloudflare_api_token"
2
Create Vectorize v2 Index
The
nusa-embedding-0001 model outputs 768-dimensional vectors. Initialize the index using cosine distance:def create_vectorize_index(account_id, token, index_name, dimension=768):
url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/vectorize/v2/indexes"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
data = {
"name": index_name,
"config": {
"dimensions": dimension,
"metric": "cosine"
}
}
r = requests.post(url, headers=headers, json=data)
if r.status_code in [200, 201]:
print(f"Index '{index_name}' created successfully.")
elif r.status_code == 409:
print(f"Index '{index_name}' already exists.")
else:
r.raise_for_status()
3
Vectorize & Ingest (NDJSON)
Cloudflare Vectorize v2 accepts batch ingestion via Newline Delimited JSON (NDJSON). Each record contains an ID, values array, and metadata payload:
def insert_documents(account_id, token, index_name, documents):
# 1. Generate embeddings via Neosantara
emb_res = client.embeddings.create(
model="nusa-embedding-0001",
input=documents
)
# 2. Build NDJSON payload
ndjson_lines = []
for i, doc_text in enumerate(documents):
record = {
"id": f"doc_{i}",
"values": emb_res.data[i].embedding,
"metadata": {"text": doc_text}
}
ndjson_lines.append(json.dumps(record))
ndjson_payload = "\n".join(ndjson_lines)
# 3. Stream to Cloudflare insert endpoint
url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/vectorize/v2/indexes/{index_name}/insert"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/x-ndjson"
}
res = requests.post(url, headers=headers, data=ndjson_payload)
res.raise_for_status()
print(f"Indexed {len(documents)} documents into Vectorize.")
4
Query & Grounded Generation
Retrieve semantic matches from Vectorize and pass them into Neosantara’s chat completions API:
def query_rag(query_text: str):
# 1. Embed query
q_emb = client.embeddings.create(
model="nusa-embedding-0001",
input=query_text
).data[0].embedding
# 2. Query Vectorize
url = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/vectorize/v2/indexes/{INDEX_NAME}/query"
r = requests.post(url, headers=cf_headers, json={
"vector": q_emb,
"topK": 2,
"returnMetadata": "all"
}).json()
matches = r.get("result", {}).get("matches", [])
if not matches:
return "No relevant context found."
context = "\n---\n".join([m["metadata"]["text"] for m in matches])
# 3. Chat completion
ans = client.chat.completions.create(
model="deepseek-v4.1-flash",
messages=[
{"role": "system", "content": "Provide factual answers strictly derived from the reference context."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query_text}"}
],
temperature=0.1
)
return ans.choices[0].message.content
Complete Production Script
rag_vectorize_complete.py
import os
import json
import requests
from openai import OpenAI
NEOSANTARA_API_KEY = os.environ["NEOSANTARA_API_KEY"]
CF_ACCOUNT_ID = os.environ["CLOUDFLARE_ACCOUNT_ID"]
CF_API_TOKEN = os.environ["CLOUDFLARE_API_TOKEN"]
INDEX_NAME = "neosantara-rag-kb"
client = OpenAI(
base_url="https://api.neosantara.xyz/v1",
api_key=NEOSANTARA_API_KEY
)
docs = [
"SLA Guidelines: Neosantara maintains 99.95% availability across regional Jakarta clusters.",
"Rate Limits: Free tier defaults to 10 RPM, 30,000 ITPM, and 8,000 OTPM.",
"Billing: Credit top-ups are processed via Mayar QRIS and Virtual Accounts in Indonesian Rupiah."
]
def main():
# 1. Create Index
url_create = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/vectorize/v2/indexes"
headers = {"Authorization": f"Bearer {CF_API_TOKEN}", "Content-Type": "application/json"}
requests.post(url_create, headers=headers, json={
"name": INDEX_NAME,
"config": {"dimensions": 768, "metric": "cosine"}
})
# 2. Ingest Data
emb_res = client.embeddings.create(model="nusa-embedding-0001", input=docs)
ndjson = "\n".join([
json.dumps({"id": f"doc_{i}", "values": emb_res.data[i].embedding, "metadata": {"text": t}})
for i, t in enumerate(docs)
])
url_insert = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/vectorize/v2/indexes/{INDEX_NAME}/insert"
requests.post(url_insert, headers={"Authorization": f"Bearer {CF_API_TOKEN}", "Content-Type": "application/x-ndjson"}, data=ndjson)
# 3. Query
question = "What are the Free tier rate limits on Neosantara?"
q_vec = client.embeddings.create(model="nusa-embedding-0001", input=question).data[0].embedding
url_query = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/vectorize/v2/indexes/{INDEX_NAME}/query"
matches = requests.post(url_query, headers=headers, json={"vector": q_vec, "topK": 1, "returnMetadata": "all"}).json()["result"]["matches"]
passage = matches[0]["metadata"]["text"]
response = client.chat.completions.create(
model="deepseek-v4.1-flash",
messages=[
{"role": "system", "content": "Provide a concise answer based strictly on the reference."},
{"role": "user", "content": f"Reference:\n{passage}\n\nQuestion: {question}"}
]
)
print("AI Answer:", response.choices[0].message.content)
if __name__ == "__main__":
main()
Because Cloudflare Vectorize stores metadata payloads alongside each vector record (
"metadata": {"text": doc_text}), you do not need an external relational database to resolve vector IDs back to raw document strings.