转载&存档丨AI 标准协议及调用（3）

实战示例：使用 OpenAI 协议调用 DeepSeek

示例 1：使用 Go

安装依赖

go get [github.com/sashabaranov/go-openai](https://github.com/sashabaranov/go-openai)
go get [github.com/joho/godotenv](https://github.com/joho/godotenv)

代码实现

package main

import (
    "context"
    "fmt"
    "log"
    "os"

    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
    openai "[github.com/sashabaranov/go-openai](https://github.com/sashabaranov/go-openai)"
)

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    config := openai.DefaultConfig(os.Getenv("DEEPSEEK_API_KEY"))
    config.BaseURL = "[https://api.deepseek.com/v1](https://api.deepseek.com/v1)"
    client := openai.NewClientWithConfig(config)

    // 与 AI 进行对话
    question := "什么是 RESTful API？"
    answer, err := chatWithAI(client, question)
    if err != nil {
        log.Fatalf("调用 API 时发生错误：%v", err)
    }

    fmt.Printf("问题：%s\n", question)
    fmt.Printf("回答：%s\n", answer)
}

// chatWithAI 与 AI 进行对话
func chatWithAI(client *openai.Client, userMessage string) (string, error) {
    ctx := context.Background()

    resp, err := client.CreateChatCompletion(
        ctx,
        openai.ChatCompletionRequest{
            Model: "deepseek-chat",
            Messages: []openai.ChatCompletionMessage{
                {
                    Role:    openai.ChatMessageRoleSystem,
                    Content: "你是一个有帮助的 AI 助手，擅长回答技术问题。",
                },
                {
                    Role:    openai.ChatMessageRoleUser,
                    Content: userMessage,
                },
            },
            Temperature: 0.7,
            MaxTokens:   2000,
        },
    )

    if err != nil {
        return "", fmt.Errorf("ChatCompletion error: %v", err)
    }

    return resp.Choices[0].Message.Content, nil
}

流式输出示例

package main

import (
    "context"
    "errors"
    "fmt"
    "io"
    "log"
    "os"

    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
    openai "[github.com/sashabaranov/go-openai](https://github.com/sashabaranov/go-openai)"
)

func chatWithStream(client *openai.Client, userMessage string) error {
    ctx := context.Background()

    req := openai.ChatCompletionRequest{
        Model: "deepseek-chat",
        Messages: []openai.ChatCompletionMessage{
            {
                Role:    openai.ChatMessageRoleSystem,
                Content: "你是一个有帮助的 AI 助手，擅长回答技术问题。",
            },
            {
                Role:    openai.ChatMessageRoleUser,
                Content: userMessage,
            },
        },
        Stream: true,
    }

    stream, err := client.CreateChatCompletionStream(ctx, req)
    if err != nil {
        return fmt.Errorf("ChatCompletionStream error: %v", err)
    }
    defer stream.Close()

    fmt.Print("AI: ")

    for {
        response, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            fmt.Println()
            return nil
        }

        if err != nil {
            return fmt.Errorf("Stream error: %v", err)
        }

        if len(response.Choices) > 0 {
            fmt.Print(response.Choices[0].Delta.Content)
        }
    }
}

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    config := openai.DefaultConfig(os.Getenv("DEEPSEEK_API_KEY"))
    config.BaseURL = "[https://api.deepseek.com/v1](https://api.deepseek.com/v1)"
    client := openai.NewClientWithConfig(config)

    // 使用流式输出
    if err := chatWithStream(client, "介绍一下 Go 的 goroutine"); err != nil {
        log.Fatalf("Error: %v", err)
    }
}

Tool Call 示例

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"

    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
    openai "[github.com/sashabaranov/go-openai](https://github.com/sashabaranov/go-openai)"
)

// WeatherResult 天气查询结果
type WeatherResult struct {
    Temperature int    `json:"temperature"`
    Condition   string `json:"condition"`
}

// getWeather 模拟获取天气信息
func getWeather(city string) WeatherResult {
    weatherData := map[string]WeatherResult{
        "北京": {Temperature: 15, Condition: "晴朗"},
        "上海": {Temperature: 20, Condition: "多云"},
    }

    if result, ok := weatherData[city]; ok {
        return result
    }
    return WeatherResult{Temperature: 0, Condition: "未知"}
}

