From b94089b2cb75ce407b4834c80f29ffa7ba511ee6 Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 23 Jul 2026 15:47:38 -0500 Subject: [PATCH 1/5] Add langfuse_tracing sample: Temporal traces in Langfuse via OpenTelemetry Demonstrates the recommended way to get Temporal workflow traces into Langfuse: OpenTelemetryPlugin(add_temporal_spans=True) with a replay-safe tracer provider, plus a standard OTLP/HTTP exporter pointed at Langfuse's native OpenTelemetry endpoint. No Langfuse-specific SDK or plugin needed. - ticket_triage/: LLM triage workflow (two LLM activities, a plain activity, and a human-approval update) with --replay-stress and worker-restart demos; one correctly nested Langfuse trace per run with GENERATION observations carrying model, token usage, and content. - verify_trace.py: asserts the whole observation tree, types, usage, and no-duplicates through the Langfuse public API. - langfuse/docker-compose.yml: pinned self-hosted Langfuse with headless org/project/API-key provisioning. - naive_guide_style/: deliberately broken anti-pattern (spans created in workflow code, sandbox disabled) showing duplicated, fragmented traces under replay. - RECOMMENDATION.md: customer-shareable write-up of the approach. - tests/langfuse_tracing/: CI-safe tests with mocked LLM activities, an in-memory exporter, the workflow cache disabled, and a Replayer pass asserting replay emits zero new spans. --- README.md | 1 + langfuse_tracing/.env.example | 24 ++ langfuse_tracing/README.md | 170 +++++++++++ langfuse_tracing/RECOMMENDATION.md | 185 ++++++++++++ langfuse_tracing/__init__.py | 0 langfuse_tracing/langfuse/docker-compose.yml | 180 ++++++++++++ langfuse_tracing/naive_guide_style/README.md | 54 ++++ .../naive_guide_style/__init__.py | 0 langfuse_tracing/naive_guide_style/starter.py | 34 +++ langfuse_tracing/naive_guide_style/worker.py | 85 ++++++ .../naive_guide_style/workflow.py | 75 +++++ langfuse_tracing/telemetry.py | 113 ++++++++ langfuse_tracing/ticket_triage/README.md | 63 ++++ langfuse_tracing/ticket_triage/__init__.py | 0 langfuse_tracing/ticket_triage/activities.py | 140 +++++++++ langfuse_tracing/ticket_triage/starter.py | 114 ++++++++ langfuse_tracing/ticket_triage/worker.py | 64 ++++ langfuse_tracing/ticket_triage/workflows.py | 80 +++++ langfuse_tracing/verify_trace.py | 273 ++++++++++++++++++ pyproject.toml | 9 + tests/langfuse_tracing/__init__.py | 0 tests/langfuse_tracing/conftest.py | 19 ++ tests/langfuse_tracing/helpers.py | 30 ++ tests/langfuse_tracing/test_ticket_triage.py | 167 +++++++++++ uv.lock | 87 +++++- 25 files changed, 1966 insertions(+), 1 deletion(-) create mode 100644 langfuse_tracing/.env.example create mode 100644 langfuse_tracing/README.md create mode 100644 langfuse_tracing/RECOMMENDATION.md create mode 100644 langfuse_tracing/__init__.py create mode 100644 langfuse_tracing/langfuse/docker-compose.yml create mode 100644 langfuse_tracing/naive_guide_style/README.md create mode 100644 langfuse_tracing/naive_guide_style/__init__.py create mode 100644 langfuse_tracing/naive_guide_style/starter.py create mode 100644 langfuse_tracing/naive_guide_style/worker.py create mode 100644 langfuse_tracing/naive_guide_style/workflow.py create mode 100644 langfuse_tracing/telemetry.py create mode 100644 langfuse_tracing/ticket_triage/README.md create mode 100644 langfuse_tracing/ticket_triage/__init__.py create mode 100644 langfuse_tracing/ticket_triage/activities.py create mode 100644 langfuse_tracing/ticket_triage/starter.py create mode 100644 langfuse_tracing/ticket_triage/worker.py create mode 100644 langfuse_tracing/ticket_triage/workflows.py create mode 100644 langfuse_tracing/verify_trace.py create mode 100644 tests/langfuse_tracing/__init__.py create mode 100644 tests/langfuse_tracing/conftest.py create mode 100644 tests/langfuse_tracing/helpers.py create mode 100644 tests/langfuse_tracing/test_ticket_triage.py diff --git a/README.md b/README.md index d39428dc7..26264b8b1 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ Some examples require extra dependencies. See each sample's directory for specif * [hello_standalone_nexus](hello_standalone_nexus) - Use Nexus Operations without using a workflow. * [hello_standalone_activity](hello_standalone_activity) - Use activities without using a workflow. * [lambda_worker](lambda_worker) - Run a Temporal Worker inside an AWS Lambda function. +* [langfuse_tracing](langfuse_tracing) - Trace Temporal workflows in Langfuse with the OpenTelemetry plugin and OTLP export. * [langgraph_plugin](langgraph_plugin) - Run LangGraph workflows as durable Temporal workflows (Graph API and Functional API). * [langsmith_tracing](langsmith_tracing) - Trace Temporal workflows with LangSmith via the LangSmith plugin. * [message_passing/introduction](message_passing/introduction/) - Introduction to queries, signals, and updates. diff --git a/langfuse_tracing/.env.example b/langfuse_tracing/.env.example new file mode 100644 index 000000000..52517b782 --- /dev/null +++ b/langfuse_tracing/.env.example @@ -0,0 +1,24 @@ +# Copy to .env and adjust. Load with: set -a; source langfuse_tracing/.env; set +a + +# Langfuse — these defaults match the headless-init values baked into +# langfuse_tracing/langfuse/docker-compose.yml (local demo stack). +LANGFUSE_HOST=http://localhost:3000 +LANGFUSE_PUBLIC_KEY=pk-lf-temporal-demo-0000 +LANGFUSE_SECRET_KEY=sk-lf-temporal-demo-0000 + +# LLM — any OpenAI-compatible endpoint works. +OPENAI_API_KEY=sk-... +MODEL_CLASSIFY=gpt-4o-mini +MODEL_DRAFT=gpt-4o-mini +# To use a local OpenAI-compatible gateway (e.g. a LiteLLM proxy) instead: +# OPENAI_BASE_URL=http://localhost:4000/v1 +# OPENAI_API_KEY= +# MODEL_CLASSIFY= +# MODEL_DRAFT= + +# LLM span instrumentation flavor: openinference (default) or openai-v2. +# See README for the trade-offs. +LLM_INSTRUMENTATION=openinference +# Required only for LLM_INSTRUMENTATION=openai-v2 content capture: +# OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental +# OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only diff --git a/langfuse_tracing/README.md b/langfuse_tracing/README.md new file mode 100644 index 000000000..f51c89999 --- /dev/null +++ b/langfuse_tracing/README.md @@ -0,0 +1,170 @@ +# Langfuse Tracing + +This sample shows the recommended way to get Temporal workflow traces into +[Langfuse](https://langfuse.com/): Temporal's +[`OpenTelemetryPlugin`](https://python.temporal.io/temporalio.contrib.opentelemetry.html) +plus a standard OTLP/HTTP exporter pointed at Langfuse's native OpenTelemetry +endpoint. No Langfuse SDK or Langfuse-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. + +Read [RECOMMENDATION.md](RECOMMENDATION.md) for the full rationale, including +measurements of what goes wrong with the alternative "run everything in the +workflow and disable the sandbox" approach. + +Contents: + +- **[ticket_triage/](ticket_triage/)** — the recommended pattern: an LLM + ticket-triage workflow (two LLM activities, one plain activity, one human + approval delivered as a workflow update). +- **[naive_guide_style/](naive_guide_style/)** — a deliberately broken + anti-pattern that creates spans in workflow code with the sandbox disabled, + to demonstrate why that fails. Do not copy it. +- **[verify_trace.py](verify_trace.py)** — checks a trace via the Langfuse + public API: whole-tree equality, observation types, token usage, and + no-duplicates. +- **[langfuse/docker-compose.yml](langfuse/docker-compose.yml)** — pinned + self-hosted Langfuse with org/project/API keys provisioned headlessly. +- **[telemetry.py](telemetry.py)** — the OpenTelemetry wiring (replay-safe + tracer provider, OTLP exporter with Langfuse auth, LLM instrumentation). + +## Prerequisites + +- Docker (for Langfuse), 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 Langfuse (first pull takes a few minutes) +cd langfuse_tracing/langfuse +docker compose up -d +curl -sf http://localhost:3000/api/public/health # repeat until {"status":"OK",...} +# UI: http://localhost:3000 — login demo@temporal.io / langfuse-demo-pw-1 + +# 2. Install dependencies and set environment (repo root) +cd ../.. +uv sync --group langfuse-tracing +cp langfuse_tracing/.env.example langfuse_tracing/.env # edit the LLM settings +set -a; source langfuse_tracing/.env; set +a + +# 3. Run the sample (two terminals, same environment) +uv run python -m langfuse_tracing.ticket_triage.worker +uv run python -m langfuse_tracing.ticket_triage.starter + +# 4. Verify the trace through the Langfuse API (uses the printed trace ID) +uv run python -m langfuse_tracing.verify_trace --trace-id +``` + +The starter prints a direct link to the trace in the Langfuse UI. You should +see one trace shaped like this (types as Langfuse derives them): + +``` +ticket-triage SPAN (root; session/user/tags) +├─ StartWorkflow:TicketTriageWorkflow SPAN +│ └─ RunWorkflow:TicketTriageWorkflow SPAN +│ ├─ triage SPAN (custom span from workflow code) +│ │ ├─ StartActivity:classify_ticket → RunActivity:classify_ticket +│ │ │ └─ ChatCompletion GENERATION (model, tokens, cost) +│ │ └─ StartActivity:lookup_account → RunActivity:lookup_account +│ └─ StartActivity:draft_reply → RunActivity:draft_reply +│ └─ ChatCompletion GENERATION +└─ StartWorkflowUpdate:approve SPAN + ├─ ValidateUpdate:approve SPAN + └─ HandleUpdate:approve SPAN +``` + +## Prove the replay-safety claims + +Durable execution means workflow code re-executes (replays) on worker +restarts and cache evictions. These two experiments show the trace is +unaffected — rerun `verify_trace.py` after each and it still passes with the +identical tree: + +```bash +# Replay stress: disable the workflow cache so EVERY workflow task replays +# the workflow from the start of history. +uv run python -m langfuse_tracing.ticket_triage.worker --replay-stress +uv run python -m langfuse_tracing.ticket_triage.starter + +# Worker restart mid-workflow: the starter waits 20s before sending the +# approval; kill the worker while the workflow is parked, start a new one, +# and watch the workflow (and its trace) complete cleanly. +uv run python -m langfuse_tracing.ticket_triage.starter --pause-before-approval 20 +# ... ctrl+c the worker, then start it again in another terminal +``` + +Then see the failure mode these protect against: + +```bash +# ANTI-PATTERN demo (see naive_guide_style/README.md): one run of a +# guide-style workflow scatters duplicated spans across ~6 disconnected traces. +uv run python -m langfuse_tracing.naive_guide_style.worker +uv run python -m langfuse_tracing.naive_guide_style.starter +uv run python -m langfuse_tracing.verify_trace --count-generations --since-minutes 3 +``` + +## Where spans come from + +| Span | Emitted by | Where it runs | +|---|---|---| +| `ticket-triage` (root) + `langfuse.*` trace attributes | 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` / `chat ` GENERATIONs | OpenAI auto-instrumentation | worker (activity) | + +## Where tracing works + +| Location | Works? | Notes | +|---|---|---| +| Activity bodies | ✅ | Plain OpenTelemetry + any auto-instrumentation, no restrictions. This is where LLM calls (and their GENERATION spans) 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 Langfuse trace-level attributes on your root span. | + +## LLM instrumentation flavors + +`LLM_INSTRUMENTATION` selects how OpenAI calls are instrumented (both are +verified against Langfuse by this sample): + +| | `openinference` (default) | `openai-v2` | +|---|---|---| +| Package | `openinference-instrumentation-openai` | `opentelemetry-instrumentation-openai-v2` | +| Semantic conventions | OpenInference | OpenTelemetry GenAI (`gen_ai.*`) | +| GENERATION type, model, token usage, cost | ✅ | ✅ | +| Prompt/completion content | ✅ by default | Requires `OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental` and `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only` | +| GENERATION span name | `ChatCompletion` | `chat ` | + +## Operational notes + +- Langfuse's OTLP endpoint is HTTP-only — this sample uses + `opentelemetry-exporter-otlp-proto-http` (the gRPC exporter will not work). +- Short-lived processes must flush: the starter and worker call + `force_flush()` on exit (see `telemetry.py`). +- One workflow run = one workflow ID = one Langfuse trace/session; never + reuse workflow IDs across runs. +- `OTEL_SDK_DISABLED=true` turns off export without code changes. +- Ingestion is asynchronous; `verify_trace.py` polls until the trace is + stable. + +## Tests + +`tests/langfuse_tracing/` runs without Langfuse, Docker, or an LLM: mocked +activities, an in-memory span exporter, a worker with the workflow cache +disabled, whole-tree span assertions, and a `Replayer` pass asserting that +replaying the finished workflow's history emits zero new spans. + +```bash +uv run --group langfuse-tracing pytest tests/langfuse_tracing -v +``` + +## Using this outside samples-python + +The sample is self-contained: copy the `langfuse_tracing/` directory, change +the absolute imports (`langfuse_tracing.ticket_triage.activities` → +`ticket_triage.activities` or similar), and install the dependencies listed +under `langfuse-tracing` in this repo's `pyproject.toml`. diff --git a/langfuse_tracing/RECOMMENDATION.md b/langfuse_tracing/RECOMMENDATION.md new file mode 100644 index 000000000..a0c7bc285 --- /dev/null +++ b/langfuse_tracing/RECOMMENDATION.md @@ -0,0 +1,185 @@ +# Recommended: Temporal → Langfuse via OpenTelemetry + +**TL;DR** — Use Temporal's built-in OpenTelemetry integration +([`temporalio.contrib.opentelemetry`](https://python.temporal.io/temporalio.contrib.opentelemetry.html)) +and export OTLP directly to Langfuse's native OpenTelemetry endpoint. No +Langfuse-specific plugin or SDK is required, workflow code stays deterministic +and sandboxed, and traces come out correctly typed, correctly nested, and +duplicate-free — even across worker restarts and workflow replay. The +runnable sample next to this document demonstrates and verifies all of it. + +## Why not the Langfuse Temporal integration guide + +Langfuse's [Temporal guide](https://langfuse.com/integrations/frameworks/temporal) +takes an approach that conflicts with Temporal's durable execution model on +three points: + +1. **It runs agent/LLM logic inside workflow code and disables the workflow + sandbox** (`UnsandboxedWorkflowRunner`). The sandbox is what protects + workflow determinism; turning it off to make I/O-performing libraries + importable in workflow code trades away durability, replayability, and + Temporal's retry semantics for the part of your application that needs them + most. Model calls belong in activities. + +2. **It instruments the LLM SDK globally, so spans are created from workflow + code.** Temporal workflows re-execute ("replay") their code on worker + restarts, deploys, and cache evictions — that is how durable execution + works. A plain OpenTelemetry tracer knows nothing about replay: every + replay re-creates the same spans with fresh random span IDs and re-exports + them. Langfuse does not deduplicate unrelated span IDs, so duplicates + accumulate. + +3. **It hand-propagates trace context** (`gen_trace_id()` passed through + workflow arguments) instead of using Temporal's header-based context + propagation, so spans from activities — which usually run on other + processes or machines — don't reliably nest under the workflow's trace. + +### Measured, not argued + +The sample contains a deliberately broken `naive_guide_style/` variant that +mirrors that architecture (spans created in workflow code, sandbox off, no +Temporal OpenTelemetry integration), run against a worker whose workflow cache +is disabled so that replay — which in production happens intermittently — +happens on every workflow task. The result of **one** workflow run with two +LLM steps: + +| | Recommended pattern (`ticket_triage/`) | Guide-style (`naive_guide_style/`) | +|---|---|---| +| Langfuse traces per workflow run | **1** | **6**, disconnected | +| Observations | **15**, correctly nested tree | **12 copies of 5 logical spans** (`research-agent` ×4, `agent-step-1` ×4, `agent-step-2` ×2) | +| LLM GENERATIONs | nested under their activity | orphaned in their own traces, no workflow context | +| Worker killed mid-workflow | same single clean tree | more duplicates, more fragments | + +Because replay in production is triggered by cache pressure and restarts, the +guide-style breakage is *intermittent*: traces look fine in a quick test and +degrade in production. The Langfuse symptoms reported as "wrong event types +and nesting" are exactly what this failure mode looks like from the UI. + +## The recommended architecture + +``` +starter process worker process +┌─────────────────────────┐ ┌──────────────────────────────────────┐ +│ root span (langfuse.*) │ │ OpenTelemetryPlugin │ +│ └ StartWorkflow ───────┼──────────┼─→ RunWorkflow (workflow, sandboxed) │ +│ └ StartWorkflowUpdate ─┼─headers──┼─→ └ activities (all LLM calls) │ +│ │ │ └ GENERATION spans (auto- │ +│ OTLP/HTTP ↓ │ │ instrumented OpenAI client) │ +└─────────────────────────┘ │ OTLP/HTTP ↓ │ + └──────────────────────────────────────┘ + ↓ ↓ + Langfuse /api/public/otel (Basic auth: public/secret key) +``` + +Three pieces, all standard: + +1. **`OpenTelemetryPlugin(add_temporal_spans=True)`** on the Temporal client + in every process (workers inherit it from the client). It emits spans for + Temporal operations (`StartWorkflow`, `RunWorkflow`, `RunActivity`, + `HandleUpdate`, …) and propagates trace context across every + client/workflow/activity boundary in Temporal headers, which are persisted + in workflow history — so parenting survives replay, worker restarts, and + multi-worker execution. Its tracer provider + (`create_tracer_provider()`) generates span IDs deterministically from + workflow state and suppresses re-export during replay: each logical span is + emitted once, and even a crash-forced re-export carries the same IDs, which + Langfuse upserts rather than duplicates. Workflow code can use plain + OpenTelemetry APIs (`tracer.start_as_current_span(...)`) and stays fully + sandboxed — the plugin adds the necessary passthrough itself. + +2. **An OTel auto-instrumentation for your LLM SDK, applied in the worker.** + LLM calls live in activities, so instrumentation spans are created in + ordinary (non-replaying) code — no replay concerns at all. Any + instrumentation that emits GenAI/OpenInference-style attributes works; + Langfuse derives the observation type GENERATION plus model, token usage, + and cost from them. + +3. **A standard OTLP/HTTP exporter pointed at Langfuse** — Langfuse ingests + native OpenTelemetry: + + ```python + OTLPSpanExporter( + endpoint=f"{LANGFUSE_HOST}/api/public/otel/v1/traces", # HTTP only; no gRPC + headers={ + "Authorization": f"Basic {base64(public_key + ':' + secret_key)}", + "x-langfuse-ingestion-version": "4", # real-time ingestion + }, + ) + ``` + +## What you see in Langfuse + +One trace per workflow run, with observation types derived automatically — +Temporal operations as SPANs with real durations, LLM calls as GENERATIONs +with model/usage/cost, and update handlers visible as first-class events: + +``` +ticket-triage SPAN (trace root; session/user/tags set here) +├─ StartWorkflow:TicketTriageWorkflow SPAN (client) +│ └─ RunWorkflow:TicketTriageWorkflow SPAN (workflow, real duration) +│ ├─ triage SPAN (custom span in workflow code) +│ │ ├─ StartActivity:classify_ticket → RunActivity:classify_ticket +│ │ │ └─ ChatCompletion GENERATION (model, tokens, cost, input/output) +│ │ └─ StartActivity:lookup_account → RunActivity:lookup_account +│ └─ StartActivity:draft_reply → RunActivity:draft_reply +│ └─ ChatCompletion GENERATION +└─ StartWorkflowUpdate:approve SPAN (client) + ├─ ValidateUpdate:approve SPAN + └─ HandleUpdate:approve SPAN +``` + +The sample's `verify_trace.py` asserts this tree — including types, token +usage, and the absence of duplicates — through the Langfuse public API, and +passes after: a normal run, a run with the workflow cache disabled (replay on +every workflow task), and a run where the worker was killed mid-workflow and +replaced. The trace was identical in all three. + +### Trace-level enrichment + +Langfuse trace fields are set with span attributes on the trace's root span +(the starter is the natural place): + +```python +with tracer.start_as_current_span("ticket-triage", attributes={ + "langfuse.trace.name": "ticket-triage", + "langfuse.session.id": workflow_id, # group runs; find traces by workflow ID + "langfuse.user.id": user_id, + "langfuse.trace.tags": ["temporal"], + "langfuse.trace.metadata.temporal_workflow_id": workflow_id, +}): + handle = await client.start_workflow(...) +``` + +Temporal's spans also carry `temporalWorkflowID`/`temporalRunID` attributes, +so any observation can be tied back to the exact workflow execution in the +Temporal UI. + +## Operational notes + +- **Langfuse's OTLP endpoint is HTTP-only** (protobuf or JSON). Use + `opentelemetry-exporter-otlp-proto-http`, not the gRPC exporter. +- **Flush before short-lived processes exit.** `BatchSpanProcessor` buffers; + call `provider.force_flush()` at the end of starters and on worker shutdown + or the last spans are silently dropped. +- **Use a fresh workflow ID per run.** It becomes the natural Langfuse + session ID, and it avoids ever-growing traces from ID reuse. +- **Long-running workflows:** activity and LLM observations stream into the + trace as they complete; the enclosing `RunWorkflow` span is exported when + the run finishes. +- **Kill switch:** `OTEL_SDK_DISABLED=true` disables export without code + changes. +- **Self-hosting:** Langfuse is MIT-licensed; the sample ships a pinned + `docker-compose.yml` with headless provisioning (org, project, API keys) + for a zero-click local setup. + +## Other SDKs + +The same architecture applies beyond Python: Temporal's TypeScript SDK ships +OpenTelemetry interceptors (`@temporalio/interceptors-opentelemetry`), and +equivalent OpenTelemetry interceptors exist for the Go, Java, and .NET SDKs. +Any of them can feed Langfuse's OTLP endpoint the same way. + +--- + +*See `README.md` in this directory for the runnable sample and the +verification runbook.* diff --git a/langfuse_tracing/__init__.py b/langfuse_tracing/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/langfuse_tracing/langfuse/docker-compose.yml b/langfuse_tracing/langfuse/docker-compose.yml new file mode 100644 index 000000000..67056dae0 --- /dev/null +++ b/langfuse_tracing/langfuse/docker-compose.yml @@ -0,0 +1,180 @@ +# Self-hosted Langfuse for the langfuse_tracing sample. +# +# Based on the upstream https://github.com/langfuse/langfuse/blob/v3.224.1/docker-compose.yml +# with these changes for a local demo: +# - Langfuse images pinned to a specific release instead of the floating `:3` tag. +# - Only the Langfuse UI/API (port 3000) is published on the host. Postgres, ClickHouse, +# Redis, and MinIO stay on the compose network (upstream binds them to 127.0.0.1, which +# can collide with other local stacks, e.g. another Postgres on 5432). +# - Headless initialization (LANGFUSE_INIT_*) provisions the org, project, API keys, and +# login user on first start, so the sample works without any UI clicking. Idempotent. +# - All secrets below are throwaway demo values. Do not reuse them outside local demos. +# +# Usage: +# docker compose up -d +# curl -sf http://localhost:3000/api/public/health # poll until 200 +# UI login: demo@temporal.io / langfuse-demo-pw-1 +name: langfuse-demo +services: + langfuse-worker: + image: docker.io/langfuse/langfuse-worker:3.224.1 + restart: always + depends_on: &langfuse-depends-on + postgres: + condition: service_healthy + minio: + condition: service_healthy + redis: + condition: service_healthy + clickhouse: + condition: service_healthy + environment: &langfuse-worker-env + NEXTAUTH_URL: ${NEXTAUTH_URL:-http://localhost:3000} + DATABASE_URL: ${DATABASE_URL:-postgresql://postgres:postgres@postgres:5432/postgres} + SALT: ${SALT:-langfuse-demo-salt} + ENCRYPTION_KEY: ${ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000} + TELEMETRY_ENABLED: ${TELEMETRY_ENABLED:-false} + LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES: ${LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES:-false} + CLICKHOUSE_MIGRATION_URL: ${CLICKHOUSE_MIGRATION_URL:-clickhouse://clickhouse:9000} + CLICKHOUSE_URL: ${CLICKHOUSE_URL:-http://clickhouse:8123} + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} + CLICKHOUSE_CLUSTER_ENABLED: ${CLICKHOUSE_CLUSTER_ENABLED:-false} + LANGFUSE_USE_AZURE_BLOB: ${LANGFUSE_USE_AZURE_BLOB:-false} + LANGFUSE_USE_OCI_NATIVE_OBJECT_STORAGE: ${LANGFUSE_USE_OCI_NATIVE_OBJECT_STORAGE:-false} + LANGFUSE_OCI_AUTH_TYPE: ${LANGFUSE_OCI_AUTH_TYPE:-workload_identity} + LANGFUSE_S3_EVENT_UPLOAD_BUCKET: ${LANGFUSE_S3_EVENT_UPLOAD_BUCKET:-langfuse} + LANGFUSE_S3_EVENT_UPLOAD_REGION: ${LANGFUSE_S3_EVENT_UPLOAD_REGION:-auto} + LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY:-miniosecret} + LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: ${LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT:-http://minio:9000} + LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE:-true} + LANGFUSE_S3_EVENT_UPLOAD_PREFIX: ${LANGFUSE_S3_EVENT_UPLOAD_PREFIX:-events/} + LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: ${LANGFUSE_S3_MEDIA_UPLOAD_BUCKET:-langfuse} + LANGFUSE_S3_MEDIA_UPLOAD_REGION: ${LANGFUSE_S3_MEDIA_UPLOAD_REGION:-auto} + LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY:-miniosecret} + LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://localhost:9090} + LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE:-true} + LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: ${LANGFUSE_S3_MEDIA_UPLOAD_PREFIX:-media/} + LANGFUSE_S3_BATCH_EXPORT_ENABLED: ${LANGFUSE_S3_BATCH_EXPORT_ENABLED:-false} + LANGFUSE_S3_BATCH_EXPORT_BUCKET: ${LANGFUSE_S3_BATCH_EXPORT_BUCKET:-langfuse} + LANGFUSE_S3_BATCH_EXPORT_PREFIX: ${LANGFUSE_S3_BATCH_EXPORT_PREFIX:-exports/} + LANGFUSE_S3_BATCH_EXPORT_REGION: ${LANGFUSE_S3_BATCH_EXPORT_REGION:-auto} + LANGFUSE_S3_BATCH_EXPORT_ENDPOINT: ${LANGFUSE_S3_BATCH_EXPORT_ENDPOINT:-http://minio:9000} + LANGFUSE_S3_BATCH_EXPORT_EXTERNAL_ENDPOINT: ${LANGFUSE_S3_BATCH_EXPORT_EXTERNAL_ENDPOINT:-http://localhost:9090} + LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID: ${LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY: ${LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY:-miniosecret} + LANGFUSE_S3_BATCH_EXPORT_FORCE_PATH_STYLE: ${LANGFUSE_S3_BATCH_EXPORT_FORCE_PATH_STYLE:-true} + LANGFUSE_INGESTION_QUEUE_DELAY_MS: ${LANGFUSE_INGESTION_QUEUE_DELAY_MS:-} + LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS: ${LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS:-} + REDIS_HOST: ${REDIS_HOST:-redis} + REDIS_PORT: ${REDIS_PORT:-6379} + REDIS_AUTH: ${REDIS_AUTH:-myredissecret} + LANGFUSE_BULLMQ_SKIP_REDIS_VERSION_CHECK: ${LANGFUSE_BULLMQ_SKIP_REDIS_VERSION_CHECK:-false} + REDIS_TLS_ENABLED: ${REDIS_TLS_ENABLED:-false} + REDIS_TLS_CA: ${REDIS_TLS_CA:-/certs/ca.crt} + REDIS_TLS_CERT: ${REDIS_TLS_CERT:-/certs/redis.crt} + REDIS_TLS_KEY: ${REDIS_TLS_KEY:-/certs/redis.key} + EMAIL_FROM_ADDRESS: ${EMAIL_FROM_ADDRESS:-} + SMTP_CONNECTION_URL: ${SMTP_CONNECTION_URL:-} + + langfuse-web: + image: docker.io/langfuse/langfuse:3.224.1 + restart: always + depends_on: *langfuse-depends-on + ports: + - 3000:3000 + environment: + <<: *langfuse-worker-env + NEXTAUTH_SECRET: ${NEXTAUTH_SECRET:-langfuse-demo-nextauth-secret} + # Headless initialization: org, project, API keys, and login user are created on + # first startup. These keys must match langfuse_tracing/.env.example. + LANGFUSE_INIT_ORG_ID: ${LANGFUSE_INIT_ORG_ID:-temporal-demo} + LANGFUSE_INIT_ORG_NAME: ${LANGFUSE_INIT_ORG_NAME:-Temporal Demo} + LANGFUSE_INIT_PROJECT_ID: ${LANGFUSE_INIT_PROJECT_ID:-langfuse-tracing-demo} + LANGFUSE_INIT_PROJECT_NAME: ${LANGFUSE_INIT_PROJECT_NAME:-langfuse-tracing-demo} + LANGFUSE_INIT_PROJECT_PUBLIC_KEY: ${LANGFUSE_INIT_PROJECT_PUBLIC_KEY:-pk-lf-temporal-demo-0000} + LANGFUSE_INIT_PROJECT_SECRET_KEY: ${LANGFUSE_INIT_PROJECT_SECRET_KEY:-sk-lf-temporal-demo-0000} + LANGFUSE_INIT_USER_EMAIL: ${LANGFUSE_INIT_USER_EMAIL:-demo@temporal.io} + LANGFUSE_INIT_USER_NAME: ${LANGFUSE_INIT_USER_NAME:-Demo User} + LANGFUSE_INIT_USER_PASSWORD: ${LANGFUSE_INIT_USER_PASSWORD:-langfuse-demo-pw-1} + + clickhouse: + image: docker.io/clickhouse/clickhouse-server + restart: always + user: "101:101" + environment: + CLICKHOUSE_DB: default + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} + volumes: + - langfuse_clickhouse_data:/var/lib/clickhouse + - langfuse_clickhouse_logs:/var/log/clickhouse-server + healthcheck: + test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1 + interval: 5s + timeout: 5s + retries: 10 + start_period: 1s + + minio: + image: cgr.dev/chainguard/minio + restart: always + entrypoint: sh + # create the 'langfuse' bucket before starting the service + command: -c 'mkdir -p /data/langfuse && minio server --address ":9000" --console-address ":9001" /data' + environment: + MINIO_ROOT_USER: ${MINIO_ROOT_USER:-minio} + MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-miniosecret} + volumes: + - langfuse_minio_data:/data + healthcheck: + test: ["CMD", "mc", "ready", "local"] + interval: 1s + timeout: 5s + retries: 5 + start_period: 1s + + redis: + image: docker.io/redis:7 + restart: always + command: > + --requirepass ${REDIS_AUTH:-myredissecret} + --maxmemory-policy noeviction + volumes: + - langfuse_redis_data:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 3s + timeout: 10s + retries: 10 + + postgres: + image: docker.io/postgres:${POSTGRES_VERSION:-17} + restart: always + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 3s + timeout: 3s + retries: 10 + environment: + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} + POSTGRES_DB: ${POSTGRES_DB:-postgres} + TZ: UTC + PGTZ: UTC + volumes: + - langfuse_postgres_data:/var/lib/postgresql/data + +volumes: + langfuse_postgres_data: + driver: local + langfuse_clickhouse_data: + driver: local + langfuse_clickhouse_logs: + driver: local + langfuse_minio_data: + driver: local + langfuse_redis_data: + driver: local diff --git a/langfuse_tracing/naive_guide_style/README.md b/langfuse_tracing/naive_guide_style/README.md new file mode 100644 index 000000000..769a0685a --- /dev/null +++ b/langfuse_tracing/naive_guide_style/README.md @@ -0,0 +1,54 @@ +# ⚠️ ANTI-PATTERN: spans from workflow code, sandbox disabled + +**This variant is deliberately broken. Do not copy it.** It exists to +demonstrate empirically why observability code must not run inside workflow +code — the architecture suggested by Langfuse's Temporal integration guide. + +What it does (mirroring that guide): + +- Creates OpenTelemetry spans **inside workflow code** with a **plain** + tracer provider (random span IDs, unconditional export). +- **Disables the workflow sandbox** so the tracing/LLM libraries are + importable from workflow code. +- Uses **no Temporal OpenTelemetry integration** — no replay awareness, no + context propagation into activities. + +The worker disables the workflow cache (`max_cached_workflows=0`) so that +replay — which production triggers intermittently via worker restarts, +deploys, and cache evictions — happens on every workflow task and the +breakage is immediately visible. + +## Run the experiment + +```bash +uv run python -m langfuse_tracing.naive_guide_style.worker +uv run python -m langfuse_tracing.naive_guide_style.starter +uv run python -m langfuse_tracing.verify_trace --count-generations --since-minutes 3 +``` + +## Measured result + +One workflow run (two LLM steps) produced **6 disconnected Langfuse traces** +containing **12 copies of 5 logical spans**: + +``` +trace 1 research-agent, agent-step-1, agent-step-2 (the only complete-looking copy) +trace 2 research-agent, agent-step-1 (replay duplicate) +trace 3 research-agent, agent-step-1 (replay duplicate) +trace 4 research-agent, agent-step-1, agent-step-2 (replay duplicate) +trace 5 ChatCompletion [GENERATION] (orphaned — no workflow context) +trace 6 ChatCompletion [GENERATION] (orphaned — no workflow context) +``` + +Why: every replay re-executes the workflow function from the top, and a plain +tracer re-creates the same spans with fresh random IDs and re-exports them — +each replay becomes a new partial trace. Meanwhile the LLM spans from the +activities have no propagated parent, so they land in traces of their own. +This is precisely the "wrong event types and broken nesting" symptom, plus +duplicates that grow with every replay. + +Compare with [../ticket_triage/](../ticket_triage/): same worker-cache +setting, one trace, every span exactly once — because the +`OpenTelemetryPlugin`'s tracer provider derives span IDs deterministically +from workflow state, suppresses re-export during replay, and propagates +context through Temporal headers persisted in workflow history. diff --git a/langfuse_tracing/naive_guide_style/__init__.py b/langfuse_tracing/naive_guide_style/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/langfuse_tracing/naive_guide_style/starter.py b/langfuse_tracing/naive_guide_style/starter.py new file mode 100644 index 000000000..6564f7db9 --- /dev/null +++ b/langfuse_tracing/naive_guide_style/starter.py @@ -0,0 +1,34 @@ +"""ANTI-PATTERN starter — see workflow.py and README.md.""" + +import asyncio +import uuid + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig + +from langfuse_tracing.naive_guide_style.workflow import NaiveGuideStyleWorkflow + +TASK_QUEUE = "langfuse-naive-guide-style-task-queue" + + +async def main() -> None: + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + + workflow_id = f"naive-guide-style-{uuid.uuid4().hex[:8]}" + result = await client.execute_workflow( + NaiveGuideStyleWorkflow.run, + "durable execution", + id=workflow_id, + task_queue=TASK_QUEUE, + ) + print(f"Workflow {workflow_id} result:\n{result}") + print( + "Now look at Langfuse: one workflow produced several disconnected traces " + "with duplicated agent spans. Compare with the ticket_triage sample." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/langfuse_tracing/naive_guide_style/worker.py b/langfuse_tracing/naive_guide_style/worker.py new file mode 100644 index 000000000..cdff6e857 --- /dev/null +++ b/langfuse_tracing/naive_guide_style/worker.py @@ -0,0 +1,85 @@ +"""ANTI-PATTERN worker — deliberately broken. Do not copy. + +Mirrors the Langfuse Temporal guide's setup: a plain (not replay-safe) +OpenTelemetry tracer provider exporting to Langfuse, global OpenAI +instrumentation, and no Temporal OpenTelemetry integration. The workflow +above additionally opts out of the sandbox, as the guide instructs. + +The worker runs with the workflow cache disabled (``max_cached_workflows=0``) +so that replay — which in production happens on worker restarts, deploys, and +cache evictions — happens on every workflow task, making the breakage +immediately visible. See README.md for the experiment. +""" + +import asyncio +import base64 +import logging +import os + +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from temporalio.client import Client +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from langfuse_tracing.naive_guide_style.workflow import ( + NaiveGuideStyleWorkflow, + naive_llm_step, +) +from langfuse_tracing.telemetry import instrument_openai + +TASK_QUEUE = "langfuse-naive-guide-style-task-queue" + + +def setup_plain_tracing() -> None: + # ANTI-PATTERN: a plain TracerProvider. Span IDs are random and export is + # unconditional, so workflow replay re-emits spans as brand-new data. + host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") + auth = base64.b64encode( + f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode() + ).decode() + provider = TracerProvider( + resource=Resource.create({SERVICE_NAME: "naive-guide-style-worker"}) + ) + provider.add_span_processor( + BatchSpanProcessor( + OTLPSpanExporter( + endpoint=f"{host}/api/public/otel/v1/traces", + headers={ + "Authorization": f"Basic {auth}", + "x-langfuse-ingestion-version": "4", + }, + ), + schedule_delay_millis=500, + ) + ) + trace.set_tracer_provider(provider) + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + setup_plain_tracing() + instrument_openai() + + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + # ANTI-PATTERN: no OpenTelemetryPlugin — no replay safety, no context + # propagation across the workflow/activity boundary. + client = await Client.connect(**config) + + worker = Worker( + client, + task_queue=TASK_QUEUE, + workflows=[NaiveGuideStyleWorkflow], + activities=[naive_llm_step], + max_cached_workflows=0, + ) + print("ANTI-PATTERN worker started (do not copy this setup), ctrl+c to exit") + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/langfuse_tracing/naive_guide_style/workflow.py b/langfuse_tracing/naive_guide_style/workflow.py new file mode 100644 index 000000000..2f982850d --- /dev/null +++ b/langfuse_tracing/naive_guide_style/workflow.py @@ -0,0 +1,75 @@ +"""ANTI-PATTERN — deliberately broken. Do not copy. + +This workflow reproduces the architecture that Langfuse's Temporal guide +suggests: observability spans are created directly inside workflow code with a +plain OpenTelemetry tracer, and the worker disables the workflow sandbox to +allow it. It exists only to demonstrate, empirically, why that approach breaks +under Temporal's durable execution model. See ``ticket_triage/`` for the +correct pattern. + +What goes wrong (see README.md for the experiment): + +- Workflow code re-executes on every replay (worker restart, cache eviction). + A plain tracer re-creates these spans each time with fresh random span IDs + and re-exports them: Langfuse accumulates duplicates. +- Without Temporal's OpenTelemetry integration there is no context propagation + into activities, so the LLM GENERATION observations land in separate, + disconnected traces — the "nesting" is gone. +""" + +import os +from dataclasses import dataclass +from datetime import timedelta + +from openai import AsyncOpenAI + +# ANTI-PATTERN: unrestricted imports in workflow code (the guide disables the +# sandbox entirely, so nothing stops these). +from opentelemetry import trace +from temporalio import activity, workflow + + +@dataclass +class ResearchStep: + question: str + + +@activity.defn +async def naive_llm_step(step: ResearchStep) -> str: + client = AsyncOpenAI( + base_url=os.environ.get("OPENAI_BASE_URL"), + api_key=os.environ.get("OPENAI_API_KEY"), + max_retries=0, + ) + response = await client.chat.completions.create( + model=os.environ.get("MODEL_CLASSIFY", "gpt-4o-mini"), + messages=[{"role": "user", "content": step.question}], + timeout=30, + ) + return response.choices[0].message.content or "" + + +@workflow.defn(sandboxed=False) +class NaiveGuideStyleWorkflow: + @workflow.run + async def run(self, topic: str) -> str: + tracer = trace.get_tracer(__name__) + # ANTI-PATTERN: spans created in workflow code with a plain tracer. + # Every replay re-runs this function from the top and re-emits them. + with tracer.start_as_current_span("research-agent"): + with tracer.start_as_current_span("agent-step-1"): + first = await workflow.execute_activity( + naive_llm_step, + ResearchStep(question=f"In one sentence: what is {topic}?"), + start_to_close_timeout=timedelta(seconds=60), + ) + # A timer forces a second workflow task, hence a replay when the + # workflow cache is disabled. + await workflow.sleep(2) + with tracer.start_as_current_span("agent-step-2"): + second = await workflow.execute_activity( + naive_llm_step, + ResearchStep(question=f"In one sentence: why does {topic} matter?"), + start_to_close_timeout=timedelta(seconds=60), + ) + return f"{first}\n{second}" diff --git a/langfuse_tracing/telemetry.py b/langfuse_tracing/telemetry.py new file mode 100644 index 000000000..bef366b37 --- /dev/null +++ b/langfuse_tracing/telemetry.py @@ -0,0 +1,113 @@ +"""Shared OpenTelemetry-to-Langfuse wiring for the langfuse_tracing samples. + +Langfuse natively ingests OpenTelemetry traces, so no Langfuse SDK is needed: +spans are exported over OTLP/HTTP to Langfuse's ``/api/public/otel`` endpoint, +authenticated with a project's public/secret API key pair. + +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 Langfuse. +""" + +import base64 +import logging +import os + +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from temporalio.contrib.opentelemetry import create_tracer_provider + +logger = logging.getLogger(__name__) + + +def _langfuse_exporter() -> OTLPSpanExporter: + host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") + public_key = os.environ["LANGFUSE_PUBLIC_KEY"] + secret_key = os.environ["LANGFUSE_SECRET_KEY"] + auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() + return OTLPSpanExporter( + # Langfuse's OTLP endpoint is HTTP-only (protobuf or JSON); it has no gRPC + # listener, so this must be the http exporter, not the grpc one. + endpoint=f"{host}/api/public/otel/v1/traces", + headers={ + "Authorization": f"Basic {auth}", + # Documented by Langfuse: opts into real-time ingestion. Without it, + # ingested traces can take several minutes to appear in the UI. + "x-langfuse-ingestion-version": "4", + }, + timeout=10, + ) + + +def setup_tracing(service_name: str) -> None: + """Install a replay-safe tracer provider that exports spans to Langfuse. + + 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 worker requires it) but no exporter is attached. + """ + provider = create_tracer_provider( + resource=Resource.create({SERVICE_NAME: service_name}) + ) + if os.environ.get("OTEL_SDK_DISABLED", "").lower() != "true": + # A short schedule delay so demo spans show up in Langfuse 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(_langfuse_exporter(), schedule_delay_millis=500) + ) + else: + logger.info("OTEL_SDK_DISABLED=true - spans will not be exported") + trace.set_tracer_provider(provider) + + +def force_flush() -> None: + """Flush any buffered spans to Langfuse 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() -> str: + """Instrument the OpenAI client library once, in the worker process. + + Every OpenAI API call made from an activity then emits a span that nests + under that activity's span and that Langfuse renders as a GENERATION + observation with model, token usage, and cost. + + Two OpenTelemetry instrumentation flavors are supported via the + ``LLM_INSTRUMENTATION`` env var: + + - ``openinference`` (default): OpenInference semantic conventions. Prompt + and completion content are recorded on span attributes, which Langfuse + maps to the observation's input/output. + - ``openai-v2``: the OpenTelemetry GenAI semantic conventions + (``gen_ai.*``). Content capture additionally requires + ``OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental`` and + ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only``. + """ + flavor = os.environ.get("LLM_INSTRUMENTATION", "openinference") + if flavor == "openinference": + from openinference.instrumentation.openai import OpenAIInstrumentor + + OpenAIInstrumentor().instrument() + elif flavor == "openai-v2": + from opentelemetry.instrumentation.openai_v2 import ( + OpenAIInstrumentor as OpenAIV2Instrumentor, + ) + + OpenAIV2Instrumentor().instrument() + else: + raise ValueError( + f"Unknown LLM_INSTRUMENTATION {flavor!r}; use 'openinference' or 'openai-v2'" + ) + return flavor diff --git a/langfuse_tracing/ticket_triage/README.md b/langfuse_tracing/ticket_triage/README.md new file mode 100644 index 000000000..f5219454b --- /dev/null +++ b/langfuse_tracing/ticket_triage/README.md @@ -0,0 +1,63 @@ +# Ticket Triage + +An LLM support-ticket triage workflow demonstrating the recommended +Temporal → Langfuse tracing setup (see [../README.md](../README.md) for the +full runbook and [../RECOMMENDATION.md](../RECOMMENDATION.md) for the +rationale). + +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 | +| `worker.py` | Worker with `OpenTelemetryPlugin(add_temporal_spans=True)`; `--replay-stress` disables the workflow cache | +| `starter.py` | Opens the root span with `langfuse.*` trace attributes, starts the workflow, sends the approval update, prints the Langfuse trace link; `--decline`, `--pause-before-approval N` | + +## Run + +With Langfuse up, dependencies synced, and the environment loaded (see +[../README.md](../README.md)): + +```bash +uv run python -m langfuse_tracing.ticket_triage.worker +uv run python -m langfuse_tracing.ticket_triage.starter +uv run python -m langfuse_tracing.verify_trace --trace-id +``` + +Variants: + +```bash +uv run python -m langfuse_tracing.ticket_triage.starter --decline +uv run python -m langfuse_tracing.verify_trace --trace-id --expect declined + +# Replay stress: every workflow task replays the workflow from history — +# the Langfuse trace must come out identical. +uv run python -m langfuse_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 langfuse_tracing.ticket_triage.starter --pause-before-approval 20 +``` + +## Expected trace + +``` +ticket-triage SPAN (root; session=workflow id, user, tags) +├─ StartWorkflow:TicketTriageWorkflow SPAN +│ └─ RunWorkflow:TicketTriageWorkflow SPAN +│ ├─ triage SPAN +│ │ ├─ StartActivity:classify_ticket → RunActivity:classify_ticket +│ │ │ └─ ChatCompletion GENERATION +│ │ └─ StartActivity:lookup_account → RunActivity:lookup_account +│ └─ StartActivity:draft_reply → RunActivity:draft_reply +│ └─ ChatCompletion GENERATION +└─ StartWorkflowUpdate:approve SPAN + ├─ ValidateUpdate:approve SPAN + └─ HandleUpdate:approve SPAN +``` + +With `--decline`, the `draft_reply` subtree is absent. diff --git a/langfuse_tracing/ticket_triage/__init__.py b/langfuse_tracing/ticket_triage/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/langfuse_tracing/ticket_triage/activities.py b/langfuse_tracing/ticket_triage/activities.py new file mode 100644 index 000000000..77c30b547 --- /dev/null +++ b/langfuse_tracing/ticket_triage/activities.py @@ -0,0 +1,140 @@ +"""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 a child span of the activity span, +which Langfuse displays as a GENERATION observation with model and token usage. +""" + +import json +import os +from dataclasses import dataclass +from typing import Optional + +from openai import AsyncOpenAI +from temporalio import activity + + +@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. + 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") + + +@activity.defn +async def classify_ticket(ticket: Ticket) -> Classification: + 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 "") + + +@activity.defn +async def lookup_account(customer_email: str) -> AccountInfo: + # A deterministic, non-LLM activity: appears in Langfuse as a plain SPAN + # observation alongside the GENERATION observations from the LLM activities. + 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/langfuse_tracing/ticket_triage/starter.py b/langfuse_tracing/ticket_triage/starter.py new file mode 100644 index 000000000..007f0fdf6 --- /dev/null +++ b/langfuse_tracing/ticket_triage/starter.py @@ -0,0 +1,114 @@ +"""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 +Langfuse trace. Langfuse trace-level attributes (name, session, user, tags) +are set on this root span. +""" + +import argparse +import asyncio +import os +import uuid + +from opentelemetry import trace +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.envconfig import ClientConfig + +from langfuse_tracing.telemetry import force_flush, setup_tracing +from langfuse_tracing.ticket_triage.activities import ApprovalDecision, Ticket +from langfuse_tracing.ticket_triage.workflows import TicketTriageWorkflow + +TASK_QUEUE = "langfuse-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 " + "Langfuse trace still comes out as a single clean tree.", + ) + 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)], + ) + + # A unique workflow ID per run gives one Langfuse trace per run. + workflow_id = 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?" + ), + ) + + tracer = trace.get_tracer(__name__) + try: + with tracer.start_as_current_span( + "ticket-triage", + attributes={ + # langfuse.* attributes on the trace's root span set Langfuse + # trace-level fields, enabling filtering by session/user/tags. + "langfuse.trace.name": "ticket-triage", + "langfuse.session.id": workflow_id, + "langfuse.user.id": os.environ.get("LANGFUSE_DEMO_USER", "demo-user"), + "langfuse.trace.tags": ["temporal", "ticket-triage"], + "langfuse.trace.metadata.temporal_workflow_id": workflow_id, + }, + ) 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() + + print(f"Workflow status: {result.status}") + if result.reply: + print(f"Drafted reply:\n{result.reply}") + + host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") + print(f"Trace ID: {trace_id}") + print(f"Langfuse trace: {host}/project/langfuse-tracing-demo/traces/{trace_id}") + finally: + # 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() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/langfuse_tracing/ticket_triage/worker.py b/langfuse_tracing/ticket_triage/worker.py new file mode 100644 index 000000000..641a2834e --- /dev/null +++ b/langfuse_tracing/ticket_triage/worker.py @@ -0,0 +1,64 @@ +"""Worker for the ticket triage sample.""" + +import asyncio +import logging +import sys + +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from langfuse_tracing.telemetry import force_flush, instrument_openai, setup_tracing +from langfuse_tracing.ticket_triage.activities import ( + classify_ticket, + draft_reply, + lookup_account, +) +from langfuse_tracing.ticket_triage.workflows import TicketTriageWorkflow + +TASK_QUEUE = "langfuse-ticket-triage-task-queue" + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + + # --replay-stress disables 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 Langfuse must look + # identical with or without this flag. + replay_stress = "--replay-stress" in sys.argv + + setup_tracing("ticket-triage-worker") + flavor = 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 replay_stress else 1000, + # No plugins here: workers inherit them from the client. + ) + + mode = "replay-stress (workflow cache disabled)" if replay_stress else "normal" + print(f"Worker started (mode={mode}, llm_instrumentation={flavor}), ctrl+c to exit") + try: + await worker.run() + finally: + force_flush() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/langfuse_tracing/ticket_triage/workflows.py b/langfuse_tracing/ticket_triage/workflows.py new file mode 100644 index 000000000..92709aeb5 --- /dev/null +++ b/langfuse_tracing/ticket_triage/workflows.py @@ -0,0 +1,80 @@ +"""Ticket triage workflow with Langfuse 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 + +with workflow.unsafe.imports_passed_through(): + from opentelemetry import trace + + from langfuse_tracing.ticket_triage.activities import ( + ApprovalDecision, + Classification, + DraftReplyInput, + Ticket, + TriageResult, + classify_ticket, + draft_reply, + lookup_account, + ) + + +@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 generation spans inside them) nest underneath it. + with trace.get_tracer(__name__).start_as_current_span("triage") as span: + classification: Classification = await workflow.execute_activity( + classify_ticket, + ticket, + start_to_close_timeout=timedelta(seconds=60), + ) + 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), + ) + 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") diff --git a/langfuse_tracing/verify_trace.py b/langfuse_tracing/verify_trace.py new file mode 100644 index 000000000..5aad011c5 --- /dev/null +++ b/langfuse_tracing/verify_trace.py @@ -0,0 +1,273 @@ +"""Verify a ticket-triage trace in Langfuse via the public API. + +Fetches the trace, reconstructs the observation tree, and deep-compares it +against the expected shape — including observation types — then checks that +every GENERATION carries a model and token usage, and that no observation was +duplicated (the replay-stress test would surface duplicates here). + +Usage: + python -m langfuse_tracing.verify_trace --trace-id + python -m langfuse_tracing.verify_trace --workflow-id + python -m langfuse_tracing.verify_trace --trace-id --expect declined + python -m langfuse_tracing.verify_trace --count-generations --since-minutes 10 + +Stdlib-only on purpose so it is trivially copy-out-able. +""" + +import argparse +import base64 +import json +import os +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +# Expected observation trees as (depth, name, type) rows, children sorted by +# name under each parent. LLM spans are normalized to "" because +# their name depends on the instrumentation flavor ("ChatCompletion" for +# openinference, "chat " for openai-v2); their type must always be +# GENERATION. +EXPECTED_APPROVED = [ + (0, "ticket-triage", "SPAN"), + (1, "StartWorkflow:TicketTriageWorkflow", "SPAN"), + (2, "RunWorkflow:TicketTriageWorkflow", "SPAN"), + (3, "StartActivity:draft_reply", "SPAN"), + (4, "RunActivity:draft_reply", "SPAN"), + (5, "", "GENERATION"), + (3, "triage", "SPAN"), + (4, "StartActivity:classify_ticket", "SPAN"), + (5, "RunActivity:classify_ticket", "SPAN"), + (6, "", "GENERATION"), + (4, "StartActivity:lookup_account", "SPAN"), + (5, "RunActivity:lookup_account", "SPAN"), + (1, "StartWorkflowUpdate:approve", "SPAN"), + (2, "HandleUpdate:approve", "SPAN"), + (2, "ValidateUpdate:approve", "SPAN"), +] +EXPECTED_DECLINED = [ + (0, "ticket-triage", "SPAN"), + (1, "StartWorkflow:TicketTriageWorkflow", "SPAN"), + (2, "RunWorkflow:TicketTriageWorkflow", "SPAN"), + (3, "triage", "SPAN"), + (4, "StartActivity:classify_ticket", "SPAN"), + (5, "RunActivity:classify_ticket", "SPAN"), + (6, "", "GENERATION"), + (4, "StartActivity:lookup_account", "SPAN"), + (5, "RunActivity:lookup_account", "SPAN"), + (1, "StartWorkflowUpdate:approve", "SPAN"), + (2, "HandleUpdate:approve", "SPAN"), + (2, "ValidateUpdate:approve", "SPAN"), +] + + +def _api_get(path: str, params: Optional[dict[str, str]] = None) -> Any: + host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") + public_key = os.environ["LANGFUSE_PUBLIC_KEY"] + secret_key = os.environ["LANGFUSE_SECRET_KEY"] + auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() + url = f"{host}{path}" + if params: + url += "?" + urllib.parse.urlencode(params) + request = urllib.request.Request(url, headers={"Authorization": f"Basic {auth}"}) + with urllib.request.urlopen(request, timeout=15) as response: + return json.loads(response.read()) + + +def _fetch_trace(trace_id: str) -> Optional[dict[str, Any]]: + try: + return _api_get(f"/api/public/traces/{trace_id}") + except urllib.error.HTTPError as err: + if err.code == 404: + return None + raise + + +def _resolve_trace_id_by_workflow_id(workflow_id: str) -> Optional[str]: + # The starter sets langfuse.session.id to the workflow ID on the root span. + result = _api_get("/api/public/traces", {"sessionId": workflow_id, "limit": "10"}) + data = result.get("data") or [] + return data[0]["id"] if data else None + + +def _normalized_name(observation: dict[str, Any]) -> str: + if observation.get("type") == "GENERATION": + return "" + return str(observation["name"]) + + +def _build_tree(observations: list[dict[str, Any]]) -> list[tuple[int, str, str]]: + children: dict[Optional[str], list[dict[str, Any]]] = {} + for observation in observations: + children.setdefault(observation.get("parentObservationId"), []).append( + observation + ) + rows: list[tuple[int, str, str]] = [] + + def walk(observation: dict[str, Any], depth: int) -> None: + rows.append((depth, _normalized_name(observation), str(observation["type"]))) + for child in sorted( + children.get(observation["id"], []), key=lambda o: str(o["name"]) + ): + walk(child, depth + 1) + + for root in sorted(children.get(None, []), key=lambda o: str(o["name"])): + walk(root, 0) + return rows + + +def _print_tree(rows: list[tuple[int, str, str]]) -> None: + for depth, name, type_ in rows: + print(f" {' ' * depth}{name} [{type_}]") + + +def _poll_stable_trace(trace_id: str, timeout_seconds: int) -> dict[str, Any]: + """Poll until the trace exists and its observation count is stable. + + Langfuse 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: + trace = _fetch_trace(trace_id) + if trace is not None: + count = len(trace.get("observations") or []) + if count > 0 and count == previous_count: + return trace + previous_count = count + time.sleep(2) + raise SystemExit( + f"FAIL: trace {trace_id} not fully ingested within {timeout_seconds}s" + ) + + +def _verify_trace(args: argparse.Namespace) -> int: + trace_id = args.trace_id + if not trace_id: + trace_id = _resolve_trace_id_by_workflow_id(args.workflow_id) + if not trace_id: + print(f"FAIL: no trace found for workflow id {args.workflow_id}") + return 1 + + trace = _poll_stable_trace(trace_id, args.timeout) + observations = trace.get("observations") or [] + failures: list[str] = [] + + # 1. No duplicate observations (replay must not re-emit spans). + ids = [o["id"] for o in observations] + if len(set(ids)) != len(ids): + failures.append("duplicate observation ids present") + actual = _build_tree(observations) + if len(set(actual)) != len(actual): + failures.append( + "duplicate (depth, name, type) rows — replay produced duplicates" + ) + + # 2. Whole-tree deep equality, types included. + expected = EXPECTED_DECLINED if args.expect == "declined" else EXPECTED_APPROVED + print(f"Trace {trace_id}: {len(observations)} observations") + _print_tree(actual) + if actual != expected: + failures.append("tree mismatch") + print(" Expected:") + _print_tree(expected) + + # 3. Every GENERATION has a model and token usage. + for observation in observations: + if observation["type"] != "GENERATION": + continue + usage = observation.get("usage") or {} + if not observation.get("model"): + failures.append(f"GENERATION {observation['id']} missing model") + if not usage.get("input") or not usage.get("output"): + failures.append(f"GENERATION {observation['id']} missing token usage") + if args.require_content and ( + observation.get("input") is None or observation.get("output") is None + ): + failures.append( + f"GENERATION {observation['id']} missing input/output content" + ) + + # 4. Trace-level enrichment from the starter's root span. + if trace.get("name") != "ticket-triage": + failures.append( + f"trace name is {trace.get('name')!r}, expected 'ticket-triage'" + ) + if not trace.get("sessionId"): + failures.append("trace has no session id (expected the workflow id)") + if not trace.get("userId"): + failures.append("trace has no user id") + if "temporal" not in (trace.get("tags") or []): + failures.append("trace missing 'temporal' tag") + + if failures: + for failure in failures: + print(f"FAIL: {failure}") + return 1 + print("PASS: tree shape, observation types, generations, and enrichment all match") + return 0 + + +def _count_generations(args: argparse.Namespace) -> int: + """Count recent GENERATION observations (used for the anti-pattern demo).""" + from_time = datetime.now(timezone.utc) - timedelta(minutes=args.since_minutes) + result = _api_get( + "/api/public/observations", + { + "type": "GENERATION", + "fromStartTime": from_time.isoformat(), + "limit": "100", + }, + ) + data = result.get("data") or [] + by_trace: dict[str, int] = {} + for observation in data: + by_trace[observation.get("traceId") or "?"] = ( + by_trace.get(observation.get("traceId") or "?", 0) + 1 + ) + print( + f"{len(data)} GENERATION observations in the last {args.since_minutes}m " + f"across {len(by_trace)} trace(s)" + ) + for trace_id, count in sorted(by_trace.items()): + print(f" trace {trace_id}: {count} generation(s)") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--trace-id", help="Trace ID printed by the starter") + parser.add_argument("--workflow-id", help="Workflow ID (resolved via session id)") + parser.add_argument( + "--expect", choices=["approved", "declined"], default="approved" + ) + parser.add_argument("--timeout", type=int, default=60) + parser.add_argument( + "--require-content", + action="store_true", + default=os.environ.get("LLM_INSTRUMENTATION", "openinference") + == "openinference", + help="Assert GENERATIONs carry input/output content " + "(default true for openinference)", + ) + parser.add_argument( + "--count-generations", + action="store_true", + help="Just count recent GENERATION observations project-wide", + ) + parser.add_argument("--since-minutes", type=int, default=10) + args = parser.parse_args() + + if args.count_generations: + return _count_generations(args) + if not args.trace_id and not args.workflow_id: + parser.error("one of --trace-id or --workflow-id is required") + return _verify_trace(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index e8f3ebd87..e77dcd459 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,14 @@ external-storage = [ external-storage-redis = ["redis>=5.0.0,<8"] gevent = ["gevent>=25.4.2 ; python_version >= '3.8'"] google-adk = ["temporalio[google-adk] >= 1.30.0", "google-adk>=1.27.0,<2"] +langfuse-tracing = [ + "openai>=1.4.0", + "temporalio[opentelemetry]>=1.30.0,<2", + # Langfuse's OTLP endpoint is HTTP-only (no gRPC), so use the http exporter. + "opentelemetry-exporter-otlp-proto-http>=1.30.0,<2", + "openinference-instrumentation-openai>=0.1.52", + "opentelemetry-instrumentation-openai-v2>=2.1b0", +] langsmith-tracing = [ "openai>=1.4.0", "langsmith>=0.7.0", @@ -108,6 +116,7 @@ packages = [ "external_storage_redis", "gevent_async", "hello", + "langfuse_tracing", "langgraph_plugin", "langsmith_tracing", "message_passing", diff --git a/tests/langfuse_tracing/__init__.py b/tests/langfuse_tracing/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/langfuse_tracing/conftest.py b/tests/langfuse_tracing/conftest.py new file mode 100644 index 000000000..3a542c9e4 --- /dev/null +++ b/tests/langfuse_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/langfuse_tracing/helpers.py b/tests/langfuse_tracing/helpers.py new file mode 100644 index 000000000..7c224ae9b --- /dev/null +++ b/tests/langfuse_tracing/helpers.py @@ -0,0 +1,30 @@ +"""Test helpers for the langfuse_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/langfuse_tracing/test_ticket_triage.py b/tests/langfuse_tracing/test_ticket_triage.py new file mode 100644 index 000000000..315a4d1ac --- /dev/null +++ b/tests/langfuse_tracing/test_ticket_triage.py @@ -0,0 +1,167 @@ +"""Tests for the ticket triage sample. + +These run without Langfuse 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 uuid +from typing import Any + +import opentelemetry.trace +from opentelemetry import trace +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from temporalio import activity +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider +from temporalio.worker import Replayer, Worker + +from langfuse_tracing.ticket_triage.activities import ( + AccountInfo, + ApprovalDecision, + Classification, + DraftReplyInput, + Ticket, +) +from langfuse_tracing.ticket_triage.workflows import TicketTriageWorkflow +from tests.langfuse_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.", +) + + +@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="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(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_workflow(client: Client, task_queue: str, approved: bool) -> Any: + handle = await client.start_workflow( + TicketTriageWorkflow.run, + TICKET, + id=f"ticket-triage-test-{uuid.uuid4()}", + task_queue=task_queue, + ) + await handle.execute_update( + TicketTriageWorkflow.approve, + ApprovalDecision(approved=approved, reviewer="test-reviewer"), + ) + await handle.result() + 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", +] + + +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()}" + + 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, + ): + with trace.get_tracer(__name__).start_as_current_span("ticket-triage test"): + handle = await _run_workflow(new_client, task_queue, 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) + + # 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, + ): + with trace.get_tracer(__name__).start_as_current_span("ticket-triage test"): + await _run_workflow(new_client, task_queue, 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 diff --git a/uv.lock b/uv.lock index 6063f75a3..2205dd8c9 100644 --- a/uv.lock +++ b/uv.lock @@ -736,7 +736,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 = [ @@ -3132,6 +3132,48 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/64/96/d7dfe1cc0be2df22d7a97ffb0f8bb00b10d92749aa6e64ffa7cc9a041580/openapi_spec_validator-0.8.5-py3-none-any.whl", hash = "sha256:3669106361856934153991e30714616a294865a33f6411a4c25d1dc2d08cfbc2", size = 50334, upload-time = "2026-04-24T15:25:19.65Z" }, ] +[[package]] +name = "openinference-instrumentation" +version = "0.1.54" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "openinference-semantic-conventions" }, + { name = "opentelemetry-api" }, + { 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" } +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" }, +] + +[[package]] +name = "openinference-instrumentation-openai" +version = "0.1.52" +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/f2/90/c81c7426cb9dca075cef1b5fe16f58c68604b7518f7aaa14d837ec80fde3/openinference_instrumentation_openai-0.1.52.tar.gz", hash = "sha256:5a3dceb742209463e33ab2a4aa82f56bedfa20c25c738de28248509931044ea1", size = 23087, upload-time = "2026-06-11T17:13:35.517Z" } +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-semantic-conventions" +version = "0.1.30" +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" } +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" }, +] + [[package]] name = "opentelemetry-api" version = "1.41.1" @@ -3253,6 +3295,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4d/0f/45adbaea1f81b847cffdcee4f4b5f89297e42facf7fac78c7aaac4c38e75/opentelemetry_instrumentation-0.62b1-py3-none-any.whl", hash = "sha256:976fc6e640f2006599e97429c949e622c108d0c17c2059347d1e6c93c707f257", size = 34163, upload-time = "2026-04-24T13:21:31.722Z" }, ] +[[package]] +name = "opentelemetry-instrumentation-openai-v2" +version = "2.4b0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-genai" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/07/5f/d034617f70dbf2a92048b0c3536bc1cce10f88adefd43fd75abd51d918b1/opentelemetry_instrumentation_openai_v2-2.4b0.tar.gz", hash = "sha256:571a2febd05b15808d7c777455d5a74c9abe08c694cb9827a98c5b4258f2adf1", size = 190742, upload-time = "2026-05-01T17:41:50.929Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c3/ed/2e8f3348dfad1dc18c01c2bbc275694e6e6f0c85ccb56257b5516e5af702/opentelemetry_instrumentation_openai_v2-2.4b0-py3-none-any.whl", hash = "sha256:c0c65fe4593fdcb466b55c047138ec71d28a5a3f36c3f0c2c5738343daa31d5c", size = 28027, upload-time = "2026-05-01T17:41:49.768Z" }, +] + [[package]] name = "opentelemetry-instrumentation-threading" version = "0.62b1" @@ -3321,6 +3378,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/eb/a6/83dc2ab6fa397ee66fba04fe2e74bdf7be3b3870005359ceb7689103c058/opentelemetry_semantic_conventions-0.62b1-py3-none-any.whl", hash = "sha256:cf506938103d331fbb78eded0d9788095f7fd59016f2bda813c3324e5a74a93c", size = 231620, upload-time = "2026-04-24T13:15:35.454Z" }, ] +[[package]] +name = "opentelemetry-util-genai" +version = "0.4b0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e8/54/545527aba649f6b8aba7b70c855db9089a2b8f234bd6c19beffa73a3163d/opentelemetry_util_genai-0.4b0.tar.gz", hash = "sha256:0235b03c5b3cb5efe5d3c16a5a68e82be34e6530d6707cf1cf122413578c2036", size = 47385, upload-time = "2026-05-01T17:29:17.055Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e8/50/0b86c4159a74802a917fcc2adf22f1af522f03805cc049415221207249e8/opentelemetry_util_genai-0.4b0-py3-none-any.whl", hash = "sha256:ac26db52ad1d86ce3e4ac183f204c37a6e66fdb6d86b71feee60468bcb32ef13", size = 42848, upload-time = "2026-05-01T17:29:15.839Z" }, +] + [[package]] name = "orjson" version = "3.11.9" @@ -5051,6 +5122,13 @@ google-adk = [ { name = "google-adk" }, { name = "temporalio", extra = ["google-adk"] }, ] +langfuse-tracing = [ + { name = "openai" }, + { name = "openinference-instrumentation-openai" }, + { name = "opentelemetry-exporter-otlp-proto-http" }, + { name = "opentelemetry-instrumentation-openai-v2" }, + { name = "temporalio", extra = ["opentelemetry"] }, +] langgraph = [ { name = "langchain" }, { name = "langchain-anthropic" }, @@ -5139,6 +5217,13 @@ google-adk = [ { name = "google-adk", specifier = ">=1.27.0,<2" }, { name = "temporalio", extras = ["google-adk"], specifier = ">=1.30.0" }, ] +langfuse-tracing = [ + { name = "openai", specifier = ">=1.4.0" }, + { name = "openinference-instrumentation-openai", specifier = ">=0.1.52" }, + { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.30.0,<2" }, + { name = "opentelemetry-instrumentation-openai-v2", specifier = ">=2.1b0" }, + { name = "temporalio", extras = ["opentelemetry"], specifier = ">=1.30.0,<2" }, +] langgraph = [ { name = "langchain", specifier = ">=0.3.0" }, { name = "langchain-anthropic", specifier = ">=0.3.0" }, From d3baba78300fa878cbf4c4eb9decad79a6690739 Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 23 Jul 2026 22:27:06 -0500 Subject: [PATCH 2/5] Slim the langfuse_tracing sample to the core pattern Drop the recommendation write-up and the comparison variant, and remove references to them from the READMEs and verify_trace.py. --- langfuse_tracing/README.md | 17 -- langfuse_tracing/RECOMMENDATION.md | 185 ------------------ langfuse_tracing/naive_guide_style/README.md | 54 ----- .../naive_guide_style/__init__.py | 0 langfuse_tracing/naive_guide_style/starter.py | 34 ---- langfuse_tracing/naive_guide_style/worker.py | 85 -------- .../naive_guide_style/workflow.py | 75 ------- langfuse_tracing/ticket_triage/README.md | 3 +- langfuse_tracing/verify_trace.py | 36 ---- 9 files changed, 1 insertion(+), 488 deletions(-) delete mode 100644 langfuse_tracing/RECOMMENDATION.md delete mode 100644 langfuse_tracing/naive_guide_style/README.md delete mode 100644 langfuse_tracing/naive_guide_style/__init__.py delete mode 100644 langfuse_tracing/naive_guide_style/starter.py delete mode 100644 langfuse_tracing/naive_guide_style/worker.py delete mode 100644 langfuse_tracing/naive_guide_style/workflow.py diff --git a/langfuse_tracing/README.md b/langfuse_tracing/README.md index f51c89999..d920cf44c 100644 --- a/langfuse_tracing/README.md +++ b/langfuse_tracing/README.md @@ -8,18 +8,11 @@ endpoint. No Langfuse SDK or Langfuse-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. -Read [RECOMMENDATION.md](RECOMMENDATION.md) for the full rationale, including -measurements of what goes wrong with the alternative "run everything in the -workflow and disable the sandbox" approach. - Contents: - **[ticket_triage/](ticket_triage/)** — the recommended pattern: an LLM ticket-triage workflow (two LLM activities, one plain activity, one human approval delivered as a workflow update). -- **[naive_guide_style/](naive_guide_style/)** — a deliberately broken - anti-pattern that creates spans in workflow code with the sandbox disabled, - to demonstrate why that fails. Do not copy it. - **[verify_trace.py](verify_trace.py)** — checks a trace via the Langfuse public API: whole-tree equality, observation types, token usage, and no-duplicates. @@ -96,16 +89,6 @@ uv run python -m langfuse_tracing.ticket_triage.starter --pause-before-approval # ... ctrl+c the worker, then start it again in another terminal ``` -Then see the failure mode these protect against: - -```bash -# ANTI-PATTERN demo (see naive_guide_style/README.md): one run of a -# guide-style workflow scatters duplicated spans across ~6 disconnected traces. -uv run python -m langfuse_tracing.naive_guide_style.worker -uv run python -m langfuse_tracing.naive_guide_style.starter -uv run python -m langfuse_tracing.verify_trace --count-generations --since-minutes 3 -``` - ## Where spans come from | Span | Emitted by | Where it runs | diff --git a/langfuse_tracing/RECOMMENDATION.md b/langfuse_tracing/RECOMMENDATION.md deleted file mode 100644 index a0c7bc285..000000000 --- a/langfuse_tracing/RECOMMENDATION.md +++ /dev/null @@ -1,185 +0,0 @@ -# Recommended: Temporal → Langfuse via OpenTelemetry - -**TL;DR** — Use Temporal's built-in OpenTelemetry integration -([`temporalio.contrib.opentelemetry`](https://python.temporal.io/temporalio.contrib.opentelemetry.html)) -and export OTLP directly to Langfuse's native OpenTelemetry endpoint. No -Langfuse-specific plugin or SDK is required, workflow code stays deterministic -and sandboxed, and traces come out correctly typed, correctly nested, and -duplicate-free — even across worker restarts and workflow replay. The -runnable sample next to this document demonstrates and verifies all of it. - -## Why not the Langfuse Temporal integration guide - -Langfuse's [Temporal guide](https://langfuse.com/integrations/frameworks/temporal) -takes an approach that conflicts with Temporal's durable execution model on -three points: - -1. **It runs agent/LLM logic inside workflow code and disables the workflow - sandbox** (`UnsandboxedWorkflowRunner`). The sandbox is what protects - workflow determinism; turning it off to make I/O-performing libraries - importable in workflow code trades away durability, replayability, and - Temporal's retry semantics for the part of your application that needs them - most. Model calls belong in activities. - -2. **It instruments the LLM SDK globally, so spans are created from workflow - code.** Temporal workflows re-execute ("replay") their code on worker - restarts, deploys, and cache evictions — that is how durable execution - works. A plain OpenTelemetry tracer knows nothing about replay: every - replay re-creates the same spans with fresh random span IDs and re-exports - them. Langfuse does not deduplicate unrelated span IDs, so duplicates - accumulate. - -3. **It hand-propagates trace context** (`gen_trace_id()` passed through - workflow arguments) instead of using Temporal's header-based context - propagation, so spans from activities — which usually run on other - processes or machines — don't reliably nest under the workflow's trace. - -### Measured, not argued - -The sample contains a deliberately broken `naive_guide_style/` variant that -mirrors that architecture (spans created in workflow code, sandbox off, no -Temporal OpenTelemetry integration), run against a worker whose workflow cache -is disabled so that replay — which in production happens intermittently — -happens on every workflow task. The result of **one** workflow run with two -LLM steps: - -| | Recommended pattern (`ticket_triage/`) | Guide-style (`naive_guide_style/`) | -|---|---|---| -| Langfuse traces per workflow run | **1** | **6**, disconnected | -| Observations | **15**, correctly nested tree | **12 copies of 5 logical spans** (`research-agent` ×4, `agent-step-1` ×4, `agent-step-2` ×2) | -| LLM GENERATIONs | nested under their activity | orphaned in their own traces, no workflow context | -| Worker killed mid-workflow | same single clean tree | more duplicates, more fragments | - -Because replay in production is triggered by cache pressure and restarts, the -guide-style breakage is *intermittent*: traces look fine in a quick test and -degrade in production. The Langfuse symptoms reported as "wrong event types -and nesting" are exactly what this failure mode looks like from the UI. - -## The recommended architecture - -``` -starter process worker process -┌─────────────────────────┐ ┌──────────────────────────────────────┐ -│ root span (langfuse.*) │ │ OpenTelemetryPlugin │ -│ └ StartWorkflow ───────┼──────────┼─→ RunWorkflow (workflow, sandboxed) │ -│ └ StartWorkflowUpdate ─┼─headers──┼─→ └ activities (all LLM calls) │ -│ │ │ └ GENERATION spans (auto- │ -│ OTLP/HTTP ↓ │ │ instrumented OpenAI client) │ -└─────────────────────────┘ │ OTLP/HTTP ↓ │ - └──────────────────────────────────────┘ - ↓ ↓ - Langfuse /api/public/otel (Basic auth: public/secret key) -``` - -Three pieces, all standard: - -1. **`OpenTelemetryPlugin(add_temporal_spans=True)`** on the Temporal client - in every process (workers inherit it from the client). It emits spans for - Temporal operations (`StartWorkflow`, `RunWorkflow`, `RunActivity`, - `HandleUpdate`, …) and propagates trace context across every - client/workflow/activity boundary in Temporal headers, which are persisted - in workflow history — so parenting survives replay, worker restarts, and - multi-worker execution. Its tracer provider - (`create_tracer_provider()`) generates span IDs deterministically from - workflow state and suppresses re-export during replay: each logical span is - emitted once, and even a crash-forced re-export carries the same IDs, which - Langfuse upserts rather than duplicates. Workflow code can use plain - OpenTelemetry APIs (`tracer.start_as_current_span(...)`) and stays fully - sandboxed — the plugin adds the necessary passthrough itself. - -2. **An OTel auto-instrumentation for your LLM SDK, applied in the worker.** - LLM calls live in activities, so instrumentation spans are created in - ordinary (non-replaying) code — no replay concerns at all. Any - instrumentation that emits GenAI/OpenInference-style attributes works; - Langfuse derives the observation type GENERATION plus model, token usage, - and cost from them. - -3. **A standard OTLP/HTTP exporter pointed at Langfuse** — Langfuse ingests - native OpenTelemetry: - - ```python - OTLPSpanExporter( - endpoint=f"{LANGFUSE_HOST}/api/public/otel/v1/traces", # HTTP only; no gRPC - headers={ - "Authorization": f"Basic {base64(public_key + ':' + secret_key)}", - "x-langfuse-ingestion-version": "4", # real-time ingestion - }, - ) - ``` - -## What you see in Langfuse - -One trace per workflow run, with observation types derived automatically — -Temporal operations as SPANs with real durations, LLM calls as GENERATIONs -with model/usage/cost, and update handlers visible as first-class events: - -``` -ticket-triage SPAN (trace root; session/user/tags set here) -├─ StartWorkflow:TicketTriageWorkflow SPAN (client) -│ └─ RunWorkflow:TicketTriageWorkflow SPAN (workflow, real duration) -│ ├─ triage SPAN (custom span in workflow code) -│ │ ├─ StartActivity:classify_ticket → RunActivity:classify_ticket -│ │ │ └─ ChatCompletion GENERATION (model, tokens, cost, input/output) -│ │ └─ StartActivity:lookup_account → RunActivity:lookup_account -│ └─ StartActivity:draft_reply → RunActivity:draft_reply -│ └─ ChatCompletion GENERATION -└─ StartWorkflowUpdate:approve SPAN (client) - ├─ ValidateUpdate:approve SPAN - └─ HandleUpdate:approve SPAN -``` - -The sample's `verify_trace.py` asserts this tree — including types, token -usage, and the absence of duplicates — through the Langfuse public API, and -passes after: a normal run, a run with the workflow cache disabled (replay on -every workflow task), and a run where the worker was killed mid-workflow and -replaced. The trace was identical in all three. - -### Trace-level enrichment - -Langfuse trace fields are set with span attributes on the trace's root span -(the starter is the natural place): - -```python -with tracer.start_as_current_span("ticket-triage", attributes={ - "langfuse.trace.name": "ticket-triage", - "langfuse.session.id": workflow_id, # group runs; find traces by workflow ID - "langfuse.user.id": user_id, - "langfuse.trace.tags": ["temporal"], - "langfuse.trace.metadata.temporal_workflow_id": workflow_id, -}): - handle = await client.start_workflow(...) -``` - -Temporal's spans also carry `temporalWorkflowID`/`temporalRunID` attributes, -so any observation can be tied back to the exact workflow execution in the -Temporal UI. - -## Operational notes - -- **Langfuse's OTLP endpoint is HTTP-only** (protobuf or JSON). Use - `opentelemetry-exporter-otlp-proto-http`, not the gRPC exporter. -- **Flush before short-lived processes exit.** `BatchSpanProcessor` buffers; - call `provider.force_flush()` at the end of starters and on worker shutdown - or the last spans are silently dropped. -- **Use a fresh workflow ID per run.** It becomes the natural Langfuse - session ID, and it avoids ever-growing traces from ID reuse. -- **Long-running workflows:** activity and LLM observations stream into the - trace as they complete; the enclosing `RunWorkflow` span is exported when - the run finishes. -- **Kill switch:** `OTEL_SDK_DISABLED=true` disables export without code - changes. -- **Self-hosting:** Langfuse is MIT-licensed; the sample ships a pinned - `docker-compose.yml` with headless provisioning (org, project, API keys) - for a zero-click local setup. - -## Other SDKs - -The same architecture applies beyond Python: Temporal's TypeScript SDK ships -OpenTelemetry interceptors (`@temporalio/interceptors-opentelemetry`), and -equivalent OpenTelemetry interceptors exist for the Go, Java, and .NET SDKs. -Any of them can feed Langfuse's OTLP endpoint the same way. - ---- - -*See `README.md` in this directory for the runnable sample and the -verification runbook.* diff --git a/langfuse_tracing/naive_guide_style/README.md b/langfuse_tracing/naive_guide_style/README.md deleted file mode 100644 index 769a0685a..000000000 --- a/langfuse_tracing/naive_guide_style/README.md +++ /dev/null @@ -1,54 +0,0 @@ -# ⚠️ ANTI-PATTERN: spans from workflow code, sandbox disabled - -**This variant is deliberately broken. Do not copy it.** It exists to -demonstrate empirically why observability code must not run inside workflow -code — the architecture suggested by Langfuse's Temporal integration guide. - -What it does (mirroring that guide): - -- Creates OpenTelemetry spans **inside workflow code** with a **plain** - tracer provider (random span IDs, unconditional export). -- **Disables the workflow sandbox** so the tracing/LLM libraries are - importable from workflow code. -- Uses **no Temporal OpenTelemetry integration** — no replay awareness, no - context propagation into activities. - -The worker disables the workflow cache (`max_cached_workflows=0`) so that -replay — which production triggers intermittently via worker restarts, -deploys, and cache evictions — happens on every workflow task and the -breakage is immediately visible. - -## Run the experiment - -```bash -uv run python -m langfuse_tracing.naive_guide_style.worker -uv run python -m langfuse_tracing.naive_guide_style.starter -uv run python -m langfuse_tracing.verify_trace --count-generations --since-minutes 3 -``` - -## Measured result - -One workflow run (two LLM steps) produced **6 disconnected Langfuse traces** -containing **12 copies of 5 logical spans**: - -``` -trace 1 research-agent, agent-step-1, agent-step-2 (the only complete-looking copy) -trace 2 research-agent, agent-step-1 (replay duplicate) -trace 3 research-agent, agent-step-1 (replay duplicate) -trace 4 research-agent, agent-step-1, agent-step-2 (replay duplicate) -trace 5 ChatCompletion [GENERATION] (orphaned — no workflow context) -trace 6 ChatCompletion [GENERATION] (orphaned — no workflow context) -``` - -Why: every replay re-executes the workflow function from the top, and a plain -tracer re-creates the same spans with fresh random IDs and re-exports them — -each replay becomes a new partial trace. Meanwhile the LLM spans from the -activities have no propagated parent, so they land in traces of their own. -This is precisely the "wrong event types and broken nesting" symptom, plus -duplicates that grow with every replay. - -Compare with [../ticket_triage/](../ticket_triage/): same worker-cache -setting, one trace, every span exactly once — because the -`OpenTelemetryPlugin`'s tracer provider derives span IDs deterministically -from workflow state, suppresses re-export during replay, and propagates -context through Temporal headers persisted in workflow history. diff --git a/langfuse_tracing/naive_guide_style/__init__.py b/langfuse_tracing/naive_guide_style/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/langfuse_tracing/naive_guide_style/starter.py b/langfuse_tracing/naive_guide_style/starter.py deleted file mode 100644 index 6564f7db9..000000000 --- a/langfuse_tracing/naive_guide_style/starter.py +++ /dev/null @@ -1,34 +0,0 @@ -"""ANTI-PATTERN starter — see workflow.py and README.md.""" - -import asyncio -import uuid - -from temporalio.client import Client -from temporalio.envconfig import ClientConfig - -from langfuse_tracing.naive_guide_style.workflow import NaiveGuideStyleWorkflow - -TASK_QUEUE = "langfuse-naive-guide-style-task-queue" - - -async def main() -> None: - config = ClientConfig.load_client_connect_config() - config.setdefault("target_host", "localhost:7233") - client = await Client.connect(**config) - - workflow_id = f"naive-guide-style-{uuid.uuid4().hex[:8]}" - result = await client.execute_workflow( - NaiveGuideStyleWorkflow.run, - "durable execution", - id=workflow_id, - task_queue=TASK_QUEUE, - ) - print(f"Workflow {workflow_id} result:\n{result}") - print( - "Now look at Langfuse: one workflow produced several disconnected traces " - "with duplicated agent spans. Compare with the ticket_triage sample." - ) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/langfuse_tracing/naive_guide_style/worker.py b/langfuse_tracing/naive_guide_style/worker.py deleted file mode 100644 index cdff6e857..000000000 --- a/langfuse_tracing/naive_guide_style/worker.py +++ /dev/null @@ -1,85 +0,0 @@ -"""ANTI-PATTERN worker — deliberately broken. Do not copy. - -Mirrors the Langfuse Temporal guide's setup: a plain (not replay-safe) -OpenTelemetry tracer provider exporting to Langfuse, global OpenAI -instrumentation, and no Temporal OpenTelemetry integration. The workflow -above additionally opts out of the sandbox, as the guide instructs. - -The worker runs with the workflow cache disabled (``max_cached_workflows=0``) -so that replay — which in production happens on worker restarts, deploys, and -cache evictions — happens on every workflow task, making the breakage -immediately visible. See README.md for the experiment. -""" - -import asyncio -import base64 -import logging -import os - -from opentelemetry import trace -from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter -from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor -from temporalio.client import Client -from temporalio.envconfig import ClientConfig -from temporalio.worker import Worker - -from langfuse_tracing.naive_guide_style.workflow import ( - NaiveGuideStyleWorkflow, - naive_llm_step, -) -from langfuse_tracing.telemetry import instrument_openai - -TASK_QUEUE = "langfuse-naive-guide-style-task-queue" - - -def setup_plain_tracing() -> None: - # ANTI-PATTERN: a plain TracerProvider. Span IDs are random and export is - # unconditional, so workflow replay re-emits spans as brand-new data. - host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") - auth = base64.b64encode( - f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode() - ).decode() - provider = TracerProvider( - resource=Resource.create({SERVICE_NAME: "naive-guide-style-worker"}) - ) - provider.add_span_processor( - BatchSpanProcessor( - OTLPSpanExporter( - endpoint=f"{host}/api/public/otel/v1/traces", - headers={ - "Authorization": f"Basic {auth}", - "x-langfuse-ingestion-version": "4", - }, - ), - schedule_delay_millis=500, - ) - ) - trace.set_tracer_provider(provider) - - -async def main() -> None: - logging.basicConfig(level=logging.INFO) - setup_plain_tracing() - instrument_openai() - - config = ClientConfig.load_client_connect_config() - config.setdefault("target_host", "localhost:7233") - # ANTI-PATTERN: no OpenTelemetryPlugin — no replay safety, no context - # propagation across the workflow/activity boundary. - client = await Client.connect(**config) - - worker = Worker( - client, - task_queue=TASK_QUEUE, - workflows=[NaiveGuideStyleWorkflow], - activities=[naive_llm_step], - max_cached_workflows=0, - ) - print("ANTI-PATTERN worker started (do not copy this setup), ctrl+c to exit") - await worker.run() - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/langfuse_tracing/naive_guide_style/workflow.py b/langfuse_tracing/naive_guide_style/workflow.py deleted file mode 100644 index 2f982850d..000000000 --- a/langfuse_tracing/naive_guide_style/workflow.py +++ /dev/null @@ -1,75 +0,0 @@ -"""ANTI-PATTERN — deliberately broken. Do not copy. - -This workflow reproduces the architecture that Langfuse's Temporal guide -suggests: observability spans are created directly inside workflow code with a -plain OpenTelemetry tracer, and the worker disables the workflow sandbox to -allow it. It exists only to demonstrate, empirically, why that approach breaks -under Temporal's durable execution model. See ``ticket_triage/`` for the -correct pattern. - -What goes wrong (see README.md for the experiment): - -- Workflow code re-executes on every replay (worker restart, cache eviction). - A plain tracer re-creates these spans each time with fresh random span IDs - and re-exports them: Langfuse accumulates duplicates. -- Without Temporal's OpenTelemetry integration there is no context propagation - into activities, so the LLM GENERATION observations land in separate, - disconnected traces — the "nesting" is gone. -""" - -import os -from dataclasses import dataclass -from datetime import timedelta - -from openai import AsyncOpenAI - -# ANTI-PATTERN: unrestricted imports in workflow code (the guide disables the -# sandbox entirely, so nothing stops these). -from opentelemetry import trace -from temporalio import activity, workflow - - -@dataclass -class ResearchStep: - question: str - - -@activity.defn -async def naive_llm_step(step: ResearchStep) -> str: - client = AsyncOpenAI( - base_url=os.environ.get("OPENAI_BASE_URL"), - api_key=os.environ.get("OPENAI_API_KEY"), - max_retries=0, - ) - response = await client.chat.completions.create( - model=os.environ.get("MODEL_CLASSIFY", "gpt-4o-mini"), - messages=[{"role": "user", "content": step.question}], - timeout=30, - ) - return response.choices[0].message.content or "" - - -@workflow.defn(sandboxed=False) -class NaiveGuideStyleWorkflow: - @workflow.run - async def run(self, topic: str) -> str: - tracer = trace.get_tracer(__name__) - # ANTI-PATTERN: spans created in workflow code with a plain tracer. - # Every replay re-runs this function from the top and re-emits them. - with tracer.start_as_current_span("research-agent"): - with tracer.start_as_current_span("agent-step-1"): - first = await workflow.execute_activity( - naive_llm_step, - ResearchStep(question=f"In one sentence: what is {topic}?"), - start_to_close_timeout=timedelta(seconds=60), - ) - # A timer forces a second workflow task, hence a replay when the - # workflow cache is disabled. - await workflow.sleep(2) - with tracer.start_as_current_span("agent-step-2"): - second = await workflow.execute_activity( - naive_llm_step, - ResearchStep(question=f"In one sentence: why does {topic} matter?"), - start_to_close_timeout=timedelta(seconds=60), - ) - return f"{first}\n{second}" diff --git a/langfuse_tracing/ticket_triage/README.md b/langfuse_tracing/ticket_triage/README.md index f5219454b..eef7d7cc2 100644 --- a/langfuse_tracing/ticket_triage/README.md +++ b/langfuse_tracing/ticket_triage/README.md @@ -2,8 +2,7 @@ An LLM support-ticket triage workflow demonstrating the recommended Temporal → Langfuse tracing setup (see [../README.md](../README.md) for the -full runbook and [../RECOMMENDATION.md](../RECOMMENDATION.md) for the -rationale). +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 diff --git a/langfuse_tracing/verify_trace.py b/langfuse_tracing/verify_trace.py index 5aad011c5..a80fe05aa 100644 --- a/langfuse_tracing/verify_trace.py +++ b/langfuse_tracing/verify_trace.py @@ -9,7 +9,6 @@ python -m langfuse_tracing.verify_trace --trace-id python -m langfuse_tracing.verify_trace --workflow-id python -m langfuse_tracing.verify_trace --trace-id --expect declined - python -m langfuse_tracing.verify_trace --count-generations --since-minutes 10 Stdlib-only on purpose so it is trivially copy-out-able. """ @@ -23,7 +22,6 @@ import urllib.error import urllib.parse import urllib.request -from datetime import datetime, timedelta, timezone from typing import Any, Optional # Expected observation trees as (depth, name, type) rows, children sorted by @@ -212,32 +210,6 @@ def _verify_trace(args: argparse.Namespace) -> int: return 0 -def _count_generations(args: argparse.Namespace) -> int: - """Count recent GENERATION observations (used for the anti-pattern demo).""" - from_time = datetime.now(timezone.utc) - timedelta(minutes=args.since_minutes) - result = _api_get( - "/api/public/observations", - { - "type": "GENERATION", - "fromStartTime": from_time.isoformat(), - "limit": "100", - }, - ) - data = result.get("data") or [] - by_trace: dict[str, int] = {} - for observation in data: - by_trace[observation.get("traceId") or "?"] = ( - by_trace.get(observation.get("traceId") or "?", 0) + 1 - ) - print( - f"{len(data)} GENERATION observations in the last {args.since_minutes}m " - f"across {len(by_trace)} trace(s)" - ) - for trace_id, count in sorted(by_trace.items()): - print(f" trace {trace_id}: {count} generation(s)") - return 0 - - def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--trace-id", help="Trace ID printed by the starter") @@ -254,16 +226,8 @@ def main() -> int: help="Assert GENERATIONs carry input/output content " "(default true for openinference)", ) - parser.add_argument( - "--count-generations", - action="store_true", - help="Just count recent GENERATION observations project-wide", - ) - parser.add_argument("--since-minutes", type=int, default=10) args = parser.parse_args() - if args.count_generations: - return _count_generations(args) if not args.trace_id and not args.workflow_id: parser.error("one of --trace-id or --workflow-id is required") return _verify_trace(args) From e8d03713e3cc23d472dfd36ae4449f717b32501d Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 23 Jul 2026 23:49:21 -0500 Subject: [PATCH 3/5] Address review feedback in the langfuse_tracing sample - Bound LLM activity retries (RetryPolicy(maximum_attempts=3)) so a misconfigured endpoint or API key fails fast instead of retrying forever, and note that each retry attempt adds its own RunActivity span; verify_trace's duplicate-row message now mentions activity retries alongside replay as a possible cause of extra spans. - Ignore .env files repo-wide so the runbook-created env file holding a real API key cannot be committed by accident. - Make the Langfuse project ID in the starter's printed trace link configurable via LANGFUSE_PROJECT_ID (defaults to the project the bundled docker-compose provisions). - Parse worker flags with argparse so typos and --help behave as expected instead of silently starting a normal-mode worker. - Replace bare os.environ lookups for Langfuse credentials with an actionable error message in telemetry.py and verify_trace.py. --- .gitignore | 1 + langfuse_tracing/.env.example | 3 +++ langfuse_tracing/telemetry.py | 10 ++++++++-- langfuse_tracing/ticket_triage/starter.py | 5 ++++- langfuse_tracing/ticket_triage/worker.py | 18 ++++++++++++------ langfuse_tracing/ticket_triage/workflows.py | 8 ++++++++ langfuse_tracing/verify_trace.py | 14 +++++++++++--- 7 files changed, 47 insertions(+), 12 deletions(-) diff --git a/.gitignore b/.gitignore index 9b0b43524..1b9a5ae45 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ __pycache__ .mypy_cache/ **/client.key **/client.pem +.env diff --git a/langfuse_tracing/.env.example b/langfuse_tracing/.env.example index 52517b782..4a4045e10 100644 --- a/langfuse_tracing/.env.example +++ b/langfuse_tracing/.env.example @@ -5,6 +5,9 @@ LANGFUSE_HOST=http://localhost:3000 LANGFUSE_PUBLIC_KEY=pk-lf-temporal-demo-0000 LANGFUSE_SECRET_KEY=sk-lf-temporal-demo-0000 +# Used only for the trace link the starter prints; set to your project ID +# when pointing at your own Langfuse instance or Langfuse Cloud. +LANGFUSE_PROJECT_ID=langfuse-tracing-demo # LLM — any OpenAI-compatible endpoint works. OPENAI_API_KEY=sk-... diff --git a/langfuse_tracing/telemetry.py b/langfuse_tracing/telemetry.py index bef366b37..b17a7d7df 100644 --- a/langfuse_tracing/telemetry.py +++ b/langfuse_tracing/telemetry.py @@ -26,8 +26,14 @@ def _langfuse_exporter() -> OTLPSpanExporter: host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") - public_key = os.environ["LANGFUSE_PUBLIC_KEY"] - secret_key = os.environ["LANGFUSE_SECRET_KEY"] + public_key = os.environ.get("LANGFUSE_PUBLIC_KEY") + secret_key = os.environ.get("LANGFUSE_SECRET_KEY") + if not public_key or not secret_key: + raise SystemExit( + "LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set. Copy " + "langfuse_tracing/.env.example to langfuse_tracing/.env and load it " + "in this terminal: set -a; source langfuse_tracing/.env; set +a" + ) auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() return OTLPSpanExporter( # Langfuse's OTLP endpoint is HTTP-only (protobuf or JSON); it has no gRPC diff --git a/langfuse_tracing/ticket_triage/starter.py b/langfuse_tracing/ticket_triage/starter.py index 007f0fdf6..a1eb4235b 100644 --- a/langfuse_tracing/ticket_triage/starter.py +++ b/langfuse_tracing/ticket_triage/starter.py @@ -101,8 +101,11 @@ async def main() -> None: print(f"Drafted reply:\n{result.reply}") host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") + # The UI link needs the Langfuse project ID; the default matches the + # project provisioned by langfuse_tracing/langfuse/docker-compose.yml. + project = os.environ.get("LANGFUSE_PROJECT_ID", "langfuse-tracing-demo") print(f"Trace ID: {trace_id}") - print(f"Langfuse trace: {host}/project/langfuse-tracing-demo/traces/{trace_id}") + print(f"Langfuse trace: {host}/project/{project}/traces/{trace_id}") finally: # The starter is short-lived; flush so its spans (the trace root and # the client-side StartWorkflow/StartWorkflowUpdate spans) are not diff --git a/langfuse_tracing/ticket_triage/worker.py b/langfuse_tracing/ticket_triage/worker.py index 641a2834e..db33e1e8d 100644 --- a/langfuse_tracing/ticket_triage/worker.py +++ b/langfuse_tracing/ticket_triage/worker.py @@ -1,8 +1,8 @@ """Worker for the ticket triage sample.""" +import argparse import asyncio import logging -import sys from temporalio.client import Client from temporalio.contrib.opentelemetry import OpenTelemetryPlugin @@ -23,11 +23,17 @@ async def main() -> None: logging.basicConfig(level=logging.INFO) - # --replay-stress disables 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 Langfuse must look - # identical with or without this flag. - replay_stress = "--replay-stress" in sys.argv + 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 Langfuse must look identical " + "with or without this flag.", + ) + args = parser.parse_args() + replay_stress = args.replay_stress setup_tracing("ticket-triage-worker") flavor = instrument_openai() diff --git a/langfuse_tracing/ticket_triage/workflows.py b/langfuse_tracing/ticket_triage/workflows.py index 92709aeb5..7ec9b897a 100644 --- a/langfuse_tracing/ticket_triage/workflows.py +++ b/langfuse_tracing/ticket_triage/workflows.py @@ -11,6 +11,12 @@ 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. Note that each retry +# attempt records its own RunActivity span in the trace. +LLM_RETRY_POLICY = RetryPolicy(maximum_attempts=3) with workflow.unsafe.imports_passed_through(): from opentelemetry import trace @@ -42,6 +48,7 @@ async def run(self, ticket: Ticket) -> TriageResult: classify_ticket, ticket, start_to_close_timeout=timedelta(seconds=60), + retry_policy=LLM_RETRY_POLICY, ) account = await workflow.execute_activity( lookup_account, @@ -64,6 +71,7 @@ async def run(self, ticket: Ticket) -> TriageResult: 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 diff --git a/langfuse_tracing/verify_trace.py b/langfuse_tracing/verify_trace.py index a80fe05aa..f9302841d 100644 --- a/langfuse_tracing/verify_trace.py +++ b/langfuse_tracing/verify_trace.py @@ -64,8 +64,14 @@ def _api_get(path: str, params: Optional[dict[str, str]] = None) -> Any: host = os.environ.get("LANGFUSE_HOST", "http://localhost:3000").rstrip("/") - public_key = os.environ["LANGFUSE_PUBLIC_KEY"] - secret_key = os.environ["LANGFUSE_SECRET_KEY"] + public_key = os.environ.get("LANGFUSE_PUBLIC_KEY") + secret_key = os.environ.get("LANGFUSE_SECRET_KEY") + if not public_key or not secret_key: + raise SystemExit( + "LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set. Copy " + "langfuse_tracing/.env.example to langfuse_tracing/.env and load it " + "in this terminal: set -a; source langfuse_tracing/.env; set +a" + ) auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() url = f"{host}{path}" if params: @@ -162,7 +168,9 @@ def _verify_trace(args: argparse.Namespace) -> int: actual = _build_tree(observations) if len(set(actual)) != len(actual): failures.append( - "duplicate (depth, name, type) rows — replay produced duplicates" + "duplicate (depth, name, type) rows — extra spans present (workflow " + "replay must never re-emit spans; activity retries also add a " + "RunActivity span per attempt)" ) # 2. Whole-tree deep equality, types included. From 4f45c8b5ea9288403e91b3b972dae0baad16bccc Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 28 Jul 2026 13:39:32 -0500 Subject: [PATCH 4/5] Add langfuse_tracing to the AI SDK team's CODEOWNERS entries --- .github/CODEOWNERS | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 403c731de..8a90220d3 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -16,11 +16,13 @@ # 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. /google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk +/langfuse_tracing/ @temporalio/sdk @temporalio/ai-sdk /langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk /langsmith_tracing/ @temporalio/sdk @temporalio/ai-sdk /openai_agents/ @temporalio/sdk @temporalio/ai-sdk /strands_plugin/ @temporalio/sdk @temporalio/ai-sdk /tests/google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk +/tests/langfuse_tracing/ @temporalio/sdk @temporalio/ai-sdk /tests/langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk /tests/langsmith_tracing/ @temporalio/sdk @temporalio/ai-sdk /tests/strands_plugin/ @temporalio/sdk @temporalio/ai-sdk From c2c23cc11c978be090b60f1577319d2be9d46be0 Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 28 Jul 2026 13:48:51 -0500 Subject: [PATCH 5/5] Remove duplicate langgraph_plugin CODEOWNERS entry The AI SDK block below already contains the identical pattern and owners, and CODEOWNERS resolves by last matching pattern, so the earlier entry was fully shadowed. --- .github/CODEOWNERS | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 8a90220d3..776007ed7 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -1,5 +1,4 @@ * @temporalio/sdk -/langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk # SDK & Nexus own the README, pyproject.toml, and uv.lock /README.md @temporalio/sdk @temporalio/nexus