Observability with concurrent threads

[DavideSanvito]

Hello,
I am using Langfuse to trace the execution of a pipeline with mutliple LLM calls and I would have a question about the observability when running some tasks in parallel.
The main pipeline method, decorated with observe(), calls multiple stages, each one implemented as a decorated method.
As a reference, let's assume the pipeline has 5 stages where B1, B2 and B3 can be parallelized, i.e. A - B1/B2/B3 - C.

When running the pipeline sequentially, everything works as expected: the execution of each stage appears as SPAN within the TRACE of the main method.

When paralellizing the 3 Bx stages, I observed the following behaviour:

• with asyncio.gather everything still works, i.e. the pipeline is recorded as a single trace and the stages are nested within it.
• with ThreadPoolExecutor the main method still records the entire duration of the pipeline, in terms of time, but the B1/B2/B3 methods running in different threads are recorded as separated traces.

In order to obtain a single trace, based on https://langfuse.com/docs/sdk/python/decorators#set-custom-ids, I had to manually pass to each B* method an extra langfuse_parent_trace_id argument with the trace id retrieved with langfuse_context.get_current_trace_id().

Is this the proper way or is there a simpler approach to obtain a single trace with threads?

Thanks a lot!

[dosubot[bot]]

Hello, @DavideSanvito!
I'm here to help you with any issues or questions you have. Let's work together to solve this!

I found a similar discussion that might be helpful. In a previous thread, it was confirmed that when using ThreadPoolExecutor, you need to manually pass the langfuse_parent_trace_id and langfuse_parent_observation_id as keyword arguments to each method to maintain a single trace across threads. This approach ensures that the context is correctly maintained across threads ^[1].

Here's an example of how you can implement this:

from concurrent.futures import ThreadPoolExecutor, as_completed
from langfuse.decorators import langfuse_context, observe

@observe()
def execute_task(*args):
    return args

@observe()
def execute_groups(task_args):
    trace_id = langfuse_context.get_current_trace_id()
    observation_id = langfuse_context.get_current_observation_id()

    with ThreadPoolExecutor(3) as executor:
        futures = [
            executor.submit(
                execute_task,
                *task_arg,
                langfuse_parent_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            )
            for task_arg in task_args
        ]

        for future in as_completed(futures):
            future.result()

    return [f.result() for f in futures]

@observe()
def main():
    task_args = [["a", "b"], ["c", "d"]]

    execute_groups(task_args)

main()

langfuse_context.flush()

This method is recommended over manually copying the context, as it directly addresses the issue of maintaining the trace hierarchy across threads ^[1].

To continue talking to Dosu, mention @dosu.

[marcklingen]

@DavideSanvito we have collected some notes on this here: https://langfuse.com/docs/sdk/python/decorators#using-threadpoolexecutors-or-processpoolexecutors

Hope this helps. Let me know in case you have any additional questions!

[DavideSanvito]

That's great, thanks a lot!