func chatWithTools(client *openai.Client, userMessage string) (string, error) {
    ctx := context.Background()

    // 定义工具
    tools := []openai.Tool{
        {
            Type: openai.ToolTypeFunction,
            Function: &openai.FunctionDefinition{
                Name:        "get_weather",
                Description: "获取指定城市的天气信息",
                Parameters: json.RawMessage(`{
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "城市名称"
                        }
                    },
                    "required": ["city"]
                }`),
            },
        },
    }

    messages := []openai.ChatCompletionMessage{
        {
            Role:    openai.ChatMessageRoleUser,
            Content: userMessage,
        },
    }

    // 第一次调用：让模型决定是否需要调用工具
    resp, err := client.CreateChatCompletion(
        ctx,
        openai.ChatCompletionRequest{
            Model:    "deepseek-chat",
            Messages: messages,
            Tools:    tools,
        },
    )

    if err != nil {
        return "", fmt.Errorf("ChatCompletion error: %v", err)
    }

    responseMessage := resp.Choices[0].Message
    messages = append(messages, responseMessage)

    // 检查是否需要调用工具
    if len(responseMessage.ToolCalls) > 0 {
        // 执行工具调用
        for _, toolCall := range responseMessage.ToolCalls {
            if toolCall.Function.Name == "get_weather" {
                // 解析参数
                var args map[string]interface{}
                if err := json.Unmarshal([]byte(toolCall.Function.Arguments), &args); err != nil {
                    return "", fmt.Errorf("Failed to parse arguments: %v", err)
                }

                city := args["city"].(string)
                fmt.Printf("调用工具: %s, 参数: %s\n", toolCall.Function.Name, city)

                // 调用实际函数
                weatherResult := getWeather(city)
                resultJSON, _ := json.Marshal(weatherResult)

                // 将工具结果添加到消息历史
                messages = append(messages, openai.ChatCompletionMessage{
                    Role:       openai.ChatMessageRoleTool,
                    Content:    string(resultJSON),
                    ToolCallID: toolCall.ID,
                })
            }
        }

        // 第二次调用：让模型基于工具结果生成最终回复
        finalResp, err := client.CreateChatCompletion(
            ctx,
            openai.ChatCompletionRequest{
                Model:    "deepseek-chat",
                Messages: messages,
            },
        )

        if err != nil {
            return "", fmt.Errorf("ChatCompletion error: %v", err)
        }

        return finalResp.Choices[0].Message.Content, nil
    }

    return responseMessage.Content, nil
}

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    config := openai.DefaultConfig(os.Getenv("DEEPSEEK_API_KEY"))
    config.BaseURL = "[https://api.deepseek.com/v1](https://api.deepseek.com/v1)"
    client := openai.NewClientWithConfig(config)

    // 使用工具调用
    result, err := chatWithTools(client, "北京今天天气怎么样？")
    if err != nil {
        log.Fatalf("Error: %v", err)
    }

    fmt.Println(result)
}

示例 2：使用原生 HTTP 请求

JavaScript 使用 fetch

// chat-http.js
import fetch from "node-fetch";
import dotenv from "dotenv";

dotenv.config();

async function chatWithHTTP(userMessage) {
  const url = "[https://api.deepseek.com/v1/chat/completions](https://api.deepseek.com/v1/chat/completions)";

  const headers = {
    "Content-Type": "application/json",
    Authorization: `Bearer ${process.env.DEEPSEEK_API_KEY}`,
  };

  const data = {
    model: "deepseek-chat",
    messages: [
      {
        role: "system",
        content: "你是一个有帮助的助手。",
      },
      {
        role: "user",
        content: userMessage,
      },
    ],
    temperature: 0.7,
    max_tokens: 2000,
  };

  try {
    const response = await fetch(url, {
      method: "POST",
      headers: headers,
      body: JSON.stringify(data),
    });

    if (!response.ok) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }

    const result = await response.json();
    return result.choices[0].message.content;
  } catch (error) {
    console.error("请求失败：", error);
    throw error;
  }
}

// 使用
chatWithHTTP("什么是 Kubernetes？")
  .then((answer) => console.log(answer))
  .catch((error) => console.error(error));

使用 curl 命令

curl [https://api.deepseek.com/v1/chat/completions](https://api.deepseek.com/v1/chat/completions) \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $DEEPSEEK_API_KEY" \
  -d '{
    "model": "deepseek-chat",
    "messages": [
      {
        "role": "system",
        "content": "你是一个有帮助的助手。"
      },
      {
        "role": "user",
        "content": "什么是微服务架构？"
      }
    ],
    "temperature": 0.7,
    "max_tokens": 2000
  }'

实战示例：使用 Claude API

Go 示例

安装依赖

