Does langfuse trace parallel api calls? I tried a sample with parallel call but it didn't traced.

[ksundarraj-c-rpx]

This is my function to run gpt-4o and I used langfuse openai to trace this in UI,

prc = langfuse.get_prompt("relevant_companies")
prompt_1 = prc.prompt
prompt_1

def get_gpt_result_1(prompt_query, response_schema, concat_text):
    
    output_parser = StructuredOutputParser.from_response_schemas(response_schema)
    format_instructions = output_parser.get_format_instructions()
    
    prompt = SystemMessagePromptTemplate.from_template(template=prompt_query)
    _input =prompt.format_messages(concat_text=concat_text, format_instructions=format_instructions)
    
    message = {"role": "system", "content": _input[0].content}
    
    try:
        
        response = openai.chat.completions.create(
          model=gpt_model,
          messages=[message],
          langfuse_prompt=prc                 #This is the prompt I got from UI using prompt.get
        )
    
        res = {'gpt_response':response.choices[0].message.content, 
               'prompt_tokens':response.usage.prompt_tokens, 
               'completion_tokens':response.usage.completion_tokens
              }
        
    except Exception as e:
        print(str(e))
    
    return res

I use another function which will call this to get gpt result, so it works fine if I perform sequence of calls I can see the traces in UI but, if I call the other function in parallel way to speed up like,

def get_prc(rows):
    
    text = rows['concat_text']
    
    response = get_gpt_result_1(prompt_query=prompt_1, concat_text=text, 
                                response_schema=response_schema_1)
    
    prompt_1_res = response['gpt_response']
    prompt_1_res = post_process(prompt_1_res)
    rows['prompt_1_res'] = prompt_1_res
    rows['prompt_1_prompt_tokens'] = response['prompt_tokens']
    rows['prompt_1_output_tokens'] = response['completion_tokens']
    rows['prompt_1_cost'] = ((response['prompt_tokens']*5)+(response['completion_tokens']*15))/1000000
    
    return rows


def get_prc_apply(rows):
    rows = rows.apply(get_prc,axis=1)
    return rows
    
