Kimi K3 长文本溢出参数调整方案，2026年最新百万Token上下文实战指南

from openai import OpenAI

client = OpenAI(
    api_key="YOUR_MOONSHOT_API_KEY",      # 从 platform.moonshot.cn 获取
    base_url="https://api.moonshot.cn/v1", # Kimi API 端点
)

# 读取并估算输入 Token
long_document = open("your_large_doc.txt", "r", encoding="utf-8").read()

response = client.chat.completions.create(
    model="moonshot-v1-k3",               # K3 模型标识
    messages=[
        {
            "role": "system",
            "content": "你是一位专业的技术文档分析师，擅长从长文本中提取关键信息。"
        },
        {
            "role": "user",
            "content": f"请分析以下文档：\n\n{long_document}"
        }
    ],
    temperature=0.3,                      # 降低随机性，确保输出稳定
    max_tokens=8192,                       # 根据任务类型调整，详见上表
)

print(f"Finish reason: {response.choices[0].finish_reason}")
print(f"输出内容: {response.choices[0].message.content}")
print(f"实际使用Token: 输入={response.usage.prompt_tokens}, 输出={response.usage.completion_tokens}")

response = client.chat.completions.create(
    model="moonshot-v1-k3",
    messages=[
        {
            "role": "system",
            "content": "简洁高效回答，不需要深度推理。"
        },
        {
            "role": "user",
            "content": "解释一下什么是MoE架构"
        }
    ],
    max_tokens=512,
    # 关闭思考模式：会自动降级路由到 K2.6，推理速度更快，Token消耗更低
    extra_body={
        "thinking": {
            "type": "off"    # 关闭深度思考，路由到K2.6
        }
    }
)

# 如果需要在 K3 上使用低思考强度
response = client.chat.completions.create(
    model="moonshot-v1-k3",
    messages=[
        {"role": "user", "content": "将以下文本分类为正面/负面/中性"}
    ],
    max_tokens=64,                         # 低思考强度场景不需要大输出
    extra_body={
        "thinking": {
            "type": "low"                  # Low档，牺牲深度换速度和Token节省
        }
    }
)

import tiktoken

# Kimi 使用与 GPT 同源的 tiktoken 分词器
enc = tiktoken.get_encoding("cl100k_base")  # 对应 K3 使用的分词器

text = open("large_document.txt", "r", encoding="utf-8").read()
token_count = len(enc.encode(text))
print(f"文档Token数: {token_count}")
print(f"剩余可用输出空间: {1_048_576 - token_count}")

# 自动设置合理的 max_tokens
available = 1_048_576 - token_count
max_tokens = min(available // 2, 16384)  # 不超过可用空间的一半，上限16K
print(f"建议 max_tokens: {max_tokens}")

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

def split_text_by_tokens(text: str, max_tokens: int, overlap_tokens: int = 500):
    """
    按Token数量分段，保持段落完整性
    overlap_tokens: 分段重叠Token数，防止边界信息丢失
    """
    tokens = enc.encode(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = start + max_tokens
        chunk_tokens = tokens[start:end]
        chunk_text = enc.decode(chunk_tokens)
        chunks.append(chunk_text)
        # 重叠移动，避免段落边界信息丢失
        start = end - overlap_tokens
        if start >= len(tokens):
            break
    return chunks

def analyze_long_document(doc_path: str, task_prompt: str, max_input_tokens: int = 80000):
    """
    通用长文档分析流水线
    max_input_tokens: 每段输入Token上限（留空间给prompt和输出）
    """
    with open(doc_path, "r", encoding="utf-8") as f:
        document = f.read()

    # 分段
    segments = split_text_by_tokens(document, max_input_tokens, overlap_tokens=500)
    print(f"文档分段数: {len(segments)}")

    all_results = []
    for i, segment in enumerate(segments):
        print(f"处理段落 {i+1}/{len(segments)} ...")
        response = client.chat.completions.create(
            model="moonshot-v1-k3",
            messages=[
                {"role": "system", "content": task_prompt},
                {"role": "user", "content": f"【段落 {i+1}/{len(segments)}】\n{segment}"}
            ],
            max_tokens=4096,             # 适中输出，避免截断
            temperature=0.3,
        )
        all_results.append({
            "segment": i+1,
            "result": response.choices[0].message.content,
            "usage": {
                "prompt": response.usage.prompt_tokens,
                "completion": response.usage.completion_tokens
            }
        })

    # 最终聚合
    final_prompt = (
        f"以下是长文档分段分析的结果，请整合为完整报告。\n\n"
        + "\n\n".join([f"【段落{r['segment']}】{r['result']}" for r in all_results])
    )
    final_response = client.chat.completions.create(
        model="moonshot-v1-k3",
        messages=[
            {"role": "system", "content": "你是专业的文档整合分析师，将多个段落分析结果整合为连贯完整的报告。"},
            {"role": "user", "content": final_prompt}
        ],
        max_tokens=8192,
        temperature=0.3,
    )
    return final_response.choices[0].message.content

# 使用示例
report = analyze_long_document(
    doc_path="large_research_paper.txt",
    task_prompt="从以下文本中提取：核心论点、研究方法、关键数据、结论与局限性。用中文回答。",
    max_input_tokens=80000    # 80K输入，留约24K给prompt和输出空间
)
print(report)

# 深度推理场景：正确分配思考与回答的Token预算
response = client.chat.completions.create(
    model="moonshot-v1-k3",
    messages=[
        {
            "role": "system",
            "content": "你是一个严谨的研究分析助手。先进行深度思考推理，再给出最终答案。"
        },
        {
            "role": "user",
            "content": "分析以下论文的核心贡献和潜在局限：\n\n[paste paper content here]"
        }
    ],
    max_tokens=16384,                          # 足够大的输出容纳推理链+答案
    temperature=0.3,
)

# 判断是否因Token耗尽而截断
if response.choices[0].finish_reason == "length":
    print("⚠️ 输出被截断，建议增大max_tokens或降低思考强度")

def safe_kimi_call(messages, max_tokens, thinking="max"):
    """
    带截断检测和安全重试的 K3 调用封装
    """
    for attempt in range(2):
        resp = client.chat.completions.create(
            model="moonshot-v1-k3",
            messages=messages,
            max_tokens=max_tokens,
            temperature=0.3,
            extra_body={"thinking": {"type": thinking}} if thinking != "max" else {}
        )
        if resp.choices[0].finish_reason == "length":
            print(f"⚠️ 尝试 {attempt+1} 输出截断，max_tokens={max_tokens} 不足，增大重试...")
            max_tokens = int(max_tokens * 1.5)
        else:
            return resp
    return resp  # 第二次仍截断则返回结果供人工处理

# 使用示例
result = safe_kimi_call(
    messages=[{"role": "user", "content": "详细解释Transformer架构"}],
    max_tokens=4096,
    thinking="max"
)
print(result.choices[0].message.content)