异步客户端
机器翻译译自英文(docs.typesafe.ai)本页 57 段已全部译完
文档索引
在以下地址获取完整的文档索引:https://docs.typesafe.ai/llms.txt 使用此文件可以在进一步探索之前发现所有可用页面。
异步客户端
使用 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 客户端。
显式选项优先于环境变量;为空或仅包含空白字符的环境变量值将被忽略。
Tip
日志配置
SDK 会将日志记录到 typesafe_sdk 日志器;你可以通过标准日志配置进行设置,或者设置 TYPESAFE_LOG_LEVEL(debug、info 等)以快速定义默认值。日志输出中的机密标头会被隐藏;请求和响应主体则不会。
参数:
-
api_key(str | None, 默认:None) –必需的 API 密钥;可通过
TYPESAFE_API_KEY环境变量设置。 -
model(str | None, 默认:None) –模型名称;可通过
TYPESAFE_DEFAULT_MODEL环境变量设置。 -
retry(RetryPolicy | None, 默认:None) –控制重试行为的
RetryPolicy;请参阅RetryPolicy了解可用选项及其默认值。传递RetryPolicy(max_retries=0)以禁用重试。 -
timeout(float | httpx2.Timeout | None, 默认:None) –HTTP 操作的超时时间。如果提供,则继承
http_client.timeout,否则使用 SDK 默认值。 -
headers(Mapping[str, str] | None, 默认:None) –要设置的额外请求标头。
-
transport(httpx2.AsyncBaseTransport | None, 默认:None) –可选的自定义 HTTP 传输,在此 SDK 客户端关闭时关闭。
-
http_client(httpx2.AsyncClient | None, 默认:None) –可选的
httpx2.AsyncClient;与transport互斥。在此 SDK 客户端关闭时关闭。 -
base_url(str | None, 默认: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: AsyncModelsModels 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 | ResponseTsystem_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,
) -> SystemOneResponsesystem_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 对象或数组。请参阅 state 了解详情。
-
questions(Mapping[str, Question]) –将名称映射到问题对象或原始字典的非空映射。
-
model(str | None, 默认:None) –模型覆盖;
None将继承客户端默认值。 -
retry(RetryPolicy | None, 默认:None) –可选的重试策略,仅用于覆盖此调用的客户端级值。
-
timeout(float | httpx2.Timeout | None, 默认:None) –HTTP 操作的可选超时,单位秒,仅用于覆盖此调用的客户端级值。
-
extra_headers(Mapping[str, str] | None, 默认:None) –要设置的额外请求标头。
-
extra_body(Mapping[str, JSONValue | None] | None, 默认:None) –额外的顶级请求主体字段,在设置
state、model和questions后与主体浅合并。合并规则为后者优先:与state、model或questions冲突的键将覆盖它们,对象值将被替换而非深合并。 -
response_model(type[ResponseT] | None, 默认:None) –可选的 Pydantic
BaseModel类型,用于描述 JSON 响应主体,包括任何嵌套的答案模型。
返回:
-
SystemOneResponse | ResponseT–response_model的实例,或者答案按问题索引的SystemOneResponse。 -
SystemOneResponse | ResponseT–未提供自定义模型时,包含名称、模型和 token 使用详情。
引发:
-
问题为空,或者评分问题的评分标准列表为空。
-
服务器在任何重试后返回不成功的 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, 默认值:None) –可选的重试策略,仅覆盖此次调用的客户端级值。
-
timeout(float | httpx2.Timeout | None, 默认值:None) –每次操作的超时覆盖;
None表示继承客户端设置。 -
extra_headers(Mapping[str, str] | None, 默认值: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()