Parse the completion, and put it into the raw_response.
(
self,
completion: Union[ChatCompletion, Generator[ChatCompletionChunk, None, None]],
)
| 216 | # return GeneratorOutput(data=None, error=str(e), raw_response=completion) |
| 217 | |
| 218 | def parse_chat_completion( |
| 219 | self, |
| 220 | completion: Union[ChatCompletion, Generator[ChatCompletionChunk, None, None]], |
| 221 | ) -> "GeneratorOutput": |
| 222 | """Parse the completion, and put it into the raw_response.""" |
| 223 | log.debug(f"completion: {completion}, parser: {self.chat_completion_parser}") |
| 224 | try: |
| 225 | data = self.chat_completion_parser(completion) |
| 226 | except Exception as e: |
| 227 | log.error(f"Error parsing the completion: {e}") |
| 228 | return GeneratorOutput(data=None, error=str(e), raw_response=completion) |
| 229 | |
| 230 | try: |
| 231 | usage = self.track_completion_usage(completion) |
| 232 | return GeneratorOutput( |
| 233 | data=None, error=None, raw_response=data, usage=usage |
| 234 | ) |
| 235 | except Exception as e: |
| 236 | log.error(f"Error tracking the completion usage: {e}") |
| 237 | return GeneratorOutput(data=None, error=str(e), raw_response=data) |
| 238 | |
| 239 | def track_completion_usage( |
| 240 | self, |
nothing calls this directly
no test coverage detected