windlass.providers.llm.groq

groq

Groq adapter.

Groq serves open-weight models (Llama, Mixtral, Gemma) on custom silicon at very high token rates. Its API is OpenAI-compatible, so this adapter reuses the message translation from :mod:windlass.providers.llm.openai and only differs in client construction and error mapping.

Install with::

pip install "windlass[groq]"

Example

from windlass import Windlass # doctest: +SKIP Windlass.llm("groq", model="llama-3.3-70b-versatile") # doctest: +SKIP

GroqLLM

GroqLLM(
    model: str = "",
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    max_retries: int = 0,
    **config: Any
)

Bases: LLM

Chat completions via the official groq SDK.

Parameters:

Name	Type	Description	Default
model	str	Model id, e.g. llama-3.3-70b-versatile.	''
api_key	str | None	Credential. Falls back to GROQ_API_KEY.	None
base_url	str | None	Endpoint override.	None
max_retries	int	SDK-level retries; 0 because Windlass retries itself.	0
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Raises:

Type	Description
MissingDependencyError	When groq is not installed.
AuthenticationError	When no API key can be found.

Source code in src\windlass\providers\llm\groq.py

def __init__(
    self,
    model: str = "",
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    max_retries: int = 0,
    **config: Any,
) -> None:
    super().__init__(model=model, **config)
    self._sdk = require("groq", extra="groq", feature="The Groq provider")
    key = api_key or settings().require_secret(
        "groq_api_key", provider="groq", env_var="GROQ_API_KEY"
    )
    kwargs: dict[str, Any] = {
        "api_key": key,
        "timeout": self.timeout,
        "max_retries": max_retries,
    }
    if base_url:
        kwargs["base_url"] = base_url
    self._client = self._sdk.AsyncGroq(**kwargs)

default_model classmethod

default_model() -> str

Return "llama-3.3-70b-versatile".

Source code in src\windlass\providers\llm\groq.py

@classmethod
def default_model(cls) -> str:
    """Return ``"llama-3.3-70b-versatile"``."""
    return "llama-3.3-70b-versatile"

native

native() -> Any

Return the underlying groq.AsyncGroq client.

Source code in src\windlass\providers\llm\groq.py

def native(self) -> Any:
    """Return the underlying ``groq.AsyncGroq`` client."""
    return self._client

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Request one chat completion.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	OpenAI-format tool definitions.	None
**kwargs	Any	Request overrides.	{}

Returns:

Type	Description
Completion	The completion.

Raises:

Type	Description
ProviderError	For any API failure.

Source code in src\windlass\providers\llm\groq.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Request one chat completion.

    Args:
        messages: The conversation.
        tools: OpenAI-format tool definitions.
        **kwargs: Request overrides.

    Returns:
        The completion.

    Raises:
        ProviderError: For any API failure.
    """
    payload: dict[str, Any] = {
        "model": self.model,
        "messages": to_openai_messages(messages),
        **self._merged(**kwargs),
    }
    if tools:
        payload["tools"] = tools
    try:
        response = await self._client.chat.completions.create(**payload)
    except Exception as exc:
        raise self._translate(exc) from exc

    try:
        choice = response.choices[0]
    except (AttributeError, IndexError) as exc:
        raise ResponseError("Groq returned no choices.", provider="groq") from exc

    calls = [
        ToolCall(
            id=tc.id,
            name=tc.function.name,
            arguments=_loads(tc.function.arguments),
            raw_arguments=tc.function.arguments,
        )
        for tc in (choice.message.tool_calls or [])
    ]
    usage = Usage()
    if getattr(response, "usage", None):
        usage = Usage(
            prompt_tokens=response.usage.prompt_tokens or 0,
            completion_tokens=response.usage.completion_tokens or 0,
        )
    return Completion(
        content=choice.message.content or "",
        tool_calls=calls,
        finish_reason=self._finish_reason(choice.finish_reason),
        model=getattr(response, "model", self.model),
        usage=usage,
        raw=response,
    )

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream a chat completion.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	OpenAI-format tool definitions.	None
**kwargs	Any	Request overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Text deltas, then completed tool calls, then done.

Source code in src\windlass\providers\llm\groq.py

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream a chat completion.

    Args:
        messages: The conversation.
        tools: OpenAI-format tool definitions.
        **kwargs: Request overrides.

    Yields:
        Text deltas, then completed tool calls, then ``done``.
    """
    payload: dict[str, Any] = {
        "model": self.model,
        "messages": to_openai_messages(messages),
        "stream": True,
        **self._merged(**kwargs),
    }
    if tools:
        payload["tools"] = tools

    partials: dict[int, dict[str, Any]] = {}
    finish: str | None = None
    try:
        stream = await self._client.chat.completions.create(**payload)
        async for chunk in stream:
            if not chunk.choices:
                continue
            choice = chunk.choices[0]
            if getattr(choice.delta, "content", None):
                yield StreamEvent(type="text", delta=choice.delta.content, raw=chunk)
            for fragment in getattr(choice.delta, "tool_calls", None) or []:
                slot = partials.setdefault(
                    fragment.index, {"id": "", "name": "", "arguments": ""}
                )
                if fragment.id:
                    slot["id"] = fragment.id
                if fragment.function and fragment.function.name:
                    slot["name"] = fragment.function.name
                if fragment.function and fragment.function.arguments:
                    slot["arguments"] += fragment.function.arguments
            if choice.finish_reason:
                finish = choice.finish_reason
    except Exception as exc:
        raise self._translate(exc) from exc

    for slot in partials.values():
        yield StreamEvent(type="tool_call", tool_call=_assemble(slot))
    yield StreamEvent(type="done", finish_reason=self._finish_reason(finish))

aclose async

aclose() -> None

Close the SDK's HTTP connection pool.

Source code in src\windlass\providers\llm\groq.py

async def aclose(self) -> None:
    """Close the SDK's HTTP connection pool."""
    close = getattr(self._client, "close", None)
    if close is not None:
        await close()