#SundayT1998/chroma-client

#Docs

• docs/CONTRIBUTING.md — setup, the checks a change has to pass, the two test layers, code style, commit and PR conventions.
• docs/chroma-api-coverage.md — which Chroma v2 endpoints the client covers, which it deliberately does not, and why.
• docs/protocol-constraints.md — the rules Chroma imposes that a thin endpoint wrapper gets wrong (columnar records, batch limits, destructive filters, where spelling, short query rows), with the input/output of each and where it is enforced.
• docs/serialization-perf.md — what a large record batch costs to serialize, and why the write path is not streamed.

#Design

• AsyncTransport — asynchronous, with async fn send_async. This is the one the bundled AsyncHttpTransport implements, because moonbitlang/async/http only offers async functions and an async call cannot be made from a synchronous one.

• Transport — synchronous. MemoryTransport implements it, so scripted responses cover the whole request/response cycle in ordinary moon test runs.

• portable — the same code runs on native, wasm and js. The transport limits the library to the backends moonbitlang/async supports; wasm-gc has no async runtime yet, so it is not a supported target;
• testable — MemoryTransport implements the synchronous Transport and replays scripted responses, so the whole request/response cycle is covered by ordinary moon test runs without a server;
• replaceable — AsyncTransport is a two-method-wide seam, so an application that already owns an HTTP stack can plug it in instead of using the bundled one.

#Install

moon add SundayT1998/chroma-client

#Quick start

///|
async fn quick_start() -> Unit raise ChromaError {
  // AsyncHttpTransport is bundled; point the config at your Chroma server.
  let client = ChromaClient::new(
    AsyncHttpTransport::new(),
    config=Config::new(endpoint="http://localhost:8000"),
  )
  let collection = client.get_or_create_collection_async("notes")
  // Chroma 1.x stores vectors, not text: every record needs coordinates.
  client.add_async(collection.id, [
    Record::of_embedding("r1", [1.0, 0.0, 0.0]).with_document(
      "MoonBit is a language for the web",
    ),
    Record::of_embedding("r2", [0.0, 1.0, 0.0]).with_document(
      "Chroma stores embeddings",
    ),
  ])
  let result = client.query_embeddings_async(
    collection.id,
    [[1.0, 0.0, 0.0]],
    n_results=2,
  )
  for id in result.ids {
    println(id)
  }
}

Chroma 1.x has no server-side embedding. An add whose records carry documents but no vectors is rejected by the server, so the client refuses it locally and points at Record::with_embedding. Produce the vectors with whatever embedding model you already use; Record::of_document(...).with_embedding(vec) is the usual shape.

#Choosing what a query returns

///|
async fn query_with_everything(
  client : ChromaClient[AsyncHttpTransport],
  collection_id : String,
) -> Unit raise ChromaError {
  ignore(
    client.query_embeddings_async(
      collection_id,
      [[0.1, 0.9, 0.0]],
      n_results=5,
      include_fields=Include::all(),
    ),
  )
}

#Filtering

///|
async fn filtered_get(
  client : ChromaClient[AsyncHttpTransport],
  collection_id : String,
) -> Unit raise ChromaError {
  let filter = Filter::new()
    .with_metadata("source", MString("handbook"))
    .with_limit(20)
  ignore(client.get_async(collection_id, filter~))
}

///|
let clause = where_and([
  where_eq("topic", MString("ai")),
  where_gt("year", MInt(2020)),
])

///|
let text_clause = where_document_or([
  where_document_contains("moon"),
  where_document_regex("chroma.*client"),
])

#Metadata types

///|
let greeting : MetadataValue = MString("hello")

///|
let answer : MetadataValue = MInt(42)

///|
let ratio : MetadataValue = MFloat(0.5)

///|
let pinned : MetadataValue = MBool(true)

///|
let nothing : MetadataValue = MNull

#Adding a real transport

///|
impl AsyncTransport for MyTransport with fn send_async(self, request) {
  let headers : @ah.Headers = Map([])
  for entry in request.headers {
    let (key, value) = entry
    headers[key] = value
  }
  let payload : &@aio.Data = match request.body {
    Some(body) => body
    None => ""
  }
  let (response, response_body) = @ah.request(
    request.url,
    my_method(request.verb),
    headers,
    payload,
  )
  let out_headers : Map[String, String] = Map([])
  for entry in response.headers {
    let (key, value) = entry
    out_headers[key.to_string().to_lower()] = value
  }
  { status: response.code, headers: out_headers, body: response_body.text(), }
}

///|
pub extend MyTransport with AsyncTransport::{send_async}

#Native transport

moon run cmd/main --target native -- http://localhost:8000 my-collection

#Authentication

///|
let authed_client = ChromaClient::new(
  AsyncHttpTransport::new(),
  config=Config::new()
    .with_token("my-api-token")
    .with_basic_auth("user", "password"),
)

#Timeouts

///|
let timed_client = ChromaClient::new(
  AsyncHttpTransport::new(),
  config=Config::new(endpoint="http://localhost:8000", timeout_ms=500),
)

///|
let timed_sync_client = ChromaClient::new(
  TimeoutTransport::new(MemoryTransport::new(), 5000),
)

///|
let timed_async_client = ChromaClient::new(
  TimeoutAsyncTransport::new(AsyncHttpTransport::new(), 5000),
)

