diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 403c731de..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 @@ -16,11 +15,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 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/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..4a4045e10 --- /dev/null +++ b/langfuse_tracing/.env.example @@ -0,0 +1,27 @@ +# 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 +# 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-... +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..d920cf44c --- /dev/null +++ b/langfuse_tracing/README.md @@ -0,0 +1,153 @@ +# 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. + +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). +- **[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 +``` + +## 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/__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/telemetry.py b/langfuse_tracing/telemetry.py new file mode 100644 index 000000000..b17a7d7df --- /dev/null +++ b/langfuse_tracing/telemetry.py @@ -0,0 +1,119 @@ +"""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.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 + # 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..eef7d7cc2 --- /dev/null +++ b/langfuse_tracing/ticket_triage/README.md @@ -0,0 +1,62 @@ +# Ticket Triage + +An LLM support-ticket triage workflow demonstrating the recommended +Temporal → Langfuse tracing setup (see [../README.md](../README.md) for the +full runbook). + +Flow: `classify_ticket` (LLM) and `lookup_account` (plain activity) run under +a custom `triage` span, the workflow then waits for a human decision delivered +as a workflow **update** (`approve`, with a validator), and on approval +`draft_reply` (LLM) produces the customer reply. + +| File | Purpose | +|---|---| +| `workflows.py` | `TicketTriageWorkflow` — deterministic, sandboxed; uses plain OpenTelemetry APIs for the `triage` span; `approve` update handler + validator | +| `activities.py` | The two LLM activities and the plain lookup activity; all I/O lives here | +| `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..a1eb4235b --- /dev/null +++ b/langfuse_tracing/ticket_triage/starter.py @@ -0,0 +1,117 @@ +"""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("/") + # 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/{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 + # 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..db33e1e8d --- /dev/null +++ b/langfuse_tracing/ticket_triage/worker.py @@ -0,0 +1,70 @@ +"""Worker for the ticket triage sample.""" + +import argparse +import asyncio +import logging + +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from 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) + + 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() + + 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..7ec9b897a --- /dev/null +++ b/langfuse_tracing/ticket_triage/workflows.py @@ -0,0 +1,88 @@ +"""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 +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 + + 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), + retry_policy=LLM_RETRY_POLICY, + ) + account = await workflow.execute_activity( + lookup_account, + ticket.customer_email, + start_to_close_timeout=timedelta(seconds=10), + ) + span.set_attribute("triage.category", classification.category) + span.set_attribute("triage.priority", classification.priority) + + # Wait for a human approval, delivered as a workflow update. + await workflow.wait_condition(lambda: self._approval is not None) + approval = self._approval + assert approval is not None + if not approval.approved: + return TriageResult(status="declined", classification=classification) + + reply = await workflow.execute_activity( + draft_reply, + DraftReplyInput( + ticket=ticket, classification=classification, account=account + ), + start_to_close_timeout=timedelta(seconds=60), + retry_policy=LLM_RETRY_POLICY, + ) + return TriageResult( + status="replied", classification=classification, reply=reply + ) + + @workflow.update + async def approve(self, decision: ApprovalDecision) -> str: + self._approval = decision + return "approved" if decision.approved else "declined" + + @approve.validator + def approve_validator(self, decision: ApprovalDecision) -> None: + if decision.approved and not decision.reviewer: + raise ValueError("approval requires a reviewer") diff --git a/langfuse_tracing/verify_trace.py b/langfuse_tracing/verify_trace.py new file mode 100644 index 000000000..f9302841d --- /dev/null +++ b/langfuse_tracing/verify_trace.py @@ -0,0 +1,245 @@ +"""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 + +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 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.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: + 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 — 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. + 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 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)", + ) + args = parser.parse_args() + + if not args.trace_id and not args.workflow_id: + parser.error("one of --trace-id or --workflow-id is required") + 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" },