Skip to main content
This example shows how the Launchpad integrates with Langfuse to trace every workflow step and every LLM call. It ships as LangfuseTracingWorkflow in app/launchpad/workflows/examples/langfuse_tracing/ and is registered as WorkflowRegistry.LANGFUSE_TRACING.
For how tracing is wired into the core Workflow class, see Langfuse Integration.

​
What the workflow does

A simple moderation pipeline for user comments:
  1. ViolationDetectionNode — an AgentNode that classifies whether a comment violates policy.
  2. ContextSummaryResult — an AgentNode that summarizes the comment for the audit log.
  3. RemoveCommentNode — a plain Node that deletes the comment when the previous step flagged it.
Each node runs inside its own Langfuse span when enable_tracing=True, so you can see timings, inputs, outputs, and LLM calls for the whole run in the Langfuse dashboard.

​
Schema

class LangfuseTracingEventSchema(BaseModel):
    event: str
    timestamp: datetime
    comment_id: str
    thread_id: str
    user_id: str
    content: str

​
Workflow definition

class LangfuseTracingWorkflow(Workflow):
    workflow_schema = WorkflowSchema(
        description="",
        event_schema=LangfuseTracingEventSchema,
        start=ViolationDetectionNode,
        nodes=[
            NodeConfig(
                node=ViolationDetectionNode,
                connections=[ContextSummaryResult],
            ),
            NodeConfig(
                node=ContextSummaryResult,
                connections=[RemoveCommentNode],
            ),
            NodeConfig(
                node=RemoveCommentNode,
                connections=[],
            ),
        ],
    )

​
Violation detection node

class ViolationDetectionNode(AgentNode):
    class OutputType(AgentNode.OutputType):
        comment_id: str
        violation: bool
        reason: Optional[str] = None

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            instructions=(
                "Determine whether the comment is a violation or not. If it is a "
                "violation, provide a reason for violation. If it is not a "
                "violation, provide a reason for non-violation."
            ),
            output_type=self.OutputType,
            deps_type=LangfuseTracingEventSchema,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-5.4-mini",
            instrument=True,
        )

    async def process(self, task_context: TaskContext) -> TaskContext:
        event: LangfuseTracingEventSchema = task_context.event

        @self.agent.instructions
        async def add_context() -> str:
            return event.model_dump_json()

        result = await self.agent.run(user_prompt=event.model_dump_json())
        self.save_output(result.output)
        return task_context
The other two nodes are thin — ContextSummaryResult follows the same pattern with a summarization prompt, and RemoveCommentNode reads ViolationDetectionNode.OutputType via get_output() and logs the deletion.

​
Running the example

1

Set Langfuse credentials

Add to .env (or your shell environment):
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com
2

Run the playground script

uv run playground/langfuse_tracing.py
The script loads app/launchpad/workflows/examples/langfuse_tracing/request_examples/violation.json, instantiates the workflow via WorkflowRegistry.LANGFUSE_TRACING.value(), and runs it.
3

Inspect the trace

Open the Langfuse dashboard. You should see a trace named LangfuseTracingWorkflow with child spans for each node (ViolationDetectionNode, ContextSummaryResult, RemoveCommentNode) and the underlying LLM generations.
The playground instantiates the workflow without arguments, which defaults to enable_tracing=False. To capture traces, update the script to WorkflowRegistry.LANGFUSE_TRACING.value(enable_tracing=True), or instantiate LangfuseTracingWorkflow(enable_tracing=True) directly.

​
Example event

app/launchpad/workflows/examples/langfuse_tracing/request_examples/violation.json: 
{
  "event": "comment_posted",
  "timestamp": "2026-04-17T12:00:00Z",
  "comment_id": "comment-123",
  "thread_id": "thread-abc",
  "user_id": "user-42",
  "content": "This is a test comment that should be evaluated for policy violations."
}