Skip to main content

Arize integration

View Markdown

Temporal's OpenTelemetry plugin sends Workflow, Activity, and LLM spans to Arize, either the open-source Arize Phoenix or the Arize AX platform, as one trace per Workflow interaction. Arize reads the OpenInference semantic conventions, so LLM calls appear with their model, token usage, and messages, and traces group into sessions by Workflow Id.

Temporal gives your agent code Durable Execution. Arize adds the observability side: inspect LLM inputs and outputs, follow a request from the Client through the Workflow to the model, evaluate outputs, and compare runs over time.

This guide configures the existing OpenTelemetryPlugin for Arize. There is no Arize-specific Temporal plugin and no Arize SDK is required. Arize maintains a matching guide on its side: Temporal tracing in Arize AX.

All code snippets in this guide are taken from the Arize tracing sample. Refer to the sample for complete code, including a one-container Phoenix setup, a second scenario built on the OpenAI Agents SDK, and a script that verifies traces through the Phoenix API.

Prerequisites​

  • This guide assumes you are already familiar with Arize. If you aren't, refer to the Phoenix documentation or the Arize AX documentation for more details.
  • If you are new to Temporal, we recommend reading Understanding Temporal or taking the Temporal 101 course.
  • Ensure you have set up your local development environment by following the Set up your local development environment guide. When you're done, leave the Temporal Development Server running if you want to test your code locally.
  • A place to send traces: a local Phoenix (docker run -p 6006:6006 arizephoenix/phoenix:version-20.9.0, or uvx --from "arize-phoenix==20.9.0" phoenix serve), or an Arize AX space ID and API key.

Install​

Install the Temporal Python SDK with the OpenTelemetry extra, an OTLP/HTTP exporter, the OpenInference semantic conventions, and the OpenInference instrumentation for the LLM SDK you call from Activities.

uv add "temporalio[opentelemetry]>=1.32.0" opentelemetry-exporter-otlp-proto-http \
  openinference-semantic-conventions openinference-instrumentation-openai

For the OpenAI Agents SDK variant, also add "temporalio[openai-agents]" and openinference-instrumentation-openai-agents.

Configure the tracer provider and exporter​

The plugin requires the replay-safe tracer provider from create_tracer_provider(), installed as the global provider before you connect the Client, in every process that traces. The provider generates span identifiers deterministically from Workflow state and does not export spans while a Workflow replays, which is what keeps Arize free of duplicates.

The openinference.project.name resource attribute selects the Phoenix or Arize AX project that receives the spans.

arize_tracing/telemetry.py

def setup_tracing(service_name: str) -> None:
    """Install a replay-safe tracer provider that exports spans to Arize.

    Must be called once at process start, before connecting the Temporal
    client, in every process that traces (worker and starter alike).

    Honors ``OTEL_SDK_DISABLED=true`` as a kill switch: the provider is still
    installed (the Temporal plugin requires it) but no exporter is attached.
    """
    provider = create_tracer_provider(
        resource=Resource.create(
            {
                SERVICE_NAME: service_name,
                # Selects the Phoenix / Arize AX project that receives the spans.
                ResourceAttributes.PROJECT_NAME: project_name(),
            }
        )
    )
    provider.add_span_processor(OpenInferenceEnrichmentProcessor())
    if os.environ.get("OTEL_SDK_DISABLED", "").lower() != "true":
        # A short schedule delay so demo spans show up in Arize quickly.
        # Buffered spans are also flushed at process exit (the provider
        # registers a shutdown hook), but call force_flush() before reading
        # traces back to avoid racing the batch.
        provider.add_span_processor(
            BatchSpanProcessor(_exporter(), schedule_delay_millis=500)
        )
    else:
        logger.info("OTEL_SDK_DISABLED=true - spans will not be exported")
    trace.set_tracer_provider(provider)

Phoenix and Arize AX both ingest OTLP over HTTP. One exporter helper covers both: Phoenix by default, Arize AX when a space ID and API key are present.

arize_tracing/telemetry.py

