大模型 API 开发入门

# 1. 装 SDK（版本必须 >= 1.0.0，旧版写法完全不同）
pip install --upgrade "openai>=1.0"

# 验证装好了
python3 -c "import openai; print(openai.__version__)"

# 2. 配好三个环境变量（写进 ~/.bashrc 或 .env，不要写进源码）
export LLM_API_KEY="你在控制台申请的key"
export LLM_BASE_URL="https://api.deepseek.com"
export LLM_MODEL="deepseek-flash"

# 3. 写代码之前，先用 curl 把这三个值验通
curl "$LLM_BASE_URL/chat/completions" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $LLM_API_KEY" \
  -d '{
        "model": "'"$LLM_MODEL"'",
        "messages": [{"role": "user", "content": "你好"}],
        "stream": false
      }'

# 能看到带 choices 的 JSON 就说明通了，可以进下一步。
# 注意：Authorization 头里必须用变量 $LLM_API_KEY，
#      不要把密钥明文粘进命令行——它会留在 shell 历史里。

"""最小可运行的一次大模型调用：OpenAI 兼容协议。

换任何一家 OpenAI 兼容的服务，只改 base_url 和 model 两处，
下面的业务代码一个字都不用动。

运行前先配好环境变量：
    export LLM_API_KEY="你的key"
    export LLM_BASE_URL="https://api.deepseek.com"
    export LLM_MODEL="deepseek-flash"
依赖：pip install "openai>=1.0"
"""
import os

from openai import OpenAI

# ① 密钥只从环境变量读，绝不硬编码进源码
client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
)

# ② messages 是一个列表，顺序就是对话发生的顺序
messages = [
    {"role": "system", "content": "你是一个简洁的中文助手，回答不超过两句话。"},
    {"role": "user", "content": "用一句话说明什么是大语言模型。"},
]

# ③ 发请求。temperature 控制随机性，max_tokens 给输出长度封顶
response = client.chat.completions.create(
    model=os.environ.get("LLM_MODEL", "deepseek-flash"),
    messages=messages,
    temperature=0.7,
    max_tokens=200,
)

# ④ 取正文：choices 是个数组，默认只生成 1 条，所以取 [0]
print("回复:", response.choices[0].message.content)

# ⑤ 两个必看字段：结束原因和 token 账单
print("结束原因:", response.choices[0].finish_reason)
print("token 用量: 输入 %d + 输出 %d = %d"
      % (response.usage.prompt_tokens,
         response.usage.completion_tokens,
         response.usage.total_tokens))

"""流式输出：让字一个个吐出来，而不是等十几秒蹦出一整段。

关键差别只有两处：
  请求时加 stream=True；响应从「一个对象」变成「一串 chunk」。
运行前配置 LLM_API_KEY / LLM_BASE_URL / LLM_MODEL 三个环境变量。
"""
import os
import sys

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
)

# stream_options 让最后一个 chunk 带上 usage，否则流式下拿不到 token 账单
stream = client.chat.completions.create(
    model=os.environ.get("LLM_MODEL", "deepseek-flash"),
    messages=[{"role": "user", "content": "用三句话讲讲 Transformer 为什么快。"}],
    temperature=0.7,
    stream=True,
    stream_options={"include_usage": True},
)

pieces = []      # 攒完整答案，落库/写进下一轮 messages 都要用它
usage = None
finish_reason = None

for chunk in stream:
    # 带 usage 的收尾 chunk 里 choices 是空数组，先判空再取下标
    if chunk.usage is not None:
        usage = chunk.usage
    if not chunk.choices:
        continue

    choice = chunk.choices[0]
    # delta 是「增量」：本次新增的那几个字，不是累计值
    piece = choice.delta.content
    if piece:
        pieces.append(piece)
        sys.stdout.write(piece)   # 边收边打印，这就是打字机效果
        sys.stdout.flush()        # 不 flush 会被缓冲住，白做流式
    if choice.finish_reason:
        finish_reason = choice.finish_reason

