Source code for xpark.dataset.processors.text_generate

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Iterable

from xpark.dataset.constants import NOT_SET
from xpark.dataset.datatype import DataType
from xpark.dataset.expressions import BatchColumnClassProtocol, udf
from xpark.dataset.import_utils import lazy_import
from xpark.dataset.utils import LLMChatCompletions, skip_empty_texts

if TYPE_CHECKING:
    import pyarrow as pa
    from openai.types.chat.chat_completion_message_param import ChatCompletionMessageParam
else:
    openai = lazy_import("openai")
    pa = lazy_import("pyarrow", rename="pa")

logger = logging.getLogger("ray")

# prompt modify from https://github.com/apache/doris/blob/4.0.2-rc01/be/src/vec/functions/ai/ai_generate.h
SYSTEM_ROLE_PROMPT = (
    "You are a creative llm model. You will generate a concise and highly relevant response based on the user's input. "
    "Aim for maximum brevity—cut every non-essential word."
)

PROMPT_TEMPLATE = """
Input:
{}
"""


def build_prompt(text: str) -> Iterable[ChatCompletionMessageParam]:
    from openai.types.chat.chat_completion_message_param import (
        ChatCompletionSystemMessageParam,
        ChatCompletionUserMessageParam,
    )

    return [
        ChatCompletionSystemMessageParam(role="system", content=SYSTEM_ROLE_PROMPT),
        ChatCompletionUserMessageParam(role="user", content=PROMPT_TEMPLATE.format(text)),
    ]




@udf(return_dtype=DataType.string())
class TextGenerate(BatchColumnClassProtocol):
    """TextGenerate processor generates content based on the input parameters. generate by llm model.

    Args:
        base_url: The base URL of the LLM server.
        model: The request model name.
        api_key: The request API key.
        max_qps: The maximum number of requests per second.
        max_retries: The maximum number of retries per request in the event of failures.
            We retry with exponential backoff upto this specific maximum retries.
        fallback_response: The response value to return when the LLM request fails.
            If set to None, the exception will be raised instead.
        **kwargs: Keyword arguments to pass to the `openai.AsyncClient.chat.completions.create
            <https://github.com/openai/openai-python/blob/main/src/openai/resources/chat/completions/completions.py>`_ API.

    Examples:
        .. code-block:: python

            from xpark.dataset.expressions import col
            from xpark.dataset import TextGenerate, from_items

            ds = from_items([""])
            ds = ds.with_column(
                "generated_text",
                TextGenerate(
                    model="deepseek-v3-0324",
                    base_url=os.getenv("LLM_ENDPOINT"),
                    api_key=os.getenv("LLM_API_KEY"),
                )
                .options(num_workers={"IO": 1}, batch_size=1)
                .with_column(col("item")),
            )

            print(ds.take_all())

    """

    def __init__(
        self,
        /,
        *,
        base_url: str,
        model: str,
        api_key: str = NOT_SET,
        max_qps: int | None = None,
        max_retries: int = 0,
        fallback_response: str | None = None,
        **kwargs: dict[str, Any],
    ):
        self.model = LLMChatCompletions(
            base_url=base_url,
            model=model,
            api_key=api_key,
            max_qps=max_qps,
            max_retries=max_retries,
            fallback_response=fallback_response,
            response_format="text",
            **kwargs,
        )

    @skip_empty_texts
    async def __call__(self, texts: pa.ChunkedArray) -> pa.Array:
        return await self.model.batch_generate(
            texts=texts,
            build_prompt=build_prompt,
        )