go get [github.com/anthropics/anthropic-sdk-go](https://github.com/anthropics/anthropic-sdk-go)
go get [github.com/joho/godotenv](https://github.com/joho/godotenv)

基础对话示例

package main

import (
    "context"
    "fmt"
    "log"
    "os"

    "[github.com/anthropics/anthropic-sdk-go](https://github.com/anthropics/anthropic-sdk-go)"
    "[github.com/anthropics/anthropic-sdk-go/option](https://github.com/anthropics/anthropic-sdk-go/option)"
    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
)

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    client := anthropic.NewClient(
        option.WithAPIKey(os.Getenv("ANTHROPIC_API_KEY")),
    )

    // 与 Claude 对话
    answer, err := chatWithClaude(client, "解释一下什么是 GraphQL")
    if err != nil {
        log.Fatalf("调用 API 时发生错误：%v", err)
    }

    fmt.Println(answer)
}

// chatWithClaude 与 Claude 对话
// 注意：Claude API 的 system 参数是独立的，不在 messages 中
func chatWithClaude(client *anthropic.Client, userMessage string) (string, error) {
    ctx := context.Background()

    message, err := client.Messages.New(ctx, anthropic.MessageNewParams{
        Model:     anthropic.F(anthropic.ModelClaude3_5Sonnet20241022),
        MaxTokens: anthropic.F(int64(2000)), // Claude 要求必须指定
        System: anthropic.F([]anthropic.TextBlockParam{
            anthropic.NewTextBlock("你是一个有帮助的 AI 助手，擅长技术问题。"),
        }),
        Messages: anthropic.F([]anthropic.MessageParam{
            anthropic.NewUserMessage(anthropic.NewTextBlock(userMessage)),
        }),
        Temperature: anthropic.F(0.7),
    })

    if err != nil {
        return "", fmt.Errorf("Messages.New error: %v", err)
    }

    // 提取文本内容
    if len(message.Content) > 0 {
        if textBlock, ok := message.Content[0].AsUnion().(anthropic.TextBlock); ok {
            return textBlock.Text, nil
        }
    }

    return "", fmt.Errorf("no text content in response")
}

流式输出

package main

import (
    "context"
    "fmt"
    "log"
    "os"

    "[github.com/anthropics/anthropic-sdk-go](https://github.com/anthropics/anthropic-sdk-go)"
    "[github.com/anthropics/anthropic-sdk-go/option](https://github.com/anthropics/anthropic-sdk-go/option)"
    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
)

func chatWithClaudeStream(client *anthropic.Client, userMessage string) error {
    ctx := context.Background()

    stream := client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
        Model:     anthropic.F(anthropic.ModelClaude3_5Sonnet20241022),
        MaxTokens: anthropic.F(int64(2000)),
        Messages: anthropic.F([]anthropic.MessageParam{
            anthropic.NewUserMessage(anthropic.NewTextBlock(userMessage)),
        }),
    })

    fmt.Print("Claude: ")

    // 处理流式响应
    for stream.Next() {
        event := stream.Current()

        // 处理内容块增量事件
        if delta, ok := event.AsUnion().(anthropic.MessageStreamEventContentBlockDelta); ok {
            if textDelta, ok := delta.Delta.AsUnion().(anthropic.TextDelta); ok {
                fmt.Print(textDelta.Text)
            }
        }
    }

    fmt.Println()

    if err := stream.Err(); err != nil {
        return fmt.Errorf("stream error: %v", err)
    }

    return nil
}

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    client := anthropic.NewClient(
        option.WithAPIKey(os.Getenv("ANTHROPIC_API_KEY")),
    )

    // 使用流式输出
    if err := chatWithClaudeStream(client, "介绍一下 WebSocket"); err != nil {
        log.Fatalf("Error: %v", err)
    }
}

Tool Use（Claude 的函数调用）

Claude 的工具调用机制与 OpenAI 类似，但有一些独特之处：

Claude Tool Use 的特点

1. 工具定义：使用 input_schema 而非 parameters
2. 工具调用：在 content 数组中，类型为 tool_use
3. 工具结果：返回时使用 tool_result 类型，包含 tool_use_id
4. 消息结构：工具结果作为新的 user 消息发送

完整的 Tool Use 示例

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"

    "[github.com/anthropics/anthropic-sdk-go](https://github.com/anthropics/anthropic-sdk-go)"
    "[github.com/anthropics/anthropic-sdk-go/option](https://github.com/anthropics/anthropic-sdk-go/option)"
    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
)

// WeatherResult 天气查询结果
type WeatherResult struct {
    Temperature int    `json:"temperature"`
    Condition   string `json:"condition"`
}

// getWeather 模拟获取天气信息
func getWeather(city string) WeatherResult {
    weatherData := map[string]WeatherResult{
        "北京": {Temperature: 15, Condition: "晴朗"},
        "上海": {Temperature: 20, Condition: "多云"},
    }

    if result, ok := weatherData[city]; ok {
        return result
    }
    return WeatherResult{Temperature: 0, Condition: "未知"}
}

