Provider — Downcity City SDK Docs

Provider 是抽象基类，封装第三方 AI 提供商的 action 实现和连接信息。通过 .model() 生成模型配置，传给 AIService.use() 注册。

快速开始：OpenAI-compatible 文本 Provider

继承 Provider，覆盖 createClient，即可自动获得 text / stream 能力：

import { createOpenAICompatible } from "@ai-sdk/openai-compatible";
import { Provider, type OpenAICompatibleClientConfig } from "@downcity/city";

class DeepSeekProvider extends Provider {
  constructor() {
    super({
      id: "deepseek",
      envKey: "DEEPSEEK_API_KEY",
      baseURL: "https://api.deepseek.com/v1",
      passthroughModel: "deepseek-chat",
    });
  }

  protected createClient({ apiKey, baseURL }: OpenAICompatibleClientConfig) {
    return createOpenAICompatible({ apiKey, baseURL, name: "deepseek" });
  }
}

const deepseek = new DeepSeekProvider();

const ai = new AIService();
ai.use(deepseek.model({ id: "deepseek-v4-flash", name: "DeepSeek V4 Flash" }));
base.use(ai);

注册后自动生成路由：

POST /v1/ai/text              → SDK 通路
POST /v1/ai/stream            → SDK 通路
POST /v1/ai/chat/completions  → 自动透传到 DeepSeek
GET  /v1/ai/models            → 模型目录

passthroughModel 控制自动透传时发往上游的 model 值。

模型上下文窗口

通过 context_window 声明模型支持的总上下文 token 容量：

ai.use(deepseek.model({
  id: "deepseek-v4-flash",
  name: "DeepSeek V4 Flash",
  context_window: 128_000,
}));

该值会通过 /v1/ai/models 返回给客户端。@downcity/agent 使用 City 模型时，会默认在上下文达到窗口的 80% 时触发 session compact，为 system、工具调用和本轮输出保留余量。未配置 context_window 的模型继续使用 Agent 的 128k 默认 compact 预算；自定义 maxInputTokensApprox 仍拥有最高优先级。

模型推理强度

支持推理强度的模型通过 reasoning 声明可选档位和默认档位：

ai.use(openai.model({
  id: "gpt-5.6-sol",
  name: "GPT-5.6 Sol",
  reasoning: {
    efforts: [
      { id: "low", name: "低" },
      { id: "medium", name: "标准" },
      { id: "high", name: "高" },
      { id: "xhigh", name: "极高" },
    ],
    default_effort: "medium",
  },
}));

这些档位会通过 /v1/ai/models 返回。调用方只传一个字段，值必须是当前模型 reasoning.efforts[].id 中的一项：

await client.ai.text({
  model: "gpt-5.6-sol",
  prompt: "分析这个问题",
  reasoning_effort: "high",
});

AIService 会先完成模型 fallback，再按照最终模型校验 reasoning_effort。请求没有指定档位时使用 default_effort；模型没有配置默认档位时使用上游默认行为。不支持的档位会返回 422。

默认 OpenAI-compatible text / stream 实现会把已校验档位转换成 AI SDK providerOptions 的 reasoningEffort。字段结构不同的 Provider 可以覆盖 build_reasoning_provider_options(ctx, model)，例如 Anthropic 映射到 anthropic.thinking / anthropic.effort，Gemini 映射到 google.thinkingConfig。自定义实现应通过 read_resolved_reasoning(ctx) 读取 AIService 已校验的结果，不要重新解析原始请求。

/v1/ai/chat/completions 已经是 OpenAI-compatible 协议，不转换成 providerOptions；校验后的 reasoning_effort 会留在请求体中交给 openai action 或自动透传。

覆盖默认 text / stream

如果默认的 OpenAI-compatible 实现不满足需求，可以直接覆盖 text / stream 方法：

import { Provider, type Context, type AIProviderChargedOutput } from "@downcity/city";
import type { UIMessage } from "ai";

class CustomTextProvider extends Provider {
  constructor() {
    super({ id: "custom", envKey: "CUSTOM_API_KEY" });
  }

  async text(ctx: Context): Promise<AIProviderChargedOutput<UIMessage>> {
    // 自定义实现
  }

  async stream(ctx: Context) {
    // 自定义实现
  }
}

多模态转文本

覆盖 extractPrompt 可以把多模态 messages 规整为纯文本：

class DeepSeekProvider extends Provider {
  // ...
  protected extractPrompt(input: { prompt?: string; messages?: Array<{ role: string; parts?: Array<{ type: string; text?: string }> }> }): string {
    if (typeof input.prompt === "string") return input.prompt;
    return input.messages
      ?.find((m) => m.role === "user")
      ?.parts?.map((part) => (part.type === "text" ? part.text : ""))
      .filter(Boolean)
      .join("\n") ?? "";
  }
}

