长文输出实战建议

import json
import httpx

def generate_long_text(prompt, api_key, model="claude-opus-5", max_tokens=64000):
    url = "https://api.apiyi.com/v1/messages"
    headers = {
        "x-api-key": api_key,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
        "accept": "text/event-stream",
    }
    payload = {
        "model": model,
        "max_tokens": max_tokens,
        "stream": True,                        # ← 关键：长输出必须流式
        "thinking": {"type": "adaptive"},      # 自适应思考，模型自行决定思考深度
        "messages": [{"role": "user", "content": prompt}],
    }
    # 三段式超时：read 只覆盖事件间隔，不覆盖整段生成
    timeout = httpx.Timeout(connect=30, write=120, read=120, pool=30)

    text, stop_reason = [], None
    with httpx.Client(timeout=timeout) as c:
        with c.stream("POST", url, json=payload, headers=headers) as r:
            if r.status_code != 200:
                raise RuntimeError(f"HTTP {r.status_code}: {r.read()[:400]}")
            event, data = None, []
            for line in r.iter_lines():
                if line == "":                 # 事件以空行分隔
                    if data:
                        d = json.loads("\n".join(data))
                        t = d.get("type") or event
                        if t == "content_block_delta" and d.get("delta", {}).get("type") == "text_delta":
                            text.append(d["delta"]["text"])
                        elif t == "message_delta":
                            stop_reason = d.get("delta", {}).get("stop_reason") or stop_reason
                    event, data = None, []
                    continue
                if line.startswith("event:"):
                    event = line[6:].strip()
                elif line.startswith("data:"):
                    data.append(line[5:].strip())
    # Claude 流以 message_stop 结束，没有 [DONE]
    if stop_reason == "max_tokens":
        raise RuntimeError(f"被 max_tokens 截断，已产出 {len(''.join(text))} 字，调大 max_tokens 后重试")
    return "".join(text).strip()

# connect：建连；write：上行发请求体；read：两次读之间的上限
timeout = httpx.Timeout(connect=30, write=120, read=120, pool=30)

messages = [{
    "role": "user",
    "content": (
        f"{original_prompt}\n\n"
        f"以下是已经写好的部分：\n\n{partial_text}\n\n"
        "请从上文最后一句之后直接接着写，不要重复已有内容，不要加任何说明。"
    ),
}]