`symfonic.services.models`¶

models ¶

Provider and model-resolution service (T3.1.1).

The one place that answers, for any provider and any call:

which wire dialect it speaks (:func:detect_provider_family) — and therefore which prompt-cache dialect applies (:func:cache_dialect_for);
what it can do (:func:describe_provider) — thinking, streaming, forced tool choice and the reason for a refusal, timeout kwargs, sampling knobs;
which model serves this call (:class:ModelResolutionService) — one precedence chain over per-dispatch hooks, role tables, caller defaults and the provider's own declaration, recording which of them won;
whether an adapter is well formed (:func:check_provider_contract).

Everything here is pure: no I/O, no client construction, no credential reads.

ModelResolutionService ¶

ModelResolutionService(provider: Any, *, default_config: ModelConfig | None = None, role_models: Mapping[str, ModelConfig] | None = None, dispatch_resolver: Any = None)

Resolve a role (and optionally one dispatch) to a model and its capabilities.

Precedence, first match wins:

dispatch_resolver(ctx) — consulted only when a dispatch context is supplied, so a service constructed with a resolver still answers static questions statically.
role_models[role] — the static per-role table.
default_config — the caller's explicit snapshot default.
the provider's own declaration (default_model_config / _symfonic_default_model), followed through routing wrappers.
ModelConfig() — the framework default.

Source code in src/symfonic/services/models/resolution.py

def __init__(
    self,
    provider: Any,
    *,
    default_config: ModelConfig | None = None,
    role_models: Mapping[str, ModelConfig] | None = None,
    dispatch_resolver: Any = None,
) -> None:
    self._provider = provider
    self._default_config = default_config
    # Snapshot then freeze: an adopter's live ``FrameworkConfig`` dict must
    # not be able to retune routing under a run that already started.
    self._role_models: Mapping[str, ModelConfig] = MappingProxyType(
        dict(role_models or {})
    )
    self._dispatch_resolver = dispatch_resolver

role_models `property` ¶

role_models: Mapping[str, ModelConfig]

Read-only view of the role table this service was built with.

resolve ¶

resolve(role: str = DEFAULT_ROLE, *, dispatch_context: Any = None) -> ResolvedModel

Resolve role — and, when a context is given, this dispatch.

Source code in src/symfonic/services/models/resolution.py

def resolve(
    self, role: str = DEFAULT_ROLE, *, dispatch_context: Any = None
) -> ResolvedModel:
    """Resolve ``role`` — and, when a context is given, this dispatch."""
    config, source = self._choose(role, dispatch_context)
    capabilities = describe_provider(self._provider, config)
    return ResolvedModel(
        config=config,
        role=role,
        source=source,
        family=capabilities.family,
        capabilities=capabilities,
    )

resolve_config ¶

resolve_config(role: str = DEFAULT_ROLE, *, dispatch_context: Any = None) -> ModelConfig

The config alone, for call sites that need nothing else.

Source code in src/symfonic/services/models/resolution.py

def resolve_config(
    self, role: str = DEFAULT_ROLE, *, dispatch_context: Any = None
) -> ModelConfig:
    """The config alone, for call sites that need nothing else."""
    return self._choose(role, dispatch_context)[0]

ProviderCapabilities `dataclass` ¶

ProviderCapabilities(label: str, family: ProviderFamily, cache_dialect: CacheDialect, thinking: ThinkingSupport, streaming: bool, forced_tool_choice: bool, forced_tool_choice_refusal: str | None, timeouts: TimeoutBinding, sampling: SamplingSupport)

Everything the framework knows about serving one call with one adapter.

Produced by describe_provider from the static descriptor row plus the adapter's own supports_* introspection for the specific ModelConfig in hand. Frozen because callers cache it per turn.

ProviderContractViolation `dataclass` ¶

ProviderContractViolation(provider: str, rule: str, message: str)

One way an adapter fails the contract.

ProviderDescriptor `dataclass` ¶

ProviderDescriptor(label: str, family: ProviderFamily, default_model: str | None, thinking: bool, streaming: bool, thinking_blocks_forced_tool_choice: bool, timeouts: TimeoutBinding, sampling: SamplingSupport)

The static half of a provider's capabilities.

default_model = None is an explicit abstention, not an omission: a gateway adapter (OpenRouter, Bedrock, the OAuth adapters) serves whatever model the caller names, so declaring one would be a guess. The contract checker requires the declaration to exist, which is what turns "nobody thought about it" into a reviewed decision.

ResolvedModel `dataclass` ¶

