构建按 Token 计费的 AI 工具

你将构建什么

一个 AI 写作助手：用户输入主题，AI 生成文章草稿。每次调用按实际 token 消耗计费——你只需要一个 API Key，Profy SDK 处理模型调用和用量统计。最终效果：

使用 Profy SDK 的 chat.completions.create 调用 AI 模型
支持流式和非流式两种模式
SDK 自动处理 SSE 解析、错误映射和超时
Token 用量在响应中返回，平台自动统计

调用方式对比

维度	SDK `chat.completions`（推荐）	直接 HTTP	OpenAI SDK 兼容
代码量	最少——SDK 封装了鉴权、解析、错误	最多——需手动处理 SSE、错误	中等——需配置 baseURL
鉴权	API Key 自动注入	手动设 Authorization header	传 apiKey
错误处理	类型化异常	手动解析 status code	OpenAI 异常体系
流式	`AsyncGenerator<ChatChunk>`	手动解析 SSE	OpenAI Stream

本教程使用 Profy SDK 的 chat.completions。如果你的场景需要调用已发布的专家（含 persona + tools + memory），参考专家调用。

前置条件

已有袋袋开发者账号
在 Platform Console 创建了 API Key（sk-pro- 开头）
安装了 SDK：npm install @profy-ai/sdk 或 pip install profy-sdk

Step 1: 初始化客户端

import { Profy } from "@profy-ai/sdk";

const client = new Profy({
  apiKey: process.env.PROFY_API_KEY!,
});

from profy import Profy

client = Profy(api_key="sk-pro-...")

API Key 是服务端凭证，不要暴露到前端代码或 Git 仓库中。通过环境变量注入。

Step 2: 非流式调用

const response = await client.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "你是一个专业的写作助手。" },
    { role: "user", content: "写一篇关于 AI 未来趋势的文章" },
  ],
  temperature: 0.7,
  maxTokens: 2000,
});

console.log(response.text);
console.log(`Tokens: ${response.usage.totalTokens}`);

async with Profy(api_key="sk-pro-...") as client:
    response = await client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "你是一个专业的写作助手。"},
            {"role": "user", "content": "写一篇关于 AI 未来趋势的文章"},
        ],
        temperature=0.7,
        max_tokens=2000,
    )

    print(response.text)
    print(f"Tokens: {response.usage.total_tokens}")

ChatResponse 的关键字段：

字段	类型	说明
`text`	`string`	完整的生成文本
`usage.promptTokens`	`number`	输入 token 数
`usage.completionTokens`	`number`	输出 token 数
`usage.totalTokens`	`number`	总 token 数
`model`	`string`	实际使用的模型

Step 3: 流式调用

设置 stream: true 获取实时流式输出，适合打字机效果的聊天界面。

const stream = await client.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "你是一个专业的写作助手。" },
    { role: "user", content: "写一篇关于 AI 的文章" },
  ],
  stream: true,
  temperature: 0.7,
  maxTokens: 2000,
});

for await (const chunk of stream) {
  if (chunk.text) {
    process.stdout.write(chunk.text);
  }
}

async with Profy(api_key="sk-pro-...") as client:
    stream = await client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "你是一个专业的写作助手。"},
            {"role": "user", "content": "写一篇关于 AI 的文章"},
        ],
        stream=True,
        temperature=0.7,
        max_tokens=2000,
    )

    async for chunk in stream:
        if chunk.text:
            print(chunk.text, end="", flush=True)

Step 4: 错误处理

SDK 根据 HTTP 状态码抛出类型化异常：

异常类	HTTP 状态码	含义	处理方式
`InvalidRequestError`	400	请求参数错误或模型不可用	检查 `model` 名称和请求体
`AuthenticationError`	401	API Key 无效	检查 Key 配置
`InsufficientBalanceError`	402	余额不足	提示充值，不重试
`RateLimitError`	429	请求频率过高	指数退避重试

import {
  Profy,
  AuthenticationError,
  InsufficientBalanceError,
  RateLimitError,
  InvalidRequestError,
} from "@profy-ai/sdk";

async function chatWithRetry(
  client: Profy,
  messages: Array<{ role: string; content: string }>,
  maxRetries = 3,
) {
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await client.chat.completions.create({
        model: "deepseek-chat",
        messages,
      });
    } catch (err) {
      if (err instanceof InsufficientBalanceError) {
        throw new Error("余额不足，请充值后重试");
      }
      if (err instanceof RateLimitError && attempt < maxRetries) {
        const delay = Math.min(1000 * 2 ** attempt, 10_000);
        await new Promise((r) => setTimeout(r, delay));
        continue;
      }
      if (err instanceof InvalidRequestError) {
        throw new Error(`请求参数错误: ${err.message}`);
      }
      throw err;
    }
  }
}

import asyncio
from profy import (
    Profy,
    InsufficientBalanceError,
    RateLimitError,
    InvalidRequestError,
)

