资讯详情

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

📅 2026/10/28 10:41:43 | 华诺云谱 👁 阅读
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_KEYsk-你的密钥 export DEEPSEEK_BASE_URLhttps://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: fBearer {KEY}} def chat(messages, modeldeepseek-chat, temperature0.3, max_tokens1024, timeout60, retries3): 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, headersHEADERS, jsonpayload, timeouttimeout) 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 ftimeout: {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 直接抛异常因为参数写错了重试一百次还是错。参数里timeout60是单次请求上限长文本任务可以放到 120retries3是重试次数批量场景下不建议超过 5否则一条坏数据会拖慢整个队列。temperature0.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, encodingutf-8, dtypestr, keep_default_naFalse) # 只保留和当前问题相关的列列越多 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], errorscoerce) df df.fillna({city: 未知, status: 未知}) print(df.shape) print(df.dtypes.to_dict())dtypestr让所有列先按字符串进来避免order_id这种带前导零的编号被自动转成数字后丢掉零keep_default_naFalse防止NA、null这类字面值被误判成缺失errorscoerce把转不了的值变成 NaN后面统一填充比让异常直接抛出来更可控。裁剪之后先打印行列数和类型这两个数字对不上的时候问题基本都在读取阶段不用往下走。3.2 长表切块表头复读与行窗口行数一多就必须切。模型有上下文上限超了要么报错要么被截断截断是更可怕的那种因为它不报错。切块的核心不是简单对半砍而是每一块都要带表头否则模型看到一堆没有列名的值只能靠猜。def chunk_rows(df, window200, overlap20, max_chars6000): header | .join(df.columns) lines df.astype(str).agg( | .join, axis1).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 chunkswindow200是每块的行数overlap20是相邻块的重叠行用来兜住跨越边界的记录——比如一条订单的备注被拆在两块里重叠段能让至少一块看到完整内容。max_chars6000是兜底红线中文字符和 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, modeldeepseek-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, headersHEADERS, jsonpayload, timeout90) 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 Noneresponse_format是结构化输出的开关temperature0.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, encodingutf-8) as f: for line in f: done.add(json.loads(line)[id]) return done def run(rows, out_pathresult.jsonl, workers4, qps3): 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, encodingutf-8) as f, \ ThreadPoolExecutor(max_workersworkers) 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_asciiFalse) \n) f.flush() # 立刻落盘进程被杀也不丢workers4是并发线程数qps3是客户端限速两个参数要一起调并发高但限速低线程大部分时间在 sleep限速高但并发低吞吐上不去。我的经验是先从workers4, qps3起步看返回里有没有 429有就往下降没有就慢慢往上加。f.flush()这行别省它是断点续跑能成立的前提没有它缓冲区的数据在进程异常退出时会丢。workers和qps的关系值得再说一句它们不是同一个东西。workers控制同时在飞的请求数qps控制每秒发起的新请求数。前者太高会瞬间打满后者太高会持续打满两种都会触发限流但表现不一样前者是突发性的 429 峰值后者是持续性的。排查的时候看 429 是集中在开头还是均匀分布就能判断该调哪个。4.3 长文摘要的分级压缩超过上下文的长文一次喂进去必然截断。常见做法是分级压缩先按段落切成小块每块单独出摘要再把这些摘要拼起来做二次摘要。这叫 map-reduce落地很简单。def summarize_long(text, seg3000, rounds2): # 第一级按字符切段段间留一点重叠避免句子被切断 parts [text[i:i seg] for i in range(0, len(text), seg - 200)] summaries [chat([{role: user, content: f用 100 字概括这段内容保留数字和人名\n{p}}], temperature0.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}}], temperature0.3) return mergedseg3000是每段字符数seg - 200的重叠是为了不让一句话被切成两半导致语义断裂。rounds2是最大压缩轮数一般两轮足够把一份长文档压到能一次处理。这个方法的代价是 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能吃掉 BOMGBK 源文件显式指定encodinggbk。读完之后统一做一次列名清洗去掉不可见字符再往下走。别靠肉眼看写一行打印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记录版本便于回溯这套表的价值在改版的时候体现得最明显。没有它你改完提示词只能凭感觉说“好像好了一点”有了它通过率从 24/30 掉到 21/30 这种变化是藏不住的。从那以后我每次动提示词哪怕只改一个词都强制走一遍这三十条验收跑完看通过率再决定要不要合并。这个习惯救过我不少次尤其是那种“只加了一句约束、结果把另一类样本带崩”的情况。这份资源里提到的批量处理和自定义模板按上面的路子接起来就是一条能长期跑的流水线脚本和验收表我整理在一起了。希望帮到你。本文还有配套的精品资源点击获取
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