模型 fallback

可以直接在 model 配置上挂 fallback：

const kimi = kimiProvider.model({
  id: "kimi-k2.6",
  name: "Kimi K2.6",
});

const deepseek = deepseekProvider.model({
  id: "deepseek-v4-flash",
  name: "DeepSeek V4 Flash",
  fallback: [
    {
      match: (media) => media.media_type.startsWith("image/"),
      model: kimi,
    },
  ],
});

当请求里出现 file 媒体输入，而当前模型不支持时，AIService 会在 text、stream、chat/completions 三条通路上按 fallback 数组顺序执行 match(media)，第一条命中且目标模型可用的规则会成为实际执行模型。

UIMessage 通路只会从 parts 里的 { type: "file" } 提取 media.media_type、media.filename 和 media.url；图片、音频、视频、PDF 等分流逻辑都应该写在 match 里。OpenAI-compatible 通路的 image_url / input_image 会被归一成 media.media_type === "image/*" 后参与同一套规则。

如果没有配置 fallback，或没有任何规则命中，AIService 会保持原模型不变，继续交给 provider 自己处理。

图片 Provider

@downcity/city 不内置 OpenAI、DeepSeek 或其他具体图片 Provider 构造器。City 包提供 Provider 抽象、AIService 注册机制，以及 client.ai.image_create() / client.ai.image_result() 的统一任务协议。

图片任务存储由 AIService 负责：它内置声明 async_jobs 表，Federation 初始化 Service 时会自动建表。Provider 只负责创建上游任务、查询上游状态，并把上游响应归一化成固定协议返回。

归一化的意思是：Provider 不能把上游原始响应直接交给 AIService 猜字段。不同上游可能把状态放在 status、data.status、result.status，也可能用 status_code + err + data 表示 pending 或成功。Provider 必须在 image_fetch() 内把这些结构翻译成 Downcity 唯一认可的结果：

type AIImageProviderFetchResult = {
  job_id: string;
  status: "queued" | "running" | "succeeded" | "failed";
  result?: UIMessage;
  error?: string;
  message?: string;
  poll_after_ms?: number;
  metadata?: Record<string, unknown>;
};

例如，上游返回 result pending 时，Provider 应返回 status: "running"、明确的 message、下一次 poll_after_ms，并在 metadata.raw 或类似字段保留原始响应摘要；上游成功时，Provider 必须返回 status: "succeeded" 和可落盘的 AI SDK UIMessage file parts；上游失败时，Provider 必须返回 status: "failed" 和可展示的 error。

一句话：AIService 管 job 存储、Queue 调度、状态缓存和计费；Provider 管上游协议解析、pending/succeeded/failed 映射、usage 提取和图片结果转换。

AIService 还会为图片任务提供平台级 pending 兜底：任务保持 queued 或 running 超过 image_max_pending_duration_ms 后，会被标记为 failed，error 和 message 都是 upstream timeout。默认值是 2 小时。Provider 不需要自己判断 Downcity 任务是否超时，只需要在上游仍未完成时持续返回 queued 或 running。

完整示例

import {
  AIService,
  Provider,
  type AIImageProviderCreateResult,
  type AIImageProviderFetchResult,
  type Context,
} from "@downcity/city";

class MyImageProvider extends Provider {
  constructor() {
    super({ id: "my-image", envKey: "MY_IMAGE_API_KEY" });
  }

  async image_create(ctx: Context): Promise<AIImageProviderCreateResult> {
    const input = ctx.input as { prompt?: string; size?: string };
    const api_key = ctx.env("MY_IMAGE_API_KEY");
    const response = await fetch("https://image.example.com/jobs", {
      method: "POST",
      headers: {
        authorization: `Bearer ${api_key}`,
        "content-type": "application/json",
      },
      body: JSON.stringify({
        prompt: input.prompt,
        size: input.size,
      }),
    });
    const data = await response.json() as { id?: string; error?: string };
    if (!response.ok || !data.id) {
      return {
        job_id: `img_${crypto.randomUUID()}`,
        status: "failed",
        error: data.error ?? response.statusText,
      };
    }
    return {
      job_id: `img_${crypto.randomUUID()}`,
      status: "running",
      message: "submitted",
      poll_after_ms: 3000,
      metadata: {
        upstream_job_id: data.id,
        upstream_model: "image-basic",
      },
    };
  }

