(
body: dict[str, Any],
*,
model: str,
session_id: str | None,
step_type: str,
)
| 445 | |
| 446 | |
| 447 | def _stable_prompt_cache_key( |
| 448 | body: dict[str, Any], |
| 449 | *, |
| 450 | model: str, |
| 451 | session_id: str | None, |
| 452 | step_type: str, |
| 453 | ) -> str: |
| 454 | seed_parts = [model, session_id or "stateless", step_type] |
| 455 | tools = body.get("tools") or body.get("customTools") or [] |
| 456 | if tools: |
| 457 | try: |
| 458 | seed_parts.append(json.dumps(tools, sort_keys=True, ensure_ascii=False, separators=(",", ":"))) |
| 459 | except TypeError: |
| 460 | seed_parts.append(str(tools)) |
| 461 | system_parts: list[str] = [] |
| 462 | for msg in body.get("messages") or []: |
| 463 | if not isinstance(msg, dict): |
| 464 | continue |
| 465 | if msg.get("role") == "system": |
| 466 | system_parts.append(_message_text(msg)) |
| 467 | if system_parts: |
| 468 | seed_parts.append("\n".join(system_parts)[:4096]) |
| 469 | digest = hashlib.sha256("\n".join(seed_parts).encode("utf-8")).hexdigest()[:24] |
| 470 | return f"ur:{digest}" |
| 471 | |
| 472 | |
| 473 | def _is_openai_cache_retention_model(model: str) -> bool: |
no test coverage detected