How can I set or edit the top-level trace name to match the first executed span?

[ystkfujii]

Description

I’m trying to understand how the top-level trace name is determined when using the Langfuse Python SDK.

In the example below, the top-level trace name in the Langfuse UI always appears as process_task, even though the first executed function (and the one decorated with observe) is main. I also tried updating the trace name with update_current_trace(name="Demo"), but it didn’t change the displayed name of the top-level trace.

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from langfuse import observe,get_client

@observe()
def process_task(task_id: int) -> str:
    print(f"Task {task_id} started")
    time.sleep(1)  # Sleep for 1 second
    result = f"Task {task_id} completed"
    print(result)
    return result

@observe()
def main():
    num_tasks = 5
    max_workers = 3

    langfuse_client = get_client()
    observation_id = langfuse_client.get_current_observation_id()
    trace_id = langfuse_client.get_current_trace_id()
    
    langfuse_client.update_current_trace(
        name="Demo",
    )

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
            ]

        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

if __name__ == "__main__":
    main()

Question

How can I ensure that the top-level trace name in Langfuse reflects the first executed span (main) or a custom name (e.g., "Demo") instead of the function name of a child span like process_task?

Environment

• Langfuse Python SDK: v3.8.1
• Python: 3.12.3
• Langfuse deployment: SaaS (Langfuse Cloud)

[inkeep-ai-assistant[bot]]

Based on the sources, the issue you're experiencing is related to how trace names are determined when using the @observe() decorator with the Langfuse Python SDK v3(1)(2).

How Trace Names Work

When using @observe(), the decorator creates observations (spans) that are nested based on OpenTelemetry context propagation(1). The trace name is typically derived from the root observation. In your case, because you're manually passing langfuse_trace_id and langfuse_parent_observation_id to the threaded process_task calls, the SDK may be treating the first process_task execution as the root-level span name(1).

Solution: Use update_current_trace() with name Parameter

The update_current_trace() method should work to set a custom trace name(1). However, based on your code, you're calling it inside the main() function. The key is ensuring this update happens at the trace level.

Try this approach:

from langfuse import observe, get_client

@observe()
def process_task(task_id: int) -> str:
    print(f"Task {task_id} started")
    time.sleep(1)
    result = f"Task {task_id} completed"
    print(result)
    return result

@observe()
def main():
    num_tasks = 5
    max_workers = 3

    langfuse_client = get_client()
    
    # Update the trace name at the beginning
    langfuse_client.update_current_trace(name="Demo")
    
    observation_id = langfuse_client.get_current_observation_id()
    trace_id = langfuse_client.get_current_trace_id()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
        ]

        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

if __name__ == "__main__":
    main()
    
    # Flush events in short-lived applications
    langfuse = get_client()
    langfuse.flush()

(1)(2)

Alternative: Use Context Managers for More Control

If the decorator approach doesn't give you the desired trace name, you can use context managers for explicit control over trace naming:

from langfuse import get_client

langfuse = get_client()

with langfuse.start_as_current_span(name="Demo") as root_span:
    # Your processing logic here
    num_tasks = 5
    max_workers = 3
    
    observation_id = langfuse.get_current_observation_id()
    trace_id = langfuse.get_current_trace_id()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
        ]
        
        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

langfuse.flush()

(1)(2)

This approach explicitly creates a span named "Demo" as the root span, which should become your trace name in the Langfuse UI(1).

📚 Sources:

1. Instrumentation
2. Overview

---

Have another question? Just tag @inkeep.

[ystkfujii]

I tried both approaches, but the behavior was still unclear. 😭

Result of the first solution

import time
from langfuse import observe, get_client
from concurrent.futures import ThreadPoolExecutor, as_completed

@observe()
def process_task(task_id: int) -> str:
    print(f"Task {task_id} started")
    time.sleep(1)
    result = f"Task {task_id} completed"
    print(result)
    return result

@observe()
def main2():
    num_tasks = 5
    max_workers = 3

    langfuse_client = get_client()
    
    # Update the trace name at the beginning
    langfuse_client.update_current_trace(name="Demo")
    
    observation_id = langfuse_client.get_current_observation_id()
    trace_id = langfuse_client.get_current_trace_id()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
        ]

        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

if __name__ == "__main__":
    main2()
    
    # Flush events in short-lived applications
    langfuse = get_client()
    langfuse.flush()

Result of the alternative approach

import time
from langfuse import observe, get_client
from concurrent.futures import ThreadPoolExecutor, as_completed

from langfuse import get_client

langfuse = get_client()

@observe()
def process_task(task_id: int) -> str:
    print(f"Task {task_id} started")
    time.sleep(1)
    result = f"Task {task_id} completed"
    print(result)
    return result

with langfuse.start_as_current_span(name="Demo") as root_span:
    # Your processing logic here
    num_tasks = 5
    max_workers = 3
    
    observation_id = langfuse.get_current_observation_id()
    trace_id = langfuse.get_current_trace_id()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
        ]
        
        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

langfuse.flush()

[jannikmaierhoefer]

Hi @ystkfujii, could you please try adding a placeholder span to define the trace name:

langfuse.start_observation(name="Agent started").update_trace(name="Trace name").end()

[ystkfujii]

Hi @jannikmaierhoefer, Thanks for checking!
Where should I set the placeholder?
I tried the following code, but it didn’t work:

Details

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from langfuse import observe,get_client

@observe()
def process_task(task_id: int) -> str:
    print(f"Task {task_id} started")
    time.sleep(1)  # Sleep for 1 second
    result = f"Task {task_id} completed"
    print(result)
    return result

@observe()
def main():
    num_tasks = 5
    max_workers = 3

    langfuse_client = get_client()
    
    placeholder = langfuse_client.start_observation(name="Agent started")
    placeholder.update_trace(name="Demo")
    placeholder.end()
    
    observation_id = langfuse_client.get_current_observation_id()
    trace_id = langfuse_client.get_current_trace_id()
    
    langfuse_client.update_current_trace(
        name="Demo",
    )

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                process_task,
                i,
                langfuse_trace_id=trace_id,
                langfuse_parent_observation_id=observation_id,
            ) 
            for i in range(num_tasks)
            ]

        results = []
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
    
    langfuse_client.flush() 

if __name__ == "__main__":
    main()

[ystkfujii]

@jannikmaierhoefer

Any updates on this?
It might be related to the same root cause as #10430 🤔

[jannikmaierhoefer]

Hi @ystkfujii, yes, this seems to be related. Lets continue the discisson on the GitHub issue.