fix: Performance optimizations and value error when streaming in langfuse

[Redna]

Related Issues

• fixes langfuse: value error for usage when streaming is enabled #797
• fixes Optimise LangfuseTracer - move flush to separate thread #782

Proposed Changes:

• flushing the data can be controlled now by an environment variable HAYSTACK_LANGFUSE_ENFORCE_FLUSH (Added a hint on the documentation that it needs to be done manually instead
• flushing will be done now only once before closing the trace
• traces are now independent for each pipeline run (Remove the last span from the context list in order to properly trace down the time of one pipeline run independently)
• Usage will be set to None when it is an empty dict to solve the langfuse value error

How did you test it?

• Unit tests are checking for enabling and disabling HAYSTACK_LANGFUSE_ENFORCE_FLUSH
• Unit tests check if the last trace will be ended and removed from the context

** Manual tests: **

import os

os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"]="true"
os.environ["LANGFUSE_SECRET_KEY"]="<secret_key>"
os.environ["LANGFUSE_PUBLIC_KEY"]="<public_key>"
os.environ["LANGFUSE_HOST"]="https://cloud.langfuse.com"


from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore

from haystack.tracing import tracer

from haystack.utils import Secret

from haystack_integrations.components.connectors.langfuse import LangfuseConnector

def print_callback(data):
    print(data.content, end="")

docstore = InMemoryDocumentStore()
docstore.write_documents([Document(content="Rome is the capital of Italy"), Document(content="Paris is the capital of France")])

query = "What is the capital of France?"

template = """
Given the following information, answer the question.

Context: 
{% for document in documents %}
    {{ document.content }}
{% endfor %}

Question: {{ query }}?
"""
pipe = Pipeline()

langfuse_connector = LangfuseConnector(name="TEST")

pipe.add_component("tracer", langfuse_connector)

pipe.add_component("retriever", InMemoryBM25Retriever(document_store=docstore))
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component("llm", OpenAIGenerator(api_key=Secret.from_token("<your-api-key>"),
                                          api_base_url="http:https://localhost:30091/v1",
                                          streaming_callback=print_callback))

pipe.connect("retriever", "prompt_builder.documents")
pipe.connect("prompt_builder", "llm")

try: 
    res=pipe.run({
        "prompt_builder": {
            "query": query
        },
        "retriever": {
            "query": query
        }
    })

    print(res)
finally:
    tracer.actual_tracer.flush()

*** Enforce flush disabled ***

*** Enforce flush enabled ***

Notes for the reviewer

Notice the 4s to 40ms difference between the two settings. I think it is crucial to enable the user the option to flush the data lazy.

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:.

[anakin87]

@masci feel free to review it if/when you have time.

[masci]

Thanks for the PR, it looks good to me!