How to pass video file to evaluator function?

[MichaelMMeskhi]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

def call_llm(model: Any, text: str, video_uri: str, sys_prompt: str) -> Any:
    """
    Call the LLM model with the given text and system prompt.

    Args:
        model (Any): LLM model.
        text (str): Input text.
        sys_prompt (str): System prompt.

    Returns:
        Any: Result of the LLM model invocation.
    """
    parser = StrOutputParser()
    media_message = {
        "type": "image_url",
        "image_url": {
            "url": video_uri,
        },
    }
    prompt_template = ChatPromptTemplate.from_messages(
        [
            ("system", sys_prompt),
            ("human", [media_message, "{input_text}"]),
        ]
    )
    chain = prompt_template | model | parser
    result = chain.invoke({"input_text": text})
    return result

helpfull_criteria_evaluator = LangChainStringEvaluator(
    "labeled_criteria",
    config={
        "criteria": "helpfulness", # irrelevant function name
        "llm": gemini_llm,
    },
    prepare_data=lambda run, example: {
        "prediction": run.outputs["output"],
        "reference": example.outputs["label"],
        "input": example.inputs["text"],
    },
)

Description

I am using a custom criteria evaluator to evaluate an answer of Gemini with respect to input video file. How can I pass that same input + video to the evaluator which is also Gemini in this case.

This is how I call my base LLM that takes in a question + video:

def call_llm(model: Any, text: str, video_uri: str, sys_prompt: str) -> Any:
    """
    Call the LLM model with the given text and system prompt.

    Args:
        model (Any): LLM model.
        text (str): Input text.
        sys_prompt (str): System prompt.

    Returns:
        Any: Result of the LLM model invocation.
    """
    parser = StrOutputParser()
    media_message = {
        "type": "image_url",
        "image_url": {
            "url": video_uri,
        },
    }
    prompt_template = ChatPromptTemplate.from_messages(
        [
            ("system", sys_prompt),
            ("human", [media_message, "{input_text}"]),
        ]
    )
    chain = prompt_template | model | parser
    result = chain.invoke({"input_text": text})
    return result

And lets assume this is my criteria function:

criteria_evaluator = LangChainStringEvaluator(
    "criteria",
    config={
        "criteria": {
            "short_video": "Is the video shorter than 10 seconds?"
        },
        "llm": gemini_llm,
        prepare_data=lambda run, example: {
        "prediction": run.outputs["output"],
        "reference": example.outputs["label"],
        "input": example.inputs["text"],
    },
    },
)

My dataset csv file is simple 2 columns:

question, video_url

I am having trouble understanding how to get this working.

System Info

langchain==0.2.6
langchain-chroma==0.1.2
langchain-community==0.2.1
langchain-core==0.2.23
langchain-google-genai==1.0.7
langchain-google-vertexai==1.0.7
langchain-openai==0.1.13
langchain-text-splitters==0.2.0
langchainhub==0.1.20
openinference-instrumentation-langchain==0.1.22

[dosubot[bot]]

Hello, @MichaelMMeskhi! I'm here to help you with your LangChain issue. If you need assistance, feel free to ask.

To pass a video file to an evaluator function in LangChain, you can use the process method of the ImageProcessor class to handle the video file. Additionally, you need to modify the prepare_data function in the LangChainStringEvaluator to include video file inputs along with text inputs.

Here's how you can adapt your code:

1. Process the Video File:
   Use the ImageProcessor class to process the video file and extract notable frames.

2. Modify the prepare_data Function:
   Update the prepare_data function to handle video files and integrate them into the existing input handling mechanism.

Here is an example of how you can achieve this:

Step 1: Process the Video File

from langchain_experimental.video_captioning.services.image_service import ImageProcessor

# Initialize the ImageProcessor
image_processor = ImageProcessor()

# Path to the video file
video_file_path = "path/to/your/video.mp4"

# Process the video file
video_models = image_processor.process(video_file_path)

# video_models now contains the processed video data

Step 2: Modify the prepare_data Function

import os
from langchain_experimental.video_captioning.base import VideoCaptioningChain
from langchain_core.llms import BaseLanguageModel

