网络不稳定或 API 限速导致请求断开时，Kimi API 不会自动重连，需要业务层自己实现重试逻辑。本文提供可直接复用的 Python 重试封装，支持自定义重试次数和间隔，适合生产环境长时间运行的 Agent 场景。

Kimi API 自动断线重连

为什么需要重连逻辑

Kimi API 在以下情况会中断连接：

• 网络抖动或临时不可用
• 并发请求超过 RPM/TPM 限制
• 服务端临时过载

对于生产级 Agent 或长批量任务，直接抛异常会导致整个任务失败。正确做法是在业务层封装重试逻辑。

与 OpenAI SDK 对比：OpenAI SDK 内置了指数退避重试（默认2次），Moonshot 的 Python SDK 也有内置重试，但自定义重试逻辑更灵活，可以针对特定错误类型决定是否重试。

---

基础重试封装

from openai import OpenAI
import time

client = OpenAI(
    api_key="$MOONSHOT_API_KEY",
    base_url="https://api.moonshot.cn/v1",
)

def chat_once(messages: list) -> str:
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=messages
    )
    return response.choices[0].message.content

def chat(user_input: str, max_attempts: int = 100) -> str | None:
    messages = [
        {
            "role": "system",
            "content": "你是 Kimi，由 Moonshot AI 提供的人工智能助手。",
        },
        {
            "role": "user",
            "content": user_input,
        },
    ]

    start_time = time.time()
    for attempt in range(max_attempts):
        print(f"尝试 {attempt + 1}/{max_attempts}")
        try:
            result = chat_once(messages)
            elapsed = time.time() - start_time
            print(f"成功！耗时 {elapsed:.2f}s")
            return result
        except Exception as e:
            print(f"失败: {e}")
            time.sleep(1)

    print("达到最大重试次数，任务失败。")
    return None

print(chat("你好，请给我讲一个童话故事。"))

---

TypeScript 版本

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.MOONSHOT_API_KEY,
  baseURL: "https://api.moonshot.cn/v1",
});

async function chatWithRetry(
  userInput: string,
  maxAttempts = 20
): Promise<string | null> {
  const messages = [
    { role: "system" as const, content: "你是 Kimi，由 Moonshot AI 提供的人工智能助手。" },
    { role: "user" as const, content: userInput },
  ];

  for (let i = 0; i < maxAttempts; i++) {
    try {
      const response = await client.chat.completions.create({
        model: "kimi-k2.6",
        messages,
      });
      return response.choices[0].message.content ?? "";
    } catch (err) {
      console.error(`第 ${i + 1} 次失败:`, err);
      await new Promise((resolve) => setTimeout(resolve, 1000));
    }
  }

  return null;
}

---

生产环境建议

参数	测试环境	生产环境
max_attempts	5	20~100
重试间隔	1s 固定	指数退避（1s → 2s → 4s...）
超时设置	不限	设置 timeout 参数（如 60s）

指数退避示例：

import math

sleep_time = min(2 ** attempt, 60)  # 最长等待 60s
time.sleep(sleep_time)

---

常见问题

Q: 重试时是否会重复计费？

A: 是的，每次成功的请求都会计费。请求过程中断（未收到完整响应）通常不计费，但重试成功的请求会正常计费。

Q: 遇到 rate_limit_exceeded 错误，应该等多久再重试？

A: 建议至少等 5~10 秒，并根据错误响应中的 retry-after 头字段（如有）来确定等待时间。

Q: 流式输出（stream=True）中途断开怎么办？

A: 记录已接收的内容，使用 Partial Mode 续写。重试时将已有内容作为 partial=True 的 assistant 消息传入。