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.

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