#Counting and previewing

///|
fn count_and_preview(
  client : ChromaClient[MemoryTransport],
  collection_id : String,
) -> Unit raise ChromaError {
  let total = client.count(collection_id)
  let sample = client.peek(collection_id, limit=5)
  println(
    total.to_string() + " records, showing " + sample.ids.length().to_string(),
  )
}

#Renaming and re-tagging a collection

///|
fn rename_collection(
  client : ChromaClient[MemoryTransport],
) -> Unit raise ChromaError {
  client.modify_collection(
    "notes",
    new_name=Some("notes-v2"),
    metadata=Some([("topic", MString("docs"))]),
  )
}

#Pagination

///|
async fn page(
  client : ChromaClient[AsyncHttpTransport],
  collection_id : String,
) -> Unit raise ChromaError {
  ignore(
    client.get_async(
      collection_id,
      filter=Filter::new(limit=Some(100), offset=Some(200)),
      include_fields=Include::get_default(),
    ),
  )
}

#Records and vectors

///|
// Vector first, text attached afterwards.
let with_text = Record::of_embedding("r1", [1.0, 0.0]).with_document("hello")

///|
// Text first, vector attached afterwards.
let with_vector = Record::of_document("r2", "world").with_embedding([0.0, 1.0])

///|
// Metadata, either way.
let with_metadata = Record::of_embedding("r3", [1.0, 1.0]).with_metadata([
  ("source", MString("handbook")),
])

#Errors

• Transport(String) — the request could not be sent;
• Api(ApiError) — the server answered non-2xx with Chroma's error body;
• Status(Int, String) — the server answered non-2xx with a body that is not a Chroma error envelope (a proxy page, say);
• Decode(String) — the response body was not the expected JSON;
• InvalidArgument(String) — the arguments would be rejected by Chroma, so nothing was sent.

///|
fn report(result : Result[QueryResult, ChromaError]) -> Unit {
  match result {
    Err(Transport(message)) => println("could not send: " + message)
    Err(api_error) =>
      if api_error.status_code() == Some(404) {
        println("nothing there")
      } else if api_error.is_retryable() {
        println("429 or 5xx, worth trying again: " + api_error.message())
      } else {
        println(api_error.message())
      }
    Ok(value) => println(value.ids.length().to_string() + " rows")
  }
}

#Retrying

///|
let retrying_client = ChromaClient::new(
  RetryTransport::new(
    AsyncHttpTransport::new(),
    policy=RetryPolicy::new(
      enabled=true,
      max_attempts=3,
      initial_delay_ms=200,
      max_delay_ms=5000,
      backoff_factor=2.0,
    ),
  ),
)

///|
fn retry_with_logging(
  transport : MemoryTransport,
) -> RetryTransport[MemoryTransport] {
  RetryTransport::new(transport, policy=RetryPolicy::new(enabled=true)).on_retry((
      attempt,
      delay_ms,
      error,
    ) => {
      println(
        "retry " +
        attempt.to_string() +
        " in " +
        delay_ms.to_string() +
        "ms: " +
        error.to_string(),
      )
    },
  )
}

#Response size

#Local validation

• an empty records batch on add / upsert / update;
• a record with an empty id, or with neither a document nor an embedding (and, on add / upsert, a document without a vector);
• an empty embeddings list on query_embeddings;
• n_results <= 0, a negative limit or offset, and a delete whose filter names nothing, or whose limit comes without a where clause;
• a write batch that names the same id twice (both positions are reported — upsert is exempt, because writing an id twice is defined there), or whose vectors do not all share one width.

#Decoding rules

• required and missing ⇒ Decode. Collection.id / Collection.name and the ids of a get result are never silently defaulted, because an empty list and a broken response are different answers;
• optional and missing ⇒ empty default. metadata, configuration, log_position, version, and everything include controls (documents, embeddings, metadatas, distances) may legitimately be absent.

#Examples

moon run cmd/main --target native -- http://localhost:8000 my-collection

#Testing

sh scripts/ci.sh

sh scripts/ci.sh fmt check     # just formatting and type-checking
sh scripts/ci.sh test coverage # just tests and coverage

sh scripts/integration.sh      # starts chromadb/chroma, runs, tears down

CHROMA_ENDPOINT=http://localhost:8000 \
  moon test --target native -p SundayT1998/chroma-client

#Alignment with chroma-go

• delete needs a filter. chroma-go's Collection.Delete refuses a call with no ids / where / where_document; this client raises InvalidArgument instead of emptying the collection.
• limit and offset are independent. The offset < limit rule this client used to enforce is gone: chroma-go pages with WithLimit(100), WithOffset(200).
• include is per call. get, get_async and peek send Include::get_default() (documents and metadata) unless told otherwise, and a query sends Include::default() (documents, metadata, distances) — the defaults chroma-go documents. POST /get cannot carry distances (distances are not supported in get), so the query default would be a bug there.
• A query takes ids. Filter::new(ids=[...]) restricts a query to known records, like WithIDs; limit / offset are still dropped there, because a query has no paging in Chroma.
• preflight and identity exist. chroma-go calls PreFlight before a batch and exposes GetIdentity; both are now part of this client, so the limits a server advertises are readable from MoonBit.

#Status

#License