def _exporter() -> OTLPSpanExporter:
    """OTLP/HTTP exporter for Arize AX or, by default, Phoenix."""
    space_id = os.environ.get("ARIZE_SPACE_ID")
    api_key = os.environ.get("ARIZE_API_KEY")
    if space_id and api_key:
        # Arize AX (SaaS). For the EU region set
        # ARIZE_OTLP_ENDPOINT=https://otlp.eu-west-1a.arize.com/v1/traces
        return OTLPSpanExporter(
            endpoint=os.environ.get("ARIZE_OTLP_ENDPOINT", DEFAULT_ARIZE_ENDPOINT),
            headers={"space_id": space_id, "api_key": api_key},
            timeout=10,
        )
    # Phoenix (self-hosted or Phoenix Cloud). PHOENIX_COLLECTOR_ENDPOINT is the
    # base URL, as in Phoenix's own tooling; OTLP/HTTP traces go to /v1/traces.
    base = phoenix_base_url()
    endpoint = base if base.endswith("/v1/traces") else f"{base}/v1/traces"
    headers: dict[str, str] = {}
    phoenix_api_key = os.environ.get("PHOENIX_API_KEY")
    if phoenix_api_key:
        headers["Authorization"] = f"Bearer {phoenix_api_key}"
    return OTLPSpanExporter(endpoint=endpoint, headers=headers, timeout=10)

Short-lived processes must flush. The sample's starter and Worker call force_flush() on the provider before they exit so the last spans are not dropped.

Configure the Client and Worker​

Register the plugin on the Client. Workers created from that Client inherit it, so the same registration covers the starter and the Worker. add_temporal_spans=True creates spans for the Temporal operations themselves (StartWorkflow, RunWorkflow, StartActivity, RunActivity, Update handlers, and so on) in addition to propagating trace context; the default False propagates context only.

arize_tracing/ticket_triage/worker.py

setup_tracing("ticket-triage-worker")
instrument_openai()

config = ClientConfig.load_client_connect_config()
config.setdefault("target_host", "localhost:7233")

# add_temporal_spans=True emits spans for Temporal operations (StartWorkflow,
# RunWorkflow, RunActivity, HandleUpdate, ...) in addition to propagating
# trace context across the client/workflow/activity boundaries.
client = await Client.connect(
    **config,
    plugins=[OpenTelemetryPlugin(add_temporal_spans=True)],
)

worker = Worker(
    client,
    task_queue=TASK_QUEUE,
    workflows=[TicketTriageWorkflow],
    activities=[classify_ticket, lookup_account, draft_reply],
    max_cached_workflows=0 if args.replay_stress else 1000,
    # No plugins here: workers inherit them from the client.
)

The starter uses the same setup_tracing() and plugin registration, then wraps the whole interaction in a root span as shown in Add OpenInference attributes.

Trace LLM calls in Activities​

Model calls are network I/O, so they belong in Activities, not in Workflow code. That also makes them easy to trace: instrument the LLM SDK once in the Worker process, and every call made from an Activity produces an OpenInference LLM span nested under that Activity's span, with the model name, token counts, and messages that Arize renders.

Disable the LLM client's own retries and let the Activity retry policy own them. Each attempt then appears as its own RunActivity span, which is how Arize shows a retry.

arize_tracing/ticket_triage/activities.py

@activity.defn
async def classify_ticket(ticket: Ticket) -> Classification:
    if FAIL_FIRST_ATTEMPT and activity.info().attempt == 1:
        # Demo: a transient failure on the first attempt. Temporal retries the
        # activity; Arize shows one RunActivity span per attempt, the first
        # with an error status.
        raise ApplicationError(
            "Simulated transient LLM failure on attempt 1", type="SimulatedFailure"
        )
    for _ in range(SLOW_CLASSIFY_SECONDS):
        # Demo: a slow attempt you can kill the worker during. Heartbeats let
        # Temporal detect the dead worker via the heartbeat timeout.
        activity.heartbeat()
        await asyncio.sleep(1)
    response = await _openai_client().chat.completions.create(
        model=os.environ.get("MODEL_CLASSIFY", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": CLASSIFY_PROMPT},
            {"role": "user", "content": f"{ticket.subject}\n\n{ticket.body}"},
        ],
        timeout=30,
    )
    return parse_classification(response.choices[0].message.content or "")

