第八天，和前面七天完全不一样。前七天你在熟悉语法、折腾环境、敲练习题，第八天要把这些碎片焊在一起，产出一个真正能解决实际问题的东西：一个完整的大模型调用脚本。这篇综合实战笔记，就是八天 Python 速成后第一次“真刀真枪”干活的全过程记录。我会从知识盘点开始，一路写到接口逻辑、完整脚本代码、踩坑经历，最后聊一聊怎么把脚本变成趁手的工具。如果你是奔着做“大模型工程师”去的，这一篇可能是整个系列里最值得反复看的一篇。
1. 八天速成收尾先盘点：写大模型脚本真正依赖的 Python 能力
1.1 语法层面：七成代码只用到了这几样
很多人在第八天容易犯一个错误：觉得前面七天学得不够，想再花时间补语法。实际上，写一个大模型调用脚本，真正高频用到的 Python 能力非常有限。把这句话反过来说更有价值：如果你时间紧张，只需要把下面这几项练熟，就能应付绝大部分调用场景。
第一项是字典与 JSON 的互相转换。大模型接口的请求体是 JSON 结构，Python 里天然用字典来组织。组装 messages 消息列表、读取返回的 content 字段、把 usage 用量存进日志，全部都是 dict 的增删改查加 json.dumps / json.loads。第二项是 HTTP 请求库的使用。无论是 requests 还是标准库 urllib，至少要能把 GET、POST 请求发出去，能携带 header、能设置超时。第三项是异常处理。网络请求一定会失败，服务端一定会报错，没有 try / except 就谈不上是一个工程脚本。第四项是生成器 yield。流式返回是大模型接口最常用的模式，逐行读取响应体时 yield 能帮你把“接收数据”和“消费数据”彻底分离，代码会简洁很多。第五项是文件操作与环境变量读取，后面做配置分离、批量任务时离不开。
举个例子，组装一段最简单的对话消息，你只需要：
messages = [ {"role": "system", "content": "你是一个严谨的Python工程师"}, {"role": "user", "content": "请帮我解释一下什么是上下文窗口"} ]就这么一个结构，后面所有复杂功能都是围绕它扩展的。八天里学到的循环、函数、类，在脚本里当然也会用到，但核心骨架其实是“发请求、收响应、处理异常、读写数据”这四件事。
1.2 工程层面：从“能跑”到“能维护”相差的是习惯
速成学习最容易养成的坏习惯是：代码能跑就行。第八天之前的练习，跑通就结束没问题；到了综合实战，这个习惯必须改掉，因为你写的脚本大概率会被反复使用、调试、修改。
我建议从三个习惯开始养。第一，使用虚拟环境。每个项目单独建一个 venv，别让全局环境里堆满各种版本的包。这不是什么高深技巧，但能避免很多“昨天还能跑，今天突然不行了”的诡异问题。第二，配置与代码分离。API Key、模型名称、接口地址这些信息不要直接写在脚本里，通过环境变量读取。这样脚本可以在不同环境之间迁移，代码也能放心分享给别人。第三，用 logging 而不是 print 输出信息。print 在脚本里一旦输出内容多起来，很难分清哪条是调试信息、哪条是业务结果。logging 模块自带级别和格式控制，后面排查问题的时候体验完全不同。
这三个习惯单独看都不难，难的是在第八天就强迫自己做到。实际体验下来，凡是第一天就坚持虚拟环境和配置分离的人，后面做批量任务、部署服务时都顺利得多；凡是“先跑通再说”的，基本都要返工一次。
1.3 思维层面：把大模型当成一个不稳定的远程函数
前七天你学的是“我的代码，我说了算”。第八天开始调用大模型接口后，思维要切换：大模型是一个不稳定的远程函数。输入是 prompt，输出是概率性的结果，网络还随时可能中断。这种不确定性是写脚本时必须考虑的。
举个实际例子：你让模型返回一个 JSON 结构，理论上应该得到规范 JSON，但实际可能夹杂解释性文字、Markdown 代码块标记，甚至干脆是空内容。做工程实现时就要对这种输出做容错处理，不能假设接口一定按文档返回。另外，大模型接口和本地函数调用最大的区别是延迟：本地函数毫秒级返回，大模型接口通常几秒甚至几十秒。这意味着你的脚本天然就需要超时控制、进度反馈、断点续跑这些设计。把这些想明白了，写脚本的时候就不会只盯着“如何发请求”，而是会主动考虑失败、重试、记录这类工程问题。
2. 写脚本之前，先把大模型接口的三层逻辑看穿
2.1 一次请求的完整生命周期
第八天动手写代码前，我建议先搞清楚一次大模型调用到底发生了什么。整个过程并不复杂：你向接口地址发送一个 HTTP POST 请求，请求体里携带模型名称、对话消息、生成参数；服务端收到后完成推理，再把结果返回给你。理解这条链路，写脚本时才有全局视角。
以目前最通用的 Chat 风格接口为例，请求体长这样：
{ "model": "your-model-name", "messages": [ {"role": "system", "content": "你是一个Python助手"}, {"role": "user", "content": "写一段读取CSV文件的代码"} ], "temperature": 0.7, "max_tokens": 1000 }model 指定用哪个模型，messages 是对话历史，temperature 控制随机性，max_tokens 限制本次生成的最大长度。响应里最重要的字段是 choices 数组，里面装着模型生成的文本以及结束原因。如果接口返回了 usage 字段，还会带上本次请求消耗的 prompt_tokens、completion_tokens 和 total_tokens。
这里特别提醒一个细节：messages 里通常包含 system、user、assistant 三种角色。系统消息用来设定行为，用户消息是当前问题，历史对话消息则充当模型记忆。很多人第一次写脚本时只放 user 一条消息，后面需要多轮对话时才发现上下文根本没有传全。所以从一开始就按数组结构组织消息，后面做会话管理会省很多事。
2.2 流式输出：为什么大模型接口默认推荐它
大模型生成文本需要时间。如果等全部生成完再返回，用户等待时间会很长，体验很差。所以接口几乎都支持流式输出模式：模型每生成一小段内容，服务端就立刻推送一次，客户端边接收边展示，用户看到的是逐字“冒出来”的效果。
流式返回的报文格式一般长这样：
data: {"choices": [{"delta": {"content": "你好"}}]} data: {"choices": [{"delta": {"content": "，我是"}}]} data: {"choices": [{"delta": {"content": "AI助手"}}]} data: [DONE]每一行以 data: 开头，携带一小段增量内容，最后以 [DONE] 表示结束。用 Python 处理时，需要逐行读取响应、去掉 data: 前缀、解析 JSON、提取 delta.content，然后累积起来或者直接输出。这就是生成器 yield 的用武之地：让解析函数 yield 出每一段文本，调用方按需消费。
非流式模式适合离线批量任务，比如批量给文章生成摘要，结果一次性返回后写入文件，不需要即时展示。但在交互式场景里，流式基本是默认选项。所以你的调用脚本里最好两种模式都支持，用一个参数切换，而不是只实现其中一种。
2.3 鉴权、限流与成本预估：写代码之前就得算清楚
这部分很多人容易等到报错才去了解，我建议提前搞明白。大模型接口通常通过请求头中的 Authorization 字段传递 API Key 完成鉴权，Key 本身必须保密，绝不能被提交到代码仓库里。
限流是另一回事。接口会限制你在单位时间内的请求次数和消耗的 Token 数，分别叫 RPM 和 TPM。超过限制后会返回 429 状态码。写脚本时必须处理这种响应，做好退避重试。很多初学者第一版脚本不考虑限流，跑批量任务时请求发得太快，结果大面积的 429，然后整个循环崩溃，教训非常深刻。
成本预估则关乎你的钱包。模型计费通常按 Token 数计算，每 1000 个 Token 的价格因模型和服务商而异。在写脚本前，我建议先按任务量估算一下：单次请求大约消耗多少输入 Token、期望生成长度是多少、总共要跑多少次。这样心里有个预算底线，避免脚本跑完一看账单吃惊。后面有一节专门讲成本控制，这里先有个概念即可。
3. 完整调用脚本逐块拆解：配置、请求、解析、容错
3.1 配置分离：环境变量管理密钥与模型参数
现在开始正式写代码。第一步是建项目结构，第二步是配置管理。我习惯的项目目录长这样：
llm_toolkit/ ├── .env ├── .env.example ├── requirements.txt └── llm_call.py.env 存放真实配置，.env.example 是模板，把它提交到版本库，让其他人知道需要配置哪些变量。.env 内容大致如下：
LLM_API_KEY=sk-your-key-here LLM_BASE_URL=https://api.example.com/v1 LLM_MODEL=your-model-name LLM_TIMEOUT=60Python 里用 python-dotenv 加载：
from dotenv import load_dotenv import os load_dotenv() API_KEY = os.getenv("LLM_API_KEY") BASE_URL = os.getenv("LLM_BASE_URL") MODEL = os.getenv("LLM_MODEL") TIMEOUT = int(os.getenv("LLM_TIMEOUT", "60"))用好环境变量，脚本就能在不同环境和不同模型服务之间灵活切换，改配置不需要改代码。这是我第八天强烈建议养成的第一个工程习惯。
3.2 请求会话：超时、重试、连接池一个都不能少
网络请求不一定每次都成功。超时、连接断开、服务端 5xx 错误都可能发生。如果不做处理，脚本就会在中间死掉，前面跑的数据全部白费。
我推荐使用 requests.Session 而不是直接 requests.post。Session 会自动复用底层 TCP 连接，对连续调用同一主机的场景有明显性能提升。同时设置连接超时和读取超时：连接超时（connect timeout）指建立连接的最长时间，读取超时（read timeout）指从服务器读取数据的间隔时间。流式输出时读取超时尤其重要，它保证模型长时间没有返回新内容时脚本不会无限等下去。
重试策略推荐使用指数退避：第一次失败后等 1 秒，第二次等 2 秒，第三次等 4 秒，同时设置最大重试次数。重试时要注意区分错误类型：网络错误可以立刻重试，4xx 类错误一般不应重试（改代码或者调整参数才能解决），5xx 类错误可以重试。429 限流错误也可以通过响应头里的 Retry-After 字段决定等待时长。
下面是一个带超时和重试的请求骨架：
import time import requests def request_with_retry(session, url, payload, headers, max_retries=3): for attempt in range(max_retries): try: resp = session.post(url, json=payload, headers=headers, timeout=(10, 60)) if resp.status_code in (200, 200): return resp if resp.status_code in (429,) or resp.status_code >= 500: wait_time = 2 ** attempt print(f"状态码 {resp.status_code}，等待 {wait_time} 秒后重试") time.sleep(wait_time) continue resp.raise_for_status() except requests.exceptions.Timeout: wait_time = 2 ** attempt print(f"请求超时，等待 {wait_time} 秒后重试") time.sleep(wait_time) continue except requests.exceptions.RequestException as e: print(f"请求异常：{e}") time.sleep(2 ** attempt) continue raise RuntimeError("重试次数耗尽，请求失败")这段代码虽然还不完整，但已经包含了一个工程脚本最基本的容错骨架。
3.3 响应处理：流式逐句与完整解析双通道
响应处理是脚本的核心。流式模式下，逐行读取响应文本，把每行去掉 data: 前缀后解析成 JSON，再提取 delta 里的 content。非流式模式则直接把整个返回体解析成 JSON，从 choices 里拿 content。
流式解析函数写成生成器：
def stream_chat_response(resp): for line in resp.iter_lines(): if not line: continue line = line.decode("utf-8") if line.startswith("data:"): data_str = line[5:].strip() if data_str == "[DONE]": return data = json.loads(data_str) delta = data["choices"][0].get("delta", {}) content = delta.get("content", "") if content: yield content这个生成器会把每段增量文本逐次交给调用方，调用方可以立即打印、写入文件或进行后续处理。使用生成器的好处是代码结构清晰，内存占用也小，不会积压全部输出。
非流式解析则简单得多，但要加上一个“提取纯文本”的容错步骤。模型有时会返回带 Markdown 标记的内容，你可以按自己的需要决定保留还是去除。统一封装一个解析函数，返回最终文本和用量统计，后续调用方就不用关心响应结构了。
3.4 一个完整可跑的通用的调用脚本
把前面的模块拼起来，就是一个结构完整的调用脚本。下面这份代码我尽量做到“可以直接模仿使用”，你只需把 .env 里的配置换成自己的服务信息。
import json import os import time import logging import requests from dotenv import load_dotenv load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" ) logger = logging.getLogger(__name__) API_KEY = os.getenv("LLM_API_KEY") BASE_URL = os.getenv("LLM_BASE_URL") MODEL = os.getenv("LLM_MODEL") TIMEOUT = int(os.getenv("LLM_TIMEOUT", "60")) def build_headers(): return { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } def build_payload(messages, temperature=0.7, max_tokens=1000, stream=False): return { "model": MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": stream, } def chat_once(messages, temperature=0.7, max_tokens=1000): url = f"{BASE_URL}/chat/completions" headers = build_headers() payload = build_payload(messages, temperature, max_tokens, stream=False) for attempt in range(3): try: resp = requests.post(url, json=payload, headers=headers, timeout=(10, TIMEOUT)) if resp.status_code == 200: data = resp.json() content = data["choices"][0]["message"]["content"] usage = data.get("usage", {}) logger.info(f"请求成功，输入Token {usage.get('prompt_tokens')}，输出Token {usage.get('completion_tokens')}") return content, usage if resp.status_code in (429,) or resp.status_code >= 500: wait_time = 2 ** attempt logger.warning(f"接口返回 {resp.status_code}，{wait_time} 秒后重试") time.sleep(wait_time) continue resp.raise_for_status() except requests.exceptions.RequestException as e: logger.error(f"请求异常: {e}") wait_time = 2 ** attempt time.sleep(wait_time) raise RuntimeError("调用大模型接口失败") def stream_chat(messages, temperature=0.7, max_tokens=1000): url = f"{BASE_URL}/chat/completions" headers = build_headers() payload = build_payload(messages, temperature, max_tokens, stream=True) resp = requests.post(url, json=payload, headers=headers, timeout=(10, TIMEOUT), stream=True) if resp.status_code != 200: raise RuntimeError(f"接口返回 {resp.status_code}: {resp.text}") for line in resp.iter_lines(): if not line: continue line = line.decode("utf-8") if line.startswith("data:"): data_str = line[5:].strip() if data_str == "[DONE]": break try: data = json.loads(data_str) delta = data["choices"][0].get("delta", {}) content = delta.get("content", "") if content: yield content except json.JSONDecodeError: logger.warning("解析流式数据失败，跳过该行") if __name__ == "__main__": messages = [ {"role": "system", "content": "你是一个Python技术专家"}, {"role": "user", "content": "请用三句话解释什么是生成器"} ] print("=== 非流式调用 ===") text, _usage = chat_once(messages) print(text) print("=== 流式调用 ===") for chunk in stream_chat(messages): print(chunk, end="", flush=True) print()这个脚本已经可以独立完成一次完整的模型调用，并且兼顾了超时、重试、日志和两种响应模式。第八天的落地成果，把它作为基础版就够了。
4. 脚本跑通后，紧接着踩到的四个典型坑
4.1 上下文超长：超出窗口时返回的不是错误而是噪音
第一次跑通脚本后，很多人会兴奋地开始多轮对话。多轮对话意味着 messages 数组不断变长，历史消息全部塞进请求里。一旦超过模型的上下文窗口长度，不同服务的表现不太一样，有的是直接报错，有的则会静默截断历史消息，导致模型“失忆”，回答变得奇怪。
踩过这个坑后我做的第一件事是给 messages 设置长度上限。最简单的方案是只保留最近的 N 轮对话，这个 N 根据模型窗口大小和业务需求来定。更精细一些则用字符数或 Token 数估算，超长时优先丢弃最早的对话内容。另外，在做长文档分析时，如果文档本身超出窗口，可以考虑先做摘要压缩再送入模型，不要强塞原文。
还有一个隐蔽的问题：返回的 finish_reason 不等于 length 就说明安全。有些接口在截断超长上下文时不会明显报错，而是继续返回看似正常的内容。所以判断“上下文是否超限”，不能只看请求成功与否，还要把输入 Token 数记录到日志里，定期观察趋势。
4.2 限流 429：并发调用的正确解锁方式
批量任务脚本跑起来后，429 几乎是必然出现的。第一次看到满屏 429 时，我甚至以为是接口地址配错了。实际上就是请求频率超过了接口限制。此时最简单的解决方案是降低并发速度，而不是盲目增加重试次数。
我常用的策略是引入信号量并发控制和一个令牌桶节奏。比如限制同时最多 3 个请求，每秒最多 2 个请求。用 Python 的 threading.Semaphore 可以控制并发数量，在请求前加一个简单的 sleep 就能控制整体节奏。如果接口响应头里给了 Retry-After，则要严格按这个值等待，否则重试越快被封越久。
更稳妥的做法是把任务改成串行加小批量重试。对于几百上千条的批量任务，串行跑虽然慢，但稳定、好排查。等到对接口限制摸清楚了，再逐步提高并发。不要一开始就追求速度，稳定性远比那一两个小时的时间差重要。
4.3 日志缺失：报错的时候只能抓瞎
第一版脚本里我只用了 print 输出结果，结果某天批量任务跑到一半突然报错，屏幕上全是内容片段，根本看不出是哪条数据出了问题。吃了这个亏之后，我给脚本加上了结构化日志：每个请求开始记录任务ID和输入摘要，请求结束记录状态码、耗时、Token 数，异常时记录完整错误信息和当前处理的数据位置。
日志级别也要区分。普通进度信息用 INFO，异常警告用 WARNING，程序中断用 ERROR。这样在排查问题时可以按级别过滤，不用在几十万行输出里找线索。还要注意一点：日志里不要记录完整的 prompt 内容和 API Key，涉及业务敏感信息的数据都要脱敏，这也是工程素养的一部分。
4.4 成本失控：usage 字段是唯一靠谱的账单依据
最后一个坑是成本。使用大模型接口不是免费的，按 Token 计费的模式下，一个失控的循环可能让账单快速上升。最典型的情况是：批量任务里没有限制 max_tokens，结果模型生成了远超预期的内容，或者重试逻辑写错了，一条失败请求被重复发送几十次。
我对成本控制的做法有三条。第一，在每次响应的 usage 字段里读取实际 Token 数，自行做一个本地累加统计，而不是等到月底看账单。第二，给 max_tokens 设置一个业务上合理的最小值，宁可分多次调用，也不要一次生成过多无用内容。第三，对可能重复执行的任务做幂等设计，例如按输入内容的哈希值做缓存，已经调用过的请求直接读缓存结果。这三条做到后，成本基本就是可控的。
我做过一次简单的实验：1000 条短文本分类任务，如果不加缓存且允许无限重试，理论消耗可能是正常情况的 3 到 5 倍。而加上缓存和限流之后，不仅成本可控，整个任务的稳定性也大幅提升。
5. 从脚本到工具的进阶：命令行、批处理与服务化
5.1 用 argparse 让脚本变得可配置
第八天的成果如果只是自己在终端里跑，改 prompt 还要打开编辑器，那还差一步。把它改造成一个标准命令行工具，通过参数传入 prompt、模型参数、输出路径，这样其他人也能用。
argparse 的标准用法很简单：
import argparse parser = argparse.ArgumentParser(description="大模型调用命令行工具") parser.add_argument("--prompt", type=str, required=True, help="输入的问题") parser.add_argument("--system", type=str, default="", help="系统提示词") parser.add_argument("--output", type=str, default="", help="保存结果的路径") parser.add_argument("--temperature", type=float, default=0.7, help="采样温度") args = parser.parse_args()改造后的脚本就能这样用：
python llm_call.py --prompt "写一段二分查找的Python代码" --temperature 0.3这一步的价值是它让你的脚本从“写给自己看”变成了一个可复用的小工具。后续无论对接自动化流程还是交给同事使用，都非常方便。
5.2 批量任务：从单次调用到跑完 1000 条数据
单次调用搞定后，下一步往往是批量处理。比如给一组产品文案生成标题，给一批问题生成答案。批处理脚本的核心设计是：从文件读取输入、逐条调用模型、结果按顺序落盘、支持断点续跑。
断点续跑是我要特别强调的一点。一次性处理 1000 条数据时，中间极可能遇到网络问题或限流。如果程序崩溃后只能从头开始，时间成本和金钱成本都不可接受。所以我的批处理脚本长这样：每处理一条数据，就把结果写入以任务ID命名的输出行，同时记录该条的状态；下次启动时先扫描已经完成的ID，跳过已处理的部分。
这种设计听起来简单，但实际用下来能节省大量时间。第八天如果时间允许，我强烈建议把脚本改造成支持批处理和续跑，这比多学两个语法知识点更有实战价值。
5.3 服务化思考：当调用量上来之后的取舍
脚本再往上走，就是服务化。用 FastAPI 或 Flask 包一层 HTTP 接口，把调用脚本变成一个内部服务。这一步并不难，但不要急着做。如果你的调用量还停留在每天几十次、几百次，脚本完全够用；等到多个业务方都要接同一个模型能力时，服务化带来的接口收敛、日志统一、并发控制才有价值。
服务化和脚本化本质上是同一套核心逻辑，差别在于外层换上了 Web 框架、增加了并发安全处理和接口文档。我见过一些团队脚本都没写好就急着上服务，结果并发一高各种问题集中爆发。我的建议是先把脚本逻辑吃透，再考虑服务化。这样无论未来技术栈怎么演进，核心的大模型调用能力始终握在自己手里。
回头看看这八天，从第一天认识变量和循环，到最后能独立写出一个完整的大模型调用脚本，变化其实是巨大的。最后分享一个我个人的小习惯：脚本写完之后，刻意把它推倒重写一遍。第一次写是为了跑通，第二次写是为了结构和可维护性。两遍写下来，你会发现自己对 Python 的理解和对大模型接口的熟悉程度，比连续写十个练习脚本都要扎实。这个综合实战项目，值得你认真做，也值得你反复改。