data_1 = data.groupby(np.arange(data.shape[0])//60).parallel_apply(get_prc_apply)

So when I call like this everything is running fine I am getting output but I could not see any trace in langfuse UI.

[dosubot[bot]]

Hey @ksundarraj-c-rpx! I'm here to help you with any bugs, questions, or contributions you have. Let's work together to solve this issue!

To ensure that parallel API calls are traced in Langfuse, you should follow these steps:

1. Verify Integration: Ensure that your application is correctly integrated with Langfuse.
2. Check API Credentials: Make sure that the API credentials used in your application match those created in your Langfuse project settings.
3. Inspect Tracing Configuration: Ensure that your tracing configuration is correctly set up.
4. Review Logs: Check the logs of your application to see if there are any errors related to Langfuse.
5. Manual Flushing: If you are using short-lived applications, ensure that you are manually flushing the events before the application exits.
6. Batching Configuration: In high throughput applications, verify the batching configuration to ensure that events are being sent in a timely manner.
7. Use Nested Spans and Events: Create nested spans and events to ensure that all parallel API calls are traced. This involves creating a trace, adding spans, and then adding events or further spans within those spans. Ensure that each span and event is correctly associated with its parent trace or span ^[1][2].

Additionally, Langfuse's client SDKs and integrations are designed to queue and batch requests in the background to optimize API calls and network time. You can customize the batching behavior to suit your needs by configuring options such as flushAt and flushInterval. For short-lived applications, it is crucial to use the flush method to ensure all events are sent before the application exits ^[3].

By following these steps and utilizing Langfuse's features, you can ensure that parallel API calls are properly traced in Langfuse.

To continue talking to Dosu, mention @dosu.

[ksundarraj-c-rpx]

def get_gpt_result_1(prompt_query, response_schema, concat_text):
    
    output_parser = StructuredOutputParser.from_response_schemas(response_schema)
    format_instructions = output_parser.get_format_instructions()
    
    prompt = SystemMessagePromptTemplate.from_template(template=prompt_query)
    _input =prompt.format_messages(concat_text=concat_text, format_instructions=format_instructions)
    
    message = {"role": "system", "content": _input[0].content}
    
    try:
        
        response = openai.chat.completions.create(
        model=gpt_model,
        messages=[message],
        langfuse_prompt=prc,
        name="test-run"
        )
        
        langfuse.flush()
    
        res = {'gpt_response':response.choices[0].message.content, 
               'prompt_tokens':response.usage.prompt_tokens, 
               'completion_tokens':response.usage.completion_tokens
              }
        
    except Exception as e:
        print(str(e))
    
    return res

Have checked the config keys, I am using the correct keys only and there is no error in langfuse setup. I used to manually flush() the trace but still no traces in the UI when using a parallel run.

[dosubot[bot]]

To ensure that parallel API calls are properly traced in Langfuse when using the provided code snippet, you need to create a parent trace and pass it to each parallel call. This allows you to group multiple generations into a single trace. Here is an example of how to do this:

import os
from langfuse import Langfuse
from langfuse.decorators import observe

# Set environment variables
os.environ["LANGFUSE_PUBLIC_KEY"] = "your_public_key"
os.environ["LANGFUSE_SECRET_KEY"] = "your_secret_key"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
os.environ["OPENAI_API_KEY"] = "your_openai_api_key"

langfuse = Langfuse()

def get_gpt_result_1(prompt_query, response_schema, concat_text, parent_trace):
    output_parser = StructuredOutputParser.from_response_schemas(response_schema)
    format_instructions = output_parser.get_format_instructions()
    
    prompt = SystemMessagePromptTemplate.from_template(template=prompt_query)
    _input = prompt.format_messages(concat_text=concat_text, format_instructions=format_instructions)
    
    message = {"role": "system", "content": _input[0].content}
    
    try:
        response = openai.chat.completions.create(
            model=gpt_model,
            messages=[message],
            langfuse_prompt=prc,
            parent=parent_trace,  # Pass the parent trace here
            name="test-run"
        )
        
        res = {
            'gpt_response': response.choices[0].message.content, 
            'prompt_tokens': response.usage.prompt_tokens, 
            'completion_tokens': response.usage.completion_tokens
        }
        
    except Exception as e:
        print(str(e))
    
    return res

def get_prc(rows):
    text = rows['concat_text']
    
    # Create a parent trace for the parallel calls
    parent_trace = langfuse.trace(name="Parallel GPT Calls")
    
    response = get_gpt_result_1(prompt_query=prompt_1, concat_text=text, 
                                response_schema=response_schema_1, parent_trace=parent_trace)
    
    prompt_1_res = response['gpt_response']
    prompt_1_res = post_process(prompt_1_res)
    rows['prompt_1_res'] = prompt_1_res
    rows['prompt_1_prompt_tokens'] = response['prompt_tokens']
    rows['prompt_1_output_tokens'] = response['completion_tokens']
    rows['prompt_1_cost'] = ((response['prompt_tokens']*5)+(response['completion_tokens']*15))/1000000
    
    # Update the parent trace with the output
    parent_trace.update(output=rows)
    
    return rows

def get_prc_apply(rows):
    rows = rows.apply(get_prc, axis=1)
    return rows

data_1 = data.groupby(np.arange(data.shape[0])//60).parallel_apply(get_prc_apply)

# Ensure all traces are flushed
langfuse.flush()

This code ensures that all API calls made within the get_prc function are traced under a single parent trace, providing a comprehensive view of the interactions. Additionally, calling langfuse.flush() ensures that all traces are sent to the server before the application exits ^[1][2].

To continue talking to Dosu, mention @dosu.

[ksundarraj-c-rpx]

Did not worked

[marcklingen]

Is parallel_apply a pandas api? Depending on how it shuffles data across cores this might be an issue.

Otherwise, can you try to import OpenAI (class) from langfuse.openai and then use this across processes? this might be more stable as it will then create a langfuse instance once instead of one on each of your processes. You could also add openai.flush_langfuse() to make sure that your client waits for all data to be flushed to the langfuse api before terminating.

[ksundarraj-c-rpx]

parallel apply will send data in a parallel way, so in my code i'm sending 60 data at same time to speed up the process. I am actually using the OpenAI class from langfuse, tried various things like updating the trace manually inside the get_gpt_result_1 function itself but not sure why it is not working.

[marcklingen]

where do you initialize the openai class?