异步客户端
使用 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 客户端。
显式选项优先于环境变量;为空或仅含空白字符的环境变量值将被忽略。
日志设置
SDK 会将日志记录到 typesafe_sdk logger;可通过标准 logging 进行配置,或设置 TYPESAFE_LOG_LEVEL(debug、info 等)以快速使用默认配置。敏感请求头会从日志输出中脱敏;请求体与响应体则不会。
参数:
api_key(str | None, default:None) –必需的 API key;可通过
TYPESAFE_API_KEY环境变量设置。会去除首尾空白字符。空 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环境变量设置。
引发:
-
API key 缺失或无效,或超时设置无效。
-
同时提供了
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
</pre></div></div><div class="tab-pane" data-title="Overload 2"> <div class="sdk-signature"><pre>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</pre></div></div></div>针对文本或结构化状态回答命名的问题。
详见 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之后浅合并到请求体上。合并采用后写优先:与state、model或questions冲突的键会将其覆盖,对象值会被替换而非深度合并。response_model(type[ResponseT] | None, default:None) –可选的 Pydantic
BaseModel类型,用于描述 JSON 响应体,包括所有嵌套的答案模型。
返回:
SystemOneResponse | ResponseT–response_model的实例;或在未提供自定义模型时,返回以问题SystemOneResponse | ResponseT–名称为键的答案以及模型与 token 用量信息的
SystemOneResponse。
引发:
-
问题为空,或 Score 类型问题的标准列表为空。
-
在所有重试之后,服务器仍返回不成功的 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()