func chatWithClaudeTools(client *anthropic.Client, userMessage string) (string, error) {
    ctx := context.Background()

    // 定义工具（注意：Claude 使用 input_schema）
    tools := []anthropic.ToolParam{
        {
            Name:        anthropic.F("get_weather"),
            Description: anthropic.F("获取指定城市的天气信息"),
            InputSchema: anthropic.F(map[string]interface{}{
                "type": "object",
                "properties": map[string]interface{}{
                    "city": map[string]interface{}{
                        "type":        "string",
                        "description": "城市名称",
                    },
                },
                "required": []string{"city"},
            }),
        },
    }

    messages := []anthropic.MessageParam{
        anthropic.NewUserMessage(anthropic.NewTextBlock(userMessage)),
    }

    // 第一次调用：让 Claude 决定是否需要调用工具
    message, err := client.Messages.New(ctx, anthropic.MessageNewParams{
        Model:     anthropic.F(anthropic.ModelClaude3_5Sonnet20241022),
        MaxTokens: anthropic.F(int64(2000)),
        Tools:     anthropic.F(tools),
        Messages:  anthropic.F(messages),
    })

    if err != nil {
        return "", fmt.Errorf("Messages.New error: %v", err)
    }

    fmt.Printf("Stop reason: %s\n", message.StopReason)

    // 检查是否需要调用工具
    if message.StopReason == "tool_use" {
        // 找到工具调用
        var toolResults []anthropic.ContentBlockParamUnion

        for _, content := range message.Content {
            if toolUse, ok := content.AsUnion().(anthropic.ToolUseBlock); ok {
                fmt.Printf("\n调用工具: %s\n", toolUse.Name)
                fmt.Printf("工具 ID: %s\n", toolUse.ID)

                // 解析参数
                var args map[string]interface{}
                if err := json.Unmarshal(toolUse.Input, &args); err != nil {
                    return "", fmt.Errorf("failed to parse arguments: %v", err)
                }

                city := args["city"].(string)
                fmt.Printf("参数: %s\n", city)

                // 执行实际函数
                weatherResult := getWeather(city)
                resultJSON, _ := json.Marshal(weatherResult)

                // 构建工具结果
                toolResults = append(toolResults, anthropic.NewToolResultBlock(
                    toolUse.ID,
                    string(resultJSON),
                    false,
                ))
            }
        }

        // 构建新的消息历史
        messages = append(messages, anthropic.NewAssistantMessage(message.Content...))
        messages = append(messages, anthropic.NewUserMessage(toolResults...))

        // 第二次调用：让 Claude 基于工具结果生成最终回复
        finalMessage, err := client.Messages.New(ctx, anthropic.MessageNewParams{
            Model:     anthropic.F(anthropic.ModelClaude3_5Sonnet20241022),
            MaxTokens: anthropic.F(int64(2000)),
            Tools:     anthropic.F(tools),
            Messages:  anthropic.F(messages),
        })

        if err != nil {
            return "", fmt.Errorf("Messages.New error: %v", err)
        }

        // 提取文本内容
        if len(finalMessage.Content) > 0 {
            if textBlock, ok := finalMessage.Content[0].AsUnion().(anthropic.TextBlock); ok {
                return textBlock.Text, nil
            }
        }
    } else {
        // 不需要调用工具，直接返回
        if len(message.Content) > 0 {
            if textBlock, ok := message.Content[0].AsUnion().(anthropic.TextBlock); ok {
                return textBlock.Text, nil
            }
        }
    }

    return "", fmt.Errorf("no text content in response")
}

func main() {
    // 加载环境变量
    err := godotenv.Load()
    if err != nil {
        log.Println("Warning: .env file not found")
    }

    // 初始化客户端
    client := anthropic.NewClient(
        option.WithAPIKey(os.Getenv("ANTHROPIC_API_KEY")),
    )

    // 使用工具调用
    result, err := chatWithClaudeTools(client, "上海今天天气如何？")
    if err != nil {
        log.Fatalf("Error: %v", err)
    }

    fmt.Println(result)
}

Claude vs OpenAI Tool Call 对比

特性	OpenAI	Claude
工具定义	parameters	input_schema
工具调用位置	tool_calls 字段	content 数组中的 tool_use 块
工具调用 ID	tool_calls[].id	tool_use.id
工具结果角色	role: "tool"	role: "user"
工具结果类型	顶层 tool_call_id 和 content	content 中的 tool_result 块
工具结果 ID 字段	tool_call_id	tool_use_id
停止原因	finish_reason: "tool_calls"	stop_reason: "tool_use"

统一接口工具

LiteLLM - 一套代码调用所有模型

LiteLLM 是一个统一的 API 接口，支持 100+ 种 LLM 模型，让你用一套代码调用所有模型。

安装

pip install litellm

使用示例

from litellm import completion
import os

# 调用 OpenAI
response = completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello"}]
)

# 调用国内模型
response = completion(
    model="deepseek/deepseek-chat",
    messages=[{"role": "user", "content": "Hello"}]
)

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

优势

• 统一接口：所有模型使用相同的调用方式
• 自动重试：内置重试和错误处理
• 成本追踪：自动记录 token 使用和成本
• 负载均衡：支持多个 API Key 轮换使用

OpenRouter - AI 模型聚合平台

OpenRouter 提供统一的 OpenAI 兼容接口，可以访问多家 AI 服务商的模型。

使用方法

from openai import OpenAI

