Recursively truncate documents as needed to fit inside max_length characters.
(obj: Any, max_length: int)
| 1112 | |
| 1113 | |
| 1114 | def _truncate_documents(obj: Any, max_length: int) -> Tuple[Any, int]: |
| 1115 | """Recursively truncate documents as needed to fit inside max_length characters.""" |
| 1116 | if max_length <= 0: |
| 1117 | return None, 0 |
| 1118 | remaining = max_length |
| 1119 | if hasattr(obj, "items"): |
| 1120 | truncated: Any = {} |
| 1121 | for k, v in obj.items(): |
| 1122 | truncated_v, remaining = _truncate_documents(v, remaining) |
| 1123 | if truncated_v: |
| 1124 | truncated[k] = truncated_v |
| 1125 | if remaining <= 0: |
| 1126 | break |
| 1127 | return truncated, remaining |
| 1128 | elif hasattr(obj, "__iter__") and not isinstance(obj, (str, bytes)): |
| 1129 | truncated: Any = [] # type:ignore[no-redef] |
| 1130 | for v in obj: |
| 1131 | truncated_v, remaining = _truncate_documents(v, remaining) |
| 1132 | if truncated_v: |
| 1133 | truncated.append(truncated_v) |
| 1134 | if remaining <= 0: |
| 1135 | break |
| 1136 | return truncated, remaining |
| 1137 | else: |
| 1138 | return _truncate(obj, remaining) |
| 1139 | |
| 1140 | |
| 1141 | def _truncate(obj: Any, remaining: int) -> Tuple[Any, int]: |