Workflow code can create spans too. Under the plugin, the standard OpenTelemetry API is replay-safe inside a Workflow, and the plugin allows the opentelemetry module through the Workflow sandbox. Set an OpenInference span kind on spans you create so Arize does not show them as unknown.

arize_tracing/ticket_triage/workflows.py

@workflow.defn
class TicketTriageWorkflow:
    def __init__(self) -> None:
        self._approval: Optional[ApprovalDecision] = None

    @workflow.run
    async def run(self, ticket: Ticket) -> TriageResult:
        # A custom span grouping the two triage activities. Under the
        # OpenTelemetryPlugin this is replay-safe; the activity spans (and the
        # LLM spans inside them) nest underneath it. The OpenInference span
        # kind tells Arize how to render it.
        with trace.get_tracer(__name__).start_as_current_span(
            "triage",
            attributes={
                SpanAttributes.OPENINFERENCE_SPAN_KIND: OpenInferenceSpanKindValues.CHAIN.value
            },
        ) as span:
            classification: Classification = await workflow.execute_activity(
                classify_ticket,
                ticket,
                start_to_close_timeout=timedelta(seconds=60),
                # Heartbeats let Temporal detect a worker that died mid-attempt
                # quickly, so the retry (and its new span) starts sooner.
                heartbeat_timeout=timedelta(seconds=10),
                retry_policy=LLM_RETRY_POLICY,
            )
            account = await workflow.execute_activity(
                lookup_account,
                ticket.customer_email,
                start_to_close_timeout=timedelta(seconds=10),
            )
            span.set_attribute("triage.category", classification.category)
            span.set_attribute("triage.priority", classification.priority)

        # Wait for a human approval, delivered as a workflow update.
        await workflow.wait_condition(lambda: self._approval is not None)
        approval = self._approval
        assert approval is not None
        if not approval.approved:
            return TriageResult(status="declined", classification=classification)

        reply = await workflow.execute_activity(
            draft_reply,
            DraftReplyInput(
                ticket=ticket, classification=classification, account=account
            ),
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=LLM_RETRY_POLICY,
        )
        return TriageResult(
            status="replied", classification=classification, reply=reply
        )

    @workflow.update
    async def approve(self, decision: ApprovalDecision) -> str:
        self._approval = decision
        return "approved" if decision.approved else "declined"

    @approve.validator
    def approve_validator(self, decision: ApprovalDecision) -> None:
        if decision.approved and not decision.reviewer:
            raise ValueError("approval requires a reviewer")

Trace agents built with the OpenAI Agents SDK​

If your Workflow runs an agent through the OpenAI Agents SDK integration, enable the plugin's OpenTelemetry bridge instead of instrumenting the OpenAI client. The bridge converts the Agents SDK trace into OpenInference spans (AGENT, LLM, and TOOL, plus CHAIN spans for the Temporal operations) on the same replay-safe provider. Construct the plugin after setup_tracing(), because it checks that the global provider is the replay-safe one, and do not add OpenTelemetryPlugin as well: the two would produce overlapping spans.

arize_tracing/ticket_triage_agents/plugin.py

def agents_plugin(*, register_activities: bool = True) -> OpenAIAgentsPlugin:
    """The OpenAI Agents plugin, configured to export through OpenTelemetry.

    Construct it after ``setup_tracing()``: with ``use_otel_instrumentation=True``
    the plugin checks that the global tracer provider is Temporal's replay-safe
    provider, then bridges Agents SDK tracing to OpenTelemetry through the
    OpenInference ``openai-agents`` instrumentation. ``add_temporal_spans=True``
    adds CHAIN spans for the Temporal operations (start workflow, execute
    activity, ...) around the AGENT, LLM, and TOOL spans.
    """
    return OpenAIAgentsPlugin(
        use_otel_instrumentation=True,
        add_temporal_spans=True,
        # False for a workflow-only worker (see worker.py --role).
        register_activities=register_activities,
        model_params=ModelActivityParameters(
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=RetryPolicy(
                maximum_attempts=3, initial_interval=timedelta(seconds=1)
            ),
        ),
    )

