Kimi K3 API 接入全景指南：1M 长上下文机制、不可关闭思考流与 Agent 工业落地

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="sk-gpt345-your-actual-api-key",
    base_url="https://api.gpt345.com/v1"
)

async def run_kimi_k3_stream():
    # 通过 reasoning_effort 明确指定思考深度
    response = await client.chat.completions.create(
        model="kimi-k3",
        messages=[
            {"role": "system", "content": "你是一位资深金融分析专家。请根据上下文进行深度论证。"},
            {"role": "user", "content": "评估高杠杆企业在加息周期下的流动性危机传导路径。"}
        ],
        extra_body={"reasoning_effort": "low"}, # 调低思考强度以加快响应
        stream=True,
        max_tokens=2048
    )

    print("=== [Kimi K3 思维链推演开始] ===")
    async for chunk in response:
        delta = chunk.choices[0].delta
        
        # 1. 捕获思考增量
        if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
            print(delta.reasoning_content, end="", flush=True)
            
        # 2. 捕获正式文本输出
        if hasattr(delta, 'content') and delta.content:
            print(delta.content, end="", flush=True)

asyncio.run(run_kimi_k3_stream())

import time
import random
import requests

def call_kimi_resilient(messages, api_key, max_retries=3):
    url = "https://api.gpt345.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "kimi-k3",
        "messages": messages,
        "reasoning_effort": "low",
        "stream": False
    }

    for attempt in range(max_retries):
        try:
            resp = requests.post(url, json=payload, headers=headers, timeout=60)
            if resp.status_code == 200:
                return resp.json()
            elif resp.status_code in [429, 502, 503, 504]:
                sleep_s = (2 ** attempt) + random.uniform(0.5, 1.5)
                print(f"遇到瞬态状态码 {resp.status_code}，等待 {sleep_s:.2f}s 后重试...")
                time.sleep(sleep_s)
            else:
                resp.raise_for_status()
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise e
            time.sleep(1.0)

# 测试调用
# res = call_kimi_resilient([{"role": "user", "content": "ping"}], "sk-your-key")