Run Helpers¶

langsmith.run_helpers ¶

Decorator for creating a run tree from functions.

FUNCTION	DESCRIPTION
`get_current_run_tree`	Get the current run tree.
`get_tracing_context`	Get the current tracing context.
`tracing_context`	Set the tracing context for a block of code.
`is_traceable_function`	Check if a function is @traceable decorated.
`ensure_traceable`	Ensure that a function is traceable.
`is_async`	Inspect function or wrapped function to see if it is async.
`traceable`	Trace a function with langsmith.
`as_runnable`	Convert a function wrapped by the LangSmith @traceable decorator to a Runnable.

LangSmithExtra ¶

Bases: TypedDict

Any additional info to be injected into the run dynamically.

name `instance-attribute` ¶

name: Optional[str]

Optional name for the run.

reference_example_id `instance-attribute` ¶

reference_example_id: Optional[ID_TYPE]

Optional ID of a reference example.

run_extra `instance-attribute` ¶

run_extra: Optional[dict]

Optional additional run information.

parent `instance-attribute` ¶

parent: Optional[Union[RunTree, str, Mapping]]

Optional parent run, can be a RunTree, string, or mapping.

run_tree `instance-attribute` ¶

run_tree: Optional[RunTree]

Optional run tree (deprecated).

project_name `instance-attribute` ¶

project_name: Optional[str]

Optional name of the project.

metadata `instance-attribute` ¶

metadata: Optional[dict[str, Any]]

Optional metadata for the run.

tags `instance-attribute` ¶

tags: Optional[list[str]]

Optional list of tags for the run.

run_id `instance-attribute` ¶

run_id: Optional[ID_TYPE]

Optional ID for the run.

client `instance-attribute` ¶

client: Optional[Client]

Optional LangSmith client.

on_end `instance-attribute` ¶

on_end: Optional[Callable[[RunTree], Any]]

Optional callback function to be called after the run ends and is sent.

SupportsLangsmithExtra ¶

Bases: Protocol, Generic[P, R]

Implementations of this Protocol accept an optional langsmith_extra parameter.

PARAMETER	DESCRIPTION
`*args`	Variable length arguments.
`langsmith_extra`	Optional dictionary of additional parameters for Langsmith. TYPE: `Optional[LangSmithExtra`
`**kwargs`	Keyword arguments.

RETURNS	DESCRIPTION
`R`	The return type of the callable.

METHOD	DESCRIPTION
`__call__`	Call the instance when it is called as a function.

call ¶

__call__(
    *args: args, langsmith_extra: Optional[LangSmithExtra] = None, **kwargs: kwargs
) -> R

Call the instance when it is called as a function.

PARAMETER	DESCRIPTION
`*args`	Variable length argument list. TYPE: `args` DEFAULT: `()`
`langsmith_extra`	Optional dictionary containing additional parameters specific to Langsmith. TYPE: `Optional[LangSmithExtra]` DEFAULT: `None`
`**kwargs`	Arbitrary keyword arguments. TYPE: `kwargs` DEFAULT: `{}`

RETURNS	DESCRIPTION
`R`	The return value of the method. TYPE: `R`

trace ¶

Manage a LangSmith run in context.

This class can be used as both a synchronous and asynchronous context manager.

PARAMETER	DESCRIPTION
`name`	Name of the run. TYPE: `str`
`run_type`	Type of run (e.g., "chain", "llm", "tool"). Defaults to "chain". TYPE: `RUN_TYPE_T` DEFAULT: `'chain'`
`inputs`	Initial input data for the run. Defaults to None. TYPE: `Optional[Dict]` DEFAULT: `None`
`project_name`	Project name to associate the run with. Defaults to None. TYPE: `Optional[str]` DEFAULT: `None`
`parent`	Parent run. Can be a RunTree, dotted order string, or tracing headers. Defaults to None. TYPE: `Optional[Union[RunTree, str, Mapping]]` DEFAULT: `None`
`tags`	List of tags for the run. Defaults to None. TYPE: `Optional[List[str]]` DEFAULT: `None`
`metadata`	Additional metadata for the run. Defaults to None. TYPE: `Optional[Mapping[str, Any]]` DEFAULT: `None`
`client`	LangSmith client for custom settings. Defaults to None. TYPE: `Optional[Client]` DEFAULT: `None`
`run_id`	Preset identifier for the run. Defaults to None. TYPE: `Optional[ID_TYPE]` DEFAULT: `None`
`reference_example_id`	Associates run with a dataset example. Only for root runs in evaluation. Defaults to None. TYPE: `Optional[ID_TYPE]` DEFAULT: `None`
`exceptions_to_handle`	Exception types to ignore. Defaults to None. TYPE: `Optional[Tuple[Type[BaseException], ...]]` DEFAULT: `None`
`extra`	Extra data to send to LangSmith. Use 'metadata' instead. Defaults to None. TYPE: `Optional[Dict]` DEFAULT: `None`

