(self, messages: list[dict], **kwargs)
| 165 | ] |
| 166 | |
| 167 | def _query(self, messages: list[dict], **kwargs): |
| 168 | assert messages[0]["role"] == "system" |
| 169 | api_messages = self._set_cache_control_on_last_message( |
| 170 | [{"role": m["role"], "content": self._strip_display_text(m["content"])} for m in messages[1:]] |
| 171 | ) |
| 172 | extra_model_kwargs = self.config.model_kwargs | kwargs |
| 173 | if self.config.drop_none_model_kwargs: |
| 174 | extra_model_kwargs = {k: v for k, v in extra_model_kwargs.items() if v is not None} |
| 175 | return self.client.messages.create( |
| 176 | model=self.config.model_name, |
| 177 | max_tokens=self.config.max_tokens, |
| 178 | system=messages[0]["content"], |
| 179 | messages=api_messages, |
| 180 | tools=[ANTHROPIC_BASH_TOOL], |
| 181 | **extra_model_kwargs, |
| 182 | ) |
| 183 | |
| 184 | def query(self, messages: list[dict[str, str]], **kwargs) -> dict: |
| 185 | for attempt in retry(logger=logger, abort_exceptions=self.abort_exceptions): |
no test coverage detected