print("\n" + "-" * 40)
answer = "".join(pieces)
print("完整答案长度:", len(answer))
print("结束原因:", finish_reason)
if usage:
    print("token 用量: 输入 %d + 输出 %d" % (usage.prompt_tokens, usage.completion_tokens))
else:
    print("token 用量: 本次未随流返回")

"""带重试的调用封装：把「网络抖一下就整个任务失败」这件事解决掉。

三条工程原则写进了代码：
  1. 只重试「重试了可能会好」的错误，配额不足、参数写错重试一万次也没用；
  2. 指数退避 + 随机抖动，避免一批请求在同一毫秒集体重来把服务打垮；
  3. 尊重服务端给的 Retry-After，它比你的猜测准。
运行前配置 LLM_API_KEY / LLM_BASE_URL / LLM_MODEL。
"""
import os
import random
import time

from openai import (APIConnectionError, APITimeoutError, InternalServerError,
                    OpenAI, RateLimitError)

# 这几类是「瞬时故障」，退避后重试有意义
RETRYABLE = (RateLimitError, APITimeoutError, APIConnectionError, InternalServerError)

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
    timeout=60.0,      # 单次请求超时，不设的话可能挂死几分钟
    max_retries=0,     # 关掉 SDK 自带重试，避免和下面的逻辑叠加成指数级请求
)


def retry_after_seconds(err):
    """服务端在 429 响应头里给的等待时间，拿得到就优先用它。"""
    response = getattr(err, "response", None)
    if response is None:
        return None
    value = response.headers.get("retry-after")
    if not value:
        return None
    try:
        return float(value)
    except ValueError:
        return None


def chat(messages, model=None, max_attempts=5, base_delay=1.0, **kwargs):
    """调用模型，失败时按指数退避重试；全部失败则把最后一个异常抛出去。"""
    model = model or os.environ.get("LLM_MODEL", "deepseek-flash")
    last_error = None

    for attempt in range(1, max_attempts + 1):
        try:
            response = client.chat.completions.create(
                model=model, messages=messages, **kwargs)
            # 被长度截断时提醒一句，否则容易把半句话当完整答案入库
            if response.choices[0].finish_reason == "length":
                print("[warn] 输出被 max_tokens 截断，答案可能不完整")
            return response
        except RETRYABLE as err:
            last_error = err
            if attempt == max_attempts:
                break
            # 指数退避：1s、2s、4s、8s……再叠一个随机抖动打散并发
            delay = retry_after_seconds(err) or base_delay * (2 ** (attempt - 1))
            delay += random.uniform(0, 0.5)
            print("[retry] 第 %d 次失败（%s），%.1fs 后重试"
                  % (attempt, type(err).__name__, delay))
            time.sleep(delay)
        except Exception as err:
            # 参数错误、鉴权失败、余额不足等，重试没有意义，直接抛
            print("[fatal] 不可重试的错误:", type(err).__name__, err)
            raise

    raise RuntimeError("重试 %d 次仍失败" % max_attempts) from last_error


if __name__ == "__main__":
    resp = chat(
        [{"role": "user", "content": "一句话解释什么叫指数退避。"}],
        temperature=0.3,
        max_tokens=120,
    )
    print(resp.choices[0].message.content)

"""命令行多轮对话：一个能真正聊起来的最小完整程序。

它演示三件在最小调用里看不到的事：
  1. 多轮记忆靠的是「把历史原样带回去」，接口本身不记事；
  2. 上下文会被撑爆，所以要有裁剪策略；
  3. 流式 + 重试要能叠在一起用。
运行前配置 LLM_API_KEY / LLM_BASE_URL / LLM_MODEL，然后 python3 chat_cli.py。
输入 exit 退出，输入 clear 清空记忆。
"""
import os
import sys

from robust_call import chat

SYSTEM_PROMPT = "你是一个中文技术助手，回答简洁、准确，不确定时直说不知道。"

# 只保留最近 N 轮，防止 messages 无限膨胀把上下文和钱一起烧光
MAX_TURNS = 8