Examples:

Synchronous usage:

.. code-block:: python

>>> with trace("My Operation", run_type="tool", tags=["important"]) as run:
...     result = "foo"  # Perform operation
...     run.metadata["some-key"] = "some-value"
...     run.end(outputs={"result": result})

Asynchronous usage:

.. code-block:: python

>>> async def main():
...     async with trace("Async Operation", run_type="tool", tags=["async"]) as run:
...         result = "foo"  # Await async operation
...         run.metadata["some-key"] = "some-value"
...         # "end" just adds the outputs and sets error to None
...         # The actual patching of the run happens when the context exits
...         run.end(outputs={"result": result})
>>> asyncio.run(main())

Handling specific exceptions:

.. code-block:: python

>>> import pytest
>>> import sys
>>> with trace("Test", exceptions_to_handle=(pytest.skip.Exception,)):
...     if sys.platform == "win32": # Just an example
...         pytest.skip("Skipping test for windows")
...     result = "foo"  # Perform test operation

METHOD	DESCRIPTION
`__init__`	Initialize the trace context manager.
`__enter__`	Enter the context manager synchronously.
`__exit__`	Exit the context manager synchronously.
`__aenter__`	Enter the context manager asynchronously.
`__aexit__`	Exit the context manager asynchronously.

init ¶

__init__(
    name: str,
    run_type: RUN_TYPE_T = "chain",
    *,
    inputs: Optional[dict] = None,
    extra: Optional[dict] = None,
    project_name: Optional[str] = None,
    parent: Optional[Union[RunTree, str, Mapping, Literal["ignore"]]] = None,
    tags: Optional[list[str]] = None,
    metadata: Optional[Mapping[str, Any]] = None,
    client: Optional[Client] = None,
    run_id: Optional[ID_TYPE] = None,
    reference_example_id: Optional[ID_TYPE] = None,
    exceptions_to_handle: Optional[tuple[type[BaseException], ...]] = None,
    attachments: Optional[Attachments] = None,
    **kwargs: Any,
)

Initialize the trace context manager.

Warns if unsupported kwargs are passed.

enter ¶

__enter__() -> RunTree

Enter the context manager synchronously.

RETURNS	DESCRIPTION
`RunTree`	run_trees.RunTree: The newly created run.

exit ¶

__exit__(
    exc_type: Optional[type[BaseException]] = None,
    exc_value: Optional[BaseException] = None,
    traceback: Optional[TracebackType] = None,
) -> None

Exit the context manager synchronously.

PARAMETER	DESCRIPTION
`exc_type`	The type of the exception that occurred, if any. TYPE: `Optional[type[BaseException]]` DEFAULT: `None`
`exc_value`	The exception instance that occurred, if any. TYPE: `Optional[BaseException]` DEFAULT: `None`
`traceback`	The traceback object associated with the exception, if any. TYPE: `Optional[TracebackType]` DEFAULT: `None`

aenter `async` ¶

__aenter__() -> RunTree

Enter the context manager asynchronously.

RETURNS	DESCRIPTION
`RunTree`	run_trees.RunTree: The newly created run.

aexit `async` ¶

__aexit__(
    exc_type: Optional[type[BaseException]] = None,
    exc_value: Optional[BaseException] = None,
    traceback: Optional[TracebackType] = None,
) -> None

Exit the context manager asynchronously.

PARAMETER	DESCRIPTION
`exc_type`	The type of the exception that occurred, if any. TYPE: `Optional[type[BaseException]]` DEFAULT: `None`
`exc_value`	The exception instance that occurred, if any. TYPE: `Optional[BaseException]` DEFAULT: `None`
`traceback`	The traceback object associated with the exception, if any. TYPE: `Optional[TracebackType]` DEFAULT: `None`

