Skip to content

GitLab

Commit 058c3bd8 authored by 陈曦's avatar 陈曦
Browse files

添加支持模型的调用API文档

parent f9944efd
Pipeline #82018 failed with stage
in 1 minute and 2 seconds
Hide whitespace changes
Inline Side-by-side
TrafficAPI调用文档.md 0 → 100644
View file @ 058c3bd8
# TrafficAPI 调用文档
> **适用版本**: TrafficAPI v0.1.1+
> **文档日期**: 2026-04-09
---
## 目录
1. [基本信息](#1-基本信息)
2. [认证方式](#2-认证方式)
3. [Claude / Anthropic API](#3-claude--anthropic-api)
4. [Gemini API](#4-gemini-api)
5. [OpenAI 兼容 API](#5-openai-兼容-api)
6. [Antigravity 路由（Claude + Gemini 混合）](#6-antigravity-路由claude--gemini-混合)
7. [公共接口](#7-公共接口)
8. [错误处理](#8-错误处理)
---
## 1. 基本信息
### 1.1 Base URL
```
https://<your-trafficapi-domain>
```
### 1.2 支持的提供商与路由
| 提供商 | 平台标识 | 主要路由前缀 |
|--------|----------|-------------|
| Anthropic (Claude) | `anthropic` | `/v1` |
| Google Gemini | `gemini` | `/v1beta` |
| OpenAI | `openai` | `/v1` (兼容层) |
| Antigravity (混合) | `antigravity` | `/antigravity` |
### 1.3 支持的模型
**Claude 系列**
| 模型名 | 说明 |
|--------|------|
| `claude-opus-4-6` | Opus 最强版 |
| `claude-opus-4-6-thinking` | Opus 扩展思考版 |
| `claude-sonnet-4-6` | Sonnet 均衡版 |
| `claude-sonnet-4-5` | Sonnet 上一代 |
| `claude-sonnet-4-5-thinking` | Sonnet 扩展思考版 |
| `claude-haiku-4-5` | Haiku 轻量版 |
**Gemini 系列**
| 模型名 | 说明 |
|--------|------|
| `gemini-2.5-flash` | Gemini 2.5 快速版 |
| `gemini-2.5-pro` | Gemini 2.5 旗舰版 |
| `gemini-2.5-flash-thinking` | Gemini 2.5 思考版 |
| `gemini-3-flash` | Gemini 3 快速版 |
| `gemini-3-pro-high` | Gemini 3 Pro 高性能 |
| `gemini-3-pro-low` | Gemini 3 Pro 经济版 |
| `gemini-3.1-pro-high` | Gemini 3.1 Pro 高性能 |
---
## 2. 认证方式
所有 AI 调用接口均通过 **API Key** 鉴权，支持以下三种方式（任选其一）：
| 方式 | Header/参数 | 示例 |
|------|------------|------|
| Bearer Token（推荐） | `Authorization` | `Authorization: Bearer sk-xxxxx` |
| x-api-key | `x-api-key` | `x-api-key: sk-xxxxx` |
| Gemini 兼容 | `x-goog-api-key` | `x-goog-api-key: sk-xxxxx` |
> **注意**: API Key 在平台控制台创建，格式通常为 `sk-` 前缀的字符串。
---
## 3. Claude / Anthropic API
完全兼容 Anthropic 官方 API 格式，可直接使用官方 SDK，只需替换 `base_url`。
### 3.0 请求参数说明
#### 通用 Header 参数
| Header | 必填 | 说明 |
|--------|------|------|
| `Authorization` | 是¹ | Bearer 认证，格式：`Bearer sk-xxxxx` |
| `x-api-key` | 是¹ | API Key 直接放入 header，与 `Authorization` 二选一 |
| `Content-Type` | 是 | 固定为 `application/json` |
| `anthropic-version` | 否 | Anthropic API 版本，推荐填写 `2023-06-01` |
| `anthropic-beta` | 否 | 开启 Beta 特性，如 `interleaved-thinking-2025-05-14`（扩展思考）、`prompt-caching-2024-07-31`（提示缓存） |
| `User-Agent` | 否 | 客户端标识，影响粘性会话（Sticky Session）亲和力 |
> ¹ `Authorization`（Bearer）、`x-api-key`、`x-goog-api-key` 三者填一即可。
---
#### `POST /v1/messages` Body 参数
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| `model` | string | **是** | — | 模型名称，如 `claude-sonnet-4-6` |
| `messages` | array | **是** | — | 对话消息数组，至少一条 |
| `messages[].role` | string | **是** | — | 消息角色，`user` 或 `assistant` |
| `messages[].content` | string \| array | **是** | — | 消息内容；字符串或内容块数组（支持 `text`、`image`、`tool_use`、`tool_result` 等类型） |
| `max_tokens` | integer | **是** | — | 最大输出 token 数，不同模型上限不同（Opus/Sonnet: 32000，Haiku: 8192） |
| `system` | string \| array | 否 | — | 系统提示；字符串或内容块数组（支持缓存控制 `cache_control`） |
| `stream` | boolean | 否 | `false` | 是否开启 SSE 流式输出 |
| `temperature` | number | 否 | `1.0` | 随机性控制，范围 `0.0~1.0`；越低越确定性 |
| `top_p` | number | 否 | — | Nucleus 采样概率阈值，范围 `0.0~1.0` |
| `top_k` | integer | 否 | — | Top-K 采样，仅保留概率最高的 K 个 token |
| `stop_sequences` | array\<string\> | 否 | — | 遇到这些字符串时停止生成 |
| `tools` | array | 否 | — | 工具定义列表（Function Calling）；每项包含 `name`、`description`、`input_schema` |
| `tool_choice` | object | 否 | — | 工具调用策略：`{"type": "auto"}` / `{"type": "any"}` / `{"type": "tool", "name": "..."}` / `{"type": "none"}` |
| `thinking` | object | 否 | — | 扩展思考配置（需同时传 `anthropic-beta: interleaved-thinking-2025-05-14`） |
| `thinking.type` | string | 否 | — | `enabled`（固定预算）或 `adaptive`（自适应预算） |
| `thinking.budget_tokens` | integer | 否 | — | 思考 token 预算，最小 `1024`，推荐 `32000` |
| `metadata` | object | 否 | — | 请求元数据 |
| `metadata.user_id` | string | 否 | — | 用户标识，用于粘性会话（Sticky Session）路由亲和 |
| `output_config` | object | 否 | — | 输出配置 |
| `output_config.effort` | string | 否 | — | 推理强度：`low` / `medium` / `high` / `max` |
| `context_management` | object | 否 | — | 上下文管理策略（扩展 Beta 功能） |
**`tools` 单项结构**
```json
{
"name": "tool_name",
"description": "工具描述",
"input_schema": {
"type": "object",
"properties": {
"param1": { "type": "string", "description": "参数说明" }
},
"required": ["param1"]
}
}
```
**响应体结构（非流式）**
```json
{
"id": "msg_xxxxx",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [
{ "type": "text", "text": "回复内容" }
],
"stop_reason": "end_turn",
"stop_sequence": null,
"usage": {
"input_tokens": 25,
"output_tokens": 80,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0
}
}
```
| `stop_reason` 取值 | 说明 |
|-------------------|------|
| `end_turn` | 模型正常结束 |
| `max_tokens` | 达到 `max_tokens` 上限 |
| `stop_sequence` | 命中 `stop_sequences` 中的字符串 |
| `tool_use` | 模型请求调用工具 |
**SSE 流式事件序列（`stream: true`）**
```
event: message_start # 消息开始，包含 usage 初始值
event: content_block_start # 内容块开始
event: content_block_delta # 内容增量（text_delta / thinking_delta）
event: content_block_stop # 内容块结束
event: message_delta # 消息结束，包含最终 stop_reason 和 usage
event: message_stop # 流结束标志
```
---
#### `POST /v1/messages/count_tokens` Body 参数
与 `/v1/messages` 相同，但忽略 `stream` 字段，且不实际调用模型，不计费。
---
### 3.1 发送消息 `POST /v1/messages`
#### Curl
```bash
BASE_URL="https://<your-trafficapi-domain>"
API_KEY="sk-xxxxx"
curl -X POST "$BASE_URL/v1/messages" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "你好，介绍一下自己。"
}
]
}'
```
**流式输出（SSE）**
```bash
curl -X POST "$BASE_URL/v1/messages" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
--no-buffer \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"stream": true,
"messages": [
{"role": "user", "content": "写一首七言绝句"}
]
}'
```
**多轮对话**
```bash
curl -X POST "$BASE_URL/v1/messages" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 2048,
"system": "你是一位专业的代码审查员，回答简洁精准。",
"messages": [
{"role": "user", "content": "如何优化 Python 列表推导式？"},
{"role": "assistant", "content": "列表推导式比等效的 for 循环快约 35%，因为..."},
{"role": "user", "content": "给我一个具体例子"}
]
}'
```
#### Python（官方 SDK）
```python
import anthropic
client = anthropic.Anthropic(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>",
)
# 普通调用
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "你好，介绍一下自己。"}
]
)
print(message.content[0].text)
# 系统提示 + 多轮对话
message = client.messages.create(
model="claude-opus-4-6",
max_tokens=2048,
system="你是一位资深 Go 语言专家。",
messages=[
{"role": "user", "content": "解释 goroutine 和 channel 的关系"},
]
)
print(message.content[0].text)
```
**流式输出**
```python
import anthropic
client = anthropic.Anthropic(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>",
)
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "写一首关于春天的诗"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # 换行
```
**工具调用（Function Calling）**
```python
import anthropic
import json
client = anthropic.Anthropic(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>",
)
tools = [
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"input_schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，例如：北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
]
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
tools=tools,
messages=[{"role": "user", "content": "北京今天天气怎么样？"}]
)
# 处理工具调用结果
for block in response.content:
if block.type == "tool_use":
print(f"调用工具: {block.name}")
print(f"参数: {json.dumps(block.input, ensure_ascii=False, indent=2)}")
```
#### Go（官方 SDK）
```go
package main
import (
"context"
"fmt"
"log"
"github.com/anthropics/anthropic-sdk-go"
"github.com/anthropics/anthropic-sdk-go/option"
)
func main() {
client := anthropic.NewClient(
option.WithAPIKey("sk-xxxxx"),
option.WithBaseURL("https://<your-trafficapi-domain>"),
)
// 普通调用
message, err := client.Messages.New(context.Background(), anthropic.MessageNewParams{
Model: anthropic.F(anthropic.ModelClaudeSonnet4_6),
MaxTokens: anthropic.F(int64(1024)),
Messages: anthropic.F([]anthropic.MessageParam{
anthropic.UserMessageParam([]anthropic.ContentBlockParamUnion{
anthropic.TextBlockParam{Text: anthropic.F("你好，介绍一下自己。")},
}),
}),
})
if err != nil {
log.Fatal(err)
}
fmt.Println(message.Content[0].(anthropic.TextBlock).Text)
}
```
**流式输出（Go）**
```go
package main
import (
"context"
"fmt"
"log"
"github.com/anthropics/anthropic-sdk-go"
"github.com/anthropics/anthropic-sdk-go/option"
)
func main() {
client := anthropic.NewClient(
option.WithAPIKey("sk-xxxxx"),
option.WithBaseURL("https://<your-trafficapi-domain>"),
)
stream := client.Messages.NewStreaming(context.Background(), anthropic.MessageNewParams{
Model: anthropic.F(anthropic.ModelClaudeSonnet4_6),
MaxTokens: anthropic.F(int64(1024)),
Messages: anthropic.F([]anthropic.MessageParam{
anthropic.UserMessageParam([]anthropic.ContentBlockParamUnion{
anthropic.TextBlockParam{Text: anthropic.F("写一首关于春天的诗")},
}),
}),
})
for stream.Next() {
event := stream.Current()
switch delta := event.Delta.(type) {
case anthropic.ContentBlockDeltaEventDelta:
if textDelta, ok := delta.AsUnion().(anthropic.TextDelta); ok {
fmt.Print(textDelta.Text)
}
}
}
fmt.Println()
if err := stream.Err(); err != nil {
log.Fatal(err)
}
}
```
**纯 HTTP 调用（Go，无 SDK）**
```go
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
)
type MessageRequest struct {
Model string `json:"model"`
MaxTokens int `json:"max_tokens"`
Messages []Message `json:"messages"`
Stream bool `json:"stream,omitempty"`
}
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
func main() {
const (
baseURL = "https://<your-trafficapi-domain>"
apiKey = "sk-xxxxx"
)
reqBody := MessageRequest{
Model: "claude-sonnet-4-6",
MaxTokens: 1024,
Messages: []Message{
{Role: "user", Content: "你好，介绍一下自己。"},
},
}
data, _ := json.Marshal(reqBody)
req, _ := http.NewRequest("POST", baseURL+"/v1/messages", bytes.NewReader(data))
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("anthropic-version", "2023-06-01")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
fmt.Println(string(body))
}
```
### 3.2 Token 计数 `POST /v1/messages/count_tokens`
#### Curl
```bash
curl -X POST "$BASE_URL/v1/messages/count_tokens" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-sonnet-4-6",
"messages": [
{"role": "user", "content": "你好，介绍一下自己。"}
]
}'
```
**响应**
```json
{
"input_tokens": 18
}
```
#### Python
```python
response = client.messages.count_tokens(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "你好，介绍一下自己。"}],
)
print(f"Token 数: {response.input_tokens}")
```
#### Go
```go
count, err := client.Messages.CountTokens(context.Background(), anthropic.MessageCountTokensParams{
Model: anthropic.F(anthropic.ModelClaudeSonnet4_6),
Messages: anthropic.F([]anthropic.MessageParam{
anthropic.UserMessageParam([]anthropic.ContentBlockParamUnion{
anthropic.TextBlockParam{Text: anthropic.F("你好，介绍一下自己。")},
}),
}),
})
if err != nil {
log.Fatal(err)
}
fmt.Printf("Token 数: %d\n", count.InputTokens)
```
### 3.3 列出模型 `GET /v1/models`
#### Curl
```bash
curl "$BASE_URL/v1/models" \
-H "Authorization: Bearer $API_KEY" \
-H "anthropic-version: 2023-06-01"
```
---
## 4. Gemini API
完全兼容 Google Gemini 原生 API 格式。
### 4.0 请求参数说明
#### 通用 Header 参数
| Header | 必填 | 说明 |
|--------|------|------|
| `x-goog-api-key` | 是¹ | Gemini 兼容认证方式（推荐） |
| `Authorization` | 是¹ | Bearer 认证，格式：`Bearer sk-xxxxx` |
| `x-api-key` | 是¹ | 与 `Authorization` 等价 |
| `Content-Type` | 是 | 固定为 `application/json` |
> ¹ 三种认证方式选一即可。
---
#### `POST /v1beta/models/{model}:generateContent` 路径参数
| 参数 | 位置 | 必填 | 说明 |
|------|------|------|------|
| `{model}` | URL 路径 | **是** | 模型名，如 `gemini-2.5-flash`、`gemini-2.5-pro` |
---
#### `POST /v1beta/models/{model}:generateContent` Body 参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `contents` | array | **是** | 对话消息数组 |
| `contents[].role` | string | **是** | 消息角色：`user` 或 `model` |
| `contents[].parts` | array | **是** | 内容块数组 |
| `contents[].parts[].text` | string | 否 | 文本内容 |
| `contents[].parts[].inlineData` | object | 否 | 内联媒体数据（图片等），含 `mimeType` 和 `data`（base64） |
| `system_instruction` | object | 否 | 系统指令（等同于 system prompt） |
| `system_instruction.parts` | array | 否 | 系统指令内容块，结构同 `contents[].parts` |
| `generationConfig` | object | 否 | 生成配置 |
| `generationConfig.temperature` | number | 否 | 随机性，范围 `0.0~2.0` |
| `generationConfig.maxOutputTokens` | integer | 否 | 最大输出 token 数 |
| `generationConfig.topP` | number | 否 | Nucleus 采样，范围 `0.0~1.0` |
| `generationConfig.topK` | integer | 否 | Top-K 采样 |
| `generationConfig.stopSequences` | array\<string\> | 否 | 遇到这些字符串时停止生成 |
| `generationConfig.candidateCount` | integer | 否 | 返回候选结果数量，默认 `1` |
| `tools` | array | 否 | 工具定义（Function Calling） |
| `tool_config` | object | 否 | 工具调用模式配置 |
| `tool_config.function_calling_config.mode` | string | 否 | `AUTO` / `ANY` / `NONE` |
| `safetySettings` | array | 否 | 安全过滤设置，每项含 `category` 和 `threshold` |
**响应体结构（非流式）**
```json
{
"candidates": [
{
"content": {
"parts": [{ "text": "回复内容" }],
"role": "model"
},
"finishReason": "STOP",
"index": 0
}
],
"usageMetadata": {
"promptTokenCount": 25,
"candidatesTokenCount": 80,
"totalTokenCount": 105
}
}
```
| `finishReason` 取值 | 说明 |
|--------------------|------|
| `STOP` | 正常结束 |
| `MAX_TOKENS` | 达到 `maxOutputTokens` 上限 |
| `SAFETY` | 触发安全过滤 |
| `RECITATION` | 内容引用检测 |
> **流式端点**：将路径中的 `generateContent` 替换为 `streamGenerateContent`，响应为 SSE 流，每条数据为完整的 JSON 对象（非增量），客户端需自行拼接文本。
---
### 4.1 生成内容 `POST /v1beta/models/{model}:generateContent`
#### Curl
```bash
BASE_URL="https://<your-trafficapi-domain>"
API_KEY="sk-xxxxx"
MODEL="gemini-2.5-flash"
# x-api-key 方式（推荐用于 Gemini）
curl -X POST "$BASE_URL/v1beta/models/$MODEL:generateContent" \
-H "x-goog-api-key: $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"contents": [
{
"role": "user",
"parts": [{"text": "你好，介绍一下自己。"}]
}
]
}'
```
**含系统指令**
```bash
curl -X POST "$BASE_URL/v1beta/models/$MODEL:generateContent" \
-H "x-goog-api-key: $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"system_instruction": {
"parts": [{"text": "你是一位精通中文的写作助手，回答简洁优美。"}]
},
"contents": [
{
"role": "user",
"parts": [{"text": "写一首关于秋天的五言绝句"}]
}
],
"generationConfig": {
"temperature": 0.9,
"maxOutputTokens": 512
}
}'
```
**流式输出**
```bash
curl -X POST "$BASE_URL/v1beta/models/$MODEL:streamGenerateContent" \
-H "x-goog-api-key: $API_KEY" \
-H "Content-Type: application/json" \
--no-buffer \
-d '{
"contents": [
{"role": "user", "parts": [{"text": "解释量子纠缠"}]}
]
}'
```
#### Python（google-generativeai SDK）
```python
import google.generativeai as genai
genai.configure(
api_key="sk-xxxxx",
# 通过环境变量或直接设置 transport
# 需要将请求路由到 TrafficAPI
)
# 推荐：使用 google.generativeai 的 transport 覆盖
import google.ai.generativelanguage as glm
from google.api_core import gapic_v1
from google.auth.credentials import AnonymousCredentials
# 方式一：直接使用 HTTP 请求（最简单）
import httpx
BASE_URL = "https://<your-trafficapi-domain>"
API_KEY = "sk-xxxxx"
MODEL = "gemini-2.5-flash"
response = httpx.post(
f"{BASE_URL}/v1beta/models/{MODEL}:generateContent",
headers={"x-goog-api-key": API_KEY},
json={
"contents": [
{"role": "user", "parts": [{"text": "你好，介绍一下自己。"}]}
]
}
)
data = response.json()
print(data["candidates"][0]["content"]["parts"][0]["text"])
```
**流式输出（Python HTTP）**
```python
import httpx
import json
BASE_URL = "https://<your-trafficapi-domain>"
API_KEY = "sk-xxxxx"
MODEL = "gemini-2.5-flash"
with httpx.stream(
"POST",
f"{BASE_URL}/v1beta/models/{MODEL}:streamGenerateContent",
headers={"x-goog-api-key": API_KEY, "Content-Type": "application/json"},
json={
"contents": [
{"role": "user", "parts": [{"text": "解释量子纠缠，500字以内"}]}
]
}
) as stream:
for line in stream.iter_lines():
line = line.strip()
if not line or line == "data: [DONE]":
continue
if line.startswith("data: "):
line = line[6:]
try:
chunk = json.loads(line)
for candidate in chunk.get("candidates", []):
for part in candidate.get("content", {}).get("parts", []):
if "text" in part:
print(part["text"], end="", flush=True)
except json.JSONDecodeError:
pass
print()
```
#### Go（纯 HTTP）
```go
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
)
type GeminiRequest struct {
Contents []GeminiContent `json:"contents"`
GenerationConfig *GeminiGenerationConfig `json:"generationConfig,omitempty"`
}
type GeminiContent struct {
Role string `json:"role"`
Parts []GeminiPart `json:"parts"`
}
type GeminiPart struct {
Text string `json:"text"`
}
type GeminiGenerationConfig struct {
Temperature float64 `json:"temperature,omitempty"`
MaxOutputTokens int `json:"maxOutputTokens,omitempty"`
}
func main() {
const (
baseURL = "https://<your-trafficapi-domain>"
apiKey = "sk-xxxxx"
model = "gemini-2.5-flash"
)
reqBody := GeminiRequest{
Contents: []GeminiContent{
{
Role: "user",
Parts: []GeminiPart{{Text: "你好，介绍一下自己。"}},
},
},
GenerationConfig: &GeminiGenerationConfig{
Temperature: 0.7,
MaxOutputTokens: 1024,
},
}
data, _ := json.Marshal(reqBody)
url := fmt.Sprintf("%s/v1beta/models/%s:generateContent", baseURL, model)
req, _ := http.NewRequest("POST", url, bytes.NewReader(data))
req.Header.Set("x-goog-api-key", apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
var result map[string]any
json.NewDecoder(resp.Body).Decode(&result)
candidates := result["candidates"].([]any)
content := candidates[0].(map[string]any)["content"].(map[string]any)
parts := content["parts"].([]any)
text := parts[0].(map[string]any)["text"].(string)
fmt.Println(text)
}
```
**流式输出（Go）**
```go
package main
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"log"
"net/http"
"strings"
)
func main() {
const (
baseURL = "https://<your-trafficapi-domain>"
apiKey = "sk-xxxxx"
model = "gemini-2.5-flash"
)
body := map[string]any{
"contents": []map[string]any{
{
"role": "user",
"parts": []map[string]any{{"text": "解释量子纠缠，500字以内"}},
},
},
}
data, _ := json.Marshal(body)
url := fmt.Sprintf("%s/v1beta/models/%s:streamGenerateContent", baseURL, model)
req, _ := http.NewRequest("POST", url, bytes.NewReader(data))
req.Header.Set("x-goog-api-key", apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
raw := strings.TrimPrefix(line, "data: ")
if raw == "[DONE]" {
break
}
var chunk map[string]any
if err := json.Unmarshal([]byte(raw), &chunk); err != nil {
continue
}
candidates, _ := chunk["candidates"].([]any)
for _, c := range candidates {
cd := c.(map[string]any)
content := cd["content"].(map[string]any)
parts := content["parts"].([]any)
for _, p := range parts {
if text, ok := p.(map[string]any)["text"].(string); ok {
fmt.Print(text)
}
}
}
}
fmt.Println()
}
```
### 4.2 列出模型 `GET /v1beta/models`
#### Curl
```bash
curl "$BASE_URL/v1beta/models" \
-H "x-goog-api-key: $API_KEY"
```
---
## 5. OpenAI 兼容 API
完全兼容 OpenAI Chat Completions 格式，可直接使用 OpenAI 官方 SDK，只需替换 `base_url`。
后端会自动将请求路由到相应平台（Claude/OpenAI），并将响应转换回 OpenAI 格式。
### 5.0 请求参数说明
#### 通用 Header 参数
| Header | 必填 | 说明 |
|--------|------|------|
| `Authorization` | 是 | Bearer 认证，格式：`Bearer sk-xxxxx` |
| `Content-Type` | 是 | 固定为 `application/json` |
| `User-Agent` | 否 | 客户端标识，影响粘性会话亲和力 |
---
#### `POST /v1/chat/completions` Body 参数
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| `model` | string | **是** | — | 模型名，如 `claude-sonnet-4-6`、`gpt-oss-120b-medium` |
| `messages` | array | **是** | — | 对话消息数组 |
| `messages[].role` | string | **是** | — | 消息角色：`system` / `user` / `assistant` / `tool` |
| `messages[].content` | string \| array | **是** | — | 消息内容；支持多模态内容块（`type: text` / `type: image_url`） |
| `messages[].name` | string | 否 | — | 发送者名称（`tool` 角色时为工具名） |
| `messages[].tool_call_id` | string | 否 | — | 工具调用 ID（`tool` 角色回复时使用） |
| `max_tokens` | integer | 否 | 模型默认 | 最大输出 token 数 |
| `stream` | boolean | 否 | `false` | 是否开启 SSE 流式输出 |
| `temperature` | number | 否 | `1.0` | 随机性，范围 `0.0~2.0` |
| `top_p` | number | 否 | `1.0` | Nucleus 采样概率阈值 |
| `frequency_penalty` | number | 否 | `0.0` | 频率惩罚，范围 `-2.0~2.0`，减少重复词 |
| `presence_penalty` | number | 否 | `0.0` | 存在惩罚，范围 `-2.0~2.0`，鼓励新话题 |
| `stop` | string \| array | 否 | — | 遇到这些字符串时停止生成，最多 4 项 |
| `n` | integer | 否 | `1` | 生成候选回复数量 |
| `tools` | array | 否 | — | 工具定义列表（Function Calling） |
| `tools[].type` | string | 否 | — | 固定为 `"function"` |
| `tools[].function.name` | string | 否 | — | 工具名称 |
| `tools[].function.description` | string | 否 | — | 工具描述 |
| `tools[].function.parameters` | object | 否 | — | 工具参数 JSON Schema |
| `tool_choice` | string \| object | 否 | `"auto"` | 工具调用策略：`"none"` / `"auto"` / `"required"` / `{"type": "function", "function": {"name": "..."}}` |
| `response_format` | object | 否 | — | 响应格式，如 `{"type": "json_object"}` 强制 JSON 输出 |
| `seed` | integer | 否 | — | 随机种子（尽力保证确定性输出） |
| `user` | string | 否 | — | 终端用户标识，用于监控滥用行为 |
| `stream_options` | object | 否 | — | 流式选项，如 `{"include_usage": true}` 在流末追加用量统计 |
**响应体结构（非流式）**
```json
{
"id": "chatcmpl-xxxxx",
"object": "chat.completion",
"created": 1700000000,
"model": "claude-sonnet-4-6",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "回复内容"
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 25,
"completion_tokens": 80,
"total_tokens": 105
}
}
```
| `finish_reason` 取值 | 说明 |
|---------------------|------|
| `stop` | 模型正常结束 |
| `length` | 达到 `max_tokens` 上限 |
| `tool_calls` | 模型请求调用工具 |
| `content_filter` | 内容过滤触发 |
**SSE 流式数据块格式（`stream: true`）**
```
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"你好"},"finish_reason":null}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
data: [DONE]
```
---
### 5.1 Chat Completions `POST /v1/chat/completions`
#### Curl
```bash
BASE_URL="https://<your-trafficapi-domain>"
API_KEY="sk-xxxxx"
curl -X POST "$BASE_URL/v1/chat/completions" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"messages": [
{"role": "system", "content": "你是一位资深软件工程师。"},
{"role": "user", "content": "解释 CAP 定理"}
],
"max_tokens": 1024,
"temperature": 0.7
}'
```
**流式输出（SSE）**
```bash
curl -X POST "$BASE_URL/v1/chat/completions" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
--no-buffer \
-d '{
"model": "claude-sonnet-4-6",
"messages": [
{"role": "user", "content": "写一段 Python 快速排序代码"}
],
"stream": true
}'
```
**函数调用**
```bash
curl -X POST "$BASE_URL/v1/chat/completions" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"messages": [
{"role": "user", "content": "北京现在几点了？"}
],
"tools": [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取指定时区的当前时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如 Asia/Shanghai"
}
},
"required": ["timezone"]
}
}
}
],
"tool_choice": "auto"
}'
```
#### Python（OpenAI SDK）
```python
from openai import OpenAI
client = OpenAI(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>/v1",
)
# 普通调用
response = client.chat.completions.create(
model="claude-sonnet-4-6",
messages=[
{"role": "system", "content": "你是一位资深软件工程师。"},
{"role": "user", "content": "解释 CAP 定理"},
],
max_tokens=1024,
temperature=0.7,
)
print(response.choices[0].message.content)
# 流式输出
stream = client.chat.completions.create(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "写一段 Python 快速排序代码"}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
print()
```
**函数调用（Python OpenAI SDK）**
```python
from openai import OpenAI
import json
client = OpenAI(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>/v1",
)
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
},
"required": ["city"],
},
},
}
]
response = client.chat.completions.create(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "上海天气怎么样？"}],
tools=tools,
tool_choice="auto",
)
message = response.choices[0].message
if message.tool_calls:
for call in message.tool_calls:
print(f"工具: {call.function.name}")
print(f"参数: {json.loads(call.function.arguments)}")
```
#### Go（OpenAI SDK）
```go
package main
import (
"context"
"fmt"
"log"
"github.com/openai/openai-go"
"github.com/openai/openai-go/option"
)
func main() {
client := openai.NewClient(
option.WithAPIKey("sk-xxxxx"),
option.WithBaseURL("https://<your-trafficapi-domain>/v1"),
)
// 普通调用
completion, err := client.Chat.Completions.New(context.Background(), openai.ChatCompletionNewParams{
Model: openai.F("claude-sonnet-4-6"),
Messages: openai.F([]openai.ChatCompletionMessageParamUnion{
openai.SystemMessage("你是一位资深软件工程师。"),
openai.UserMessage("解释 CAP 定理"),
}),
MaxTokens: openai.F(int64(1024)),
Temperature: openai.F(0.7),
})
if err != nil {
log.Fatal(err)
}
fmt.Println(completion.Choices[0].Message.Content)
}
```
**流式输出（Go）**
```go
package main
import (
"context"
"fmt"
"log"
"github.com/openai/openai-go"
"github.com/openai/openai-go/option"
)
func main() {
client := openai.NewClient(
option.WithAPIKey("sk-xxxxx"),
option.WithBaseURL("https://<your-trafficapi-domain>/v1"),
)
stream := client.Chat.Completions.NewStreaming(context.Background(), openai.ChatCompletionNewParams{
Model: openai.F("claude-sonnet-4-6"),
Messages: openai.F([]openai.ChatCompletionMessageParamUnion{
openai.UserMessage("写一段 Go 实现的二叉树遍历"),
}),
})
for stream.Next() {
chunk := stream.Current()
if len(chunk.Choices) > 0 {
fmt.Print(chunk.Choices[0].Delta.Content)
}
}
fmt.Println()
if err := stream.Err(); err != nil {
log.Fatal(err)
}
}
```
**纯 HTTP 调用（Go，无 SDK）**
```go
package main
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"strings"
)
func main() {
const (
baseURL = "https://<your-trafficapi-domain>"
apiKey = "sk-xxxxx"
)
body := map[string]any{
"model": "claude-sonnet-4-6",
"messages": []map[string]string{
{"role": "user", "content": "解释 CAP 定理"},
},
"max_tokens": 1024,
"stream": true,
}
data, _ := json.Marshal(body)
req, _ := http.NewRequest("POST", baseURL+"/v1/chat/completions", bytes.NewReader(data))
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
raw := strings.TrimPrefix(line, "data: ")
if raw == "[DONE]" {
break
}
var chunk map[string]any
if err := json.Unmarshal([]byte(raw), &chunk); err != nil {
continue
}
choices := chunk["choices"].([]any)
delta := choices[0].(map[string]any)["delta"].(map[string]any)
if content, ok := delta["content"].(string); ok {
fmt.Print(content)
}
}
fmt.Println()
// 非流式调用
body2 := map[string]any{
"model": "claude-sonnet-4-6",
"messages": []map[string]string{
{"role": "user", "content": "你好"},
},
}
data2, _ := json.Marshal(body2)
req2, _ := http.NewRequest("POST", baseURL+"/v1/chat/completions", bytes.NewReader(data2))
req2.Header.Set("Authorization", "Bearer "+apiKey)
req2.Header.Set("Content-Type", "application/json")
resp2, err := http.DefaultClient.Do(req2)
if err != nil {
log.Fatal(err)
}
defer resp2.Body.Close()
body3, _ := io.ReadAll(resp2.Body)
var result map[string]any
json.Unmarshal(body3, &result)
choices := result["choices"].([]any)
message := choices[0].(map[string]any)["message"].(map[string]any)
fmt.Println(message["content"])
}
```
### 5.2 Responses API `POST /v1/responses`
Responses API 兼容 OpenAI Responses 格式（含 WebSocket 流式支持）。
```bash
curl -X POST "$BASE_URL/v1/responses" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"input": "你好，介绍一下自己。",
"stream": false
}'
```
---
## 6. Antigravity 路由（Claude + Gemini 混合）
Antigravity 平台同时支持 Claude 和 Gemini 模型，通过 `/antigravity` 前缀路由。
### 6.1 Claude via Antigravity
```bash
curl -X POST "$BASE_URL/antigravity/v1/messages" \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{"role": "user", "content": "你好"}
]
}'
```
### 6.2 Gemini via Antigravity
```bash
MODEL="gemini-2.5-flash"
curl -X POST "$BASE_URL/antigravity/v1beta/models/$MODEL:generateContent" \
-H "x-goog-api-key: $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"contents": [
{"role": "user", "parts": [{"text": "你好"}]}
]
}'
```
### 6.3 Antigravity 支持的模型（含别名映射）
| 请求模型名 | 实际路由模型 |
|-----------|-------------|
| `claude-opus-4-6` | `claude-opus-4-6-thinking` |
| `claude-sonnet-4-6` | `claude-sonnet-4-6` |
| `claude-haiku-4-5` | `claude-sonnet-4-6`（升级） |
| `gemini-2.5-flash` | `gemini-2.5-flash` |
| `gemini-2.5-pro` | `gemini-2.5-pro` |
| `gemini-3-flash-preview` | `gemini-3-flash` |
| `gemini-3-pro-preview` | `gemini-3-pro-high` |
| `gemini-3.1-pro-preview` | `gemini-3.1-pro-high` |
---
## 7. 公共接口
### 7.1 健康检查 `GET /health`
```bash
curl "$BASE_URL/health"
```
**响应**
```json
{"status": "ok"}
```
### 7.2 获取使用量 `GET /v1/usage`
```bash
curl "$BASE_URL/v1/usage" \
-H "Authorization: Bearer $API_KEY"
```
---
## 8. 错误处理
### 8.1 错误格式
不同平台返回不同格式的错误，网关会根据调用端点自动适配：
**Anthropic 格式**（`/v1/messages`）
```json
{
"type": "error",
"error": {
"type": "authentication_error",
"message": "invalid x-api-key"
}
}
```
**OpenAI 格式**（`/v1/chat/completions`）
```json
{
"error": {
"type": "authentication_error",
"message": "invalid api key"
}
}
```
**Gemini 格式**（`/v1beta/...`）
```json
{
"error": {
"code": 401,
"message": "API key not valid.",
"status": "UNAUTHENTICATED"
}
}
```
### 8.2 常见 HTTP 状态码
| 状态码 | 含义 | 处理建议 |
|--------|------|----------|
| `200` | 成功 | - |
| `400` | 请求格式错误 | 检查请求体参数 |
| `401` | 认证失败 | 检查 API Key 是否正确 |
| `403` | 权限不足 / IP 被限制 | 联系管理员 |
| `429` | 请求过于频繁 / 并发超限 | 降低请求频率，等待后重试 |
| `500` | 服务器内部错误 | 联系管理员 |
| `529` | 上游服务过载 | 稍后重试 |
### 8.3 Python 错误处理示例
```python
import anthropic
client = anthropic.Anthropic(
api_key="sk-xxxxx",
base_url="https://<your-trafficapi-domain>",
)
try:
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "你好"}],
)
print(message.content[0].text)
except anthropic.AuthenticationError as e:
print(f"认证失败: {e}")
except anthropic.RateLimitError as e:
print(f"速率限制: {e}")
except anthropic.APIStatusError as e:
print(f"API 错误 {e.status_code}: {e.message}")
```
### 8.4 Go 错误处理示例
```go
message, err := client.Messages.New(ctx, params)
if err != nil {
var apiErr *anthropic.Error
if errors.As(err, &apiErr) {
switch apiErr.StatusCode {
case 401:
log.Println("认证失败，请检查 API Key")
case 429:
log.Println("请求频率过高，请稍后重试")
default:
log.Printf("API 错误 %d: %s", apiErr.StatusCode, apiErr.Message)
}
} else {
log.Printf("网络错误: %v", err)
}
return
}
```
---
## 附录：快速参考
### A. SDK 安装
```bash
# Python - Anthropic 官方 SDK
pip install anthropic
# Python - OpenAI 官方 SDK
pip install openai
# Python - HTTP 客户端
pip install httpx
# Go - Anthropic SDK
go get github.com/anthropics/anthropic-sdk-go
# Go - OpenAI SDK
go get github.com/openai/openai-go
```
### B. 环境变量配置（推荐）
```bash
# .env 或 shell 配置
export TRAFFICAPI_BASE_URL="https://<your-trafficapi-domain>"
export TRAFFICAPI_API_KEY="sk-xxxxx"
```
```python
import os
import anthropic
client = anthropic.Anthropic(
api_key=os.environ["TRAFFICAPI_API_KEY"],
base_url=os.environ["TRAFFICAPI_BASE_URL"],
)
```
### C. 各 API 路由速查
| 操作 | 方法 | 路径 |
|------|------|------|
| Claude 发送消息 | `POST` | `/v1/messages` |
| Claude 计算 Token | `POST` | `/v1/messages/count_tokens` |
| Claude 模型列表 | `GET` | `/v1/models` |
| Gemini 生成内容 | `POST` | `/v1beta/models/{model}:generateContent` |
| Gemini 流式生成 | `POST` | `/v1beta/models/{model}:streamGenerateContent` |
| Gemini 模型列表 | `GET` | `/v1beta/models` |
| OpenAI Chat | `POST` | `/v1/chat/completions` |
| OpenAI Responses | `POST` | `/v1/responses` |
| Antigravity Claude | `POST` | `/antigravity/v1/messages` |
| Antigravity Gemini | `POST` | `/antigravity/v1beta/models/{model}:generateContent` |
| 健康检查 | `GET` | `/health` |
| 使用量查询 | `GET` | `/v1/usage` |
frontend/src/components/layout/AppSidebar.vue
View file @ 058c3bd8
......@@ -385,6 +385,21 @@ const BellIcon = {
)
}
const BookOpenIcon = {
render: () =>
h(
'svg',
{ fill: 'none', viewBox: '0 0 24 24', stroke: 'currentColor', 'stroke-width': '1.5' },
[
h('path', {
'stroke-linecap': 'round',
'stroke-linejoin': 'round',
d: 'M12 6.042A8.967 8.967 0 006 3.75c-1.052 0-2.062.18-3 .512v14.25A8.987 8.987 0 016 18c2.305 0 4.408.867 6 2.292m0-14.25a8.966 8.966 0 016-2.292c1.052 0 2.062.18 3 .512v14.25A8.987 8.987 0 0018 18a8.967 8.967 0 00-6 2.292m0-14.25v14.25'
})
]
)
}
const TicketIcon = {
render: () =>
h(
......@@ -499,6 +514,7 @@ const userNavItems = computed((): NavItem[] => {
: []),
{ path: '/redeem', label: t('nav.redeem'), icon: GiftIcon, hideInSimpleMode: true },
{ path: '/profile', label: t('nav.profile'), icon: UserIcon },
{ path: '/docs', label: t('nav.docs'), icon: BookOpenIcon },
...customMenuItemsForUser.value.map((item): NavItem => ({
path: `/custom/${item.id}`,
label: item.label,
......@@ -527,6 +543,7 @@ const personalNavItems = computed((): NavItem[] => {
: []),
{ path: '/redeem', label: t('nav.redeem'), icon: GiftIcon, hideInSimpleMode: true },
{ path: '/profile', label: t('nav.profile'), icon: UserIcon },
{ path: '/docs', label: t('nav.docs'), icon: BookOpenIcon },
...customMenuItemsForUser.value.map((item): NavItem => ({
path: `/custom/${item.id}`,
label: item.label,
......
frontend/src/router/index.ts
View file @ 058c3bd8
......@@ -201,6 +201,17 @@ const routes: RouteRecordRaw[] = [
descriptionKey: 'purchase.description'
}
},
{
path: '/docs',
name: 'Docs',
component: () => import('@/views/user/DocsView.vue'),
meta: {
requiresAuth: true,
requiresAdmin: false,
title: 'API Docs',
titleKey: 'nav.docs'
}
},
{
path: '/custom/:id',
name: 'CustomPage',
......
frontend/src/views/user/DocsView.vue 0 → 100644
View file @ 058c3bd8
<template>
<AppLayout>
<div class="docs-page-root">
<!-- TOC sidebar — sticky within the page -->
<aside v-if="toc.length > 0" class="docs-toc">
<div class="docs-toc-inner">
<p class="toc-title">目录</p>
<nav class="space-y-0.5">
<a
v-for="item in toc"
:key="item.id"
:href="`#${item.id}`"
class="toc-link"
:class="[
item.level === 1 ? 'toc-h1' : item.level === 2 ? 'toc-h2' : 'toc-h3',
activeId === item.id ? 'toc-link-active' : 'toc-link-idle'
]"
@click.prevent="scrollTo(item.id)"
>{{ item.text }}</a>
</nav>
</div>
</aside>
<!-- Doc content -->
<div class="docs-scroll">
<article
ref="articleEl"
class="docs-content"
v-html="renderedContent"
></article>
</div>
</div>
</AppLayout>
</template>
<script setup lang="ts">
import { onMounted, onUnmounted, ref, nextTick } from 'vue'
import { Marked, Renderer } from 'marked'
import DOMPurify from 'dompurify'
import AppLayout from '@/components/layout/AppLayout.vue'
import docRaw from '../../../../trafficapi_ai_call_documentation.md?raw'
// ─── Slug ─────────────────────────────────────────────────────────────────────
function slugify(text: string): string {
return text
.toLowerCase()
.replace(/\s+/g, '-')
.replace(/[^\w\u4e00-\u9fff-]/g, '')
.replace(/-+/g, '-')
.replace(/^-|-$/g, '')
}
interface TocItem { id: string; text: string; level: number }
// ─── Parse once ───────────────────────────────────────────────────────────────
const toc: TocItem[] = []
const slugCount: Record<string, number> = {}
const renderer = new Renderer()
// eslint-disable-next-line @typescript-eslint/no-explicit-any
renderer.heading = function (token: any) {
const text: string = token.text ?? ''
const depth: number = token.depth ?? 1
const raw = text.replace(/<[^>]+>/g, '')
let slug = slugify(raw)
if (slugCount[slug] !== undefined) { slugCount[slug]++; slug = `${slug}-${slugCount[slug]}` }
else { slugCount[slug] = 0 }
toc.push({ id: slug, text: raw, level: depth })
return `<h${depth} id="${slug}">${text}</h${depth}>`
}
const markedInstance = new Marked({ renderer, breaks: true, gfm: true })
const html = markedInstance.parse(docRaw) as string
const renderedContent = DOMPurify.sanitize(html, { ADD_ATTR: ['id'] })
// ─── Active heading — page-level scroll ───────────────────────────────────────
// AppLayout: header h-16 (64px) + main p-4~p-8. Use 88px as safe offset.
const SCROLL_OFFSET = 88
const activeId = ref(toc[0]?.id ?? '')
const articleEl = ref<HTMLElement | null>(null)
let observer: IntersectionObserver | null = null
function setupObserver() {
if (observer) observer.disconnect()
if (!articleEl.value) return
const headings = articleEl.value.querySelectorAll('h1[id], h2[id], h3[id], h4[id]')
if (!headings.length) return
// root: null → observe against the viewport (page scrolls, not an inner div)
observer = new IntersectionObserver(
(entries) => {
const visible = entries
.filter((e) => e.isIntersecting)
.sort((a, b) => a.boundingClientRect.top - b.boundingClientRect.top)
if (visible.length > 0) activeId.value = visible[0].target.id
},
{ root: null, rootMargin: `-${SCROLL_OFFSET}px 0px -60% 0px`, threshold: 0 }
)
headings.forEach((el) => observer!.observe(el))
}
function scrollTo(id: string) {
const el = document.getElementById(id)
if (!el) return
// Scroll at page (window) level, offset by sticky header height
const top = el.getBoundingClientRect().top + window.scrollY - SCROLL_OFFSET
window.scrollTo({ top, behavior: 'smooth' })
activeId.value = id
}
onMounted(async () => { await nextTick(); setupObserver() })
onUnmounted(() => { observer?.disconnect() })
</script>
<style scoped>
/* ═══════════════════════════════════════════════════════════════
Layout — page scrolls naturally; TOC is sticky
═══════════════════════════════════════════════════════════════ */
.docs-page-root {
display: flex;
align-items: flex-start;
border-radius: 1rem;
border: 1px solid #e5e7eb;
background: #ffffff;
overflow: visible; /* let page scroll */
}
:global(.dark) .docs-page-root {
background: #111827;
border-color: rgba(255, 255, 255, 0.08);
}
.docs-toc {
width: 14rem;
flex-shrink: 0;
border-right: 1px solid #e5e7eb;
display: none;
background: #f9fafb;
/* sticky: stays in view as page scrolls */
position: sticky;
top: 88px; /* header (64px) + main padding (24px) */
max-height: calc(100vh - 100px);
overflow-y: auto;
}
@media (min-width: 1280px) {
.docs-toc { display: block; }
}
:global(.dark) .docs-toc {
border-right-color: rgba(255, 255, 255, 0.07);
background: #0d1525;
}
.docs-toc-inner {
padding: 1rem;
}
.docs-scroll {
min-width: 0;
flex: 1;
}
.docs-content {
max-width: 56rem;
margin-left: auto;
margin-right: auto;
padding: 2rem 1.5rem;
}
/* Give headings scroll-margin so they aren't hidden behind the sticky header */
.docs-content :deep(h1),
.docs-content :deep(h2),
.docs-content :deep(h3),
.docs-content :deep(h4) {
scroll-margin-top: 92px;
}
/* ═══════════════════════════════════════════════════════════════
TOC
═══════════════════════════════════════════════════════════════ */
.toc-title {
margin-bottom: 0.75rem;
font-size: 0.7rem;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.08em;
color: #9ca3af;
}
.toc-link {
display: block;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
border-radius: 0.375rem;
padding: 0.25rem 0.5rem;
font-size: 0.8rem;
text-decoration: none;
transition: background-color 0.1s, color 0.1s;
}
.toc-h1 { font-weight: 600; }
.toc-h2 { padding-left: 1rem; }
.toc-h3 { padding-left: 1.5rem; font-size: 0.72rem; }
.toc-link-idle { color: #6b7280; }
.toc-link-idle:hover { background: #f3f4f6; color: #111827; }
:global(.dark) .toc-link-idle { color: #9ca3af; }
:global(.dark) .toc-link-idle:hover { background: rgba(255,255,255,0.07); color: #e2e8f0; }
.toc-link-active { background: #ede9fe; color: #6d28d9; }
:global(.dark) .toc-link-active { background: rgba(109,40,217,0.2); color: #a78bfa; }
/* ═══════════════════════════════════════════════════════════════
Markdown typography — Light
═══════════════════════════════════════════════════════════════ */
.docs-content :deep(h1) {
margin: 2rem 0 1rem;
font-size: 1.875rem;
font-weight: 700;
color: #111827;
line-height: 2.25rem;
}
.docs-content :deep(h2) {
margin: 2rem 0 0.75rem;
font-size: 1.5rem;
font-weight: 600;
color: #1f2937;
padding-bottom: 0.5rem;
border-bottom: 1px solid #e5e7eb;
}
.docs-content :deep(h3) {
margin: 1.5rem 0 0.5rem;
font-size: 1.2rem;
font-weight: 600;
color: #374151;
}
.docs-content :deep(h4) {
margin: 1rem 0 0.5rem;
font-size: 1rem;
font-weight: 600;
color: #4b5563;
}
.docs-content :deep(p) {
margin-bottom: 1rem;
line-height: 1.75;
color: #374151;
}
.docs-content :deep(ul),
.docs-content :deep(ol) {
margin-bottom: 1rem;
padding-left: 1.5rem;
color: #374151;
}
.docs-content :deep(ul) { list-style-type: disc; }
.docs-content :deep(ol) { list-style-type: decimal; }
.docs-content :deep(li) {
margin-bottom: 0.25rem;
line-height: 1.75;
}
.docs-content :deep(a) {
color: #6d28d9;
text-decoration: underline;
}
.docs-content :deep(a:hover) { color: #5b21b6; }
.docs-content :deep(blockquote) {
margin-bottom: 1rem;
border-left: 4px solid #c4b5fd;
padding-left: 1rem;
font-style: italic;
color: #6b7280;
}
.docs-content :deep(strong) { font-weight: 600; color: #111827; }
.docs-content :deep(hr) {
margin: 2rem 0;
border: none;
border-top: 1px solid #e5e7eb;
}
/* Table — light */
.docs-content :deep(table) {
margin-bottom: 1rem;
width: 100%;
border-collapse: collapse;
font-size: 0.875rem;
}
.docs-content :deep(th) {
border: 1px solid #d1d5db;
background: #f3f4f6;
padding: 0.5rem 0.75rem;
text-align: left;
font-weight: 600;
color: #374151;
}
.docs-content :deep(td) {
border: 1px solid #e5e7eb;
padding: 0.5rem 0.75rem;
color: #374151;
}
.docs-content :deep(tr:nth-child(even) td) { background: #f9fafb; }
/* Inline code — light */
.docs-content :deep(code) {
background: #fef2f2;
color: #be123c;
padding: 0.125rem 0.375rem;
border-radius: 0.25rem;
font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
font-size: 0.85em;
}
/* Code block — always dark bg */
.docs-content :deep(pre) {
margin-bottom: 1rem;
overflow-x: auto;
border-radius: 0.75rem;
background: #1e293b;
border: 1px solid rgba(255,255,255,0.12);
padding: 1rem;
font-size: 0.875rem;
line-height: 1.6;
}
.docs-content :deep(pre code) {
background: transparent !important;
color: #e2e8f0 !important;
padding: 0 !important;
font-size: inherit;
border-radius: 0;
}
/* ═══════════════════════════════════════════════════════════════
Markdown typography — Dark
═══════════════════════════════════════════════════════════════ */
:global(.dark) .docs-content :deep(h1) { color: #f1f5f9; }
:global(.dark) .docs-content :deep(h2) { color: #e2e8f0; border-bottom-color: rgba(255,255,255,0.1); }
:global(.dark) .docs-content :deep(h3) { color: #cbd5e1; }
:global(.dark) .docs-content :deep(h4) { color: #94a3b8; }
:global(.dark) .docs-content :deep(p) { color: #cbd5e1; }
:global(.dark) .docs-content :deep(ul),
:global(.dark) .docs-content :deep(ol) { color: #cbd5e1; }
:global(.dark) .docs-content :deep(li) { color: #cbd5e1; }
:global(.dark) .docs-content :deep(a) { color: #a78bfa; }
:global(.dark) .docs-content :deep(a:hover) { color: #c4b5fd; }
:global(.dark) .docs-content :deep(blockquote) {
border-left-color: #7c3aed;
color: #9ca3af;
}
:global(.dark) .docs-content :deep(strong) { color: #f1f5f9; }
:global(.dark) .docs-content :deep(hr) { border-top-color: rgba(255,255,255,0.1); }
/* Table — dark */
:global(.dark) .docs-content :deep(th) {
border-color: rgba(255,255,255,0.12);
background: rgba(255,255,255,0.06);
color: #e2e8f0;
}
:global(.dark) .docs-content :deep(td) {
border-color: rgba(255,255,255,0.08);
color: #cbd5e1;
}
:global(.dark) .docs-content :deep(tr:nth-child(even) td) {
background: rgba(255,255,255,0.03);
}
/* Inline code — dark */
:global(.dark) .docs-content :deep(code) {
background: rgba(239,68,68,0.12);
color: #fca5a5;
}
/* Code block dark — slightly lighter to be distinct from page */
:global(.dark) .docs-content :deep(pre) {
background: #0f172a;
border-color: rgba(255,255,255,0.1);
}
</style>
trafficapi_ai_call_documentation.md 0 → 100644
View file @ 058c3bd8
# TrafficAPI AI 模型调用文档
## 概述
TrafficAPI 是一个 AI API 网关平台，用于分发和管理 AI 产品订阅的 API 配额。它支持 Claude、OpenAI、Gemini 等多种 AI 模型，提供统一的 API 接口和智能调度功能。
## 接入地址
| 项目 | 值 |
|------|----|
| Base URL | `https://trafficapi.fcluadecodex.xyz` |
| Claude 端点 | `https://trafficapi.fcluadecodex.xyz/v1/messages` |
| OpenAI 端点 | `https://trafficapi.fcluadecodex.xyz/v1/chat/completions` |
| Gemini 端点 | `https://trafficapi.fcluadecodex.xyz/v1beta/models/{model}%3AgenerateContent` |
> **注意**：Gemini 端点路径中的 `%3A` 是冒号 `:` 的 URL 编码，curl 及各语言 HTTP 客户端需使用编码后的形式。
## 环境变量约定
文档中所有调用示例均通过以下环境变量引用凭据，请在运行前预先配置：
```bash
export TRAFFICAPI_BASE_URL="https://trafficapi.fcluadecodex.xyz"
export CLAUDE_API_KEY="your-claude-api-key"
export OPENAI_API_KEY="your-openai-api-key"
export GEMINI_API_KEY="your-gemini-api-key"
```
## 目录
1. [TrafficAPI 基本原理](#TrafficAPI-基本原理)
2. [认证和 API Key 获取方式](#认证和-api-key-获取方式)
3. [Claude 模型调用](#claude-模型调用)
- [HTTP API 调用](#http-api-调用)
- [SDK 调用](#sdk-调用)
4. [OpenAI 模型调用](#openai-模型调用)
- [HTTP API 调用](#http-api-调用-1)
- [SDK 调用](#sdk-调用-1)
5. [Gemini 模型调用](#gemini-模型调用)
- [HTTP API 调用](#http-api-调用-2)
- [SDK 调用](#sdk-调用-2)
6. [流式响应处理](#流式响应处理)
7. [错误处理和重试机制](#错误处理和重试机制)
8. [性能优化建议](#性能优化建议)
## TrafficAPI 基本原理
### 架构设计
TrafficAPI 作为 AI API 中转服务，其核心原理如下：
1. **统一网关层**：提供标准化的 API 端点，兼容 Claude、OpenAI、Gemini 等不同厂商的 API 协议
2. **智能调度**：根据账号状态、负载情况、成本等因素智能选择上游账号
3. **粘性会话**：支持会话级别的账号绑定，确保同一会话的请求路由到同一上游账号
4. **配额管理**：精确追踪 Token 使用量，实现按量计费
5. **并发控制**：用户级和账号级的并发限制，防止资源滥用
### 核心功能
- **多账号管理**：支持多种上游账号类型（OAuth、API Key）
- **API Key 分发**：为用户生成和管理 API Key
- **精确计费**：Token 级别的用量追踪和成本计算
- **智能调度**：智能账号选择，支持粘性会话
- **并发控制**：用户级和账号级并发限制
- **速率限制**：可配置的请求和 Token 速率限制
## 认证和 API Key 获取方式
### 1. 管理员获取 API Key
1. 登录 TrafficAPI 管理后台
2. 进入「API Keys」管理页面
3. 点击「Create API Key」生成新的 API Key
4. 复制生成的 API Key（格式：`sk-xxxxxxxxxxxxxxxx`）
### 2. 用户获取 API Key
1. 用户注册并登录 TrafficAPI 平台
2. 在用户面板中查看或生成 API Key
3. 管理员可以为用户分配 API Key
### 3. API Key 使用方式
在 HTTP 请求头中添加：
```http
Authorization: Bearer $YOUR_API_KEY
```
或者使用 `x-api-key` 头：
```http
x-api-key: $YOUR_API_KEY
```
## HTTP 请求参数详解
### 请求头参数
| 参数名 | 是否必填 | 说明 | 示例值 |
|--------|----------|------|--------|
| Authorization | 是 | Bearer Token 认证 | `Bearer sk-xxx` |
| Content-Type | 是 | 请求体格式 | `application/json` |
| Accept | 否 | 响应格式 | `application/json` 或 `text/event-stream` |
| User-Agent | 否 | 客户端标识 | `MyApp/1.0` |
| X-Request-ID | 否 | 请求唯一标识 | `req_123456` |
### 请求体参数（通用）
| 参数名 | 类型 | 是否必填 | 说明 | 示例值 |
|--------|------|----------|------|--------|
| model | string | 是 | 模型名称 | `claude-3-sonnet-20241022` |
| messages | array | 是 | 消息数组 | `[{"role": "user", "content": "Hello"}]` |
| max_tokens | integer | 否 | 最大生成 token 数 | `1000` |
| temperature | number | 否 | 温度参数（0.0-2.0） | `0.7` |
| stream | boolean | 否 | 是否启用流式响应 | `true` |
| top_p | number | 否 | 核采样参数（0.0-1.0） | `0.9` |
| frequency_penalty | number | 否 | 频率惩罚（-2.0-2.0） | `0.0` |
| presence_penalty | number | 否 | 存在惩罚（-2.0-2.0） | `0.0` |
### Claude 特有参数
| 参数名 | 类型 | 是否必填 | 说明 | 示例值 |
|--------|------|----------|------|--------|
| system | string | 否 | 系统提示 | `"You are a helpful assistant."` |
| max_tokens | integer | 否 | 最大生成 token 数 | `4096` |
### OpenAI 特有参数
| 参数名 | 类型 | 是否必填 | 说明 | 示例值 |
|--------|------|----------|------|--------|
| n | integer | 否 | 生成多个选择 | `1` |
| stop | string/array | 否 | 停止序列 | `["\n", "Human:"]` |
| logprobs | integer | 否 | 返回 token 对数概率 | `null` |
### Gemini 特有参数
| 参数名 | 类型 | 是否必填 | 说明 | 示例值 |
|--------|------|----------|------|--------|
| safetySettings | array | 否 | 安全设置 | `[{"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"}]` |
| generationConfig | object | 否 | 生成配置 | `{"temperature": 0.7, "maxOutputTokens": 1000}` |
## 响应参数详解
### 成功响应
```json
{
"id": "chatcmpl-123",
"object": "chat.completion",
"created": 1677652288,
"model": "claude-3-sonnet-20241022",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello! How can I help you today?"
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 15,
"total_tokens": 25
}
}
```
### 响应字段说明
| 字段名 | 类型 | 说明 |
|--------|------|------|
| id | string | 请求唯一标识 |
| object | string | 对象类型 |
| created | integer | 创建时间戳 |
| model | string | 使用的模型 |
| choices | array | 生成结果数组 |
| usage | object | Token 使用统计 |
| choices[].index | integer | 选择索引 |
| choices[].message | object | 消息内容 |
| choices[].finish_reason | string | 完成原因 |
| usage.prompt_tokens | integer | 提示 token 数 |
| usage.completion_tokens | integer | 生成 token 数 |
| usage.total_tokens | integer | 总 token 数 |
## Claude 模型调用
### HTTP API 调用
#### 基础信息
- **端点**：`/v1/messages`
- **协议**：兼容 Anthropic Claude API
- **Content-Type**：`application/json`
#### curl 示例
```bash
# 非流式调用
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1/messages" \
-H "Authorization: Bearer $CLAUDE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "Hello, how are you?"
}
]
}'
# 流式调用
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1/messages" \
-H "Authorization: Bearer $CLAUDE_API_KEY" \
-H "Content-Type: application/json" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "Hello, how are you?"
}
],
"stream": true
}'
```
#### Python 示例
```python
import requests
import json
class TrafficAPIClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"anthropic-version": "2023-06-01"
}
def create_message(self, model, messages, max_tokens=1024, stream=False):
url = f"{self.base_url}/v1/messages"
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": messages,
"stream": stream
}
response = requests.post(url, headers=self.headers, json=payload, stream=stream)
if stream:
return self._handle_stream_response(response)
else:
return response.json()
def _handle_stream_response(self, response):
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
yield json.loads(data)
except json.JSONDecodeError:
continue
# 使用示例
import os
client = TrafficAPIClient(
base_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz"),
api_key=os.environ["CLAUDE_API_KEY"]
)
# 非流式调用
response = client.create_message(
model="claude-sonnet-4-6",
messages=[
{"role": "user", "content": "Hello, how are you?"}
],
max_tokens=1024,
stream=False
)
print(response)
# 流式调用
for chunk in client.create_message(
model="claude-sonnet-4-6",
messages=[
{"role": "user", "content": "Hello, how are you?"}
],
max_tokens=1024,
stream=True
):
print(chunk)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### Go 示例
```go
package main
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strings"
)
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
type ClaudeRequest struct {
Model string `json:"model"`
MaxTokens int `json:"max_tokens"`
Messages []Message `json:"messages"`
Stream bool `json:"stream"`
}
type ClaudeResponse struct {
ID string `json:"id"`
Type string `json:"type"`
Role string `json:"role"`
Content []struct {
Type string `json:"type"`
Text string `json:"text"`
} `json:"content"`
Model string `json:"model"`
StopReason string `json:"stop_reason"`
StopSequence string `json:"stop_sequence"`
Usage struct {
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
} `json:"usage"`
}
type TrafficAPIClient struct {
BaseURL string
APIKey string
}
func NewTrafficAPIClient(baseURL, apiKey string) *TrafficAPIClient {
return &TrafficAPIClient{
BaseURL: baseURL,
APIKey: apiKey,
}
}
func (c *TrafficAPIClient) CreateMessage(req ClaudeRequest) (*ClaudeResponse, error) {
url := c.BaseURL + "/v1/messages"
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("anthropic-version", "2023-06-01")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
var claudeResp ClaudeResponse
if err := json.NewDecoder(resp.Body).Decode(&claudeResp); err != nil {
return nil, err
}
return &claudeResp, nil
}
func (c *TrafficAPIClient) CreateMessageStream(req ClaudeRequest) (<-chan string, error) {
req.Stream = true
url := c.BaseURL + "/v1/messages"
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("anthropic-version", "2023-06-01")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
ch := make(chan string)
go func() {
defer resp.Body.Close()
defer close(ch)
reader := bufio.NewReader(resp.Body)
for {
line, err := reader.ReadString('\n')
if err != nil {
if err == io.EOF {
break
}
return
}
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "data: ") {
data := line[6:]
if data == "[DONE]" {
break
}
ch <- data
}
}
}()
return ch, nil
}
// 使用示例
func main() {
baseURL := os.Getenv("TRAFFICAPI_BASE_URL")
if baseURL == "" {
baseURL = "https://trafficapi.fcluadecodex.xyz"
}
client := NewTrafficAPIClient(baseURL, os.Getenv("CLAUDE_API_KEY"))
// 非流式调用
req := ClaudeRequest{
Model: "claude-sonnet-4-6",
MaxTokens: 1024,
Messages: []Message{
{Role: "user", Content: "Hello, how are you?"},
},
Stream: false,
}
resp, err := client.CreateMessage(req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
fmt.Printf("Response: %+v\n", resp)
// 流式调用
req.Stream = true
ch, err := client.CreateMessageStream(req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
for chunk := range ch {
fmt.Printf("Chunk: %s\n", chunk)
}
}
```
**Go SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址
- `http_client`: 可选，自定义 HTTP 客户端
**聊天完成参数：**
- `Model`: 必填，模型名称
- `Messages`: 必填，消息数组
- `MaxTokens`: 可选，最大生成 token 数
- `Temperature`: 可选，温度参数
- `Stream`: 可选，是否启用流式响应
- `TopP`: 可选，核采样参数
- `FrequencyPenalty`: 可选，频率惩罚
- `PresencePenalty`: 可选，存在惩罚
### SDK 调用
#### 使用官方 Anthropic SDK
```python
import anthropic
from anthropic import Anthropic
# 配置 TrafficAPI 作为代理
import os
client = Anthropic(
api_key=os.environ["CLAUDE_API_KEY"],
base_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz") # 无需 /v1 后缀，SDK 会自动添加
)
# 调用 Claude
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "Hello, how are you?"}
]
)
print(response.content[0].text)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### 使用 LangChain
```python
from langchain.chat_models import ChatAnthropic
from langchain.schema import HumanMessage
# 配置 LangChain
import os
chat = ChatAnthropic(
anthropic_api_key=os.environ["CLAUDE_API_KEY"],
anthropic_api_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz"),
model="claude-sonnet-4-6"
)
# 调用
messages = [HumanMessage(content="Hello, how are you?")]
response = chat(messages)
print(response.content)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
## OpenAI 模型调用
### HTTP API 调用
#### 基础信息
- **端点**：`/v1/chat/completions`
- **协议**：兼容 OpenAI Chat Completions API
- **Content-Type**：`application/json`
#### curl 示例
```bash
# 非流式调用
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1/chat/completions" \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "gpt-5.2",
"messages": [
{
"role": "user",
"content": "Hello, how are you?"
}
],
"max_tokens": 1024
}'
# 流式调用
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1/chat/completions" \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "gpt-5.2",
"messages": [
{
"role": "user",
"content": "Hello, how are you?"
}
],
"max_tokens": 1024,
"stream": true
}'
```
#### Python 示例
```python
import requests
import json
class TrafficAPIOpenAIClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_chat_completion(self, model, messages, max_tokens=1024, stream=False):
url = f"{self.base_url}/v1/chat/completions"
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"stream": stream
}
response = requests.post(url, headers=self.headers, json=payload, stream=stream)
if stream:
return self._handle_stream_response(response)
else:
return response.json()
def _handle_stream_response(self, response):
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
yield json.loads(data)
except json.JSONDecodeError:
continue
# 使用示例
import os
client = TrafficAPIOpenAIClient(
base_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz"),
api_key=os.environ["OPENAI_API_KEY"]
)
# 非流式调用
response = client.create_chat_completion(
model="gpt-5.2",
messages=[
{"role": "user", "content": "Hello, how are you?"}
],
max_tokens=1024,
stream=False
)
print(response["choices"][0]["message"]["content"])
# 流式调用
for chunk in client.create_chat_completion(
model="gpt-5.2",
messages=[
{"role": "user", "content": "Hello, how are you?"}
],
max_tokens=1024,
stream=True
):
if "choices" in chunk and chunk["choices"]:
delta = chunk["choices"][0].get("delta", {})
if "content" in delta:
print(delta["content"], end="", flush=True)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### Go 示例
```go
package main
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strings"
)
type OpenAIMessage struct {
Role string `json:"role"`
Content string `json:"content"`
}
type OpenAIRequest struct {
Model string `json:"model"`
Messages []OpenAIMessage `json:"messages"`
MaxTokens int `json:"max_tokens"`
Stream bool `json:"stream"`
}
type OpenAIResponse struct {
ID string `json:"id"`
Object string `json:"object"`
Created int64 `json:"created"`
Model string `json:"model"`
Choices []struct {
Index int `json:"index"`
Message struct {
Role string `json:"role"`
Content string `json:"content"`
} `json:"message"`
FinishReason string `json:"finish_reason"`
} `json:"choices"`
Usage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
TotalTokens int `json:"total_tokens"`
} `json:"usage"`
}
type TrafficAPIOpenAIClient struct {
BaseURL string
APIKey string
}
func NewTrafficAPIOpenAIClient(baseURL, apiKey string) *TrafficAPIOpenAIClient {
return &TrafficAPIOpenAIClient{
BaseURL: baseURL,
APIKey: apiKey,
}
}
func (c *TrafficAPIOpenAIClient) CreateChatCompletion(req OpenAIRequest) (*OpenAIResponse, error) {
url := c.BaseURL + "/v1/chat/completions"
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
var openaiResp OpenAIResponse
if err := json.NewDecoder(resp.Body).Decode(&openaiResp); err != nil {
return nil, err
}
return &openaiResp, nil
}
func (c *TrafficAPIOpenAIClient) CreateChatCompletionStream(req OpenAIRequest) (<-chan string, error) {
req.Stream = true
url := c.BaseURL + "/v1/chat/completions"
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
ch := make(chan string)
go func() {
defer resp.Body.Close()
defer close(ch)
reader := bufio.NewReader(resp.Body)
for {
line, err := reader.ReadString('\n')
if err != nil {
if err == io.EOF {
break
}
return
}
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "data: ") {
data := line[6:]
if data == "[DONE]" {
break
}
ch <- data
}
}
}()
return ch, nil
}
// 使用示例
func main() {
baseURL := os.Getenv("TRAFFICAPI_BASE_URL")
if baseURL == "" {
baseURL = "https://trafficapi.fcluadecodex.xyz"
}
client := NewTrafficAPIOpenAIClient(baseURL, os.Getenv("OPENAI_API_KEY"))
// 非流式调用
req := OpenAIRequest{
Model: "gpt-5.2",
Messages: []OpenAIMessage{
{Role: "user", Content: "Hello, how are you?"},
},
MaxTokens: 1024,
Stream: false,
}
resp, err := client.CreateChatCompletion(req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
fmt.Printf("Response: %+v\n", resp)
// 流式调用
req.Stream = true
ch, err := client.CreateChatCompletionStream(req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
for chunk := range ch {
fmt.Printf("Chunk: %s\n", chunk)
}
}
```
**Go SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址
- `http_client`: 可选，自定义 HTTP 客户端
**聊天完成参数：**
- `Model`: 必填，模型名称
- `Messages`: 必填，消息数组
- `MaxTokens`: 可选，最大生成 token 数
- `Temperature`: 可选，温度参数
- `Stream`: 可选，是否启用流式响应
- `TopP`: 可选，核采样参数
- `FrequencyPenalty`: 可选，频率惩罚
- `PresencePenalty`: 可选，存在惩罚
### SDK 调用
#### 使用官方 OpenAI SDK
```python
import openai
# 配置 TrafficAPI 作为代理
import os
client = openai.OpenAI(
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz") + "/v1"
)
# 调用 GPT
response = client.chat.completions.create(
model="gpt-5.2",
messages=[
{"role": "user", "content": "Hello, how are you?"}
],
max_tokens=1024
)
print(response.choices[0].message.content)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### 使用 LangChain
```python
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
# 配置 LangChain
import os
chat = ChatOpenAI(
openai_api_key=os.environ["OPENAI_API_KEY"],
openai_api_base=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz") + "/v1",
model_name="gpt-5.2"
)
# 调用
messages = [HumanMessage(content="Hello, how are you?")]
response = chat(messages)
print(response.content)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
## Gemini 模型调用
### HTTP API 调用
#### 基础信息
- **端点**：`/v1beta/models/{model}:generateContent`
- **协议**：兼容 Google Gemini API
- **Content-Type**：`application/json`
#### curl 示例
```bash
# 非流式调用
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1beta/models/gemini-3.1-flash-lite-preview%3AgenerateContent" \
-H "Authorization: Bearer $GEMINI_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"contents": [
{
"parts": [
{
"text": "Hello, how are you?"
}
]
}
]
}'
# 流式调用（使用 streamGenerateContent 端点，注意 %3A 是 : 的 URL 编码）
curl -X POST "https://trafficapi.fcluadecodex.xyz/v1beta/models/gemini-3.1-flash-lite-preview%3AstreamGenerateContent" \
-H "Authorization: Bearer $GEMINI_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"contents": [
{
"parts": [
{
"text": "Hello, how are you?"
}
]
}
]
}'
```
#### Python 示例
```python
import requests
import json
class TrafficAPIGeminiClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def generate_content(self, model, contents):
# 注意：路径中的 : 需要 URL 编码为 %3A
url = f"{self.base_url}/v1beta/models/{model}%3AgenerateContent"
payload = {"contents": contents}
response = requests.post(url, headers=self.headers, json=payload)
return response.json()
def generate_content_stream(self, model, contents):
# 注意：路径中的 : 需要 URL 编码为 %3A
url = f"{self.base_url}/v1beta/models/{model}%3AstreamGenerateContent"
payload = {"contents": contents}
response = requests.post(url, headers=self.headers, json=payload, stream=True)
return self._handle_stream_response(response)
def _handle_stream_response(self, response):
# streamGenerateContent 返回 SSE 格式，每行以 "data: " 开头
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
yield json.loads(data)
except json.JSONDecodeError:
continue
# 使用示例
import os
client = TrafficAPIGeminiClient(
base_url=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz"),
api_key=os.environ["GEMINI_API_KEY"]
)
# 非流式调用
response = client.generate_content(
model="gemini-3.1-flash-lite-preview",
contents=[
{
"parts": [
{"text": "Hello, how are you?"}
]
}
],
stream=False
)
print(response["candidates"][0]["content"]["parts"][0]["text"])
# 流式调用（使用 streamGenerateContent 端点）
for chunk in client.generate_content_stream(
model="gemini-3.1-flash-lite-preview",
contents=[
{
"parts": [
{"text": "Hello, how are you?"}
]
}
]
):
if "candidates" in chunk and chunk["candidates"]:
for candidate in chunk["candidates"]:
if "content" in candidate and "parts" in candidate["content"]:
for part in candidate["content"]["parts"]:
if "text" in part:
print(part["text"], end="", flush=True)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### Go 示例
```go
package main
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strings"
)
type GeminiPart struct {
Text string `json:"text"`
}
type GeminiContent struct {
Parts []GeminiPart `json:"parts"`
}
type GeminiRequest struct {
Contents []GeminiContent `json:"contents"`
GenerationConfig *struct {
CandidateCount int `json:"candidateCount"`
} `json:"generationConfig,omitempty"`
}
type GeminiResponse struct {
Candidates []struct {
Content struct {
Parts []GeminiPart `json:"parts"`
} `json:"content"`
FinishReason string `json:"finishReason"`
} `json:"candidates"`
UsageMetadata *struct {
PromptTokenCount int `json:"promptTokenCount"`
CandidatesTokenCount int `json:"candidatesTokenCount"`
TotalTokenCount int `json:"totalTokenCount"`
} `json:"usageMetadata"`
}
type TrafficAPIGeminiClient struct {
BaseURL string
APIKey string
}
func NewTrafficAPIGeminiClient(baseURL, apiKey string) *TrafficAPIGeminiClient {
return &TrafficAPIGeminiClient{
BaseURL: baseURL,
APIKey: apiKey,
}
}
func (c *TrafficAPIGeminiClient) GenerateContent(model string, req GeminiRequest) (*GeminiResponse, error) {
// 注意：路径中的 : 需要 URL 编码为 %3A（%%3A 在 Sprintf 中输出字面量 %3A）
url := fmt.Sprintf("%s/v1beta/models/%s%%3AgenerateContent", c.BaseURL, model)
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
var geminiResp GeminiResponse
if err := json.NewDecoder(resp.Body).Decode(&geminiResp); err != nil {
return nil, err
}
return &geminiResp, nil
}
func (c *TrafficAPIGeminiClient) GenerateContentStream(model string, req GeminiRequest) (<-chan string, error) {
url := fmt.Sprintf("%s/v1beta/models/%s%%3AstreamGenerateContent", c.BaseURL, model)
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", "Bearer "+c.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("API error: %s, body: %s", resp.Status, string(body))
}
ch := make(chan string)
go func() {
defer resp.Body.Close()
defer close(ch)
// streamGenerateContent 返回 SSE 格式，每行以 "data: " 开头
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
line := scanner.Text()
if strings.HasPrefix(line, "data: ") {
data := line[6:]
if data == "[DONE]" {
break
}
if data != "" {
ch <- data
}
}
}
}()
return ch, nil
}
// 使用示例
func main() {
baseURL := os.Getenv("TRAFFICAPI_BASE_URL")
if baseURL == "" {
baseURL = "https://trafficapi.fcluadecodex.xyz"
}
client := NewTrafficAPIGeminiClient(baseURL, os.Getenv("GEMINI_API_KEY"))
// 非流式调用
req := GeminiRequest{
Contents: []GeminiContent{
{
Parts: []GeminiPart{
{Text: "Hello, how are you?"},
},
},
},
}
resp, err := client.GenerateContent("gemini-3.1-flash-lite-preview", req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
fmt.Printf("Response: %+v\n", resp)
// 流式调用
ch, err := client.GenerateContentStream("gemini-3.1-flash-lite-preview", req)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
for chunk := range ch {
fmt.Printf("Chunk: %s\n", chunk)
}
}
```
**Go SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址
- `http_client`: 可选，自定义 HTTP 客户端
**聊天完成参数：**
- `Model`: 必填，模型名称
- `Messages`: 必填，消息数组
- `MaxTokens`: 可选，最大生成 token 数
- `Temperature`: 可选，温度参数
- `Stream`: 可选，是否启用流式响应
- `TopP`: 可选，核采样参数
- `FrequencyPenalty`: 可选，频率惩罚
- `PresencePenalty`: 可选，存在惩罚
### SDK 调用
#### 使用官方 Google Generative AI SDK
```python
import google.generativeai as genai
# 配置 TrafficAPI 作为代理
import os
_base = os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz")
genai.configure(
api_key=os.environ["GEMINI_API_KEY"],
transport="rest",
client_options={
"api_endpoint": f"{_base}/v1beta"
}
)
# 创建模型
model = genai.GenerativeModel("gemini-3.1-flash-lite-preview")
# 调用
response = model.generate_content("Hello, how are you?")
print(response.text)
# 流式调用
response = model.generate_content("Hello, how are you?", stream=True)
for chunk in response:
print(chunk.text, end="", flush=True)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### 使用 LangChain
```python
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema import HumanMessage
# 配置 LangChain
import os
chat = ChatGoogleGenerativeAI(
google_api_key=os.environ["GEMINI_API_KEY"],
google_api_base=os.environ.get("TRAFFICAPI_BASE_URL", "https://trafficapi.fcluadecodex.xyz") + "/v1beta",
model="gemini-3.1-flash-lite-preview"
)
# 调用
messages = [HumanMessage(content="Hello, how are you?")]
response = chat(messages)
print(response.content)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
## 流式响应处理
### 通用流式响应处理模式
TrafficAPI 支持所有模型的流式响应，处理模式类似但略有不同：
#### Claude 流式响应格式
```
event: message_start
data: {"type":"message_start","message":{"model":"claude-sonnet-4-6","id":"msg_xxx","type":"message","role":"assistant","content":[],...}}
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" there"}}
event: message_stop
data: {"type":"message_stop"}
```
#### OpenAI 流式响应格式
```
data: {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1677652288, "model": "gpt-5.2", "choices": [{"index": 0, "delta": {"content": "Hello"}, "finish_reason": null}]}
data: {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1677652288, "model": "gpt-5.2", "choices": [{"index": 0, "delta": {"content": " there"}, "finish_reason": null}]}
data: [DONE]
```
#### Gemini 流式响应格式（streamGenerateContent 端点，SSE 格式）
```
data: {"candidates":[{"content":{"role":"model","parts":[{"text":"Hello"}]},"finishReason":"","index":0}],"usageMetadata":{...},"modelVersion":"gemini-3.1-flash-lite-preview"}
data: {"candidates":[{"content":{"role":"model","parts":[{"text":" there"}]},"finishReason":"STOP","index":0}],"usageMetadata":{...}}
```
### 流式处理最佳实践
1. **缓冲区管理**：适当缓冲流式数据，避免频繁的 UI 更新
2. **错误处理**：流式响应中可能包含错误信息，需要实时检测
3. **连接保持**：确保网络连接稳定，必要时实现重连机制
4. **资源清理**：及时关闭流式连接，释放资源
### Python 通用流式处理函数
```python
def handle_stream_response(response, callback):
"""
通用流式响应处理函数
Args:
response: requests.Response 对象
callback: 处理每个数据块的函数，接收解析后的 JSON 数据
"""
buffer = ""
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
# 处理 SSE 格式（Claude/OpenAI）
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
callback(json_data)
except json.JSONDecodeError:
# 可能是 Gemini 的纯 JSON 流
buffer += data
try:
json_data = json.loads(buffer)
callback(json_data)
buffer = ""
except json.JSONDecodeError:
continue
else:
# 处理 Gemini 的纯 JSON 流
buffer += line
try:
json_data = json.loads(buffer)
callback(json_data)
buffer = ""
except json.JSONDecodeError:
continue
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
## 错误处理和重试机制
### 常见错误码
| 状态码 | 错误类型 | 描述 | 处理建议 |
|--------|----------|------|----------|
| 400 | Bad Request | 请求参数错误 | 检查请求格式和参数 |
| 401 | Unauthorized | API Key 无效或过期 | 检查 API Key 有效性 |
| 403 | Forbidden | 权限不足 | 检查用户权限和配额 |
| 429 | Too Many Requests | 请求频率超限 | 降低请求频率，实现退避重试 |
| 500 | Internal Server Error | 服务器内部错误 | 等待后重试，联系管理员 |
| 502 | Bad Gateway | 上游服务不可用 | 等待后重试 |
| 503 | Service Unavailable | 服务暂时不可用 | 等待后重试 |
### 重试策略实现
#### Python 重试装饰器
```python
import time
import random
from functools import wraps
def retry_with_exponential_backoff(
max_retries=5,
initial_delay=1,
exponential_base=2,
jitter=True,
retry_status_codes=[429, 500, 502, 503, 504]
):
"""指数退避重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
# 检查是否需要重试
should_retry = False
if hasattr(e, 'status_code'):
should_retry = e.status_code in retry_status_codes
elif hasattr(e, 'response') and hasattr(e.response, 'status_code'):
should_retry = e.response.status_code in retry_status_codes
if not should_retry or attempt == max_retries - 1:
raise
# 计算延迟时间
delay *= exponential_base ** attempt
if jitter:
delay = random.uniform(0, delay)
print(f"Attempt {attempt + 1} failed, retrying in {delay:.2f} seconds...")
time.sleep(delay)
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例
@retry_with_exponential_backoff(max_retries=3)
def call_api_with_retry():
response = requests.post(...)
response.raise_for_status()
return response.json()
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### Go 重试实现
```go
package main
import (
"fmt"
"math/rand"
"time"
)
type RetryConfig struct {
MaxRetries int
InitialDelay time.Duration
ExponentialBase float64
Jitter bool
RetryStatusCodes []int
}
func NewDefaultRetryConfig() *RetryConfig {
return &RetryConfig{
MaxRetries: 5,
InitialDelay: time.Second,
ExponentialBase: 2.0,
Jitter: true,
RetryStatusCodes: []int{429, 500, 502, 503, 504},
}
}
func RetryWithExponentialBackoff(config *RetryConfig, fn func() error) error {
var lastErr error
delay := config.InitialDelay
for attempt := 0; attempt < config.MaxRetries; attempt++ {
err := fn()
if err == nil {
return nil
}
lastErr = err
// 检查是否需要重试
shouldRetry := false
if apiErr, ok := err.(*APIError); ok {
for _, code := range config.RetryStatusCodes {
if apiErr.StatusCode == code {
shouldRetry = true
break
}
}
}
if !shouldRetry || attempt == config.MaxRetries-1 {
break
}
// 计算延迟时间
delay = time.Duration(float64(delay) * config.ExponentialBase)
if config.Jitter {
delay = time.Duration(float64(delay) * (0.5 + rand.Float64()))
}
fmt.Printf("Attempt %d failed, retrying in %v...\n", attempt+1, delay)
time.Sleep(delay)
}
return lastErr
}
// 使用示例
func callAPIWithRetry() error {
config := NewDefaultRetryConfig()
return RetryWithExponentialBackoff(config, func() error {
resp, err := http.Post(...)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return &APIError{
StatusCode: resp.StatusCode,
Message: "API request failed",
}
}
return nil
})
}
```
**Go SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址
- `http_client`: 可选，自定义 HTTP 客户端
**聊天完成参数：**
- `Model`: 必填，模型名称
- `Messages`: 必填，消息数组
- `MaxTokens`: 可选，最大生成 token 数
- `Temperature`: 可选，温度参数
- `Stream`: 可选，是否启用流式响应
- `TopP`: 可选，核采样参数
- `FrequencyPenalty`: 可选，频率惩罚
- `PresencePenalty`: 可选，存在惩罚
### 错误监控和告警
1. **错误率监控**：监控 API 调用错误率，设置阈值告警
2. **延迟监控**：监控 API 响应时间，识别性能问题
3. **配额监控**：监控用户配额使用情况，提前预警
4. **健康检查**：定期进行健康检查，确保服务可用性
## 性能优化建议
### 1. 连接池管理
#### Python 连接池配置
```python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 创建会话并配置连接池
session = requests.Session()
# 配置重试策略
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
# 配置适配器
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10, # 连接池大小
pool_maxsize=10, # 最大连接数
pool_block=False # 是否阻塞等待连接
)
session.mount("http://", adapter)
session.mount("https://", adapter)
# 使用会话调用 API
response = session.post(
"https://trafficapi.fcluadecodex.xyz/v1/chat/completions",
headers=headers,
json=payload,
timeout=30
)
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
#### Go 连接池配置
```go
package main
import (
"net/http"
"time"
)
func createHTTPClient() *http.Client {
transport := &http.Transport{
MaxIdleConns: 100, // 最大空闲连接数
MaxIdleConnsPerHost: 10, // 每个主机的最大空闲连接数
MaxConnsPerHost: 10, // 每个主机的最大连接数
IdleConnTimeout: 90 * time.Second, // 空闲连接超时时间
TLSHandshakeTimeout: 10 * time.Second, // TLS 握手超时时间
}
return &http.Client{
Transport: transport,
Timeout: 30 * time.Second, // 请求超时时间
}
}
// 使用共享的 HTTP 客户端
var httpClient = createHTTPClient()
func callAPI() (*http.Response, error) {
req, err := http.NewRequest("POST", "https://trafficapi.fcluadecodex.xyz/v1/chat/completions", nil)
if err != nil {
return nil, err
}
return httpClient.Do(req)
}
```
**Go SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址
- `http_client`: 可选，自定义 HTTP 客户端
**聊天完成参数：**
- `Model`: 必填，模型名称
- `Messages`: 必填，消息数组
- `MaxTokens`: 可选，最大生成 token 数
- `Temperature`: 可选，温度参数
- `Stream`: 可选，是否启用流式响应
- `TopP`: 可选，核采样参数
- `FrequencyPenalty`: 可选，频率惩罚
- `PresencePenalty`: 可选，存在惩罚
### 2. 请求批量化
对于多个独立的请求，可以考虑批量处理：
```python
from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_process_requests(requests_data, max_workers=5):
"""批量处理请求"""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交所有任务
future_to_request = {
executor.submit(process_single_request, data): data
for data in requests_data
}
# 收集结果
for future in as_completed(future_to_request):
try:
result = future.result()
results.append(result)
except Exception as e:
print(f"Request failed: {e}")
results.append(None)
return results
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
### 3. 缓存策略
对于重复的查询，实现缓存机制：
```python
import hashlib
import pickle
from datetime import datetime, timedelta
class APICache:
def __init__(self, ttl_seconds=300):
self.cache = {}
self.ttl = ttl_seconds
def get_cache_key(self, endpoint, params):
"""生成缓存键"""
key_data = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, endpoint, params):
"""获取缓存"""
key = self.get_cache_key(endpoint, params)
if key in self.cache:
cached_data, timestamp = self.cache[key]
if datetime.now() - timestamp < timedelta(seconds=self.ttl):
return cached_data
return None
def set(self, endpoint, params, data):
"""设置缓存"""
key = self.get_cache_key(endpoint, params)
self.cache[key] = (data, datetime.now())
def clear(self):
"""清空缓存"""
self.cache.clear()
# 使用缓存
cache = APICache(ttl_seconds=600)
def cached_api_call(endpoint, params):
# 检查缓存
cached_result = cache.get(endpoint, params)
if cached_result:
return cached_result
# 调用 API
result = call_api(endpoint, params)
# 缓存结果
cache.set(endpoint, params, result)
return result
```
**Python SDK 参数说明：**
**初始化参数：**
- `api_key`: 必填，TrafficAPI 的 API Key
- `base_url`: 可选，TrafficAPI 服务地址，默认为官方地址
- `timeout`: 可选，请求超时时间（秒）
- `max_retries`: 可选，最大重试次数
**聊天完成参数：**
- `model`: 必填，模型名称
- `messages`: 必填，消息列表，格式为 `[{"role": "user", "content": "..."}]`
- `max_tokens`: 可选，最大生成 token 数
- `temperature`: 可选，温度参数（0.0-2.0）
- `stream`: 可选，是否启用流式响应
- `top_p`: 可选，核采样参数
- `frequency_penalty`: 可选，频率惩罚
- `presence_penalty`: 可选，存在惩罚
### 4. 安全最佳实践
1. **API Key 管理**：
- 定期轮换 API Key
- 使用环境变量存储敏感信息
- 实现 Key 的访问控制
2. **请求验证**：
- 验证输入参数
- 限制请求大小
- 实现速率限制
3. **日志和审计**：
- 记录所有 API 调用
- 实现操作审计
- 监控异常行为
## 注意事项
### 1. 服务条款合规性
使用 TrafficAPI 调用 AI 模型服务时，请注意：
- 遵守上游服务提供商（Anthropic、OpenAI、Google）的服务条款
- 了解并遵守相关法律法规
- 合理使用 API 资源，避免滥用
### 2. 安全性考虑
1. **API Key 保护**：
- 不要在客户端代码中硬编码 API Key
- 使用环境变量或密钥管理服务
- 定期轮换 API Key
2. **请求安全**：
- 验证用户输入，防止注入攻击
- 实现请求速率限制
- 监控异常请求模式
3. **数据隐私**：
- 避免传输敏感个人信息
- 了解数据在传输和存储中的加密情况
- 遵守数据保护法规（如 GDPR、CCPA）
### 3. 成本控制
1. **用量监控**：
- 实时监控 Token 使用量
- 设置用量告警阈值
- 定期分析成本趋势
2. **优化策略**：
- 使用缓存减少重复请求
- 优化提示词减少 Token 消耗
- 考虑使用成本更低的模型
3. **预算管理**：
- 设置月度预算限制
- 实现自动化的成本控制
- 定期审查成本效益
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment