文档导航

用法

用法

使用 TypeSafe Python SDK 的指南与模式。

调用 System One API

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul, Score

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            "I was charged twice. Please help ASAP.",
            {
                "billing": Noul(instructions="Is this about billing?"),
                "tone": Choice(
                    instructions="What is the tone?",
                    criteria={"calm": None, "angry": None},
                ),
                "urgency": Score(
                    instructions="How urgent is this?",
                    criteria=["low", "medium", "high"],
                ),
            },
        )
        print(
            result.nouls["billing"].noul,
            result.choices["tone"].choice,
            result.scores["urgency"].score,
        )

asyncio.run(main())
from typesafe_sdk import Choice, Noul, Score, TypeSafeClient

client = TypeSafeClient()
state = "I was charged twice. Please help ASAP."
questions = {
    "billing": Noul(instructions="Is this about billing?"),
    "tone": Choice(
        instructions="What is the tone?", criteria={"calm": None, "angry": None}
    ),
    "urgency": Score(
        instructions="How urgent is this?", criteria=["low", "medium", "high"]
    ),
}
result = client.system_one(state, questions)
print(
    result.nouls["billing"].noul,
    result.choices["tone"].choice,
    result.scores["urgency"].score,
)

带类型的 system_one 响应

可以给 system_one 传入一个响应模型,让响应用起来更 type-safe(类型安全):

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Noul, NoulAnswer, SystemOneResponse

class BillingResponse(SystemOneResponse):
    billing: NoulAnswer

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
            response_model=BillingResponse,
        )
        assert 0 <= result.billing.noul <= 1
        assert result.billing == result.nouls["billing"]
        print(result.request_id)

asyncio.run(main())
from typesafe_sdk import Noul, NoulAnswer, SystemOneResponse, TypeSafeClient

class BillingResponse(SystemOneResponse):
    billing: NoulAnswer

with TypeSafeClient() as client:
    result = client.system_one(
        "I was charged twice.",
        {"billing": Noul(instructions="Is this about billing?")},
        response_model=BillingResponse,
    )
    assert 0 <= result.billing.noul <= 1
    assert result.billing == result.nouls["billing"]
    print(result.request_id)

自定义响应类型

也可以完全从头定义一个全新的响应模型,不必继承 SystemOneResponse:

import asyncio

from pydantic import BaseModel

from typesafe_sdk import AsyncTypeSafeClient, Noul, NoulAnswer

class BillingAnswers(BaseModel):
    billing: NoulAnswer

class BillingResponse(BaseModel):
    answers: BillingAnswers

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
            response_model=BillingResponse,
        )
        assert 0 <= result.answers.billing.noul <= 1

asyncio.run(main())
from pydantic import BaseModel

from typesafe_sdk import Noul, NoulAnswer, TypeSafeClient

class BillingAnswers(BaseModel):
    billing: NoulAnswer

class BillingResponse(BaseModel):
    answers: BillingAnswers

result = TypeSafeClient().system_one(
    "I was charged twice.",
    {"billing": Noul(instructions="Is this about billing?")},
    response_model=BillingResponse,
)
assert 0 <= result.answers.billing.noul <= 1

选择模型

查看可用模型:

import asyncio

from typesafe_sdk import AsyncTypeSafeClient

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        print(await client.models.list())

asyncio.run(main())
from typesafe_sdk import TypeSafeClient

print(TypeSafeClient().models.list())

在构造客户端时选择模型:

client = AsyncTypeSafeClient(model="jev")
client = TypeSafeClient(model="jev")

详见 Models 资源参考。

配置 base URL

要用 SDK 连到另一个 API 地址,可以在客户端上设置 base_url,或设置 TYPESAFE_BASE_URL 环境变量。这要求替代的 API 遵循 TypeSafe OpenAPI 规范。

例如,通过某个 AI 网关连接,使用它自己的 API key 和模型 ID。