ResolvedModel(config: ModelConfig, role: str, source: ResolutionSource, family: ProviderFamily, capabilities: ProviderCapabilities)

One resolution, with the reason it came out that way.

source is the load-bearing field. A resolution that cannot say why it chose a model is indistinguishable from a resolution that ignored the caller's configuration, which is the class of bug this service exists to make impossible to ship silently.

SamplingSupport `dataclass` ¶

SamplingSupport(top_p: bool = True, top_k: bool = False)

Which sampling knobs survive to the wire for this adapter.

ThinkingSupport `dataclass` ¶

ThinkingSupport(enabled: bool = False, blocks_forced_tool_choice: bool = False)

Extended-thinking capability, and what enabling it costs.

blocks_forced_tool_choice records the published Anthropic constraint: with thinking enabled the API accepts only {type:auto} / {type:none} and returns HTTP 400 for a forced tool. It is a property of the adapter, evaluated per ModelConfig by capabilities.describe_provider.

TimeoutBinding `dataclass` ¶

TimeoutBinding(timeout_kwarg: str | None, http_client_kwarg: str | None = None, http_client_support: HttpClientSupport = 'unsupported')

Which LangChain constructor kwargs an adapter's timeouts thread through.

timeout_kwarg differs per adapter (default_request_timeout on Anthropic, timeout everywhere else) and is None for adapters that do not thread ModelConfig.timeout_seconds at construction at all (Bedrock puts request timeouts on the boto3 client). http_client_kwarg is None for adapters whose LangChain class builds its own transport.

cache_dialect_for ¶

cache_dialect_for(family: ProviderFamily) -> CacheDialect

The prompt-cache annotation dialect for a wire family.

Total over the closed family set, and derived rather than assigned: an adapter cannot claim cache_control support without claiming the Anthropic wire family that actually accepts it.

Source code in src/symfonic/services/models/descriptors.py

def cache_dialect_for(family: ProviderFamily) -> CacheDialect:
    """The prompt-cache annotation dialect for a wire family.

    Total over the closed family set, and derived rather than assigned: an
    adapter cannot claim ``cache_control`` support without claiming the
    Anthropic wire family that actually accepts it.
    """
    return _CACHE_DIALECTS.get(family, "none")

check_provider_contract ¶

check_provider_contract(target: Any) -> tuple[ProviderContractViolation, ...]

Check one provider class or instance against the adapter contract.

Accepts either, because adapters differ in whether they can be constructed without credentials. Given a class, the probes that need an instance run against object.__new__(cls) — no __init__, so no credential read, no network, no side effects. The Protocol already requires supports_forced_tool_choice to be a pure function of its ModelConfig argument, so an adapter that cannot answer from an uninitialised instance is itself outside the contract; those probes are skipped rather than guessed at, and the structural checks still apply.

Source code in src/symfonic/services/models/contract.py

def check_provider_contract(target: Any) -> tuple[ProviderContractViolation, ...]:
    """Check one provider class or instance against the adapter contract.

    Accepts either, because adapters differ in whether they can be constructed
    without credentials. Given a class, the probes that need an instance run
    against ``object.__new__(cls)`` — no ``__init__``, so no credential read,
    no network, no side effects. The Protocol already requires
    ``supports_forced_tool_choice`` to be a pure function of its ``ModelConfig``
    argument, so an adapter that cannot answer from an uninitialised instance
    is itself outside the contract; those probes are skipped rather than
    guessed at, and the structural checks still apply.
    """
    cls = target if isinstance(target, type) else type(target)
    name = cls.__name__
    violations: list[ProviderContractViolation] = []

    for method in _REQUIRED_METHODS:
        if not callable(getattr(cls, method, None)):
            violations.append(
                ProviderContractViolation(
                    name,
                    "missing-method",
                    f"does not implement {method}(); the ModelProvider Protocol "
                    "requires it",
                )
            )

    violations.extend(_check_declarations(cls, name))

    instance = target if not isinstance(target, type) else _uninitialised(cls)
    if instance is not None:
        violations.extend(_check_probes(instance, name))
    return tuple(violations)

describe_provider ¶

describe_provider(provider: object, resolved_config: Any = None) -> ProviderCapabilities

Describe how provider will serve resolved_config.

Pure: no I/O, no provider-state mutation, safe to call every turn. The family and descriptor come from the leaf that will actually serve the config (routing wrappers are followed); the supports_* probes are put to the original provider, because a router knows how to delegate them and a leaf does not know it is being routed to.

