not able to track entire request end to end in langchain + langfuse

[jaiswalvineet]

I am using langchain with langfuse and its working fine for individual LLM call tracking but my end goal is to track end to end request in one trace, and I am not able to find my method to do that activity

here are sample code

```
 class Demo:
def create_handler(self, user_identifier, trace_name):
    langfuse_handler = CallbackHandler(
        session_id=user_identifier,
        metadata={"a": "b"},
        trace_name=trace_name,
    )
    return langfuse_handler

def method1(self, question, llm, prompt, user_identifier):
    langfuse_handler = self.create_handler(user_identifier, "topic_classification")
    retrieval_chain = (
        {"question": RunnablePassthrough()} | prompt | llm
    )
    output_data = retrieval_chain.invoke(
        {"question": question},
        config={
            "callbacks": [langfuse_handler],
        },
    )
    return output_data  # Simulating classification output like "sports", "politics", etc.

def method2(self, question, llm, prompt, user_identifier):
    langfuse_handler = self.create_handler(user_identifier, "RAG_sports")
    retrieval_chain = (
        {"question": RunnablePassthrough()} | prompt | llm
    )
    output_data = retrieval_chain.invoke(
        {"question": question},
        config={
            "callbacks": [langfuse_handler],
        },
    )
    return output_data  # Return RAG-based response for "sports"

def method3(self, question, llm, prompt, user_identifier):
    langfuse_handler = self.create_handler(user_identifier, "RAG_other")
    retrieval_chain = (
        {"question": RunnablePassthrough()} | prompt | llm
    )
    output_data = retrieval_chain.invoke(
        {"question": question},
        config={
            "callbacks": [langfuse_handler],
        },
    )
    return output_data  # Return RAG-based response for non-sports


   class RequestInput(BaseModel):
   question: str
   user_identifier: str
   demo_instance = Demo()

   llm = "Llama"  
    prompt_classification = "classification_prompt"  
    prompt_rag_sports = "sport_promot"  
    prompt_rag_other = "rag_prompt" 


   @app.post("/get_response/")
  async def get_response(request: RequestInput):

class_identification = demo_instance.method1(
    question=request.question,
    llm=llm,
    prompt=prompt_classification,
    user_identifier=request.user_identifier
)
if class_identification == "sports":
    final_output = demo_instance.method2(
        question=request.question,
        llm=llm,
        prompt=prompt_rag_sports,
        user_identifier=request.user_identifier
    )
else:
    final_output = demo_instance.method3(
        question=request.question,
        llm=llm,
        prompt=prompt_rag_other,
        user_identifier=request.user_identifier
    )
return {"final_output": final_output} ```

here for each request, it first goes to classification and suppose it classify as sports then go to method2 else method3, now I want to track entire user journey in trace, ideally langfuse should support it as I am using same handler but its creating two different rows as trace, If I track in last state then also it misses the first classification.

Can someone please let me know what I am missing ?

PS: I am using langgraph for classification + RAG switching, for sake of simplicity, I am not added in this code

[dosubot[bot]]

To track the entire user journey in one trace using Langchain and Langfuse, you need to ensure that all stages of the request are grouped into a single trace. Here is how you can modify your code to achieve this:

1. Create a single CallbackHandler instance for the entire request:

   • Instead of creating a new CallbackHandler in each method, create it once and pass it to all methods.

2. Update your methods to accept the CallbackHandler as a parameter:

   • This ensures that the same handler is used throughout the entire request.

Here is the updated code:

class Demo:
    def create_handler(self, user_identifier, trace_name):
        langfuse_handler = CallbackHandler(
            session_id=user_identifier,
            metadata={"a": "b"},
            trace_name=trace_name,
        )
        return langfuse_handler

    def method1(self, question, llm, prompt, langfuse_handler):
        retrieval_chain = (
            {"question": RunnablePassthrough()} | prompt | llm
        )
        output_data = retrieval_chain.invoke(
            {"question": question},
            config={
                "callbacks": [langfuse_handler],
            },
        )
        return output_data  # Simulating classification output like "sports", "politics", etc.

    def method2(self, question, llm, prompt, langfuse_handler):
        retrieval_chain = (
            {"question": RunnablePassthrough()} | prompt | llm
        )
        output_data = retrieval_chain.invoke(
            {"question": question},
            config={
                "callbacks": [langfuse_handler],
            },
        )
        return output_data  # Return RAG-based response for "sports"

    def method3(self, question, llm, prompt, langfuse_handler):
        retrieval_chain = (
            {"question": RunnablePassthrough()} | prompt | llm
        )
        output_data = retrieval_chain.invoke(
            {"question": question},
            config={
                "callbacks": [langfuse_handler],
            },
        )
        return output_data  # Return RAG-based response for non-sports


class RequestInput(BaseModel):
    question: str
    user_identifier: str

demo_instance = Demo()

llm = "Llama"  
prompt_classification = "classification_prompt"  
prompt_rag_sports = "sport_promot"  
prompt_rag_other = "rag_prompt" 


