主题模式

路由策略 ​

路由策略 ​

AI Ping 的路由策略目标是把请求调度到最合适的服务商，用户可通过修改对话生成接口中的 provider 字段定制路由策略。

为确保接口稳定可用，若最优服务商接口调用失败，系统将按优先级重试所有可用服务商接口，如全部尝试均未成功，则返回失败。

默认策略 ​

基于接口的可靠性、价格、性能，为用户选取当前最合适的服务商。

不传 extra_body.provider 时按标准价 × 1.00 计费。一旦指定 only / ignore / order / sort / 倍率或性能、上下文范围，将按实际执行路线的倍率计费。

不需要调整任何参数，默认采用的参数

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

价格优先策略 ​

sort 取：price_multiplier（推荐）、input_price、output_price

price_multiplier 按目录输入价与输出价综合排序（不是按计费倍率排序）。input_price / output_price 仍可用，分别表示输入价优先、输出价优先；一种价格相同时再比较另一种。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "sort": "price_multiplier"
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

吞吐优先策略 ​

sort 取：throughput

根据实时测得的各服务商的吞吐数据，优先选择吞吐高的服务商

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "sort": "throughput"
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

延迟优先策略 ​

sort 取：latency

根据实时测得的各服务商的延迟数据，优先选择延迟低的服务商

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "sort": "latency"
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

最大输入长度优先策略 ​

sort 取：input_length

在长输入的场景中，经常存在服务商提供的最大输入长度不足的情况，使用最大输入长度优先策略，系统将优先调度给长最大输入长度的服务商。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "sort": "input_length"
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

多关键字混合策略 ​

provider.sort 可以兼容列表格式，支持多种策略，优先级按次序降低
例如：[ "price_multiplier" , "throughput" ]，在优先考虑目录价格的基础上，将选择吞吐更高的接口。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "sort": ["price_multiplier", "throughput"] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

指定服务商 ​

provider.only 取指定的服务商，调度时将优先调度给指定的服务商。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "only": ["阿里云百炼"] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

服务商排序 ​

provider.order 取指定的服务商，优先级按次序降低。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "order": ["阿里云百炼", "火山方舟"] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

筛选倍率范围 ​

provider.effective_multiplier_range 取 [ low , high ]，调度时只调度给计费倍率在范围内的服务商。区间两端可省略，如 [None, 1] 表示不超过标准价。空列表 [] 视为未设置，与不传该字段相同。

目录表上的输入/输出元价格请勿填入该字段。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "effective_multiplier_range": [None, 1] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

筛选吞吐范围 ​

provider.throughput_range 取 [ low_throughput , high_throughput ]，调度时只调度给模型吞吐在 low_throughput <= throughput <= high_throughput 范围内的服务商

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "throughput_range": [30, 100] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

筛选延迟范围 ​

provider.latency_range 取 [ low_latency , high_latency ]，调度时只调度给模型延迟在 low_latency <= latency <= high_latency 范围内的服务商

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "latency_range": [0, 2] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

筛选最大输入长度范围 ​

provider.input_length 取 [ low_input_length, high_input_length ]，调度时只调度给最大输入长度在 low_input_length <= input_length <= high_input_length 范围内的服务商

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "input_length": [64 * 1024, 1024 * 1024] 
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)

失败重试 ​

provider.allow_fallbacks 用于控制筛选无匹配服务商时，是否自动兜底调度其他服务商。

字段含义 ​

  • true：开启兜底降级。当系统按筛选策略（延迟、吞吐、最大输入长度）过滤后无符合条件的服务商时，会忽略筛选限制，按默认策略调度其他可用服务商，保障调用成功。
  • false：关闭兜底降级。筛选后无匹配服务商时，直接返回路由失败报错，不切换任何其他服务商。

平台默认规则（用户未设置 allow_fallbacks） ​

  • 请求配置了 provider.only 字段时，默认生效值为 false。此时仅允许使用 only 指定的服务商；若筛选规则不满足该服务商，系统将直接报错，不会自动切换其他服务商。
  • 请求未配置 provider.only 字段时，默认生效值为 true。此时筛选后无匹配服务商时，系统会自动兜底调度其他渠道，以提升接口可用性。

用户显式设置 allow_fallbacks ​

用户一旦显式传入 allow_fallbacks，均以用户传入值为准，优先级高于平台默认规则。

python
from openai import OpenAI

openai_client = OpenAI(
  base_url="https://aiping.cn/api/v1",
  api_key="<API_KEY>",
)

stream = openai_client.chat.completions.create(
    model="DeepSeek-R1-0528",
    stream=True,
    extra_body={
        "provider": {
            "allow_fallbacks": False
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Hello"
        }
    ]
)

for chunk in stream:
    if not getattr(chunk, "choices", None):
        continue

    content = getattr(chunk.choices[0].delta, "content", None)
    if content:
        print(content, end="", flush=True)