用法
以下示例都假设你已经安装 SDK 并设置了 XIANGXIN_API_KEY。
调用 System One API
system_one 的前两个参数是 state 和 questions,既可以按位置传,也可以按关键字传。questions 的键是你给问题起的名字,答案会以同样的名字返回。
import asyncio
from xiangxin import AsyncXiangxinClient, Choice, Noul, Score
async def main() -> None:
async with AsyncXiangxinClient() as client:
result = await client.system_one(
"我被重复扣费了两次,请尽快处理!",
{
"billing": Noul(instructions="这是否与扣费有关?"),
"tone": Choice(
instructions="语气是?",
criteria={"calm": None, "angry": None},
),
"urgency": Score(
instructions="有多紧急?",
criteria=["低", "中", "高"],
),
},
)
print(
result.nouls["billing"].noul,
result.choices["tone"].choice,
result.scores["urgency"].score,
)
asyncio.run(main())from xiangxin import Choice, Noul, Score, XiangxinClient
client = XiangxinClient()
state = "我被重复扣费了两次,请尽快处理!"
questions = {
"billing": Noul(instructions="这是否与扣费有关?"),
"tone": Choice(instructions="语气是?", criteria={"calm": None, "angry": None}),
"urgency": Score(instructions="有多紧急?", criteria=["低", "中", "高"]),
}
result = client.system_one(state, questions)
print(
result.nouls["billing"].noul,
result.choices["tone"].choice,
result.scores["urgency"].score,
)result.answers 按问题名收录全部答案,result["billing"] 是 result.answers["billing"] 的简写;nouls、choices、scores 则按类型筛好了答案,类型检查器能直接推断出 .noul、.choice、.score 等字段。每个字段的含义见答案与响应。
提示
问题名只用来在响应里对应答案。把完整的意思写进 instructions,不要指望模型从名字里猜。同一个 state 上的多个问题尽量放进一次调用:模型只读一遍 state,这就是推测式扇出。
每个响应还带有用量和耗时信息:
print(result.model) # 实际应答的模型版本,例如 xiangxin-1.0.0
print(result.usage.input_tokens) # 计费依据,见 /models#pricing
print(result.request_id) # x-request-id,联系技术支持时请附上
print(result.model_ms, result.total_ms) # 模型推理耗时 / 网关总耗时(毫秒)带类型的 system_one 响应
给 system_one 传入 response_model,就能以属性的方式、带类型地访问答案。最简单的做法是继承 SystemOneResponse,声明与问题同名的字段:
from xiangxin import Noul, NoulAnswer, SystemOneResponse, XiangxinClient
class BillingResponse(SystemOneResponse):
billing: NoulAnswer
with XiangxinClient() as client:
result = client.system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
response_model=BillingResponse,
)
assert 0 <= result.billing.noul <= 1
assert result.billing == result.nouls["billing"]
print(result.request_id)自定义响应类型
也可以完全不继承 SystemOneResponse,用任意 pydantic 模型描述响应 JSON:
from pydantic import BaseModel
from xiangxin import Noul, NoulAnswer, XiangxinClient
class BillingAnswers(BaseModel):
billing: NoulAnswer
class BillingResponse(BaseModel):
answers: BillingAnswers
result = XiangxinClient().system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
response_model=BillingResponse,
)
assert 0 <= result.answers.billing.noul <= 1这种模型不继承 XiangxinResponse,因此没有 request_id、raw_http_response 等属性;需要响应头时请改用原始响应视图。响应不符合模型时抛出 APIResponseValidationError。
选择模型
查看可用模型:
from xiangxin import XiangxinClient
for m in XiangxinClient().models.list().models:
print(m.name, m.release_date, m.description)创建客户端时指定模型:
client = XiangxinClient(model="xiangxin-1.0.0")也可以在单次调用中用 model= 覆盖。默认的 xiangxin-latest 是别名,会随新版本发布而移动;如果你的置信度阈值是在某个版本上调出来的,建议固定版本号。详见 Models 资源和模型。
配置 base URL
要让 SDK 连接其他地址(例如私有化部署或内网网关),在客户端上设置 base_url,或设置环境变量 XIANGXIN_BASE_URL。需要走企业代理或自定义 TLS 设置时,传入自己配置好的 httpx 客户端:
import os
from xiangxin import Noul, XiangxinClient
with XiangxinClient(
api_key=os.environ["XIANGXIN_API_KEY"],
base_url="https://xiangxin.internal.example.com",
) as client:
result = client.system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
)
print(result.nouls["billing"].noul)import httpx
from xiangxin import Noul, XiangxinClient
with XiangxinClient(
http_client=httpx.Client(proxy="http://10.0.0.8:3128"),
) as client:
result = client.system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
)
print(result.nouls["billing"].noul)base_url 必须以 http:// 或 https:// 开头,末尾的斜杠会被去掉。替代地址需要实现与象信 HTTP API 相同的接口。
重试
SDK 默认会对限流、过载、5xx、连接错误和超时自动重试。需要调整时,在客户端上或单次调用中传入自定义的 RetryPolicy。API 密钥缺失或格式非法时,创建客户端就会抛出 XiangxinError,不会发出请求,也不会重试。
from xiangxin import RetryPolicy, XiangxinClient
client = XiangxinClient(retry=RetryPolicy(max_retries=3, backoff_max=2.0, timeout=10.0))from xiangxin import RetryPolicy
client.system_one(
state, questions, retry=RetryPolicy(max_retries=3, backoff_max=2.0, timeout=10.0)
)另有 timeout 参数控制单次 HTTP 操作的超时(默认 30 秒,可传秒数或 httpx.Timeout),而 RetryPolicy.timeout(默认 60 秒)限制的是一次 SDK 调用含所有重试与等待的总预算。
错误处理
处理 SDK 抛出的异常:
from xiangxin import APIError
try:
client.system_one(state, questions)
except APIError as error:
print(error.status_code, error.detail, error.request_id)所有异常都继承自 XiangxinError;余额不足(InsufficientBalanceError)、请求未通过校验(UnprocessableEntityError)等情况有各自的子类。
日志
SDK 使用名为 xiangxin 的 logger,按标准 logging 的方式配置即可:
import logging
logging.getLogger("xiangxin").setLevel(logging.DEBUG)也可以在导入 SDK 之前把 XIANGXIN_LOG 设为 debug、info、warning、error 或 off。
info 为每个请求输出一行摘要(方法、路径、状态码、耗时、请求 ID);debug 还会输出请求与响应的头和体。每次重试前会输出一条 WARNING。日志中的鉴权类请求头——Authorization、API 密钥、Cookie,以及名字里含 token 或 secret 的头——会被隐去;请求体和响应体不会脱敏,处理敏感数据时请谨慎开启 debug。
环境变量
SDK 会读取以下环境变量:
| 变量 | 配置项 | 默认值 |
|---|---|---|
XIANGXIN_API_KEY | API 密钥(必需) | — |
XIANGXIN_BASE_URL | API 根地址 | https://api.xiangxinai.cn |
XIANGXIN_DEFAULT_MODEL | 默认模型 | xiangxin-latest |
XIANGXIN_LOG | xiangxin logger 的级别,导入时生效一次 | 未设置 |
显式传入的构造参数优先于环境变量,值为空或只有空白的环境变量视为未设置。SDK 的默认值见常量。
通过 api_key 或 XIANGXIN_API_KEY 提供的密钥会去掉首尾空白(包括从密钥文件读入的换行)。空密钥、中间含空白、控制字符或非 ASCII 字符的密钥会在发出请求前被拒绝。显式传入空字符串不会回退到环境变量。
前向兼容
象信 API 演进时,SDK 仍能继续工作:在 SDK 发版正式支持之前,你就可以先用上 API 的新功能。
额外的请求字段
用 extra_body 发送额外的请求体字段。它会浅合并到请求体顶层,同名键会覆盖 state、model、questions。下面的 new_option 只是示意,请只发送 API 支持的字段。
from xiangxin import Noul, XiangxinClient
with XiangxinClient() as client:
client.system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
extra_body={"new_option": 4},
)需要附加请求头时用 extra_headers;Authorization 与 Accept 由 SDK 设置,不能被覆盖。
原始问题字典
问题类会拒绝未知字段;问题字典则会原样发送,可以带上 SDK 尚未建模的字段(下例中的 weight 同样只是示意):
from xiangxin import XiangxinClient
with XiangxinClient() as client:
client.system_one(
"我被重复扣费了两次。",
{"billing": {"type": "noul", "instructions": "这是否与扣费有关?", "weight": 2}},
)提示
未知字段只是前向兼容的应急出口。请忽略它们带来的类型检查报错,并优先考虑升级 SDK。
未知的答案类型
遇到无法识别的答案类型时,SDK 会记录一条警告并跳过该答案。用 raw_http_response 可以查看完整的 API 响应,包括被跳过的答案:
from xiangxin import Noul, XiangxinClient
result = XiangxinClient().system_one(
"我被重复扣费了两次。",
{"billing": Noul(instructions="这是否与扣费有关?")},
)
raw_answers = result.raw_http_response.json()["answers"]未知的响应字段
已识别的响应中出现的未知字段会被忽略。
测试
单元测试里不需要真的调用 API。SDK 基于 httpx,传入 httpx.MockTransport 即可在本地返回固定响应(异步客户端同理):
import httpx
from xiangxin import Noul, XiangxinClient
FAKE = {
"model": "xiangxin-1.0.0",
"answers": {"is_spam": {"type": "noul", "noul": 0.97}},
"usage": {"input_tokens": 42, "output_tokens": 0},
}
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/v1/systemone"
return httpx.Response(200, json=FAKE, headers={"x-request-id": "req_test"})
def test_spam_filter() -> None:
client = XiangxinClient(api_key="sk-xx-test", transport=httpx.MockTransport(handler))
resp = client.system_one("加我领券", {"is_spam": Noul(instructions="是否广告?")})
assert resp.nouls["is_spam"].noul > 0.9
assert resp.request_id == "req_test"