get_current_run_tree ¶

get_current_run_tree() -> Optional[RunTree]

Get the current run tree.

get_tracing_context ¶

get_tracing_context(context: Optional[Context] = None) -> dict[str, Any]

Get the current tracing context.

tracing_context ¶

tracing_context(
    *,
    project_name: Optional[str] = None,
    tags: Optional[list[str]] = None,
    metadata: Optional[dict[str, Any]] = None,
    parent: Optional[Union[RunTree, Mapping, str, Literal[False]]] = None,
    enabled: Optional[Union[bool, Literal["local"]]] = None,
    client: Optional[Client] = None,
    replicas: Optional[Sequence[WriteReplica]] = None,
    distributed_parent_id: Optional[str] = None,
    **kwargs: Any,
) -> Generator[None, None, None]

Set the tracing context for a block of code.

PARAMETER	DESCRIPTION
`project_name`	The name of the project to log the run to. Defaults to None. TYPE: `Optional[str]` DEFAULT: `None`
`tags`	The tags to add to the run. Defaults to None. TYPE: `Optional[list[str]]` DEFAULT: `None`
`metadata`	The metadata to add to the run. Defaults to None. TYPE: `Optional[dict[str, Any]]` DEFAULT: `None`
`parent`	The parent run to use for the context. Can be a Run/RunTree object, request headers (for distributed tracing), or the dotted order string. Defaults to None. TYPE: `Optional[Union[RunTree, Mapping, str, Literal[False]]]` DEFAULT: `None`
`client`	The client to use for logging the run to LangSmith. Defaults to None, TYPE: `Optional[Client]` DEFAULT: `None`
`enabled`	Whether tracing is enabled. Defaults to None, meaning it will use the current context value or environment variables. TYPE: `Optional[Union[bool, Literal['local']]]` DEFAULT: `None`
`replicas`	A sequence of WriteReplica dictionaries to send runs to. Example: [{"api_url": "https://api.example.com", "api_key": "key", "project_name": "proj"}] or [{"project_name": "my_experiment", "updates": {"reference_example_id": None}}] TYPE: `Optional[Sequence[WriteReplica]]` DEFAULT: `None`
`distributed_parent_id`	The distributed parent ID for distributed tracing. Defaults to None. TYPE: `Optional[str]` DEFAULT: `None`

is_traceable_function ¶

is_traceable_function(func: Any) -> TypeGuard[SupportsLangsmithExtra[P, R]]

Check if a function is @traceable decorated.

ensure_traceable ¶

ensure_traceable(
    func: Callable[P, R],
    *,
    name: Optional[str] = None,
    metadata: Optional[Mapping[str, Any]] = None,
    tags: Optional[list[str]] = None,
    client: Optional[Client] = None,
    reduce_fn: Optional[Callable[[Sequence], Union[dict, str]]] = None,
    project_name: Optional[str] = None,
    process_inputs: Optional[Callable[[dict], dict]] = None,
    process_outputs: Optional[Callable[..., dict]] = None,
    process_chunk: Optional[Callable] = None,
) -> SupportsLangsmithExtra[P, R]

Ensure that a function is traceable.

is_async ¶

is_async(func: Callable) -> bool

Inspect function or wrapped function to see if it is async.

traceable ¶

traceable(*args: Any, **kwargs: Any) -> Union[Callable, Callable[[Callable], Callable]]

Trace a function with langsmith.

PARAMETER	DESCRIPTION
`run_type`	The type of run (span) to create. Examples: llm, chain, tool, prompt, retriever, etc. Defaults to "chain".
`name`	The name of the run. Defaults to the function name.
`metadata`	The metadata to add to the run. Defaults to None.
`tags`	The tags to add to the run. Defaults to None.
`client`	The client to use for logging the run to LangSmith. Defaults to None, which will use the default client.
`reduce_fn`	A function to reduce the output of the function if the function returns a generator. Defaults to None, which means the values will be logged as a list. Note: if the iterator is never exhausted (e.g. the function returns an infinite generator), this will never be called, and the run itself will be stuck in a pending state.
`project_name`	The name of the project to log the run to. Defaults to None, which will use the default project.
`process_inputs`	Custom serialization / processing function for inputs. Defaults to None.
`process_outputs`	Custom serialization / processing function for outputs. Defaults to None.
`dangerously_allow_filesystem`	Whether to allow filesystem access for attachments. Defaults to False. Traces that reference local filepaths will be uploaded to LangSmith. In general, network-hosted applications should not be using this because referenced files are usually on the user's machine, not the host machine.

