ccproxy.services.adapters.delta_utils

ccproxy.services.adapters.delta_utils

Delta accumulation utilities following OpenAI SDK patterns.

accumulate_delta

accumulate_delta(accumulated, delta)

Recursively merge delta into accumulated following OpenAI's rules.

This function implements the same accumulation logic as OpenAI's SDK: - Concatenate strings - Add numbers (int/float) - Recursively merge dictionaries - Extend primitive lists - Merge object lists by 'index' key - Preserve 'index' and 'type' keys without modification

Parameters:

Name	Type	Description	Default
accumulated	dict[str, Any]	The accumulated state to merge into	required
delta	dict[str, Any]	The delta to merge	required

Returns:

Type	Description
dict[str, Any]	The merged result (may modify accumulated in-place)

Raises:

Type	Description
TypeError	For unsupported data types
ValueError	For invalid list structures

Source code in ccproxy/services/adapters/delta_utils.py

def accumulate_delta(
    accumulated: dict[str, Any], delta: dict[str, Any]
) -> dict[str, Any]:
    """Recursively merge delta into accumulated following OpenAI's rules.

    This function implements the same accumulation logic as OpenAI's SDK:
    - Concatenate strings
    - Add numbers (int/float)
    - Recursively merge dictionaries
    - Extend primitive lists
    - Merge object lists by 'index' key
    - Preserve 'index' and 'type' keys without modification

    Args:
        accumulated: The accumulated state to merge into
        delta: The delta to merge

    Returns:
        The merged result (may modify accumulated in-place)

    Raises:
        TypeError: For unsupported data types
        ValueError: For invalid list structures
    """
    # Handle None/empty cases
    if not delta:
        return accumulated
    if not accumulated:
        return dict(delta)

    # Work on a copy to avoid mutating input
    result = dict(accumulated)

    for key, delta_value in delta.items():
        if key not in result:
            # New key, just set it
            result[key] = delta_value
            continue

        current_value = result[key]

        # Handle different data type combinations
        if isinstance(current_value, str) and isinstance(delta_value, str):
            # Concatenate strings
            result[key] = current_value + delta_value

        elif isinstance(current_value, int | float) and isinstance(
            delta_value, int | float
        ):
            # Add numbers
            result[key] = current_value + delta_value

        elif isinstance(current_value, dict) and isinstance(delta_value, dict):
            # Recursively merge dictionaries
            result[key] = accumulate_delta(current_value, delta_value)

        elif isinstance(current_value, list) and isinstance(delta_value, list):
            # Handle list merging
            result[key] = _accumulate_list(current_value, delta_value)

        else:
            # For any other case, delta value overwrites
            result[key] = delta_value

    return result