def trim(messages, max_turns=MAX_TURNS):
    """裁剪历史：system 永远留着，其余只保留最近 max_turns 轮问答。"""
    system = messages[:1]
    body = messages[1:]
    keep = max_turns * 2          # 一轮 = 一条 user + 一条 assistant
    if len(body) > keep:
        body = body[-keep:]
        # 裁剪后第一条必须是 user，否则有些服务会拒绝
        while body and body[0]["role"] != "user":
            body.pop(0)
    return system + body


def stream_answer(messages):
    """流式打印回答，并返回完整文本，便于写回历史。"""
    response = chat(messages, temperature=0.7, stream=True)
    pieces = []
    for chunk in response:
        if not chunk.choices:
            continue
        piece = chunk.choices[0].delta.content
        if piece:
            pieces.append(piece)
            sys.stdout.write(piece)
            sys.stdout.flush()
    print()
    return "".join(pieces)


def main():
    if not os.environ.get("LLM_API_KEY"):
        print("请先设置环境变量 LLM_API_KEY")
        return
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    print("开始对话（exit 退出，clear 清空记忆）")

    while True:
        try:
            user_input = input("\n你: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见")
            return
        if not user_input:
            continue
        if user_input == "exit":
            print("再见")
            return
        if user_input == "clear":
            messages = [{"role": "system", "content": SYSTEM_PROMPT}]
            print("记忆已清空")
            continue

        messages.append({"role": "user", "content": user_input})
        messages = trim(messages)

        print("助手: ", end="")
        answer = stream_answer(messages)
        # 关键一步：把模型的回答也追加进历史，下一轮它才知道自己说过什么
        messages.append({"role": "assistant", "content": answer})


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""多轮对话的成本增长与三种历史裁剪策略对比。

只用标准库，直接 python3 token_cost.py 就能跑，不需要 API Key。
它不调用任何模型，只做账本推演——目的是让「越聊越贵」变成具体数字。

⚠️ token 数用固定估算值代替真实 tokenizer，只为演示增长趋势。
   真实用量必须读响应里的 usage 字段。
