temporalio.contrib.langsmith.LangSmithInterceptor

class documentation

class LangSmithInterceptor(temporalio.client.Interceptor, temporalio.worker.Interceptor): (source)

Constructor: LangSmithInterceptor(client, project_name, add_temporal_runs, default_metadata, ...)

View In Hierarchy

Interceptor that supports client and worker LangSmith run creation and context propagation.

Warning

This class is experimental and may change in future versions. Use with caution in production environments.

Method __init__ Initialize the LangSmith interceptor with tracing configuration.
Method intercept_activity Create an activity inbound interceptor for LangSmith tracing.
Method intercept_client Create a client outbound interceptor for LangSmith tracing.
Method intercept_nexus_operation Create a Nexus operation inbound interceptor for LangSmith tracing.
Method maybe_run Create a LangSmith run with this interceptor's config already applied.
Method workflow_interceptor_class Return the workflow interceptor class with config bound.
Instance Variable _add_temporal_runs Undocumented
Instance Variable _client Undocumented
Instance Variable _default_metadata Undocumented
Instance Variable _default_tags Undocumented
Instance Variable _executor Undocumented
Instance Variable _project_name Undocumented
def __init__(self, *, client: langsmith.Client | None = None, project_name: str | None = None, add_temporal_runs: bool = False, default_metadata: dict[str, Any] | None = None, default_tags: list[str] | None = None): (source)

Initialize the LangSmith interceptor with tracing configuration.

def intercept_activity(self, next: temporalio.worker.ActivityInboundInterceptor) -> temporalio.worker.ActivityInboundInterceptor: (source)
overrides temporalio.worker.Interceptor.intercept_activity

Create an activity inbound interceptor for LangSmith tracing.

def intercept_client(self, next: temporalio.client.OutboundInterceptor) -> temporalio.client.OutboundInterceptor: (source)
overrides temporalio.client.Interceptor.intercept_client

Create a client outbound interceptor for LangSmith tracing.

def intercept_nexus_operation(self, next: temporalio.worker.NexusOperationInboundInterceptor) -> temporalio.worker.NexusOperationInboundInterceptor: (source)
overrides temporalio.worker.Interceptor.intercept_nexus_operation

Create a Nexus operation inbound interceptor for LangSmith tracing.

@contextmanager
def maybe_run(self, name: str, *, run_type: str = 'chain', parent: RunTree | None = None, extra_metadata: dict[str, Any] | None = None) -> Iterator[None]: (source)

Create a LangSmith run with this interceptor's config already applied.

def workflow_interceptor_class(self, input: temporalio.worker.WorkflowInterceptorClassInput) -> type[_LangSmithWorkflowInboundInterceptor]: (source)
overrides temporalio.worker.Interceptor.workflow_interceptor_class

Return the workflow interceptor class with config bound.

_add_temporal_runs = (source)

Undocumented

_client = (source)

Undocumented

_default_metadata = (source)

Undocumented

_default_tags = (source)

Undocumented

_executor = (source)

Undocumented

_project_name = (source)

Undocumented