Skip to content

datarobot_genai.core.router

router

merge_streaming_tool_calls 

merge_streaming_tool_calls(tool_calls_seen: list[Any]) -> list[dict]

Merge streaming tool-call delta objects into complete tool-call dicts.

Each element of tool_calls_seen must have .index, .id, and .function attributes matching the litellm/OpenAI streaming schema. Returns a list of OpenAI-format tool-call dicts ready to forward to the framework or serialize as JSON.

Source code in datarobot_genai/core/router.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def merge_streaming_tool_calls(tool_calls_seen: list[Any]) -> list[dict]:
    """Merge streaming tool-call delta objects into complete tool-call dicts.

    Each element of *tool_calls_seen* must have ``.index``, ``.id``,
    and ``.function`` attributes matching the litellm/OpenAI streaming schema.
    Returns a list of OpenAI-format tool-call dicts ready to forward to the
    framework or serialize as JSON.
    """
    merged: dict[int, dict] = {}
    for tc in tool_calls_seen:
        idx = tc.index
        if idx not in merged:
            merged[idx] = {"id": "", "name": "", "arguments": ""}
        if tc.id:
            merged[idx]["id"] = tc.id
        if tc.function and tc.function.name:
            merged[idx]["name"] = tc.function.name
        if tc.function and tc.function.arguments is not None:
            merged[idx]["arguments"] += tc.function.arguments
    return [
        {
            "id": v["id"],
            "type": "function",
            "function": {"name": v["name"], "arguments": v["arguments"]},
        }
        for v in merged.values()
    ]

build_litellm_router 

build_litellm_router(primary: LLMConfig, fallbacks: list[LLMConfig], router_settings: dict | None = None) -> litellm.Router

Build a litellm.Router with automatic failover.

Parameters:

Name Type Description Default
primary LLMConfig

LLMConfig for the primary model.

 required
fallbacks list[LLMConfig]

Ordered list of LLMConfig fallback models.

 required
router_settings dict | None

Extra keyword arguments forwarded to litellm.Router (e.g. num_retries).

 None
Returns
A configured ``litellm.Router`` that tries ``primary`` first and
cascades through ``fallback_0``, ``fallback_1``, … on failure.
Source code in datarobot_genai/core/router.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def build_litellm_router(
    primary: LLMConfig,
    fallbacks: list[LLMConfig],
    router_settings: dict | None = None,
) -> litellm.Router:
    """Build a ``litellm.Router`` with automatic failover.

    Args:
        primary: ``LLMConfig`` for the primary model.
        fallbacks: Ordered list of ``LLMConfig`` fallback models.
        router_settings: Extra keyword arguments forwarded to ``litellm.Router``
            (e.g. ``num_retries``).

    Returns
    -------
        A configured ``litellm.Router`` that tries ``primary`` first and
        cascades through ``fallback_0``, ``fallback_1``, … on failure.
    """
    import litellm

    model_list = [
        {"model_name": "primary", "litellm_params": primary.to_litellm_params()},
        *[
            {"model_name": f"fallback_{i}", "litellm_params": c.to_litellm_params()}
            for i, c in enumerate(fallbacks)
        ],
    ]
    fallbacks_cfg = [{"primary": [f"fallback_{i}" for i in range(len(fallbacks))]}]
    settings = router_settings or {}
    return litellm.Router(
        model_list=model_list,
        fallbacks=fallbacks_cfg,
        **settings,
    )