def prepare_data(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """
    Prepare data for evaluation, including handling text and video file inputs.

    Args:
        inputs: The input dictionary containing text and/or video file paths.

    Returns:
        A dictionary with prepared inputs.
    """
    prepared_inputs = {}

    # Handle text inputs
    if 'text' in inputs:
        prepared_inputs['text'] = inputs['text']
    
    # Handle video file inputs
    if 'video_files' in inputs:
        video_files = inputs['video_files']
        if not isinstance(video_files, list):
            raise InputFormatError("Expected list of video file paths for 'video_files'.")
        
        prepared_videos = []
        for video_file in video_files:
            if not os.path.isfile(video_file):
                raise InputFormatError(f"Video file not found: {video_file}")
            # Process the video file using VideoCaptioningChain
            video_captioning_chain = VideoCaptioningChain(
                llm=BaseLanguageModel(),  # Replace with actual language model instance
                assemblyai_key="your_assemblyai_key"  # Replace with actual API key
            )
            srt_content = video_captioning_chain._call({"video_file_path": video_file})
            prepared_videos.append(srt_content)
        
        prepared_inputs['video_files'] = prepared_videos

    return prepared_inputs

Step 3: Update the Evaluator

criteria_evaluator = LangChainStringEvaluator(
    "criteria",
    config={
        "criteria": {
            "short_video": "Is the video shorter than 10 seconds?"
        },
        "llm": gemini_llm,
        "prepare_data": prepare_data,
    },
)

Example Usage

def call_llm(model: Any, text: str, video_uri: str, sys_prompt: str) -> Any:
    """
    Call the LLM model with the given text and system prompt.

    Args:
        model (Any): LLM model.
        text (str): Input text.
        sys_prompt (str): System prompt.

    Returns:
        Any: Result of the LLM model invocation.
    """
    parser = StrOutputParser()
    media_message = {
        "type": "image_url",
        "image_url": {
            "url": video_uri,
        },
    }
    prompt_template = ChatPromptTemplate.from_messages(
        [
            ("system", sys_prompt),
            ("human", [media_message, "{input_text}"]),
        ]
    )
    chain = prompt_template | model | parser
    result = chain.invoke({"input_text": text})
    return result

In this setup, the prepare_data function is capable of handling both text and video file inputs, leveraging the VideoCaptioningChain for video processing ^[1][2].

To continue talking to Dosu, mention @dosu.

[MichaelMMeskhi]

@dosu prepare_data takes in only 3 keys not sure this works!

[MichaelMMeskhi]

@dosu thats no right

[dosubot[bot]]

To extend the LangChainStringEvaluator to handle video files, you need to modify the prepare_data function to accommodate additional keys for media data. Here's how you can achieve this:

1. Extend the LLMStringRunMapper class to handle media files:

class MediaLLMStringRunMapper(LLMStringRunMapper):
    """Extract items to evaluate from the run object, including media files."""

    def serialize_media(self, media: Dict) -> str:
        """Serialize media data."""
        return f"Media: {media['url']}"

    def serialize_inputs(self, inputs: Dict) -> str:
        if "media" in inputs:
            input_ = self.serialize_media(inputs["media"])
        else:
            input_ = super().serialize_inputs(inputs)
        return input_

    def serialize_outputs(self, outputs: Dict) -> str:
        if "media" in outputs:
            output_ = self.serialize_media(outputs["media"])
        else:
            output_ = super().serialize_outputs(outputs)
        return output_

2. Modify the prepare_data function to handle video files:

def prepare_data(run: Run, example: Optional[Example] = None) -> Dict[str, str]:
    """Prepare data for evaluation, including handling media files."""
    run_mapper = MediaLLMStringRunMapper()
    evaluate_strings_inputs = run_mapper(run)
    
    if example:
        example_mapper = StringExampleMapper()
        evaluate_strings_inputs.update(example_mapper(example))
    
    return evaluate_strings_inputs

3. Update the evaluator to use the new prepare_data function:

criteria_evaluator = LangChainStringEvaluator(
    "criteria",
    config={
        "criteria": {
            "short_video": "Is the video shorter than 10 seconds?"
        },
        "llm": gemini_llm,
        "prepare_data": prepare_data,
    },
)

With these modifications, the LangChainStringEvaluator can now handle media files such as videos by serializing the media data and including it in the evaluation process ^[1].

To continue talking to Dosu, mention @dosu.