jianmu.model

适用对象：模型接入开发者 / Provider 维护者 / 高阶使用者 是否必读：按需 相关模块：jianmu.message, jianmu.node, jianmu.swarm

1. 模块职责

jianmu.model 是 Jianmu 的模型系统，负责把“调用大模型”抽象成框架内部可复用、可替换、可观测的一层能力。

它主要覆盖以下职责：

• model provider 抽象
• 统一调用门面
• 流式与非流式返回统一
• 消息编码与解码协议
• provider 注册、解析、重试与降级
• token usage 与流式增量遥测发射

它不是具体业务节点，而是 LLM 调用能力的抽象层。

2. 适合查什么

• 基础类型：ModelConfig、MessageChunk
• provider 协议：ModelProviderProtocol
• 消息编码：MessageEncoderProtocol
• 主入口：ModelClient
• 注册机制：ModelClient.register_provider()、ModelClient.resolve()
• 内置 provider：OpenAIProvider、LiteLLMProvider

如果你是第一次进入这个模块，建议按这个顺序看：

1. ModelConfig
2. ModelProviderProtocol
3. ModelClient
4. invoke_model()
5. OpenAIProvider / LiteLLMProvider
6. OpenAIChatEncoder / LiteLLMChatEncoder

3. 使用建议

• 普通应用通常直接使用 ModelClient.resolve() 即可，不必先手动实例化所有 provider
• 需要接入新模型后端时，再实现 provider 协议并通过 ModelClient.register_provider() 注册
• 想理解一次统一调用该传什么参数时，优先看 ModelConfig
• 想理解为什么流式 tool call 能恢复成统一消息对象，继续看 invoke_model() 和 stream.py

4. 注意事项

• OpenAIProvider 和 LiteLLMProvider 在这个模块中是 lazy export
• 实际可用性依赖环境变量、端点配置和对应依赖
• 这里的编码协议和 jianmu.message 的消息领域模型是分层设计，不应混用职责
• ModelClient 是入口，但它不等于整个模型系统；真正的调用链还包括 invoke.py、stream.py、codecs.py 和 providers/

5. 最小示例

from jianmu.model import ModelClient, ModelConfig

client = ModelClient.resolve()
config = ModelConfig(model="gpt-4.1-mini", temperature=0.2)

6. 常见入口

• 想拿默认 provider：看 ModelClient.resolve()
• 想注册自定义 provider：看 ModelClient.register_provider()
• 想理解统一调用参数：看 ModelConfig

7. API 参考

model

LLM provider interfaces, model client facade, and registry helpers for Jianmu.

MessageChunk dataclass

