Function get_field_value_from_trace_tree_v2

api/oss/src/utils/traces.py:366–413

Retrieve the value of the key from the trace tree. Args: tree (Dict[str, Any]): The nested dictionary to retrieve the value from. i.e. inline trace e.g. tree["spans"]["rag"]["spans"]["retriever"]["internals"]["prompt"] key (str): The dot-separated ke

(
    tree: Dict[str, Any],
    field: str,
)

def get_field_value_from_trace_tree_v2(
    tree: Dict[str, Any],
    field: str,
):
    """
    Retrieve the value of the key from the trace tree.

    Args:
        tree (Dict[str, Any]): The nested dictionary to retrieve the value from.
            i.e. inline trace
            e.g. tree["spans"]["rag"]["spans"]["retriever"]["internals"]["prompt"]
        key (str): The dot-separated key to access the value.
            e.g. rag.summarizer[0].outputs.report

    Returns:
        Dict[str, Any]: The retrieved value or None if the key does not exist or an error occurs.
    """

    separate_by_spans_key = True

    parts = field.split(".")

    try:
        for part in parts:
            # by default, expects something like 'retriever'
            key, idx = _parse_field_part(part)

            # before 'SPECIAL_KEYS', spans are nested within a 'spans' key
            # e.g. trace["spans"]["rag"]["spans"]["retriever"]...
            if key in SPECIAL_KEYS:
                separate_by_spans_key = False

            # after 'SPECIAL_KEYS', it is a normal dict.
            # e.g. trace[...]["internals"]["prompt"]
            if separate_by_spans_key:
                tree = tree[SPANS_KEY]

            tree = tree[key]

            if idx is not None:
                tree = tree[idx]

        return tree

    # Suppress all Exception and leave Exception management to the caller.
    except Exception as e:
        log.error(f"Error retrieving trace value from key: {traceback.format_exc()}")
        return None


def get_field_value_from_trace_tree_v3(trace_data: Dict[str, any], key: str):