Source code in src/symfonic/services/models/capabilities.py

def describe_provider(
    provider: object, resolved_config: Any = None
) -> ProviderCapabilities:
    """Describe how ``provider`` will serve ``resolved_config``.

    Pure: no I/O, no provider-state mutation, safe to call every turn. The
    family and descriptor come from the leaf that will actually serve the
    config (routing wrappers are followed); the ``supports_*`` probes are put
    to the *original* provider, because a router knows how to delegate them
    and a leaf does not know it is being routed to.
    """
    row = descriptor_for_provider(provider, resolved_config)
    family = detect_provider_family(provider, resolved_config)
    config = resolved_config if resolved_config is not None else _default_config()

    # A gateway adapter reassigns its own family from the model it is actually
    # serving -- ``AWSBedrockProvider`` rewrites ``_symfonic_provider_family``
    # on every ``get_chat_model`` call -- so the row reached by class name can
    # describe a different wire than this call will use. Family-derived facts
    # (thinking, and whether thinking blocks a forced tool choice) follow the
    # DETECTED family; adapter-specific facts (label, timeouts, sampling) stay
    # on the row, because those are properties of the client being built and
    # not of the model being served. Without this split one
    # ``ProviderCapabilities`` reports two providers: Bedrock serving a Llama
    # model would carry ``family="openai"`` next to Anthropic's
    # extended-thinking constraint.
    served = row if row.family == family else descriptor_for_family(family)
    blocks_forced = served.thinking_blocks_forced_tool_choice

    forced = _probe_forced_tool_choice(provider, config)
    return ProviderCapabilities(
        label=row.label,
        family=family,
        cache_dialect=cache_dialect_for(family),
        thinking=ThinkingSupport(
            enabled=_probe_bool(provider, "supports_thinking", served.thinking),
            blocks_forced_tool_choice=blocks_forced,
        ),
        streaming=_probe_bool(provider, "supports_streaming", row.streaming),
        forced_tool_choice=forced,
        forced_tool_choice_refusal=(
            None
            if forced
            else _probe_refusal(provider, config, row.label, blocks_forced)
        ),
        timeouts=row.timeouts,
        sampling=row.sampling,
    )

descriptor_for_family ¶

descriptor_for_family(family: ProviderFamily) -> ProviderDescriptor

The fallback row for a family. Total over the closed family set.

Source code in src/symfonic/services/models/descriptors.py

def descriptor_for_family(family: ProviderFamily) -> ProviderDescriptor:
    """The fallback row for a family. Total over the closed family set."""
    return PROVIDER_DESCRIPTORS[_FAMILY_FALLBACK.get(family, "unknown")]

descriptor_for_provider ¶

descriptor_for_provider(provider: object, resolved_config: Any = None) -> ProviderDescriptor

Best row for provider, descending routing wrappers like the classifier.

Shipped adapters match by class name; everything else falls back to the row for the family detect_provider_family assigns, so an adopter's own provider still gets a coherent (if generic) capability answer.

Source code in src/symfonic/services/models/descriptors.py

def descriptor_for_provider(
    provider: object, resolved_config: Any = None
) -> ProviderDescriptor:
    """Best row for ``provider``, descending routing wrappers like the classifier.

    Shipped adapters match by class name; everything else falls back to the row
    for the family ``detect_provider_family`` assigns, so an adopter's own
    provider still gets a coherent (if generic) capability answer.
    """
    from symfonic.services.models.family import detect_provider_family, leaf_provider

    leaf = leaf_provider(provider, resolved_config)
    label = PROVIDER_LABELS_BY_CLASS.get(type(leaf).__name__)
    if label is not None:
        return PROVIDER_DESCRIPTORS[label]
    return descriptor_for_family(detect_provider_family(provider, resolved_config))

detect_provider_family ¶

detect_provider_family(provider: object, resolved_config: Any = None) -> ProviderFamily

Classify provider into the content-block dialect it expects.

resolved_config is optional and, when given, decides routing wrappers: the family returned is the family of the leaf that will serve that config, not the wrapper's default leaf.

Source code in src/symfonic/services/models/family.py

