(
self,
completion: Union[ChatCompletion, Generator[ChatCompletionChunk, None, None]],
)
| 237 | return GeneratorOutput(data=None, error=str(e), raw_response=data) |
| 238 | |
| 239 | def track_completion_usage( |
| 240 | self, |
| 241 | completion: Union[ChatCompletion, Generator[ChatCompletionChunk, None, None]], |
| 242 | ) -> CompletionUsage: |
| 243 | |
| 244 | try: |
| 245 | usage: CompletionUsage = CompletionUsage( |
| 246 | completion_tokens=completion.usage.completion_tokens, |
| 247 | prompt_tokens=completion.usage.prompt_tokens, |
| 248 | total_tokens=completion.usage.total_tokens, |
| 249 | ) |
| 250 | return usage |
| 251 | except Exception as e: |
| 252 | log.error(f"Error tracking the completion usage: {e}") |
| 253 | return CompletionUsage( |
| 254 | completion_tokens=None, prompt_tokens=None, total_tokens=None |
| 255 | ) |
| 256 | |
| 257 | def parse_embedding_response( |
| 258 | self, response: CreateEmbeddingResponse |
no outgoing calls
no test coverage detected