| 2502 | |
| 2503 | |
| 2504 | class CreateCompletionRequest(BaseModel): |
| 2505 | model_config = ConfigDict(extra="ignore") |
| 2506 | |
| 2507 | prompt: CompletionPrompt = "" |
| 2508 | suffix: Optional[str] = None |
| 2509 | max_tokens: Optional[int] = Field(default=16, ge=0) |
| 2510 | temperature: float = 0.8 |
| 2511 | top_p: float = Field(default=0.95, ge=0.0, le=1.0) |
| 2512 | echo: bool = False |
| 2513 | stop: Optional[Union[str, List[str]]] = None |
| 2514 | stream: bool = False |
| 2515 | logprobs: Optional[int] = Field(default=None, ge=0) |
| 2516 | presence_penalty: Optional[float] = Field(default=0.0, ge=-2.0, le=2.0) |
| 2517 | frequency_penalty: Optional[float] = Field(default=0.0, ge=-2.0, le=2.0) |
| 2518 | logit_bias: Optional[Dict[str, float]] = None |
| 2519 | seed: Optional[int] = None |
| 2520 | model: Optional[str] = None |
| 2521 | n: int = Field(default=1, ge=1) |
| 2522 | best_of: Optional[int] = Field(default=None, ge=1) |
| 2523 | user: Optional[str] = None |
| 2524 | |
| 2525 | @field_validator("logit_bias") |
| 2526 | @classmethod |
| 2527 | def validate_logit_bias(cls, value: Optional[Dict[str, float]]) -> Optional[Dict[str, float]]: |
| 2528 | if value is None: |
| 2529 | return None |
| 2530 | result: Dict[str, float] = {} |
| 2531 | for key, bias in value.items(): |
| 2532 | int(key) |
| 2533 | result[key] = float(bias) |
| 2534 | return result |
| 2535 | |
| 2536 | @model_validator(mode="after") |
| 2537 | def validate_after(self) -> "CreateCompletionRequest": |
| 2538 | if self.best_of is None: |
| 2539 | self.best_of = self.n |
| 2540 | if self.best_of < self.n: |
| 2541 | raise ValueError("best_of must be greater than or equal to n") |
| 2542 | if self.stream and self.best_of > 1: |
| 2543 | raise ValueError("best_of is not supported for streaming completions") |
| 2544 | if len(self.normalized_prompt()) > 1 and self.stream: |
| 2545 | raise ValueError("streaming does not support multiple prompts") |
| 2546 | return self |
| 2547 | |
| 2548 | def normalized_prompt(self) -> List[Union[str, List[int]]]: |
| 2549 | if isinstance(self.prompt, str): |
| 2550 | return [self.prompt] |
| 2551 | if all(isinstance(token, int) for token in self.prompt): |
| 2552 | return [cast(List[int], self.prompt)] |
| 2553 | if all(isinstance(prompt, str) for prompt in self.prompt): |
| 2554 | return cast(List[Union[str, List[int]]], list(cast(List[str], self.prompt))) |
| 2555 | if all( |
| 2556 | isinstance(prompt, list) |
| 2557 | and all(isinstance(token, int) for token in prompt) |
| 2558 | for prompt in self.prompt |
| 2559 | ): |
| 2560 | return cast( |
| 2561 | List[Union[str, List[int]]], |
no outgoing calls
no test coverage detected
searching dependent graphs…