非同步客戶端
用 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()