MessageChunk(
    text: str = "",
    tool_calls: Optional[List[Dict[str, Any]]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    raw: Optional[Any] = None,
)

Single streamed chunk emitted by a model provider.

属性：

名称	类型	描述
text	str	Incremental text content emitted in this chunk.
tool_calls	Optional[List[Dict[str, Any]]]	Optional tool-call deltas emitted by the provider.
metadata	Optional[Dict[str, Any]]	Optional provider-specific metadata for the chunk.
raw	Optional[Any]	Optional raw provider payload preserved for advanced consumers.

ModelConfig dataclass

ModelConfig(
    model: str,
    temperature: float = 0.7,
    top_p: float = 0.95,
    top_k: int = 40,
    max_tokens: Optional[int] = None,
    timeout: float = 120.0,
    tools: Optional[List[Dict[str, Any]]] = None,
    tool_choice: Optional[Any] = None,
    strict_tools: bool = False,
    response_format: Optional[Dict[str, Any]] = None,
    max_retries: Optional[int] = None,
    fallback_model: Optional[str] = None,
    disable_fallback: bool = False,
    extra: Dict[str, Any] = dict(),
)

Configuration for one provider model call.

ModelConfig is the transport-neutral call configuration handed to model providers. It keeps common generation settings, tool-calling settings, and provider-specific extensions in one object.

属性：

名称	类型	描述
model	str	Provider-specific model name or identifier.
temperature	float	Sampling temperature for generation.
top_p	float	Nucleus-sampling cutoff.
top_k	int	Top-k sampling cutoff when supported.
max_tokens	Optional[int]	Optional maximum generated-token limit.
timeout	float	Request timeout in seconds.
tools	Optional[List[Dict[str, Any]]]	Optional tool schema list exposed to the model.
tool_choice	Optional[Any]	Optional provider-specific tool selection directive.
strict_tools	bool	Whether tool arguments must strictly match the schema.
response_format	Optional[Dict[str, Any]]	Optional structured-output configuration.
extra	Dict[str, Any]	Provider-specific extra request options.

with_tools

with_tools(
    tools: List[Dict[str, Any]], strict: bool = False
) -> "ModelConfig"

Return a copy with tool-calling settings replaced.

参数：

名称	类型	描述	默认
tools	List[Dict[str, Any]]	The list of tool schemas to use.	必需
strict	bool	Whether to enforce strict schema adherence. Defaults to False.	False

返回：

类型	描述
'ModelConfig'	A new ModelConfig instance with updated tool configuration.

源代码位于： jianmu/model/types.py

def with_tools(self, tools: List[Dict[str, Any]], strict: bool = False) -> "ModelConfig":
    """Return a copy with tool-calling settings replaced.

    Args:
        tools: The list of tool schemas to use.
        strict: Whether to enforce strict schema adherence. Defaults to False.

    Returns:
        A new ``ModelConfig`` instance with updated tool configuration.
    """
    return ModelConfig(
        model=self.model,
        temperature=self.temperature,
        top_p=self.top_p,
        top_k=self.top_k,
        max_tokens=self.max_tokens,
        timeout=self.timeout,
        tools=tools,
        tool_choice=self.tool_choice,
        strict_tools=strict,
        response_format=self.response_format,
        max_retries=self.max_retries,
        fallback_model=self.fallback_model,
        disable_fallback=self.disable_fallback,
        extra=self.extra,
    )

with_extra

with_extra(**kwargs: Any) -> 'ModelConfig'

Return a copy with merged provider-specific extras.

参数：

名称	类型	描述	默认
kwargs	Any	Key-value pairs to merge into the extra configuration dictionary.	{}

返回：

类型	描述
'ModelConfig'	A new ModelConfig instance with updated extra dictionary.

源代码位于： jianmu/model/types.py

def with_extra(self, **kwargs: Any) -> "ModelConfig":
    """Return a copy with merged provider-specific extras.

    Args:
        kwargs: Key-value pairs to merge into the extra configuration dictionary.

    Returns:
        A new ``ModelConfig`` instance with updated extra dictionary.
    """
    new_extra = {**self.extra, **kwargs}
    return ModelConfig(
        model=self.model,
        temperature=self.temperature,
        top_p=self.top_p,
        top_k=self.top_k,
        max_tokens=self.max_tokens,
        timeout=self.timeout,
        tools=self.tools,
        tool_choice=self.tool_choice,
        strict_tools=self.strict_tools,
        response_format=self.response_format,
        max_retries=self.max_retries,
        fallback_model=self.fallback_model,
        disable_fallback=self.disable_fallback,
        extra=new_extra,
    )

ModelProviderProtocol

Bases: Protocol

Protocol for model transport clients used by Jianmu.

generate async

generate(
    messages: List[Message], config: ModelConfig
) -> Message

Generate a single assistant message.

参数：

名称	类型	描述	默认
messages	List[Message]	Ordered conversation history for the current request.	必需
config	ModelConfig	Model configuration, including tool and response-format options.	必需

返回：

类型	描述
Message	A decoded assistant message in Jianmu's message format.

源代码位于： jianmu/model/base.py

async def generate(
    self,
    messages: List[Message],
    config: ModelConfig,
) -> Message:
    """Generate a single assistant message.

    Args:
        messages: Ordered conversation history for the current request.
        config: Model configuration, including tool and response-format options.

    Returns:
        A decoded assistant message in Jianmu's message format.
    """
    ...

stream async

stream(
    messages: List[Message], config: ModelConfig
) -> AsyncIterator[MessageChunk]

Stream assistant output as incremental message chunks.

参数：

名称	类型	描述	默认
messages	List[Message]	Ordered conversation history for the current request.	必需
config	ModelConfig	Model configuration, including tool and response-format options.	必需

产生：

类型	描述
AsyncIterator[MessageChunk]	Incremental output chunks decoded into Jianmu's streaming format.

返回：

类型	描述
AsyncIterator[MessageChunk]	An async iterator of message chunks.

源代码位于： jianmu/model/base.py

async def stream(
    self,
    messages: List[Message],
    config: ModelConfig,
) -> AsyncIterator[MessageChunk]:
    """Stream assistant output as incremental message chunks.

    Args:
        messages: Ordered conversation history for the current request.
        config: Model configuration, including tool and response-format options.

    Yields:
        Incremental output chunks decoded into Jianmu's streaming format.

    Returns:
        An async iterator of message chunks.
    """
    ...

ModelClient

ModelClient(provider: ModelProviderProtocol)

Invocation facade bound to one resolved model provider.

ModelClient is the object to use when callers want Jianmu's normalized invocation surface via :meth:invoke.

Resolution helpers are intentionally split by return type:

1. :meth:resolve_provider returns the raw provider instance
2. :meth:resolve returns a ModelClient wrapping that provider

属性：

名称	类型	描述
provider		Bound provider implementation used to service invocations.

Bind the client to a concrete provider instance.

源代码位于： jianmu/model/client.py

def __init__(self, provider: ModelProviderProtocol):
    """Bind the client to a concrete provider instance."""
    self.provider = provider

resolve classmethod

resolve(
    name: str | None = None,
    preference: list[str] | None = None,
    env_override: bool = False,
    **kwargs: Any,
) -> "ModelClient"

Resolve a provider and wrap it in a ModelClient.

Use this when the caller wants to invoke the model through :meth:invoke. Callers that need the provider object itself should use :meth:resolve_provider directly.

参数：

名称	类型	描述	默认
name	str | None	Name used by the operation.	None
preference	list[str] | None	The preference value.	None
env_override	bool	The env_override value.	False
**kwargs	Any	Additional keyword arguments.	{}

返回：

类型	描述
'ModelClient'	The resulting 'ModelClient' value.

源代码位于： jianmu/model/client.py

@classmethod
def resolve(
    cls,
    name: str | None = None,
    preference: list[str] | None = None,
    env_override: bool = False,
    **kwargs: Any,
) -> "ModelClient":
    """Resolve a provider and wrap it in a ``ModelClient``.

    Use this when the caller wants to invoke the model through
    :meth:`invoke`. Callers that need the provider object itself should use
    :meth:`resolve_provider` directly.

    Args:
        name: Name used by the operation.
        preference: The `preference` value.
        env_override: The `env_override` value.
        **kwargs: Additional keyword arguments.

    Returns:
        The resulting `'ModelClient'` value.
    """
    return cls(cls.resolve_provider(name=name, preference=preference, env_override=env_override, **kwargs))

resolve_provider classmethod

resolve_provider(
    name: str | None = None,
    preference: list[str] | None = None,
    env_override: bool = False,
    **kwargs: Any,
) -> ModelProviderProtocol

Resolve and instantiate the first usable provider.

This returns the raw provider instance rather than a ModelClient. Use it when framework code or examples need a provider object to inject into nodes, runtimes, or contexts.

参数：

名称	类型	描述	默认
name	str | None	Explicit provider name. When omitted, providers are selected from preference by available credentials.	None
preference	list[str] | None	Provider resolution order.	None
env_override	bool	Whether project .env values should override existing process environment variables.	False
**kwargs	Any	Extra provider-specific constructor arguments.	{}

返回：

类型	描述
ModelProviderProtocol	The resulting ModelProviderProtocol value.

引发：

类型	描述
RuntimeError	If the operation cannot be completed at runtime.

源代码位于： jianmu/model/client.py

@classmethod
def resolve_provider(
    cls,
    name: str | None = None,
    preference: list[str] | None = None,
    env_override: bool = False,
    **kwargs: Any,
) -> ModelProviderProtocol:
    """Resolve and instantiate the first usable provider.

    This returns the raw provider instance rather than a ``ModelClient``.
    Use it when framework code or examples need a provider object to inject
    into nodes, runtimes, or contexts.

    Args:
        name: Explicit provider name. When omitted, providers are selected
            from ``preference`` by available credentials.
        preference: Provider resolution order.
        env_override: Whether project ``.env`` values should override
            existing process environment variables.
        **kwargs: Extra provider-specific constructor arguments.

    Returns:
        The resulting `ModelProviderProtocol` value.

    Raises:
        RuntimeError: If the operation cannot be completed at runtime.
    """
    try:
        from jianmu.config.loader import load_project_env
    except Exception:
        load_project_env = None
    if load_project_env is not None:
        load_project_env(override=env_override)

    if name:
        return _create_provider(name, **kwargs)

    order = preference or ["openai", "litellm"]

    for provider_name in order:
        if _has_credentials(provider_name):
            try:
                return _create_provider(provider_name, **kwargs)
            except Exception:
                continue

    for provider_name in _REGISTRY:
        if provider_name not in order:
            try:
                return _create_provider(provider_name, **kwargs)
            except Exception:
                continue

    raise RuntimeError(
        "No model provider configured. Set OPENAI_API_KEY or API_KEY for the native "
        "OpenAI-compatible provider, or configure LiteLLM credentials such as "
        "OPENAI_API_KEY, OPENROUTER_API_KEY, ANTHROPIC_API_KEY, GOOGLE_API_KEY, "
        "GEMINI_API_KEY, XAI_API_KEY, DEEPSEEK_API_KEY, GROQ_API_KEY, or OLLAMA_API_BASE. "
        "Or register a custom provider with ModelClient.register_provider()."
    )

invoke async

invoke(
    messages: list[Message],
    config: ModelConfig,
    *,
    stream: bool = False,
    on_text_update: Callable[[str], None] | None = None,
    trace_node: str | None = None,
) -> tuple[Message, str]

Invoke the bound provider using Jianmu's normalized call path.

参数：

名称	类型	描述	默认
messages	list[Message]	Messages to process.	必需
config	ModelConfig	Configuration object for the operation.	必需
stream	bool	The stream value.	False
on_text_update	Callable[[str], None] | None	The on_text_update value.	None
trace_node	str | None	The trace_node value.	None

返回：

类型	描述
tuple[Message, str]	The resulting tuple value.

引发：

类型	描述
RuntimeError	If the operation cannot be completed at runtime.
last_exc	If the operation fails.

源代码位于： jianmu/model/client.py

async def invoke(
    self,
    messages: list[Message],
    config: ModelConfig,
    *,
    stream: bool = False,
    on_text_update: Callable[[str], None] | None = None,
    trace_node: str | None = None,
) -> tuple[Message, str]:
    """Invoke the bound provider using Jianmu's normalized call path.

    Args:
        messages: Messages to process.
        config: Configuration object for the operation.
        stream: The `stream` value.
        on_text_update: The `on_text_update` value.
        trace_node: The `trace_node` value.

    Returns:
        The resulting tuple value.

    Raises:
        RuntimeError: If the operation cannot be completed at runtime.
        last_exc: If the operation fails.
    """
    from jianmu.model.invoke import invoke_model
    messages = sanitize_messages_for_model_invoke(messages, trace_node=trace_node)
    max_retries = max(0, int(_resolve_max_retries(config)))
    last_exc: Exception | None = None
    last_exc_retryable = False

    for attempt in range(max_retries + 1):
        try:
            return await invoke_model(
                provider=self.provider,
                messages=messages,
                config=config,
                stream=stream,
                on_text_update=on_text_update,
                trace_node=trace_node,
            )
        except Exception as exc:
            last_exc = exc
            last_exc_retryable = _is_retryable_model_error(exc)
            if not last_exc_retryable or attempt >= max_retries:
                break
            logger.warning(
                "⚠️ [ModelClient] Retryable model error on attempt {}/{} for model '{}': {}",
                attempt + 1,
                max_retries + 1,
                config.model,
                exc,
            )

    fallback_model = _resolve_fallback_model(config)
    if (
        fallback_model
        and not config.disable_fallback
        and fallback_model != config.model
        and _is_same_provider_model_family(config.model, fallback_model)
        and last_exc is not None
        and last_exc_retryable
    ):
        logger.warning(
            "⚠️ [ModelClient] Falling back from model '{}' to '{}' after retry exhaustion.",
            config.model,
            fallback_model,
        )
        return await invoke_model(
            provider=self.provider,
            messages=messages,
            config=ModelConfig(
                model=fallback_model,
                temperature=config.temperature,
                top_p=config.top_p,
                top_k=config.top_k,
                max_tokens=config.max_tokens,
                timeout=config.timeout,
                tools=config.tools,
                tool_choice=config.tool_choice,
                strict_tools=config.strict_tools,
                response_format=config.response_format,
                max_retries=0,
                fallback_model=None,
                disable_fallback=True,
                extra=config.extra,
            ),
            stream=stream,
            on_text_update=on_text_update,
            trace_node=trace_node,
        )

    if last_exc is not None:
        raise last_exc
    raise RuntimeError("Model invocation failed without a captured exception.")

register_provider staticmethod

register_provider(
    name: str,
    provider: type[ModelProviderProtocol]
    | Callable[..., ModelProviderProtocol],
) -> None

Register a custom provider factory in the shared registry.

参数：

名称	类型	描述	默认
name	str	Name used by the operation.	必需
provider	type[ModelProviderProtocol] | Callable[..., ModelProviderProtocol]	The provider value.	必需

源代码位于： jianmu/model/client.py

@staticmethod
def register_provider(
    name: str,
    provider: type[ModelProviderProtocol] | Callable[..., ModelProviderProtocol],
) -> None:
    """Register a custom provider factory in the shared registry.

    Args:
        name: Name used by the operation.
        provider: The `provider` value.
    """
    _REGISTRY[name] = provider

unregister_provider staticmethod

unregister_provider(name: str) -> bool

Remove a custom provider factory from the shared registry.

参数：

名称	类型	描述	默认
name	str	Name used by the operation.	必需

返回：

类型	描述
bool	True if the operation succeeds; otherwise False.

源代码位于： jianmu/model/client.py

@staticmethod
def unregister_provider(name: str) -> bool:
    """Remove a custom provider factory from the shared registry.

    Args:
        name: Name used by the operation.

    Returns:
        True if the operation succeeds; otherwise False.
    """
    return _REGISTRY.pop(name, None) is not None

list_providers staticmethod

list_providers() -> list[str]

List builtin and custom provider names.

返回：

类型	描述
list[str]	The resulting list of values.

源代码位于： jianmu/model/client.py

@staticmethod
def list_providers() -> list[str]:
    """List builtin and custom provider names.

    Returns:
        The resulting list of values.
    """
    builtin = ["openai", "litellm"]
    custom = list(_REGISTRY.keys())
    return builtin + [n for n in custom if n not in builtin]

MessageEncoderProtocol

Bases: Protocol

Encode jianmu Message objects into provider-specific payloads.

encode_messages

encode_messages(
    messages: Sequence[Message],
) -> List[Dict[str, Any]]

Encode normalized messages into provider request payloads.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	A sequence of jianmu Message objects.	必需

返回：

类型	描述
List[Dict[str, Any]]	A list of dictionary payloads formatted for the provider API.

源代码位于： jianmu/model/codecs.py

def encode_messages(self, messages: Sequence[Message]) -> List[Dict[str, Any]]:
    """Encode normalized messages into provider request payloads.

    Args:
        messages: A sequence of jianmu Message objects.

    Returns:
        A list of dictionary payloads formatted for the provider API.
    """
    ...

OpenAIChatEncoder

Bases: MessageEncoderProtocol

Encoder for OpenAI-compatible chat providers.

encode_messages

encode_messages(
    messages: Sequence[Message],
) -> List[Dict[str, Any]]

Encode Jianmu messages into OpenAI-compatible chat payloads.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	A sequence of jianmu Message objects.	必需

返回：

类型	描述
List[Dict[str, Any]]	A list of dictionary payloads formatted for OpenAI API request.

源代码位于： jianmu/model/codecs.py

def encode_messages(self, messages: Sequence[Message]) -> List[Dict[str, Any]]:
    """Encode Jianmu messages into OpenAI-compatible chat payloads.

    Args:
        messages: A sequence of jianmu Message objects.

    Returns:
        A list of dictionary payloads formatted for OpenAI API request.
    """
    safe_messages = sanitize_tool_call_history(messages).messages
    encoded: List[Dict[str, Any]] = []
    for msg in self._normalize_system_roles(safe_messages):
        item: Dict[str, Any] = {
            "role": msg.role,
            "content": self._normalize_content(msg.content),
        }

        if msg.role == "assistant" and msg.tool_calls:
            converted_calls = []
            for idx, call in enumerate(msg.tool_calls):
                call_id = call.get("id") if isinstance(call, dict) else None
                call_name = call.get("name") if isinstance(call, dict) else ""
                call_args = call.get("arguments", "{}") if isinstance(call, dict) else "{}"
                if isinstance(call_args, dict):
                    call_args = json.dumps(call_args)
                elif call_args is None:
                    call_args = "{}"
                converted_calls.append(
                    {
                        "id": call_id or f"call_{idx}",
                        "type": "function",
                        "function": {"name": call_name, "arguments": call_args},
                    }
                )
            item["tool_calls"] = converted_calls

        if msg.role == "tool":
            tool_call_id = msg.tool_call_id
            if tool_call_id:
                item["tool_call_id"] = tool_call_id

        encoded.append(item)
    return encoded

LiteLLMChatEncoder

Bases: OpenAIChatEncoder

Encoder for LiteLLM completion payloads.

Keep parity with OpenAIChatEncoder so switching providers does not change Jianmu's message normalization semantics.

ProviderErrorInfo dataclass

ProviderErrorInfo(
    code: str,
    message: str,
    retryable: bool = False,
    user_fixable: bool = False,
)

Normalized provider/model error semantics.

require_model_client

require_model_client(
    model_client: Any, *, source: str
) -> "ModelClient"

Validate that a value is Jianmu's ModelClient facade type.

参数：

名称	类型	描述	默认
model_client	Any	Candidate value expected to be a ModelClient.	必需
source	str	Human-readable source label used in the validation error.	必需

返回：

类型	描述
'ModelClient'	The validated ModelClient instance.

引发：

类型	描述
TypeError	If model_client is not a ModelClient instance.

源代码位于： jianmu/model/client.py

def require_model_client(model_client: Any, *, source: str) -> "ModelClient":
    """Validate that a value is Jianmu's ``ModelClient`` facade type.

    Args:
        model_client: Candidate value expected to be a ``ModelClient``.
        source: Human-readable source label used in the validation error.

    Returns:
        The validated ``ModelClient`` instance.

    Raises:
        TypeError: If ``model_client`` is not a ``ModelClient`` instance.
    """
    if isinstance(model_client, ModelClient):
        return model_client
    raise TypeError(
        f"{source} must be a jianmu.model.ModelClient instance, "
        f"got {type(model_client).__name__}. Wrap providers with ModelClient(...)."
    )

classify_provider_exception

classify_provider_exception(
    exc: Exception,
) -> ProviderErrorInfo | None

Classify one provider/model exception into a stable error code.

参数：

名称	类型	描述	默认
exc	Exception	Provider- or model-facing exception raised during resolution or invocation.	必需

返回：

类型	描述
ProviderErrorInfo | None	A normalized ProviderErrorInfo when the exception matches a known
ProviderErrorInfo | None	provider/model failure category; otherwise None.

源代码位于： jianmu/model/errors.py

def classify_provider_exception(exc: Exception) -> ProviderErrorInfo | None:
    """Classify one provider/model exception into a stable error code.

    Args:
        exc: Provider- or model-facing exception raised during resolution or
            invocation.

    Returns:
        A normalized ``ProviderErrorInfo`` when the exception matches a known
        provider/model failure category; otherwise ``None``.
    """
    message = str(exc).strip() or exc.__class__.__name__
    lowered = message.lower()

    if "api key is invalid" in lowered or "error code: 401" in lowered:
        return ProviderErrorInfo(
            code="invalid_api_key",
            message=message,
            user_fixable=True,
        )
    if "balance is insufficient" in lowered or ("insufficient" in lowered and "balance" in lowered):
        return ProviderErrorInfo(
            code="insufficient_balance",
            message=message,
            user_fixable=True,
        )
    if "model not found" in lowered:
        return ProviderErrorInfo(
            code="model_not_found",
            message=message,
            user_fixable=True,
        )
    if "unknown provider" in lowered:
        return ProviderErrorInfo(
            code="unknown_provider",
            message=message,
            user_fixable=True,
        )
    if any(
        marker in lowered
        for marker in (
            "connection error",
            "api connection error",
            "failed to resolve",
            "name or service not known",
            "nodename nor servname provided",
            "connection reset",
            "connection aborted",
            "peer closed connection",
            "incomplete chunked read",
            "temporarily unavailable",
            "timeout",
            "timed out",
        )
    ):
        return ProviderErrorInfo(
            code="connection_error",
            message=message,
            retryable=True,
        )
    return None