Llamaindex Workflow does't work with Instrumentation Module

[ntdanhtma]

Hi there!

After many days of waiting to be able to try LlamaIndex Integration (Instrumentation Module) it traces Llamaindex very well 👍. For example:
Code snippet

with instrumentor.observe(
    trace_name='trace-name',
    session_id='session-id',
    user_id='dev',
    tags=['test']
) as trace:
    retrieved_nodes = await retriever.aretrieve(user_query)
    instrumentor.flush()
    response = response_synthesizer.synthesize(
        user_query, nodes=retrieved_nodes
    )

It worked fine,

---

I tried to build a LlamaIndex Workflow for this, and the script runs as expected. However, the Langfuse instrumentor is not sending traces to the Langfuse server, and no trace appears on the dashboard.

class QueryRAGWorkflow(Workflow):
    
    @step(pass_context=True)
    async def prepare_for_retrieval(
        self,
        context: Context,
        event: StartEvent   
    ) -> PrepEvent | None:
        """Prepare for retrieval"""
        
        query: str | None = event.get('query', None)
        retriever_kwargs: dict | None = event.get('retriever_kwargs', None)
        synthesizer_kwargs: dict | None = event.get('synthesizer_kwargs', None)
        if not(query or retriever_kwargs or synthesizer_kwargs):
            return None

        await context.set('query', query)
        await context.set('retriever_kwargs', retriever_kwargs)
        await context.set('synthesizer_kwargs', synthesizer_kwargs)
        return PrepEvent()
    
    @step(pass_context=True)
    async def retrieve(
        self,
        context: Context,
        event: PrepEvent,
    ) -> RetrievalEvent | None:
        """Retrieve the relevant node for the query."""
        query: str | None = await context.get('query', None)
        if query is None:
            return None
        retriever_kwargs = await context.get('retriever_kwargs', {})
        retriever_helper = RetrieverHelper()
        retriever = retriever_helper.get_retriever(
            **retriever_kwargs
        )
        retrieved_nodes = await retriever.aretrieve(query)
        return RetrievalEvent(retrieved_nodes=retrieved_nodes)
    
    @step
    async def node_postprocess(
        self,
        event: RetrievalEvent
    ) -> NodePostProcessEvent | None:
        nodes = event.retrieved_nodes
        postprocessor = SentenceEmbeddingOptimizer(
            embed_model=embedding, # Settings.embed_model
            percentile_cutoff=0.5
        )
        selected_nodes = postprocessor.postprocess_nodes(nodes)
        return NodePostProcessEvent(selected_nodes=selected_nodes)
    
    @step(pass_context=True)
    async def synthesize(
        self,
        context: Context,
        event: NodePostProcessEvent
    ) -> StopEvent | None:
        query: str | None = await context.get('query', None)
        if query is None:
            return None
        nodes = event.selected_nodes
        synthesizer_kwargs = await context.get('synthesizer_kwargs', {})
        synthesizer = get_response_synthesizer(
            **synthesizer_kwargs
        )
        response = await synthesizer.asynthesize(query=query, nodes=nodes)
        return StopEvent(result=response)


query = 'What is ABC?'
retriever_kwargs = {
    'vector_store_query_mode': 'hybrid',
    'index_ids': ['1f51ed30-9e2e-4d26-ab93-20fec231ebd7']
}
synthesizer_kwargs = {
    'llm': llm,
    'response_mode': ResponseMode.COMPACT,
    'use_async': True,
    'streaming': False
}

rag_workflow = QueryRAGWorkflow(verbose=True, timeout=10)
with instrumentor.observe(
    trace_name='trace-name',
    session_id='session-id',
    user_id='dev',
    tags=['test']
) as trace:
    result = await rag_workflow.run(
        query=query,
        retriever_kwargs=retriever_kwargs,
        synthesizer_kwargs=synthesizer_kwargs
    )
print(result)

---

Moreover, I also tested by trying simpler Workflows (JokeFlow) in the LlamaIndex Workflow, it seems that it only works on completion functions with some serialization errors.

the instrumentation module for the LLamaindex Workflow appears to be incomplete. I'm unsure if it's an actual bug. Could you assist me in investigating this further and help confirm if this is an issue or a known limitation?

