Issue with Langfuse tracing. All traces don't seem to show up.

[rahulkrprajapati]

My code is defined as something like:

At the top level above the class.

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler

langfuse_callback_handler = LlamaIndexCallbackHandler(
    public_key="pk-...",
    secret_key="sk-.....",
    host="https://cloud.langfuse.com"
)
Settings.callback_manager = CallbackManager([langfuse_callback_handler])

Then Inside the class:

I have two functions that I want to enable trace for:

@observe()
    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            bm24_retriever = BM25Retriever.from_defaults(
                docstore=...
            )
            fusion_retriever = QueryFusionRetriever(
                ...
            )
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = fusion_retriever.retrieve(query_bundle)
            langfuse_callback_handler.flush()
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
               ...
            )
            langfuse_callback_handler.flush()
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            langfuse_callback_handler.flush()
            # px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")
    @observe()
    def process_streamlit_retrieved_nodes_using_llm(self, query="", reranked_nodes=[]):
        try:
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            response = self.response_synthesizer.synthesize(
                ...
            )
            langfuse_callback_handler.flush()
            return response
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred generating response from LLM: {e}")

This does not work.

[rahulkrprajapati]

@dosu Could you make changes to the code to make sure the trace works.

[dosubot[bot]]

The issue might be that the observe decorator you are using does not properly integrate with the CallbackManager for tracing. Instead, you should use the trace_method decorator provided by LlamaIndex to ensure that the methods are properly traced.

Additionally, ensure that you have set the global handler for Langfuse as described in the Langfuse integration documentation.

Here is how you can modify your code to use the trace_method decorator and set the global handler:

At the top level above the class:

from llama_index.core import Settings, set_global_handler
from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler

# Set the global handler for Langfuse
set_global_handler("langfuse")

langfuse_callback_handler = LlamaIndexCallbackHandler(
    public_key="pk-...",
    secret_key="sk-.....",
    host="https://cloud.langfuse.com"
)
Settings.callback_manager = CallbackManager([langfuse_callback_handler])

Inside the class, modify the functions to use trace_method:

from llama_index.core.callbacks.utils import trace_method

class YourClass:
    @trace_method("process_streamlit_fetch_query_results")
    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            bm24_retriever = BM25Retriever.from_defaults(
                docstore=self.docstore, similarity_top_k=self.similarity_top_k
            )
            fusion_retriever = QueryFusionRetriever(
                [self.retriever, bm24_retriever],
                similarity_top_k=self.similarity_top_k,
                num_queries=1,  # set this to 1 to disable query generation
                mode="reciprocal_rerank",
                use_async=True,
                verbose=verbose,
            )
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = fusion_retriever.retrieve(query_bundle)
            langfuse_callback_handler.flush()
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            langfuse_callback_handler.flush()
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            langfuse_callback_handler.flush()
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

    @trace_method("process_streamlit_retrieved_nodes_using_llm")
    def process_streamlit_retrieved_nodes_using_llm(self, query="", reranked_nodes=[]):
        try:
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            Settings.callback_manager = CallbackManager([langfuse_handler])
            response = self.response_synthesizer.synthesize(
                query=query, nodes=reranked_nodes
            )
            langfuse_callback_handler.flush()
            return response
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred generating response from LLM: {e}")

This should ensure that the methods are properly traced and the traces show up as expected. Additionally, make sure you have set your environment variables LANGFUSE_SECRET_KEY, LANGFUSE_PUBLIC_KEY, and LANGFUSE_HOST as shown in your Langfuse project settings ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

To resolve the ValueError and ensure that the CallbackManager does not attempt to add multiple handlers of the same type, you need to check for existing handlers of the same type before adding a new handler. Here is how you can modify your function to include this check:

from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler

class YourClass:
    @trace_method("process_streamlit_fetch_query_results")
    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            langfuse_handler = langfuse_context.get_current_llama_index_handler()
            callback_manager = Settings.callback_manager

            # Check if a handler of the same type already exists
            for existing_handler in callback_manager.handlers:
                if isinstance(existing_handler, type(langfuse_handler)):
                    raise ValueError(
                        "Cannot add two handlers of the same type "
                        f"{type(langfuse_handler)} to the callback manager."
                    )

            # Add the new handler if no handler of the same type exists
            callback_manager.add_handler(langfuse_handler)
            Settings.callback_manager = callback_manager

            bm24_retriever = BM25Retriever.from_defaults(
                docstore=self.docstore, similarity_top_k=self.similarity_top_k
            )
            fusion_retriever = QueryFusionRetriever(
                [self.retriever, bm24_retriever],
                similarity_top_k=self.similarity_top_k,
                num_queries=1,  # set this to 1 to disable query generation
                mode="reciprocal_rerank",
                use_async=True,
                verbose=verbose,
            )
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = fusion_retriever.retrieve(query_bundle)
            langfuse_callback_handler.flush()

            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            langfuse_callback_handler.flush()

            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            langfuse_callback_handler.flush()

            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

This code ensures that before adding a new handler, it checks if a handler of the same type already exists in the CallbackManager. If it does, it raises a ValueError. If not, it proceeds to add the new handler ^[1].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu

Traceback (most recent call last):
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/llama_index/core/callbacks/utils.py", line 31, in wrapper
    callback_manager = getattr(self, callback_manager_attr)
AttributeError: 'DocCore' object has no attribute 'callback_manager'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/rahulprajapati/Developer/ProtoSearchBackend/DocCore/picore.py", line 376, in process_streamlit_fetch_query_results
    langfuse_handler = langfuse_context.get_current_llama_index_handler()
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/langfuse/decorators/langfuse_decorator.py", line 513, in get_current_llama_index_handler
    observation = _observation_stack_context.get()[-1]