使用 OpenRouter 的 API key 和一个 OpenRouter 模型 ID:

import asyncio
import os

from typesafe_sdk import AsyncTypeSafeClient, Noul

async def main() -> None:
    async with AsyncTypeSafeClient(
        api_key=os.environ["OPENROUTER_API_KEY"],
        base_url="https://openrouter.ai/api",
        model="~typesafe/jev-latest",
    ) as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
        )
        print(result.nouls["billing"].noul)

asyncio.run(main())
import os

from typesafe_sdk import Noul, TypeSafeClient

with TypeSafeClient(
    api_key=os.environ["OPENROUTER_API_KEY"],
    base_url="https://openrouter.ai/api",
    model="~typesafe/jev-latest",
) as client:
    result = client.system_one(
        "I was charged twice.",
        {"billing": Noul(instructions="Is this about billing?")},
    )
    print(result.nouls["billing"].noul)

Vercel 提供的兼容 TypeSafe 的 API 可以与 SDK 一起使用:

import asyncio
import os

from typesafe_sdk import AsyncTypeSafeClient, Noul

async def main() -> None:
    async with AsyncTypeSafeClient(
        api_key=os.environ["AI_GATEWAY_API_KEY"],
        base_url="https://ai-gateway.vercel.sh/typesafe",
        model="typesafe-ai/jev",
    ) as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
        )
        print(result.nouls["billing"].noul)

asyncio.run(main())
import os

from typesafe_sdk import Noul, TypeSafeClient

with TypeSafeClient(
    api_key=os.environ["AI_GATEWAY_API_KEY"],
    base_url="https://ai-gateway.vercel.sh/typesafe",
    model="typesafe-ai/jev",
) as client:
    result = client.system_one(
        "I was charged twice.",
        {"billing": Noul(instructions="Is this about billing?")},
    )
    print(result.nouls["billing"].noul)

使用 Pydantic AI Gateway 的 API key:

import asyncio
import os

from typesafe_sdk import AsyncTypeSafeClient, Noul

async def main() -> None:
    async with AsyncTypeSafeClient(
        api_key=os.environ["PYDANTIC_AI_GATEWAY_API_KEY"],
        base_url="https://gateway-us.pydantic.dev/proxy/typesafe",
        model="jev-latest",
    ) as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
        )
        print(result.nouls["billing"].noul)

asyncio.run(main())
import os

from typesafe_sdk import Noul, TypeSafeClient

with TypeSafeClient(
    api_key=os.environ["PYDANTIC_AI_GATEWAY_API_KEY"],
    base_url="https://gateway-us.pydantic.dev/proxy/typesafe",
    model="jev-latest",
) as client:
    result = client.system_one(
        "I was charged twice.",
        {"billing": Noul(instructions="Is this about billing?")},
    )
    print(result.nouls["billing"].noul)

HTTP/2

import httpx2

from typesafe_sdk import AsyncTypeSafeClient

client = AsyncTypeSafeClient(http_client=httpx2.AsyncClient(http2=True))
import httpx2

from typesafe_sdk import TypeSafeClient

client = TypeSafeClient(http_client=httpx2.Client(http2=True))

重试

把一个自定义的 RetryPolicy 作为 retry 传给客户端,或按单次调用传入。无效的 API key 会在创建客户端时就抛出 TypeSafeError,早于任何请求或重试。

在客户端上:

from typesafe_sdk import AsyncTypeSafeClient, RetryPolicy

client = AsyncTypeSafeClient(
    retry=RetryPolicy(max_retries=3, backoff_max=0.2, timeout=1.0)
)
from typesafe_sdk import RetryPolicy, TypeSafeClient

client = TypeSafeClient(retry=RetryPolicy(max_retries=3, backoff_max=0.2, timeout=1.0))

按单次调用:

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, RetryPolicy

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        await client.system_one(
            state,
            questions,
            retry=RetryPolicy(max_retries=3, backoff_max=0.2, timeout=1.0),
        )

