feat: Unify traces of sub-pipelines within pipelines with Langfuse

[sjrl]

Related Issues

• fixes AsyncPipeline Creates Multiple Traces Instead of a Single Unified Trace in Langfuse #1604 and Support Unified Tracing with LangfuseConnector Across Main and Sub-Pipelines in Haystack #1605

Proposed Changes:

• Correctly pass the parent_span in the situation we have a sub-pipeline within a pipeline

How did you test it?

• Added integration test and tested locally
• Also expanded unit tests and did some refactoring

Notes for the reviewer

Here is some example code

import os

os.environ["OPENAI_API_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true"

from haystack import Pipeline, SuperComponent
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.utils.auth import Secret

from haystack_integrations.components.connectors.langfuse import LangfuseConnector


sub_pipe = Pipeline()
sub_pipe.add_component("llm", OpenAIChatGenerator())
super_llm = SuperComponent(
    pipeline=sub_pipe,
)

top_sub_pipe = Pipeline()
top_sub_pipe.add_component("prompt_builder", ChatPromptBuilder())
top_sub_pipe.add_component("super_llm", super_llm)
top_sub_pipe.connect("prompt_builder.prompt", "super_llm.messages")
super_prompt_and_llm = SuperComponent(
    pipeline=top_sub_pipe,
)

# Main pipeline
pipe = Pipeline()
pipe.add_component("tracer", LangfuseConnector("Sub-Sub Pipeline Test"))
pipe.add_component("super_prompt_and_llm", super_prompt_and_llm)

messages = [
    ChatMessage.from_system("Always respond in German even if some input data is in other languages."),
    ChatMessage.from_user("Tell me about {{location}}"),
]

response = pipe.run(
    data={
        "super_prompt_and_llm": {
            "template_variables": {"location": "Berlin"},
            "template": messages,
        },
        "tracer": {
            "invocation_context": {"some_key": "some_value"},
        },
    }
)
print(response["super_prompt_and_llm"]["replies"][0])
print(response["tracer"]["trace_url"])

which produces this trace in Langfuse

Checklist

• I have read the contributors guidelines and the code of conduct
• I have updated the related issue with new insights and changes
• I added unit tests and updated the docstrings
• I've used one of the conventional commit types for my PR title: fix:, feat:, build:, chore:, ci:, docs:, style:, refactor:, perf:, test:.

[sjrl]

This was moved from the test_tracing.py file

[sjrl]

This file got refactored into the test_tracer.py file, following our standard practice of having the test files follow the same name structure as our module files.

[sjrl]

This is the new test, that checks that when using sub-pipelines within a pipeline we correctly collect the traces together.

[sjrl]

This line is what fixed the issue

[sjrl]

pinging @vblagoje in case you are interested

[vblagoje]

Very nice @sjrl

[Amnah199]

Great to see that you've introduced multiple test cases. Maybe in future we can cover some edge casestest_invalid_span_handler to verify how the system behaves. But its your call.

[sjrl]

Good point! I think that is something we can do in the future.

[Amnah199]

Great work! Although I may not have full context on Langfuse, I reviewed the code based on my understanding. I am guessing @vblagoje has also had a chance to take a look.