"""

# 每轮对话的估算 token 数（演示用的固定值）
SYSTEM_TOKENS = 120
USER_TOKENS = 60
ASSISTANT_TOKENS = 220

TURNS = 20
CONTEXT_LIMIT = 4096          # 假设模型上下文上限
KEEP_RECENT = 3               # 滑动窗口保留最近几轮
SUMMARY_TOKENS = 150          # 摘要压缩后的固定开销


def full_history(turns):
    """策略一：不裁剪，每轮把全部历史重发。"""
    prompt = SYSTEM_TOKENS
    rows = []
    for t in range(1, turns + 1):
        prompt += USER_TOKENS                      # 本轮提问
        rows.append((t, prompt, ASSISTANT_TOKENS))
        prompt += ASSISTANT_TOKENS                 # 模型回答追加进历史
    return rows


def sliding_window(turns, keep=KEEP_RECENT):
    """策略二：只保留 system + 最近 keep 轮。"""
    rows = []
    for t in range(1, turns + 1):
        kept = min(t - 1, keep)                    # 已完成的历史轮数
        prompt = SYSTEM_TOKENS + kept * (USER_TOKENS + ASSISTANT_TOKENS) + USER_TOKENS
        rows.append((t, prompt, ASSISTANT_TOKENS))
    return rows


def summarize_old(turns, keep=KEEP_RECENT):
    """策略三：最近 keep 轮保留原文，更早的历史压缩成一段摘要。"""
    rows = []
    for t in range(1, turns + 1):
        done = t - 1
        kept = min(done, keep)
        has_summary = done > keep
        prompt = SYSTEM_TOKENS
        if has_summary:
            prompt += SUMMARY_TOKENS
        prompt += kept * (USER_TOKENS + ASSISTANT_TOKENS) + USER_TOKENS
        rows.append((t, prompt, ASSISTANT_TOKENS))
    return rows


def report(name, rows, limit=CONTEXT_LIMIT):
    total_prompt = sum(r[1] for r in rows)
    total_out = sum(r[2] for r in rows)
    last_prompt = rows[-1][1]
    overflow = next((r[0] for r in rows if r[1] > limit), None)
    print("【%s】" % name)
    print("   第 1 轮输入 %d token，第 %d 轮输入 %d token（涨了 %.1f 倍）"
          % (rows[0][1], len(rows), last_prompt, last_prompt / rows[0][1]))
    print("   累计输入 %d token，累计输出 %d token" % (total_prompt, total_out))
    if overflow:
        print("   ❌ 第 %d 轮就超过 %d 的上下文上限" % (overflow, limit))
    else:
        print("   ✅ %d 轮下来始终未超过 %d 的上下文上限" % (len(rows), limit))
    print()
    return total_prompt


def main():
    print("=" * 70)
    print("设定：system %d token，每轮提问 %d token，每轮回答 %d token，共 %d 轮"
          % (SYSTEM_TOKENS, USER_TOKENS, ASSISTANT_TOKENS, TURNS))
    print("=" * 70)
    print()

    print("一、不裁剪时，每轮输入 token 是怎么涨的")
    print("%-8s %-14s %s" % ("轮次", "本轮输入", "占上下文上限"))
    print("-" * 46)
    rows = full_history(TURNS)
    for t, prompt, _ in rows:
        if t in (1, 2, 3, 5, 10, 15, 20):
            bar = "█" * int(prompt / CONTEXT_LIMIT * 30)
            print("%-8d %-14d %5.1f%% %s" % (t, prompt, prompt / CONTEXT_LIMIT * 100, bar))
    print()
    print("注意这是线性增长，不是常数——第 20 轮的输入是第 1 轮的十几倍。")
    print()

    print("=" * 70)
    print("二、三种策略的总账对比")
    print("=" * 70)
    a = report("策略一：全量历史", full_history(TURNS))
    b = report("策略二：滑动窗口（保留最近 %d 轮）" % KEEP_RECENT, sliding_window(TURNS))
    c = report("策略三：旧历史压缩成摘要", summarize_old(TURNS))

    print("=" * 70)
    print("三、结论")
    print("=" * 70)
    print("累计输入 token：全量 %d / 滑窗 %d / 摘要 %d" % (a, b, c))
    print("滑窗相比全量省了 %.0f%%，摘要省了 %.0f%%"
          % ((1 - b / a) * 100, (1 - c / a) * 100))
    print()
    print("滑窗最省，但它把更早的信息直接丢了——用户翻回去问前面的事就答不上来。")
    print("摘要贵一点，但保住了早期信息的要点。")
    print("选哪个取决于业务：客服问答适合滑窗，长程任务助手适合摘要。")


if __name__ == "__main__":
    main()

======================================================================
设定：system 120 token，每轮提问 60 token，每轮回答 220 token，共 20 轮
======================================================================

一、不裁剪时，每轮输入 token 是怎么涨的
轮次       本轮输入           占上下文上限
----------------------------------------------
1        180              4.4% █
2        460             11.2% ███
3        740             18.1% █████
5        1300            31.7% █████████
10       2700            65.9% ███████████████████
15       4100           100.1% ██████████████████████████████
20       5500           134.3% ████████████████████████████████████████

注意这是线性增长，不是常数——第 20 轮的输入是第 1 轮的十几倍。

======================================================================
二、三种策略的总账对比
======================================================================
【策略一：全量历史】
   第 1 轮输入 180 token，第 20 轮输入 5500 token（涨了 30.6 倍）
   累计输入 56800 token，累计输出 4400 token
   ❌ 第 15 轮就超过 4096 的上下文上限

【策略二：滑动窗口（保留最近 3 轮）】
   第 1 轮输入 180 token，第 20 轮输入 1020 token（涨了 5.7 倍）
   累计输入 18720 token，累计输出 4400 token
   ✅ 20 轮下来始终未超过 4096 的上下文上限

【策略三：旧历史压缩成摘要】
   第 1 轮输入 180 token，第 20 轮输入 1170 token（涨了 6.5 倍）
   累计输入 21120 token，累计输出 4400 token
   ✅ 20 轮下来始终未超过 4096 的上下文上限

======================================================================
三、结论
======================================================================
累计输入 token：全量 56800 / 滑窗 18720 / 摘要 21120
滑窗相比全量省了 67%，摘要省了 63%

滑窗最省，但它把更早的信息直接丢了——用户翻回去问前面的事就答不上来。
摘要贵一点，但保住了早期信息的要点。
选哪个取决于业务：客服问答适合滑窗，长程任务助手适合摘要。

"""通用 LLM 客户端骨架：换供应商只改配置，不改业务代码。

