文档导航

异步客户端

异步客户端

用 AsyncTypeSafeClient 提问、列出模型,并配置异步的 TypeSafe API 请求。

typesafe_sdk.AsyncTypeSafeClient

AsyncTypeSafeClient(
    *,
    api_key: str | None = None,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float
    | httpx2.Timeout
    | None = None,
    headers: Mapping[str, str] | None = None,
    transport: httpx2.AsyncBaseTransport
    | None = None,
    http_client: httpx2.AsyncClient
    | None = None,
    base_url: str | None = None,
)

为 TypeSafe AI API 创建一个异步 HTTP 客户端。

显式选项优先于环境变量;留空或只有空白字符的环境变量值会被忽略。

参数:

  • api_key (str | None, default: None ) –

    必需的 API 密钥;可通过 TYPESAFE_API_KEY 环境变量设置。首尾空白会被去除。空密钥、内部空白、控制字符和非 ASCII 字符都会被拒绝。

  • model (str | None, default: None ) –

    模型名;可通过 TYPESAFE_DEFAULT_MODEL 环境变量设置。

  • retry (RetryPolicy | None, default: None ) –

    一个控制重试行为的 RetryPolicy;可选参数及其默认值见 RetryPolicy。传 RetryPolicy(max_retries=0) 可关闭重试。

  • timeout (float | httpx2.Timeout | None, default: None ) –

    HTTP 操作的超时。传入 HTTP client 时继承它的 http_client.timeout,否则用 SDK 默认值。

  • headers (Mapping[str, str] | None, default: None ) –

    要额外设置的请求头。

  • transport (httpx2.AsyncBaseTransport | None, default: None ) –

    可选的自定义 HTTP transport,在此 SDK 客户端关闭时一并关闭。

  • http_client (httpx2.AsyncClient | None, default: None ) –

    可选的 httpx2.AsyncClient;与 transport 互斥。在此 SDK 客户端关闭时一并关闭。

  • base_url (str | None, default: None ) –

    API 根地址;可通过 TYPESAFE_BASE_URL 环境变量设置。

抛出:

  • TypeSafeError –

    API 密钥缺失或非法,或者超时值非法。

  • ValueError –

    同时传入了 transport 和 http_client。

示例:

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            state="I was charged twice. Please help.",
            questions={
                "billing": Noul(instructions="Is this about billing?"),
                "tone": Choice(
                    instructions="What is the tone?",
                    criteria={"calm": None, "angry": None},
                ),
            },
        )
        assert 0 <= result.nouls["billing"].noul <= 1
        assert result.choices["tone"].choice in {"calm", "angry"}

asyncio.run(main())

models

cached property

models: AsyncModels

Models API 资源的访问入口。

示例:

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        models = await client.models.list()

system_one

async

system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float
    | httpx2.Timeout
    | None = None,
    extra_headers: Mapping[str, str]
    | None = None,
    extra_body: Mapping[str, JSONValue | None]
    | None = None,
    response_model: type[ResponseT]
    | None = None,
) -> SystemOneResponse | ResponseT
system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float
    | httpx2.Timeout
    | None = None,
    extra_headers: Mapping[str, str]
    | None = None,
    extra_body: Mapping[str, JSONValue | None]
    | None = None,
    response_model: None = None,
) -> SystemOneResponse
system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float
    | httpx2.Timeout
    | None = None,
    extra_headers: Mapping[str, str]
    | None = None,
    extra_body: Mapping[str, JSONValue | None]
    | None = None,
    response_model: type[ResponseT],
) -> ResponseT

针对文本或结构化状态回答具名问题。

详见 System One。

Parameters:

  • state (JSONContent) –

    待求值的文本、JSON 对象或数组。详见 state。

  • questions (Mapping[str, Question]) –

    非空的映射,把名字对应到问题对象或原始字典。

  • model (str | None, default: None ) –

    模型覆盖项;为 None 时继承客户端默认值。

  • retry (RetryPolicy | None, default: None ) –

    可选的重试策略,仅对本次调用覆盖客户端级别的取值。

  • timeout (float | httpx2.Timeout | None, default: None ) –

    可选的 http 操作超时,仅对本次调用覆盖客户端级别的取值,单位为秒。

  • extra_headers (Mapping[str, str] | None, default: None ) –

    要额外设置的请求头。

  • extra_body (Mapping[str, JSONValue | None] | None, default: None ) –

    额外的顶层请求体字段,在设置好 state、model 和 questions 之后浅合并到请求体上。合并遵循后写覆盖:与 state、model 或 questions 冲突的键会覆盖它们,对象值整体替换、不做深合并。

  • response_model (type[ResponseT] | None, default: None ) –

    可选的 Pydantic BaseModel 类型,用于描述 JSON 响应体,包括其中嵌套的答案模型。

返回值:

  • SystemOneResponse | ResponseT –

    response_model 的实例;未提供自定义模型时,则是按问题名索引答案的 SystemOneResponse,

  • SystemOneResponse | ResponseT –

    其中还包含模型与 token 用量详情。

抛出:

示例:

用具名参数创建问题:

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            state="I was charged twice. Please help.",
            questions={
                "billing": Noul(instructions="Is this about billing?"),
                "tone": Choice(
                    instructions="What is the tone?",
                    criteria={"calm": None, "angry": None},
                ),
            },
        )
        assert 0 <= result.nouls["billing"].noul <= 1
        assert result.choices["tone"].choice in {"calm", "angry"}

把问题作为字典传入:

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            state={"message": "I was charged twice. Please help."},
            questions={
                "billing": {"type": "noul", "instructions": "Is this about billing?"},
                "tone": {
                    "type": "choice",
                    "instructions": "What is the tone?",
                    "criteria": {"calm": None, "angry": None},
                },
            },
        )
        assert 0 <= result.nouls["billing"].noul <= 1
        assert result.choices["tone"].choice in {"calm", "angry"}

aclose

async

aclose() -> None

释放网络资源并关闭底层 HTTP 客户端,包括外部传入的那个。

Models 资源

通过 AsyncTypeSafeClient.models 访问。

typesafe_sdk.AsyncModels

访问该账号可用的模型,通过 AsyncTypeSafeClient.models 进入。

list

async

list(
    *,
    retry: RetryPolicy | None = None,
    timeout: float
    | httpx2.Timeout
    | None = None,
    extra_headers: Mapping[str, str]
    | None = None,
) -> ListModelsResponse

列出该账号可用的模型。

Parameters:

  • retry (RetryPolicy | None, default: None ) –

    可选的重试策略,仅对本次调用覆盖客户端级别的取值。

  • timeout (float | httpx2.Timeout | None, default: None ) –

    按操作覆盖的超时;None 表示继承客户端设置。

  • extra_headers (Mapping[str, str] | None, default: None ) –

    对额外请求头的覆盖;身份认证、SDK 标识和 Accept 仍受保护。

返回值:

抛出:

示例:

from typesafe_sdk import AsyncTypeSafeClient

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        models = await client.models.list()