langfuse.api.resources.commons.errors.not_found_error.NotFoundError: status_code: 404, body: {'message': 'Observation not found'}

[emailic]

I'm trying to run evaluation of my RAG system, but I'm struggling with this Error that I don't fully understand.

import argparse
import os
import sys

sys.path.append(os.path.abspath(os.path.join("cloud-functions", "python")))

from ...langchain_chat_model_factory import LangchainChatModelFactory
from ...llm_chat_service import LLMChatService
from ...pinecone_service import PineconeConfig, PineconeService
from ...firestore_chat_repository import FirestoreChatRepository
from langchain.chat_models.openai import ChatOpenAI
from langchain.smith import RunEvalConfig
from langsmith import Client
from langfuse import Langfuse
import random
from ...model_config import ModelConfig, ModelParams, Platform

from utils import (
    CallbackHandlerMock,
    get_feature_flag_service,
    get_rag_result,
    get_retrieval_filter,
    get_num_retrieval_results,
)


def get_arguments():
    parser = argparse.ArgumentParser(description="Parse the dataset path.")
    parser.add_argument(
        "-d",
        "--dataset",
        default="noa_rag_dataset",
        help="Name of the the dataset.",
        choices=["mini_esg_bench", "mini_squadv2", "noa_rag_dataset"],
    )
    parser.add_argument(
        "--is_organisation_system_prompt_enabled",
        type=bool,
        default=False,
        help="Enables or disables organisation system prompt.",
    )
  
    return parser.parse_args()


def main():
    args = get_arguments()
    feature_flag_service = get_feature_flag_service(
        is_organisation_system_prompt_enabled=args.is_organisation_system_prompt_enabled,
    )
    model_name = (
        "gpt-4"
    )
    langfuse = Langfuse()


    callbacks = CallbackHandlerMock(
        trace_name=args.dataset + "_turbo",
        user_id=os.environ["USER"],
        session_id=langfuse.__repr__(),
    )
    chat_model_factory = LangchainChatModelFactory(
        chat_repository=FirestoreChatRepository(
            idempotency_key="some-idemp-key",
            document_path="documents/some-document-path",
        ),
        vector_db=PineconeService(
            config=PineconeConfig(
                environment=os.environ["PINECONE_ENV"],
                index_name=os.environ["PINECONE_INDEX_NAME"],
            )
        ),
        feature_flag_service=feature_flag_service,
        get_callbacks=callbacks.get_callbacks,
    )
    llm_chat_service = LLMChatService(chat_model_factory)
    num_retrieval_results = get_num_retrieval_results(
        feature_flag_service.is_cohere_rerank_enabled()
    )

 
    langfuse = Langfuse()
    retrieval_filter = get_retrieval_filter(args.dataset)

    def get_result (input):
        return get_rag_result(
            question=input,
            retrieval_filter=retrieval_filter,
            llm_chat_service=llm_chat_service,
            num_retrieval_results=num_retrieval_results,
            trace_name=args.dataset,
            user_id=os.environ["USER"],
            session_id=langfuse.__repr__(),
            model_config=ModelConfig(
                platform=Platform.OPENAI,
                model_name=model_name,
            ),
        )

    dataset = langfuse.get_dataset(args.dataset)
    def my_eval_fn(input, output, expected_output):
        return random.random()


    for item in dataset.items:
        # execute application function and get Langfuse parent observation (span/generation/event)
        # output also returned as it is used to evaluate the run
        generation = get_result(item.input)
        output = item.expected_output
    
        # link the execution trace to the dataset item and give it a run_name
        item.link(generation, "test run") #f"{args.dataset}_test_run"
    
        # optionally, evaluate the output to compare different runs more easily
        generation.score(
        name="<example_eval_test>",
        # any float value
        value=my_eval_fn(
            item.input,
            output,
            item.expected_output
        )
    )
        


if __name__ == "__main__":
    main()

Just for the reference, this is how the callback handler looks like:

class CallbackHandlerMock:
    def __init__(self, trace_name: str, user_id: str, session_id: str):
        self.trace_name = trace_name
        self.user_id = user_id
        self.session_id = session_id

    def get_callbacks(
        self,
        model_config: ModelConfig,
        callback_handler: CallbackHandler = None,
    ) -> list[BaseCallbackHandler]:
        return [
            LangfuseCallbackHandler(
                public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
                secret_key=os.environ["LANGFUSE_SECRET_KEY"],
                host=os.environ["LANGFUSE_HOST"],
                trace_name=self.trace_name,
                user_id=self.user_id,
                model_config=model_config,
                session_id=self.session_id,
            )
        ]

The error I'm getting when running this script is the following:

Traceback (most recent call last):
  File "...qa_evaluation.py", line 190, in <module>
    main()
  File ".../qa_evaluation.py", line 174, in main
    item.link(generation, "test run") #f"{args.dataset}_test_run"
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/email/Library/Caches/pypoetry/virtualenvs/noa-77-C5xh0-py3.11/lib/python3.11/site-packages/langfuse/client.py", line 2258, in link
    self.langfuse.client.dataset_run_items.create(
  File "/Users/email/Library/Caches/pypoetry/virtualenvs/noa-77-C5xh0-py3.11/lib/python3.11/site-packages/langfuse/api/resources/dataset_run_items/client.py", line 61, in create
    raise NotFoundError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
langfuse.api.resources.commons.errors.not_found_error.NotFoundError: status_code: 404, body: {'message': 'Observation not found'}

In Traces and Trace detail I can see correct output from my RAG system which contains two generations.
However, in the UI > Datasets > noa_rag_dataset I can se no runs for the dataset I'm using. Any idea why are there no runs being logged, and why am I seeing this error? Thank you!

[marcklingen]

What does get_result() do here? This seems to be the most interesting part to debug this

[emailic]

Hi Marc, thank you for getting back to me.

get_result calls our get_rag_result function. In this issue, for the sake of reproducing the error, i created a generic function, and the error still persists.

[marcklingen]

does it return the langfuse generation?

[emailic]

It was returning just the result of the rag function, that was the issue. I later understood it needs to return a generaiton or a span. Now that it's returning a span, the problem is solved. thank you!