codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import atexit4import contextlib5import contextvars6import datetime7import functools8import hashlib9import importlib10import inspect11import logging12import os13import threading14import time15import uuid16import warnings17from collections.abc import Generator, Sequence18from concurrent.futures import Future19from pathlib import Path20from typing import (21 Any,22 Callable,23 Optional,24 TypeVar,25 Union,26 cast,27 overload,28)29 30from typing_extensions import TypedDict31 32from langsmith import client as ls_client33from langsmith import env as ls_env34from langsmith import run_helpers as rh35from langsmith import run_trees36from langsmith import run_trees as rt37from langsmith import schemas as ls_schemas38from langsmith import utils as ls_utils39from langsmith._internal import _orjson40from langsmith._internal._serde import dumps_json41from langsmith.client import ID_TYPE42 43try:44 import pytest # type: ignore45 46 SkipException = pytest.skip.Exception47except ImportError:48 49 class SkipException(Exception): # type: ignore[no-redef]50 pass51 52 53logger = logging.getLogger(__name__)54 55# UUID5 namespace used for generating consistent example IDs56UUID5_NAMESPACE = uuid.UUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8")57 58T = TypeVar("T")59U = TypeVar("U")60 61 62def _object_hash(obj: Any) -> str:63 """Hash an object to generate a consistent hash string."""64 # Use the existing serialization infrastructure with consistent ordering65 serialized = _stringify(obj)66 return hashlib.sha256(serialized.encode()).hexdigest()67 68 69@overload70def test(71 func: Callable,72) -> Callable: ...73 74 75@overload76def test(77 *,78 id: Optional[uuid.UUID] = None,79 output_keys: Optional[Sequence[str]] = None,80 client: Optional[ls_client.Client] = None,81 test_suite_name: Optional[str] = None,82 metadata: Optional[dict] = None,83 experiment_metadata: Optional[dict] = None,84 repetitions: Optional[int] = None,85 split: Optional[Union[str | list[str]]] = None,86 cached_hosts: Optional[Sequence[str]] = None,87) -> Callable[[Callable], Callable]: ...88 89 90def test(*args: Any, **kwargs: Any) -> Callable:91 """Trace a pytest test case in LangSmith.92 93 This decorator is used to trace a pytest test to LangSmith. It ensures94 that the necessary example data is created and associated with the test function.95 The decorated function will be executed as a test case, and the results will be96 recorded and reported by LangSmith.97 98 Args:99 - id (Optional[uuid.UUID]): A unique identifier for the test case. If not100 provided, an ID will be generated based on the test function's module101 and name.102 - output_keys (Optional[Sequence[str]]): A list of keys to be considered as103 the output keys for the test case. These keys will be extracted from the104 test function's inputs and stored as the expected outputs.105 - client (Optional[ls_client.Client]): An instance of the LangSmith client106 to be used for communication with the LangSmith service. If not provided,107 a default client will be used.108 - test_suite_name (Optional[str]): The name of the test suite to which the109 test case belongs. If not provided, the test suite name will be determined110 based on the environment or the package name.111 - cached_hosts (Optional[Sequence[str]]): A list of hosts or URL prefixes to112 cache requests to during testing. If not provided, all requests will be113 cached (default behavior). This is useful for caching only specific114 API calls (e.g., ["api.openai.com"] or ["https://api.openai.com"]).115 116 Returns:117 Callable: The decorated test function.118 119 Environment:120 - `LANGSMITH_TEST_CACHE`: If set, API calls will be cached to disk to121 save time and costs during testing. Recommended to commit the122 cache files to your repository for faster CI/CD runs.123 Requires the 'langsmith[vcr]' package to be installed.124 - `LANGSMITH_TEST_TRACKING`: Set this variable to the path of a directory125 to enable caching of test results. This is useful for re-running tests126 without re-executing the code. Requires the 'langsmith[vcr]' package.127 128 Example:129 For basic usage, simply decorate a test function with `@pytest.mark.langsmith`.130 Under the hood this will call the `test` method:131 132 ```python133 import pytest134 135 136 # Equivalently can decorate with `test` directly:137 # from langsmith import test138 # @test139 @pytest.mark.langsmith140 def test_addition():141 assert 3 + 4 == 7142 ```143 144 145 Any code that is traced (such as those traced using `@traceable`146 or `wrap_*` functions) will be traced within the test case for147 improved visibility and debugging.148 149 ```python150 import pytest151 from langsmith import traceable152 153 154 @traceable155 def generate_numbers():156 return 3, 4157 158 159 @pytest.mark.langsmith160 def test_nested():161 # Traced code will be included in the test case162 a, b = generate_numbers()163 assert a + b == 7164 ```165 166 LLM calls are expensive! Cache requests by setting167 `LANGSMITH_TEST_CACHE=path/to/cache`. Check in these files to speed up168 CI/CD pipelines, so your results only change when your prompt or requested169 model changes.170 171 Note that this will require that you install langsmith with the `vcr` extra:172 173 `pip install -U "langsmith[vcr]"`174 175 Caching is faster if you install libyaml. See176 https://vcrpy.readthedocs.io/en/latest/installation.html#speed for more details.177 178 ```python179 # os.environ["LANGSMITH_TEST_CACHE"] = "tests/cassettes"180 import openai181 import pytest182 from langsmith import wrappers183 184 oai_client = wrappers.wrap_openai(openai.Client())185 186 187 @pytest.mark.langsmith188 def test_openai_says_hello():189 # Traced code will be included in the test case190 response = oai_client.chat.completions.create(191 model="gpt-3.5-turbo",192 messages=[193 {"role": "system", "content": "You are a helpful assistant."},194 {"role": "user", "content": "Say hello!"},195 ],196 )197 assert "hello" in response.choices[0].message.content.lower()198 ```199 200 You can also specify which hosts to cache by using the `cached_hosts` parameter.201 This is useful when you only want to cache specific API calls:202 203 ```python204 @pytest.mark.langsmith(cached_hosts=["https://api.openai.com"])205 def test_openai_with_selective_caching():206 # Only OpenAI API calls will be cached, other API calls will not207 # be cached208 response = oai_client.chat.completions.create(209 model="gpt-3.5-turbo",210 messages=[211 {"role": "system", "content": "You are a helpful assistant."},212 {"role": "user", "content": "Say hello!"},213 ],214 )215 assert "hello" in response.choices[0].message.content.lower()216 ```217 218 LLMs are stochastic. Naive assertions are flakey. You can use langsmith's219 `expect` to score and make approximate assertions on your results.220 221 ```python222 import pytest223 from langsmith import expect224 225 226 @pytest.mark.langsmith227 def test_output_semantically_close():228 response = oai_client.chat.completions.create(229 model="gpt-3.5-turbo",230 messages=[231 {"role": "system", "content": "You are a helpful assistant."},232 {"role": "user", "content": "Say hello!"},233 ],234 )235 # The embedding_distance call logs the embedding distance to LangSmith236 expect.embedding_distance(237 prediction=response.choices[0].message.content,238 reference="Hello!",239 # The following optional assertion logs a240 # pass/fail score to LangSmith241 # and raises an AssertionError if the assertion fails.242 ).to_be_less_than(1.0)243 # Compute damerau_levenshtein distance244 expect.edit_distance(245 prediction=response.choices[0].message.content,246 reference="Hello!",247 # And then log a pass/fail score to LangSmith248 ).to_be_less_than(1.0)249 ```250 251 The `@test` decorator works natively with pytest fixtures.252 The values will populate the "inputs" of the corresponding example in LangSmith.253 254 ```python255 import pytest256 257 258 @pytest.fixture259 def some_input():260 return "Some input"261 262 263 @pytest.mark.langsmith264 def test_with_fixture(some_input: str):265 assert "input" in some_input266 ```267 268 You can still use `pytest.parametrize()` as usual to run multiple test cases269 using the same test function.270 271 ```python272 import pytest273 274 275 @pytest.mark.langsmith(output_keys=["expected"])276 @pytest.mark.parametrize(277 "a, b, expected",278 [279 (1, 2, 3),280 (3, 4, 7),281 ],282 )283 def test_addition_with_multiple_inputs(a: int, b: int, expected: int):284 assert a + b == expected285 ```286 287 By default, each test case will be assigned a consistent, unique identifier288 based on the function name and module. You can also provide a custom identifier289 using the `id` argument:290 291 ```python292 import pytest293 import uuid294 295 example_id = uuid.uuid4()296 297 298 @pytest.mark.langsmith(id=str(example_id))299 def test_multiplication():300 assert 3 * 4 == 12301 ```302 303 By default, all test inputs are saved as "inputs" to a dataset.304 You can specify the `output_keys` argument to persist those keys305 within the dataset's "outputs" fields.306 307 ```python308 import pytest309 310 311 @pytest.fixture312 def expected_output():313 return "input"314 315 316 @pytest.mark.langsmith(output_keys=["expected_output"])317 def test_with_expected_output(some_input: str, expected_output: str):318 assert expected_output in some_input319 ```320 321 322 To run these tests, use the pytest CLI. Or directly run the test functions.323 324 ```python325 test_output_semantically_close()326 test_addition()327 test_nested()328 test_with_fixture("Some input")329 test_with_expected_output("Some input", "Some")330 test_multiplication()331 test_openai_says_hello()332 test_addition_with_multiple_inputs(1, 2, 3)333 ```334 """335 cached_hosts = kwargs.pop("cached_hosts", None)336 cache_dir = ls_utils.get_cache_dir(kwargs.pop("cache", None))337 338 # Validate cached_hosts usage339 if cached_hosts and not cache_dir:340 raise ValueError(341 "cached_hosts parameter requires caching to be enabled. "342 "Please set the LANGSMITH_TEST_CACHE environment variable "343 "to a cache directory path, "344 "or pass a cache parameter to the test decorator. "345 "Example: LANGSMITH_TEST_CACHE='tests/cassettes' "346 "or @pytest.mark.langsmith(cache='tests/cassettes', cached_hosts=[...])"347 )348 349 langtest_extra = _UTExtra(350 id=kwargs.pop("id", None),351 output_keys=kwargs.pop("output_keys", None),352 client=kwargs.pop("client", None),353 test_suite_name=kwargs.pop("test_suite_name", None),354 cache=cache_dir,355 metadata=kwargs.pop("metadata", None),356 experiment_metadata=kwargs.pop("experiment_metadata", None),357 repetitions=kwargs.pop("repetitions", None),358 split=kwargs.pop("split", None),359 cached_hosts=cached_hosts,360 )361 if kwargs:362 warnings.warn(f"Unexpected keyword arguments: {kwargs.keys()}")363 disable_tracking = ls_utils.test_tracking_is_disabled()364 if disable_tracking:365 logger.info(366 "LANGSMITH_TEST_TRACKING is set to 'false'."367 " Skipping LangSmith test tracking."368 )369 370 def decorator(func: Callable) -> Callable:371 # Handle repetitions372 repetitions = langtest_extra.get("repetitions", 1) or 1373 374 if inspect.iscoroutinefunction(func):375 376 @functools.wraps(func)377 async def async_wrapper(378 *test_args: Any, request: Any = None, **test_kwargs: Any379 ):380 if disable_tracking:381 return await func(*test_args, **test_kwargs)382 383 # Run test multiple times for repetitions384 for i in range(repetitions):385 repetition_extra = langtest_extra.copy()386 await _arun_test(387 func,388 *test_args,389 pytest_request=request,390 **test_kwargs,391 langtest_extra=repetition_extra,392 )393 394 return async_wrapper395 396 @functools.wraps(func)397 def wrapper(*test_args: Any, request: Any = None, **test_kwargs: Any):398 if disable_tracking:399 return func(*test_args, **test_kwargs)400 401 # Run test multiple times for repetitions402 for i in range(repetitions):403 repetition_extra = langtest_extra.copy()404 _run_test(405 func,406 *test_args,407 pytest_request=request,408 **test_kwargs,409 langtest_extra=repetition_extra,410 )411 412 return wrapper413 414 if args and callable(args[0]):415 return decorator(args[0])416 417 return decorator418 419 420## Private functions421 422 423def _get_experiment_name(test_suite_name: str) -> str:424 # If this is a pytest-xdist multi-process run then we need to create the same425 # experiment name across processes. We can do this by accessing the426 # PYTEST_XDIST_TESTRUNID env var.427 if os.environ.get("PYTEST_XDIST_TESTRUNUID") and importlib.util.find_spec("xdist"):428 id_name = test_suite_name + os.environ["PYTEST_XDIST_TESTRUNUID"]429 id_ = str(uuid.uuid5(uuid.NAMESPACE_DNS, id_name).hex[:8])430 else:431 id_ = str(uuid.uuid4().hex[:8])432 433 if os.environ.get("LANGSMITH_EXPERIMENT"):434 prefix = os.environ["LANGSMITH_EXPERIMENT"]435 else:436 prefix = ls_utils.get_tracer_project(False) or "TestSuiteResult"437 name = f"{prefix}:{id_}"438 return name439 440 441def _get_test_suite_name(func: Callable) -> str:442 test_suite_name = ls_utils.get_env_var("TEST_SUITE")443 if test_suite_name:444 return test_suite_name445 repo_name = ls_env.get_git_info()["repo_name"]446 try:447 mod = inspect.getmodule(func)448 if mod:449 return f"{repo_name}.{mod.__name__}"450 except BaseException:451 logger.debug("Could not determine test suite name from file path.")452 453 raise ValueError("Please set the LANGSMITH_TEST_SUITE environment variable.")454 455 456def _get_test_suite(457 client: ls_client.Client, test_suite_name: str458) -> ls_schemas.Dataset:459 if client.has_dataset(dataset_name=test_suite_name):460 return client.read_dataset(dataset_name=test_suite_name)461 else:462 repo = ls_env.get_git_info().get("remote_url") or ""463 description = "Test suite"464 if repo:465 description += f" for {repo}"466 try:467 return client.create_dataset(468 dataset_name=test_suite_name,469 description=description,470 metadata={"__ls_runner": "pytest"},471 )472 except ls_utils.LangSmithConflictError:473 return client.read_dataset(dataset_name=test_suite_name)474 475 476def _start_experiment(477 client: ls_client.Client,478 test_suite: ls_schemas.Dataset,479 experiment_metadata: Optional[dict] = None,480) -> ls_schemas.TracerSession:481 experiment_name = _get_experiment_name(test_suite.name)482 # User-provided experiment_metadata is merged first so that system keys483 # (revision_id, __ls_runner) always take precedence.484 metadata = {485 **(experiment_metadata or {}),486 "revision_id": ls_env.get_langchain_env_var_metadata().get("revision_id"),487 "__ls_runner": "pytest",488 }489 try:490 return client.create_project(491 experiment_name,492 reference_dataset_id=test_suite.id,493 description="Test Suite Results.",494 metadata=metadata,495 )496 except ls_utils.LangSmithConflictError:497 return client.read_project(project_name=experiment_name)498 499 500def _get_example_id(501 dataset_id: str,502 inputs: dict,503 outputs: Optional[dict] = None,504) -> uuid.UUID:505 """Generate example ID based on inputs, outputs, and dataset ID."""506 identifier_obj = (dataset_id, _object_hash(inputs), _object_hash(outputs or {}))507 identifier = _stringify(identifier_obj)508 return uuid.uuid5(UUID5_NAMESPACE, identifier)509 510 511def _get_example_id_legacy(512 func: Callable, inputs: Optional[dict], suite_id: uuid.UUID513) -> tuple[uuid.UUID, str]:514 try:515 file_path = str(Path(inspect.getfile(func)).relative_to(Path.cwd()))516 except ValueError:517 # Fall back to module name if file path is not available518 file_path = func.__module__519 identifier = f"{suite_id}{file_path}::{func.__name__}"520 # If parametrized test, need to add inputs to identifier:521 if hasattr(func, "pytestmark") and any(522 m.name == "parametrize" for m in func.pytestmark523 ):524 identifier += _stringify(inputs)525 return uuid.uuid5(uuid.NAMESPACE_DNS, identifier), identifier[len(str(suite_id)) :]526 527 528def _end_tests(test_suite: _LangSmithTestSuite):529 git_info = ls_env.get_git_info() or {}530 test_suite.shutdown()531 dataset_version = test_suite.get_dataset_version()532 dataset_id = test_suite._dataset.id533 # User-provided experiment_metadata is merged first so that system keys534 # always take precedence.535 test_suite.client.update_project(536 test_suite.experiment_id,537 metadata={538 **(test_suite.experiment_metadata or {}),539 **git_info,540 "dataset_version": dataset_version,541 "revision_id": ls_env.get_langchain_env_var_metadata().get("revision_id"),542 "__ls_runner": "pytest",543 },544 )545 if dataset_version and git_info["commit"] is not None:546 test_suite.client.update_dataset_tag(547 dataset_id=dataset_id,548 as_of=dataset_version,549 tag=f"git:commit:{git_info['commit']}",550 )551 if dataset_version and git_info["branch"] is not None:552 test_suite.client.update_dataset_tag(553 dataset_id=dataset_id,554 as_of=dataset_version,555 tag=f"git:branch:{git_info['branch']}",556 )557 558 559VT = TypeVar("VT", bound=Optional[dict])560 561 562def _serde_example_values(values: VT) -> VT:563 if values is None:564 return cast(VT, values)565 bts = ls_client._dumps_json(values)566 return _orjson.loads(bts)567 568 569class _LangSmithTestSuite:570 _instances: Optional[dict] = None571 _lock = threading.RLock()572 573 def __init__(574 self,575 client: Optional[ls_client.Client],576 experiment: ls_schemas.TracerSession,577 dataset: ls_schemas.Dataset,578 experiment_metadata: Optional[dict] = None,579 ):580 self.client = client or rt.get_cached_client()581 self._experiment = experiment582 self._dataset = dataset583 self._dataset_version: Optional[datetime.datetime] = dataset.modified_at584 self._executor = ls_utils.ContextThreadPoolExecutor()585 self.experiment_metadata = experiment_metadata586 atexit.register(_end_tests, self)587 588 @property589 def id(self):590 return self._dataset.id591 592 @property593 def experiment_id(self):594 return self._experiment.id595 596 @property597 def experiment(self):598 return self._experiment599 600 @classmethod601 def from_test(602 cls,603 client: Optional[ls_client.Client],604 func: Callable,605 test_suite_name: Optional[str] = None,606 experiment_metadata: Optional[dict] = None,607 ) -> _LangSmithTestSuite:608 client = client or rt.get_cached_client()609 test_suite_name = test_suite_name or _get_test_suite_name(func)610 with cls._lock:611 if not cls._instances:612 cls._instances = {}613 if test_suite_name not in cls._instances:614 test_suite = _get_test_suite(client, test_suite_name)615 experiment = _start_experiment(client, test_suite, experiment_metadata)616 cls._instances[test_suite_name] = cls(617 client, experiment, test_suite, experiment_metadata618 )619 return cls._instances[test_suite_name]620 621 @property622 def name(self):623 return self._experiment.name624 625 def get_dataset_version(self):626 return self._dataset_version627 628 def submit_result(629 self,630 run_id: uuid.UUID,631 error: Optional[str] = None,632 skipped: bool = False,633 pytest_plugin: Any = None,634 pytest_nodeid: Any = None,635 ) -> None:636 if skipped:637 score = None638 status = "skipped"639 elif error:640 score = 0641 status = "failed"642 else:643 score = 1644 status = "passed"645 if pytest_plugin and pytest_nodeid:646 pytest_plugin.update_process_status(pytest_nodeid, {"status": status})647 self._executor.submit(self._submit_result, run_id, score)648 649 def _submit_result(self, run_id: uuid.UUID, score: Optional[int]) -> None:650 # trace_id will always be run_id here because the feedback is on the root651 # test run652 self.client.create_feedback(run_id, key="pass", score=score, trace_id=run_id)653 654 def sync_example(655 self,656 example_id: uuid.UUID,657 *,658 inputs: Optional[dict] = None,659 outputs: Optional[dict] = None,660 metadata: Optional[dict] = None,661 split: Optional[Union[str, list[str]]] = None,662 pytest_plugin=None,663 pytest_nodeid=None,664 ) -> None:665 inputs = inputs or {}666 if pytest_plugin and pytest_nodeid:667 update = {"inputs": inputs, "reference_outputs": outputs}668 update = {k: v for k, v in update.items() if v is not None}669 pytest_plugin.update_process_status(pytest_nodeid, update)670 metadata = metadata.copy() if metadata else metadata671 inputs = _serde_example_values(inputs)672 outputs = _serde_example_values(outputs)673 try:674 example = self.client.read_example(example_id=example_id)675 except ls_utils.LangSmithNotFoundError:676 try:677 example = self.client.create_example(678 example_id=example_id,679 inputs=inputs,680 outputs=outputs,681 dataset_id=self.id,682 metadata=metadata,683 split=split,684 created_at=self._experiment.start_time,685 )686 except ls_utils.LangSmithConflictError:687 # Another worker (e.g. pytest-xdist) created this example688 # concurrently between our read and create. Read the existing one.689 example = self.client.read_example(example_id=example_id)690 else:691 normalized_split = split692 if isinstance(normalized_split, str):693 normalized_split = [normalized_split]694 if normalized_split and metadata:695 metadata["dataset_split"] = normalized_split696 existing_dataset_split = (example.metadata or {}).pop("dataset_split")697 if (698 (inputs != example.inputs)699 or (outputs is not None and outputs != example.outputs)700 or (metadata is not None and metadata != example.metadata)701 or str(example.dataset_id) != str(self.id)702 or (703 normalized_split is not None704 and existing_dataset_split != normalized_split705 )706 ):707 self.client.update_example(708 example_id=example.id,709 inputs=inputs,710 outputs=outputs,711 metadata=metadata,712 split=split,713 dataset_id=self.id,714 )715 example = self.client.read_example(example_id=example.id)716 if self._dataset_version is None:717 self._dataset_version = example.modified_at718 elif (719 example.modified_at720 and self._dataset_version721 and example.modified_at > self._dataset_version722 ):723 self._dataset_version = example.modified_at724 725 def _submit_feedback(726 self,727 run_id: ID_TYPE,728 feedback: Union[dict, list],729 pytest_plugin: Any = None,730 pytest_nodeid: Any = None,731 **kwargs: Any,732 ):733 feedback = feedback if isinstance(feedback, list) else [feedback]734 for fb in feedback:735 if pytest_plugin and pytest_nodeid:736 val = fb["score"] if "score" in fb else fb["value"]737 pytest_plugin.update_process_status(738 pytest_nodeid, {"feedback": {fb["key"]: val}}739 )740 self._executor.submit(741 self._create_feedback, run_id=run_id, feedback=fb, **kwargs742 )743 744 def _create_feedback(self, run_id: ID_TYPE, feedback: dict, **kwargs: Any) -> None:745 # trace_id will always be run_id here because the feedback is on the root746 # test run747 self.client.create_feedback(run_id, **feedback, **kwargs, trace_id=run_id)748 749 def shutdown(self):750 self._executor.shutdown()751 752 def end_run(753 self,754 run_tree,755 example_id,756 outputs,757 reference_outputs,758 metadata,759 split,760 pytest_plugin=None,761 pytest_nodeid=None,762 ) -> Future:763 return self._executor.submit(764 self._end_run,765 run_tree=run_tree,766 example_id=example_id,767 outputs=outputs,768 reference_outputs=reference_outputs,769 metadata=metadata,770 split=split,771 pytest_plugin=pytest_plugin,772 pytest_nodeid=pytest_nodeid,773 )774 775 def _end_run(776 self,777 run_tree,778 example_id,779 outputs,780 reference_outputs,781 metadata,782 split,783 pytest_plugin,784 pytest_nodeid,785 ) -> None:786 # TODO: remove this hack so that run durations are correct787 # Ensure example is fully updated788 self.sync_example(789 example_id,790 inputs=run_tree.inputs,791 outputs=reference_outputs,792 split=split,793 metadata=metadata,794 )795 run_tree.reference_example_id = example_id796 run_tree.end(outputs=outputs, metadata={"reference_example_id": example_id})797 run_tree.patch()798 799 800class _TestCase:801 def __init__(802 self,803 test_suite: _LangSmithTestSuite,804 run_id: uuid.UUID,805 example_id: Optional[uuid.UUID] = None,806 metadata: Optional[dict] = None,807 split: Optional[Union[str, list[str]]] = None,808 pytest_plugin: Any = None,809 pytest_nodeid: Any = None,810 inputs: Optional[dict] = None,811 reference_outputs: Optional[dict] = None,812 ) -> None:813 self.test_suite = test_suite814 self.example_id = example_id815 self.run_id = run_id816 self.metadata = metadata817 self.split = split818 self.pytest_plugin = pytest_plugin819 self.pytest_nodeid = pytest_nodeid820 self.inputs = inputs821 self.reference_outputs = reference_outputs822 self._logged_reference_outputs: Optional[dict] = None823 self._logged_outputs: Optional[dict] = None824 825 if pytest_plugin and pytest_nodeid:826 pytest_plugin.add_process_to_test_suite(827 test_suite._dataset.name, pytest_nodeid828 )829 if inputs:830 self.log_inputs(inputs)831 if reference_outputs:832 self.log_reference_outputs(reference_outputs)833 834 def submit_feedback(self, *args, **kwargs: Any):835 self.test_suite._submit_feedback(836 *args,837 **{838 **kwargs,839 **dict(840 pytest_plugin=self.pytest_plugin,841 pytest_nodeid=self.pytest_nodeid,842 ),843 },844 )845 846 def log_inputs(self, inputs: dict) -> None:847 self.inputs = inputs848 if self.pytest_plugin and self.pytest_nodeid:849 self.pytest_plugin.update_process_status(850 self.pytest_nodeid, {"inputs": inputs}851 )852 853 def log_outputs(self, outputs: dict) -> None:854 self._logged_outputs = outputs855 if self.pytest_plugin and self.pytest_nodeid:856 self.pytest_plugin.update_process_status(857 self.pytest_nodeid, {"outputs": outputs}858 )859 860 def log_reference_outputs(self, reference_outputs: dict) -> None:861 self._logged_reference_outputs = reference_outputs862 if self.pytest_plugin and self.pytest_nodeid:863 self.pytest_plugin.update_process_status(864 self.pytest_nodeid, {"reference_outputs": reference_outputs}865 )866 867 def submit_test_result(868 self,869 error: Optional[str] = None,870 skipped: bool = False,871 ) -> None:872 return self.test_suite.submit_result(873 self.run_id,874 error=error,875 skipped=skipped,876 pytest_plugin=self.pytest_plugin,877 pytest_nodeid=self.pytest_nodeid,878 )879 880 def start_time(self) -> None:881 if self.pytest_plugin and self.pytest_nodeid:882 self.pytest_plugin.update_process_status(883 self.pytest_nodeid, {"start_time": time.time()}884 )885 886 def end_time(self) -> None:887 if self.pytest_plugin and self.pytest_nodeid:888 self.pytest_plugin.update_process_status(889 self.pytest_nodeid, {"end_time": time.time()}890 )891 892 def end_run(self, run_tree, outputs: Any) -> None:893 if not (outputs is None or isinstance(outputs, dict)):894 outputs = {"output": outputs}895 example_id = self.example_id or _get_example_id(896 dataset_id=str(self.test_suite.id),897 inputs=self.inputs or {},898 outputs=outputs,899 )900 self.test_suite.end_run(901 run_tree,902 example_id,903 outputs,904 reference_outputs=self._logged_reference_outputs,905 metadata=self.metadata,906 split=self.split,907 pytest_plugin=self.pytest_plugin,908 pytest_nodeid=self.pytest_nodeid,909 )910 911 912_TEST_CASE = contextvars.ContextVar[Optional[_TestCase]]("_TEST_CASE", default=None)913 914 915class _UTExtra(TypedDict, total=False):916 client: Optional[ls_client.Client]917 id: Optional[uuid.UUID]918 output_keys: Optional[Sequence[str]]919 test_suite_name: Optional[str]920 cache: Optional[str]921 metadata: Optional[dict]922 experiment_metadata: Optional[dict]923 repetitions: Optional[int]924 split: Optional[Union[str, list[str]]]925 cached_hosts: Optional[Sequence[str]]926 927 928def _create_test_case(929 func: Callable,930 *args: Any,931 pytest_request: Any,932 langtest_extra: _UTExtra,933 **kwargs: Any,934) -> _TestCase:935 client = langtest_extra["client"] or rt.get_cached_client()936 output_keys = langtest_extra["output_keys"]937 metadata = langtest_extra["metadata"]938 split = langtest_extra["split"]939 # Resolve experiment_metadata: explicit kwarg > env var940 experiment_metadata = langtest_extra.get("experiment_metadata")941 if experiment_metadata is None:942 env_val = os.environ.get("LANGSMITH_EXPERIMENT_METADATA")943 if env_val:944 try:945 experiment_metadata = _orjson.loads(env_val)946 except Exception as e:947 msg = f"LANGSMITH_EXPERIMENT_METADATA env var is not valid JSON: {e}"948 raise ValueError(msg) from e949 signature = inspect.signature(func)950 inputs = rh._get_inputs_safe(signature, *args, **kwargs) or None951 outputs = None952 if output_keys:953 outputs = {}954 if not inputs:955 msg = (956 "'output_keys' should only be specified when marked test function has "957 "input arguments."958 )959 raise ValueError(msg)960 for k in output_keys:961 outputs[k] = inputs.pop(k, None)962 test_suite = _LangSmithTestSuite.from_test(963 client, func, langtest_extra.get("test_suite_name"), experiment_metadata964 )965 example_id = langtest_extra["id"]966 dataset_sdk_version = (967 test_suite._dataset.metadata968 and test_suite._dataset.metadata.get("runtime")969 and test_suite._dataset.metadata.get("runtime", {}).get("sdk_version")970 )971 if not dataset_sdk_version or not ls_utils.is_version_greater_or_equal(972 dataset_sdk_version, "0.4.33"973 ):974 legacy_example_id, example_name = _get_example_id_legacy(975 func, inputs, test_suite.id976 )977 example_id = example_id or legacy_example_id978 pytest_plugin = (979 pytest_request.config.pluginmanager.get_plugin("langsmith_output_plugin")980 if pytest_request981 else None982 )983 pytest_nodeid = pytest_request.node.nodeid if pytest_request else None984 if pytest_plugin:985 pytest_plugin.test_suite_urls[test_suite._dataset.name] = (986 cast(str, test_suite._dataset.url)987 + "/compare?selectedSessions="988 + str(test_suite.experiment_id)989 )990 test_case = _TestCase(991 test_suite,992 run_id=uuid.uuid4(),993 example_id=example_id,994 metadata=metadata,995 split=split,996 inputs=inputs,997 reference_outputs=outputs,998 pytest_plugin=pytest_plugin,999 pytest_nodeid=pytest_nodeid,1000 )1001 return test_case1002 1003 1004def _run_test(1005 func: Callable,1006 *test_args: Any,1007 pytest_request: Any,1008 langtest_extra: _UTExtra,1009 **test_kwargs: Any,1010) -> None:1011 test_case = _create_test_case(1012 func,1013 *test_args,1014 **test_kwargs,1015 pytest_request=pytest_request,1016 langtest_extra=langtest_extra,1017 )1018 _TEST_CASE.set(test_case)1019 1020 def _test():1021 test_case.start_time()1022 with rh.trace(1023 name=getattr(func, "__name__", "Test"),1024 run_id=test_case.run_id,1025 inputs=test_case.inputs,1026 metadata={1027 # Experiment run metadata is prefixed with "ls_example_" in1028 # the ingest backend, but we must reproduce this behavior here1029 # because the example may not have been created before the trace1030 # starts.1031 f"ls_example_{k}": v1032 for k, v in (test_case.metadata or {}).items()1033 },1034 project_name=test_case.test_suite.name,1035 exceptions_to_handle=(SkipException,),1036 _end_on_exit=False,1037 ) as run_tree:1038 try:1039 result = func(*test_args, **test_kwargs)1040 except SkipException as e:1041 test_case.submit_test_result(error=repr(e), skipped=True)1042 test_case.end_run(run_tree, {"skipped_reason": repr(e)})1043 raise e1044 except BaseException as e:1045 test_case.submit_test_result(error=repr(e))1046 test_case.end_run(run_tree, None)1047 raise e1048 else:1049 test_case.end_run(run_tree, result)1050 finally:1051 test_case.end_time()1052 try:1053 test_case.submit_test_result()1054 except BaseException as e:1055 logger.warning(1056 f"Failed to create feedback for run_id {test_case.run_id}:\n{e}"1057 )1058 1059 if langtest_extra["cache"]:1060 cache_path = Path(langtest_extra["cache"]) / f"{test_case.test_suite.id}.yaml"1061 else:1062 cache_path = None1063 current_context = rh.get_tracing_context()1064 metadata = {1065 **(current_context["metadata"] or {}),1066 **{1067 "experiment": test_case.test_suite.experiment.name,1068 },1069 }1070 # Handle cached_hosts parameter1071 ignore_hosts = [test_case.test_suite.client.api_url]1072 allow_hosts = langtest_extra.get("cached_hosts") or None1073 1074 with (1075 rh.tracing_context(**{**current_context, "metadata": metadata}),1076 ls_utils.with_optional_cache(1077 cache_path, ignore_hosts=ignore_hosts, allow_hosts=allow_hosts1078 ),1079 ):1080 _test()1081 1082 1083async def _arun_test(1084 func: Callable,1085 *test_args: Any,1086 pytest_request: Any,1087 langtest_extra: _UTExtra,1088 **test_kwargs: Any,1089) -> None:1090 test_case = _create_test_case(1091 func,1092 *test_args,1093 **test_kwargs,1094 pytest_request=pytest_request,1095 langtest_extra=langtest_extra,1096 )1097 _TEST_CASE.set(test_case)1098 1099 async def _test():1100 test_case.start_time()1101 with rh.trace(1102 name=getattr(func, "__name__", "Test"),1103 run_id=test_case.run_id,1104 reference_example_id=test_case.example_id,1105 inputs=test_case.inputs,1106 metadata={1107 # Experiment run metadata is prefixed with "ls_example_" in1108 # the ingest backend, but we must reproduce this behavior here1109 # because the example may not have been created before the trace1110 # starts.1111 f"ls_example_{k}": v1112 for k, v in (test_case.metadata or {}).items()1113 },1114 project_name=test_case.test_suite.name,1115 exceptions_to_handle=(SkipException,),1116 _end_on_exit=False,1117 ) as run_tree:1118 try:1119 result = await func(*test_args, **test_kwargs)1120 except SkipException as e:1121 test_case.submit_test_result(error=repr(e), skipped=True)1122 test_case.end_run(run_tree, {"skipped_reason": repr(e)})1123 raise e1124 except BaseException as e:1125 test_case.submit_test_result(error=repr(e))1126 test_case.end_run(run_tree, None)1127 raise e1128 else:1129 test_case.end_run(run_tree, result)1130 finally:1131 test_case.end_time()1132 try:1133 test_case.submit_test_result()1134 except BaseException as e:1135 logger.warning(1136 f"Failed to create feedback for run_id {test_case.run_id}:\n{e}"1137 )1138 1139 if langtest_extra["cache"]:1140 cache_path = Path(langtest_extra["cache"]) / f"{test_case.test_suite.id}.yaml"1141 else:1142 cache_path = None1143 current_context = rh.get_tracing_context()1144 metadata = {1145 **(current_context["metadata"] or {}),1146 **{1147 "experiment": test_case.test_suite.experiment.name,1148 "reference_example_id": str(test_case.example_id),1149 },1150 }1151 # Handle cached_hosts parameter1152 ignore_hosts = [test_case.test_suite.client.api_url]1153 cached_hosts = langtest_extra.get("cached_hosts")1154 allow_hosts = cached_hosts if cached_hosts else None1155 1156 with (1157 rh.tracing_context(**{**current_context, "metadata": metadata}),1158 ls_utils.with_optional_cache(1159 cache_path, ignore_hosts=ignore_hosts, allow_hosts=allow_hosts1160 ),1161 ):1162 await _test()1163 1164 1165# For backwards compatibility1166unit = test1167 1168 1169def log_inputs(inputs: dict, /) -> None:1170 """Log run inputs from within a pytest test run.1171 1172 Should only be used in pytest tests decorated with @pytest.mark.langsmith.1173 1174 Args:1175 inputs: Inputs to log.1176 1177 Example:1178 ```python1179 from langsmith import testing as t1180 1181 1182 @pytest.mark.langsmith1183 def test_foo() -> None:1184 x = 01185 y = 11186 t.log_inputs({"x": x, "y": y})1187 assert foo(x, y) == 21188 ```1189 """1190 if ls_utils.test_tracking_is_disabled():1191 logger.info("LANGSMITH_TEST_TRACKING is set to 'false'. Skipping log_inputs.")1192 return1193 run_tree = rh.get_current_run_tree()1194 test_case = _TEST_CASE.get()1195 if not run_tree or not test_case:1196 msg = (1197 "log_inputs should only be called within a pytest test decorated with "1198 "@pytest.mark.langsmith, and with tracing enabled (by setting the "1199 "LANGSMITH_TRACING environment variable to 'true')."1200 )