  async image_fetch(ctx: Context): Promise<AIImageProviderFetchResult> {
    const image_job = ctx.locals.ai_image_job as {
      record: { job_id: string; user_id?: string | null; city_id?: string | null };
      state?: { upstream_job_id?: string; upstream_model?: string };
    };
    const upstream_job_id = image_job.state?.upstream_job_id;
    if (!upstream_job_id) {
      return {
        job_id: image_job.record.job_id,
        status: "failed",
        error: "Missing upstream image job id",
      };
    }

    const api_key = ctx.env("MY_IMAGE_API_KEY");
    const response = await fetch(`https://image.example.com/jobs/${upstream_job_id}`, {
      headers: {
        authorization: `Bearer ${api_key}`,
      },
    });
    const data = await response.json() as {
      status?: "queued" | "running" | "succeeded" | "failed";
      image_url?: string;
      error?: string;
      usage?: unknown;
    };

    if (!response.ok || data.status === "failed") {
      return {
        job_id: image_job.record.job_id,
        status: "failed",
        error: data.error ?? response.statusText,
        metadata: image_job.state,
      };
    }

    if (data.status === "queued" || data.status === "running" || !data.image_url) {
      return {
        job_id: image_job.record.job_id,
        status: data.status ?? "running",
        message: "generating",
        poll_after_ms: 3000,
        metadata: image_job.state,
      };
    }

    return {
      job_id: image_job.record.job_id,
      status: "succeeded",
      result: {
        id: `msg_${crypto.randomUUID()}`,
        role: "assistant",
        parts: [{
          type: "file",
          mediaType: "image/png",
          url: data.image_url,
        }],
      },
      metadata: {
        ...image_job.state,
        user_id: image_job.record.user_id,
        city_id: image_job.record.city_id,
        usage: data.usage,
      },
    };
  }
}

const ai = new AIService({ balance });
const image_provider = new MyImageProvider();

ai.use(image_provider.model({
  id: "image-basic",
  name: "Image Basic",
  bill(ctx, output) {
    return {
      user_id: output.metadata?.user_id,
      credits: 50_000,
      ref: `image:${output.job_id}`,
      note: "image generation",
      metadata: { usage: output.metadata?.usage },
    };
  },
}));

把 balance 传给 AIService 后，AIService 也会在真实消费模型的 action 前执行默认余额检查：如果用户钱包已经是负数，请求会在调用 Provider 前被拦截；Provider 成功后再按实际 usage 扣费。

Queue 是 Federation 级能力，不需要传给 AIService：

fed.queue.use(cloudflareQueue(env.DOWNCITY_QUEUE));
fed.use(ai);

Cloudflare Queue consumer 收到消息后，把消息交回 Federation：

for (const message of batch.messages) {
  await fed.queue.call(message.body);
  message.ack();
}

调用端仍然保持统一：

const job = await client.ai.image_create({
  model: "image-basic",
  prompt: "A clean product photo of a ceramic mug",
  ratio: "1:1",
  quality: "standard",
  count: 1,
});

const current = await client.ai.image_result({ job_id: job.job_id });

如果任务还在 queued 或 running，前端按 poll_after_ms 继续调用 image_result()。image_result() 只读取 async_jobs 缓存；后台 Queue 会调用 ai.image/fetch，由 AIService 调 Provider 的 image_fetch() 抓取上游状态并写回 async_jobs。

Provider 任务协议

调用方只使用 client.ai.image_create() 和 client.ai.image_result()。Provider 实现侧通常只需要两个方法：

class ImageProvider extends Provider {
  async image_create(ctx: Context): Promise<AIImageProviderCreateResult> {
    // 创建/启动上游任务。AIService 会把 ctx.input 和 metadata 保存到 async_jobs。
    return { job_id: "upstream_or_local_job_id", status: "running" };
  }

  async image_fetch(ctx: Context): Promise<AIImageProviderFetchResult> {
    // 读取 ctx.locals.ai_image_job，查询上游，并返回最新状态或最终结果。
    return { job_id: String(ctx.input.job_id), status: "running", poll_after_ms: 2000 };
  }
}

image_fetch 是图片模型的必需方法。模型只有同时提供 image_create 和 image_fetch 时，才会在模型目录里暴露 image capability。

AIService 负责 job 存储、Queue 调度、状态缓存、最终结果落库和幂等计费。Provider 只负责上游调用、上游轮询、usage 解析，以及把结果转换成 AI SDK UIMessage file parts。

自动透传

有 baseURL + envKey 但没有 openai 方法时，AIService 自动生成透传 action：

第三方工具的 OpenAI body 原样转发到上游
上游 Response 原样返回
passthroughModel 替换 body.model（有则替换）

// 这行不用写 — AIService 自动做
// async openai(ctx) { return fetch(`${baseURL}/chat/completions`, ...) }

自定义 OpenAI 兼容 Handler

当 OpenAI 兼容上游需要自定义请求或响应处理时，可以写 openai 方法：

