Class HttpxLanguageModel

python/src/typechat/_internal/model.py:38–111

]

class HttpxLanguageModel(TypeChatLanguageModel, AsyncContextManager):
    url: str
    headers: dict[str, str]
    default_params: dict[str, str]
    # Specifies the maximum number of retry attempts.
    max_retry_attempts: int = 3
    # Specifies the delay before retrying in milliseconds.
    retry_pause_seconds: float = 1.0
    # Specifies how long a request should wait in seconds
    # before timing out with a Failure.
    timeout_seconds = 10
    _async_client: httpx.AsyncClient

    def __init__(self, url: str, headers: dict[str, str], default_params: dict[str, str]):
        super().__init__()
        self.url = url
        self.headers = headers
        self.default_params = default_params
        self._async_client = httpx.AsyncClient()

    @override
    async def complete(self, prompt: str | list[PromptSection]) -> Success[str] | Failure:
        headers = {
            "Content-Type": "application/json",
            **self.headers,
        }

        if isinstance(prompt, str):
            prompt = [{"role": "user", "content": prompt}]

        body = {
            **self.default_params,
            "messages": prompt,
            "temperature": 0.0,
            "n": 1,
        }
        retry_count = 0
        while True:
            try:
                response = await self._async_client.post(
                    self.url,
                    headers=headers,
                    json=body,
                    timeout=self.timeout_seconds
                )
                if response.is_success:
                    json_result = cast(
                        dict[Literal["choices"], list[dict[Literal["message"], PromptSection]]],
                        response.json()
                    )
                    return Success(json_result["choices"][0]["message"]["content"] or "")

                if response.status_code not in _TRANSIENT_ERROR_CODES or retry_count >= self.max_retry_attempts:
                    return Failure(f"REST API error {response.status_code}: {response.reason_phrase}")
            except Exception as e:
                if retry_count >= self.max_retry_attempts:
                    return Failure(str(e) or f"{repr(e)} raised from within internal TypeChat language model.")