Function create_completion

examples/server/server.py:15942–15995

(  # pyright: ignore[reportUnusedFunction]
        http_request: Request,
        body: CreateCompletionRequest,
    )

    @app.post("/v1/completions")
    async def create_completion(  # pyright: ignore[reportUnusedFunction]
        http_request: Request,
        body: CreateCompletionRequest,
    ):
        service: CompletionService = app.state.service
        formatter = service.formatter
        prompts = body.normalized_prompt()
        if len(prompts) > 1:
            if body.stream:
                raise HTTPException(
                    status_code=400,
                    detail="streaming does not support multiple prompts",
                )
            try:
                submissions = [
                    service.submit(body.model_copy(update={"prompt": prompt}))
                    for prompt in prompts
                ]
            except CompletionRequestValidationError as exc:
                raise bad_request(exc) from exc
            results = await collect_completion_results(
                formatter, http_request, submissions
            )
            if isinstance(results, Response):
                return results
            return JSONResponse(
                formatter.aggregate_completion_results(results).model_dump(
                    mode="json",
                    exclude_none=True,
                )
            )
        try:
            stream, cancel = service.submit(
                body.model_copy(update={"prompt": prompts[0]})
            )
        except CompletionRequestValidationError as exc:
            raise bad_request(exc) from exc
        if body.stream:
            return StreamingResponse(
                stream_sse_chunks(
                    formatter,
                    http_request,
                    stream,
                    cancel,
                    lambda chunk: [chunk],
                ),
                media_type="text/event-stream",
            )
        result = await collect_completion_result(
            formatter, http_request, stream, cancel
        )
        if isinstance(result, Response):
            return result
        return JSONResponse(result.model_dump(mode="json", exclude_none=True))

    @app.post("/v1/embeddings", response_model=CreateEmbeddingResponse)
    async def create_embedding(  # pyright: ignore[reportUnusedFunction]