(message: Dict[str, Any], delta: Dict[str, Any])
| 7483 | |
| 7484 | @staticmethod |
| 7485 | def _apply_message_delta(message: Dict[str, Any], delta: Dict[str, Any]) -> None: |
| 7486 | if "role" in delta: |
| 7487 | message["role"] = delta["role"] |
| 7488 | for key, value in delta.items(): |
| 7489 | if key in {"role", "tool_calls", "function_call"}: |
| 7490 | continue |
| 7491 | if isinstance(value, str): |
| 7492 | existing = message.get(key) |
| 7493 | if isinstance(existing, str): |
| 7494 | message[key] = existing + value |
| 7495 | else: |
| 7496 | message[key] = value |
| 7497 | else: |
| 7498 | message[key] = value |
| 7499 | tool_call_deltas = delta.get("tool_calls") |
| 7500 | if not isinstance(tool_call_deltas, list): |
| 7501 | return |
| 7502 | tool_calls = cast(List[Dict[str, Any]], message.setdefault("tool_calls", [])) |
| 7503 | for tool_delta in tool_call_deltas: |
| 7504 | if not isinstance(tool_delta, dict): |
| 7505 | continue |
| 7506 | index = tool_delta.get("index") |
| 7507 | if not isinstance(index, int): |
| 7508 | continue |
| 7509 | while len(tool_calls) <= index: |
| 7510 | tool_calls.append({"function": {"name": "", "arguments": ""}}) |
| 7511 | tool_call = tool_calls[index] |
| 7512 | if "id" in tool_delta: |
| 7513 | tool_call["id"] = tool_delta["id"] |
| 7514 | if "type" in tool_delta: |
| 7515 | tool_call["type"] = tool_delta["type"] |
| 7516 | function_delta = tool_delta.get("function") |
| 7517 | if not isinstance(function_delta, dict): |
| 7518 | continue |
| 7519 | function = cast(Dict[str, Any], tool_call.setdefault("function", {})) |
| 7520 | name_delta = function_delta.get("name") |
| 7521 | if isinstance(name_delta, str): |
| 7522 | function["name"] = cast(str, function.get("name", "")) + name_delta |
| 7523 | arguments_delta = function_delta.get("arguments") |
| 7524 | if isinstance(arguments_delta, str): |
| 7525 | function["arguments"] = cast(str, function.get("arguments", "")) + arguments_delta |
| 7526 | if tool_calls: |
| 7527 | message["function_call"] = dict(cast(Dict[str, Any], tool_calls[0]["function"])) |
| 7528 | |
| 7529 | def parse_completion_message(self, response_text: str) -> Dict[str, Any]: |
| 7530 | parsed = self.parse_chat_response(response_text, partial=False) |
no test coverage detected