Verwendung

Anleitungen und Muster für die Arbeit mit dem TypeSafe Python SDK.

Die System One-API aufrufen

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,
)

Typisierte `system_one`-Antworten

Es ist möglich, system_one ein Antwortmodell bereitzustellen, damit die Verwendung der Antwort typsicherer wird:

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)

Benutzerdefinierte Antworttypen

Es ist auch möglich, ein völlig neues Antwortmodell zu definieren, ohne von SystemOneResponse zu erben:

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

Ein Modell auswählen

Sieh dir die verfügbaren Modelle an:

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())

Wähle das Modell beim Erstellen eines Clients aus:

client = AsyncTypeSafeClient(model="jev")

client = TypeSafeClient(model="jev")

Siehe die Referenz zur Ressource Models für Details.

Die Basis-URL konfigurieren

Um das SDK mit einer anderen API-URL zu verwenden, setze base_url am Client oder die Umgebungsvariable TYPESAFE_BASE_URL. Dies setzt voraus, dass die alternative API die TypeSafe OpenAPI-Spezifikation befolgt.

Verbinde dich zum Beispiel über ein KI-Gateway mit dessen API-Schlüssel und Modell-ID.

Verwende einen OpenRouter-API-Schlüssel und eine OpenRouter-Modell-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)

Vercels TypeSafe-kompatible API kann mit dem SDK verwendet werden:

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)

Verwende einen Pydantic AI Gateway-API-Schlüssel:

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))

Wiederholungen

Übergib eine benutzerdefinierte RetryPolicy als retry am Client oder pro Aufruf. Ungültige API-Schlüssel lösen bei der Client-Erstellung einen TypeSafeError aus, bevor eine Anfrage oder Wiederholung erfolgt.

Auf dem Client:

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))

Pro Aufruf:

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)
)

Fehlerbehandlung

Behandle Ausnahmen, die vom SDK ausgelöst werden:

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)

Protokollierung

Das SDK protokolliert in den Logger typesafe_sdk. Konfiguriere ihn gemäß der Anleitung zur Standardprotokollierung:

import logging

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

Oder setze TYPESAFE_LOG_LEVEL vor dem Importieren des SDK auf einen der Werte debug, info, warning, error oder off.

info protokolliert eine Zusammenfassungszeile pro Anfrage; debug protokolliert zusätzlich die Header und Bodys von Anfragen und Antworten. Geheime Header – Autorisierung, API-Schlüssel, Cookies und jeder Header, dessen Name token oder secret enthält – werden aus der Protokollausgabe entfernt. Anfrage- und Antwort-Bodys werden nicht entfernt.

Umgebungsvariablen

Das SDK liest und verwendet die folgenden Umgebungsvariablen:

Variable	Konfiguriert	Standard
`TYPESAFE_API_KEY`	API-Schlüssel (erforderlich)	—
`TYPESAFE_BASE_URL`	API-Wurzel-URL	`https://api.typesafe.ai`
`TYPESAFE_DEFAULT_MODEL`	Standardmodell	`jev-latest`
`TYPESAFE_LOG_LEVEL`	Grad des Loggers `typesafe_sdk`, einmal beim Import angewendet	nicht gesetzt

Siehe die Referenz zu Konstanten für die SDK-Standardwerte.

API-Schlüssel, die über api_key oder TYPESAFE_API_KEY übergeben werden, werden von führenden und abschließenden Leerzeichen befreit, einschließlich Zeilenumbrüchen aus Schlüsseldateien. Leere Schlüssel, interne Leerzeichen, Steuerzeichen und Nicht-ASCII-Zeichen werden vor dem Senden einer Anfrage abgelehnt. Ein ausdrücklich leerer Schlüssel greift nicht auf die Umgebung zurück.

Vorwärtskompatibilität

Das SDK funktioniert weiter, während sich die TypeSafe-API weiterentwickelt, sodass du neue API-Funktionen übernehmen kannst, bevor eine SDK-Version erstklassige Unterstützung dafür hinzufügt.

Zusätzliche Anfragefelder

Sende zusätzliche API-Anfragefelder mit extra_body. Das Feld beam_width unten ist illustrativ; sende nur Felder, die von der API unterstützt werden.

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},
    )

Rohe Fragewörterbücher

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}},
    )

Unbekannte Antwortarten

Das SDK protokolliert eine Warnung und überspringt nicht erkannte Antwortarten. Verwende raw_http_response, um die vollständige API-Antwort einschließlich dieser Antworten zu untersuchen:

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"]

Unbekannte Antwortfelder

Unbekannte zusätzliche Felder in erkannten Antworten werden ignoriert.

Verwendung

Die System One-API aufrufen

Typisierte system_one-Antworten

Benutzerdefinierte Antworttypen

Ein Modell auswählen

Die Basis-URL konfigurieren

HTTP/2

Wiederholungen

Fehlerbehandlung

Protokollierung

Umgebungsvariablen

Vorwärtskompatibilität

Zusätzliche Anfragefelder

Rohe Fragewörterbücher

Unbekannte Antwortarten

Unbekannte Antwortfelder

Typisierte `system_one`-Antworten