外观

象信控制台
客户端

异步客户端

用 AsyncXiangxinClient 提问、列出模型，并配置异步的象信 API 请求。

异步客户端 ​

xiangxin.AsyncXiangxinClient ​

python
AsyncXiangxinClient(
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float | httpx.Timeout | None = None,
    headers: Mapping[str, str] | None = None,
    transport: httpx.AsyncBaseTransport | None = None,
    http_client: httpx.AsyncClient | None = None,
)

创建一个调用象信 API 的 asyncio HTTP 客户端，基于 httpx.AsyncClient，适合 FastAPI、aiohttp 等异步服务，以及需要高并发的批处理脚本。参数、方法、返回值都与同步客户端一一对应，区别只是方法需要 await，且 transport / http_client 需要传异步版本。

所有参数都只能按关键字传入。显式传入的参数优先于环境变量；值为空或只有空白的环境变量视为未设置。构造函数本身是同步的，可以在模块顶层创建，但请在事件循环里使用和关闭它。

日志配置

SDK 的日志写入 xiangxin logger，可以按标准 logging 方式配置，也可以设置 XIANGXIN_LOG（debug、info 等）快速开启。鉴权类请求头会被隐去，请求体和响应体不会。详见用法 · 日志。

参数

api_keystr | None默认 None

API 密钥，必需；未传入时读取环境变量 XIANGXIN_API_KEY。首尾空白会被去掉；空密钥、中间含空白、控制字符或非 ASCII 字符的密钥会被拒绝。

base_urlstr | None默认 None

API 根地址；未传入时读取 XIANGXIN_BASE_URL，否则为 https://api.xiangxinai.cn。必须以 http:// 或 https:// 开头，末尾斜杠会被去掉。

modelstr | None默认 None

该客户端的默认模型；未传入时读取 XIANGXIN_DEFAULT_MODEL，否则为 xiangxin-latest。

retryRetryPolicy | None默认 None

控制重试行为的 RetryPolicy；None 使用 RetryPolicy() 的默认值。传 RetryPolicy(max_retries=0) 关闭重试。

timeoutfloat | httpx.Timeout | None默认 None

单次 HTTP 操作的超时，正数（秒）或 httpx.Timeout。传入 http_client 时沿用它的超时设置，否则默认 30 秒。

headersMapping[str, str] | None默认 None

附加到每个请求的请求头。Authorization 和 Accept 由 SDK 设置，不能被覆盖。

transporthttpx.AsyncBaseTransport | None默认 None

自定义的 HTTP transport，常用于测试（httpx.MockTransport 同时支持同步与异步）。

http_clienthttpx.AsyncClient | None默认 None

自己构造好的 httpx.AsyncClient（例如配置了代理）；与 transport 互斥。关闭 SDK 客户端时它也会被关闭。

异常

  • XiangxinError：API 密钥缺失或非法、base_url 或 timeout 非法。
  • ValueError：同时传入了 transport 和 http_client。

示例

python
import asyncio

from xiangxin import AsyncXiangxinClient, Choice, Noul


async def main() -> None:
    async with AsyncXiangxinClient() as client:
        result = await client.system_one(
            state="我被重复扣费了两次，请帮忙处理。",
            questions={
                "billing": Noul(instructions="这是否与扣费有关？"),
                "tone": Choice(
                    instructions="语气是？",
                    criteria={"calm": None, "angry": None},
                ),
            },
        )
        assert 0 <= result.nouls["billing"].noul <= 1
        assert result.choices["tone"].choice in {"calm", "angry"}


asyncio.run(main())

models ​

属性

python
models: AsyncModels

Models 资源的入口。

python
async with AsyncXiangxinClient() as client:
    models = await client.models.list()

system_one ​

python
async system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float | httpx.Timeout | None = None,
    extra_headers: Mapping[str, str] | None = None,
    extra_body: Mapping[str, Any] | None = None,
    response_model: type[Any] | None = None,
) -> Any
python
async system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float | httpx.Timeout | None = None,
    extra_headers: Mapping[str, str] | None = None,
    extra_body: Mapping[str, Any] | None = None,
    response_model: None = None,
) -> SystemOneResponse
python
async system_one(
    state: JSONContent,
    questions: Mapping[str, Question],
    *,
    model: str | None = None,
    retry: RetryPolicy | None = None,
    timeout: float | httpx.Timeout | None = None,
    extra_headers: Mapping[str, str] | None = None,
    extra_body: Mapping[str, Any] | None = None,
    response_model: type[ResponseT],
) -> ResponseT

对文本或结构化的 state 提出一组命名问题，调用 POST /v1/systemone，一次返回全部答案。与同步版相同，只是需要 await；重试等待使用 asyncio.sleep，不会阻塞事件循环。原理见系统一。

参数

stateJSONContent必填

要评估的文本、JSON 对象或数组，不能为 None。见状态。

questionsMapping[str, Question]必填

非空映射：问题名 → 问题对象或问题字典。

modelstr | None默认 None

仅本次调用使用的模型；None 沿用客户端默认值。

retryRetryPolicy | None默认 None

仅本次调用使用的重试策略，覆盖客户端级设置。

timeoutfloat | httpx.Timeout | None默认 None

仅本次调用使用的单次 HTTP 操作超时（秒），覆盖客户端级设置。

extra_headersMapping[str, str] | None默认 None

仅本次调用附加的请求头；Authorization 与 Accept 仍受保护。

extra_bodyMapping[str, Any] | None默认 None

额外的请求体顶层字段，在 state、model、questions 设置之后浅合并进去。后写者生效：与这三个键同名时会覆盖它们。用法见前向兼容。

response_modeltype[ResponseT] | None默认 None

描述响应 JSON 的 pydantic BaseModel 类型（包括其中嵌套的答案模型）。见带类型的响应。

