장문 출력 실무 가이드

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,                        # key: long output must stream
        "thinking": {"type": "adaptive"},      # adaptive thinking; the model decides depth
        "messages": [{"role": "user", "content": prompt}],
    }
    # Three-part timeout: read only covers the inter-event gap, not the whole generation
    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 == "":                 # events are separated by a blank 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())
    # A Claude stream ends on message_stop, never [DONE]
    if stop_reason == "max_tokens":
        raise RuntimeError(f"truncated by max_tokens after {len(''.join(text))} chars; raise max_tokens and retry")
    return "".join(text).strip()

# connect: establish; write: upload the request body; read: max gap between reads
timeout = httpx.Timeout(connect=30, write=120, read=120, pool=30)

messages = [{
    "role": "user",
    "content": (
        f"{original_prompt}\n\n"
        f"Here is what has been written so far:\n\n{partial_text}\n\n"
        "Continue directly from the last sentence above. Do not repeat existing content or add any commentary."
    ),
}]