(
cls,
response_schema: Optional[Dict[str, Any]],
)
| 5173 | |
| 5174 | @classmethod |
| 5175 | def _cached_stream_plan( |
| 5176 | cls, |
| 5177 | response_schema: Optional[Dict[str, Any]], |
| 5178 | ) -> Optional[Dict[str, Any]]: |
| 5179 | if not isinstance(response_schema, dict): |
| 5180 | return None |
| 5181 | cache_key = id(response_schema) |
| 5182 | cached = cls._STREAM_PLAN_CACHE.get(cache_key) |
| 5183 | if cached is not None and cached[0] is response_schema: |
| 5184 | return cached[1] |
| 5185 | plan = cls._compile_stream_plan(response_schema) |
| 5186 | cls._STREAM_PLAN_CACHE[cache_key] = (response_schema, plan) |
| 5187 | return plan |
| 5188 | |
| 5189 | @staticmethod |
| 5190 | def _tool_schema_map( |
no test coverage detected