返回

  • SystemOneResponse：未传 response_model 时返回，包含按问题名索引的答案、实际模型与 token 用量。
  • ResponseT：传入 response_model 时返回该模型的实例。

异常

  • XiangxinError：请求发出前的本地校验失败，例如 state 为 None、questions 为空、问题字典缺少 type、Choice / Score 字典的 criteria 为空。
  • APIError：重试后服务端仍返回不成功的 HTTP 响应。
  • APIConnectionError：重试后仍无法连接或超时。
  • APIResponseValidationError：响应体与响应模型不符。

示例

用问题类构造问题：

python
async with AsyncXiangxinClient() as client:
    result = await client.system_one(
        state="我被重复扣费了两次，请帮忙处理。",
        questions={
            "billing": Noul(instructions="这是否与扣费有关？"),
            "tone": Choice(
                instructions="语气是？",
                criteria={"calm": None, "angry": None},
            ),
        },
    )
    assert 0 <= result.nouls["billing"].noul <= 1
    assert result.choices["tone"].choice in {"calm", "angry"}

以字典形式传入问题：

python
async with AsyncXiangxinClient() as client:
    result = await client.system_one(
        state={"message": "我被重复扣费了两次，请帮忙处理。"},
        questions={
            "billing": {"type": "noul", "instructions": "这是否与扣费有关？"},
            "tone": {
                "type": "choice",
                "instructions": "语气是？",
                "criteria": {"calm": None, "angry": None},
            },
        },
    )
    assert 0 <= result.nouls["billing"].noul <= 1
    assert result.choices["tone"].choice in {"calm", "angry"}

with_raw_response ​

属性

system_one 与 models.list 的“原始响应”版本：参数完全相同，但返回 RawResponse。与同步版一样，只是需要 await：

python
raw = await client.with_raw_response.system_one(state, questions)

print(raw.status_code, raw.headers.get("x-xiangxin-total-ms"))
resp = raw.parse()  # parse() 本身是同步方法

models = (await client.with_raw_response.models.list()).parse()

base_url / model / retry ​

属性

python
base_url: str
model: str
retry: RetryPolicy

解析后的 API 根地址、默认模型和客户端级重试策略。

aclose ​

python
async aclose() -> None

释放网络资源，关闭底层 HTTP 客户端（包括你传入的 http_client）。用 async with AsyncXiangxinClient() as client: 时会自动调用。

close ​

python
async close() -> None

aclose 的别名，同样需要 await。

Models 资源 ​

通过 AsyncXiangxinClient.models 访问。

xiangxin.AsyncModels ​

访问当前账号可用的模型，通过 AsyncXiangxinClient.models 获得。

list ​

python
async list(
    *,
    retry: RetryPolicy | None = None,
    timeout: float | httpx.Timeout | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> ListModelsResponse

调用 GET /v1/models，列出当前账号可以填进 model 的名称。

参数

retryRetryPolicy | None默认 None

仅本次调用使用的重试策略，覆盖客户端级设置。

timeoutfloat | httpx.Timeout | None默认 None

仅本次调用使用的超时；None 沿用客户端设置。

extra_headersMapping[str, str] | None默认 None

附加请求头；Authorization 与 Accept 仍受保护。

返回

  • ListModelsResponse：其 models 字段列出每个模型的名称、说明和发布日期。

异常

示例

python
from xiangxin import AsyncXiangxinClient

async with AsyncXiangxinClient() as client:
    models = await client.models.list()
    for m in models.models:
        print(m.name, m.release_date)

控制并发 ​

异步客户端最常见的用途是同时处理大量不同的 state。请用 asyncio.Semaphore 限制同时在途的请求数，否则一次性发出几千个请求很容易撞上速率限制：

python
import asyncio

from xiangxin import AsyncXiangxinClient, Choice

INTENT = Choice(
    instructions="用户这句话想做什么？",
    criteria={
        "check_order": "查询订单或物流状态",
        "return_goods": "退货、换货、退款",
        "invoice": "开发票、改发票抬头",
        "human": "要求转人工",
        "other": "以上都不是",
    },
)


async def route_messages(messages: list[str]) -> list[tuple[str, str, float]]:
    sem = asyncio.Semaphore(16)
    async with AsyncXiangxinClient() as client:

        async def one(msg: str) -> tuple[str, str, float]:
            async with sem:
                resp = await client.system_one(state=msg, questions={"intent": INTENT})
            ans = resp.choices["intent"]
            return msg, ans.choice, ans.confidence

        return await asyncio.gather(*(one(m) for m in messages))


rows = asyncio.run(route_messages(["我的快递到哪了", "发票抬头开错了能改吗", "叫你们人工来"]))
for msg, intent, conf in rows:
    print(f"{intent:12} {conf:.2f}  {msg}")

限流（429）与过载（529）会被自动退避重试，见重试。某一条最终仍失败时 asyncio.gather 会抛出异常；想让其余条目继续，可以传 return_exceptions=True，事后单独处理失败项。同一个 state 上的多个问题本来就在一次请求里并行计算，并发应加在不同 state 之间。

在 FastAPI 中复用 ​

在异步 Web 服务里创建一个进程级客户端，应用关闭时释放：

python
from contextlib import asynccontextmanager

from fastapi import FastAPI
from xiangxin import AsyncXiangxinClient, Noul


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.xx = AsyncXiangxinClient()
    yield
    await app.state.xx.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/moderate")
async def moderate(text: str) -> dict:
    resp = await app.state.xx.system_one(
        state=text,
        questions={"abusive": Noul(instructions="这段文字是否包含辱骂或人身攻击？")},
    )
    return {"abusive": resp.nouls["abusive"].noul, "request_id": resp.request_id}