asyncio.run(main())
from typesafe_sdk import RetryPolicy

client.system_one(
    state, questions, retry=RetryPolicy(max_retries=3, backoff_max=0.2, timeout=1.0)
)

错误处理

处理 SDK 抛出的异常:

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, TypeSafeAPIError

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        try:
            await client.system_one(state, questions)
        except TypeSafeAPIError as error:
            print(error.status, error.request_id)

asyncio.run(main())
from typesafe_sdk import TypeSafeAPIError

try:
    client.system_one(state, questions)
except TypeSafeAPIError as error:
    print(error.status, error.request_id)

日志

SDK 会向 typesafe_sdk 这个 logger 记录日志。按标准 logging 指南来配置它:

import logging

logging.getLogger("typesafe_sdk").setLevel(logging.DEBUG)

或者在导入 SDK 之前,把 TYPESAFE_LOG_LEVEL 设为 debug、info、warning、error 或 off 之一。

info 会为每次请求记录一行摘要;debug 还会记录请求和响应的头与请求体。机密头 —— 授权信息、API key、cookie,以及任何名字里含 token 或 secret 的头 —— 都会在日志输出中被脱敏。请求体和响应体不会脱敏。

环境变量

SDK 会读取并使用以下环境变量:

变量 配置项 默认值
TYPESAFE_API_KEY API key(必填) —
TYPESAFE_BASE_URL API 根 URL https://api.typesafe.ai
TYPESAFE_DEFAULT_MODEL 默认模型 jev-latest
TYPESAFE_LOG_LEVEL typesafe_sdk logger 的级别,在导入时应用一次 未设置

SDK 的默认值见常量参考。

通过 api_key 或 TYPESAFE_API_KEY 提供的 API key 会去掉首尾空白,包括来自 key 文件的换行。空 key、内部空白、控制字符和非 ASCII 字符会在发送请求之前被拒绝。显式传入的空 key 不会回退到环境变量。

前向兼容

随着 TypeSafe API 演进,SDK 会持续可用,因此你可以在某次 SDK 发布正式支持某个新 API 特性之前,就先把它用起来。

额外的请求字段

用 extra_body 发送额外的 API 请求字段。下面的 beam_width 字段只是示意;只发送 API 支持的字段。

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Noul

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="About billing?")},
            extra_body={"beam_width": 4},
        )

asyncio.run(main())
from typesafe_sdk import Noul, TypeSafeClient

with TypeSafeClient() as client:
    client.system_one(
        "I was charged twice.",
        {"billing": Noul(instructions="About billing?")},
        extra_body={"beam_width": 4},
    )

原始问题字典

import asyncio

from typesafe_sdk import AsyncTypeSafeClient

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        await client.system_one(
            "I was charged twice.",
            {
                "billing": {
                    "type": "noul",
                    "instructions": "About billing?",
                    "weight": 2,
                }
            },
        )

asyncio.run(main())
from typesafe_sdk import TypeSafeClient

with TypeSafeClient() as client:
    client.system_one(
        "I was charged twice.",
        {"billing": {"type": "noul", "instructions": "About billing?", "weight": 2}},
    )

未知的答案种类

对于无法识别的答案种类,SDK 会记录一条警告并跳过。用 raw_http_response 检查完整的 API 响应,其中包含那些答案:

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Noul

async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            "I was charged twice.",
            {"billing": Noul(instructions="Is this about billing?")},
        )
        print(result.raw_http_response.json()["answers"])

asyncio.run(main())
from typesafe_sdk import Noul, TypeSafeClient

result = TypeSafeClient().system_one(
    "I was charged twice.",
    {"billing": Noul(instructions="Is this about billing?")},
)
raw_answers = result.raw_http_response.json()["answers"]

未知的响应字段

已识别响应上的未知额外字段会被忽略。