| 1 | # from . import files |
| 2 | |
| 3 | import json |
| 4 | |
| 5 | |
| 6 | def truncate_text(agent, output, threshold=1000): |
| 7 | threshold = int(threshold) |
| 8 | if not threshold or len(output) <= threshold: |
| 9 | return output |
| 10 | |
| 11 | # Adjust the file path as needed |
| 12 | placeholder = agent.read_prompt( |
| 13 | "fw.msg_truncated.md", length=(len(output) - threshold) |
| 14 | ) |
| 15 | # placeholder = files.read_file("./prompts/default/fw.msg_truncated.md", length=(len(output) - threshold)) |
| 16 | |
| 17 | start_len = (threshold - len(placeholder)) // 2 |
| 18 | end_len = threshold - len(placeholder) - start_len |
| 19 | |
| 20 | truncated_output = output[:start_len] + placeholder + output[-end_len:] |
| 21 | return truncated_output |
| 22 | |
| 23 | |
| 24 | def truncate_dict_by_ratio(agent, data: dict|list|str, threshold_chars: int, truncate_to: int): |
| 25 | threshold_chars = int(threshold_chars) |
| 26 | truncate_to = int(truncate_to) |
| 27 | |
| 28 | def process_item(item): |
| 29 | if isinstance(item, dict): |
| 30 | truncated_dict = {} |
| 31 | cumulative_size = 0 |
| 32 | |
| 33 | for key, value in item.items(): |
| 34 | processed_value = process_item(value) |
| 35 | serialized_value = json.dumps(processed_value, ensure_ascii=False) |
| 36 | size = len(serialized_value) |
| 37 | |
| 38 | if cumulative_size + size > threshold_chars: |
| 39 | truncated_dict[key] = truncate_text( |
| 40 | agent, serialized_value, truncate_to |
| 41 | ) |
| 42 | else: |
| 43 | cumulative_size += size |
| 44 | truncated_dict[key] = processed_value |
| 45 | |
| 46 | return truncated_dict |
| 47 | |
| 48 | elif isinstance(item, list): |
| 49 | truncated_list = [] |
| 50 | cumulative_size = 0 |
| 51 | |
| 52 | for value in item: |
| 53 | processed_value = process_item(value) |
| 54 | serialized_value = json.dumps(processed_value, ensure_ascii=False) |
| 55 | size = len(serialized_value) |
| 56 | |
| 57 | if cumulative_size + size > threshold_chars: |
| 58 | truncated_list.append( |
| 59 | truncate_text(agent, serialized_value, truncate_to) |
| 60 | ) |
| 61 | else: |
| 62 | cumulative_size += size |
| 63 | truncated_list.append(processed_value) |
| 64 | |
| 65 | return truncated_list |
| 66 | |
| 67 | elif isinstance(item, str): |
| 68 | if len(item) > threshold_chars: |
| 69 | return truncate_text(agent, item, truncate_to) |
| 70 | return item |
| 71 | |
| 72 | else: |
| 73 | return item |
| 74 | |
| 75 | return process_item(data) |