Function create_chat_completion

examples/server/server.py:16009–16084

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

    @app.post("/v1/chat/completions")
    async def create_chat_completion(  # pyright: ignore[reportUnusedFunction]
        http_request: Request, body: CreateChatCompletionRequest
    ):
        service: CompletionService = app.state.service
        formatter = service.formatter
        try:
            parts = formatter.completion_request_from_chat_request(
                body,
            )
        except CompletionRequestValidationError as exc:
            raise bad_request(exc) from exc
        template_functions = (
            [function.to_template_function() for function in body.functions]
            if body.functions is not None
            else None
        )
        template_tools = (
            [tool.to_template_tool() for tool in body.tools]
            if body.tools is not None
            else None
        )
        try:
            request = service.request_from_prepared(
                payload=parts.payload,
                prompt_text=parts.prompt_text,
                prompt_plan=parts.prompt_plan,
                grammar_text=parts.grammar_text,
                chat_tool_name=parts.tool_name,
            )
            stream, cancel = service.submit_request(request)
        except CompletionRequestValidationError as exc:
            raise bad_request(exc) from exc
        if body.stream:
            started_indices: set[int] = set()

            def chat_chunk_payloads(
                completion_chunk: CompletionChunk,
            ) -> Iterable[BaseModel | Dict[str, Any]]:
                return formatter.convert_completion_chunk_to_chat_chunks(
                    completion_chunk,
                    started_indices,
                    parts.tool_name,
                    functions=template_functions,
                    tools=template_tools,
                    parsed_states=parsed_states,
                    generation_prompt=parts.generation_prompt,
                )

            parsed_states: Dict[int, Dict[str, Any]] = {}
            return StreamingResponse(
                stream_sse_chunks(
                    formatter,
                    http_request,
                    stream,
                    cancel,
                    chat_chunk_payloads,
                ),
                media_type="text/event-stream",