异步客户端
异步客户端
用 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环境变量设置。
抛出:
-
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。
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 用量详情。
抛出:
-
问题为空,或某个 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
列出该账号可用的模型。
Parameters:
-
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()