企业官网建设流程全解析

简介：大模型API正在重塑文本处理的效率边界。通过Python脚本调用DeepSeek等模型的接口，开发者可以将分类、摘要、关键词抽取等重复性工作封装为自动化流水线。理解API调用的基础原理——密钥管理、流式输出与temperature、max_tokens等参数调优——是构建可靠脚本的前提。在此基础上，借助线程池控制并发、设计指数退避重试、实现JSONL断点续跑，能显著提升批量任务的吞吐与容错性。无论是客服评论情感分析，还是海量文档结构化，这套方法都把原本按天计的体力劳动压缩到分钟级。围绕DeepSeek API的工程实践，系统梳理从单次调用到批量落盘的关键环节与避坑经验，为开发者提供一条可复制的文本批处理最短路径。

1. 大模型 API 自动化脚本：为什么说它是文本批处理的最短路径

如果你手头有几百上千条文本要做分类、摘要、抽取关键词，逐条复制到对话网页里问大模型，点几次就崩溃。大模型 API 自动化脚本解决的正是这个问题：用 Python 直连大模型接口，把重复的"提问—回答—复制"变成一条命令跑完。这套脚本围绕 DeepSeek 大模型 API 展开，覆盖密钥配置、流式输出、多轮上下文、批量并发和断点续跑，适合需要批量处理文本的开发者、运营和分析师。花半小时改改参数，就能把原本两天的体力活压缩成几分钟。

2. DeepSeek API 基础调用：密钥、流式输出与三种必调参数

2.1 为什么选 API 而不是本地起模型

本地部署一个同量级的模型，先要解决显存问题。量化后跑起来，推理速度又受显卡性能限制，并发一上来就排队。API 方式把算力放在远端，本地只需要发 HTTP 请求，按 token 计费，成本可控，也不用操心显卡驱动和环境冲突。我第一次用 API 跑批量任务时，最大的感触是：延迟比预想低，成本比预想少，真正花时间的反而是写脚本和处理返回格式。

DeepSeek 的接口兼容 OpenAI 的调用风格，这意味着不需要为它单独学一套 SDK，直接用常见的 openai 库，改一下 base_url 就能跑。对于已经写过 OpenAI 接口脚本的开发者，迁移成本几乎为零。对于没写过的，照着官方示例改几行也能上手。我一般习惯把 API 调用封装成一个函数，这样后续批量任务、重试、日志都可以挂在同一个入口上。

2.2 环境准备与密钥管理

密钥不要硬编码在脚本里。常见做法是在项目根目录放一个.env文件，用 python-dotenv 加载。这样脚本可以分享，密钥留在本地，进不了版本库。

pip install openai python-dotenv

# .env 文件内容 DEEPSEEK_API_KEY=sk-你的密钥 DEEPSEEK_BASE_URL=https://api.deepseek.com

import os from dotenv import load_dotenv load_dotenv() api_key = os.getenv("DEEPSEEK_API_KEY") base_url = os.getenv("DEEPSEEK_BASE_URL")

这段代码的逻辑很简单：load_dotenv 读取.env文件里的键值对，os.getenv 把它们取出来。注意.env文件要写进.gitignore，否则推到仓库就等于公开了密钥。我见过某同学的脚本因为密钥硬编码被传到仓库，几分钟内就被爬虫扫走，账单直接爆掉。从那以后我所有的 API 脚本第一行必然是加载环境变量。

2.3 第一次调用：非流式与流式两种写法

非流式调用适合结果较短的任务，比如分类、情感判断、实体抽取。请求发出去，等完整响应回来再处理。流式调用适合生成长文本，比如写文章、翻译长段落，逐字返回，用户看到的是"打字机"效果，也能规避超长输出时连接中断的问题。

from openai import OpenAI client = OpenAI(api_key=api_key, base_url=base_url) response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是文本分类助手，只输出类别名称。"}, {"role": "user", "content": "这台设备开机后频繁重启，售后一直不回复。"} ], temperature=0, max_tokens=64 ) reply = response.choices[0].message.content print(reply)

这个例子做了三件事：创建客户端、发起聊天补全请求、取出模型回复。messages 里 system 是系统指令，user 是输入文本。temperature 设 0 是为了让分类结果稳定，max_tokens 限 64 足够输出一个类别名。返回的 response 对象里还有 usage 字段，能看到本次请求消耗的 token 数，批量跑的时候我习惯把 usage 也记进日志。

流式调用写法差异不大，关键在 stream=True 和遍历响应块：

