(
self,
messages: List[Union[Message, Dict]],
tools: Optional[List[Union[Tool, Dict]]] = None,
)
| 23 | return self.pool |
| 24 | |
| 25 | def generate( |
| 26 | self, |
| 27 | messages: List[Union[Message, Dict]], |
| 28 | tools: Optional[List[Union[Tool, Dict]]] = None, |
| 29 | ) -> Tuple[Message, DebugInfo]: |
| 30 | if len(messages) > 0 and isinstance(messages[0], Message): |
| 31 | kwargs = dict( |
| 32 | messages=[message2dict(msg) for msg in messages], |
| 33 | ) |
| 34 | elif isinstance(messages[0], dict): |
| 35 | kwargs = dict( |
| 36 | messages=messages, |
| 37 | ) |
| 38 | kwargs.update(self.config["model_args"]) |
| 39 | if tools is not None: |
| 40 | if len(tools) > 0 and isinstance(tools[0], Tool): |
| 41 | kwargs["tools"] = [tool2dict(tool) for tool in tools] |
| 42 | kwargs["tool_choice"] = "auto" |
| 43 | else: |
| 44 | kwargs["tools"] = tools |
| 45 | kwargs["tool_choice"] = "auto" |
| 46 | |
| 47 | new_message, debug_info = self._post_request(kwargs) |
| 48 | |
| 49 | new_message = Message(**new_message.model_dump()) |
| 50 | return new_message, debug_info |
| 51 | |
| 52 | @retry(stop=stop_after_attempt(10), wait=wait_random(min=5, max=10)) |
| 53 | def _post_request(self, kwargs: Dict[str, Any]) -> Tuple[OpenAIMessage, DebugInfo]: |
nothing calls this directly
no test coverage detected