(
self,
*,
chunk_id: str,
created: int,
model: str,
deltas: List[Dict[str, Any]],
finish_reason: Optional[str],
logprobs: Optional[Dict[str, Any]],
leading_delta: Optional[Dict[str, Any]] = None,
)
| 7611 | } |
| 7612 | |
| 7613 | def _chunk_payloads( |
| 7614 | self, |
| 7615 | *, |
| 7616 | chunk_id: str, |
| 7617 | created: int, |
| 7618 | model: str, |
| 7619 | deltas: List[Dict[str, Any]], |
| 7620 | finish_reason: Optional[str], |
| 7621 | logprobs: Optional[Dict[str, Any]], |
| 7622 | leading_delta: Optional[Dict[str, Any]] = None, |
| 7623 | ) -> List[Dict[str, Any]]: |
| 7624 | payloads: List[Dict[str, Any]] = [] |
| 7625 | logprobs_sent = False |
| 7626 | chat_chunk_id = "chat" + chunk_id |
| 7627 | index = self._choice_index |
| 7628 | if leading_delta is not None: |
| 7629 | payloads.append( |
| 7630 | { |
| 7631 | "id": chat_chunk_id, |
| 7632 | "object": "chat.completion.chunk", |
| 7633 | "created": created, |
| 7634 | "model": model, |
| 7635 | "choices": [ |
| 7636 | { |
| 7637 | "index": index, |
| 7638 | "delta": leading_delta, |
| 7639 | "finish_reason": None, |
| 7640 | } |
| 7641 | ], |
| 7642 | } |
| 7643 | ) |
| 7644 | for delta in deltas: |
| 7645 | payload_logprobs = None |
| 7646 | if ( |
| 7647 | logprobs is not None |
| 7648 | and delta.get("role") != "assistant" |
| 7649 | and not logprobs_sent |
| 7650 | ): |
| 7651 | payload_logprobs = logprobs |
| 7652 | logprobs_sent = True |
| 7653 | choice: Dict[str, Any] = { |
| 7654 | "index": index, |
| 7655 | "delta": delta, |
| 7656 | "finish_reason": None, |
| 7657 | } |
| 7658 | if payload_logprobs is not None: |
| 7659 | choice["logprobs"] = payload_logprobs |
| 7660 | payloads.append( |
| 7661 | { |
| 7662 | "id": chat_chunk_id, |
| 7663 | "object": "chat.completion.chunk", |
| 7664 | "created": created, |
| 7665 | "model": model, |
| 7666 | "choices": [choice], |
| 7667 | } |
| 7668 | ) |
| 7669 | if finish_reason is not None: |
| 7670 | payloads.append( |
no test coverage detected