DeepSeek API 批量处理与数据清洗实战指南-育师

简介：《DeepSeek详细使用教学》面向希望快速上手DeepSeek的初学者、办公人员与数据分析学习者，围绕账号注册登录、核心功能认知和实际操作展开，帮助读者建立从数据导入、清洗、分析、可视化到文本生成与摘要的完整使用路径。压缩包内共1个doc文档，约38KB，以文字教程形式集中呈现功能入口、操作顺序与关键参数说明，便于按章节查阅和对照练习。内容还涉及任务自动化、工作流优化、模型训练与部署，并补充批量处理、自定义模板、优化模式及简洁表达需求等进阶技巧，对数据导入错误、生成结果不符预期、自动化任务执行失败等问题给出排查思路。已有667人学习，适合需要一份结构清晰、可直接照做的DeepSeek入门与功能速查材料的读者。

1. 别急着点按钮：DeepSeek 这份教学真正的分水岭在第一次脚本调用

大多数人拿到 DeepSeek 的使用教学，路径都一样：照着截图点一遍菜单，试两句问答，关掉，然后觉得这东西跟普通聊天框没区别。分水岭其实不在你点了哪个按钮，而在你第一次把它接进自己的脚本、让它批量处理一批真实数据的那一刻。这份资源把注册登录、数据导入清洗、文本生成摘要、任务自动化、模型训练部署串成了一条线，覆盖的正是从“试一试”到“真跑起来”的那一段。它适合手上有 CSV 或 Excel 要整理、有大量文本要摘要、有重复劳动想自动化的从业者；如果你只想要一个聊天的东西，那这份东西的价值你大概率吃不到一半。

2. 接入形态选型：对话页、API、自托管权重各管哪一段

先把一个常见的误解摆正。教学里写的“数据导入”“数据清洗”“模型训练与部署”，落到真实工作里，绝大部分不是你点一下按钮系统就全自动做完，而是你把数据整理成模型能读的文本，用对话页或接口把问题问对，再把输出接回自己的表格、脚本或数据库。界面上的模块名和位置在不同时间会有调整，这条链路不会变。所以选型要选的是“我用哪种形态接进去”，不是“我点哪个菜单”。

2.1 三种形态的边界与成本

对话页、API、自托管权重，这三样不是替代关系，是分工关系。对话页负责探索：你不确定该怎么切字段、怎么措辞、输出长什么样的时候，在那儿试错最便宜，改了立刻看到结果。API 负责量产：一旦提示词稳定了，就把它搬进脚本，让它一次跑几百上千条。自托管权重负责边界：数据完全不能出本地、或者要离线跑在隔离环境里的时候才考虑，代价是硬件和运维。

形态	上手成本	数据是否出本地	适合做的事	不适合做的事
对话页	极低	是	试提示词、一次性整理、小样本对照	批量、可重复、需要留痕
API	中	是	批量清洗、摘要、生成、接进流水线	需要零外网的场景
自托管权重	高	否	数据不出域、离线、定制微调	快速验证、算力紧张时

我一般的顺序是：对话页把提示词磨到“输入同样的东西，输出结构稳定”，再花半小时搬成 API 脚本，最后才想自托管。跳过前两步直接上自托管，八成是给自己找麻烦，模型跑起来了，提示词还没定型，返工成本翻倍。

还有一点值得单独说：界面上那些“深度思考”“优化模式”一类的开关，不要一上来全开。全开的好处是复杂问题表现好一些，坏处是慢、贵，而且你分不清效果提升来自提示词还是来自开关。常见做法是拿十条固定样本，逐个开关做对照，把每个开关的实际收益记下来，再决定生产环境开哪个。这部分没有标准答案，靠自己的样本说话。

2.2 账号与密钥：注册登录之后第一件要做的事

注册登录本身没什么可讲的，邮箱或手机号走一遍流程就行。真正该在第一天就定下来的是密钥怎么管。密钥一旦硬编码进脚本、又提交到了代码仓库，后面只能靠轮换收拾，属于典型的“后悔药”场景。我习惯的做法是环境变量加本地配置文件，配置文件进忽略清单，密钥定期轮换。