The Agents SDK trace opened by the starter becomes the root AGENT span of the Arize trace, so the OpenInference trace-level attributes go onto it directly. Set the output before the trace ends, because ending the trace ends the span.

arize_tracing/ticket_triage_agents/starter.py

# Starting an Agents SDK trace outside a worker requires the plugin's
# tracing context. The OpenInference instrumentation turns the trace into
# the root AGENT span and makes it the current OpenTelemetry span, so the
# OpenInference trace-level attributes are set on it directly.
try:
    with plugin.tracing_context():
        with agents_trace("Ticket triage agents", group_id=workflow_id):
            root = trace.get_current_span()
            root.set_attributes(
                {
                    SpanAttributes.SESSION_ID: workflow_id,
                    SpanAttributes.USER_ID: args.user,
                    SpanAttributes.INPUT_VALUE: json.dumps(
                        dataclasses.asdict(ticket)
                    ),
                    SpanAttributes.INPUT_MIME_TYPE: OpenInferenceMimeTypeValues.JSON.value,
                    SpanAttributes.METADATA: json.dumps(
                        {
                            "temporal.workflow_id": workflow_id,
                            "temporal.task_queue": TASK_QUEUE,
                            "temporal.namespace": client.namespace,
                        }
                    ),
                    SpanAttributes.TAG_TAGS: [
                        "temporal",
                        "ticket-triage",
                        "openai-agents",
                    ],
                }
            )
            trace_id = format(root.get_span_context().trace_id, "032x")

            handle = await client.start_workflow(
                TicketTriageAgentsWorkflow.run,
                request,
                id=workflow_id,
                task_queue=TASK_QUEUE,
            )
            print(f"Started workflow: {workflow_id}")
            if args.pause_before_approval:
                print(f"Pausing {args.pause_before_approval}s before approving ...")
                await asyncio.sleep(args.pause_before_approval)
            update_result = await handle.execute_update(
                TicketTriageAgentsWorkflow.approve,
                ApprovalDecision(approved=approved, reviewer="demo-reviewer"),
            )
            print(f"Approval update: {update_result}")
            result = await handle.result()
            # Set before the Agents SDK trace ends; ending it ends the span.
            root.set_attribute(
                SpanAttributes.OUTPUT_VALUE, json.dumps(dataclasses.asdict(result))
            )
            root.set_attribute(
                SpanAttributes.OUTPUT_MIME_TYPE,
                OpenInferenceMimeTypeValues.JSON.value,
            )
finally:
    force_flush()

The bridge is Public Preview in the Temporal SDK. The sample's telemetry.py includes a small logging filter for a known, harmless "Failed to detach context" message that the bridge produces when an agent span outlives a Workflow Task.

caution

Until temporalio/sdk-python#1852 is fixed, a Worker process that runs both the Workflow and its Activities exports the temporal:startActivity spans with a parent that is not in the trace, so the model-call and tool subtrees appear detached from the agent turns in Arize. Run the Workflow and the Activities in separate Worker processes, as the sample's --role workflows and --role activities Worker flags do.

Add OpenInference attributes​

Arize lists a trace by the input.value and output.value of its root span, groups traces into sessions by session.id, and filters by user.id, metadata, and tag.tags. Set them on a root span in the starter, and use the Workflow Id as the session so every interaction with one Workflow Execution lands in one session. Setting session.id and user.id as OpenTelemetry baggage as well lets Temporal carry them into the Workflow and Activity contexts through its trace-context header.

arize_tracing/ticket_triage/starter.py