def detect_provider_family(
    provider: object,
    resolved_config: Any = None,
) -> ProviderFamily:
    """Classify ``provider`` into the content-block dialect it expects.

    ``resolved_config`` is optional and, when given, decides routing wrappers:
    the family returned is the family of the leaf that will serve *that*
    config, not the wrapper's default leaf.
    """
    seen: set[int] = set()
    current: Any = provider
    for _ in range(_MAX_DEPTH):
        if id(current) in seen:
            break
        seen.add(id(current))

        routed = _route_leaf(current, resolved_config)
        if routed is not current:
            # A route decision is a leaf decision: re-enter classification on
            # the picked provider with no config, so a router wrapping a router
            # cannot loop on the same pick forever.
            return detect_provider_family(routed, None)

        declared = _declared_family(current)
        if declared is not None:
            return declared

        inner = getattr(current, "_default", None) or getattr(current, "default", None)
        if inner is None or inner is current:
            break
        current = inner
    return "unknown"

leaf_provider ¶

leaf_provider(provider: Any, resolved_config: Any = None) -> Any

The provider that will actually serve resolved_config.

Follows route maps first (when a config is supplied), then plain wrapper _default / default chains, under the same depth bound and cycle guard as :func:detect_provider_family.

Source code in src/symfonic/services/models/family.py

def leaf_provider(provider: Any, resolved_config: Any = None) -> Any:
    """The provider that will actually serve ``resolved_config``.

    Follows route maps first (when a config is supplied), then plain wrapper
    ``_default`` / ``default`` chains, under the same depth bound and cycle
    guard as :func:`detect_provider_family`.
    """
    seen: set[int] = set()
    current: Any = provider
    for _ in range(_MAX_DEPTH):
        if id(current) in seen:
            break
        seen.add(id(current))

        routed = _route_leaf(current, resolved_config)
        if routed is not current:
            current = routed
            continue
        if _declared_family(current) is not None:
            return current

        inner = getattr(current, "_default", None) or getattr(current, "default", None)
        if inner is None or inner is current:
            break
        current = inner
    return current

provider_default_config ¶

provider_default_config(provider: Any) -> ModelConfig

The ModelConfig a provider declares for callers that name none.

Resolution order, first match wins:

provider.default_model_config — a ModelConfig or a zero-argument callable returning one. The full seam: model, temperature and token budget pinned in one place.
provider._symfonic_default_model — a model-name string. The cheap seam, and the one every shipped adapter declares.
ModelConfig() — the framework default.

Routing wrappers are followed through _default / default, so a router wrapping one provider inherits that provider's declaration.

This is the function symfonic.agent.backend.model.resolve_model_config now is: the simple Agent facade has no model= parameter, so without it Agent(OpenAIProvider()) would ask OpenAI for claude-sonnet-4-5 and fail at request time with model_not_found.

Source code in src/symfonic/services/models/resolution.py

def provider_default_config(provider: Any) -> ModelConfig:
    """The ``ModelConfig`` a provider declares for callers that name none.

    Resolution order, first match wins:

    1. ``provider.default_model_config`` — a ``ModelConfig`` or a zero-argument
       callable returning one. The full seam: model, temperature and token
       budget pinned in one place.
    2. ``provider._symfonic_default_model`` — a model-name string. The cheap
       seam, and the one every shipped adapter declares.
    3. ``ModelConfig()`` — the framework default.

    Routing wrappers are followed through ``_default`` / ``default``, so a
    router wrapping one provider inherits that provider's declaration.

    This is the function ``symfonic.agent.backend.model.resolve_model_config``
    now *is*: the simple ``Agent`` facade has no ``model=`` parameter, so
    without it ``Agent(OpenAIProvider())`` would ask OpenAI for
    ``claude-sonnet-4-5`` and fail at request time with ``model_not_found``.
    """
    return _declared_default(provider) or ModelConfig()

symfonic.services.models¶

models ¶

ModelResolutionService ¶

role_models property ¶

resolve ¶

resolve_config ¶

ProviderCapabilities dataclass ¶

ProviderContractViolation dataclass ¶

ProviderDescriptor dataclass ¶

ResolvedModel dataclass ¶

SamplingSupport dataclass ¶

ThinkingSupport dataclass ¶

TimeoutBinding dataclass ¶

cache_dialect_for ¶

check_provider_contract ¶

describe_provider ¶

descriptor_for_family ¶

descriptor_for_provider ¶

detect_provider_family ¶

leaf_provider ¶

provider_default_config ¶

`symfonic.services.models`¶

role_models `property` ¶

ProviderCapabilities `dataclass` ¶

ProviderContractViolation `dataclass` ¶

ProviderDescriptor `dataclass` ¶

ResolvedModel `dataclass` ¶

SamplingSupport `dataclass` ¶

ThinkingSupport `dataclass` ¶

TimeoutBinding `dataclass` ¶