import {
  Provider,
  type AIProviderChargedResponse,
  type Context,
  normalizeOpenAICompatibleBody,
  readRequiredEnv,
  trimTrailingSlash,
} from "@downcity/city";

class OpenAICustomProvider extends Provider {
  constructor() {
    super({ id: "openai-custom", envKey: "OPENAI_API_KEY", baseURL: "https://api.openai.com/v1" });
  }

  async openai(ctx: Context): Promise<AIProviderChargedResponse> {
    const apiKey = readRequiredEnv(ctx, this.envKey ?? "");
    const body = normalizeOpenAICompatibleBody(
      ctx.input as Record<string, unknown>,
      this.passthroughModel ?? "",
    );
    const response = await fetch(`${trimTrailingSlash(this.baseURL ?? "")}/chat/completions`, {
      method: "POST",
      headers: { authorization: `Bearer ${apiKey}`, "content-type": "application/json" },
      body: JSON.stringify(body),
    });
    return { response };
  }
}

Provider 出账

当 Provider 或某个具体 model 需要把一次完成的 AI 调用转换成账单草稿时，使用 bill(ctx, output)。bill 不直接扣余额；AIService 会读取它返回的 charge line，并通过配置的 BalanceService bridge 调用 balance.charge()。这次成功调用可以把钱包扣成负数；下一次真实消费模型的 AI 请求会被 AIService 的前置检查拦截。

图片任务的后台 image/fetch 会先原子领取任务，同一时间只有一个 worker 调用 Provider。Provider 返回成功后，AIService 先用稳定的任务幂等键完成扣费，再写入 status: "succeeded" 和 result_json；扣费失败时任务保持可重试，重复执行不会重复扣款。image_create() 不扣费；用户 image_result() 只读 async_jobs 缓存。Queue 消费没有当前用户上下文时，AIService 会自动使用任务创建时保存的 owner；charge line 仍可用 user_id 显式覆盖。

流式 action 返回异步 charge 时，Worker 使用 waitUntil 完成结算；Node 等没有 ExecutionContext 的运行时会把结算绑定到 Response 流收尾，扣费完成后才关闭响应流。普通异步 charge 则会在请求结束前直接完成。

通常优先级是：

action 显式返回 charge
model 提供 bill(ctx, output)
provider 提供 bill(ctx, output)
不产生扣费

import { normalizeAIUsage } from "@downcity/city";

class DeepSeekProvider extends Provider {
  // ...
  protected bill(ctx: Context, output: unknown) {
    const usage = output?.metadata?.usage;
    const normalized = normalizeAIUsage(usage);
    // 根据 usage 计算金额
    return {
      credits: 10_000,
      note: `DeepSeek ${ctx.variant?.id}`,
      metadata: { raw_usage: usage },
    };
  }
}

单个 model 也可以覆盖 provider 的出账逻辑：

ai.use(deepseek.model({
  id: "deepseek-v4-pro",
  name: "DeepSeek V4 Pro",
  bill(ctx, output) {
    return {
      credits: 20_000,
      note: "DeepSeek V4 Pro request",
      metadata: {
        model_id: ctx.variant?.id,
        usage: output?.metadata?.usage,
      },
    };
  },
}));

也可以在 action 方法里直接返回 charge：

async text(ctx: Context) {
  const result = await callUpstream(ctx.input);
  return {
    output: toUIMessage(result),
    charge: {
      credits: priceUpstreamUsage(result.usage),
      note: "custom text",
    },
  };
}

Provider 字段与方法

字段	必填	说明
`id`	是	Provider 唯一 ID
`env`	否	环境变量声明
`baseURL`	透传时必填	上游 API 地址
`envKey`	透传时必填	API Key 环境变量名
`passthroughModel`	否	透传时替换 body.model

方法	说明
`createClient`	需要 text/stream 时覆盖，返回 OpenAI-compatible client
`text`	SDK 文本生成 action（基类默认提供）
`stream`	SDK 流式生成 action（基类默认提供）
`image_create`	创建并启动图片 provider 任务
`image_fetch`	后台查询图片 provider 任务状态并返回最终结果
`video`	SDK 视频生成 action
`openai`	/chat/completions action（未提供则自动透传）
`bill`	为本次完成的调用返回账单草稿
`extractPrompt`	从输入提取 prompt（可覆盖）

工具函数

@downcity/city 提供一些通用工具函数，方便在 Provider 子类里复用：

readRequiredEnv(ctx, key)
resolveUpstreamModel(ctx, fallback)
trimTrailingSlash(url)
buildAssistantMessage(text, ctx, result, charge?)
buildImageMessage(ctx, images, metadata)
normalizeOpenAICompatibleBody(input, model)
readOpenAICompatibleSseUsage(body)
normalizeAIUsage(usage)
buildToolSet(items)