# session.id / user.id as OpenTelemetry baggage: Temporal propagates baggage
# through its trace-context header, so the enrichment processor can stamp
# both onto every span on the worker as well, LLM spans included.
ctx = baggage.set_baggage(SpanAttributes.SESSION_ID, workflow_id)
ctx = baggage.set_baggage(SpanAttributes.USER_ID, args.user, context=ctx)
token = context.attach(ctx)
tracer = trace.get_tracer(__name__)
try:
    with tracer.start_as_current_span(
        "ticket-triage",
        attributes={
            # OpenInference attributes on the trace's root span: Arize lists
            # the trace by its input/output and groups it by session.
            SpanAttributes.OPENINFERENCE_SPAN_KIND: OpenInferenceSpanKindValues.CHAIN.value,
            SpanAttributes.SESSION_ID: workflow_id,
            SpanAttributes.USER_ID: args.user,
            SpanAttributes.INPUT_VALUE: json.dumps(dataclasses.asdict(ticket)),
            SpanAttributes.INPUT_MIME_TYPE: OpenInferenceMimeTypeValues.JSON.value,
            SpanAttributes.METADATA: json.dumps(
                {
                    "temporal.workflow_id": workflow_id,
                    "temporal.task_queue": TASK_QUEUE,
                    "temporal.namespace": client.namespace,
                }
            ),
            SpanAttributes.TAG_TAGS: ["temporal", "ticket-triage"],
        },
    ) as root:
        trace_id = format(root.get_span_context().trace_id, "032x")
        handle = await client.start_workflow(
            TicketTriageWorkflow.run,
            ticket,
            id=workflow_id,
            task_queue=TASK_QUEUE,
        )
        print(f"Started workflow: {workflow_id}")

        if args.pause_before_approval:
            print(f"Pausing {args.pause_before_approval}s before approving ...")
            await asyncio.sleep(args.pause_before_approval)

        update_result = await handle.execute_update(
            TicketTriageWorkflow.approve,
            ApprovalDecision(approved=approved, reviewer="demo-reviewer"),
        )
        print(f"Approval update: {update_result}")

        result = await handle.result()
        root.set_attribute(
            SpanAttributes.OUTPUT_VALUE, json.dumps(dataclasses.asdict(result))
        )
        root.set_attribute(
            SpanAttributes.OUTPUT_MIME_TYPE, OpenInferenceMimeTypeValues.JSON.value
        )
finally:
    context.detach(token)
    # The starter is short-lived; flush so its spans (the trace root and
    # the client-side StartWorkflow/StartWorkflowUpdate spans) are not
    # dropped at process exit.
    force_flush()

Temporal's own spans have no OpenInference kind, so Phoenix and Arize AX show them as unknown. A small span processor gives them a kind, copies the Temporal identifiers into OpenInference metadata, sets the session from the Workflow Id, records the Activity attempt number, and applies the baggage to every span, LLM spans included.

arize_tracing/telemetry.py

class OpenInferenceEnrichmentProcessor(SpanProcessor):
    """Make Temporal's spans first-class citizens in Arize.

    Temporal's ``OpenTelemetryPlugin`` emits plain spans (``RunWorkflow:*``,
    ``RunActivity:*``, ...). When such a span starts, this processor:

    - marks it ``openinference.span.kind=CHAIN`` so Phoenix and Arize AX stop
      showing it as UNKNOWN,
    - copies the Temporal identifiers into OpenInference ``metadata`` and sets
      ``session.id`` to the Workflow Id, so every trace of a Workflow Execution
      lands in one Arize session,
    - records the attempt number on ``RunActivity:*`` spans (the SDK emits one
      span per attempt but no attempt attribute).

    It also copies ``session.id`` / ``user.id`` from OpenTelemetry baggage onto
    any span that lacks them. The starter sets that baggage; Temporal carries it
    across the client/workflow/activity boundaries in its trace-context header.

    The processor only sets attributes, so it is replay-neutral: spans that are
    re-created during replay are never ended and therefore never exported.
    """

    def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
        attributes = dict(span.attributes or {})
        scope = span.instrumentation_scope.name if span.instrumentation_scope else ""
        if scope.startswith(TEMPORAL_SCOPE_PREFIX):
            span.set_attribute(
                SpanAttributes.OPENINFERENCE_SPAN_KIND,
                OpenInferenceSpanKindValues.CHAIN.value,
            )
            metadata: dict[str, Any] = {
                key: value
                for key, value in attributes.items()
                if key.startswith("temporal")
            }
            if activity.in_activity():
                attempt = activity.info().attempt
                span.set_attribute(ACTIVITY_ATTEMPT_ATTRIBUTE, attempt)
                metadata["temporalActivityAttempt"] = attempt
            if metadata:
                span.set_attribute(
                    SpanAttributes.METADATA,
                    json.dumps(metadata, sort_keys=True, default=str),
                )
            workflow_id = attributes.get("temporalWorkflowID")
            if workflow_id and SpanAttributes.SESSION_ID not in attributes:
                span.set_attribute(SpanAttributes.SESSION_ID, str(workflow_id))
                attributes[SpanAttributes.SESSION_ID] = workflow_id
        for key in (SpanAttributes.SESSION_ID, SpanAttributes.USER_ID):
            if key in attributes:
                continue
            value = baggage.get_baggage(key, parent_context)
            if value is not None:
                span.set_attribute(key, str(value))

    def on_end(self, span: ReadableSpan) -> None:
        pass

    def shutdown(self) -> None:
        pass

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        return True