RETURNS	DESCRIPTION
`Union[Callable, Callable[[Callable], Callable]]`	Union[Callable, Callable[[Callable], Callable]]: The decorated function.

Note

Requires that LANGSMITH_TRACING_V2 be set to 'true' in the environment.

Examples:

Basic usage:

.. code-block:: python

@traceable
def my_function(x: float, y: float) -> float:
    return x + y


my_function(5, 6)


@traceable
async def my_async_function(query_params: dict) -> dict:
    async with httpx.AsyncClient() as http_client:
        response = await http_client.get(
            "https://api.example.com/data",
            params=query_params,
        )
        return response.json()


asyncio.run(my_async_function({"param": "value"}))

Streaming data with a generator:

.. code-block:: python

@traceable
def my_generator(n: int) -> Iterable:
    for i in range(n):
        yield i


for item in my_generator(5):
    print(item)

Async streaming data:

.. code-block:: python

@traceable
async def my_async_generator(query_params: dict) -> Iterable:
    async with httpx.AsyncClient() as http_client:
        response = await http_client.get(
            "https://api.example.com/data",
            params=query_params,
        )
        for item in response.json():
            yield item


async def async_code():
    async for item in my_async_generator({"param": "value"}):
        print(item)


asyncio.run(async_code())

Specifying a run type and name:

.. code-block:: python

@traceable(name="CustomName", run_type="tool")
def another_function(a: float, b: float) -> float:
    return a * b


another_function(5, 6)

Logging with custom metadata and tags:

.. code-block:: python

@traceable(
    metadata={"version": "1.0", "author": "John Doe"}, tags=["beta", "test"]
)
def tagged_function(x):
    return x**2


tagged_function(5)

Specifying a custom client and project name:

.. code-block:: python

custom_client = Client(api_key="your_api_key")


@traceable(client=custom_client, project_name="My Special Project")
def project_specific_function(data):
    return data


project_specific_function({"data": "to process"})

Manually passing langsmith_extra:

.. code-block:: python

@traceable
def manual_extra_function(x):
    return x**2


manual_extra_function(5, langsmith_extra={"metadata": {"version": "1.0"}})

as_runnable ¶

as_runnable(traceable_fn: Callable) -> Runnable

Convert a function wrapped by the LangSmith @traceable decorator to a Runnable.

PARAMETER	DESCRIPTION
`traceable_fn`	The function wrapped by the @traceable decorator. TYPE: `Callable`

RETURNS	DESCRIPTION
`Runnable`	A Runnable object that maintains a consistent LangSmith tracing context. TYPE: `Runnable`

RAISES	DESCRIPTION
`ImportError`	If langchain module is not installed.
`ValueError`	If the provided function is not wrapped by the @traceable decorator.

Example

@traceable ... def my_function(input_data): ... # Function implementation ... pass runnable = as_runnable(my_function)

Run Helpers¶

langsmith.run_helpers ¶

LangSmithExtra ¶

name instance-attribute ¶

reference_example_id instance-attribute ¶

run_extra instance-attribute ¶

parent instance-attribute ¶

run_tree instance-attribute ¶

project_name instance-attribute ¶

metadata instance-attribute ¶

tags instance-attribute ¶

run_id instance-attribute ¶

client instance-attribute ¶

on_end instance-attribute ¶

SupportsLangsmithExtra ¶

__call__ ¶

trace ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

__aenter__ async ¶

__aexit__ async ¶

get_current_run_tree ¶

get_tracing_context ¶

tracing_context ¶

is_traceable_function ¶

ensure_traceable ¶

is_async ¶

traceable ¶

as_runnable ¶

name `instance-attribute` ¶

reference_example_id `instance-attribute` ¶

run_extra `instance-attribute` ¶

parent `instance-attribute` ¶

run_tree `instance-attribute` ¶

project_name `instance-attribute` ¶

metadata `instance-attribute` ¶

tags `instance-attribute` ¶

run_id `instance-attribute` ¶

client `instance-attribute` ¶

on_end `instance-attribute` ¶

call ¶

init ¶

enter ¶

exit ¶

aenter `async` ¶

aexit `async` ¶