# 密钥和地址都走环境变量，脚本里不出现明文字符串 export DEEPSEEK_API_KEY="sk-你的密钥" export DEEPSEEK_BASE_URL="https://api.deepseek.com/v1" # 本地放一份 .env 供开发用，务必确认它已被版本控制忽略 grep -n "env" .gitignore || echo ".env" >> .gitignore

这两行命令的含义很直白：第一段把密钥和接口地址注入当前 shell 会话，子进程能继承；第二段检查忽略清单里有没有.env，没有就补上。参数上只有两个需要留意，DEEPSEEK_BASE_URL末尾带不带/v1会影响拼接出来的完整路径，同一个项目里统一一种写法，别一半带一半不带，排查起来会非常难受。密钥的权限范围尽量按最小化给，能只读就别给写。

2.3 用 curl 打通第一次调用

第一次调用建议用 curl，把中间层全部剥掉，出问题能立刻定位到是网络、鉴权还是参数。请求体走的是业界通用的对话补全风格，messages是一个数组，system定角色，user放任务，多轮就继续往后追加assistant和user。

curl -s "$DEEPSEEK_BASE_URL/chat/completions" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer $DEEPSEEK_API_KEY" \ -d '{ "model": "deepseek-chat", "messages": [ {"role": "system", "content": "你是一名数据整理助手，回答只给结论，不解释过程。"}, {"role": "user", "content": "把下面三行订单状态归成 待付款/已发货/已完成 三类：已下单未支付、包裹已寄出、交易关闭"} ], "temperature": 1.0, "max_tokens": 512, "stream": false }'

这段请求里，model决定走哪条能力线，通用对话和推理类模型在同一个接口下用不同名字区分；temperature控制随机性，做分类、抽取这类要稳定结果的任务我一般压到 0 到 0.3，做文案发散才往上调；max_tokens是输出上限，设太小会被截断，症状是句子说到一半没了，很多人第一反应是模型坏了，其实是这个值不够；stream设成false时一次性返回完整结果，设成true会分块吐，适合前端要打字机效果的场景，脚本批处理反而建议关掉。

跑通之后把返回的 JSON 拿去看一眼结构，choices[0].message.content是正文，usage里有输入输出 token 的计数。这个计数是你后面估成本的唯一依据，别凭感觉。

2.4 Python 封装：超时、重试与流式输出的取舍

curl 通了就搬 Python。封装的时候有三个东西必须加进去：超时、重试、以及一个能区分“可重试错误”和“不可重试错误”的判断。没有超时的请求在网络抖动时会把整个批处理卡死，这是血泪经验。

