Langfuse Workflows Via OpenInference

Export OpenTelemetry spans from a conditional research workflow to Langfuse via the OpenInference Agno instrumentor.

Demonstrates tracing a multi-step Agno workflow in Langfuse.

langfuse_via_openinference_workflows.py
"""
Langfuse Workflows Via OpenInference
====================================

Demonstrates tracing a multi-step Agno workflow in Langfuse.
"""

import base64
import os

from agno.agent import Agent
from agno.tools.websearch import WebSearchTools
from agno.workflow.condition import Condition
from agno.workflow.step import Step
from agno.workflow.types import StepInput
from agno.workflow.workflow import Workflow
from openinference.instrumentation.agno import AgnoInstrumentor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
LANGFUSE_AUTH = base64.b64encode(
    f"{os.getenv('LANGFUSE_PUBLIC_KEY')}:{os.getenv('LANGFUSE_SECRET_KEY')}".encode()
).decode()
# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
#     "https://us.cloud.langfuse.com/api/public/otel"  # US data region
# )
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    "https://cloud.langfuse.com/api/public/otel"  # EU data region
)
# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:3000/api/public/otel"  # Local deployment (>= v3.22.0)

os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter()))

# Start instrumenting agno
AgnoInstrumentor().instrument(tracer_provider=tracer_provider)


# ---------------------------------------------------------------------------
# Create Workflow
# ---------------------------------------------------------------------------
# Basic agents
researcher = Agent(
    name="Researcher",
    instructions="Research the given topic and provide detailed findings.",
    tools=[WebSearchTools()],
)

summarizer = Agent(
    name="Summarizer",
    instructions="Create a clear summary of the research findings.",
)

fact_checker = Agent(
    name="Fact Checker",
    instructions="Verify facts and check for accuracy in the research.",
    tools=[WebSearchTools()],
)

writer = Agent(
    name="Writer",
    instructions="Write a comprehensive article based on all available research and verification.",
)


# Condition evaluator
def needs_fact_checking(step_input: StepInput) -> bool:
    """Determine if the research contains claims that need fact-checking."""
    return True


# Workflow steps
research_step = Step(
    name="research",
    description="Research the topic",
    agent=researcher,
)

summarize_step = Step(
    name="summarize",
    description="Summarize research findings",
    agent=summarizer,
)

fact_check_step = Step(
    name="fact_check",
    description="Verify facts and claims",
    agent=fact_checker,
)

write_article = Step(
    name="write_article",
    description="Write final article",
    agent=writer,
)

basic_workflow = Workflow(
    name="Basic Linear Workflow",
    description="Research -> Summarize -> Condition(Fact Check) -> Write Article",
    steps=[
        research_step,
        summarize_step,
        Condition(
            name="fact_check_condition",
            description="Check if fact-checking is needed",
            evaluator=needs_fact_checking,
            steps=[fact_check_step],
        ),
        write_article,
    ],
)


# ---------------------------------------------------------------------------
# Run Workflow
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    print("Running Basic Linear Workflow Example")
    print("=" * 50)

    try:
        basic_workflow.print_response(
            input="Recent breakthroughs in quantum computing",
            stream=True,
        )
    except Exception as e:
        print(f"Error: {e}")
        import traceback

        traceback.print_exc()

The condition evaluator always returns True, so the fact-checking step always runs when execution reaches it. Replace the evaluator if you want a content-dependent decision.

Langfuse export settings

Choose the endpoint for your project's region in the copied source. For current direct OTLP ingestion, replace its header assignment before creating the exporter:

os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = (
    f"Authorization=Basic {LANGFUSE_AUTH},x-langfuse-ingestion-version=4"
)

Without the version header, traces can take up to ten minutes to appear in Langfuse's v4 data model. See Langfuse OpenTelemetry configuration.

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U agno ddgs fastapi openai openinference-instrumentation-agno opentelemetry-exporter-otlp opentelemetry-sdk

Export environment variables

export LANGFUSE_PUBLIC_KEY="your_langfuse_public_key_here"
export LANGFUSE_SECRET_KEY="your_langfuse_secret_key_here"
export OPENAI_API_KEY="your_openai_api_key_here"

Run the example

Save the code above as langfuse_via_openinference_workflows.py, then run:

python langfuse_via_openinference_workflows.py

Full source: cookbook/observability/workflows/langfuse_via_openinference_workflows.py