How run Langfuse evaluations over specifics spans?

[joaoCrulhas]

Hi, I'm continuing to improve our observability over our prompts, but I'm facing an issue that I wasn't able to find in Langfuse docs.
After executing the request to OpenAI, I execute two RunnableLambdas

This can be seen in the image above.

I'd like to run the evaluations in each RunnableLambda, but I'm having trouble determining which is the correct method.

For each Lambda, I'm executing a small task for which I'd like to evaluate the inputs and the outputs.

I tried some approaches, the config below:

Unfortunately, the evaluation configuration does not run in my trace.
Is there some way to do it?

Another way to set the lambdas is by using Langchain.
Thanks in advance.

Best,
João

[dosubot[bot]]

To run Langfuse evaluations over specific spans in each RunnableLambda to evaluate the inputs and outputs, you can follow these steps:

1. Initialize the Langfuse client SDK:

   from langfuse import Langfuse
   
   langfuse = Langfuse()

2. Define a utility function to score your trace with the metrics you chose:

   async def score_with_ragas(query, chunks, answer):
       scores = {}
       for m in metrics:
           print(f"calculating {m.name}")
           scores[m.name] = await m.ascore(
               row={"question": query, "contexts": chunks, "answer": answer}
           )
       return scores

3. Compute the score with each request:

   # Start a new trace when you get a question
   question = row['question']
   trace = langfuse.trace(name="rag")
   
   # Retrieve the relevant chunks
   contexts = row['contexts']
   trace.span(
       name="retrieval", input={'question': question}, output={'contexts': contexts}
   )
   
   # Use LLM to generate an answer with the chunks
   answer = row['answer']
   trace.span(
       name="generation", input={'question': question, 'contexts': contexts}, output={'answer': answer}
   )
   
   # Compute scores for the question, context, answer tuple
   ragas_scores = await score_with_ragas(question, contexts, answer)

4. Add the computed scores to the trace in Langfuse:

   # Send the scores
   for m in metrics:
       trace.score(name=m.name, value=ragas_scores[m.name])

This setup ensures that each span within a RunnableLambda is evaluated for its inputs and outputs, and the scores are logged in Langfuse.

For a more detailed setup, including initializing metrics with LLMs and embeddings, and handling batch scoring, refer to the Langfuse documentation and the Ragas documentation ^{[1][2][3][4][5]}.

To continue talking to Dosu, mention @dosu.

[joaoCrulhas]

Can you provide this using typescript ?

[dosubot[bot]]

To run Langfuse evaluations over specific spans in each RunnableLambda using TypeScript, you can follow these steps:

1. Initialize the Langfuse client SDK:

   import { Langfuse, observe, langfuseContext } from 'langfuse-sdk';
   
   const langfuseClient = new Langfuse({
     publicKey: process.env.LANGFUSE_PUBLIC_KEY,
     secretKey: process.env.LANGFUSE_SECRET_KEY,
     host: process.env.LANGFUSE_HOST || 'https://cloud.langfuse.com',
   });

2. Define a utility function to score your trace with the metrics you chose:

   async function scoreWithRagas(query: string, chunks: string[], answer: string) {
     const scores: Record<string, number> = {};
     for (const metric of metrics) {
       console.log(`calculating ${metric.name}`);
       scores[metric.name] = await metric.ascore({
         question: query,
         contexts: chunks,
         answer: answer,
       });
     }
     return scores;
   }

3. Compute the score with each request:

   const mainFn = observe(async (row: { question: string, contexts: string[], answer: string }) => {
     const { question, contexts, answer } = row;
   
     // Start a new trace when you get a question
     const trace = langfuseClient.trace({ name: "rag" });
   
     // Retrieve the relevant chunks
     trace.span({
       name: "retrieval",
       input: { question },
       output: { contexts },
     });
   
     // Use LLM to generate an answer with the chunks
     trace.span({
       name: "generation",
       input: { question, contexts },
       output: { answer },
     });
   
     // Compute scores for the question, context, answer tuple
     const ragasScores = await scoreWithRagas(question, contexts, answer);
   
     // Send the scores
     for (const metric of metrics) {
       trace.score({
         name: metric.name,
         value: ragasScores[metric.name],
       });
     }
   });

4. Execute the main function to generate a trace and score it:

   const row = {
     question: "What is the capital of France?",
     contexts: ["Paris is the capital of France."],
     answer: "The capital of France is Paris."
   };
   
   mainFn(row).then(() => {
     const traceId = langfuseContext.getCurrentTraceId();
     const observationId = langfuseContext.getCurrentObservationId();
     console.log(`Trace ID: ${traceId}, Observation ID: ${observationId}`);
   });

This setup ensures that each span within a RunnableLambda is evaluated for its inputs and outputs, and the scores are logged in Langfuse. For more detailed setup, including initializing metrics with LLMs and embeddings, refer to the Langfuse documentation and the Ragas documentation.

To continue talking to Dosu, mention @dosu.

[joaoCrulhas]

The way that I'm working is a little bit different; I'm attaching here the code

const model = new ChatOpenAI({
   modelName: prompt.config.modelName,
   temperature: prompt.config.temperature,
 });

 const promptTemplate = await PromptTemplate.fromTemplate(prompt.getLangchainPrompt()).partial({
   schema: parsers.getFormatInstructions(),
 });

 const extractJsonLambda = new RunnableLambda({
   func: extractJson<ConsumptionRecord>,
 }).withConfig({
   metadata: {
     traceId: trace.id,
     name: 'extractJsonLambda',
   },
 });

 const postProcessing = new RunnableLambda({
   func: postProcessing,
 }).withConfig({
   metadata: {
     traceId: trace.id,
     name: 'postProcessing',
   },
 });

 const chain = RunnableSequence.from(
   [promptTemplate, model, extractJsonLambda, postProcessing],
   'myChain',
 );


function postProcessResult(result: ConsumptionRecord): ConsumptionDraftType {
return object;
}

const extractJson = <T = any>(output: AIMessage): Array<T> => {
const text = output.content as string;
// Define the regular expression pattern to match JSON blocks
const pattern = /```json(.*?)```/gs;

// Find all non-overlapping matches of the pattern in the string
const matches = text.match(pattern);

// Process each match, attempting to parse it as JSON
try {
 return (
   matches?.map((match) => {
     // Remove the markdown code block syntax to isolate the JSON string
     const jsonStr = match.replace(/```json|```/g, '').trim();
     return JSON.parse(jsonStr);
   }) ?? []
 );
} catch (error) {
 throw new Error(`Failed to parse: ${output}`);
}
};

The lambdas are being executed in a RunnableSequence, using the RunnableLambda from langchain.

I want to know how can I run evaluations using langfuse over the RunnableLambda mentioned above.

[marcklingen]

Hi @joaoCrulhas, I'd recommend to assign a run_name to the runnable lambdas which will define the name of the resulting span in Langfuse. Thereby you should be able to select which runnable lambda to run your evals on in Langfuse

[joaoCrulhas]

@marcklingen, thank you for your feedback; I missed this property.
Thank you, it works like a charm

[marcklingen]

@marcklingen, thank you for your feedback; I missed this property.

Thank you, it works like a charm

Awesome. Thanks for confirming that this works as expected now.