Read Temporal traces in Arize​

With the pieces above, one ticket-triage interaction produces this trace. The kinds are what Arize renders.

ticket-triage                                    CHAIN  (root: session = Workflow Id, user, input, output, tags)
├─ StartWorkflow:TicketTriageWorkflow            CHAIN
│  └─ RunWorkflow:TicketTriageWorkflow           CHAIN
│     ├─ triage                                  CHAIN  (custom span from Workflow code)
│     │  ├─ StartActivity:classify_ticket → RunActivity:classify_ticket   CHAIN, attempt 1
│     │  │  └─ ChatCompletion                    LLM    (model, tokens, messages)
│     │  └─ StartActivity:lookup_account  → RunActivity:lookup_account
│     └─ StartActivity:draft_reply → RunActivity:draft_reply
│        └─ ChatCompletion                       LLM
└─ StartWorkflowUpdate:approve                   CHAIN
   ├─ ValidateUpdate:approve                     CHAIN
   └─ HandleUpdate:approve                       CHAIN
The ticket-triage trace in Phoenix

The ticket-triage trace in Phoenix

The Temporal Web UI and Arize answer different questions about the same execution. This table maps the concepts.

Temporal Web UIArize
Workflow Execution (one Workflow Id, one or more runs)Session (session.id = Workflow Id)
Run (RunWorkflow with temporalRunID)One RunWorkflow span per run in the trace
Event History and TimelineThe trace waterfall, in the same order
Pending Activities: attempt count and last failureOne RunActivity span per attempt, failed attempts with error status and temporal.activity.attempt
Update, Signal, and Query handlersHandleUpdate, HandleSignal, HandleQuery spans under the Client's span
Input and resultinput.value and output.value on the root span
Workers tab (which Worker ran what)Not recorded; Activity spans carry no Worker identity

Workflow replay​

Durable Execution re-runs Workflow code on Worker restarts and cache evictions to rebuild state. Event History records nothing for a replay, and neither does Arize: spans re-created during replay have the same deterministic identifiers as before and are never exported again, so a Workflow that replays ten times still produces one clean trace. You can confirm this with the sample's --replay-stress Worker flag, which disables the Workflow cache so every Workflow Task replays from the start of history.

Activity retries and Worker crashes​

Retries are real re-executions and stay visible: one StartActivity span with one RunActivity child per attempt, failed attempts carrying an error status and the exception. If a Worker dies while a Workflow waits, the RunWorkflow span is exported once, when the Workflow finishes on another Worker, and its duration covers the outage. If a Worker dies in the middle of an Activity attempt, that attempt's span was never ended and does not appear; the next attempt does, with the next attempt number. Spans that ended less than about half a second before a Worker process died can also be lost from the export buffer. Event History remains the source of truth.

Workflow Task failures​

A failed Workflow Task re-executes the code since the last completed Workflow Task. Spans that already ended inside the failed task are exported again with the same span identifiers, because the identifiers are deterministic, and Phoenix, which deduplicates spans by span identifier, keeps the first copy. The WorkflowTaskFailed event in the Temporal Web UI is where the failure is visible.

Continue-As-New, resets, and retries​

A Workflow that continues as new writes its current trace context into the new run, so the new run's RunWorkflow span nests under the previous run's span in the same trace. A reset or a retried Workflow reuses the original start context, so its RunWorkflow span appears as a second child of the same StartWorkflow span with its own temporalRunID. Because the session is the Workflow Id, all of these stay together in one Arize session.

Resources​