stream = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是一名文案编辑，润色以下句子。"}, {"role": "user", "content": "这个功能很实用。"} ], stream=True ) full_text = "" for chunk in stream: delta = chunk.choices[0].delta.content if delta: full_text += delta print(full_text)

流式响应的每个 chunk 只带一小段增量内容，手动拼接成完整文本。这里要注意判断 delta 是否为空，因为最后一个 chunk 往往不含内容。另外，流式模式下拿不到完整的 usage 信息，需要自己估 token 消耗，我的经验是按 full_text 的长度乘一个系数估算，误差不大。

2.4 三个必调参数：temperature、max_tokens、top_p

这三个参数是每次调用都要过一遍脑子的。temperature 控制随机性，0 到 2 之间。分类、抽取、格式化输出这类任务设 0 或 0.1；写文案、头脑风暴可以拉到 0.8 到 1.2。max_tokens 限制输出长度，设太小会截断，设太大浪费 token。top_p 是另一种采样策略，按概率累积截断候选词，和 temperature 二选一调节就行，一般不同时动。

系统提示词里也应该写清楚输出格式。比如"只输出 JSON"和"输出一个JSON对象，不要额外解释"，效果差别很大。这里有个小坑：即使你写了只输出 JSON，模型偶尔还是会带出 ```json 代码块标记，解析的时候要做兼容处理。这个坑后面专门讲。

3. 从单发到工具化：重复调用、上下文管理与指数退避重试

3.1 把调用封装成函数

单次调用没问题之后，下一步是把它封装成可复用的函数。封装的核心是统一处理：密钥从环境变量读、异常统一捕获、重试策略内置、返回格式固定。我一般这样写：

import os import time import random import json from openai import OpenAI client = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=os.getenv("DEEPSEEK_BASE_URL") ) def chat(messages, temperature=0, max_tokens=1024, retries=5): for attempt in range(retries): try: resp = client.chat.completions.create( model="deepseek-chat", messages=messages, temperature=temperature, max_tokens=max_tokens ) return resp.choices[0].message.content except Exception as e: wait = 2 ** attempt + random.uniform(0, 0.5) print(f"第 {attempt + 1} 次重试，等待 {wait:.2f}s，错误: {e}") time.sleep(wait) raise RuntimeError("重试多次仍然失败")

这个封装的逻辑是：每次请求失败先等 2 的 attempt 次方秒，再加 0 到 0.5 秒的随机抖动，然后重试，最多 5 次。随机抖动的作用是防止多个并发请求同时失败后，再同时重试造成二次拥塞。retries 参数按需调整，本地网络不稳定时我会调到 8，生产环境一般 3 到 5 就够了。

函数返回的是纯文本字符串。如果需要 JSON，可以在函数里加一个解析步骤，把模型输出转成字典。但解析逻辑单独拎出来更清晰，因为不同任务的 JSON 结构不一样。

3.2 上下文管理：多轮对话与 token 预算

批量任务大多数单轮问答就够了，但有些场景需要多轮：先让模型提炼要点，再基于要点做判断。这种场景要把历史消息累积在 messages 列表里，每次请求都带上全部上下文。

messages = [ {"role": "system", "content": "你是客服质检助手。"}, {"role": "user", "content": "这条工单里客户最不满意的点是什么？"}, {"role": "assistant", "content": "客户主要不满响应速度，等待了三天才有人处理。"}, {"role": "user", "content": "基于以上结论，给出改进建议。"} ]

messages 的结构是 system 开头，之后 user 和 assistant 交替出现。assistant 的内容来自上一轮模型输出，自己拼进去。这里最大的坑是 token 预算：上下文越长，费用越高，而且超出模型窗口会直接报错。我一般用估算函数控制长度，一个汉字大约占 1.5 到 2 个 token，1000 个汉字的对话大概 1500 到 2000 token。

控制策略是：保留 system 消息和最近几轮对话，把最早的历史消息丢弃。常见做法是设一个最大轮数，超过就砍掉最旧的。对批量任务来说，其实很少真正用到多轮，单轮加详细的 system 提示词就能解决大多数问题。

3.3 指数退避重试

重试不只是捕获异常那么简单，要区分错误的类型。429 是限流，503 是服务暂不可用，这两种值得重试。401 是密钥错误，retry 多少次都没用，直接报错更省时间。

from openai import RateLimitError, APIConnectionError def chat_with_retry(messages, retries=5): for attempt in range(retries): try: return chat(messages) except RateLimitError: wait = 2 ** attempt + random.uniform(0, 0.3) time.sleep(wait) except APIConnectionError: wait = 3 + random.uniform(0, 1) time.sleep(wait) except Exception: raise

这个版本把限流和连接错误分开处理。限流通常几十秒后会恢复，连接错误可能是网络瞬时抖动，等 3 秒左右再试。其他异常直接抛出，不给重试掩盖真正 bug 的机会。我在实际项目里踩过这个坑：把所有异常都吞掉重试，结果密钥错了也傻等 5 轮，浪费两分钟才报错。

4. 批量任务落地：并发控制、断点续跑与 JSONL 落盘

4.1 场景设定与输入输出格式

批量任务最常见的形态是：一个 CSV 或 Excel 文件，每行一条文本，需要逐条处理，结果写回另一个文件。以客户评论情感分类为例，输入是input.csv，包含id和text两列，输出要求每条数据附上情感标签和置信度理由。输出我用 JSONL 格式，每行一个 JSON 对象，方便追加和断点续跑。

id,text 1,快递迟到了两天，客服态度还很差 2,产品很好用，包装也很结实 3,退货流程太复杂，再也不买了

JSONL 的好处是每行独立，追加写入不需要重排整个文件。处理完成后还可以用 pandas 读回 DataFrame 做统计分析。

4.2 串行改并发：线程池与 QPS 的关系

一条一条跑最稳妥，但 1000 条数据串行执行，每条 2 秒，半小时就过去了。改成并发后，速度提升非常明显。常见做法是用 ThreadPoolExecutor，因为大模型 API 是网络 IO 任务，线程池足够，而且代码改动最小。

import csv import json import threading from concurrent.futures import ThreadPoolExecutor, as_completed input_file = "input.csv" output_file = "result.jsonl" lock = threading.Lock() def process_row(row): messages = [ {"role": "system", "content": "你是情感分类助手。只输出 JSON，格式：{\"label\": \"正面/负面/中性\", \"reason\": \"简要理由\"}"}, {"role": "user", "content": row["text"]} ] output = chat(messages, temperature=0, max_tokens=256) # 解析 JSON，兼容 ```json 包裹 cleaned = output.strip() if cleaned.startswith("```"): cleaned = cleaned.split("```")[1] if cleaned.startswith("json"): cleaned = cleaned[4:].strip() result = json.loads(cleaned) return row["id"], result with open(input_file, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) rows = list(reader) with open(output_file, "w", encoding="utf-8") as out: with ThreadPoolExecutor(max_workers=6) as executor: futures = {executor.submit(process_row, row): row for row in rows} for future in as_completed(futures): row_id, result = future.result() with lock: out.write(json.dumps({"id": row_id, **result}, ensure_ascii=False) + "\n") out.flush()