Thanks.

[dosubot[bot]]

@ntdanhtma Hello there! I'm here to help you with your Llamaindex Workflow issue. Let's get this sorted out together.

The issue with the Langfuse instrumentor not sending traces in the LlamaIndex Workflow could be due to several factors:

1. Asynchronous Tracing: Ensure your setup correctly handles asynchronous operations. Langfuse runs all tracing integrations asynchronously, so make sure your application is integrated according to the quickstart guide.

2. Configuration: Double-check your API credentials and tracing configuration, including LANGFUSE_HOST or LANGFUSE_BASEURL.

3. Manual Flushing: For short-lived applications, ensure events are manually flushed before the application exits to avoid losing data.

4. Serialization Errors: The serialization errors in simpler workflows might indicate a bug or limitation, especially since the integration is in beta. It's recommended to use the callback-based integration for production until the instrumentation module stabilizes.

If these steps don't resolve the issue, it might be a limitation of the current beta version of the instrumentation module ^[1][2].

To continue talking to Dosu, mention @dosu.

[ntdanhtma]

I have checked and tried the above, it might be a limitation of the current beta version of the instrumentation module. If you have any information about this please let me know. Thanks.

[dosubot[bot]]

The current beta version of the LlamaIndex instrumentation module is indeed in active development, and its behavior and APIs may change in future releases. It's recommended to use the callback-based integration for production use cases until the instrumentation module is stable. Additionally, only Python is supported at the moment, and there might be serialization issues as you've encountered ^[1][2].

To continue talking to Dosu, mention @dosu.

[marcklingen]

thanks for the feedback! @hassiebp wdyt, turn this into an issue?

[hassiebp]

Thanks a lot for your detailed feedbacke, @ntdanhtma!

• Missing traces on workflow runs: This seems to be an upstream LlamaIndex issue that workflow runs aren't properly instrumented yet. If you'd like to see this supported, please open a feature request in the LlamaIndex repo or create your own custom instrumentation.
• Serialization issues: First of all, these issues do not raise any exceptions in you app hopefully, as these should be silent warnings only. There is an open issue in the LlamaIndex repo with regards to serialization issues for some of their objects. This fix has been merged a few days ago, could you please try again with version llama-index version v0.11.17?

[ntdanhtma]

I quickly tested the setup with another observation integration, such as Arize Phoenix, to confirm whether LlamaIndex is functioning properly. It successfully traced the workflow. Just a bit of additional information for you. 😊

[marcklingen]

thank you! cc @hassiebp

[ntdanhtma]

@hassiebp Rarely, it still sends traces of the Workflow to Langfuse Server (but randomly which I can't reproduce) and loses some information (name, session, tag, etc), and includes error Spans.

I also ran the debug and saw that all Events were successfully traced (tokens, costs, parameters, etc), but it seemed like it was not sent to the Langfuse server, and errors were appearing from the Langfuse server v2.85.0 (self-hosted) logs as below

langfuse_logs.txt

[hassiebp]

Thanks a lot for digging deeper on this @ntdanhtma - I have created this issue to track progress. #3736

[plagree]

I have the same issue with tracing of LlamaIndex workflows via Langfuse. It loses some information like names or complete span events. I'm using the concurrent execution feature (see https://docs.llamaindex.ai/en/stable/understanding/workflows/concurrent_execution/#emitting-multiple-events) and it's often only reporting a subset of the several executions of the same step. In addition, when waiting for several events like this:

    @step
    async def step_three(
        self,
        ctx: Context,
        ev: StepACompleteEvent | StepBCompleteEvent | StepCCompleteEvent,
    ) -> StopEvent:
        print("Received event ", ev.result)

        # wait until we receive 3 events
        if (
            ctx.collect_events(
                ev,
                [StepCCompleteEvent, StepACompleteEvent, StepBCompleteEvent],
            )
            is None
        ):
            return None

        # do something with all 3 results together
        return StopEvent(result="Done")

it's typically reporting empty spans as well (the return None ones): is this on purpose?

FYI @hassiebp

[lambdaofgod]

I can confirm that this still persists, even for a Workflow without concurrent events.