client = OpenAI(
    base_url="[https://openrouter.ai/api/v1](https://openrouter.ai/api/v1)",
    api_key=os.getenv("OPENROUTER_API_KEY")
)

# 使用 Claude
response = client.chat.completions.create(
    model="anthropic/claude-3-opus",
    messages=[{"role": "user", "content": "Hello"}]
)

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

优势

• 一个 API Key：访问所有模型
• 按需付费：只为实际使用付费
• 模型对比：方便测试不同模型的效果
• OpenAI 兼容：无需修改现有代码

Ollama - 本地运行开源模型

Ollama 让你在本地运行开源 LLM 模型，提供 OpenAI 兼容的 API。

安装

# macOS/Linux
curl -fsSL [https://ollama.com/install.sh](https://ollama.com/install.sh) | sh

# Windows
# 从 [https://ollama.com/download](https://ollama.com/download) 下载安装包

运行模型

# 下载并运行模型
ollama run llama3

# 后台运行
ollama serve

API 调用

from openai import OpenAI

# Ollama 提供 OpenAI 兼容接口
client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # 本地运行不需要真实 key
)

response = client.chat.completions.create(
    model="llama3",
    messages=[{"role": "user", "content": "Hello"}]
)

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

优势

• 完全免费：本地运行，无 API 费用
• 数据隐私：数据不离开本地
• 离线可用：无需网络连接
• OpenAI 兼容：代码无需修改

工具对比

工具	类型	优势	适用场景
LiteLLM	统一接口库	支持最多模型，功能丰富	需要调用多种模型的应用
OpenRouter	云端聚合平台	一个 Key 访问所有模型	快速测试和对比模型
Ollama	本地运行工具	免费、隐私、离线	开发测试、隐私敏感场景

错误处理最佳实践

1. 重试机制设计

重试机制是处理临时性错误（如网络波动、速率限制）的关键策略。

指数退避（Exponential Backoff）

指数退避是一种逐渐增加重试等待时间的策略，避免在服务压力大时继续施压。

核心原理：

• 第一次重试等待时间较短（如 1 秒）
• 后续每次重试等待时间呈指数增长（1秒 → 2秒 → 4秒 → 8秒…）
• 设置最大重试次数（通常 3-5 次）避免无限重试
• 区分错误类型：速率限制、网络错误需要重试；参数错误、认证失败不应重试
• 设置合理的超时时间（如 30 秒）防止请求长时间挂起

实现要点：

• 使用循环结构进行重试，记录当前尝试次数
• 捕获不同类型的异常并分类处理
• 速率限制错误使用较长的等待时间（指数增长）
• 网络连接错误使用较短的等待时间（线性增长）
• 参数错误、认证错误等直接抛出，不进行重试

带抖动的指数退避（Jitter）

在高并发场景下，添加随机抖动可以避免”惊群效应”（多个请求同时重试）。

核心原理：

• 在指数退避的基础上，增加随机时间偏移
• 例如：基础等待 4 秒 + 随机 0-1 秒 = 实际等待 4.0-5.0 秒
• 避免大量失败请求在同一时刻同时重试，造成服务器压力激增
• 特别适用于分布式系统和高并发场景

实现要点：

• 计算基础等待时间（指数退避）
• 生成随机抖动值（通常为 0 到 1 秒之间的随机数）
• 实际等待时间 = 基础等待时间 + 随机抖动
• 对于网络错误，也可以使用随机范围内的等待时间

2. 兜底机制设计

兜底机制确保即使 AI 调用失败，系统仍能提供基本服务。

多模型兜底

当主模型失败时，自动切换到备用模型。

核心原理：

• 定义模型优先级列表（主模型 → 备用模型 1 → 备用模型 2）
• 按优先级依次尝试，直到成功或全部失败
• 记录使用的模型信息，便于监控和分析
• 考虑成本因素：主模型可选性价比高的，备用模型选稳定性好的
• 所有模型都失败时，返回友好的错误提示

实现要点：

• 创建模型配置列表，包含模型名称、API 客户端、base_url 等信息
• 使用循环遍历模型列表，依次尝试调用
• 记录最后一次错误信息，用于最终的错误报告
• 设置较短的超时时间（如 10 秒），快速切换到下一个模型
• 返回结构化的结果，包含成功状态、使用的模型、响应内容等

缓存兜底

对于常见问题，使用缓存作为兜底。

核心原理：

• 对用户输入生成唯一标识（如 MD5 哈希）作为缓存键
• 优先从缓存获取结果，命中则直接返回
• 缓存未命中时调用 AI，成功后将结果存入缓存
• AI 调用失败时，可返回缓存的历史相似回答或通用回复
• 生产环境建议使用 Redis 等分布式缓存，支持过期时间和容量管理

实现要点：

