主题模式

路由策略 ​

路由策略 ​

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

为确保接口稳定可用，系统在遇到所选最优供应商接口调用失败时，会进行最多两次自动重试，并依次切换至其他次优供应商进行调度；如三次调度均未成功，则返回失败。

默认策略 ​

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

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

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 取：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": "output_price"
        }
    },
    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 可以兼容列表格式，支持多种策略，优先级按次序降低
例如：[ "output_price" , "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": ["output_price", "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 取指定的供应商，调度时将优先调度给指定的供应商。

供应商列表见：https://aiping.cn/supplierList

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.input_price_range 取 [ low_price , high_price ]，调度时只调度给模型输入价格在 low_price <= input_price <= high_price 范围内的供应商；

筛选输出价格：
provider.output_price_range 取 [ low_price , high_price ]，调度时只调度给模型输出价格在 low_price <= output_price <= high_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": {
            "input_price_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.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，系统按照用户指定筛选策略(延迟、吞吐、最大输入长度)无法找到合适的供应商时，为了保证接口可用，将按照指定排序策略调度给其他的供应商；

将 provider.allow_fallbacks 为 False，系统在指定筛选策略无法找到合适的供应商时，将返回报错。

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)