src/backend/base/langflow/services/tracing/utils.py

from typing import Any

from langflow.schema.data import Data


def convert_to_langchain_type(value):
    from langflow.schema.message import Message

    if isinstance(value, dict):
        value = {key: convert_to_langchain_type(val) for key, val in value.items()}
    elif isinstance(value, list):
        value = [convert_to_langchain_type(v) for v in value]
    elif isinstance(value, Message):
        if "prompt" in value:
            value = value.load_lc_prompt()
        elif value.sender:
            value = value.to_lc_message()
        else:
            value = value.to_lc_document()
    elif isinstance(value, Data):
        value = value.to_lc_document() if "text" in value.data else value.data
    return value


def convert_to_langchain_types(io_dict: dict[str, Any]):
    converted = {}
    for key, value in io_dict.items():
        converted[key] = convert_to_langchain_type(value)
    return converted