Implementation:NVIDIA NeMo Curator LLMClient

Knowledge Sources	NVIDIA NeMo Curator
Domains	LLM Integration, Inference Client, Abstraction Layer
Last Updated	2026-02-14 00:00 GMT

Overview

Defines abstract interfaces for synchronous and asynchronous LLM inference clients, a conversation formatting abstraction, and a generation configuration dataclass, forming the core abstraction layer for LLM-based data annotation in the NeMo Curator pipeline.

Description

The llm_client.py module provides four key classes:

ConversationFormatter (ABC): An abstract base class for formatting conversation dictionaries into model-ready strings. Subclasses must implement format_conversation(conv: list[dict]) -> str to convert a list of message dictionaries into the format expected by the target LLM.

GenerationConfig (dataclass): A configuration class holding LLM sampling parameters:

• max_tokens (default: 2048) - Maximum number of tokens to generate
• n (default: 1) - Number of completions to generate
• seed (default: 0) - Random seed for reproducibility
• stop (default: None) - Stop sequence(s)
• stream (default: False) - Whether to stream the response
• temperature (default: 0.0) - Sampling temperature
• top_k (default: None) - Top-k sampling parameter
• top_p (default: 0.95) - Nucleus sampling parameter

LLMClient (ABC): A synchronous interface with two abstract methods:

• setup() - Initialize the client connection
• query_model(messages, model, conversation_formatter, generation_config) - Send a query and return a list of response strings

AsyncLLMClient (ABC): An asynchronous interface that wraps the subclass-implemented _query_model_impl() method with production-grade retry and concurrency management:

• Concurrency control via asyncio.Semaphore limiting simultaneous requests to max_concurrent_requests (default: 5)
• Exponential backoff with jitter for rate-limited (HTTP 429) and connection error retries, up to max_retries attempts (default: 3)
• Error classification distinguishing retryable errors (rate limits, connection failures including ReadError, BrokenResourceError, APIConnectionError) from non-retryable errors that are immediately re-raised
• Semaphore lifecycle management that detects event loop changes and reinitializes the semaphore appropriately

The retry logic uses a base delay of 1.0 second with exponential backoff (base_delay * 2^(attempt-1)) plus random jitter from secrets.randbelow(100) / 100.0.

Usage

These abstract classes serve as the foundation for concrete LLM client implementations (e.g., OpenAI, NVIDIA NIM, local inference servers). Use LLMClient for synchronous workflows and AsyncLLMClient for high-throughput asynchronous data annotation pipelines where concurrency control and retry management are essential.

Code Reference

Source Location

• Repository: NeMo-Curator
• File: nemo_curator/models/client/llm_client.py
• Lines: 1-213

Signature

class ConversationFormatter(ABC):
    @abstractmethod
    def format_conversation(self, conv: list[dict]) -> str: ...

@dataclass
class GenerationConfig:
    max_tokens: int | None = 2048
    n: int | None = 1
    seed: int | None = 0
    stop: str | None | list[str] = None
    stream: bool = False
    temperature: float | None = 0.0
    top_k: int | None = None
    top_p: float | None = 0.95

class LLMClient(ABC):
    @abstractmethod
    def setup(self) -> None: ...
    @abstractmethod
    def query_model(
        self,
        *,
        messages: Iterable,
        model: str,
        conversation_formatter: ConversationFormatter | None = None,
        generation_config: GenerationConfig | dict | None = None,
    ) -> list[str]: ...

class AsyncLLMClient(ABC):
    def __init__(
        self,
        max_concurrent_requests: int = 5,
        max_retries: int = 3,
        base_delay: float = 1.0,
    ): ...
    @abstractmethod
    def setup(self) -> None: ...
    @abstractmethod
    async def _query_model_impl(
        self,
        *,
        messages: Iterable,
        model: str,
        conversation_formatter: ConversationFormatter | None = None,
        generation_config: GenerationConfig | dict | None = None,
    ) -> list[str]: ...
    async def query_model(
        self,
        *,
        messages: Iterable,
        model: str,
        conversation_formatter: ConversationFormatter | None = None,
        generation_config: GenerationConfig | dict | None = None,
    ) -> list[str]: ...

Import

from nemo_curator.models.client.llm_client import (
    AsyncLLMClient,
    ConversationFormatter,
    GenerationConfig,
    LLMClient,
)

I/O Contract

Inputs

Name	Type	Required	Description
messages	Iterable	Yes	Conversation messages to send to the LLM (list of dicts with role/content)
model	str	Yes	Model identifier string (e.g., "gpt-4", "openai/gpt-oss-20b")
conversation_formatter	ConversationFormatter or None	No	Optional formatter to convert conversation dicts to model-ready strings
generation_config	GenerationConfig or dict or None	No	Sampling parameters; defaults to GenerationConfig() if None
max_concurrent_requests	int	No	Maximum concurrent async requests (AsyncLLMClient only, default: 5)
max_retries	int	No	Maximum retry attempts for retryable errors (AsyncLLMClient only, default: 3)
base_delay	float	No	Base delay in seconds for exponential backoff (AsyncLLMClient only, default: 1.0)

Outputs

Name	Type	Description
responses	list[str]	List of generated text responses from the LLM

Usage Examples

Implementing a Synchronous Client

from nemo_curator.models.client.llm_client import LLMClient, GenerationConfig

class MyLLMClient(LLMClient):
    def setup(self):
        # Initialize connection to the inference server
        self.client = create_connection()

    def query_model(self, *, messages, model, conversation_formatter=None, generation_config=None):
        config = generation_config or GenerationConfig()
        response = self.client.generate(
            messages=messages,
            model=model,
            temperature=config.temperature,
            max_tokens=config.max_tokens,
        )
        return [response.text]

Implementing an Asynchronous Client

from nemo_curator.models.client.llm_client import AsyncLLMClient, GenerationConfig

class MyAsyncLLMClient(AsyncLLMClient):
    def __init__(self):
        super().__init__(max_concurrent_requests=10, max_retries=5, base_delay=2.0)

    def setup(self):
        self.client = create_async_connection()

    async def _query_model_impl(self, *, messages, model, conversation_formatter=None, generation_config=None):
        config = generation_config or GenerationConfig()
        response = await self.client.agenerate(
            messages=messages,
            model=model,
            temperature=config.temperature,
        )
        return [response.text]

# The parent class query_model() automatically wraps _query_model_impl()
# with retry logic and concurrency control via asyncio.Semaphore

Using GenerationConfig

from nemo_curator.models.client.llm_client import GenerationConfig

# Default configuration (temperature=0.0 for deterministic output)
config = GenerationConfig()

# Custom sampling configuration
config = GenerationConfig(
    max_tokens=4096,
    temperature=0.7,
    top_p=0.9,
    top_k=50,
    seed=42,
    stop=["\n\n", "###"],
)

# Can also be passed as a dict
client.query_model(
    messages=messages,
    model="gpt-4",
    generation_config={"temperature": 0.5, "max_tokens": 1024},
)

Related Pages

• Environment:NVIDIA_NeMo_Curator_Python_Linux_Base