• 使用哈希算法（如 MD5）为用户消息生成缓存键
• 实现三层查询逻辑：缓存 → AI 调用 → 兜底回复
• 成功的 AI 响应应存入缓存，供后续使用
• 设置合理的缓存过期时间，避免返回过时信息
• 返回结果时标注数据来源（cache/ai/fallback）

降级服务

当 AI 完全不可用时，提供降级的基础服务。

核心原理：

• 使用规则匹配或关键词识别提供基础回复
• 针对常见问题（价格、联系方式、使用帮助）预设模板回答
• 明确告知用户当前使用简化服务，设置合理预期
• 提供人工客服联系方式作为最终兜底
• 降级服务应简单可靠，避免依赖外部服务

实现要点：

• 实现基于关键词的规则匹配函数
• 为常见问题类型准备模板回复
• 在降级响应中明确标注服务状态
• 提供替代的联系方式（客服电话、文档链接等）
• 降级逻辑应足够简单，不依赖数据库或外部 API

3. 完整的错误处理示例

结合重试和兜底机制的完整实现：

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "math"
    "math/rand"
    "os"
    "strings"
    "time"

    "[github.com/joho/godotenv](https://github.com/joho/godotenv)"
    openai "[github.com/sashabaranov/go-openai](https://github.com/sashabaranov/go-openai)"
)

// AIResult AI 调用结果
type AIResult struct {
    Success bool
    Model   string
    Mode    string
    Content string
    Notice  string
    Error   string
}

// RobustAIClient 健壮的 AI 客户端：重试 + 兜底
type RobustAIClient struct {
    primaryClient  *openai.Client
    fallbackClient *openai.Client
}

// NewRobustAIClient 创建健壮的 AI 客户端
func NewRobustAIClient() *RobustAIClient {
    // 加载环境变量
    _ = godotenv.Load()

    // 初始化主客户端（DeepSeek）
    primaryConfig := openai.DefaultConfig(os.Getenv("DEEPSEEK_API_KEY"))
    primaryConfig.BaseURL = "[https://api.deepseek.com/v1](https://api.deepseek.com/v1)"

    // 初始化备用客户端（通义千问）
    fallbackConfig := openai.DefaultConfig(os.Getenv("QWEN_API_KEY"))
    fallbackConfig.BaseURL = "[https://dashscope.aliyuncs.com/compatible-mode/v1](https://dashscope.aliyuncs.com/compatible-mode/v1)"

    return &RobustAIClient{
        primaryClient:  openai.NewClientWithConfig(primaryConfig),
        fallbackClient: openai.NewClientWithConfig(fallbackConfig),
    }
}

// Chat 完整的错误处理流程
func (c *RobustAIClient) Chat(userMessage string, maxRetries int) AIResult {
    // 1. 尝试主模型（带重试）
    result := c.tryWithRetry(c.primaryClient, "deepseek-chat", userMessage, maxRetries)
    if result.Success {
        return result
    }

    fmt.Println("主模型失败，切换到备用模型...")

    // 2. 尝试备用模型（带重试）
    result = c.tryWithRetry(c.fallbackClient, "qwen-turbo", userMessage, 2)
    if result.Success {
        return result
    }

    fmt.Println("备用模型也失败，使用降级服务...")

    // 3. 降级服务
    return AIResult{
        Success: false,
        Mode:    "degraded",
        Content: getRuleBasedResponse(userMessage),
        Notice:  "AI 服务暂时不可用，已切换到基础服务",
    }
}

// tryWithRetry 带重试的调用
func (c *RobustAIClient) tryWithRetry(
    client *openai.Client,
    model string,
    message string,
    maxRetries int,
) AIResult {
    ctx := context.Background()

    for attempt := 0; attempt < maxRetries; attempt++ {
        // 设置超时
        ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
        defer cancel()

        resp, err := client.CreateChatCompletion(
            ctxWithTimeout,
            openai.ChatCompletionRequest{
                Model: model,
                Messages: []openai.ChatCompletionMessage{
                    {
                        Role:    openai.ChatMessageRoleUser,
                        Content: message,
                    },
                },
            },
        )

        // 成功
        if err == nil {
            return AIResult{
                Success: true,
                Model:   model,
                Content: resp.Choices[0].Message.Content,
            }
        }

        // 错误处理
        errMsg := err.Error()

        // 速率限制错误
        if isRateLimitError(err) {
            if attempt < maxRetries-1 {
                // 指数退避 + 随机抖动
                baseWait := math.Pow(2, float64(attempt))
                jitter := rand.Float64()
                waitTime := time.Duration(baseWait+jitter) * time.Second

                fmt.Printf("速率限制，等待 %.2f 秒...\n", waitTime.Seconds())
                time.Sleep(waitTime)
                continue
            }
            return AIResult{Success: false, Error: "速率限制"}
        }

        // 网络连接错误
        if isConnectionError(err) {
            if attempt < maxRetries-1 {
                waitTime := time.Duration(attempt+1) * time.Second
                fmt.Printf("网络错误，等待 %d 秒...\n", int(waitTime.Seconds()))
                time.Sleep(waitTime)
                continue
            }
            return AIResult{Success: false, Error: "网络连接失败"}
        }

        // 其他错误（不重试）
        return AIResult{Success: false, Error: errMsg}
    }

    return AIResult{Success: false, Error: "未知错误"}
}

