(
self,
previous_message: Dict[str, Any],
message: Dict[str, Any],
)
| 7416 | return message |
| 7417 | |
| 7418 | def _message_deltas( |
| 7419 | self, |
| 7420 | previous_message: Dict[str, Any], |
| 7421 | message: Dict[str, Any], |
| 7422 | ) -> List[Dict[str, Any]]: |
| 7423 | deltas: List[Dict[str, Any]] = [] |
| 7424 | for key, value in message.items(): |
| 7425 | if key in {"role", "content", "tool_calls", "function_call"}: |
| 7426 | continue |
| 7427 | old_value = previous_message.get(key, "") |
| 7428 | if not isinstance(value, str): |
| 7429 | if key not in previous_message and value is not None: |
| 7430 | deltas.append({key: value}) |
| 7431 | continue |
| 7432 | if not value: |
| 7433 | continue |
| 7434 | if isinstance(old_value, str) and value.startswith(old_value): |
| 7435 | delta_value = value[len(old_value) :] |
| 7436 | else: |
| 7437 | delta_value = value |
| 7438 | if delta_value: |
| 7439 | deltas.append({key: delta_value}) |
| 7440 | |
| 7441 | new_content = message.get("content") |
| 7442 | old_content = previous_message.get("content", "") |
| 7443 | if isinstance(new_content, str) and new_content: |
| 7444 | if isinstance(old_content, str) and new_content.startswith(old_content): |
| 7445 | content_delta = new_content[len(old_content) :] |
| 7446 | else: |
| 7447 | content_delta = new_content |
| 7448 | if content_delta: |
| 7449 | deltas.append({"content": content_delta}) |
| 7450 | |
| 7451 | new_tool_calls = cast(List[Dict[str, Any]], message.get("tool_calls", [])) |
| 7452 | old_tool_calls = cast(List[Dict[str, Any]], previous_message.get("tool_calls", [])) |
| 7453 | for tool_call_index, tool_call in enumerate(new_tool_calls): |
| 7454 | old_tool_call = old_tool_calls[tool_call_index] if tool_call_index < len(old_tool_calls) else None |
| 7455 | delta_tool_call: Dict[str, Any] = {"index": tool_call_index} |
| 7456 | if old_tool_call is None: |
| 7457 | delta_tool_call["id"] = tool_call["id"] |
| 7458 | delta_tool_call["type"] = tool_call["type"] |
| 7459 | function = cast(Dict[str, Any], tool_call["function"]) |
| 7460 | old_function = ( |
| 7461 | cast(Dict[str, Any], old_tool_call["function"]) |
| 7462 | if old_tool_call is not None |
| 7463 | else {} |
| 7464 | ) |
| 7465 | function_delta: Dict[str, Any] = {} |
| 7466 | if function.get("name") and function.get("name") != old_function.get("name"): |
| 7467 | function_delta["name"] = function["name"] |
| 7468 | arguments = cast(str, function.get("arguments", "")) |
| 7469 | old_arguments = cast(str, old_function.get("arguments", "")) |
| 7470 | if old_tool_call is None and arguments == "{}": |
| 7471 | argument_delta = "" |
| 7472 | elif arguments.startswith(old_arguments): |
| 7473 | argument_delta = arguments[len(old_arguments) :] |
| 7474 | else: |
| 7475 | argument_delta = arguments |
no test coverage detected