a `g@sdZddlmZddlZddlmZddlZddlZddl Z ddl Z ddl m Z m Z mZmZmZmZmZmZmZmZmZmZmZmZddlZddlmZddlmZmZddlmZddlm Z!dd l"m#Z$dd l%m&Z&dd l'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:dd l;mZ>m?Z?e rfddl@ZAdd lBmCZCeAjDZDne ZDeEeFZGeeeHgeeHfeeHeHgeeHffZIdJddddddddddddddddddZJdKdddddddd!dd" d#d$ZKdLd%dddddddddddd&dd'd(d)ZLGd*d+d+e-ZMGd,ddZNdMd-d.d/d0d1dd2d3d4d5ZOd6d-d7d8d9ZPd d:d%d1dd;d<d=d>ZQed?ZRd@dAdBdCdDZSdEdFdGdHdIZTdS)NzV2 Evaluation Interface.) annotationsN) TYPE_CHECKINGAny AsyncIterable AsyncIterator AwaitableCallableDictIterableListOptionalSequenceTypeVarUnioncast) run_helpers) run_treesschemas)r)utils)_aiter) _warn_once) AEVALUATOR_TDATA_T EVALUATOR_TExperimentResultRow_evaluators_include_attachments_ExperimentManagerMixin_extract_feedback_keys_ForwardResults_include_attachments_is_langchain_runnable_load_examples_map_load_experiment _load_tqdm _load_traces _resolve_data_resolve_evaluators_resolve_experiment _to_pandas_wrap_summary_evaluators)SUMMARY_EVALUATOR_TEvaluationResultEvaluationResults RunEvaluator)RunnableTzNUnion[DATA_T, AsyncIterable[schemas.Example], Iterable[schemas.Example], None]z4Optional[Sequence[Union[EVALUATOR_T, AEVALUATOR_T]]]z'Optional[Sequence[SUMMARY_EVALUATOR_T]]Optional[dict] Optional[str] Optional[int]intOptional[langsmith.Client]boolz6Optional[Union[schemas.TracerSession, str, uuid.UUID]]zVUnion[ATARGET_T, AsyncIterable[dict], Runnable, str, uuid.UUID, schemas.TracerSession]rAsyncExperimentResults)data evaluatorssummary_evaluatorsmetadataexperiment_prefix descriptionmax_concurrencynum_repetitionsclientblocking experimentupload_resultstargetkwargsreturnc stt|ttjtjfr|dkt| | t|t|d}t|rhdt dd| Dd}t |t|ttjfr||n|j }t d|dt|f||||| | d | Id HSt|tt frd }t |n| rd | d }t |n|sd}t |nn| r(|r(d| d|}t |nH| s6tdt d|dt|||||||||| | | | d Id HSd S)a%Evaluate an async target system on a given dataset. Args: target (AsyncCallable[[dict], dict] | AsyncIterable[dict] | Runnable | EXPERIMENT_T | Tuple[EXPERIMENT_T, EXPERIMENT_T]): The target system or experiment(s) to evaluate. Can be an async function that takes a dict and returns a dict, a langchain Runnable, an existing experiment ID, or a two-tuple of experiment IDs. data (Union[DATA_T, AsyncIterable[schemas.Example]]): The dataset to evaluate on. Can be a dataset name, a list of examples, an async generator of examples, or an async iterable of examples. evaluators (Optional[Sequence[EVALUATOR_T]]): A list of evaluators to run on each example. Defaults to None. summary_evaluators (Optional[Sequence[SUMMARY_EVALUATOR_T]]): A list of summary evaluators to run on the entire dataset. Defaults to None. metadata (Optional[dict]): Metadata to attach to the experiment. Defaults to None. experiment_prefix (Optional[str]): A prefix to provide for your experiment name. Defaults to None. description (Optional[str]): A description of the experiment. max_concurrency (int | None): The maximum number of concurrent evaluations to run. If None then no limit is set. If 0 then no concurrency. Defaults to 0. num_repetitions (int): The number of times to run the evaluation. Each item in the dataset will be run and evaluated this many times. Defaults to 1. client (Optional[langsmith.Client]): The LangSmith client to use. Defaults to None. blocking (bool): Whether to block until the evaluation is complete. Defaults to True. experiment (Optional[schemas.TracerSession]): An existing experiment to extend. If provided, experiment_prefix is ignored. For advanced usage only. load_nested: Whether to load all child runs for the experiment. Default is to only load the top-level root runs. Should only be specified when evaluating an existing experiment. Returns: AsyncIterator[ExperimentResultRow]: An async iterator over the experiment results. Environment: - LANGSMITH_TEST_CACHE: If set, API calls will be cached to disk to save time and cost during testing. Recommended to commit the cache files to your repository for faster CI/CD runs. Requires the 'langsmith[vcr]' package to be installed. Examples: >>> from typing import Sequence >>> from langsmith import Client, aevaluate >>> from langsmith.schemas import Example, Run >>> client = Client() >>> dataset = client.clone_public_dataset( ... "https://smith.langchain.com/public/419dcab2-1d66-4b94-8901-0357ead390df/d" ... ) >>> dataset_name = "Evaluate Examples" Basic usage: >>> def accuracy(run: Run, example: Example): ... # Row-level evaluator for accuracy. ... pred = run.outputs["output"] ... expected = example.outputs["answer"] ... return {"score": expected.lower() == pred.lower()} >>> def precision(runs: Sequence[Run], examples: Sequence[Example]): ... # Experiment-level evaluator for precision. ... # TP / (TP + FP) ... predictions = [run.outputs["output"].lower() for run in runs] ... expected = [example.outputs["answer"].lower() for example in examples] ... # yes and no are the only possible answers ... tp = sum([p == e for p, e in zip(predictions, expected) if p == "yes"]) ... fp = sum([p == "yes" and e == "no" for p, e in zip(predictions, expected)]) ... return {"score": tp / (tp + fp)} >>> import asyncio >>> async def apredict(inputs: dict) -> dict: ... # This can be any async function or just an API call to your app. ... await asyncio.sleep(0.1) ... return {"output": "Yes"} >>> results = asyncio.run( ... aevaluate( ... apredict, ... data=dataset_name, ... evaluators=[accuracy], ... summary_evaluators=[precision], ... experiment_prefix="My Experiment", ... description="Evaluate the accuracy of the model asynchronously.", ... metadata={ ... "my-prompt-version": "abcd-1234", ... }, ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... Evaluating over only a subset of the examples using an async generator: >>> async def example_generator(): ... examples = client.list_examples(dataset_name=dataset_name, limit=5) ... for example in examples: ... yield example >>> results = asyncio.run( ... aevaluate( ... apredict, ... data=example_generator(), ... evaluators=[accuracy], ... summary_evaluators=[precision], ... experiment_prefix="My Subset Experiment", ... description="Evaluate a subset of examples asynchronously.", ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... Streaming each prediction to more easily + eagerly debug. >>> results = asyncio.run( ... aevaluate( ... apredict, ... data=dataset_name, ... evaluators=[accuracy], ... summary_evaluators=[precision], ... experiment_prefix="My Streaming Experiment", ... description="Streaming predictions for debugging.", ... blocking=False, ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... >>> async def aenumerate(iterable): ... async for elem in iterable: ... print(elem) >>> asyncio.run(aenumerate(results)) Running without concurrency: >>> results = asyncio.run( ... aevaluate( ... apredict, ... data=dataset_name, ... evaluators=[accuracy], ... summary_evaluators=[precision], ... experiment_prefix="My Experiment Without Concurrency", ... description="This was run without concurrency.", ... max_concurrency=0, ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... Using Async evaluators: >>> async def helpfulness(run: Run, example: Example): ... # Row-level evaluator for helpfulness. ... await asyncio.sleep(5) # Replace with your LLM API call ... return {"score": run.outputs["output"] == "Yes"} >>> results = asyncio.run( ... aevaluate( ... apredict, ... data=dataset_name, ... evaluators=[helpfulness], ... summary_evaluators=[precision], ... experiment_prefix="My Helpful Experiment", ... description="Applying async evaluators example.", ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... .. versionchanged:: 0.2.0 'max_concurrency' default updated from None (no limit on concurrency) to 0 (no concurrency at all). r/)r>rArBr;r7zReceived invalid arguments. css|]\}}|r|VqdSN).0kvrGrGk/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/langsmith/evaluation/_arunner.py zaevaluate..z? should not be specified when target is an existing experiment.z,Running evaluation over existing experiment z...)r8r9r:r=r?r@NzRunning a comparison of two existing experiments asynchronously is not currently supported. Please use the `evaluate()` method instead and make sure that your evaluators are defined as synchronous functions.zReceived unsupported arguments zC. These arguments are not supported when creating a new experiment.zDMust specify 'data' when running evaluations over a target function.zeExpected at most one of 'experiment' or 'experiment_prefix', but both were provided. Got: experiment=z, experiment_prefix=z&'upload_results' parameter is in beta.z&Running evaluation over target system ) r7r8r9r:r;r<r=r>r?r@rArB) isinstancestruuidUUIDr TracerSessionr5anyvaluestupleitems ValueErroridloggerdebugaevaluate_existinglistr _aevaluate)rCr7r8r9r:r;r<r=r>r?r@rArBrDZ invalid_argsmsgZ target_idrGrGrK aevaluateNs@        r_Fz,Union[str, uuid.UUID, schemas.TracerSession]) r8r9r:r=r? load_nestedr@rArEc s|p t}t|tjr|ntt||IdH}tjt|||dIdH} tt ||IdHfdd| D} t | | |||||||d IdHS)a Evaluate existing experiment runs asynchronously. Args: experiment (Union[str, uuid.UUID]): The identifier of the experiment to evaluate. evaluators (Optional[Sequence[EVALUATOR_T]]): Optional sequence of evaluators to use for individual run evaluation. summary_evaluators (Optional[Sequence[SUMMARY_EVALUATOR_T]]): Optional sequence of evaluators to apply over the entire dataset. metadata (Optional[dict]): Optional metadata to include in the evaluation results. max_concurrency (int | None): The maximum number of concurrent evaluations to run. If None then no limit is set. If 0 then no concurrency. Defaults to 0. client (Optional[langsmith.Client]): Optional Langsmith client to use for evaluation. load_nested: Whether to load all child runs for the experiment. Default is to only load the top-level root runs. blocking (bool): Whether to block until evaluation is complete. Returns: AsyncIterator[ExperimentResultRow]: An async iterator over the experiment results. Examples: Define your evaluators >>> from typing import Sequence >>> from langsmith.schemas import Example, Run >>> def accuracy(run: Run, example: Example): ... # Row-level evaluator for accuracy. ... pred = run.outputs["output"] ... expected = example.outputs["answer"] ... return {"score": expected.lower() == pred.lower()} >>> def precision(runs: Sequence[Run], examples: Sequence[Example]): ... # Experiment-level evaluator for precision. ... # TP / (TP + FP) ... predictions = [run.outputs["output"].lower() for run in runs] ... expected = [example.outputs["answer"].lower() for example in examples] ... # yes and no are the only possible answers ... tp = sum([p == e for p, e in zip(predictions, expected) if p == "yes"]) ... fp = sum([p == "yes" and e == "no" for p, e in zip(predictions, expected)]) ... return {"score": tp / (tp + fp)} Load the experiment and run the evaluation. >>> from langsmith import aevaluate, aevaluate_existing >>> dataset_name = "Evaluate Examples" >>> async def apredict(inputs: dict) -> dict: ... # This can be any async function or just an API call to your app. ... await asyncio.sleep(0.1) ... return {"output": "Yes"} >>> # First run inference on the dataset ... results = asyncio.run( ... aevaluate( ... apredict, ... data=dataset_name, ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... Then evaluate the results >>> experiment_name = "My Experiment:64e6e91" # Or manually specify >>> results = asyncio.run( ... aevaluate_existing( ... experiment_name, ... evaluators=[accuracy], ... summary_evaluators=[precision], ... ) ... ) # doctest: +ELLIPSIS View the evaluation results for experiment:... N)r`csg|]}|jqSrG)reference_example_id)rHrunZdata_maprGrK rMz&aevaluate_existing..)r7r8r9r:r=r?r@rA) rget_cached_clientrNrrR aitertools aio_to_threadr"r$r!r]) rAr8r9r:r=r?r`r@projectrunsr7rGrcrKr[Ts*P   r[-Union[DATA_T, AsyncIterable[schemas.Example]]zFUnion[ATARGET_T, AsyncIterable[dict], Iterable[schemas.Run], Runnable])r7r8r9r:r;r<r=r>r?r@rArBrCrEc  st|p(t|dr"t|p(t|} | p4t} | r>dntt t j |}t t| || IdH\}}t|| ||pv||||t|pt|dk|tt|t|dk| d IdH}td}|dur|IdH}t||d}nd}tj|| jgd| rd|r2|jtt|||dIdH}n|jtt||dIdH}|r||IdH}n0|r~|j||dIdH}|r||IdH}t |}| r|!IdH|WdS1s0YdS)N __aiter__rr/) r?r:rAr<r>riinclude_attachmentsreuse_attachmentsrBz.yaml)Z ignore_hostsr=)"asyncioiscoroutinefunctionhasattr iscoroutinerkr rtrerr rRunrfrgr'_AsyncExperimentManagerrrr3astartls_utilsZ get_cache_dirget_dataset_idpathlibPathZwith_optional_cacheZapi_url awith_predictions_and_evaluators ATARGET_Tawith_predictionsawith_summary_evaluatorsawith_evaluatorsr6wait)rCr7r8r9r:r;r<r=r>r?r@rArBZis_async_targetriZ experiment_manager cache_dirZdsid cache_pathresultsrGrGrKr]st        r]cseZdZdZdTdddd d d d d d d d ddd fdd ZdddddZddddZddddZddddZd dd!d"Z ddd#d$Z dddd%d&Z dUd'd(d)dd*d+d,Z dVd'd(dd-d.d/Z dd0d)d'dd1d2d3Zd4dd5d6d7Zd8dd9d:Zd;ddd?d@dAZdXdBd'd8d1dCdDZdBdEdFdEdGdHdIZd4d d5dJdKZd ddLdMZdNddOdPZdQddRdSZZS)YruaManage the execution of experiments asynchronously. Supports lazily running predictions and evaluations in parallel to facilitate result streaming and early debugging. Args: data (DATA_T): The data used for the experiment. Can be a dataset name or ID OR a generator of examples. runs (Optional[Iterable[schemas.Run]]): The runs associated with the experiment predictions. experiment (Optional[schemas.TracerSession]): The tracer session associated with the experiment. experiment_prefix (Optional[str]): The prefix for the experiment name. description (Optional[str]): The description for the experiment. metadata (Optional[dict]): Additional metadata for the experiment. client (Optional[langsmith.Client]): The Langsmith client used for the experiment. evaluation_results (Optional[Iterable[EvaluationResults]]): The evaluation sresults for the experiment. summary_results (Optional[Iterable[EvaluationResults]]): The aggregate results for the experiment. num_repetitions (Optional[int], default=1): The number of repetitions for the experiment. include_attachments (Optional[bool], default=False): Whether to include attachments. This is used for when we pull the examples for the experiment. reuse_attachments (Optional[bool], default=False): Whether to reuse attachments from examples. This is True if we need to reuse attachments across multiple target/evaluator functions. upload_results (Optional[bool], default=True): Whether to upload results to Langsmith. attachment_raw_data_dict (Optional[dict]): A dictionary to store raw data for attachments. Only used if we reuse attachments across multiple target/evaluator functions. Nr/FTz+Optional[Union[schemas.TracerSession, str]]r0zBOptional[Union[Iterable[schemas.Run], AsyncIterable[schemas.Run]]]r4z*Optional[AsyncIterable[EvaluationResults]]r1r3r5rj) rAr:rir?evaluation_resultssummary_resultsr<r>rlrmrBattachment_raw_data_dictr7csftj||||d||_d|_|dur2t|nd|_||_||_| |_ | |_ | |_ | |_ | |_ dS)N)rAr:r?r<)super__init___data _examplesrfensure_async_iterator_runs_evaluation_results_summary_results_num_repetitionsr_reuse_attachments_upload_results_attachment_raw_data_dict)selfr7rAr:rir?rrr<r>rlrmrBr __class__rGrKr/s"z _AsyncExperimentManager.__init__schemas.Example)examplerEcCst|dr|js|Si}|jD]^\}}|jdurxt|j||jvrx|dt|jt|j||dd||<q"|||<q"tj |j|j |j |j |j |j|j|j|j||j|jd S)aReset attachment readers for an example. This is only in the case that an attachment is going to be used by more than 1 callable (target + evaluators). In that case we keep a single copy of the attachment data in self._attachment_raw_data_dict, and create readers from that data. This makes it so that we don't have to keep copies of the same data in memory, instead we can just create readers from the same data. attachmentsN presigned_url mime_typerreaderr rX created_at dataset_idinputsoutputsr: modified_atri source_run_idr _host_url _tenant_id)rqrrVrrOrXioBytesIOrExamplerrrrr:rrirrr)rrnew_attachmentsname attachmentrGrGrK_reset_example_attachmentsSs:   z2_AsyncExperimentManager._reset_example_attachmentsAsyncIterator[schemas.Example]rEcsjdurtjjjd_jrXjdurXtj\}_dd|2IdH_j dkrddj2IdHt fddt j D_tjt jdt d \_}|S) Nr?rlcsFi|z<3dHW}|jpiD] \}}t|j||dqq6S)Nr)rrVrOrXread)rHervaluerGrGrK sz9_AsyncExperimentManager.aget_examples..r/csg|z3dHW}|q6SrFrGrHrrGrGrKrdrMz9_AsyncExperimentManager.aget_examples..cs"g|]}tfddDqS)csg|]}|qSrG)rrrrGrKrdszD_AsyncExperimentManager.aget_examples...)async_iter_from_list)rH_Z examples_listrrGrKrds  lock)r_aresolve_datarr?rrrrfateerasync_chain_from_iterablerangerroLock)rZ examples_copyZ examples_iterrGrrK aget_exampless,     z%_AsyncExperimentManager.aget_examplesrOcsX|jdust|jddsLt|IdHIdH}|durBtdt|jSt|jjS)Nreference_dataset_idz!No examples found in the dataset.) _experimentgetattrrfpy_anextrrWrOrr)rrrGrGrKrxs  z&_AsyncExperimentManager.get_dataset_idzAsyncIterator[schemas.Run]cCsR|jdurtdtjt|jdtd\|_}|2z3dHW}|Vq86dS)NzRuns not loaded yet.rr)rrWrfrrror)rrirbrGrGrK aget_runss  z!_AsyncExperimentManager.aget_runsz AsyncIterator[EvaluationResults]cCst|jdur4|IdH2z3dHW}dgiVq6n) rAr:r?rirrlrmrBr)rfrrStopAsyncIterationrWrZ _get_projectZ_print_experiment_startr _metadatarr?rrrrr)rZ first_examplerhrGrGrKrvs2     z_AsyncExperimentManager.astartcCsi}|jp iD]b\}}|jdurlt|j||jvrlt|jt|j|}|d||dd||<q|||<qtj|j|j |j |j |j |j |j|j|j||j|jd S)Nrrrr)rrVrrOrXrrrrrrrrr:rrirrr)rrrrrrrGrGrK_get_example_with_readerss8  z1_AsyncExperimentManager._get_example_with_readersr2r|z*Sequence[Union[EVALUATOR_T, AEVALUATOR_T]])r=rCr8rEc sttds tjdd_fdd}tj|dd}tj|dt d \}}}t d d |2j j j d d |2d d |2jjjd S)zRun predictions and evaluations in a single pipeline. This allows evaluators to process results as soon as they're available from the target function, rather than waiting for all predictions to complete first. _evaluator_executor max_workerscs^jtd2zB3dHW}|d|d}}j||dgidjd}|Vq6dS)zCreate a single task per example. That task is to run the target function and all the evaluators sequentially. r=rlNrrbrrbrrexecutor) _apredictr_arun_evaluatorsr)predrrbrr8r=rrCrGrKprocess_exampless zR_AsyncExperimentManager.awith_predictions_and_evaluators..process_examplesMbP?Z_eager_consumption_timeoutrcSs |z3dHW}|dVq6dSNrrGrHrrGrGrKrL3rMzK_AsyncExperimentManager.awith_predictions_and_evaluators..cSs |z3dHW}|dVq6dSNrbrGrrGrGrKrL7rMcSs |z3dHW}|dVq6dSNrrGrrGrGrKrL8rMrAr:r?rirrrlrB)r&rqcfThreadPoolExecutorrrfaiter_with_concurrencyrrorrurrr?rrr) rrCr8r=rexperiment_resultsr1r2r3rGrrKr{s*     z8_AsyncExperimentManager.awith_predictions_and_evaluators)r=rCrEc s`|j||t|d}tj|dtd\}}tdd|2|j|j|j dd|2|j|j dS)NrrrcSs |z3dHW}|dVq6dSrrGrHrrGrGrKrLKrMz<_AsyncExperimentManager.awith_predictions..cSs |z3dHW}|dVq6dSrrGrrGrGrKrLOrM)rAr:r?rirlrB) rrrfrrorrurrr?r)rrCr=Z_experiment_resultsrrrGrGrKr}>s  z)_AsyncExperimentManager.awith_predictionsrn)r8r=rEc stt|}|j||d}tj|dtd\}}}tdd|2|j|j|j dd|2dd|2|j |j |j d S) NrnrrcSs |z3dHW}|dVq6dSrrGrrGrGrKrL^rMz;_AsyncExperimentManager.awith_evaluators..cSs |z3dHW}|dVq6dSrrGrrGrGrKrLbrMcSs |z3dHW}|dVq6dSrrGrrGrGrKrLcrMr) r&_ascorerfrrorrurrr?rrr)rr8r=rrrrrGrGrKrTs   z(_AsyncExperimentManager.awith_evaluatorszSequence[SUMMARY_EVALUATOR_T])r9rEc sFt|}||}t|IdH|j|j|j||j||j |j d S)Nr) r)_aapply_summary_evaluatorsrurrrr?rrrr)rr9Zwrapped_evaluatorsZaggregate_feedback_genrGrGrKr~is  z0_AsyncExperimentManager.awith_summary_evaluators"AsyncIterator[ExperimentResultRow]cCsJt||IdH|2z"3dHW\}}}t|||dVq 6dS)Nr)rf async_ziprrrr)rrbrrrGrGrK aget_results{sz$_AsyncExperimentManager.aget_resultszDict[str, List[dict]]cs,|jdurdgiSddd|j2IdHiS)Nrcs(g|z3dHW}|dD]}|qq6S)NrrG)rHrresrGrGrKrds z?_AsyncExperimentManager.aget_summary_scores..)rrrGrGrKaget_summary_scoress   z+_AsyncExperimentManager.aget_summary_scoreszAsyncIterator[_ForwardResults])r=rlrCrEcsRt|fdd}tj||dd2z3dHW}|Vq*6IdHdS)NcsBIdH2z,3dHW}t|jjjVq6dSrF)r _aforwardrexperiment_namerr?)rfnrlrrGrK predict_allsz6_AsyncExperimentManager._apredict..predict_allrr)_ensure_async_traceablerfr_aend)rrCr=rlrrrGrrKrs  z!_AsyncExperimentManager._apredictzSequence[RunEvaluator]cshtjddHfdd}tj||dd2z3dHW}|Vq06Wdn1sZ0YdS)Nrrcs.2z3dHW}j|dVq6dS)Nr)rr)current_resultsr8rrrGrK score_allsz2_AsyncExperimentManager._ascore..score_allrr)rrrfr)rr8r=rrrGrrKrs z_AsyncExperimentManager._ascorerzcf.ThreadPoolExecutor)r8rrrEc st}i|dpidji}tjfii|d|js>dndjd|d|d|d }fd d }g}|D]} ||| IdHq|D]} | dur|d | qt|d WdS1s0YdS)Nr:rAr8localT project_namer:enabledr?rbrrc s(zD|jdIdH}j|}jr@jj|d|WSty"zzXt|}tfdd|Dd}j|}jrjj|d|WWYdSty}zt d|WYd}~n d}~00t j dt |dj d t d d WYdn d00dS) Nrbr)rbZ _executorcs&g|]}t|jtddidqS)errorT)keyrcommentextra)r+rXrepr)rHr)rrbrGrKrdsz[_AsyncExperimentManager._arun_evaluators.._run_single_evaluator..)rzError parsing feedback keys: zError running evaluator z on run : Texc_info)Z aevaluate_runrr?_select_eval_resultsrZ_log_evaluation_feedback Exceptionrr,rYrZrrrX) evaluatorZevaluator_responseZselected_resultsZ feedback_keysZerror_responsee2rrrbr)rrK_run_single_evaluatorsP    zG_AsyncExperimentManager._arun_evaluators.._run_single_evaluatorrr) rhget_tracing_contextrtracing_contextrr?appendextendr) rr8rrcurrent_contextr:Z eval_resultsr Z all_resultsr rrGr rKrs>   /z(_AsyncExperimentManager._arun_evaluatorsc Csgg}}t|IdH}t||2z$3dHW\}}||||q.6g}|jrj|jnd}t } i| dpi|j |d} t j fii| d| |jsdnd|j d|D]} z| ||} |j j| | jd} || |jrH| D]H}|jdhd }|d d}tj|j jfi|d||d IdHqWqty}z*tjd t| d |ddWYd}~qd}~00qWdn1s0Yd|iVdS)Nr:)rAZ experiment_idr8rTr)fn_nameZ target_run_id)excludeevaluator_info)Zrun_id project_idZ source_infoz Error running summary evaluator rrr)rfrrrrrrZ_get_experimentrXrrrrr?r__name__rdictpoprgZcreate_feedbackr rYrr)rr9riexamplesZasync_examplesrbrZaggregate_feedbackrrr:r Zsummary_eval_resultZflattened_resultsrZfeedbackrrrGrGrKrsl        >z2_AsyncExperimentManager._aapply_summary_evaluatorscsTg}|IdH2z3dHW}|jr||jq6|r@t|nd}|rP|SdSrF)rrrmax isoformat)rrrZmax_modified_atrGrGrK_get_dataset_versionLs z,_AsyncExperimentManager._get_dataset_versionzOptional[list[str]]cs~t}|IdH2z^3dHW}|jrh|jdrht|jdtrh|jdD]}t|trL||qLq|dq6t|S)NZ dataset_splitbase)setrr:getrNr\rOadd)rZsplitsrsplitrGrGrK_get_dataset_splitsWs  z+_AsyncExperimentManager._get_dataset_splitsNonecsl|js dS|j}|dur td|}|IdH|d<|IdH|d<|jj|ji|j |ddS)NzExperiment not started yet.Zdataset_versionZdataset_splits)r:) rrrWZ_get_experiment_metadatarr$r?Zupdate_projectrXr:)rrAZproject_metadatarGrGrKrgs z_AsyncExperimentManager._aend) NNNNNNNr/FFTN)N)N)NF)N)r __module__ __qualname____doc__rrrrxrrrvrr{r}rr~rrrrrrrr$r __classcell__rGrGrrKru sP',$-!   &C T7 ruc@seZdZddddZeddddZd dd d Zd dd dZdddddZd#ddddddZ ddddZ ddddZ dddd Z ddd!d"Z dS)$r6ru)experiment_managercCs4||_g|_t|_t||j|_d|_dS)Nr) _manager_resultsror_lock create_task _process_data_task_processed_count)rr*rGrGrKr{s  zAsyncExperimentResults.__init__rOrcCs|jjSrF)r+rrrGrGrKrsz&AsyncExperimentResults.experiment_namercCs|SrFrGrrGrGrKrksz AsyncExperimentResults.__aiter__rc sdddfdd }j4IdHfjtjkrfjj}jd7_|WdIdHSjrttWdIdHq1IdHs0Yttj |tjddIdHqdS)Nr3r%)indexrEcs j|krtdIdHqdS)Ng?)r1rosleep)r2rrGrK_wait_until_indexs z;AsyncExperimentResults.__anext__.._wait_until_indexr/)timeout) r-r1lenr,r0donerroshieldwait_for)rr4rrGrrK __anext__s  .z AsyncExperimentResults.__anext__r%)rrEc st}||2zR3dHW}|j4IdH"|j|WdIdHq1IdHsZ0Yq6|IdH}|j4IdH||_WdIdHq1IdHs0YdSrF)r#rr-r,rrr)rrZtqdmitemZsummary_scoresrGrGrKr/s:z$AsyncExperimentResults._process_datarNr2 DataFrame)startendrEcCst|j||dS)N)r=r>)r(r,)rr=r>rGrGrK to_pandassz AsyncExperimentResults.to_pandascCs:ddl}|jr.|jdr.|dd}|S|SdS)Nrpandas)importlib.utilr,util find_specr? _repr_html___repr__)r importlibZdfrGrGrKrEs  z"AsyncExperimentResults._repr_html_r3cCs t|jSrF)r6r,rrGrGrK__len__szAsyncExperimentResults.__len__cCsd|jdS)Nz)rrrGrGrKrFszAsyncExperimentResults.__repr__cs|jIdHdSrF)r0rrGrGrKrszAsyncExperimentResults.wait)rN)rr&r'rpropertyrrkr:r/r?rErHrFrrGrGrGrKr6zs   z,rh.SupportsLangsmithExtra[[dict], Awaitable]rrOrzlangsmith.Clientr)rrrr:r?rlrEc sddddfdd }tjddzd|r8|j|jfn|jf}||dtj|j||i|d |jrj|jn|ji|d iIdHWn:t y}z"t j d |dd d WYd}~n d}~00t t tj|dWdS1s0YdS)Nzrun_trees.RunTreer%)rrEcs|dSrFrG)rKrbrGrK_get_runsz_aforward.._get_runT)rZlangsmith_extraZexample_version)raZon_endrr:r?zError running target function: r/)r stacklevelr)rrrrZLangSmithExtrarXrrrr rYrrrrrt) rrrr:r?rlrMargsrrGrLrKrsD   rr|)rCrEcCs\t|s,t|s,t|r$tdntdt|r:|St|rH|j}tjdd|SdS)NzTarget must be an async function. For sync functions, use evaluate. Example usage: async def predict(inputs: dict) -> dict: # do work, like chain.invoke(inputs) return {...} await aevaluate(predict, ...)zTarget must be a callable async function. Received a non-callable object. Example usage: async def predict(inputs: dict) -> dict: # do work, like chain.invoke(inputs) return {...} await aevaluate(predict, ...)Z AsyncTarget)r) rorpr callablerWrZis_traceable_functionZainvokeZ traceable)rCrGrGrKrs  r)rlr)r7r?rlrEcCs(t|trt|Stt|||dS)z*Return the examples for the given dataset.r)rNrrfrr%)r7r?rlrGrGrKr s    rTzIterable[AsyncIterable[T]]zAsyncIterator[T])iterablerEcCs(|D]}|2z3dHW}|Vq 6qdS)zChain multiple async iterables.NrG)rRZ sub_iterabler;rGrGrKrsrzList[schemas.Example]zAsyncIterable[schemas.Example])rrEcCs|D] }|VqdS)z0Convert a list of examples to an async iterable.NrG)rrrGrGrKr'sr) NNNNNNrr/NTNT)NNNrNFT) NNNNNNr/NTNT)F)Ur( __future__rroconcurrent.futuresfuturesrrloggingryrPtypingrrrrrrr r r r r rrrZ langsmithrrrrrsrrwZlangsmith._internalrrfZ#langsmith._internal._beta_decoratorrZlangsmith.evaluation._runnerrrrrrrrrrr r!r"r#r$r%r&r'r(r)Zlangsmith.evaluation.evaluatorr*r+r,r-r@pdZlangchain_core.runnablesr.r< getLoggerrrYrr|r_r[r]rur6rrrrQrrrGrGrGrKs  @     T  $ ,  l*OsK0"