// getRuleBasedResponse 基于规则的降级回复
func getRuleBasedResponse(message string) string {
    messageLower := strings.ToLower(message)

    if strings.Contains(messageLower, "价格") || strings.Contains(messageLower, "多少钱") {
        return "关于价格信息，请访问我们的官网或联系客服：400-xxx-xxxx"
    }

    if strings.Contains(messageLower, "使用") || strings.Contains(messageLower, "怎么") {
        return "使用帮助请查看文档：[https://docs.example.com](https://docs.example.com)"
    }

    if strings.Contains(messageLower, "联系") || strings.Contains(messageLower, "客服") {
        return "客服热线：400-xxx-xxxx，工作时间：9:00-18:00"
    }

    return "抱歉，AI 服务暂时不可用。如需帮助，请联系客服：400-xxx-xxxx"
}

// isRateLimitError 判断是否为速率限制错误
func isRateLimitError(err error) bool {
    if err == nil {
        return false
    }
    errMsg := err.Error()
    return strings.Contains(errMsg, "rate_limit") ||
        strings.Contains(errMsg, "429") ||
        strings.Contains(errMsg, "too many requests")
}

// isConnectionError 判断是否为网络连接错误
func isConnectionError(err error) bool {
    if err == nil {
        return false
    }
    errMsg := err.Error()
    return strings.Contains(errMsg, "connection") ||
        strings.Contains(errMsg, "timeout") ||
        strings.Contains(errMsg, "network")
}

func main() {
    // 初始化随机数种子
    rand.Seed(time.Now().UnixNano())

    // 创建客户端
    client := NewRobustAIClient()

    // 使用
    result := client.Chat("什么是机器学习？", 3)

    if result.Success {
        fmt.Printf("[%s] %s\n", result.Model, result.Content)
    } else {
        fmt.Printf("[%s] %s\n", result.Mode, result.Content)
        if result.Notice != "" {
            fmt.Printf("提示: %s\n", result.Notice)
        }
    }
}

最佳实践总结

1. 重试机制：

   • 使用指数退避避免过度重试
   • 添加随机抖动防止惊群效应
   • 区分可重试错误（网络、速率限制）和不可重试错误（参数错误）

2. 兜底机制：

   • 多模型兜底：主模型失败时切换备用模型
   • 缓存兜底：常见问题使用缓存响应
   • 降级服务：提供基础的规则匹配服务

3. 监控和日志：

   • 记录每次重试和兜底的触发情况
   • 监控各模型的成功率和响应时间
   • 设置告警阈值，及时发现问题

成本优化建议

1. 选择合适的模型

# 根据任务复杂度选择模型
def choose_model(task_complexity: str) -> str:
    """
    简单任务：使用更便宜的模型
    复杂任务：使用更强大的模型
    """
    if task_complexity == "simple":
        return "deepseek-chat"  # 更便宜
    elif task_complexity == "complex":
        return "deepseek-coder"  # 更强大但更贵
    else:
        return "deepseek-chat"

2. 控制 token 使用

def optimize_tokens(messages: list) -> list:
    """优化消息历史，减少 token 消耗"""

    # 只保留最近的 N 条消息
    max_history = 10
    if len(messages) > max_history:
        # 保留 system 消息和最近的对话
        system_msg = [m for m in messages if m["role"] == "system"]
        recent_msgs = messages[-max_history:]
        messages = system_msg + recent_msgs

    return messages

3. 使用缓存

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_chat(user_message: str) -> str:
    """缓存相同问题的答案"""
    return chat_with_ai(user_message)

# 或使用更灵活的缓存
cache = {}

def chat_with_cache(user_message: str) -> str:
    # 生成缓存键
    cache_key = hashlib.md5(user_message.encode()).hexdigest()

    if cache_key in cache:
        print("使用缓存结果")
        return cache[cache_key]

    result = chat_with_ai(user_message)
    cache[cache_key] = result
    return result

4. 批量处理

def batch_process(questions: list) -> list:
    """批量处理多个问题，减少请求次数"""

    # 将多个问题合并为一个请求
    combined_prompt = "请分别回答以下问题：\n\n"
    for i, q in enumerate(questions, 1):
        combined_prompt += f"{i}. {q}\n"

    response = chat_with_ai(combined_prompt)

    # 解析响应（实际应用中需要更复杂的解析逻辑）
    return response.split("\n\n")

国内外主流 AI 服务

国内服务