这段脚本的逻辑分三步：定义单条处理函数，提交到线程池，拿到结果立即写文件。max_workers 我设 6，这是综合 QPS 限制和稳定性后的经验值。线程太多会触发限流，429 报错反而拖慢整体速度；太少又浪费带宽。实测中，6 到 8 个线程同时跑比较稳。写文件时加锁是因为多个线程同时写同一个文件对象会乱行。

为什么不用 asyncio？线程池够用，而且不用改 chat 函数为异步版本。如果任务规模大到上万条，可以考虑异步加连接复用，但对大多数场景来说，线程池是性价比最高的方案。

4.3 断点续跑的实现

批量跑到一半崩了，重头再来非常痛苦。断点续跑的思路是：启动时先扫描输出文件，记录已处理的 id，跳过这些行。崩溃后重新执行，只跑没完成的部分。

import os def load_completed_ids(output_file): if not os.path.exists(output_file): return set() done = set() with open(output_file, "r", encoding="utf-8") as f: for line in f: data = json.loads(line) done.add(data["id"]) return done completed = load_completed_ids(output_file) todo = [row for row in rows if row["id"] not in completed] print(f"共 {len(rows)} 条，已完成 {len(completed)}，剩余 {len(todo)}")

这里注意：id 在 CSV 里是字符串，在 JSONL 里取出来也是字符串，类型要对齐。断点续跑配合逐行 flush 写入，是批量任务的标准组合。我跑过最久的一次是 8000 条评论，中途断了两回，续跑只用了六分钟全部完成。

5. 避坑与排查：截断、幻觉、JSON 解析失败的五个实战记录

5.1 输出被截断，返回内容不完整

现象：max_tokens 设为 128，让模型写一段产品描述，返回内容在中间突然断掉，没有结尾。

