Trace Temporal Workflows with Langfuse

%pip install "temporalio[opentelemetry]" openai opentelemetry-exporter-otlp-proto-http openinference-instrumentation-openai

import os

# Get keys for your project from the project settings page: https://cloud.langfuse.com
os.environ.setdefault("LANGFUSE_PUBLIC_KEY", "pk-lf-...");
os.environ.setdefault("LANGFUSE_SECRET_KEY", "sk-lf-...");
os.environ.setdefault("LANGFUSE_BASE_URL", "https://cloud.langfuse.com");  # 🇪🇺 EU region
# Other Langfuse data regions include 🇺🇸 US: https://us.cloud.langfuse.com, 🇯🇵 Japan: https://jp.cloud.langfuse.com and ⚕️ HIPAA: https://hipaa.cloud.langfuse.com

# Your OpenAI key (any OpenAI-compatible endpoint also works via OPENAI_BASE_URL)
os.environ.setdefault("OPENAI_API_KEY", "sk-proj-...");

# Temporal server address (defaults to a local dev server). For Temporal
# Cloud, set TEMPORAL_ADDRESS to your Cloud gRPC endpoint, TEMPORAL_NAMESPACE
# to your Cloud namespace, and TEMPORAL_API_KEY to an API key.
os.environ.setdefault("TEMPORAL_ADDRESS", "localhost:7233");
os.environ.setdefault("TEMPORAL_NAMESPACE", "default");
TASK_QUEUE = "langfuse-tracing-task-queue"

import base64

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

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

exporter = OTLPSpanExporter(
    # Langfuse's OTLP endpoint is HTTP-based, so use the http exporter (not grpc).
    endpoint=f"{os.environ['LANGFUSE_BASE_URL'].rstrip('/')}/api/public/otel/v1/traces",
    headers={
        "Authorization": f"Basic {LANGFUSE_AUTH}",
        "x-langfuse-ingestion-version": "4",  # real-time ingestion
    },
)

# Replay-safe tracer provider from the Temporal SDK: deterministic span IDs,
# spans emitted exactly once even when Temporal replays workflow code.
provider = create_tracer_provider(
    resource=Resource.create({SERVICE_NAME: "ticket-triage"})
)
provider.add_span_processor(BatchSpanProcessor(exporter, schedule_delay_millis=500))
trace.set_tracer_provider(provider)

import json
import urllib.request

request = urllib.request.Request(
    f"{os.environ['LANGFUSE_BASE_URL'].rstrip('/')}/api/public/projects",
    headers={"Authorization": f"Basic {LANGFUSE_AUTH}"},
)
try:
    with urllib.request.urlopen(request, timeout=10):
        print("✅ Langfuse credentials verified!")
except Exception as exc:
    print(f"❌ Could not verify Langfuse credentials: {exc}")

from openinference.instrumentation.openai import OpenAIInstrumentor

OpenAIInstrumentor().instrument()

%%writefile ticket_triage_activities.py
"""Activities for the ticket triage workflow. All I/O and LLM calls live here."""

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
    reply: Optional[str] = None


def _openai_client() -> AsyncOpenAI:
    # max_retries=0: Temporal's activity retry policy owns retries instead of
    # the OpenAI client, with full visibility in the Temporal UI.
    return AsyncOpenAI(max_retries=0)


def _model() -> str:
    return os.environ.get("OPENAI_MODEL", "gpt-4o-mini")


@activity.defn
async def classify_ticket(ticket: Ticket) -> Classification:
    response = await _openai_client().chat.completions.create(
        model=_model(),
        messages=[
            {
                "role": "system",
                "content": (
                    "Classify the support ticket. Respond with ONLY a JSON object like "
                    '{"category": "billing|bug|how-to|other", "priority": "low|normal|high"}.'
                ),
            },
            {"role": "user", "content": f"{ticket.subject}\n\n{ticket.body}"},
        ],
    )
    text = response.choices[0].message.content or ""
    try:
        data = json.loads(text[text.index("{") : text.rindex("}") + 1])
        return Classification(
            category=str(data.get("category", "other")),
            priority=str(data.get("priority", "normal")),
        )
    except ValueError:
        return Classification(category="other", priority="normal")


@activity.defn
async def lookup_account(customer_email: str) -> AccountInfo:
    # A non-LLM activity: appears in Langfuse as a plain span observation
    # alongside the generation observations from the LLM activities.
    return AccountInfo(
        customer_email=customer_email, account_name="Acme Corp", plan="enterprise"
    )


