o 3IhK#@s<ddlZddlZddlmZmZmZmZmZddlZddl m Z ddl m Z m Z m Z ddlmZddlmZddlmZddlmZmZmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!m"Z"m#Z#m$Z$m%Z%ddl&m'Z'dee dee fddZ(dee dee)fddZ*edZ+edZ,GdddeZ-dS)N)CallableDictListOptionalTypeVar) overrides) GetResultMetadata QueryResult)System)Executor)Scan) CountPlanGetPlanKNNPlan)convert)QueryExecutorStub)DistributedSegmentManager)OtelInterceptor)RetryCallStateRetryingstop_after_attemptwait_exponential_jitterretry_if_exception)SpanmetadatareturncCsD|sdSi}|D] \}}|ds|||<q t|dkr dS|S)z[Remove any chroma-specific metadata keys that the client shouldn't see from a metadata map.Nzchroma:r)items startswithlen)rresultkvr#g/home/air/sanwanet/gpt-api/venv/lib/python3.10/site-packages/chromadb/execution/executor/distributed.py_clean_metadatas  r%cCs|r d|vr t|dSdS)z-Retrieve the uri (if any) from a Metadata mapz chroma:uriN)strrr#r#r$_uri&s  r(IOcseZdZUejed<eeefed<e ed<e ed<e ed<de ffdd Z d e eegefd ed efd d Zeded e fddZeded efddZeded efddZded e efddZded efddZZS)DistributedExecutor_mtx_grpc_stub_pool_manager_request_timeout_seconds_query_replication_factorsystemcsHt|t|_i|_|t|_|j d|_ |j d|_ dS)N$chroma_query_request_timeout_secondschroma_query_replication_factor) super__init__ threadingLockr,r-requirerr.settingsr/r0)selfr1 __class__r#r$r5:s    zDistributedExecutor.__init__funcsargsrc sd}ddtddffdd }ttdtddd d td d |d D].}dur.d|||t||WdS1sGwY|d7}q"td)z Retry a list of functions in a round-robin fashion until one of them succeeds. funcs: List of functions to retry args: Arguments to pass to each function rN_rcs&ddlm}|dur|ddSdS)Nr)tracerzWaiting to retry RPC) chromadb.telemetry.opentelemetryr@ start_span)r?r@ sleep_spanr#r$ before_sleepQs z.before_sleepg?)jitterTcSs$t|tjo|tjjtjjfvSN) isinstancegrpcRpcErrorcode StatusCode UNAVAILABLEUNKNOWN)xr#r#r$`s z8DistributedExecutor._round_robin_retry..)stopwaitreraiseretryrEz0Unreachable code error - should never reach here)rrrrrendr Exception)r:r=r> attempt_countrEattemptr#rCr$_round_robin_retryFs(    $ z&DistributedExecutor._round_robin_retryplancs:|j}fdd|D}|t|}t|S)Ncg|]}|jqSr#) _get_stubCount.0endpointr:r#r$ sz-DistributedExecutor.count..)_get_grpc_endpointsscanr[rto_proto_count_planfrom_proto_count_result)r:r\ endpoints count_funcs count_resultr#rcr$countps   zDistributedExecutor.countc s|j}fdd|D}|t|}t|}dd|D}|jjr/dd|Dnd}|jjr.cSg|]}|dqS)idr#rarecordr#r#r$rdcSro) embeddingr#rqr#r#r$rdrscSro)documentr#rqr#r#r$rdrscSg|]}t|dqSr'r(rqr#r#r$rdrecSrvr'r%rqr#r#r$rdre)ids embeddings documentsurisdata metadatasincluded) rfrgr[rto_proto_get_planfrom_proto_get_result projectionrtruurirrr) r:r\rj get_funcs get_resultrecordsryrzr{r|r~r#rcr$getys<  zDistributedExecutor.getc s|j}fdd|D}|t|}t|}dd|D}|jjr/dd|Dnd}|jjr.cSg|] }dd|DqS)cSg|]}|ddqS)rrrpr#rqr#r#r$rdre6DistributedExecutor.knn...r#rarr#r#r$rdcSr)cSr)rrrtr#rqr#r#r$rdrerr#rr#r#r$rd cSr)cSr)rrrur#rqr#r#r$rdrerr#rr#r#r$rdrcSr)cSg|] }t|ddqSrrrrwrqr#r#r$rdrr#rr#r#r$rdrcSr)cSrrrxrqr#r#r$rdrrr#rr#r#r$rdrcSr)cSro)distancer#rqr#r#r$rdrsrr#rr#r#r$rdr)ryrzr{r|r}r~ distancesr)rfrgr[rto_proto_knn_planfrom_proto_knn_batch_resultrrtrurrrankr r) r:r\rj knn_funcs knn_resultresultsryrzr{r|r~rr#rcr$knnsV      zDistributedExecutor.knnrgcCs |j|j|j}t||SrH)r. get_endpointsrrr0randomshuffle)r:rg grpc_urlsr#r#r$rfs  z'DistributedExecutor._get_grpc_endpointsgrpc_urlcCsx|j/||jvr&tj|ddgd}tg}tj|g|R}t||j|<|j|WdS1s5wYdS)N)zgrpc.max_concurrent_streamsi)zgrpc.max_receive_message_lengthiH)options)r,r-rJinsecure_channelrintercept_channelr)r:rchannel interceptorsr#r#r$r^s $zDistributedExecutor._get_stub)__name__ __module__ __qualname__r6r7__annotations__rr&rrintr r5rrr)r*r[rrrmrrrrr rr rfr^ __classcell__r#r#r;r$r+3s $ *'9 r+).r6rtypingrrrrrrJrchromadb.api.typesrr r chromadb.configr $chromadb.execution.executor.abstractr &chromadb.execution.expression.operatorr "chromadb.execution.expression.planrrrchromadb.protor&chromadb.proto.query_executor_pb2_grpcr)chromadb.segment.impl.manager.distributedr%chromadb.telemetry.opentelemetry.grpcrtenacityrrrrropentelemetry.tracerr%r&r(r)r*r+r#r#r#r$s*