async def chat_with_retry(
    client: Profy,
    messages: list[dict],
    max_retries: int = 3,
):
    for attempt in range(max_retries + 1):
        try:
            return await client.chat.completions.create(
                model="deepseek-chat",
                messages=messages,
            )
        except InsufficientBalanceError:
            raise ValueError("余额不足，请充值后重试")
        except RateLimitError:
            if attempt < max_retries:
                delay = min(1.0 * 2**attempt, 10.0)
                await asyncio.sleep(delay)
                continue
            raise
        except InvalidRequestError as e:
            raise ValueError(f"请求参数错误: {e}") from e

收到 InsufficientBalanceError（402）时不要重试——余额不足不会因为重试而改变。向用户展示充值入口。

Step 5: 查询可用模型

const models = await client.models.list();
for (const model of models) {
  console.log(`${model.id} — ${model.owned_by}`);
}

models = await client.models.list()
for model in models:
    print(f"{model['id']} — {model.get('owned_by', 'unknown')}")

可用模型列表取决于平台管理员的配置。常见模型包括 deepseek-chat、deepseek-reasoner、qwen-plus 等。

完整示例

一个 AI 写作助手后端，整合了流式输出和错误处理。

import express from "express";
import { Profy, InsufficientBalanceError } from "@profy-ai/sdk";

const app = express();
app.use(express.json());

const client = new Profy({ apiKey: process.env.PROFY_API_KEY! });

app.post("/api/chat", async (req, res) => {
  try {
    const stream = await client.chat.completions.create({
      model: "deepseek-chat",
      messages: [
        { role: "system", content: "你是一个专业的写作助手。" },
        ...req.body.messages,
      ],
      stream: true,
      temperature: 0.7,
      maxTokens: 2000,
    });

    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");

    for await (const chunk of stream) {
      if (chunk.text) {
        res.write(`data: ${JSON.stringify({ content: chunk.text })}\n\n`);
      }
    }

    res.write("data: [DONE]\n\n");
    res.end();
  } catch (err) {
    if (err instanceof InsufficientBalanceError) {
      return res.status(402).json({ error: "余额不足，请充值" });
    }
    return res.status(500).json({ error: String(err) });
  }
});

app.listen(3000);

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from profy import Profy, InsufficientBalanceError

app = FastAPI()
client = Profy(api_key="sk-pro-...")

class ChatRequest(BaseModel):
    message: str

@app.post("/api/chat")
async def chat(body: ChatRequest):
    async def stream_response():
        try:
            stream = await client.chat.completions.create(
                model="deepseek-chat",
                messages=[
                    {"role": "system", "content": "你是一个专业的写作助手。"},
                    {"role": "user", "content": body.message},
                ],
                stream=True,
                temperature=0.7,
                max_tokens=2000,
            )
            async for chunk in stream:
                if chunk.text:
                    yield f"data: {json.dumps({'content': chunk.text})}\n\n"
            yield "data: [DONE]\n\n"
        except InsufficientBalanceError:
            yield f"data: {json.dumps({'error': '余额不足，请充值'})}\n\n"

    return StreamingResponse(stream_response(), media_type="text/event-stream")

OpenAI SDK 兼容方式（备选）

Profy 的 /v1/chat/completions 端点兼容 OpenAI 协议，可以直接使用 OpenAI 官方 SDK：

import OpenAI from "openai";

const openai = new OpenAI({
  apiKey: process.env.PROFY_API_KEY!,
  baseURL: "https://api.profy.cn/v1",
});

const completion = await openai.chat.completions.create({
  model: "deepseek-chat",
  messages: [{ role: "user", content: "写一段产品描述" }],
});

console.log(completion.choices[0].message.content);

from openai import AsyncOpenAI

openai_client = AsyncOpenAI(
    api_key="sk-pro-...",
    base_url="https://api.profy.cn/v1",
)

completion = await openai_client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "写一段产品描述"}],
)

print(completion.choices[0].message.content)

推荐使用 Profy SDK——它提供了类型化异常、专家调用等 Profy 特有功能。OpenAI SDK 方式适合从 OpenAI 迁移过来的项目。

下一步

按次计费实战

使用 ProfyApp report_event 实现按次扣费

专家调用

调用平台上已发布的 AI 专家

Next.js 全栈集成

完整的 OAuth + 计费 + AI 调用全栈应用

SDK 快速开始

安装 SDK、创建 API Key、首次调用

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

构建按 Token 计费的 AI 工具

你将构建什么

调用方式对比

前置条件

Step 1: 初始化客户端

Step 2: 非流式调用

Step 3: 流式调用

Step 4: 错误处理

Step 5: 查询可用模型

完整示例

OpenAI SDK 兼容方式（备选）

下一步

按次计费实战

专家调用

Next.js 全栈集成

SDK 快速开始

​你将构建什么

​调用方式对比

​前置条件

​Step 1: 初始化客户端

​Step 2: 非流式调用

​Step 3: 流式调用

​Step 4: 错误处理

​Step 5: 查询可用模型

​完整示例

​OpenAI SDK 兼容方式（备选）

​下一步

按次计费实战

专家调用

Next.js 全栈集成

SDK 快速开始

你将构建什么

调用方式对比

前置条件

Step 1: 初始化客户端

Step 2: 非流式调用

Step 3: 流式调用

Step 4: 错误处理

Step 5: 查询可用模型

完整示例

OpenAI SDK 兼容方式（备选）

下一步