@app.post("/get_response/")
async def get_response(request: RequestInput):
    langfuse_handler = demo_instance.create_handler(request.user_identifier, "end_to_end_trace")

    class_identification = demo_instance.method1(
        question=request.question,
        llm=llm,
        prompt=prompt_classification,
        langfuse_handler=langfuse_handler
    )
    if class_identification == "sports":
        final_output = demo_instance.method2(
            question=request.question,
            llm=llm,
            prompt=prompt_rag_sports,
            langfuse_handler=langfuse_handler
        )
    else:
        final_output = demo_instance.method3(
            question=request.question,
            llm=llm,
            prompt=prompt_rag_other,
            langfuse_handler=langfuse_handler
        )
    return {"final_output": final_output}

By creating a single CallbackHandler instance and passing it through all methods, you ensure that the entire user journey is tracked under one trace. This should resolve the issue of having separate traces for different stages of the request ^[1].

To continue talking to Dosu, mention @dosu.

[jaiswalvineet]

the single instance handler is also creating two traces, so it will not work

[jannikmaierhoefer]

Hi @jaiswalvineet, have a look at the Langfuse python @observe() decorator. The decorator will automatically create a trace for the top-level function and spans for any nested functions (example here).

[marcklingen]

+1, it is fully interoperable with nested Langchain executions as well: https://langfuse.com/docs/sdk/python/decorators#langchain

[jaiswalvineet]

thanks both, but its not working in my case, I have LLaMA hosted somewhere on Triton and I created a wrapper class which actually inherit and implemented LLM from from "langchain.llms.base import LLM", here the pseudo code

class CustomLLM(LLM):
def __init__(self):
    self.model_name = "llama3"
@property
def _llm_type(self) -> str:
    return "custom_llm"

def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
    response = f"Response from {self.model_name}: You said '{prompt}'"
    return response`

here the reproducible code where I applied observe

def classification_agent(state):
    print(f"inside classification_agent")
    messages = state.get("messages")
    question = messages[-1]
    print(f"inside classification_agent {question}")
    llm = CustomLLM()
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "you are an AI bot who can classify"),
            (
                "user",
                "classify input either in sports_agent if related to sports or non_sports_agent if not related to sports category, input: {question}",
            ),
        ]
    )
    retrieval_chain = {"question": RunnablePassthrough()} | prompt | llm
    output_data = retrieval_chain.invoke({"question": question})
    print(f"classification_agent output: {output_data}")
    return {"messages": [output_data]}
def sports_agent(state):
    messages = state.get("messages")
    question = messages[0]
    llm = CustomLLM()
    print(f"inside sports_agent {question}")
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "you are an AI bot who can generate more details"),
            ("user", "give me more detail about given sports  input: {question}"),
        ]
    )
    retrieval_chain = {"question": RunnablePassthrough()} | prompt | llm
    output_data = retrieval_chain.invoke({"question": question})
    print(f"sports_agent output {output_data}")
    return {"messages": [output_data]}
def non_sports_agent(state):
    messages = state.get("messages")
    question = messages[0]
    llm = CustomLLM()
    print(f"inside non_sports_agent {non_sports_agent}")
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "you are an AI bot who can generate more details"),
            ("user", "give me more detail about given input: {question}"),
        ]
    )
    retrieval_chain = {"question": RunnablePassthrough()} | prompt | llm
    output_data = retrieval_chain.invoke({"question": question})
    print(f"non_sports_agent output {output_data}")
    return {"messages": [output_data]}
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
def _router(state):
    messages = state["messages"]
    last_message = messages[-1]
    return last_message
@observe()
def execture_graph(input):
    inputs = {"messages": [input]}
    graph = StateGraph(AgentState)
    graph.add_node("classification_agent", classification_agent)
    graph.add_node("sports_agent", sports_agent)
    graph.add_node("non_sports_agent", non_sports_agent)
    graph.set_entry_point("classification_agent")
    graph.add_conditional_edges(
        "classification_agent",
        _router,
        {"sports_agent": "sports_agent", "non_sports_agent": "non_sports_agent"},
    )
    graph.add_edge("sports_agent", END)
    graph.add_edge("non_sports_agent", END)
    app = graph.compile()
    output_data = app.invoke(inputs)
    final_success_message = output_data.get("messages")[-1]
    return final_success_message
output = execture_graph("football")
print(f"final outout ==> \n\n  {output}")

its not giving proper detail end to end, attached screenshot also not able to get end to end via CallbackHandler too

Can you please help us on urgent basis, it will decide our go/no-go decision with langfuse.

Thanks in advance

[marcklingen]

You need to add the Langchain Callback Handler to the invoke method in order for this to work. The code snippet that you shared does not include it all but only the decorator. Please have a look at this section in the docs again to understand how to get a langchain handler within a decorated method: https://langfuse.com/docs/sdk/python/decorators#langchain