原因：max_tokens 限制的是输出长度，不是输入长度。模型觉得需要 400 token 才能写完，但上限只有 128，生成到上限就被强制截断。API 调用本身成功了，不会报错，但内容残缺。

解决：按任务估算输出长度，拿不准就设 1024。生成完可以用 usage 字段里的 completion_tokens 和 max_tokens 对比，如果非常接近，说明可能被截断，需要调大上限重新跑。

5.2 同样的输入，两次结果不一样

现象：对同一条评论调用两次情感分类，一次是"正面"，一次是"负面"。

原因：temperature 设太高。默认值是 1.0，随机性较强，对分类这种确定性任务会翻车。另外提示词没有限定输出格式，模型自主发挥空间太大。

解决：分类和抽取任务把 temperature 设成 0。system 提示词里写明"只输出 JSON，不要解释"。如果还是不稳定，把候选标签全部列在提示词里，要求只能从列表里选。

5.3 JSON 解析报错，模型输出带代码块

现象：json.loads 抛 JSONDecodeError，打印原始输出才发现模型返回了 ```json 包裹的内容，或者输出里混入了"好的，这是结果："之类的废话。

原因：模型没有严格遵守"只输出 JSON"的指令，把内容包装进了 markdown 代码块。这在多个模型上都存在，不属于某个特定服务的 bug。

解决：解析前先剥离代码块标记。上面批量脚本里的 cleaned 处理就是这个逻辑。更稳妥的做法是加一个兜底逻辑，用正则找出第一个{到最后一个}的子串再解析。我见过有人直接用output[output.find("{"):output.rfind("}")+1]，简单粗暴但有效。

5.4 并发一高就被限流，频繁 429

现象：把 max_workers 从 6 调到 20 后，日志里开始大量出现 429 错误，整体速度反而比 6 线程时更慢。

原因：API 有 QPS 限制，短时间请求太多触发限流。重试虽然是等一段时间再发，但 20 个线程同时等，恢复后又是同时打过去，形成一波一波的脉冲请求，限流雪上加霜。

解决：降低线程数，6 到 8 是稳妥区间。另外把日志级别打开，观察 429 的密集程度，频繁出现就先降线程数。指数退避加随机抖动也能缓解脉冲效应。

5.5 密钥被提交到仓库，账单异常

现象：某开发者的脚本传到公开仓库后，第二天收到服务商的预警邮件，显示有大量陌生 IP 调用。

原因：.env没有写进.gitignore，密钥随脚本一起被提交。爬虫会自动扫描公开仓库里的敏感信息，被扫到后盗刷额度。

解决：提交前检查 git 状态，.gitignore里必须包含.env。我现在的习惯是创建项目时第一件事就把.env加进.gitignore，再写代码。如果已经泄露，立即在平台后台重置密钥，一分钟都不能拖。

6. 进阶：用回归样例集锁住 prompt 效果

批量脚本写完能跑只是第一步，改 prompt 才是无底洞。今天把 system 提示词从"输出 JSON"改成"严格输出 JSON"，可能一个分类任务的效果变好了，另一个抽取任务的格式又坏了。单测大模型的输出不现实，但我常用一个轻量回归样例集来锁住效果。

准备 15 到 20 条固定输入，覆盖典型场景和边界情况：正常短文本、超长文本、空字符串、纯数字、表情符号、特殊符号。每条标注期望输出的关键点，比如分类任务的期望标签、抽取任务必须包含的字段。每次改 prompt 或换模型版本，先跑一遍样例集，对比输出和期望的差异，差异超过预期就直接回滚。

test_cases = [ {"text": "快递迟到了两天，客服态度还很差", "expect_label": "负面"}, {"text": "产品很好用，包装也很结实", "expect_label": "正面"}, {"text": "商品一般般，没有特别好也没有特别差", "expect_label": "中性"}, {"text": "", "expect_label": None}, ] for case in test_cases: result = classify(case["text"]) if result["label"] != case["expect_label"]: print(f"不一致: {case['text'][:20]} 期望 {case['expect_label']} 实际 {result['label']}")

这个脚本不复杂，但价值很高。它把 prompt 的改动从"感觉还行"变成"可以验证"。空字符串那个用例你可能觉得没必要，但批量数据里什么脏数据都有，空值、纯符号、异常编码，模型在这些输入上的表现经常出乎意料。

从那以后，我每次改完 prompt 或者换模型版本，都强制走一遍固定样例集，确认没破坏已有任务才跑全量数据。这个习惯帮我挡掉了好几次静默回归，也让脚本在换模型版本时心里有底。希望帮到你。

本文还有配套的精品资源，点击获取