服务商	OpenAI 兼容	官网
DeepSeek	兼容	https://www.deepseek.com/
阿里云通义千问	兼容	https://tongyi.aliyun.com/
智谱 AI	兼容	https://www.zhipuai.cn/
百度文心	部分兼容	https://yiyan.baidu.com/
月之暗面	兼容	https://www.moonshot.cn/

国外服务

服务商	协议类型	官网
OpenAI	OpenAI	https://openai.com/
Anthropic	Claude	https://www.anthropic.com/
Google	Gemini	https://ai.google.dev/

其他服务：字节豆包、xAI (Grok) 等。

统一接口平台

平台	官网
LiteLLM	https://litellm.ai/
OpenRouter	https://openrouter.ai/
Ollama	https://ollama.com/

参考资源

官方文档

DeepSeek

• 官方网站：https://www.deepseek.com/
• API 文档：https://platform.deepseek.com/api-docs/
• 定价：https://platform.deepseek.com/api-docs/pricing/

阿里云通义千问

• 官方网站：https://tongyi.aliyun.com/
• API 文档：https://help.aliyun.com/zh/dashscope/
• 控制台：https://dashscope.console.aliyun.com/

智谱 AI

• 官方网站：https://www.zhipuai.cn/
• API 文档：https://open.bigmodel.cn/dev/api
• 开放平台：https://open.bigmodel.cn/

百度文心

• 官方网站：https://yiyan.baidu.com/
• API 文档：https://cloud.baidu.com/doc/WENXINWORKSHOP/index.html
• 千帆平台：https://qianfan.cloud.baidu.com/

月之暗面（Kimi）

• 官方网站：https://www.moonshot.cn/
• API 文档：https://platform.moonshot.cn/docs/

OpenAI（参考）

• 官方文档：https://platform.openai.com/docs/
• API 参考：https://platform.openai.com/docs/api-reference/

Anthropic Claude（参考）

• 官方文档：https://docs.anthropic.com/
• API 参考：https://docs.anthropic.com/claude/reference/

Google Gemini

• 官方网站：https://ai.google.dev/
• API 文档：https://ai.google.dev/docs
• 快速开始：https://ai.google.dev/tutorials/python_quickstart

统一接口工具

LiteLLM

• GitHub：https://github.com/BerriAI/litellm
• 文档：https://docs.litellm.ai/
• 支持的模型列表：https://docs.litellm.ai/docs/providers

OpenRouter

• 官方网站：https://openrouter.ai/
• 文档：https://openrouter.ai/docs
• 模型列表：https://openrouter.ai/models

Ollama

• 官方网站：https://ollama.com/
• GitHub：https://github.com/ollama/ollama
• 模型库：https://ollama.com/library

开发工具和库

Python

# OpenAI SDK（兼容多数国内服务）
pip install openai

# Anthropic SDK
pip install anthropic

# 环境变量管理
pip install python-dotenv

# HTTP 请求
pip install requests

JavaScript/TypeScript

# OpenAI SDK
npm install openai

# Anthropic SDK
npm install @anthropic-ai/sdk

# 环境变量管理
npm install dotenv

# HTTP 请求（Node.js 18+ 内置 fetch）
npm install node-fetch  # 仅旧版本需要

学习资源

• OpenAI Cookbook：https://cookbook.openai.com/
• 包含大量实用示例和最佳实践
• LangChain 文档：https://python.langchain.com/
• 构建 LLM 应用的框架
• Prompt Engineering Guide：https://www.promptingguide.ai/
• 提示词工程指南

社区和论坛

• DeepSeek 开发者社区：https://github.com/deepseek-ai
• 阿里云开发者社区：https://developer.aliyun.com/
• 智谱 AI 开发者论坛：https://open.bigmodel.cn/forum

总结

本文详细介绍了大模型 API 调用的方方面面：

1. 协议标准：理解了为什么需要标准化协议，以及 OpenAI 和 Claude 协议的区别
2. 参数详解：掌握了各个请求参数的含义和使用场景
3. Tool Call：学会了如何让 AI 调用外部函数和 API
4. Prompt 优化：了解了编写高质量提示词的技巧
5. 安全实践：认识到 API Key 安全的重要性和防护措施
6. 实战示例：通过多个完整示例学会了使用不同方式调用 API
7. 成本优化：掌握了降低 API 调用成本的方法

关键要点

• 始终使用环境变量存储 API Key，永远不要硬编码，千万不要把API Key暴露在前端
• 根据任务选择合适的模型和参数
• 实现完善的错误处理和重试机制
• 注意防范 Prompt 注入攻击
• 优化 token 使用以控制成本
• 国内服务在访问性、合规性上更有优势(但实际国外的AI比国内的要强得多)

下一步学习

• 探索 LangChain 等 LLM 应用框架
• 学习 RAG（检索增强生成）技术
• 了解 Agent 和 Multi-Agent 系统
• 实践构建完整的 AI 应用

---

最后更新：2026-3-30