| 109 | |
| 110 | |
| 111 | class OpenAIModel(BaseModel): |
| 112 | _API_KEY_FIELD = "openai_api_key" |
| 113 | _ENV_VAR = "OPENAI_API_KEY" |
| 114 | _LOG_SOURCE = "openai" |
| 115 | _MAX_RATE_LIMIT_RETRIES = 5 |
| 116 | _MAX_TRANSIENT_RETRIES = 5 |
| 117 | _DEFAULT_CONFIG_CLASS = OpenAIModelConfig |
| 118 | |
| 119 | def _request_headers(self) -> dict[str, str]: |
| 120 | return { |
| 121 | "Content-Type": "application/json", |
| 122 | "Authorization": f"Bearer {self.config.openai_api_key}", |
| 123 | } |
| 124 | |
| 125 | def _post_url(self) -> str: |
| 126 | return self.config.openai_endpoint |
| 127 | |
| 128 | def _build_payload(self, messages: list[dict[str, Any]]) -> dict[str, Any]: |
| 129 | return { |
| 130 | "model": self.config.model_name, |
| 131 | "input": _serialize_response_input(messages), |
| 132 | "max_output_tokens": self.config.max_output_tokens, |
| 133 | "text": { |
| 134 | "format": { |
| 135 | "type": "json_schema", |
| 136 | "name": "playwright_step", |
| 137 | "schema": self._response_schema(), |
| 138 | "strict": True, |
| 139 | } |
| 140 | }, |
| 141 | } |
| 142 | |
| 143 | def _build_text_payload(self, messages: list[dict[str, Any]]) -> dict[str, Any]: |
| 144 | return { |
| 145 | "model": self.config.model_name, |
| 146 | "input": _serialize_response_input(messages), |
| 147 | "max_output_tokens": self.config.max_output_tokens, |
| 148 | } |
| 149 | |
| 150 | def _request_metrics_input(self, payload: dict[str, Any]) -> list[dict[str, Any]]: |
| 151 | return payload.get("input") or [] |
| 152 | |
| 153 | def _extract_text(self, payload: dict[str, Any]) -> str: |
| 154 | return _extract_response_text(payload) |
| 155 | |
| 156 | def _usage_metrics_from_payload(self, payload: dict[str, Any]) -> dict[str, int]: |
| 157 | return _usage_metrics_from_response_payload(payload) |
nothing calls this directly
no outgoing calls
no test coverage detected