import os, time, json, requests BASE = os.environ["DEEPSEEK_BASE_URL"] KEY = os.environ["DEEPSEEK_API_KEY"] HEADERS = {"Content-Type": "application/json", "Authorization": f"Bearer {KEY}"} def chat(messages, model="deepseek-chat", temperature=0.3, max_tokens=1024, timeout=60, retries=3): payload = {"model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False} last_err = None for i in range(retries): try: r = requests.post(f"{BASE}/chat/completions", headers=HEADERS, json=payload, timeout=timeout) if r.status_code == 200: return r.json()["choices"][0]["message"]["content"] # 429 限流、5xx 服务端问题才值得重试，4xx 参数错重试没用 if r.status_code == 429 or r.status_code >= 500: last_err = f"{r.status_code} {r.text[:200]}" else: raise RuntimeError(f"不可重试: {r.status_code} {r.text[:200]}") except requests.Timeout as e: last_err = f"timeout: {e}" time.sleep(2 ** i) # 指数退避：1s、2s、4s raise RuntimeError(f"重试 {retries} 次仍失败: {last_err}") if __name__ == "__main__": out = chat([{"role": "user", "content": "用一句话说明什么是数据清洗"}]) print(out)

逻辑上分三层：正常返回直接取choices[0].message.content；限流和服务端错误进重试分支，用2 ** i做指数退避，避免短时间内反复撞墙；其余 4xx 直接抛异常，因为参数写错了重试一百次还是错。参数里timeout=60是单次请求上限，长文本任务可以放到 120，retries=3是重试次数，批量场景下不建议超过 5，否则一条坏数据会拖慢整个队列。temperature=0.3是给抽取类任务用的默认值，写文案的脚本单独传参覆盖。

流式输出在这个封装里是关掉的。批处理场景下流式没有收益，反而增加解析复杂度；只有做交互界面、需要边生成边显示的时候才值得打开。

3. 数据导入与清洗：CSV 喂进 DeepSeek 之前必须做的四件事

模型不是数据库，它读不懂你的表结构，也不会自己去做GROUP BY。教学里写的“系统自动识别缺失值和重复数据”，真实情况是：你得先把表整理干净、把该算的算完、把该裁的裁掉，再让它做它擅长的事——理解语义、归类、抽取、写摘要。这一章的四件事，做完之后你的成功率会明显不一样。

3.1 字段裁剪与编码归一

第一件事是减法。一张几十列的表直接转文本塞进去，token 成本翻几倍，噪声还会把关键信息淹掉。先想清楚你要回答的问题是什么，只留和问题相关的列。编码上，CSV 里中文列名变乱码是最常见的翻车点，源文件是 GBK 的时候不要硬用 UTF-8 读。

import pandas as pd # 源文件编码不确定时，先探测再读；常见是 utf-8 和 gbk df = pd.read_csv("orders_raw.csv", encoding="utf-8", dtype=str, keep_default_na=False) # 只保留和当前问题相关的列，列越多 token 越贵、干扰越大 keep = ["order_id", "city", "amount", "created_at", "status"] df = df[[c for c in keep if c in df.columns]] # 值归一：去空格、金额转数值、空值统一成可读占位符 df["city"] = df["city"].str.strip() df["amount"] = pd.to_numeric(df["amount"], errors="coerce") df = df.fillna({"city": "未知", "status": "未知"}) print(df.shape) print(df.dtypes.to_dict())

dtype=str让所有列先按字符串进来，避免order_id这种带前导零的编号被自动转成数字后丢掉零；keep_default_na=False防止NA、null这类字面值被误判成缺失；errors="coerce"把转不了的值变成 NaN，后面统一填充，比让异常直接抛出来更可控。裁剪之后先打印行列数和类型，这两个数字对不上的时候，问题基本都在读取阶段，不用往下走。

3.2 长表切块：表头复读与行窗口

行数一多就必须切。模型有上下文上限，超了要么报错要么被截断，截断是更可怕的那种，因为它不报错。切块的核心不是简单对半砍，而是每一块都要带表头，否则模型看到一堆没有列名的值，只能靠猜。

def chunk_rows(df, window=200, overlap=20, max_chars=6000): header = " | ".join(df.columns) lines = df.astype(str).agg(" | ".join, axis=1).tolist() chunks, i = [], 0 while i < len(lines): block = [header] + lines[i:i + window] # 每块都复读表头 text = "\n".join(block) # 字符超限就缩窗口，保证单块不撑爆上下文 while len(text) > max_chars and len(block) > 2: block = block[:-1] text = "\n".join(block) chunks.append(text) i += window - overlap # 留重叠，避免跨块的记录被切碎 return chunks

window=200是每块的行数，overlap=20是相邻块的重叠行，用来兜住跨越边界的记录——比如一条订单的备注被拆在两块里，重叠段能让至少一块看到完整内容。max_chars=6000是兜底红线，中文字符和 token 不是一比一，这个值偏保守，宁可多切几块也别赌上限。切完之后建议打印块数和最长块的长度，块数明显超过预期说明窗口设小了。

这里有个容易忽略的点：切块之后每块的结果要能拼回去。所以切的时候必须保留一个唯一键（比如order_id），让模型在输出里带上它，不然后面根本对不上号。

3.3 让输出可解析的 JSON

要求模型输出 JSON 这件事，光在提示词里写“请输出 JSON”是不够的，格式会飘。稳妥的做法是同时开接口的结构化输出开关，并在提示词里明确出现 json 这个词，两者配合命中率才高。

import json, re SYSTEM = ( "你是数据抽取助手。只输出一个 JSON 对象，不要任何解释文字、不要 markdown 代码块。" "字段固定为 order_id(string)、category(string)、confidence(number, 0-1)。" ) def extract(text, model="deepseek-chat"): msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": f"从下面的表格片段中归类，输出 json：\n{text}"}] payload = {"model": model, "messages": msgs, "temperature": 0.0, "response_format": {"type": "json_object"}} r = requests.post(f"{BASE}/chat/completions", headers=HEADERS, json=payload, timeout=90) raw = r.json()["choices"][0]["message"]["content"] try: return json.loads(raw) except json.JSONDecodeError: # 兜底：模型偶尔会包一层代码块，剥掉再试一次 m = re.search(r"\{.*\}", raw, re.S) return json.loads(m.group(0)) if m else None

response_format是结构化输出的开关，temperature=0.0进一步压随机性。兜底那段正则不是可有可无的，即便开了结构化输出，模型仍然有一定概率把结果包在代码块里，生产脚本里不加这层，解析失败率会集中在长尾上，很难查。confidence这个字段是我个人习惯加的，让模型自己标一下把握度，低于阈值的条目人工过一遍，比全量人工省力得多。

3.4 分析结论的复核口径

模型给出的“趋势”“占比”“同比增长”这类结论，一律不能直接信。原因很简单，它是在做语言层面的推断，不是真的在算。正确姿势是把计算留在 pandas 或 SQL 里，把解释留给模型：你算好每个月的环比数字，让它根据数字写一段归因描述，这样既用上了它的表达能力，又不会在数值上翻车。

具体做法是准备一张小对照表，抽查二十条，人工算一遍和模型输出比一比。差异集中在哪一类问题上，就把那一类从“模型算”改成“脚本算”。这个动作花不了多少时间，但能帮你划清边界。边界一旦划清，后面所有批量任务的可信度都会跟着上去。

4. 文本生成与批量自动化：从单次问答到能跑通夜的脚本

单次问答调好之后，下一步是让它能无人值守地跑。这一步的难点不在模型，在工程：并发怎么控、失败了怎么续、结果往哪儿放、同一个模板怎么复用。教学里提到的“批量处理”和“自定义模板”，落到代码上就是这一章的内容。

4.1 提示词四段式结构

我写生产用提示词基本固定四段：角色、约束、样例、输出格式。角色定语气和知识范围，约束写清不能做什么，样例给一两个正反例，输出格式钉死结构。四段齐全的提示词，稳定性比一句“帮我总结一下”高一个量级。

PROMPT = """你是电商售后文本的摘要助手。 约束： 1. 只保留用户诉求、涉及商品、情绪倾向三项信息； 2. 不臆测用户未提及的内容，信息缺失写“未提及”； 3. 单条摘要不超过 60 字。 样例： 输入：买了三天的耳机右耳没声音，联系客服两次都没人回，很生气。 输出：诉求=换货或退款；商品=耳机；情绪=愤怒；备注=客服两次未响应。 输出格式：严格按 JSON 输出，键为 need、product、emotion、note。 输入：{text} 输出："""

这里{text}是占位符，用的时候做字符串替换。样例那一段是最值钱的部分，它比任何形容词都更能约束输出。约束里的“信息缺失写未提及”是为了防止模型编造，不加这句，它遇到空字段会自己脑补一个合理值，而且看起来很真。

4.2 并发、限速与断点续跑

批量脚本最容易出问题的地方是跑了一半挂了，重跑要从头来。所以第一版就要把断点续跑做进去：每条任务用唯一 id 标记，成功的写进结果文件，启动时先读结果文件把已完成的 id 剔掉。

import json, os, time from concurrent.futures import ThreadPoolExecutor, as_completed def load_done(path): done = set() if os.path.exists(path): with open(path, encoding="utf-8") as f: for line in f: done.add(json.loads(line)["id"]) return done def run(rows, out_path="result.jsonl", workers=4, qps=3): done = load_done(out_path) todo = [r for r in rows if r["id"] not in done] interval = 1.0 / qps with open(out_path, "a", encoding="utf-8") as f, \ ThreadPoolExecutor(max_workers=workers) as pool: futures = {} for r in todo: futures[pool.submit(extract, PROMPT.replace("{text}", r["text"]))] = r["id"] time.sleep(interval) # 客户端侧限速 for fut in as_completed(futures): rid = futures[fut] try: res = fut.result() except Exception as e: res = {"error": str(e)} f.write(json.dumps({"id": rid, **({"result": res} if "error" not in res else res)}, ensure_ascii=False) + "\n") f.flush() # 立刻落盘，进程被杀也不丢

workers=4是并发线程数，qps=3是客户端限速，两个参数要一起调：并发高但限速低，线程大部分时间在 sleep；限速高但并发低，吞吐上不去。我的经验是先从workers=4, qps=3起步，看返回里有没有 429，有就往下降，没有就慢慢往上加。f.flush()这行别省，它是断点续跑能成立的前提，没有它缓冲区的数据在进程异常退出时会丢。

workers和qps的关系值得再说一句：它们不是同一个东西。workers控制同时在飞的请求数，qps控制每秒发起的新请求数。前者太高会瞬间打满，后者太高会持续打满，两种都会触发限流，但表现不一样，前者是突发性的 429 峰值，后者是持续性的。排查的时候看 429 是集中在开头还是均匀分布，就能判断该调哪个。

4.3 长文摘要的分级压缩

超过上下文的长文，一次喂进去必然截断。常见做法是分级压缩：先按段落切成小块，每块单独出摘要，再把这些摘要拼起来做二次摘要。这叫 map-reduce，落地很简单。

def summarize_long(text, seg=3000, rounds=2): # 第一级：按字符切段，段间留一点重叠，避免句子被切断 parts = [text[i:i + seg] for i in range(0, len(text), seg - 200)] summaries = [chat([{"role": "user", "content": f"用 100 字概括这段内容，保留数字和人名：\n{p}"}], temperature=0.3) for p in parts] merged = "\n".join(summaries) # 第二级：如果合并后仍然超长，再压一轮 for _ in range(rounds - 1): if len(merged) <= seg: break merged = chat([{"role": "user", "content": f"把下面的分点摘要合并成一段连贯摘要，去掉重复：\n{merged}"}], temperature=0.3) return merged

seg=3000是每段字符数，seg - 200的重叠是为了不让一句话被切成两半导致语义断裂。rounds=2是最大压缩轮数，一般两轮足够把一份长文档压到能一次处理。这个方法的代价是 token 翻倍，因为原文被读了两遍，所以只对真正需要的长文用，短文本直接一次过。

4.4 结果落库与模板复用

结果落 JSONL 是中间态，最后还是要进表。字段固定的任务直接建表，别用文本字段糊弄，后面查起来会想哭。

CREATE TABLE doc_summary ( doc_id VARCHAR(64) PRIMARY KEY, need VARCHAR(200), product VARCHAR(200), emotion VARCHAR(32), note TEXT, model VARCHAR(64), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 幂等写入：重复跑同一个 id 不报错，直接覆盖 INSERT INTO doc_summary (doc_id, need, product, emotion, note, model) VALUES (:doc_id, :need, :product, :emotion, :note, :model) ON CONFLICT (doc_id) DO UPDATE SET need = EXCLUDED.need, note = EXCLUDED.note;

ON CONFLICT ... DO UPDATE是幂等的关键。批量任务重跑是常态，没有这个子句，第二次跑就会主键冲突，你就得先清表再跑，清表又可能误删。model这一列一定要留，同一批数据用不同模型跑出来的结果要能区分，不然以后做效果对比无从下手。

模板复用指的是把提示词、参数、输出解析三样打包成一个函数或配置项，换任务只换配置不换代码。做上三五个任务之后你会发现，真正变的东西只有那一段样例和字段名。

5. 避坑与排查：五条踩出来的记录

下面这五条都是实际跑批量时撞上的，按现象、原因、解决写清楚。

5.1 中文列名导入后变成问号

现象：pd.read_csv读出来的列名是\ufeff订单号或者一堆问号，后面按列名取值全部 KeyError。

原因：源文件是带 BOM 的 UTF-8，或者干脆是 GBK 编码，用默认参数读的时候编码没对上，BOM 被当成了列名的一部分。

解决：先探测编码再读，utf-8-sig能吃掉 BOM，GBK 源文件显式指定encoding="gbk"。读完之后统一做一次列名清洗，去掉不可见字符再往下走。别靠肉眼看，写一行打印repr(df.columns.tolist())比什么都快。

5.2 长文档摘要后半段凭空消失

现象：一份两万字的文档，摘要出来只有前半部分的内容，后半段的观点完全没提到，但接口没有任何报错。

原因：输入超了上下文上限，被静默截断。截断不发生错误，所以最难发现。

解决：入口处加长度检查，超阈值就走分级压缩，不要赌上限。另外把每块的字符数打日志，跑批之前先跑一条最长的样本验证。这个坑我吃过一次，整整一批数据重新跑，代价很大。

5.3 批量任务跑到一半大面积超时

现象：前一百条正常，之后开始密集超时，重试也失败，但换个时间点单独跑又是好的。

原因：并发和限速参数设得太激进，触发了服务端限流，或者本地网络出口被打满。也有可能是没设超时，卡住的请求堆积把线程池占满。

解决：降并发、降 QPS，加上指数退避重试，并确保单请求有超时。判断依据是看失败时间的分布：集中在某一分钟说明是突发打满，均匀分布说明是持续超限，两种调法不一样。

5.4 让模型做算术，结论对不上

现象：让它统计表格里某列的合计或者同比，结果和 Excel 差了几百，且每次问答案还不一样。

原因：模型是在生成文本，不是在执行计算。数值一多，它倾向于给出一个“看起来合理”的数。

解决：所有计算交给 pandas 或 SQL，模型只负责解释和措辞。把算好的数字作为输入给它，明确要求“只使用给定数字，不要自己计算”。这条改完之后，数值类问题的错误率基本归零。

5.5 输出格式时好时坏，解析器天天崩

现象：同一个提示词，八十条能解析，二十条报 JSON 解析错误，错误内容五花八门，有包代码块的，有前面加一句“好的，以下是结果”的。

原因：只靠提示词约束格式，没有开结构化输出开关；或者开了但提示词里没出现 json 关键词，开关没生效。

解决：三件事一起做——开结构化输出、提示词里明确写 json、解析层加正则兜底。三件都做了还有漏网的话，把失败的原文存下来单独看，通常是某条输入里带了特殊字符把模型带偏了。

6. 进阶：把单次调用成本压下来，同时留一份可复现的验收集

跑到一定量之后，成本会变成主要矛盾。压成本有两条路，一条是把重复的前缀利用起来，一条是选对模型。前者指的是：系统提示词、样例、固定的表格表头这些东西在每次请求里都是一样的，把它们放在messages的最前面并保持逐字节一致，服务端的缓存机制才有可能命中，命中之后这部分 token 的计费会明显下来。所以别在系统提示词里写时间戳、写随机 id，那等于主动放弃缓存。

后者指的是模型分工。抽取、分类、格式转换这类任务用通用对话模型就够，快且便宜；需要多步推理、要它自己拆解问题的，才切到推理类模型。判断标准不是任务听起来难不难，而是它需不需要中间步骤——需要就换，不需要就别浪费。

真正让效果可控的是验收集。我一般固定留三十条样本，覆盖正常、边界、脏数据三类，每次改提示词或换模型，跑一遍对比。

字段	用途
sample_id	固定编号，跨版本对齐
input	原始输入，一字不改
expected	人工确认过的期望输出
actual	当次跑出来的结果
pass	是否通过，人工或规则判定
model / prompt_ver	记录版本，便于回溯