IndexError: list index out of range
2024-07-23 13:05:15.404 Uncaught app exception
Traceback (most recent call last):
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/llama_index/core/callbacks/utils.py", line 31, in wrapper
    callback_manager = getattr(self, callback_manager_attr)
AttributeError: 'DocCore' object has no attribute 'callback_manager'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/rahulprajapati/Developer/ProtoSearchBackend/DocCore/picore.py", line 376, in process_streamlit_fetch_query_results
    langfuse_handler = langfuse_context.get_current_llama_index_handler()
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/langfuse/decorators/langfuse_decorator.py", line 513, in get_current_llama_index_handler
    observation = _observation_stack_context.get()[-1]
IndexError: list index out of range

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 535, in _run_script
    exec(code, module.__dict__)
  File "/Users/rahulprajapati/Developer/ProtoSearchBackend/ragDemoApp.py", line 77, in <module>
    reranked_nodes = rag_obj.process_streamlit_fetch_query_results(query=query_text)
  File "/Users/rahulprajapati/anaconda3/envs/rag-exp/lib/python3.10/site-packages/llama_index/core/callbacks/utils.py", line 38, in wrapper
    return func(self, *args, **kwargs)
  File "/Users/rahulprajapati/Developer/ProtoSearchBackend/DocCore/picore.py", line 418, in process_streamlit_fetch_query_results
    raise Exception(f"An error occurred retrieving: {e}")
Exception: An error occurred retrieving: list index out of range
^C  Stopping...

Implement tracing for my function from scratch and tell me the steps. USe langtrace

[dosubot[bot]]

To implement tracing for your function from scratch using Langtrace, follow these steps:

1. Install the necessary packages:

   pip install llama-index llama-index-callbacks-langfuse

2. Configure your environment:
   If you haven't done so yet, sign up on Langfuse and obtain your API keys from the project settings.

   import os
   
   # Langfuse
   os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
   os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
   os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # 🇪🇺 EU region, 🇺🇸 US region: "https://us.cloud.langfuse.com"
   
   # OpenAI
   os.environ["OPENAI_API_KEY"] = "sk-..."

3. Register the Langfuse callback handler:
   You have two options to set up the Langfuse callback handler.

   Option 1: Set global LlamaIndex handler

   from llama_index.core import set_global_handler
   
   set_global_handler("langfuse")
   langfuse_callback_handler = global_handler

   Option 2: Use Langfuse callback directly

   from llama_index.core import Settings
   from llama_index.core.callbacks import CallbackManager
   from langfuse.llama_index import LlamaIndexCallbackHandler
   
   langfuse_callback_handler = LlamaIndexCallbackHandler()
   Settings.callback_manager = CallbackManager([langfuse_callback_handler])

4. Flush events to Langfuse:
   The Langfuse SDKs queue and batch events in the background to reduce the number of network requests and improve overall performance. Before exiting your application, make sure all queued events have been flushed to Langfuse servers.

   # ... your LlamaIndex calls here ...
   
   langfuse_callback_handler.flush()

Here is a complete example of how you might integrate Langfuse into your existing code:

import os
from llama_index.core import set_global_handler
from langfuse.llama_index import LlamaIndexCallbackHandler

# Configure environment
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

# Set global handler
set_global_handler("langfuse")
langfuse_callback_handler = LlamaIndexCallbackHandler()

# Your existing function or code
def process_streamlit_fetch_query_results(query_text):
    try:
        # Your existing logic
        reranked_nodes = rag_obj.process_streamlit_fetch_query_results(query=query_text)
        return reranked_nodes
    except Exception as e:
        raise Exception(f"An error occurred retrieving: {e}")

# Example usage
query_text = "Your query here"
result = process_streamlit_fetch_query_results(query_text)
print(result)

# Flush events to Langfuse
langfuse_callback_handler.flush()

Make sure to replace the placeholder values with your actual Langfuse API keys. This setup will enable tracing for your function using Langfuse ^[1][2].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu

I've defined my docstore, vector and index store using redis like:

self.redis_client = redis.from_url(self.config.get('REDIS_VECTOR_STORE'))
                self.docstore = RedisDocumentStore.from_redis_client(
                    redis_client=self.redis_client, namespace=self.namespace
                )
                # TODO: UNcomment to use redis vector store
                self.vector_store = RedisVectorStore(
                   index_name=self.namespace,
                   redis_url=self.config.get('REDIS_VECTOR_STORE'),
                   metadata_fields=['namespace']
                )

Give me two functions that I can add to the class that can do the following:

1. return a list of all documents added
2. Given the document name, return all the chunks created.

[dosubot[bot]]

Here are the two functions you can add to your class to achieve the desired functionality:

1. Return a list of all documents added:
2. Given the document name, return all the chunks created:

class YourClass:
    def get_all_documents(self) -> List[Document]:
        """
        Returns a list of all documents added.
        """
        return list(self.docstore.docs.values())

    def get_chunks_by_document_name(self, document_name: str) -> List[str]:
        """
        Returns all the chunks created given a document name.
        """
        document = next((doc for doc in self.docstore.docs.values() if doc.metadata.get("file_name") == document_name), None)
        if not document:
            return []

        file_path = document.metadata.get("file_path")
        if not file_path:
            return []

        chunk_size = 512  # or any other default chunk size
        documents = SimpleDirectoryReader(input_files=[file_path]).load_data()
        splitter = SemanticSplitterNodeParser(
            buffer_size=1,
            breakpoint_percentile_threshold=95,
            embed_model=OpenAIEmbedding(),
        )
        nodes = splitter.get_nodes_from_documents(documents)
        return [node.get_content() for node in nodes]

These functions will allow you to list all documents and retrieve chunks based on the document name ^[1][2].

To continue talking to Dosu, mention @dosu.