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 客户端。
显式传入的选项优先于环境变量;为空或仅含空白字符的环境变量值会被忽略。
日志配置
SDK 通过 typesafe_sdk 记录器输出日志;可通过标准日志机制进行配置,也可以设置 TYPESAFE_LOG_LEVEL(debug、info、……)作为快速默认值。敏感头部会在日志输出中脱敏;请求和响应主体不会。
参数:
-
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时继承其timeout,否则使用 SDK 默认值。 -
headers(Mapping[str, str] | None, default:None) –要设置的额外请求头。
-
transport(httpx2.AsyncBaseTransport | None, default:None) –可选的自定义 HTTP 传输层,随本 SDK 客户端关闭而关闭。
-
http_client(httpx2.AsyncClient | None, default:None) –可选的
httpx2.AsyncClient;与transport互斥。随本 SDK 客户端关闭而关闭。 -
base_url(str | None, default:None) –API 根地址;可通过
TYPESAFE_BASE_URL环境变量设置。
抛出:
-
API 密钥缺失或无效,或超时时间无效。
-
同时提供了
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。
参数:
-
state(JSONContent) –要评估的文本、JSON 对象或数组。详情参见状态。
-
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之后浅合并到请求体上。合并遵循后写优先(last-write-wins):与state、model或questions冲突的键会覆盖对应值,对象值会被整体替换而非深度合并。 -
response_model(type[ResponseT] | None, default:None) –可选的 Pydantic
BaseModel类型,用于描述 JSON 响应体,包括任何嵌套的答案模型。
返回值:
-
SystemOneResponse | ResponseT–response_model的实例;或SystemOneResponse,其答案以问题名称为键, -
SystemOneResponse | ResponseT–并在未提供自定义模型时包含模型和 token 用量详情。
抛出:
-
questions 为空,或 Score 问题的 criteria 列表为空。
-
服务器在所有重试之后仍返回不成功的 HTTP 响应。
-
请求在所有重试之后仍无法连接或超时。
-
TypeSafeAPIResponseValidationError–响应体与响应模型不匹配。
示例:
使用具名参数创建问题:
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
列出账户可用的模型。
参数:
-
retry(RetryPolicy | None, default:None) –可选的重试策略,仅对本次调用覆盖客户端级别的值。
-
timeout(float | httpx2.Timeout | None, default:None) –单次操作的超时覆盖值;
None继承客户端设置。 -
extra_headers(Mapping[str, str] | None, default:None) –额外请求头的覆盖值;身份验证、SDK 标识和
Accept仍受保护。
返回值:
-
一个
ListModelsResponse,其models包含每个模型的名称、描述 -
和发布日期。
抛出:
-
服务器在所有重试之后仍返回不成功的 HTTP 响应。
-
请求在所有重试之后仍无法连接或超时。
示例:
from typesafe_sdk import AsyncTypeSafeClient
async def main() -> None:
async with AsyncTypeSafeClient() as client:
models = await client.models.list()