Plain text formatter: clean prompts in, plain text out.
| 14 | |
| 15 | |
| 16 | class PlainTextFormatter(StatefulFormatter): |
| 17 | """Plain text formatter: clean prompts in, plain text out.""" |
| 18 | |
| 19 | def __init__(self): |
| 20 | super().__init__() |
| 21 | self._is_input_formatter = True |
| 22 | self._is_output_formatter = True |
| 23 | self._agent_introducer = "" |
| 24 | |
| 25 | def dump(self, message): |
| 26 | if isinstance(message, str): |
| 27 | return message |
| 28 | if isinstance(message, dict): |
| 29 | if len(message) == 1: |
| 30 | val = next(iter(message.values())) |
| 31 | if isinstance(val, str): |
| 32 | return val |
| 33 | lines = [] |
| 34 | for key, val in message.items(): |
| 35 | if val is None: |
| 36 | continue |
| 37 | rendered = val if isinstance(val, str) else str(val) |
| 38 | lines.append(f"{key}:\n{rendered}") |
| 39 | return "\n\n".join(lines) |
| 40 | return str(message) |
| 41 | |
| 42 | def format(self, message): |
| 43 | raw = message |
| 44 | if isinstance(raw, dict): |
| 45 | return self._normalize_dict(raw) |
| 46 | text = str(raw) |
| 47 | text = re.sub(r"<think.*?</think\s*>", "", text, flags=re.DOTALL).strip() |
| 48 | output_key = list(self._field_keys.keys())[0] if self._field_keys else "output" |
| 49 | for cand in re.findall(r"```json\s*(.*?)```", text, flags=re.DOTALL): |
| 50 | cand = cand.strip() |
| 51 | try: |
| 52 | parsed = _json.loads(cand) |
| 53 | return self._normalize_dict(parsed) |
| 54 | except Exception: |
| 55 | pass |
| 56 | start, end = text.find("{"), text.rfind("}") |
| 57 | if start != -1 and end != -1 and end > start: |
| 58 | try: |
| 59 | parsed = _json.loads(text[start:end + 1]) |
| 60 | return self._normalize_dict(parsed) |
| 61 | except Exception: |
| 62 | pass |
| 63 | return {output_key: text.strip()} |
| 64 | |
| 65 | def _normalize_dict(self, d): |
| 66 | output_key = list(self._field_keys.keys())[0] if self._field_keys else "output" |
| 67 | result = {output_key: ""} |
| 68 | if output_key in d: |
| 69 | result[output_key] = d[output_key] |
| 70 | else: |
| 71 | for v in d.values(): |
| 72 | if isinstance(v, str): |
| 73 | result[output_key] = v |
no outgoing calls
no test coverage detected