Warn when the LLM hit the max_tokens cap; return True if it did. ``json_repair`` will silently salvage the truncated prefix, so without this the caller can't tell a short response from a cut-off one. Callers that write a page from the response can pass ``raise_on_truncation=True`` t
(response, step_name: str, max_tokens: int | None)
| 494 | |
| 495 | |
| 496 | def _warn_if_truncated(response, step_name: str, max_tokens: int | None) -> bool: |
| 497 | """Warn when the LLM hit the max_tokens cap; return True if it did. |
| 498 | |
| 499 | ``json_repair`` will silently salvage the truncated prefix, so without |
| 500 | this the caller can't tell a short response from a cut-off one. Callers |
| 501 | that write a page from the response can pass ``raise_on_truncation=True`` |
| 502 | to ``_llm_call``/`_llm_call_async`` to turn a truncated response into a |
| 503 | skip instead of persisting partial content. |
| 504 | """ |
| 505 | try: |
| 506 | finish_reason = response.choices[0].finish_reason |
| 507 | except (AttributeError, IndexError): |
| 508 | return False |
| 509 | if finish_reason != "length": |
| 510 | return False |
| 511 | cap = f" (max_tokens={max_tokens})" if max_tokens else "" |
| 512 | logger.warning("LLM [%s] hit length limit%s — output may be truncated.", step_name, cap) |
| 513 | sys.stdout.write(f" [WARN] {step_name} hit length limit{cap} — output may be truncated.\n") |
| 514 | sys.stdout.flush() |
| 515 | return True |
| 516 | |
| 517 | |
| 518 | def _parse_json(text: str) -> list | dict: |
no test coverage detected