把「连哪家、用哪个模型、怎么重试、怎么流式」全收进一个类，
上层业务只看见 ask() 和 ask_stream() 两个方法。
复制后改 5 处 TODO 即可。
"""
import os
import random
import time

from openai import (APIConnectionError, APITimeoutError, InternalServerError,
                    OpenAI, RateLimitError)

RETRYABLE = (RateLimitError, APITimeoutError, APIConnectionError, InternalServerError)

# TODO 1：登记你要用的供应商。base_url 与模型名都以各家官方文档为准。
#         凡是标称「OpenAI 兼容」的服务，都能塞进这张表。
PROVIDERS = {
    "deepseek": {
        "base_url": "https://api.deepseek.com",
        "key_env": "DEEPSEEK_API_KEY",
        "model": "deepseek-flash",
    },
    "zhipu": {
        "base_url": "https://open.bigmodel.cn/api/paas/v4/",
        "key_env": "ZHIPU_API_KEY",
        "model": "glm-5.3-flash",
    },
    "local": {                       # Ollama / vLLM 等本地服务
        "base_url": "http://127.0.0.1:11434/v1",
        "key_env": "LOCAL_API_KEY",  # 本地服务通常不校验，给个占位值即可
        "model": "qwen3:8b",
    },
}


class LLMClient:
    """一层薄封装：统一重试、统一超时、统一默认参数。"""

    def __init__(self, provider="deepseek", model=None, timeout=60.0):
        conf = PROVIDERS[provider]
        api_key = os.environ.get(conf["key_env"])
        if not api_key:
            raise RuntimeError("缺少环境变量 %s" % conf["key_env"])

        self.model = model or conf["model"]
        self.client = OpenAI(
            api_key=api_key,
            base_url=conf["base_url"],
            timeout=timeout,
            max_retries=0,          # 重试逻辑自己管，避免叠加
        )
        # TODO 2：按业务改默认采样参数。抽取/分类类任务把 temperature 压到 0。
        self.defaults = {"temperature": 0.7, "max_tokens": 1024}

    # ----------------------------------------------------------------- 内部
    def _call(self, messages, max_attempts=5, **kwargs):
        params = dict(self.defaults)
        params.update(kwargs)
        last_error = None

        for attempt in range(1, max_attempts + 1):
            try:
                return self.client.chat.completions.create(
                    model=self.model, messages=messages, **params)
            except RETRYABLE as err:
                last_error = err
                if attempt == max_attempts:
                    break
                delay = 1.0 * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
                time.sleep(delay)
            except Exception:
                raise      # 鉴权、参数、余额类错误直接抛，重试无意义
        raise RuntimeError("重试 %d 次仍失败" % max_attempts) from last_error

    # ----------------------------------------------------------------- 对外
    def ask(self, prompt, system=None, **kwargs):
        """一问一答，返回纯文本。"""
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        response = self._call(messages, **kwargs)
        return response.choices[0].message.content

    def ask_stream(self, prompt, system=None, **kwargs):
        """生成器版本：逐段 yield，前端可以边收边渲染。"""
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        stream = self._call(messages, stream=True, **kwargs)
        for chunk in stream:
            if not chunk.choices:
                continue
            piece = chunk.choices[0].delta.content
            if piece:
                yield piece


if __name__ == "__main__":
    # TODO 3：换成你要用的那一家
    llm = LLMClient(provider="deepseek")

    # TODO 4：换成你的提示词
    print(llm.ask("用一句话解释什么是上下文长度。", system="你是简洁的中文助手。"))

    # TODO 5：需要打字机效果时改用流式
    # for piece in llm.ask_stream("讲讲流式输出的好处。"):
    #     print(piece, end="", flush=True)