@activity.defn
async def draft_reply(input: DraftReplyInput) -> str:
    response = await _openai_client().chat.completions.create(
        model=_model(),
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a support agent. Draft a short, friendly reply (under 120 "
                    "words) using the provided classification and account details."
                ),
            },
            {
                "role": "user",
                "content": (
                    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)"
                ),
            },
        ],
    )
    return response.choices[0].message.content or ""

%%writefile ticket_triage_workflow.py
"""Ticket triage workflow: deterministic, sandboxed, and fully traced."""

from datetime import timedelta
from typing import Optional

from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from opentelemetry import trace

    from ticket_triage_activities import (
        ApprovalDecision,
        Classification,
        DraftReplyInput,
        Ticket,
        TriageResult,
        classify_ticket,
        draft_reply,
        lookup_account,
    )

# Bounded retries so a misconfigured LLM endpoint or API key fails fast.
LLM_RETRY_POLICY = RetryPolicy(maximum_attempts=3)


@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 steps — plain OpenTelemetry,
        # replay-safe under the plugin's tracer provider.
        with trace.get_tracer(__name__).start_as_current_span("triage"):
            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),
            )

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

        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", 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")

temporal server start-dev

import uuid

from temporalio.client import Client
from temporalio.contrib.opentelemetry import OpenTelemetryPlugin
from temporalio.worker import Worker

from ticket_triage_activities import (
    ApprovalDecision,
    Ticket,
    classify_ticket,
    draft_reply,
    lookup_account,
)
from ticket_triage_workflow import TicketTriageWorkflow


async def main() -> None:
    # TEMPORAL_API_KEY authenticates to Temporal Cloud (TLS is enabled
    # automatically when an API key is used); leave it unset for a local
    # dev server.
    api_key = os.environ.get("TEMPORAL_API_KEY") or None
    client = await Client.connect(
        os.environ["TEMPORAL_ADDRESS"],
        namespace=os.environ["TEMPORAL_NAMESPACE"],
        api_key=api_key,
        tls=api_key is not None,
        # Emits spans for Temporal operations and propagates trace context
        # across the client, workflow, and activity boundaries.
        plugins=[OpenTelemetryPlugin(add_temporal_spans=True)],
    )

    worker = Worker(
        client,
        task_queue=TASK_QUEUE,
        workflows=[TicketTriageWorkflow],
        activities=[classify_ticket, lookup_account, draft_reply],
    )

    # A fresh workflow ID per run avoids Temporal workflow ID conflicts and
    # gives each run its own Langfuse session; the trace itself is keyed by
    # the root span opened below, which is new on every 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=(
            "My card statement shows two identical charges for our July invoice. "
            "Can you refund the duplicate?"
        ),
    )

    async with worker:
        # Langfuse trace-level attributes are set on the trace's root span.
        with trace.get_tracer(__name__).start_as_current_span(
            "ticket-triage",
            attributes={
                "langfuse.trace.name": "ticket-triage",
                "langfuse.session.id": workflow_id,
                "langfuse.user.id": "demo-user",
                "langfuse.trace.tags": ["temporal", "ticket-triage"],
            },
        ):
            handle = await client.start_workflow(
                TicketTriageWorkflow.run,
                ticket,
                id=workflow_id,
                task_queue=TASK_QUEUE,
            )
            print(f"Started workflow: {workflow_id}")

            # A human (here: this notebook) approves the triage via an update.
            decision = await handle.execute_update(
                TicketTriageWorkflow.approve,
                ApprovalDecision(approved=True, reviewer="demo-reviewer"),
            )
            print(f"Approval decision: {decision}")

            result = await handle.result()

    print(f"\nWorkflow status: {result.status}")
    print(f"\nDrafted reply:\n{result.reply}")

    # Flush buffered spans before a short-lived process exits.
    provider.force_flush()


await main()

ticket-triage                                   (trace root: session, user, tags)
├── StartWorkflow:TicketTriageWorkflow          (client)
│   └── RunWorkflow:TicketTriageWorkflow        (workflow)
│       ├── triage                              (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: model, tokens, cost)
└── StartWorkflowUpdate:approve                 (client)
    ├── ValidateUpdate:approve
    └── HandleUpdate:approve