diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 7323ce83..c3570e64 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -15,6 +15,7 @@ # The AI SDK team owns the AI integration samples and their tests. We add # @temporalio/sdk too, so the SDK team can continue to manage repo-wide concerns. +/arize_tracing/ @temporalio/sdk @temporalio/ai-sdk /deepagents_plugin/ @temporalio/sdk @temporalio/ai-sdk /google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk /google_genai/ @temporalio/sdk @temporalio/ai-sdk @@ -24,6 +25,7 @@ /litellm_activity/ @temporalio/sdk @temporalio/ai-sdk /openai_agents/ @temporalio/sdk @temporalio/ai-sdk /strands_plugin/ @temporalio/sdk @temporalio/ai-sdk +/tests/arize_tracing/ @temporalio/sdk @temporalio/ai-sdk /tests/deepagents_plugin/ @temporalio/sdk @temporalio/ai-sdk /tests/google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk /tests/google_genai/ @temporalio/sdk @temporalio/ai-sdk diff --git a/README.md b/README.md index 2954d242..d55f5779 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,7 @@ Some examples require extra dependencies. See each sample's directory for specif * [hello update](hello/hello_update.py) - Send a request to and a response from a client to a workflow execution. * [activity_worker](activity_worker) - Use Python activities from a workflow in another language. +* [arize_tracing](arize_tracing) - Trace Temporal workflows and OpenAI Agents in Arize Phoenix or Arize AX with the OpenTelemetry plugin and OpenInference. * [batch_sliding_window](batch_sliding_window) - Batch processing with a sliding window of child workflows. * [bedrock](bedrock) - Orchestrate a chatbot with Amazon Bedrock. * [cloud_export_to_parquet](cloud_export_to_parquet) - Set up schedule workflow to process exported files on an hourly basis diff --git a/arize_tracing/.env.example b/arize_tracing/.env.example new file mode 100644 index 00000000..586ff27b --- /dev/null +++ b/arize_tracing/.env.example @@ -0,0 +1,30 @@ +# Copy to .env and adjust. Load with: set -a; source arize_tracing/.env; set +a + +# Arize Phoenix (default target). Matches arize_tracing/phoenix/docker-compose.yml; +# the UI, REST API, and OTLP/HTTP collector all live on this base URL. +PHOENIX_COLLECTOR_ENDPOINT=http://localhost:6006 +# Only when Phoenix authentication is on (Phoenix Cloud, or PHOENIX_ENABLE_AUTH): +# PHOENIX_API_KEY=... + +# Arize AX (SaaS). Setting both switches the exporter from Phoenix to Arize AX. +# ARIZE_SPACE_ID=... +# ARIZE_API_KEY=... +# EU region only: +# ARIZE_OTLP_ENDPOINT=https://otlp.eu-west-1a.arize.com/v1/traces + +# Project that receives the spans (Phoenix creates it on first use). +ARIZE_PROJECT_NAME=temporal-ticket-triage +# Reported as the OpenInference user.id on each trace by the starters. +ARIZE_DEMO_USER=demo-user + +# LLM. Any OpenAI-compatible endpoint works. +OPENAI_API_KEY=sk-... +MODEL_CLASSIFY=gpt-4o-mini +MODEL_DRAFT=gpt-4o-mini +MODEL_AGENT=gpt-4o-mini +# To use a local OpenAI-compatible gateway (for example a LiteLLM proxy) instead: +# OPENAI_BASE_URL=http://localhost:4000/v1 +# OPENAI_API_KEY= +# MODEL_CLASSIFY= +# MODEL_DRAFT= +# MODEL_AGENT= diff --git a/arize_tracing/README.md b/arize_tracing/README.md new file mode 100644 index 00000000..cde4ff11 --- /dev/null +++ b/arize_tracing/README.md @@ -0,0 +1,256 @@ +# Arize Tracing + +This sample shows the recommended way to get Temporal workflow traces into +[Arize](https://arize.com/) — the open-source [Arize Phoenix](https://arize.com/docs/phoenix) +or the [Arize AX](https://arize.com/docs/ax) platform — using Temporal's +[`OpenTelemetryPlugin`](https://python.temporal.io/temporalio.contrib.opentelemetry.OpenTelemetryPlugin.html) +plus a standard OTLP/HTTP exporter and the +[OpenInference](https://github.com/Arize-ai/openinference) semantic conventions +that Arize reads. No Arize SDK or Arize-specific plugin is involved, workflow +code stays deterministic and sandboxed, and traces are correctly nested, +correctly typed, and duplicate-free across replay and worker restarts. + +Contents: + +- **[ticket_triage/](ticket_triage/)** — the framework-agnostic pattern: an LLM + ticket-triage workflow (two LLM activities, one plain activity, one human + approval delivered as a workflow update). LLM calls run in activities and are + captured by the OpenInference OpenAI instrumentation. +- **[ticket_triage_agents/](ticket_triage_agents/)** — the same workflow built + with the [OpenAI Agents SDK](https://openai.github.io/openai-agents-python/) + through Temporal's `OpenAIAgentsPlugin(use_otel_instrumentation=True)`, so + Arize shows native AGENT, LLM, and TOOL spans. +- **[verify_trace.py](verify_trace.py)** — checks a trace through the Phoenix + REST API: whole-tree equality with span kinds, enrichment attributes, LLM + token usage, activity attempts, and no duplicates. +- **[phoenix/docker-compose.yml](phoenix/docker-compose.yml)** — pinned + self-hosted Phoenix, one container. +- **[telemetry.py](telemetry.py)** — the OpenTelemetry wiring: replay-safe + tracer provider, Phoenix/Arize AX exporter, the OpenInference enrichment + processor, and LLM instrumentation. + +## Prerequisites + +- Docker (for Phoenix) or `uvx`, a local Temporal server + (`temporal server start-dev`), and `uv`. +- An OpenAI-compatible LLM endpoint: either a real `OPENAI_API_KEY`, or any + OpenAI-compatible gateway via `OPENAI_BASE_URL`. + +## Run it + +```bash +# 1. Start Phoenix (UI, REST API, and OTLP collector on port 6006) +docker compose -f arize_tracing/phoenix/docker-compose.yml up -d +curl -sf http://localhost:6006/healthz && echo ok +# ...or without Docker: uvx --from "arize-phoenix==20.9.0" phoenix serve + +# 2. Install dependencies and set environment (repo root) +uv sync --group arize-tracing +cp arize_tracing/.env.example arize_tracing/.env # edit the LLM settings +set -a; source arize_tracing/.env; set +a + +# 3. Run the sample (two terminals, same environment) +uv run python -m arize_tracing.ticket_triage.worker +uv run python -m arize_tracing.ticket_triage.starter + +# 4. Verify the trace through the Phoenix API (uses the printed trace ID) +uv run python -m arize_tracing.verify_trace --trace-id +``` + +The starter prints a direct link to the trace in Phoenix (project +`temporal-ticket-triage`). You should see one trace shaped like this, with the +OpenInference kinds 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 +``` + +![Ticket triage trace in Phoenix](phoenix-ticket-triage.png) + +The Sessions tab groups traces by Temporal workflow ID (`session.id`), so all +interactions with one workflow execution appear as one session. Start the +starter with `--workflow-id ` more than once to see several traces in a +session. + +### With the OpenAI Agents SDK + +```bash +# Two worker processes (see "Known issue" below), then the starter +uv run python -m arize_tracing.ticket_triage_agents.worker --role workflows +uv run python -m arize_tracing.ticket_triage_agents.worker --role activities +uv run python -m arize_tracing.ticket_triage_agents.starter +uv run python -m arize_tracing.verify_trace --trace-id --scenario agents +``` + +Here the Agents SDK trace itself becomes the root, and the plugin's +OpenTelemetry bridge produces the OpenInference kinds directly: + +``` +Ticket triage agents AGENT (root; session, user, input, output) +├─ temporal:startWorkflow:TicketTriageAgentsWorkflow CHAIN +│ └─ temporal:executeWorkflow CHAIN +│ ├─ Agent workflow → Triage agent AGENT +│ │ ├─ turn CHAIN +│ │ │ └─ temporal:startActivity CHAIN (model call) +│ │ │ ├─ temporal:executeActivity → response LLM +│ │ │ └─ lookup_account TOOL (a Temporal activity as an agent tool) +│ │ │ └─ temporal:startActivity → temporal:executeActivity +│ │ └─ turn → temporal:startActivity → temporal:executeActivity → response LLM +│ └─ Agent workflow → Reply agent AGENT +│ └─ turn → temporal:startActivity → temporal:executeActivity → response LLM +└─ temporal:updateWorkflow CHAIN +``` + +(The tool span nests under the preceding model call's `temporal:startActivity` +span rather than directly under the turn because the plugin leaves that span +current in the Agents SDK scope after it finishes; see +[temporalio/sdk-python#1855](https://github.com/temporalio/sdk-python/issues/1855).) + +![Ticket triage agents trace in Phoenix](phoenix-ticket-triage-agents.png) + +**Known issue.** With `use_otel_instrumentation=True`, a single 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 +([temporalio/sdk-python#1852](https://github.com/temporalio/sdk-python/issues/1852)). +`verify_trace.py` reports this as spans referencing a missing parent. Running +the workflow and the activities in separate worker processes, as above, avoids +it; `worker.py` without `--role` runs both in one process, which is fine for +the framework-agnostic scenario but not for this one until the fix lands. + +## How replay, retries, and restarts show up + +Durable execution means workflow code re-executes (replays) on worker +restarts and cache evictions, and activities retry. The rule this sample +demonstrates: **replay produces no spans, real re-executions do**, which is +exactly what the Temporal Web UI shows too (Event History records nothing for a +replay, but it does record every activity attempt). + +| What happened | Temporal Web UI | Arize | +|---|---|---| +| Workflow replayed (worker restart, cache eviction, `--replay-stress`) | Nothing: Event History is unchanged | Nothing: spans re-created during replay have the same deterministic IDs and are never exported again | +| Activity retried | Pending Activities shows attempt N and the last failure | One `RunActivity` span per attempt under one `StartActivity`, failed attempts with error status, `temporal.activity.attempt` = 1, 2, ... | +| Worker died while the workflow waited | Workers tab / Workflow Task timeouts | The `RunWorkflow` span exports once, when the workflow finishes on another worker; its duration covers the outage | +| Worker died mid-activity | The attempt is not recorded; the next attempt is | The attempt's span was never ended, so it does not appear; the next attempt does | +| Workflow Task failed (bug, non-determinism) | `WorkflowTaskFailed` events | Spans that ended inside the failed task are exported again with the same span ID; Phoenix deduplicates by span ID and keeps one copy | +| Reset or retried workflow (new run) | New run under the same Workflow Id | Another `RunWorkflow` span with its own `temporalRunID`, same trace; the session (workflow ID) groups them | +| Continue-As-New | New run under the same Workflow Id | The new run's `RunWorkflow` span nests under the previous run's, same trace | + +Reproduce each case with the flags below; every run must still verify cleanly: + +```bash +# Replay stress: disable the workflow cache so EVERY workflow task replays +# the workflow from the start of history. The trace must be identical. +uv run python -m arize_tracing.ticket_triage.worker --replay-stress +uv run python -m arize_tracing.ticket_triage.starter +uv run python -m arize_tracing.verify_trace --trace-id + +# Worker restart mid-workflow: the starter waits 20s before sending the +# approval. Give the triage activities a few seconds to finish, then kill the +# worker while the workflow durably awaits approval; start a new worker and +# watch the workflow (and its trace) complete cleanly. +uv run python -m arize_tracing.ticket_triage.starter --pause-before-approval 20 +# ... after ~5s, ctrl+c the worker, then start it again in another terminal +uv run python -m arize_tracing.verify_trace --trace-id + +# Activity retry: classify_ticket fails on its first attempt. Arize shows two +# RunActivity:classify_ticket spans, the first with an error status. +uv run python -m arize_tracing.ticket_triage.worker --fail-first-attempt +uv run python -m arize_tracing.ticket_triage.starter +uv run python -m arize_tracing.verify_trace --trace-id --expect-attempts classify_ticket=2 + +# Worker crash mid-activity: classify_ticket heartbeats for 30s first. Kill +# the worker hard (kill -9) during that time and start a new one. The first +# attempt's span was never ended, so it is absent; the retry appears with +# attempt 2 after the heartbeat timeout. +uv run python -m arize_tracing.ticket_triage.worker --slow-classify 30 +uv run python -m arize_tracing.ticket_triage.starter +uv run python -m arize_tracing.verify_trace --trace-id --expect-attempt classify_ticket=2 + +# Reset: rerun a finished workflow from its first workflow task. The reset +# reapplies the original approval update, and the new run shares the trace as +# a second RunWorkflow span with its own run ID. +temporal workflow reset --workflow-id --type FirstWorkflowTask --reason demo +uv run python -m arize_tracing.verify_trace --workflow-id --expect-runs 2 +``` + +![Activity retry attempts in Phoenix](phoenix-retry-attempts.png) + +## Where spans come from + +| Span | Emitted by | Where it runs | +|---|---|---| +| `ticket-triage` (root) + OpenInference session/user/input/output | starter code | starter | +| `StartWorkflow:*`, `StartWorkflowUpdate:*` | `OpenTelemetryPlugin` | starter (client side) | +| `RunWorkflow:*`, `StartActivity:*`, `ValidateUpdate:*`, `HandleUpdate:*` | `OpenTelemetryPlugin` | worker (workflow) | +| `triage` | plain OpenTelemetry API in workflow code | worker (workflow) | +| `RunActivity:*` | `OpenTelemetryPlugin` | worker (activity) | +| `ChatCompletion` LLM spans | `openinference-instrumentation-openai` | worker (activity) | +| `openinference.span.kind`, `session.id`, `metadata`, `temporal.activity.attempt` on Temporal spans | `OpenInferenceEnrichmentProcessor` in `telemetry.py` | every process | + +The enrichment processor is optional but recommended: without it Phoenix and +Arize AX show Temporal's spans as UNKNOWN, do not group them into sessions, +and cannot tell activity attempts apart. + +## Where tracing works + +| Location | Works? | Notes | +|---|---|---| +| Activity bodies | ✅ | Plain OpenTelemetry + any OpenInference instrumentation, no restrictions. This is where LLM calls belong. | +| Workflow bodies | ✅ | Plain OpenTelemetry APIs are replay-safe under the plugin: deterministic span IDs, no re-export on replay. Spans export when they end; the `RunWorkflow` span exports when the run completes. | +| Signal/query/update handlers | ✅ | Handled by the plugin automatically (`HandleUpdate:*` etc.). | +| Client / starter code | ✅ | Standard OpenTelemetry; put the OpenInference trace-level attributes on your root span. | + +## Sending to Arize AX instead of Phoenix + +Set `ARIZE_SPACE_ID` and `ARIZE_API_KEY` (and `ARIZE_OTLP_ENDPOINT` for the EU +region) and `telemetry.py` exports to `https://otlp.arize.com/v1/traces` with +the same spans and attributes; `ARIZE_PROJECT_NAME` selects the project. +`verify_trace.py` reads Phoenix's REST API and does not apply to Arize AX; use +the AX UI or the `ax` CLI there. + +## Operational notes + +- Phoenix and Arize AX both accept OTLP/HTTP; this sample uses + `opentelemetry-exporter-otlp-proto-http`. +- Short-lived processes must flush: the starters and workers call + `force_flush()` on exit (see `telemetry.py`). +- The workflow ID doubles as the Arize session ID. A fresh ID per run is the + default; reuse one to group runs. +- `OTEL_SDK_DISABLED=true` turns off export without code changes. +- Ingestion is asynchronous; `verify_trace.py` polls until the trace is stable. +- The OpenAI Agents SDK bridge (`use_otel_instrumentation=True`) is Public + Preview in the Temporal SDK. Run the starter, the workflow worker, and the + activity worker as separate processes, as the sample does (see the known + issue above), and see `quiet_otel_context_detach_errors()` in `telemetry.py` + for a known log-noise issue. + +## Tests + +`tests/arize_tracing/` runs without Arize, Docker, or an LLM: mocked activities +(and the SDK's `TestModel` for the agents scenario), an in-memory span +exporter, a worker with the workflow cache disabled, whole-tree span +assertions, enrichment and retry-attempt assertions, and a `Replayer` pass +asserting that replaying the finished workflow's history emits zero new spans. + +```bash +uv run --group arize-tracing pytest tests/arize_tracing -v +``` + +## Using this outside samples-python + +The sample is self-contained: copy the `arize_tracing/` directory, change the +absolute imports (`arize_tracing.ticket_triage.activities` → +`ticket_triage.activities` or similar), and install the dependencies listed +under `arize-tracing` in this repo's `pyproject.toml`. diff --git a/arize_tracing/__init__.py b/arize_tracing/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/arize_tracing/phoenix-retry-attempts.png b/arize_tracing/phoenix-retry-attempts.png new file mode 100644 index 00000000..08405171 Binary files /dev/null and b/arize_tracing/phoenix-retry-attempts.png differ diff --git a/arize_tracing/phoenix-ticket-triage-agents.png b/arize_tracing/phoenix-ticket-triage-agents.png new file mode 100644 index 00000000..4417e990 Binary files /dev/null and b/arize_tracing/phoenix-ticket-triage-agents.png differ diff --git a/arize_tracing/phoenix-ticket-triage.png b/arize_tracing/phoenix-ticket-triage.png new file mode 100644 index 00000000..ee0ab832 Binary files /dev/null and b/arize_tracing/phoenix-ticket-triage.png differ diff --git a/arize_tracing/phoenix/docker-compose.yml b/arize_tracing/phoenix/docker-compose.yml new file mode 100644 index 00000000..23be6ccc --- /dev/null +++ b/arize_tracing/phoenix/docker-compose.yml @@ -0,0 +1,21 @@ +# Self-hosted Arize Phoenix for the arize_tracing samples. +# +# One container serves the UI, the REST API, and the OTLP/HTTP collector on +# port 6006 (the OTLP gRPC listener on 4317 is not published; this sample +# exports over HTTP). Traces persist in a named volume across restarts. +# +# Start: docker compose -f arize_tracing/phoenix/docker-compose.yml up -d +# Check: curl -sf http://localhost:6006/healthz +# UI: http://localhost:6006 +name: phoenix-demo +services: + phoenix: + image: arizephoenix/phoenix:version-20.9.0 + ports: + - "6006:6006" + environment: + PHOENIX_WORKING_DIR: /mnt/data + volumes: + - phoenix_data:/mnt/data +volumes: + phoenix_data: {} diff --git a/arize_tracing/telemetry.py b/arize_tracing/telemetry.py new file mode 100644 index 00000000..70bf0d3e --- /dev/null +++ b/arize_tracing/telemetry.py @@ -0,0 +1,268 @@ +"""Shared OpenTelemetry wiring for the arize_tracing samples. + +Arize Phoenix and Arize AX ingest OpenTelemetry traces over OTLP/HTTP and read +the OpenInference semantic conventions (``openinference.span.kind``, +``session.id``, ``input.value``, ``llm.*`` ...) to render LLM traces, so no +Arize SDK is needed. The same code targets both: Phoenix by default, Arize AX +when ``ARIZE_SPACE_ID`` and ``ARIZE_API_KEY`` are set. + +The tracer provider comes from ``temporalio.contrib.opentelemetry +.create_tracer_provider()``, which is safe to use inside workflow code: span +IDs are generated deterministically from workflow state and span export is +suppressed during replay, so a workflow that replays (worker restart, cache +eviction, host failover) never produces duplicate spans in Arize. +""" + +import json +import logging +import os +import urllib.request +from typing import Any, Optional + +from openinference.semconv.resource import ResourceAttributes +from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes +from opentelemetry import baggage, trace +from opentelemetry.context import Context +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from temporalio import activity +from temporalio.contrib.opentelemetry import create_tracer_provider + +logger = logging.getLogger(__name__) + +DEFAULT_PHOENIX_ENDPOINT = "http://localhost:6006" +DEFAULT_ARIZE_ENDPOINT = "https://otlp.arize.com/v1/traces" +DEFAULT_PROJECT_NAME = "temporal-ticket-triage" + +# Instrumentation scope prefix of the spans emitted by Temporal's +# OpenTelemetryPlugin (``temporalio.contrib.opentelemetry._otel_interceptor``). +TEMPORAL_SCOPE_PREFIX = "temporalio.contrib." + +# Attribute this sample adds to RunActivity spans (one span per attempt). +ACTIVITY_ATTEMPT_ATTRIBUTE = "temporal.activity.attempt" + + +def project_name() -> str: + """The Phoenix / Arize AX project that receives the spans.""" + return ( + os.environ.get("ARIZE_PROJECT_NAME") + or os.environ.get("PHOENIX_PROJECT_NAME") + or DEFAULT_PROJECT_NAME + ) + + +def phoenix_base_url() -> str: + """Phoenix base URL (UI, REST API, and OTLP/HTTP collector share it).""" + return os.environ.get( + "PHOENIX_COLLECTOR_ENDPOINT", DEFAULT_PHOENIX_ENDPOINT + ).rstrip("/") + + +def exporting_to_arize_ax() -> bool: + return bool(os.environ.get("ARIZE_SPACE_ID") and os.environ.get("ARIZE_API_KEY")) + + +# @@@SNIPSTART python-arize-tracing-exporter +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) + + +# @@@SNIPEND + + +# @@@SNIPSTART python-arize-tracing-enrichment-processor +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 + + +# @@@SNIPEND + + +# @@@SNIPSTART python-arize-tracing-setup +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) + + +# @@@SNIPEND + + +class _IgnoreContextDetachErrors(logging.Filter): + def filter(self, record: logging.LogRecord) -> bool: + return not record.getMessage().startswith("Failed to detach context") + + +def quiet_otel_context_detach_errors() -> None: + """Silence OpenTelemetry's "Failed to detach context" errors. + + The OpenInference OpenAI Agents processor attaches an OpenTelemetry + context when an Agents SDK span starts and detaches it when the span ends. + Temporal runs each workflow task in its own ``contextvars`` context, so for + agent spans that outlive a workflow task the detach happens in a different + context and OpenTelemetry logs an error with a traceback. Tracing is + unaffected (the spans are still correct and exported once), so processes + that use ``OpenAIAgentsPlugin(use_otel_instrumentation=True)`` call this + to keep their logs readable. + """ + logging.getLogger("opentelemetry.context").addFilter(_IgnoreContextDetachErrors()) + + +def force_flush() -> None: + """Flush any buffered spans to Arize immediately.""" + # The replay-safe provider implements force_flush but the base + # opentelemetry TracerProvider type does not declare it, hence getattr. + flush = getattr(trace.get_tracer_provider(), "force_flush", None) + if callable(flush): + flush() + + +def instrument_openai() -> None: + """Instrument the OpenAI client library once, in the worker process. + + Every OpenAI API call made from an activity then emits an OpenInference LLM + span (model, token counts, prompt and completion messages) that nests under + that activity's span. + """ + from openinference.instrumentation.openai import OpenAIInstrumentor + + OpenAIInstrumentor().instrument() + + +def phoenix_request(url: str) -> urllib.request.Request: + """A Phoenix REST API request, with the API key header when configured.""" + headers: dict[str, str] = {} + api_key = os.environ.get("PHOENIX_API_KEY") + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + return urllib.request.Request(url, headers=headers) + + +def trace_url(trace_id: str) -> str: + """Best-effort deep link to a trace in the Phoenix UI. + + Phoenix trace URLs need the project's global ID, which the REST API + returns; falls back to the projects page if the lookup fails. + """ + if exporting_to_arize_ax(): + return f"https://app.arize.com (project {project_name()!r}, trace {trace_id})" + base = phoenix_base_url() + try: + with urllib.request.urlopen( + phoenix_request(f"{base}/v1/projects"), timeout=5 + ) as response: + projects = json.loads(response.read()).get("data") or [] + for project in projects: + if project.get("name") == project_name(): + return f"{base}/projects/{project['id']}/traces/{trace_id}" + except (OSError, ValueError): + pass + return f"{base}/projects" diff --git a/arize_tracing/ticket_triage/README.md b/arize_tracing/ticket_triage/README.md new file mode 100644 index 00000000..6f1bb986 --- /dev/null +++ b/arize_tracing/ticket_triage/README.md @@ -0,0 +1,66 @@ +# Ticket Triage + +An LLM support-ticket triage workflow demonstrating the recommended +Temporal → Arize tracing setup (see [../README.md](../README.md) for the full +runbook). + +Flow: `classify_ticket` (LLM) and `lookup_account` (plain activity) run under +a custom `triage` span, the workflow then waits for a human decision delivered +as a workflow **update** (`approve`, with a validator), and on approval +`draft_reply` (LLM) produces the customer reply. + +| File | Purpose | +|---|---| +| `workflows.py` | `TicketTriageWorkflow` — deterministic, sandboxed; uses plain OpenTelemetry APIs for the `triage` span; `approve` update handler + validator | +| `activities.py` | The two LLM activities and the plain lookup activity; all I/O lives here. Demo hooks for the retry experiments | +| `worker.py` | Worker with `OpenTelemetryPlugin(add_temporal_spans=True)`; `--replay-stress`, `--fail-first-attempt`, `--slow-classify N` | +| `starter.py` | Opens the root span with the OpenInference trace-level attributes (session = workflow ID, user, input, output, metadata, tags), starts the workflow, sends the approval update, prints the Phoenix trace link; `--decline`, `--pause-before-approval N`, `--workflow-id`, `--user` | + +## Run + +With Phoenix up, dependencies synced, and the environment loaded (see +[../README.md](../README.md)): + +```bash +uv run python -m arize_tracing.ticket_triage.worker +uv run python -m arize_tracing.ticket_triage.starter +uv run python -m arize_tracing.verify_trace --trace-id +``` + +Variants: + +```bash +uv run python -m arize_tracing.ticket_triage.starter --decline +uv run python -m arize_tracing.verify_trace --trace-id --expect declined + +# Replay stress: every workflow task replays the workflow from history — +# the Arize trace must come out identical. +uv run python -m arize_tracing.ticket_triage.worker --replay-stress + +# Durability demo: park the workflow awaiting approval for 20s, kill and +# restart the worker meanwhile — one clean trace regardless. +uv run python -m arize_tracing.ticket_triage.starter --pause-before-approval 20 + +# Retry demo: the first classify_ticket attempt fails; two RunActivity spans. +uv run python -m arize_tracing.ticket_triage.worker --fail-first-attempt +uv run python -m arize_tracing.verify_trace --trace-id --expect-attempts classify_ticket=2 +``` + +## Expected trace + +``` +ticket-triage CHAIN (root; session = workflow id, user, input, output, tags) +├─ StartWorkflow:TicketTriageWorkflow CHAIN +│ └─ RunWorkflow:TicketTriageWorkflow CHAIN +│ ├─ triage CHAIN +│ │ ├─ StartActivity:classify_ticket → RunActivity:classify_ticket +│ │ │ └─ ChatCompletion LLM +│ │ └─ StartActivity:lookup_account → RunActivity:lookup_account +│ └─ StartActivity:draft_reply → RunActivity:draft_reply +│ └─ ChatCompletion LLM +└─ StartWorkflowUpdate:approve CHAIN + ├─ ValidateUpdate:approve CHAIN + └─ HandleUpdate:approve CHAIN +``` + +With `--decline`, the `draft_reply` subtree is absent. diff --git a/arize_tracing/ticket_triage/__init__.py b/arize_tracing/ticket_triage/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/arize_tracing/ticket_triage/activities.py b/arize_tracing/ticket_triage/activities.py new file mode 100644 index 00000000..11536a4a --- /dev/null +++ b/arize_tracing/ticket_triage/activities.py @@ -0,0 +1,170 @@ +"""Activities for the ticket triage sample. + +All LLM and I/O work happens here, in activities — never in workflow code. +The OpenAI client is instrumented process-wide (see ``telemetry.instrument_openai``), +so each API call below automatically emits an OpenInference LLM span nested +under the activity's span, which Arize renders with model, token usage, and +the prompt/completion messages. +""" + +import asyncio +import json +import os +from dataclasses import dataclass +from typing import Optional + +from openai import AsyncOpenAI +from temporalio import activity +from temporalio.exceptions import ApplicationError + +# Demo hooks, set from worker.py flags (module state in the worker process, +# never activity inputs, which are recorded in workflow history). +FAIL_FIRST_ATTEMPT = False # --fail-first-attempt: classify_ticket fails on attempt 1 +SLOW_CLASSIFY_SECONDS = ( + 0 # --slow-classify N: classify_ticket heartbeats N seconds first +) + + +@dataclass +class Ticket: + ticket_id: str + customer_email: str + subject: str + body: str + + +@dataclass +class Classification: + category: str + priority: str + + +@dataclass +class AccountInfo: + customer_email: str + account_name: str + plan: str + + +@dataclass +class DraftReplyInput: + ticket: Ticket + classification: Classification + account: AccountInfo + + +@dataclass +class ApprovalDecision: + approved: bool + reviewer: str + + +@dataclass +class TriageResult: + status: str + classification: Classification + reply: Optional[str] = None + + +CLASSIFY_PROMPT = ( + "You are a support ticket triage assistant. Classify the ticket and respond " + 'with ONLY a JSON object like {"category": "billing|bug|how-to|other", ' + '"priority": "low|normal|high"}.' +) + +DRAFT_PROMPT = ( + "You are a support agent. Draft a short (under 120 words), friendly reply to " + "the customer's ticket. Use the provided classification and account details." +) + + +def _openai_client() -> AsyncOpenAI: + # Configuration comes from the environment, never from activity inputs + # (activity inputs are recorded in workflow history and shown in the UI). + # max_retries=0 disables the OpenAI client's built-in retries — Temporal's + # activity retry policy owns retries, with full visibility in the UI and, + # through this sample, one RunActivity span per attempt in Arize. + return AsyncOpenAI( + base_url=os.environ.get("OPENAI_BASE_URL"), + api_key=os.environ.get("OPENAI_API_KEY"), + max_retries=0, + ) + + +def parse_classification(text: str) -> Classification: + try: + data = json.loads(text[text.index("{") : text.rindex("}") + 1]) + return Classification( + category=str(data.get("category", "other")).lower(), + priority=str(data.get("priority", "normal")).lower(), + ) + except ValueError: + return Classification(category="other", priority="normal") + + +# @@@SNIPSTART python-arize-tracing-activity +@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 "") + + +# @@@SNIPEND + + +@activity.defn +async def lookup_account(customer_email: str) -> AccountInfo: + """Look up the account name and plan for a customer email address.""" + # A deterministic, non-LLM activity: appears in Arize as a plain CHAIN span + # alongside the LLM spans produced inside the LLM activities. The + # ticket_triage_agents sample also exposes it to an agent as a tool, which + # is why it has a docstring: the Agents SDK uses it as the tool description. + known_accounts = { + "ada@acme.example": AccountInfo( + customer_email="ada@acme.example", + account_name="Acme Corp", + plan="enterprise", + ), + } + return known_accounts.get( + customer_email, + AccountInfo(customer_email=customer_email, account_name="Unknown", plan="free"), + ) + + +@activity.defn +async def draft_reply(input: DraftReplyInput) -> str: + context = ( + f"Ticket: {input.ticket.subject}\n{input.ticket.body}\n\n" + f"Category: {input.classification.category}, " + f"priority: {input.classification.priority}\n" + f"Account: {input.account.account_name} ({input.account.plan} plan)" + ) + response = await _openai_client().chat.completions.create( + model=os.environ.get("MODEL_DRAFT", "gpt-4o-mini"), + messages=[ + {"role": "system", "content": DRAFT_PROMPT}, + {"role": "user", "content": context}, + ], + timeout=30, + ) + return response.choices[0].message.content or "" diff --git a/arize_tracing/ticket_triage/starter.py b/arize_tracing/ticket_triage/starter.py new file mode 100644 index 00000000..b107e544 --- /dev/null +++ b/arize_tracing/ticket_triage/starter.py @@ -0,0 +1,155 @@ +"""Starter for the ticket triage sample. + +Opens one root span around the whole interaction (start workflow, send the +approval update, await the result) so that everything — including the +workflow, activity, and LLM spans produced on the worker — lands in a single +Arize trace. OpenInference trace-level attributes (session, user, input, +output, metadata, tags) are set on this root span; Arize reads a trace's +input and output from its root span. +""" + +import argparse +import asyncio +import dataclasses +import json +import os +import uuid + +from openinference.semconv.trace import ( + OpenInferenceMimeTypeValues, + OpenInferenceSpanKindValues, + SpanAttributes, +) +from opentelemetry import baggage, context, trace +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.envconfig import ClientConfig + +from arize_tracing.telemetry import force_flush, setup_tracing, trace_url +from arize_tracing.ticket_triage.activities import ApprovalDecision, Ticket +from arize_tracing.ticket_triage.workflows import TicketTriageWorkflow + +TASK_QUEUE = "arize-ticket-triage-task-queue" + + +async def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--decline", action="store_true", help="Decline the ticket") + parser.add_argument( + "--pause-before-approval", + type=int, + default=0, + metavar="SECONDS", + help="Wait before sending the approval update. While the workflow durably " + "awaits approval you can kill and restart the worker to see that the " + "Arize trace still comes out as a single clean tree.", + ) + parser.add_argument( + "--workflow-id", + help="Reuse a workflow ID. Each run of the same ID is a new Temporal run " + "and a new Arize trace, grouped under one Arize session (the ID).", + ) + parser.add_argument( + "--user", + default=os.environ.get("ARIZE_DEMO_USER", "demo-user"), + help="Reported as the OpenInference user.id", + ) + args = parser.parse_args() + approved = not args.decline + + setup_tracing("ticket-triage-starter") + + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect( + **config, + plugins=[OpenTelemetryPlugin(add_temporal_spans=True)], + ) + + # The workflow ID doubles as the Arize session ID, so all traces of one + # Workflow Execution group together in the Sessions view. A fresh ID per + # run is the default; --workflow-id reuses one on purpose. + workflow_id = args.workflow_id or f"ticket-triage-{uuid.uuid4().hex[:8]}" + + ticket = Ticket( + ticket_id="T-1001", + customer_email="ada@acme.example", + subject="Charged twice for the July invoice", + body=( + "Hi, my card statement shows two identical charges for our July " + "invoice. Can you check what happened and refund the duplicate?" + ), + ) + + # @@@SNIPSTART python-arize-tracing-root-span + # 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() + # @@@SNIPEND + + print(f"Workflow status: {result.status}") + if result.reply: + print(f"Drafted reply:\n{result.reply}") + print(f"Trace ID: {trace_id}") + print(f"Arize trace: {trace_url(trace_id)}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/arize_tracing/ticket_triage/worker.py b/arize_tracing/ticket_triage/worker.py new file mode 100644 index 00000000..48e840da --- /dev/null +++ b/arize_tracing/ticket_triage/worker.py @@ -0,0 +1,88 @@ +"""Worker for the ticket triage sample.""" + +import argparse +import asyncio +import logging + +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from arize_tracing.telemetry import force_flush, instrument_openai, setup_tracing +from arize_tracing.ticket_triage import activities +from arize_tracing.ticket_triage.activities import ( + classify_ticket, + draft_reply, + lookup_account, +) +from arize_tracing.ticket_triage.workflows import TicketTriageWorkflow + +TASK_QUEUE = "arize-ticket-triage-task-queue" + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + + parser = argparse.ArgumentParser() + parser.add_argument( + "--replay-stress", + action="store_true", + help="Disable the workflow cache so every workflow task replays the " + "workflow from the start of history — the harshest test that tracing " + "emits each span exactly once. Traces in Arize must look identical " + "with or without this flag.", + ) + parser.add_argument( + "--fail-first-attempt", + action="store_true", + help="Make classify_ticket fail on its first attempt so the trace shows " + "one RunActivity span per attempt (the first with an error status).", + ) + parser.add_argument( + "--slow-classify", + type=int, + default=0, + metavar="SECONDS", + help="Make classify_ticket heartbeat for SECONDS before calling the LLM, " + "so you can kill the worker mid-attempt and watch the retry.", + ) + args = parser.parse_args() + activities.FAIL_FIRST_ATTEMPT = args.fail_first_attempt + activities.SLOW_CLASSIFY_SECONDS = args.slow_classify + + # @@@SNIPSTART python-arize-tracing-worker + 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. + ) + # @@@SNIPEND + + mode = "replay-stress (workflow cache disabled)" if args.replay_stress else "normal" + print(f"Worker started (mode={mode}), ctrl+c to exit") + try: + await worker.run() + finally: + force_flush() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/arize_tracing/ticket_triage/workflows.py b/arize_tracing/ticket_triage/workflows.py new file mode 100644 index 00000000..ecd9eaea --- /dev/null +++ b/arize_tracing/ticket_triage/workflows.py @@ -0,0 +1,107 @@ +"""Ticket triage workflow with Arize tracing via OpenTelemetry. + +The workflow is fully deterministic and runs inside Temporal's standard +workflow sandbox. With ``OpenTelemetryPlugin`` registered on the client, +plain OpenTelemetry APIs work in workflow code: the ``triage`` span below is +created with the regular tracer, gets a deterministic span ID, and is never +re-exported on replay. +""" + +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.common import RetryPolicy + +# Bounded retries for the LLM activities so that a misconfigured endpoint or +# API key fails fast instead of retrying forever. Each retry attempt records +# its own RunActivity span in the trace (see --fail-first-attempt on the worker). +LLM_RETRY_POLICY = RetryPolicy( + maximum_attempts=3, initial_interval=timedelta(seconds=1) +) + +with workflow.unsafe.imports_passed_through(): + from openinference.semconv.trace import ( + OpenInferenceSpanKindValues, + SpanAttributes, + ) + from opentelemetry import trace + + from arize_tracing.ticket_triage.activities import ( + ApprovalDecision, + Classification, + DraftReplyInput, + Ticket, + TriageResult, + classify_ticket, + draft_reply, + lookup_account, + ) + + +# @@@SNIPSTART python-arize-tracing-workflow +@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") + + +# @@@SNIPEND diff --git a/arize_tracing/ticket_triage_agents/README.md b/arize_tracing/ticket_triage_agents/README.md new file mode 100644 index 00000000..32b3cdb5 --- /dev/null +++ b/arize_tracing/ticket_triage_agents/README.md @@ -0,0 +1,68 @@ +# Ticket Triage with the OpenAI Agents SDK + +The same ticket-triage flow as [../ticket_triage/](../ticket_triage/), built +with the [OpenAI Agents SDK](https://openai.github.io/openai-agents-python/) +and traced to Arize through Temporal's `OpenAIAgentsPlugin` (see +[../README.md](../README.md) for the full runbook). + +A triage agent looks up the customer's account through a Temporal activity +exposed as a tool (`activity_as_tool`) and classifies the ticket; the workflow +then waits for a human approval (a workflow update), and a reply agent drafts +the reply. Model calls run as Temporal activities. + +| File | Purpose | +|---|---| +| `plugin.py` | `OpenAIAgentsPlugin(use_otel_instrumentation=True, add_temporal_spans=True, ...)`: bridges Agents SDK tracing to OpenTelemetry through the OpenInference `openai-agents` instrumentation | +| `workflows.py` | `TicketTriageAgentsWorkflow` — two agents, one activity tool, `approve` update handler + validator | +| `worker.py` | Worker registering the workflow and the `lookup_account` activity; `--role workflows|activities|all`, `--replay-stress` | +| `starter.py` | Opens the Agents SDK trace (which becomes the root AGENT span), sets the OpenInference attributes on it, starts the workflow, sends the approval; `--decline`, `--pause-before-approval N`, `--workflow-id`, `--user` | + +Differences from the framework-agnostic scenario: + +- The Agents SDK trace is the root span (kind AGENT); Temporal operations + appear as `temporal:*` CHAIN spans created by the plugin, not as the + `OpenTelemetryPlugin`'s `StartWorkflow:*`/`RunWorkflow:*` spans. Do not add + `OpenTelemetryPlugin` as well: the two would produce overlapping spans. +- `setup_tracing()` must run before the plugin is constructed: the plugin + checks that the global tracer provider is Temporal's replay-safe provider. +- The plugin propagates trace and span IDs but not OpenTelemetry baggage, so + `session.id` and `user.id` live on the root span (enough for Phoenix's + Sessions view). +- Run the starter, a workflow worker (`--role workflows`), and an activity + worker (`--role activities`) as separate processes, as shown. A single + worker process that runs both exports the `temporal:startActivity` spans + with a parent that is not in the trace, so the model-call and tool subtrees + appear detached in Arize until + [temporalio/sdk-python#1852](https://github.com/temporalio/sdk-python/issues/1852) + is fixed. The bridge is Public Preview in the Temporal SDK. + +## Run + +```bash +uv run python -m arize_tracing.ticket_triage_agents.worker --role workflows +uv run python -m arize_tracing.ticket_triage_agents.worker --role activities +uv run python -m arize_tracing.ticket_triage_agents.starter +uv run python -m arize_tracing.verify_trace --trace-id --scenario agents + +# Replay stress and the durability demo work the same way as in ticket_triage: +uv run python -m arize_tracing.ticket_triage_agents.worker --role workflows --replay-stress +uv run python -m arize_tracing.ticket_triage_agents.starter --pause-before-approval 20 +``` + +## Expected trace + +``` +Ticket triage agents AGENT (root; session, user, input, output) +├─ temporal:startWorkflow:TicketTriageAgentsWorkflow CHAIN +│ └─ temporal:executeWorkflow CHAIN +│ ├─ Agent workflow → Triage agent AGENT +│ │ ├─ turn CHAIN +│ │ │ └─ temporal:startActivity CHAIN (model call) +│ │ │ ├─ temporal:executeActivity → response LLM +│ │ │ └─ lookup_account TOOL +│ │ │ └─ temporal:startActivity → temporal:executeActivity +│ │ └─ turn → temporal:startActivity → temporal:executeActivity → response LLM +│ └─ Agent workflow → Reply agent AGENT +│ └─ turn → temporal:startActivity → temporal:executeActivity → response LLM +└─ temporal:updateWorkflow CHAIN +``` diff --git a/arize_tracing/ticket_triage_agents/__init__.py b/arize_tracing/ticket_triage_agents/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/arize_tracing/ticket_triage_agents/plugin.py b/arize_tracing/ticket_triage_agents/plugin.py new file mode 100644 index 00000000..1aecddbf --- /dev/null +++ b/arize_tracing/ticket_triage_agents/plugin.py @@ -0,0 +1,36 @@ +"""Shared OpenAI Agents plugin configuration for the agents sample.""" + +from datetime import timedelta + +from temporalio.common import RetryPolicy +from temporalio.contrib.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin + +TASK_QUEUE = "arize-ticket-triage-agents-task-queue" + + +# @@@SNIPSTART python-arize-tracing-agents-plugin +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) + ), + ), + ) + + +# @@@SNIPEND diff --git a/arize_tracing/ticket_triage_agents/starter.py b/arize_tracing/ticket_triage_agents/starter.py new file mode 100644 index 00000000..26c12092 --- /dev/null +++ b/arize_tracing/ticket_triage_agents/starter.py @@ -0,0 +1,147 @@ +"""Starter for the ticket triage agents sample. + +The Agents SDK trace opened here becomes the root AGENT span of the Arize trace: +the plugin bridges Agents SDK tracing to OpenTelemetry, so the OpenInference +trace-level attributes (session, user, input, output, metadata, tags) go onto +that span instead of a separate OpenTelemetry root. +""" + +import argparse +import asyncio +import dataclasses +import json +import os +import uuid + +from agents import trace as agents_trace +from openinference.semconv.trace import OpenInferenceMimeTypeValues, SpanAttributes +from opentelemetry import trace +from temporalio.client import Client +from temporalio.envconfig import ClientConfig + +from arize_tracing.telemetry import ( + force_flush, + quiet_otel_context_detach_errors, + setup_tracing, + trace_url, +) +from arize_tracing.ticket_triage.activities import ApprovalDecision, Ticket +from arize_tracing.ticket_triage_agents.plugin import TASK_QUEUE, agents_plugin +from arize_tracing.ticket_triage_agents.workflows import ( + AgentTicketRequest, + TicketTriageAgentsWorkflow, +) + + +async def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--decline", action="store_true", help="Decline the ticket") + parser.add_argument( + "--pause-before-approval", + type=int, + default=0, + metavar="SECONDS", + help="Wait before sending the approval update, so you can kill and " + "restart the worker while the workflow durably awaits approval.", + ) + parser.add_argument("--workflow-id", help="Reuse a workflow ID (Arize session)") + parser.add_argument( + "--user", + default=os.environ.get("ARIZE_DEMO_USER", "demo-user"), + help="Reported as the OpenInference user.id", + ) + args = parser.parse_args() + approved = not args.decline + + setup_tracing("ticket-triage-agents-starter") + quiet_otel_context_detach_errors() + plugin = agents_plugin() + + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config, plugins=[plugin]) + + workflow_id = args.workflow_id or f"ticket-triage-agents-{uuid.uuid4().hex[:8]}" + ticket = Ticket( + ticket_id="T-1001", + customer_email="ada@acme.example", + subject="Charged twice for the July invoice", + body=( + "Hi, my card statement shows two identical charges for our July " + "invoice. Can you check what happened and refund the duplicate?" + ), + ) + request = AgentTicketRequest( + ticket=ticket, model=os.environ.get("MODEL_AGENT", "gpt-4o-mini") + ) + + # @@@SNIPSTART python-arize-tracing-agents-starter + # 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() + # @@@SNIPEND + + print(f"Workflow status: {result.status}") + if result.reply: + print(f"Drafted reply:\n{result.reply}") + print(f"Trace ID: {trace_id}") + print(f"Arize trace: {trace_url(trace_id)}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/arize_tracing/ticket_triage_agents/worker.py b/arize_tracing/ticket_triage_agents/worker.py new file mode 100644 index 00000000..ea4c8556 --- /dev/null +++ b/arize_tracing/ticket_triage_agents/worker.py @@ -0,0 +1,77 @@ +"""Worker for the ticket triage agents sample.""" + +import argparse +import asyncio +import logging + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from arize_tracing.telemetry import ( + force_flush, + quiet_otel_context_detach_errors, + setup_tracing, +) +from arize_tracing.ticket_triage.activities import lookup_account +from arize_tracing.ticket_triage_agents.plugin import TASK_QUEUE, agents_plugin +from arize_tracing.ticket_triage_agents.workflows import TicketTriageAgentsWorkflow + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + + parser = argparse.ArgumentParser() + parser.add_argument( + "--replay-stress", + action="store_true", + help="Disable the workflow cache so every workflow task replays the " + "workflow (and the agent loop) from the start of history. Traces in " + "Arize must look identical with or without this flag.", + ) + parser.add_argument( + "--role", + choices=["all", "workflows", "activities"], + default="all", + help="Run workflows and activities in this process (default), or only one " + "of them. Until temporalio/sdk-python#1852 is fixed, a single process that " + "runs both exports the temporal:startActivity spans with the wrong parent, " + "so the tool and model-call subtrees appear detached in Arize. Running one " + "worker with --role workflows and another with --role activities avoids it.", + ) + args = parser.parse_args() + run_workflows = args.role in ("all", "workflows") + run_activities = args.role in ("all", "activities") + + # @@@SNIPSTART python-arize-tracing-agents-worker + # The tracer provider must exist before the plugin is constructed. + setup_tracing("ticket-triage-agents-worker") + quiet_otel_context_detach_errors() + + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + # The plugin registers the model-call activity itself (unless this is a + # workflow-only worker). + client = await Client.connect( + **config, plugins=[agents_plugin(register_activities=run_activities)] + ) + + worker = Worker( + client, + task_queue=TASK_QUEUE, + workflows=[TicketTriageAgentsWorkflow] if run_workflows else [], + activities=[lookup_account] if run_activities else [], + max_cached_workflows=0 if args.replay_stress else 1000, + ) + # @@@SNIPEND + + mode = "replay-stress (workflow cache disabled)" if args.replay_stress else "normal" + print(f"Worker started (role={args.role}, mode={mode}), ctrl+c to exit") + try: + await worker.run() + finally: + force_flush() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/arize_tracing/ticket_triage_agents/workflows.py b/arize_tracing/ticket_triage_agents/workflows.py new file mode 100644 index 00000000..07c89feb --- /dev/null +++ b/arize_tracing/ticket_triage_agents/workflows.py @@ -0,0 +1,118 @@ +"""Ticket triage with the OpenAI Agents SDK, traced to Arize via OpenTelemetry. + +The agent loop runs inside the workflow, durably. Through ``OpenAIAgentsPlugin`` +every model call runs as a Temporal activity, and ``activity_as_tool`` exposes +the ``lookup_account`` activity to the agent as a tool. With +``use_otel_instrumentation=True`` the plugin turns the Agents SDK trace into +OpenInference spans (AGENT, LLM, TOOL, and CHAIN for the Temporal operations) +on Temporal's replay-safe tracer provider, so Arize renders the agent natively +and replay never duplicates a span. +""" + +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from agents import Agent, Runner + from temporalio.contrib import openai_agents as temporal_agents + + from arize_tracing.ticket_triage.activities import ( + ApprovalDecision, + Classification, + Ticket, + TriageResult, + lookup_account, + parse_classification, + ) + + +@dataclass +class AgentTicketRequest: + ticket: Ticket + # The model name is a workflow input rather than an environment lookup in + # workflow code, which keeps the workflow deterministic. + model: str = "gpt-4o-mini" + + +TRIAGE_INSTRUCTIONS = ( + "You are a support ticket triage agent. First look up the customer's account " + "with the lookup_account tool, then classify the ticket. Respond with ONLY a " + 'JSON object like {"category": "billing|bug|how-to|other", ' + '"priority": "low|normal|high"}.' +) + +REPLY_INSTRUCTIONS = ( + "You are a support agent. Draft a short (under 120 words), friendly reply to " + "the customer's ticket, using the classification you are given." +) + + +# @@@SNIPSTART python-arize-tracing-agents-workflow +@workflow.defn +class TicketTriageAgentsWorkflow: + def __init__(self) -> None: + self._approval: Optional[ApprovalDecision] = None + + @workflow.run + async def run(self, request: AgentTicketRequest) -> TriageResult: + ticket = request.ticket + triage_agent = Agent( + name="Triage agent", + model=request.model, + instructions=TRIAGE_INSTRUCTIONS, + tools=[ + # A Temporal activity as an agent tool: Arize shows a TOOL span + # wrapping the activity's own Temporal spans. + temporal_agents.workflow.activity_as_tool( + lookup_account, start_to_close_timeout=timedelta(seconds=10) + ) + ], + ) + triage = await Runner.run( + triage_agent, + input=( + f"Ticket from {ticket.customer_email}\n" + f"Subject: {ticket.subject}\n\n{ticket.body}" + ), + ) + classification: Classification = parse_classification(str(triage.final_output)) + + # 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_agent = Agent( + name="Reply agent", model=request.model, instructions=REPLY_INSTRUCTIONS + ) + reply = await Runner.run( + reply_agent, + input=( + f"Ticket: {ticket.subject}\n{ticket.body}\n\n" + f"Category: {classification.category}, " + f"priority: {classification.priority}" + ), + ) + return TriageResult( + status="replied", + classification=classification, + reply=str(reply.final_output), + ) + + @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") + + +# @@@SNIPEND diff --git a/arize_tracing/verify_trace.py b/arize_tracing/verify_trace.py new file mode 100644 index 00000000..1aed77e6 --- /dev/null +++ b/arize_tracing/verify_trace.py @@ -0,0 +1,483 @@ +"""Verify a ticket-triage trace in Arize Phoenix through its REST API. + +Fetches every span of the trace, rebuilds the tree, and deep-compares it +against the expected shape — including OpenInference span kinds — then checks +that Temporal spans carry the enrichment attributes, that every LLM span has a +model and token usage, and that no span was duplicated (running the worker +with --replay-stress surfaces replay-caused duplicates here, if there were +any). + +Usage: + python -m arize_tracing.verify_trace --trace-id + python -m arize_tracing.verify_trace --workflow-id + python -m arize_tracing.verify_trace --trace-id --expect declined + python -m arize_tracing.verify_trace --trace-id --expect-attempts classify_ticket=2 + python -m arize_tracing.verify_trace --workflow-id --expect-runs 2 + python -m arize_tracing.verify_trace --trace-id --scenario agents + +Stdlib-only on purpose so it is trivially copy-out-able. Reads the same +environment variables as the samples: PHOENIX_COLLECTOR_ENDPOINT, +PHOENIX_API_KEY, and ARIZE_PROJECT_NAME / PHOENIX_PROJECT_NAME. +""" + +import argparse +import json +import os +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any, Optional + +DEFAULT_PHOENIX_ENDPOINT = "http://localhost:6006" +DEFAULT_PROJECT_NAME = "temporal-ticket-triage" + +TEMPORAL_SPAN_PREFIXES = ( + "StartWorkflow:", + "RunWorkflow:", + "StartActivity:", + "RunActivity:", + "StartWorkflowUpdate:", + "ValidateUpdate:", + "HandleUpdate:", + "HandleSignal:", + "HandleQuery:", + "StartChildWorkflow:", + "SignalWorkflow:", + "QueryWorkflow:", +) + +# Expected span trees as (depth, name, kind) rows; siblings sorted by name, +# then by start time. LLM spans are normalized to "" (their name depends +# on the API used, for example "ChatCompletion"); their kind must be LLM. +EXPECTED_APPROVED = [ + (0, "ticket-triage", "CHAIN"), + (1, "StartWorkflow:TicketTriageWorkflow", "CHAIN"), + (2, "RunWorkflow:TicketTriageWorkflow", "CHAIN"), + (3, "StartActivity:draft_reply", "CHAIN"), + (4, "RunActivity:draft_reply", "CHAIN"), + (5, "", "LLM"), + (3, "triage", "CHAIN"), + (4, "StartActivity:classify_ticket", "CHAIN"), + (5, "RunActivity:classify_ticket", "CHAIN"), + (6, "", "LLM"), + (4, "StartActivity:lookup_account", "CHAIN"), + (5, "RunActivity:lookup_account", "CHAIN"), + (1, "StartWorkflowUpdate:approve", "CHAIN"), + (2, "HandleUpdate:approve", "CHAIN"), + (2, "ValidateUpdate:approve", "CHAIN"), +] +# The declined path never reaches draft_reply: drop its three rows explicitly. +EXPECTED_DECLINED = [ + (0, "ticket-triage", "CHAIN"), + (1, "StartWorkflow:TicketTriageWorkflow", "CHAIN"), + (2, "RunWorkflow:TicketTriageWorkflow", "CHAIN"), + (3, "triage", "CHAIN"), + (4, "StartActivity:classify_ticket", "CHAIN"), + (5, "RunActivity:classify_ticket", "CHAIN"), + (6, "", "LLM"), + (4, "StartActivity:lookup_account", "CHAIN"), + (5, "RunActivity:lookup_account", "CHAIN"), + (1, "StartWorkflowUpdate:approve", "CHAIN"), + (2, "HandleUpdate:approve", "CHAIN"), + (2, "ValidateUpdate:approve", "CHAIN"), +] + + +def _base_url() -> str: + return os.environ.get( + "PHOENIX_COLLECTOR_ENDPOINT", DEFAULT_PHOENIX_ENDPOINT + ).rstrip("/") + + +def _project() -> str: + return ( + os.environ.get("ARIZE_PROJECT_NAME") + or os.environ.get("PHOENIX_PROJECT_NAME") + or DEFAULT_PROJECT_NAME + ) + + +def _api_get(path: str, params: Optional[dict[str, Any]] = None) -> Any: + url = f"{_base_url()}/v1{path}" + if params: + url += "?" + urllib.parse.urlencode(params, doseq=True) + headers = {} + api_key = os.environ.get("PHOENIX_API_KEY") + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + request = urllib.request.Request(url, headers=headers) + with urllib.request.urlopen(request, timeout=15) as response: + return json.loads(response.read()) + + +def _spans(params: dict[str, Any]) -> list[dict[str, Any]]: + """Page through GET /v1/projects/{project}/spans with the given filters.""" + project = urllib.parse.quote(_project(), safe="") + spans: list[dict[str, Any]] = [] + cursor: Optional[str] = None + while True: + query = dict(params, limit=100) + if cursor: + query["cursor"] = cursor + page = _api_get(f"/projects/{project}/spans", query) + spans.extend(page.get("data") or []) + cursor = page.get("next_cursor") + if not cursor: + return spans + + +def _flatten(attributes: dict[str, Any], prefix: str = "") -> dict[str, Any]: + # Phoenix returns attributes with dotted keys; nested dicts are flattened + # defensively so both shapes read the same. + flat: dict[str, Any] = {} + for key, value in attributes.items(): + name = f"{prefix}{key}" + if isinstance(value, dict): + flat.update(_flatten(value, f"{name}.")) + else: + flat[name] = value + return flat + + +def _attrs(span: dict[str, Any]) -> dict[str, Any]: + return _flatten(span.get("attributes") or {}) + + +def _trace_ids_for_workflow(workflow_id: str, scenario: str) -> list[str]: + """Trace IDs whose RunWorkflow span belongs to the workflow, newest first.""" + name = ( + "RunWorkflow:TicketTriageWorkflow" + if scenario == "ticket-triage" + else "temporal:executeWorkflow" + ) + spans = _spans({"attribute": f"temporalWorkflowID:{workflow_id}", "name": name}) + if not spans: + spans = _spans({"attribute": f"session.id:{workflow_id}", "parent_id": "null"}) + spans.sort(key=lambda s: s["start_time"], reverse=True) + trace_ids: list[str] = [] + for span in spans: + trace_id = span["context"]["trace_id"] + if trace_id not in trace_ids: + trace_ids.append(trace_id) + return trace_ids + + +def _poll_stable_trace(trace_id: str, timeout_seconds: int) -> list[dict[str, Any]]: + """Poll until the trace exists and its span count is stable. + + Phoenix ingestion is asynchronous, so a freshly finished run may land + over a few seconds even though export already succeeded. + """ + deadline = time.monotonic() + timeout_seconds + previous_count = -1 + while time.monotonic() < deadline: + spans = _spans({"trace_id": trace_id}) + if spans and len(spans) == previous_count: + return spans + previous_count = len(spans) + time.sleep(2) + raise SystemExit( + f"FAIL: trace {trace_id} not fully ingested within {timeout_seconds}s" + ) + + +def _normalized_name(span: dict[str, Any]) -> str: + return "" if span.get("span_kind") == "LLM" else str(span["name"]) + + +def _build_tree(spans: list[dict[str, Any]]) -> list[tuple[int, str, str]]: + by_id = {s["context"]["span_id"]: s for s in spans} + children: dict[Optional[str], list[dict[str, Any]]] = {} + for span in spans: + parent = span.get("parent_id") + children.setdefault(parent if parent in by_id else None, []).append(span) + rows: list[tuple[int, str, str]] = [] + + def walk(span: dict[str, Any], depth: int) -> None: + rows.append((depth, _normalized_name(span), str(span.get("span_kind")))) + for child in sorted( + children.get(span["context"]["span_id"], []), + key=lambda s: (_normalized_name(s), s["start_time"]), + ): + walk(child, depth + 1) + + for root in sorted( + children.get(None, []), key=lambda s: (str(s["name"]), s["start_time"]) + ): + walk(root, 0) + return rows + + +def _print_tree(rows: list[tuple[int, str, str]]) -> None: + for depth, name, kind in rows: + print(f" {' ' * depth}{name} [{kind}]") + + +def _expected_tree(expect: str, attempts: dict[str, int]) -> list[tuple[int, str, str]]: + rows = list(EXPECTED_DECLINED if expect == "declined" else EXPECTED_APPROVED) + for activity_name, count in attempts.items(): + # Failed attempts have no LLM child; they precede the successful one. + index = rows.index( + next(row for row in rows if row[1] == f"RunActivity:{activity_name}") + ) + depth = rows[index][0] + rows[index:index] = [(depth, f"RunActivity:{activity_name}", "CHAIN")] * ( + count - 1 + ) + return rows + + +def _trace_url(trace_id: str) -> str: + try: + projects = _api_get("/projects").get("data") or [] + for project in projects: + if project.get("name") == _project(): + return f"{_base_url()}/projects/{project['id']}/traces/{trace_id}" + except (OSError, ValueError): + pass + return f"{_base_url()}/projects" + + +def _verify_common( + trace_id: str, + spans: list[dict[str, Any]], + failures: list[str], + args: argparse.Namespace, +) -> None: + # No duplicate spans (workflow replay must never re-emit spans). + all_ids = [s["context"]["span_id"] for s in spans] + if len(set(all_ids)) != len(all_ids): + failures.append("duplicate span ids present") + + # Every child points at a span that is part of the trace. A missing parent + # means a span was exported with the wrong parent, which Arize renders as a + # detached subtree. + span_ids = {s["context"]["span_id"] for s in spans} + orphans = [ + s for s in spans if s.get("parent_id") and s["parent_id"] not in span_ids + ] + if orphans: + names = sorted({s["name"] for s in orphans}) + failures.append( + f"{len(orphans)} span(s) reference a parent that is not in the trace: {names}" + ) + + # The root carries the OpenInference trace-level attributes. + roots = [s for s in spans if not s.get("parent_id")] + if len(roots) != 1: + failures.append(f"expected exactly one root span, found {len(roots)}") + else: + root_attrs = _attrs(roots[0]) + for key in ("session.id", "user.id", "input.value", "output.value"): + if not root_attrs.get(key): + failures.append(f"root span missing {key}") + if args.workflow_id and root_attrs.get("session.id") != args.workflow_id: + failures.append("root span session.id is not the workflow id") + + # Every LLM span has a model and token usage. + for span in spans: + if span.get("span_kind") != "LLM": + continue + attrs = _attrs(span) + if not attrs.get("llm.model_name"): + failures.append( + f"LLM span {span['context']['span_id']} missing llm.model_name" + ) + if ( + attrs.get("llm.token_count.prompt") is None + or attrs.get("llm.token_count.completion") is None + ): + failures.append( + f"LLM span {span['context']['span_id']} missing token counts" + ) + + +def _verify_ticket_triage( + args: argparse.Namespace, trace_id: str, spans: list[dict[str, Any]] +) -> list[str]: + failures: list[str] = [] + _verify_common(trace_id, spans, failures, args) + + actual = _build_tree(spans) + print(f"Trace {trace_id}: {len(spans)} spans") + _print_tree(actual) + + run_workflow_spans = [s for s in spans if s["name"].startswith("RunWorkflow:")] + if args.expect_runs == 1: + expected = _expected_tree(args.expect, args.expect_attempts) + if actual != expected: + failures.append("tree mismatch") + print(" Expected:") + _print_tree(expected) + else: + # Reset / retry experiments: several runs share one trace. + run_ids = {_attrs(s).get("temporalRunID") for s in run_workflow_spans} + if ( + len(run_workflow_spans) != args.expect_runs + or len(run_ids) != args.expect_runs + ): + failures.append( + f"expected {args.expect_runs} RunWorkflow spans with distinct run ids, " + f"found {len(run_workflow_spans)} ({len(run_ids)} run ids)" + ) + if args.expect_runs == 1 and len(run_workflow_spans) != 1: + failures.append( + f"expected exactly one RunWorkflow span, found {len(run_workflow_spans)}" + ) + + # Temporal spans carry the OpenInference enrichment. + for span in spans: + if not span["name"].startswith(TEMPORAL_SPAN_PREFIXES): + continue + attrs = _attrs(span) + if span.get("span_kind") != "CHAIN": + failures.append( + f"{span['name']} has kind {span.get('span_kind')}, expected CHAIN" + ) + if not attrs.get("session.id"): + failures.append(f"{span['name']} missing session.id") + if not attrs.get("metadata.temporalWorkflowID"): + failures.append(f"{span['name']} missing metadata.temporalWorkflowID") + + # Activity attempts: one RunActivity span per attempt, failed ones first. + for activity_name, count in args.expect_attempts.items(): + runs = sorted( + (s for s in spans if s["name"] == f"RunActivity:{activity_name}"), + key=lambda s: s["start_time"], + ) + starts = [s for s in spans if s["name"] == f"StartActivity:{activity_name}"] + if len(runs) != count: + failures.append( + f"expected {count} RunActivity:{activity_name} spans, found {len(runs)}" + ) + continue + if len(starts) != 1: + failures.append( + f"expected one StartActivity:{activity_name} span, found {len(starts)}" + ) + attempts = [_attrs(s).get("temporal.activity.attempt") for s in runs] + if attempts != list(range(1, count + 1)): + failures.append( + f"RunActivity:{activity_name} attempt attributes are {attempts}" + ) + for failed in runs[:-1]: + if failed.get("status_code") != "ERROR": + failures.append( + f"failed attempt of {activity_name} does not have ERROR status" + ) + if runs[-1].get("status_code") == "ERROR": + failures.append(f"final attempt of {activity_name} has ERROR status") + + # A worker that died mid-attempt never ended that attempt's span, so only + # the later attempt is present, carrying its attempt number. + for activity_name, attempt in args.expect_attempt.items(): + runs = [s for s in spans if s["name"] == f"RunActivity:{activity_name}"] + attempts = [_attrs(s).get("temporal.activity.attempt") for s in runs] + if attempts != [attempt]: + failures.append( + f"expected one RunActivity:{activity_name} span with attempt {attempt}, " + f"found attempts {attempts}" + ) + return failures + + +def _verify_agents( + args: argparse.Namespace, trace_id: str, spans: list[dict[str, Any]] +) -> list[str]: + """Looser checks for the OpenAI Agents scenario, whose span names come from + the Agents SDK and the OpenInference instrumentation.""" + failures: list[str] = [] + _verify_common(trace_id, spans, failures, args) + actual = _build_tree(spans) + print(f"Trace {trace_id}: {len(spans)} spans") + _print_tree(actual) + kinds = {str(s.get("span_kind")) for s in spans} + for kind in ("AGENT", "LLM", "TOOL", "CHAIN"): + if kind not in kinds: + failures.append(f"no span of kind {kind}") + if "UNKNOWN" in kinds: + failures.append("spans with UNKNOWN kind present") + executes = [s for s in spans if s["name"] == "temporal:executeWorkflow"] + if len(executes) != args.expect_runs: + failures.append( + f"expected {args.expect_runs} temporal:executeWorkflow spans, found {len(executes)}" + ) + return failures + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--trace-id", help="Trace ID printed by the starter") + parser.add_argument( + "--workflow-id", help="Workflow ID (resolved through the RunWorkflow span)" + ) + parser.add_argument( + "--scenario", choices=["ticket-triage", "agents"], default="ticket-triage" + ) + parser.add_argument( + "--expect", choices=["approved", "declined"], default="approved" + ) + parser.add_argument( + "--expect-attempts", + action="append", + default=[], + metavar="ACTIVITY=N", + help="Expect N RunActivity spans (attempts) for ACTIVITY, for example classify_ticket=2", + ) + parser.add_argument( + "--expect-attempt", + action="append", + default=[], + metavar="ACTIVITY=N", + help="Expect the single surviving RunActivity span for ACTIVITY to carry attempt N " + "(earlier attempts died with a killed worker and were never exported)", + ) + parser.add_argument( + "--expect-runs", + type=int, + default=1, + help="Number of workflow runs sharing the trace (after a reset), default 1", + ) + parser.add_argument("--timeout", type=int, default=60) + args = parser.parse_args() + if not args.trace_id and not args.workflow_id: + parser.error("one of --trace-id or --workflow-id is required") + args.expect_attempts = { + item.split("=", 1)[0]: int(item.split("=", 1)[1]) + for item in args.expect_attempts + } + args.expect_attempt = { + item.split("=", 1)[0]: int(item.split("=", 1)[1]) + for item in args.expect_attempt + } + + trace_id = args.trace_id + if not trace_id: + trace_ids = _trace_ids_for_workflow(args.workflow_id, args.scenario) + if not trace_ids: + print(f"FAIL: no trace found for workflow id {args.workflow_id}") + return 1 + if len(trace_ids) > 1: + print( + f"Workflow {args.workflow_id} has {len(trace_ids)} traces; verifying the newest" + ) + trace_id = trace_ids[0] + + spans = _poll_stable_trace(trace_id, args.timeout) + verify = _verify_agents if args.scenario == "agents" else _verify_ticket_triage + failures = verify(args, trace_id, spans) + print(f"Phoenix: {_trace_url(trace_id)}") + if failures: + for failure in failures: + print(f"FAIL: {failure}") + return 1 + print("PASS: tree shape, span kinds, enrichment, and LLM spans all match") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index 3bcb3ffd..f01c1a72 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,16 @@ dev = [ "pytest-pretty>=1.3.0", "poethepoet>=0.36.0", ] +arize-tracing = [ + "openai>=1.4.0", + "openai-agents>=0.19.0", + "temporalio[opentelemetry,openai-agents]>=1.32.0,<2", + # Phoenix and Arize AX ingest OTLP/HTTP; this sample never uses gRPC. + "opentelemetry-exporter-otlp-proto-http>=1.30.0,<2", + "openinference-semantic-conventions>=0.1.30", + "openinference-instrumentation-openai>=0.1.52", + "openinference-instrumentation-openai-agents>=1.6.1,<3", +] bedrock = ["boto3>=1.34.92,<2"] deepagents = [ "deepagents>=0.6.12,<0.7 ; python_version >= '3.11'", diff --git a/tests/arize_tracing/__init__.py b/tests/arize_tracing/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/arize_tracing/conftest.py b/tests/arize_tracing/conftest.py new file mode 100644 index 00000000..3a542c9e --- /dev/null +++ b/tests/arize_tracing/conftest.py @@ -0,0 +1,19 @@ +from typing import Iterator + +import opentelemetry.trace +import pytest +from opentelemetry.util._once import Once + + +@pytest.fixture +def reset_otel_tracer_provider() -> Iterator[None]: + """Reset global OpenTelemetry tracer provider state around a test. + + OpenTelemetry only allows the global tracer provider to be set once per + process; tests that install their own provider need this reset. + """ + opentelemetry.trace._TRACER_PROVIDER_SET_ONCE = Once() + opentelemetry.trace._TRACER_PROVIDER = None + yield + opentelemetry.trace._TRACER_PROVIDER_SET_ONCE = Once() + opentelemetry.trace._TRACER_PROVIDER = None diff --git a/tests/arize_tracing/helpers.py b/tests/arize_tracing/helpers.py new file mode 100644 index 00000000..f60cf040 --- /dev/null +++ b/tests/arize_tracing/helpers.py @@ -0,0 +1,30 @@ +"""Test helpers for the arize_tracing sample tests.""" + +from typing import Iterable, List, Optional + +from opentelemetry.sdk.trace import ReadableSpan + + +def dump_spans( + spans: Iterable[ReadableSpan], + *, + parent_id: Optional[int] = None, + indent_depth: int = 0, +) -> List[str]: + """Render spans as an indented tree, one line per span. + + Mirrors the helper used by the Temporal Python SDK's own OpenTelemetry + tests so span hierarchies can be asserted with a whole-tree equality. + """ + ret: List[str] = [] + for span in spans: + if (not span.parent and parent_id is None) or ( + span.parent and span.parent.span_id == parent_id + ): + ret.append(f"{' ' * indent_depth}{span.name}") + ret += dump_spans( + spans, + parent_id=span.context.span_id if span.context else None, + indent_depth=indent_depth + 1, + ) + return ret diff --git a/tests/arize_tracing/test_ticket_triage.py b/tests/arize_tracing/test_ticket_triage.py new file mode 100644 index 00000000..b25a1ac8 --- /dev/null +++ b/tests/arize_tracing/test_ticket_triage.py @@ -0,0 +1,327 @@ +"""Tests for the ticket triage sample. + +These run without Arize or an LLM: the LLM activities are mocked (each opens a +custom span to prove trace context propagates into activities) and spans are +captured with an in-memory exporter. The worker runs with the workflow cache +disabled, so every workflow task replays the workflow from the start of +history — asserting the whole span tree with deep equality proves spans are +emitted exactly once despite replay. +""" + +import json +import uuid +from typing import Any, Sequence + +import opentelemetry.trace +from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes +from opentelemetry import baggage, context, trace +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode +from temporalio import activity +from temporalio.client import Client, WorkflowHandle +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider +from temporalio.exceptions import ApplicationError +from temporalio.worker import Replayer, Worker + +from arize_tracing.telemetry import ( + ACTIVITY_ATTEMPT_ATTRIBUTE, + TEMPORAL_SCOPE_PREFIX, + OpenInferenceEnrichmentProcessor, +) +from arize_tracing.ticket_triage.activities import ( + AccountInfo, + ApprovalDecision, + Classification, + DraftReplyInput, + Ticket, +) +from arize_tracing.ticket_triage.workflows import TicketTriageWorkflow +from tests.arize_tracing.helpers import dump_spans + +TICKET = Ticket( + ticket_id="T-1", + customer_email="ada@acme.example", + subject="Charged twice", + body="Please refund the duplicate charge.", +) +TEST_USER = "test-user" + + +@activity.defn(name="classify_ticket") +async def classify_ticket_mocked(ticket: Ticket) -> Classification: + with trace.get_tracer(__name__).start_as_current_span("mock llm classify"): + return Classification(category="billing", priority="high") + + +@activity.defn(name="classify_ticket") +async def classify_ticket_fails_once(ticket: Ticket) -> Classification: + with trace.get_tracer(__name__).start_as_current_span("mock llm classify"): + if activity.info().attempt == 1: + raise ApplicationError("simulated transient failure", type="Simulated") + return Classification(category="billing", priority="high") + + +@activity.defn(name="lookup_account") +async def lookup_account_mocked(customer_email: str) -> AccountInfo: + with trace.get_tracer(__name__).start_as_current_span("mock account lookup"): + return AccountInfo( + customer_email=customer_email, account_name="Acme Corp", plan="enterprise" + ) + + +@activity.defn(name="draft_reply") +async def draft_reply_mocked(input: DraftReplyInput) -> str: + with trace.get_tracer(__name__).start_as_current_span("mock llm draft"): + return "Sorry about that - refund on the way." + + +def _install_in_memory_exporter() -> InMemorySpanExporter: + exporter = InMemorySpanExporter() + provider = create_tracer_provider() + provider.add_span_processor(OpenInferenceEnrichmentProcessor()) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + opentelemetry.trace.set_tracer_provider(provider) + return exporter + + +def _client_with_plugin(client: Client) -> Client: + config = client.config() + config["plugins"] = [OpenTelemetryPlugin(add_temporal_spans=True)] + return Client(**config) + + +async def _run_interaction( + client: Client, task_queue: str, workflow_id: str, approved: bool +) -> WorkflowHandle[Any, Any]: + """Mirror the starter: baggage + a root span around start, update, result.""" + ctx = baggage.set_baggage(SpanAttributes.SESSION_ID, workflow_id) + ctx = baggage.set_baggage(SpanAttributes.USER_ID, TEST_USER, context=ctx) + token = context.attach(ctx) + try: + with trace.get_tracer(__name__).start_as_current_span( + "ticket-triage test", + attributes={ + SpanAttributes.OPENINFERENCE_SPAN_KIND: OpenInferenceSpanKindValues.CHAIN.value, + SpanAttributes.SESSION_ID: workflow_id, + SpanAttributes.USER_ID: TEST_USER, + }, + ): + handle = await client.start_workflow( + TicketTriageWorkflow.run, + TICKET, + id=workflow_id, + task_queue=task_queue, + ) + await handle.execute_update( + TicketTriageWorkflow.approve, + ApprovalDecision(approved=approved, reviewer="test-reviewer"), + ) + await handle.result() + finally: + context.detach(token) + return handle + + +EXPECTED_APPROVED = [ + "ticket-triage test", + " StartWorkflow:TicketTriageWorkflow", + " RunWorkflow:TicketTriageWorkflow", + " triage", + " StartActivity:classify_ticket", + " RunActivity:classify_ticket", + " mock llm classify", + " StartActivity:lookup_account", + " RunActivity:lookup_account", + " mock account lookup", + " StartActivity:draft_reply", + " RunActivity:draft_reply", + " mock llm draft", + " StartWorkflowUpdate:approve", + " ValidateUpdate:approve", + " HandleUpdate:approve", +] + + +def _temporal_spans(spans: Sequence[ReadableSpan]) -> list[ReadableSpan]: + return [ + s + for s in spans + if s.instrumentation_scope + and s.instrumentation_scope.name.startswith(TEMPORAL_SCOPE_PREFIX) + ] + + +async def test_spans_emitted_exactly_once_under_replay_stress( + client: Client, reset_otel_tracer_provider: Any +) -> None: + exporter = _install_in_memory_exporter() + new_client = _client_with_plugin(client) + task_queue = f"tq-{uuid.uuid4()}" + workflow_id = f"ticket-triage-test-{uuid.uuid4()}" + + async with Worker( + new_client, + task_queue=task_queue, + workflows=[TicketTriageWorkflow], + activities=[classify_ticket_mocked, lookup_account_mocked, draft_reply_mocked], + # Disable the workflow cache: every workflow task replays the workflow + # from the start of history. Tracing must still emit each span once. + max_cached_workflows=0, + ): + handle = await _run_interaction( + new_client, task_queue, workflow_id, approved=True + ) + + spans = exporter.get_finished_spans() + assert dump_spans(spans) == EXPECTED_APPROVED + span_ids = [s.context.span_id for s in spans if s.context] + assert len(set(span_ids)) == len(span_ids) + trace_ids = {s.context.trace_id for s in spans if s.context} + assert len(trace_ids) == 1 + + # Replaying the finished workflow's real history must emit zero new spans. + history = await handle.fetch_history() + before = len(exporter.get_finished_spans()) + replayer = Replayer( + workflows=[TicketTriageWorkflow], + plugins=[OpenTelemetryPlugin(add_temporal_spans=True)], + ) + await replayer.replay_workflow(history) + assert len(exporter.get_finished_spans()) == before + + +async def test_declined_path_span_tree( + client: Client, reset_otel_tracer_provider: Any +) -> None: + exporter = _install_in_memory_exporter() + new_client = _client_with_plugin(client) + task_queue = f"tq-{uuid.uuid4()}" + + async with Worker( + new_client, + task_queue=task_queue, + workflows=[TicketTriageWorkflow], + activities=[classify_ticket_mocked, lookup_account_mocked, draft_reply_mocked], + max_cached_workflows=0, + ): + await _run_interaction( + new_client, task_queue, f"ticket-triage-test-{uuid.uuid4()}", approved=False + ) + + expected = [ + line + for line in EXPECTED_APPROVED + if "draft" not in line # declined tickets never reach draft_reply + ] + assert dump_spans(exporter.get_finished_spans()) == expected + + +async def test_openinference_enrichment( + client: Client, reset_otel_tracer_provider: Any +) -> None: + exporter = _install_in_memory_exporter() + new_client = _client_with_plugin(client) + task_queue = f"tq-{uuid.uuid4()}" + workflow_id = f"ticket-triage-test-{uuid.uuid4()}" + + async with Worker( + new_client, + task_queue=task_queue, + workflows=[TicketTriageWorkflow], + activities=[classify_ticket_mocked, lookup_account_mocked, draft_reply_mocked], + max_cached_workflows=0, + ): + await _run_interaction(new_client, task_queue, workflow_id, approved=True) + + spans = exporter.get_finished_spans() + temporal_spans = _temporal_spans(spans) + assert len(temporal_spans) == 11 # 1 client start, 1 client update, 9 worker-side + + # Every Temporal span gets an OpenInference kind, the Workflow Id as the + # Arize session, and the Temporal identifiers as OpenInference metadata. + for span in temporal_spans: + attributes = dict(span.attributes or {}) + assert attributes[SpanAttributes.OPENINFERENCE_SPAN_KIND] == "CHAIN", span.name + assert attributes[SpanAttributes.SESSION_ID] == workflow_id, span.name + assert attributes[SpanAttributes.USER_ID] == TEST_USER, span.name + metadata = json.loads(str(attributes[SpanAttributes.METADATA])) + assert metadata["temporalWorkflowID"] == workflow_id, span.name + + # One RunActivity span per attempt, each stamped with its attempt number. + run_activity_spans = [ + s for s in temporal_spans if s.name.startswith("RunActivity:") + ] + assert len(run_activity_spans) == 3 + for span in run_activity_spans: + attributes = dict(span.attributes or {}) + assert attributes[ACTIVITY_ATTEMPT_ATTRIBUTE] == 1 + assert ( + json.loads(str(attributes[SpanAttributes.METADATA]))[ + "temporalActivityAttempt" + ] + == 1 + ) + + # Baggage set in the starter reaches spans created inside workflow and + # activity code (through Temporal's trace-context header), so Arize can + # group and filter LLM spans by session and user too. + for name in ( + "triage", + "mock llm classify", + "mock account lookup", + "mock llm draft", + ): + span = next(s for s in spans if s.name == name) + attributes = dict(span.attributes or {}) + assert attributes[SpanAttributes.SESSION_ID] == workflow_id, name + assert attributes[SpanAttributes.USER_ID] == TEST_USER, name + triage = next(s for s in spans if s.name == "triage") + assert ( + dict(triage.attributes or {})[SpanAttributes.OPENINFERENCE_SPAN_KIND] == "CHAIN" + ) + + +async def test_activity_retry_attempts_are_separate_spans( + client: Client, reset_otel_tracer_provider: Any +) -> None: + exporter = _install_in_memory_exporter() + new_client = _client_with_plugin(client) + task_queue = f"tq-{uuid.uuid4()}" + + async with Worker( + new_client, + task_queue=task_queue, + workflows=[TicketTriageWorkflow], + activities=[ + classify_ticket_fails_once, + lookup_account_mocked, + draft_reply_mocked, + ], + max_cached_workflows=0, + ): + await _run_interaction( + new_client, task_queue, f"ticket-triage-test-{uuid.uuid4()}", approved=True + ) + + spans = exporter.get_finished_spans() + expected = list(EXPECTED_APPROVED) + # The failed first attempt adds one RunActivity span (with its inner mock + # span) under the single StartActivity span. + index = expected.index(" RunActivity:classify_ticket") + expected[index:index] = [ + " RunActivity:classify_ticket", + " mock llm classify", + ] + assert dump_spans(spans) == expected + + attempts = [s for s in spans if s.name == "RunActivity:classify_ticket"] + assert [dict(s.attributes or {})[ACTIVITY_ATTEMPT_ATTRIBUTE] for s in attempts] == [ + 1, + 2, + ] + assert attempts[0].status.status_code == StatusCode.ERROR + assert any(event.name == "exception" for event in attempts[0].events) + assert attempts[1].status.status_code != StatusCode.ERROR + assert len([s for s in spans if s.name == "StartActivity:classify_ticket"]) == 1 diff --git a/tests/arize_tracing/test_ticket_triage_agents.py b/tests/arize_tracing/test_ticket_triage_agents.py new file mode 100644 index 00000000..95a8d191 --- /dev/null +++ b/tests/arize_tracing/test_ticket_triage_agents.py @@ -0,0 +1,191 @@ +"""Tests for the ticket triage agents sample. + +These run without Arize or an LLM: the model is mocked with the SDK's +``TestModel`` (a tool call, then the classification, then the reply), spans are +captured with an in-memory exporter, and the worker runs with the workflow +cache disabled so every workflow task replays the agent loop from history. + +Client and worker share one process here, and therefore one OpenInference +processor. The worker-side trace replicas that Temporal creates for context +propagation register under the same trace id as the client's trace, so in a +single process the client's root span is not the span that ends when the +Agents SDK trace ends. The sample runs starter and worker as separate +processes, where the root exports correctly; these assertions therefore focus +on the worker-side tree. +""" + +import json +import uuid +from datetime import timedelta +from typing import Any, Optional, Sequence + +import opentelemetry.trace +from agents import trace as agents_trace +from openinference.semconv.trace import SpanAttributes +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from temporalio.client import Client +from temporalio.contrib.openai_agents import ModelActivityParameters +from temporalio.contrib.openai_agents.testing import ( + AgentEnvironment, + ResponseBuilders, + TestModel, +) +from temporalio.contrib.opentelemetry import create_tracer_provider +from temporalio.worker import Replayer, Worker + +from arize_tracing.telemetry import ( + OpenInferenceEnrichmentProcessor, + quiet_otel_context_detach_errors, +) +from arize_tracing.ticket_triage.activities import ( + ApprovalDecision, + Ticket, + lookup_account, +) +from arize_tracing.ticket_triage_agents.workflows import ( + AgentTicketRequest, + TicketTriageAgentsWorkflow, +) + +TICKET = Ticket( + ticket_id="T-1", + customer_email="ada@acme.example", + subject="Charged twice", + body="Please refund the duplicate charge.", +) + + +def _model_responses() -> list[Any]: + return [ + ResponseBuilders.tool_call( + json.dumps({"customer_email": TICKET.customer_email}), "lookup_account" + ), + ResponseBuilders.output_message('{"category": "billing", "priority": "high"}'), + ResponseBuilders.output_message("Sorry about that - refund on the way."), + ] + + +def _install_in_memory_exporter() -> InMemorySpanExporter: + exporter = InMemorySpanExporter() + provider = create_tracer_provider() + provider.add_span_processor(OpenInferenceEnrichmentProcessor()) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + opentelemetry.trace.set_tracer_provider(provider) + return exporter + + +def _kind(span: ReadableSpan) -> str: + return str(dict(span.attributes or {}).get(SpanAttributes.OPENINFERENCE_SPAN_KIND)) + + +def _ancestors(span: ReadableSpan, spans: Sequence[ReadableSpan]) -> list[str]: + by_id = {s.context.span_id: s for s in spans if s.context} + names: list[str] = [] + parent: Optional[Any] = span.parent + while parent is not None and parent.span_id in by_id: + current = by_id[parent.span_id] + names.append(current.name) + parent = current.parent + return names + + +def _only(spans: Sequence[ReadableSpan], name: str) -> ReadableSpan: + matches = [s for s in spans if s.name == name] + assert len(matches) == 1, f"expected one {name!r} span, found {len(matches)}" + return matches[0] + + +async def test_agent_spans_emitted_exactly_once_under_replay_stress( + client: Client, reset_otel_tracer_provider: Any +) -> None: + quiet_otel_context_detach_errors() + # The provider must exist before the plugin is constructed. + exporter = _install_in_memory_exporter() + workflow_id = f"ticket-triage-agents-test-{uuid.uuid4()}" + task_queue = f"tq-{uuid.uuid4()}" + + async with AgentEnvironment( + model=TestModel.returning_responses(_model_responses()), + use_otel_instrumentation=True, + add_temporal_spans=True, + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=30) + ), + ) as env: + new_client = env.applied_on_client(client) + async with Worker( + new_client, + task_queue=task_queue, + workflows=[TicketTriageAgentsWorkflow], + activities=[lookup_account], + max_cached_workflows=0, + ): + with env.openai_agents_plugin.tracing_context(): + with agents_trace("Ticket triage agents test", group_id=workflow_id): + handle = await new_client.start_workflow( + TicketTriageAgentsWorkflow.run, + AgentTicketRequest(ticket=TICKET, model="test-model"), + id=workflow_id, + task_queue=task_queue, + ) + await handle.execute_update( + TicketTriageAgentsWorkflow.approve, + ApprovalDecision(approved=True, reviewer="test-reviewer"), + ) + result = await handle.result() + + assert result.status == "replied" + assert result.classification.category == "billing" + + spans = exporter.get_finished_spans() + + # Every span has an OpenInference kind, so nothing renders as UNKNOWN. + kinds: dict[str, set[str]] = {} + for span in spans: + kinds.setdefault(_kind(span), set()).add(span.name) + assert "None" not in kinds and "UNKNOWN" not in kinds + assert {"Triage agent", "Reply agent"} <= kinds["AGENT"] + assert "lookup_account" in kinds["TOOL"] + assert { + "temporal:startWorkflow:TicketTriageAgentsWorkflow", + "temporal:executeWorkflow", + "temporal:startActivity", + "temporal:executeActivity", + "temporal:updateWorkflow", + } <= kinds["CHAIN"] + + # Exactly once, in one trace, despite the workflow cache being disabled. + span_ids = [s.context.span_id for s in spans if s.context] + assert len(set(span_ids)) == len(span_ids) + assert len({s.context.trace_id for s in spans if s.context}) == 1 + + # Worker-side tree: the agents nest under the workflow execution, and the + # tool call wraps the activity that implements it. + execute = _only(spans, "temporal:executeWorkflow") + assert _ancestors(execute, spans)[:1] == [ + "temporal:startWorkflow:TicketTriageAgentsWorkflow" + ] + for agent in ("Triage agent", "Reply agent"): + assert "temporal:executeWorkflow" in _ancestors(_only(spans, agent), spans) + tool = _only(spans, "lookup_account") + assert "Triage agent" in _ancestors(tool, spans) + tool_activity_starts = [ + s + for s in spans + if s.name == "temporal:startActivity" + and s.parent + and tool.context + and s.parent.span_id == tool.context.span_id + ] + assert len(tool_activity_starts) == 1 + + # Replaying the finished workflow's real history must emit zero new spans. + history = await handle.fetch_history() + before = len(exporter.get_finished_spans()) + replayer = Replayer( + workflows=[TicketTriageAgentsWorkflow], plugins=[env.openai_agents_plugin] + ) + await replayer.replay_workflow(history) + assert len(exporter.get_finished_spans()) == before diff --git a/uv.lock b/uv.lock index 8469affe..12a7980b 100644 --- a/uv.lock +++ b/uv.lock @@ -244,14 +244,14 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "anyio", marker = "python_full_version < '3.11'" }, - { name = "distro", marker = "python_full_version < '3.11'" }, - { name = "docstring-parser", marker = "python_full_version < '3.11'" }, - { name = "httpx", marker = "python_full_version < '3.11'" }, - { name = "jiter", marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, - { name = "sniffio", marker = "python_full_version < '3.11'" }, - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/fb/57/0b758b08cf4606c94d63a997d67a0063f7438efbaf81cfedd0d7c0c69d67/anthropic-0.103.1.tar.gz", hash = "sha256:21c12f4fc0fdd87a2e80d58479cd0af640062b3cfb82bbfa01c7977acd4defeb", size = 848877, upload-time = "2026-05-19T15:43:27.698Z" } wheels = [ @@ -270,14 +270,14 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "anyio", marker = "python_full_version >= '3.11'" }, - { name = "distro", marker = "python_full_version >= '3.11'" }, - { name = "docstring-parser", marker = "python_full_version >= '3.11'" }, - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "jiter", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "sniffio", marker = "python_full_version >= '3.11'" }, - { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/0f/ca/3cb2c20ee729736fbd4546d5d8b67e818288529fe70cb7a80dbf80aef70b/anthropic-0.121.0.tar.gz", hash = "sha256:e79d6e08ab3376602fc9a70d4d5ea3540817c76cf7e16658bed790834e1833d6", size = 1013292, upload-time = "2026-08-07T17:11:07.241Z" } wheels = [ @@ -709,12 +709,12 @@ name = "deepagents" version = "0.6.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain", version = "1.3.15", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langchain-anthropic", version = "1.5.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langchain-google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langsmith", version = "0.8.18", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "wcmatch", marker = "python_full_version >= '3.11'" }, + { name = "langchain", version = "1.3.15", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-anthropic", version = "1.5.5", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-google-genai" }, + { name = "langsmith", version = "0.8.18", source = { registry = "https://pypi.org/simple" } }, + { name = "wcmatch" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e0/db/a6acdc72a9e90c3f07ed10de35c951734a02d4facb693bb59684ad368801/deepagents-0.6.12.tar.gz", hash = "sha256:1f281c0bc5a63132f62e2ee345c1dc593b23188da6e23016401f6879fbe54b5f", size = 211364, upload-time = "2026-06-25T17:26:52.775Z" } wheels = [ @@ -767,7 +767,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1689,9 +1689,9 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "langgraph", version = "1.2.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, + { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" } }, + { name = "langgraph", version = "1.2.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/11/e5/6350e77a9e2764eaafcb2d581cbf0b800f53c6bc98fdf5ebc85f3a931ded/langchain-1.3.1.tar.gz", hash = "sha256:bc283c220233230f48b8e50ab1fbf1b688bcb206d933fa448d40a9b143177f62", size = 581329, upload-time = "2026-05-15T18:14:55.368Z" } wheels = [ @@ -1710,9 +1710,9 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langgraph", version = "1.2.11", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "langgraph", version = "1.2.11", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/6a/1c/b84579174a8e82ed79f4c3e0cd5a7f2323facc5ccd4d1b8390e7d175b663/langchain-1.3.15.tar.gz", hash = "sha256:ab4b775b9703f7e37babe0b325dbbaef25573bda60ecf79f7850bc875f252795", size = 665047, upload-time = "2026-08-11T19:10:52.455Z" } wheels = [ @@ -1727,9 +1727,9 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "anthropic", version = "0.103.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, + { name = "anthropic", version = "0.103.1", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/fa/e3/d2f9dec95602524b1cfb4be2747ba5bc38d32501b2a56cb4bcb76e80bb45/langchain_anthropic-1.4.3.tar.gz", hash = "sha256:f8a2442463c0629b1b3110eaeaa56fdbdc87df2a802f8c7f5ecf611eb4874ec8", size = 685219, upload-time = "2026-05-03T17:33:27.118Z" } wheels = [ @@ -1748,9 +1748,9 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "anthropic", version = "0.121.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "anthropic", version = "0.121.0", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/38/ce/5fdff0c55c4711da9d87a4a0a066d0b14b633402dc9d441214cc64889be9/langchain_anthropic-1.5.5.tar.gz", hash = "sha256:e8697f13b93fe95b7c7c17679f5d0143c239a8fcf45c0f498349c54482322dc9", size = 720572, upload-time = "2026-08-11T19:16:38.842Z" } wheels = [ @@ -1765,15 +1765,15 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "jsonpatch", marker = "python_full_version < '3.11'" }, - { name = "langchain-protocol", version = "0.0.15", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "langsmith", version = "0.8.9", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "packaging", marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, - { name = "pyyaml", marker = "python_full_version < '3.11'" }, - { name = "tenacity", marker = "python_full_version < '3.11'" }, - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, - { name = "uuid-utils", marker = "python_full_version < '3.11'" }, + { name = "jsonpatch" }, + { name = "langchain-protocol", version = "0.0.15", source = { registry = "https://pypi.org/simple" } }, + { name = "langsmith", version = "0.8.9", source = { registry = "https://pypi.org/simple" } }, + { name = "packaging" }, + { name = "pydantic" }, + { name = "pyyaml" }, + { name = "tenacity" }, + { name = "typing-extensions" }, + { name = "uuid-utils" }, ] sdist = { url = "https://files.pythonhosted.org/packages/59/de/679a53472c25860837e32c0442c962fa86e95317a36460e2c9d5c91b17c2/langchain_core-1.4.0.tar.gz", hash = "sha256:1dc341eed802ed9c117c0df3923c991e5e9e226571e5725c194eeb5bd93d1a7f", size = 920260, upload-time = "2026-05-11T18:42:35.919Z" } wheels = [ @@ -1792,15 +1792,15 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "jsonpatch", marker = "python_full_version >= '3.11'" }, - { name = "langchain-protocol", version = "0.0.18", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langsmith", version = "0.8.18", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "packaging", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "pyyaml", marker = "python_full_version >= '3.11'" }, - { name = "tenacity", marker = "python_full_version >= '3.11'" }, - { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, - { name = "uuid-utils", marker = "python_full_version >= '3.11'" }, + { name = "jsonpatch" }, + { name = "langchain-protocol", version = "0.0.18", source = { registry = "https://pypi.org/simple" } }, + { name = "langsmith", version = "0.8.18", source = { registry = "https://pypi.org/simple" } }, + { name = "packaging" }, + { name = "pydantic" }, + { name = "pyyaml" }, + { name = "tenacity" }, + { name = "typing-extensions" }, + { name = "uuid-utils" }, ] sdist = { url = "https://files.pythonhosted.org/packages/9e/18/20c3eec05ccf2fff8e553866bec3bb2f92880aea3cb878603e5a854bd5c0/langchain_core-1.5.4.tar.gz", hash = "sha256:aa76104f30b6c7305f292cb2c364e67cb52c321940ae812d7969471dce32a89a", size = 980540, upload-time = "2026-08-11T18:02:52.239Z" } wheels = [ @@ -1812,10 +1812,10 @@ name = "langchain-google-genai" version = "4.3.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "filetype", marker = "python_full_version >= '3.11'" }, - { name = "google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "filetype" }, + { name = "google-genai" }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/11/98/39b62fb50236fc582449beb96e0392d108bd7baac7ac6158d656bfac1561/langchain_google_genai-4.3.3.tar.gz", hash = "sha256:f051b98aaf223cf9092fc27c280dfec63070fc0022dc513640b2da138c9fa2f0", size = 286010, upload-time = "2026-08-10T18:34:26.499Z" } wheels = [ @@ -1830,7 +1830,7 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" } wheels = [ @@ -1849,7 +1849,7 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d2/59/b5959aea96faa9146e2e49a7a22882b3528c62efafe9a6a95beab30c2305/langchain_protocol-0.0.18.tar.gz", hash = "sha256:ec3e11782f1ed0c9db38e5a9ed01b0e7a0d3fba406faa8aef6594b73c56a63e6", size = 6150, upload-time = "2026-06-18T17:08:26.959Z" } wheels = [ @@ -1864,12 +1864,12 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "langgraph-checkpoint", marker = "python_full_version < '3.11'" }, - { name = "langgraph-prebuilt", marker = "python_full_version < '3.11'" }, - { name = "langgraph-sdk", version = "0.3.14", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, - { name = "xxhash", marker = "python_full_version < '3.11'" }, + { name = "langchain-core", version = "1.4.0", source = { registry = "https://pypi.org/simple" } }, + { name = "langgraph-checkpoint" }, + { name = "langgraph-prebuilt" }, + { name = "langgraph-sdk", version = "0.3.14", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, + { name = "xxhash" }, ] sdist = { url = "https://files.pythonhosted.org/packages/58/61/d5d25e783035aa307d289b37e082258a6061c0fb4caa4a284f3bf1e87169/langgraph-1.2.0.tar.gz", hash = "sha256:4a9baaf62afc5d5f63144a50095140a34b9aa9b7cea695d25326d564775348e7", size = 690248, upload-time = "2026-05-12T03:46:39.164Z" } wheels = [ @@ -1888,12 +1888,12 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langgraph-checkpoint", marker = "python_full_version >= '3.11'" }, - { name = "langgraph-prebuilt", marker = "python_full_version >= '3.11'" }, - { name = "langgraph-sdk", version = "0.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "xxhash", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "langgraph-checkpoint" }, + { name = "langgraph-prebuilt" }, + { name = "langgraph-sdk", version = "0.4.2", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic" }, + { name = "xxhash" }, ] sdist = { url = "https://files.pythonhosted.org/packages/56/0d/c8e7ee98896659e1b6555db0ab115a9ca899844744645d5d894032bab1d7/langgraph-1.2.11.tar.gz", hash = "sha256:9ecfe11e50d338b34b15cf4d8a442642de103e8ae6971320efba84e4542eb363", size = 725753, upload-time = "2026-08-11T14:00:36.945Z" } wheels = [ @@ -1936,8 +1936,8 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "httpx", marker = "python_full_version < '3.11'" }, - { name = "orjson", marker = "python_full_version < '3.11'" }, + { name = "httpx" }, + { name = "orjson" }, ] sdist = { url = "https://files.pythonhosted.org/packages/02/f1/134046c20bc4a4a15d410d1d21c9e298a3e9923777b4cc867b8669bc636b/langgraph_sdk-0.3.14.tar.gz", hash = "sha256:acd1674c538e97f3cdaa610f6dd7e34bc9bad30167f0ccc482dcd563325e81f5", size = 198162, upload-time = "2026-05-05T18:40:03.524Z" } wheels = [ @@ -1956,11 +1956,11 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "langchain-protocol", version = "0.0.18", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "orjson", marker = "python_full_version >= '3.11'" }, - { name = "websockets", marker = "python_full_version >= '3.11'" }, + { name = "httpx" }, + { name = "langchain-core", version = "1.5.4", source = { registry = "https://pypi.org/simple" } }, + { name = "langchain-protocol", version = "0.0.18", source = { registry = "https://pypi.org/simple" } }, + { name = "orjson" }, + { name = "websockets" }, ] sdist = { url = "https://files.pythonhosted.org/packages/b4/2b/bd8ac26d4e97f6df88ef05ce5b6a38945a3903e1025d926f4752aa88aa97/langgraph_sdk-0.4.2.tar.gz", hash = "sha256:b88f0f5f6328ac0680d6790614a905b2bcfa257f2276dba4e38f0e86db0aa738", size = 348327, upload-time = "2026-06-01T17:51:19.856Z" } wheels = [ @@ -1975,16 +1975,16 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "httpx", marker = "python_full_version < '3.11'" }, - { name = "orjson", marker = "python_full_version < '3.11' and platform_python_implementation != 'PyPy'" }, - { name = "packaging", marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, - { name = "requests", marker = "python_full_version < '3.11'" }, - { name = "requests-toolbelt", marker = "python_full_version < '3.11'" }, - { name = "uuid-utils", marker = "python_full_version < '3.11'" }, - { name = "websockets", marker = "python_full_version < '3.11'" }, - { name = "xxhash", marker = "python_full_version < '3.11'" }, - { name = "zstandard", marker = "python_full_version < '3.11'" }, + { name = "httpx" }, + { name = "orjson", marker = "platform_python_implementation != 'PyPy'" }, + { name = "packaging" }, + { name = "pydantic" }, + { name = "requests" }, + { name = "requests-toolbelt" }, + { name = "uuid-utils" }, + { name = "websockets" }, + { name = "xxhash" }, + { name = "zstandard" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e4/dd/f4c8a12987318e505b10760d30c3c2d45e8dc87ba8f47a004c753a9e7b35/langsmith-0.8.9.tar.gz", hash = "sha256:f16e37fcd5a8a2d4db30eae0e399a866a65ce5cc86218825c59409ed57a3bf53", size = 4428684, upload-time = "2026-06-03T17:56:09.448Z" } wheels = [ @@ -2003,16 +2003,16 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "orjson", marker = "python_full_version >= '3.11' and platform_python_implementation != 'PyPy'" }, - { name = "packaging", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "requests", marker = "python_full_version >= '3.11'" }, - { name = "requests-toolbelt", marker = "python_full_version >= '3.11'" }, - { name = "uuid-utils", marker = "python_full_version >= '3.11'" }, - { name = "websockets", marker = "python_full_version >= '3.11'" }, - { name = "xxhash", marker = "python_full_version >= '3.11'" }, - { name = "zstandard", marker = "python_full_version >= '3.11'" }, + { name = "httpx" }, + { name = "orjson", marker = "platform_python_implementation != 'PyPy'" }, + { name = "packaging" }, + { name = "pydantic" }, + { name = "requests" }, + { name = "requests-toolbelt" }, + { name = "uuid-utils" }, + { name = "websockets" }, + { name = "xxhash" }, + { name = "zstandard" }, ] sdist = { url = "https://files.pythonhosted.org/packages/9a/d9/a6681aa9847bbbc5ec21abe20a5e233b94e5edcfe39624db607ac7e8ccb4/langsmith-0.8.18.tar.gz", hash = "sha256:32dde9c0e67e053e0fb738921fc8ced768af7b8fa83d7a0e3fd63597cf8776dd", size = 4526988, upload-time = "2026-06-19T13:12:17.123Z" } wheels = [ @@ -2725,7 +2725,7 @@ wheels = [ [[package]] name = "openinference-instrumentation" -version = "0.1.54" +version = "0.1.63" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "openinference-semantic-conventions" }, @@ -2733,9 +2733,9 @@ dependencies = [ { name = "opentelemetry-sdk" }, { name = "wrapt" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/c1/cc/62c1175ee7edc2cbdf95b5b73e0b0f305e759d407bf1bc353ff30a763365/openinference_instrumentation-0.1.54.tar.gz", hash = "sha256:9af9817bb38816ed32856fb4cd813c1a5d9f530ab589c3473b37e06cf406ab28", size = 33938, upload-time = "2026-06-30T19:23:15.648Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ab/41/2b15aeb99c3211cb224957e9c91d3b83384e9809486b6c5e673c29d56bde/openinference_instrumentation-0.1.63.tar.gz", hash = "sha256:74d038a62288c994be24a1983b49769382f0042a6cbc071967e6bd384fc522d0", size = 43736, upload-time = "2026-09-10T05:43:27.077Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/38/e3/c7aa7bb4845e0cfdf477ff87f9a3ec0cd2aa55a34a9f31be84d724cabbb7/openinference_instrumentation-0.1.54-py3-none-any.whl", hash = "sha256:8bc991865c90c804ac9983ef93aa6081a7a2397dd113d3d38b5465cca17892bb", size = 41197, upload-time = "2026-06-30T19:23:14.315Z" }, + { url = "https://files.pythonhosted.org/packages/e0/d9/b6c1ea9235b5c1b8652331cc812026854599adfc035441b949abe3c734a5/openinference_instrumentation-0.1.63-py3-none-any.whl", hash = "sha256:45fe2b228652be47eda61a970a87901dff87700d764edcff317421b031615799", size = 51463, upload-time = "2026-09-10T05:43:25.762Z" }, ] [[package]] @@ -2756,13 +2756,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9b/ff/10d75f37dd006072db725d36a30b343386d6404526eeda877bbe37157abe/openinference_instrumentation_openai-0.1.52-py3-none-any.whl", hash = "sha256:aa96d41cb755e0d9b3d5a09331d991b7424c017c107d0bf196b03d3e5a7dc475", size = 30532, upload-time = "2026-06-11T17:13:34.327Z" }, ] +[[package]] +name = "openinference-instrumentation-openai-agents" +version = "2.4.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "openinference-instrumentation" }, + { name = "openinference-semantic-conventions" }, + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, + { name = "wrapt" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c7/1a/5493d664baf3893f68c3441b5e7291c2f027f63d3d5d4f58e28389529419/openinference_instrumentation_openai_agents-2.4.1.tar.gz", hash = "sha256:4bcfe61d39ba914b510832096af7030472fbac0df66854bd116f60f67642adee", size = 35334, upload-time = "2026-09-10T05:43:20.312Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/79/08/65ea6e5dd68ba810e33a5c60673764efa65ddb5409807862acb5b3a2504a/openinference_instrumentation_openai_agents-2.4.1-py3-none-any.whl", hash = "sha256:c43857c9738bc785ab482d91644e0dde5be9275e3038a2000b85b3a8e1d38237", size = 38508, upload-time = "2026-09-10T05:43:18.828Z" }, +] + [[package]] name = "openinference-semantic-conventions" -version = "0.1.30" +version = "0.1.37" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/2c/51/8ba1182ee86fc79793d5ff2d11e7fdcda10ded2d01f3e46ca6fcf0568213/openinference_semantic_conventions-0.1.30.tar.gz", hash = "sha256:81fece76e09c83789e35c393b8b30523481eeabf1008745b955631a53e3221d9", size = 13391, upload-time = "2026-05-22T21:10:44.065Z" } +sdist = { url = "https://files.pythonhosted.org/packages/0b/14/b97e71d5aa73a838e9e2550e85d7a2e1a7f89831ac4de5425f93bdfcf896/openinference_semantic_conventions-0.1.37.tar.gz", hash = "sha256:ff1d7568bb75427320221ec0eee222a52a020b868e1846505be43af2b76808fd", size = 14575, upload-time = "2026-09-10T01:38:17.587Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b6/76/5b7e78cf0de38589b821bbe8e9c29c59a6e76edfb980488d0854cbb90f7c/openinference_semantic_conventions-0.1.30-py3-none-any.whl", hash = "sha256:36d946d3f95f699b7c4b12324ae9c1f02d6c7750df11eece56aa159cff430b3d", size = 10911, upload-time = "2026-05-22T21:10:43.04Z" }, + { url = "https://files.pythonhosted.org/packages/19/d7/8ddfc990d8af3a9ee328c71435ecbed85df3f6c3b363492be20a1a799777/openinference_semantic_conventions-0.1.37-py3-none-any.whl", hash = "sha256:5adb0b05c6fdd6a6cf1b5bbd5a370460ea1cb076184d46bb895b6b156a036722", size = 11692, upload-time = "2026-09-10T01:38:16.485Z" }, ] [[package]] @@ -4547,6 +4565,15 @@ dependencies = [ ] [package.dev-dependencies] +arize-tracing = [ + { name = "openai" }, + { name = "openai-agents" }, + { name = "openinference-instrumentation-openai" }, + { name = "openinference-instrumentation-openai-agents" }, + { name = "openinference-semantic-conventions" }, + { name = "opentelemetry-exporter-otlp-proto-http" }, + { name = "temporalio", extra = ["openai-agents", "opentelemetry"] }, +] bedrock = [ { name = "boto3" }, ] @@ -4668,6 +4695,15 @@ requires-dist = [ ] [package.metadata.requires-dev] +arize-tracing = [ + { name = "openai", specifier = ">=1.4.0" }, + { name = "openai-agents", specifier = ">=0.19.0" }, + { name = "openinference-instrumentation-openai", specifier = ">=0.1.52" }, + { name = "openinference-instrumentation-openai-agents", specifier = ">=1.6.1,<3" }, + { name = "openinference-semantic-conventions", specifier = ">=0.1.30" }, + { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.30.0,<2" }, + { name = "temporalio", extras = ["opentelemetry", "openai-agents"], specifier = ">=1.32.0,<2" }, +] bedrock = [{ name = "boto3", specifier = ">=1.34.92,<2" }] cloud-export-to-parquet = [ { name = "boto3", specifier = ">=1.34.89,<2" }, @@ -5221,7 +5257,7 @@ name = "wcmatch" version = "11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "bracex", marker = "python_full_version >= '3.11'" }, + { name = "bracex" }, ] sdist = { url = "https://files.pythonhosted.org/packages/16/25/1da725838132221e33568973da484ff43813662ccc06ebf7f6e3abddfcd5/wcmatch-11.0.tar.gz", hash = "sha256:55d95c2447789712774b198ceec72939e88b5618f1f8f0a9b605bf7740b63b96", size = 141360, upload-time = "2026-07-10T05:50:24.183Z" } wheels = [