如果你平时只是偶尔问几句大模型那价格波动对你的影响其实很小真正肉疼的是把 API 接进业务流程、每天跑大量重复调用的团队。DeepSeek 的 API 一直按 token 计费网上已经有消息讨论它的价格调整幅度最大能到 350% 左右。这个数字不一定准确最终要以官方定价为准但它指向一个真问题token 消耗是累进式成本单次调用便宜流程一重复就贵跑的人越多成本越失控。这次我们来看一套不复杂但很落地的省钱思路把规则明确的重复流程交给 RPA 自动处理本地负责采集、清洗、缓存只把真正需要模型理解的部分交给 DeepSeek API。目标不是不用 AI而是让每一次 token 调用都花在刀刃上。文章会覆盖 RPA 与普通脚本的区别、DeepSeek API 的调用姿势、token 消耗怎么估算、如何用缓存和批量任务压调用量、常见报错怎么排查以及哪些流程适合自动化、哪些流程不适合。1. 核心能力速览能力项说明项目主题DeepSeek API 成本优化 RPA 重复流程自动化核心思路本地 RPA 负责采集、缓存和规则处理只把必要内容发送给 DeepSeek关键收益减少无效 token 消耗降低 API 调用频率提高重复流程执行效率适用人群使用 DeepSeek API 做业务集成的开发者、运营人员、RPA 工程师自动化程度支持定时任务、批量任务、失败重试、日志记录API 依赖调用 DeepSeek chat 类接口需要 API Key是否支持本地部署不依赖本地模型RPA 脚本本地运行模型调用走云端 API批量任务支持通过队列和缓存设计实现主要风险API 定价变化、token 限额、认证失效、采集合规性参数和价格类信息请以 DeepSeek 官方公告为准本文只讨论成本和流程优化方法。2. 为什么重复流程最烧 token先明确一个概念DeepSeek 这类大模型 API 按 token 计费输入和输出都算钱上下文越长单次成本越高。很多人的误区是只盯着单次对话的价格觉得一次几分钱无所谓但真实场景里成本是这样堆起来的第一同一批数据反复传。处理 100 条数据每条都要带系统提示词和历史上下文等于每次调用都重复支付背景信息的费用。100 条就是 100 份重复 token。第二无效输出太多。让模型做分类、提取、格式化这类规则明确的任务它可能给你回一大段解释文字这些输出 token 完全没必要。第三失败重试浪费。调用超时、认证失效、返回格式异常每重试一次就多一次计费。如果脚本没有重试上限和熔断机制token 会在报错循环里白白烧掉。第四人工操作不统一。不同人操作同一个流程提示词写法不同结果格式不同返工率高调试也烧 token。RPA 解决的是后三个问题尤其是第一点。RPA 全称 Robotic Process Automation机器人流程自动化本质是让软件模拟人工操作按固定规则处理页面、文件、接口调用。和普通 Python 脚本相比RPA 更强调流程编排、界面自动化、定时触发和日志追踪适合接入业务流程。对于 DeepSeek 场景RPA 可以在本地完成数据采集、格式整理、缓存去重只把必须要模型判断的内容打包发给 API。一次批量任务只需要一次系统提示词而不是每条数据都重复一份。3. 适用场景与使用边界3.1 适合自动化的场景数据采集后整理从网页、Excel、数据库抓取字段清洗成统一格式再由 DeepSeek 做摘要或分类。固定格式文档生成本地模板 API 返回文本自动填表、自动归档。定时执行的内容处理每天固定时间处理前一天的日志、评论、工单生成汇总报告。规则明确的信息提取从一批文本里抽日期、金额、人名、订单号这类任务不需要深度推理适合批量调用。小规模测试和回归修改提示词后用同一批本地数据反复验证输出格式减少人工复制粘贴。RPA 最适合“跑一样的流程处理不同的数据”的场景。规则越固定收益越明显。3.2 不适合的场景需要深度交互和多轮推理的任务比如复杂代码调试、头脑风暴这类场景应该让人工主导而不是 RPA 硬跑。对实时性要求极高的场景RPA 流程有调度延迟不适合在线接口直接返回。采集对象有明显反爬或反自动化限制强行自动化可能违反平台规则风险大于收益。涉及未授权数据、隐私内容、版权素材的抓取和处理不要碰。3.3 使用边界提醒所有 RPA 流程都应该在平台规则、数据授权和使用者权限范围内运行。调用 DeepSeek API 时不要尝试通过非官方渠道获取密钥也不要模仿“无限制提示词”之类对抗模型安全机制的用法。处理个人数据要脱敏处理商业数据要确认合同允许做内容生成要复核版权风险。4. 方案架构RPA 负责采集DeepSeek 负责生成整体流程分三层数据源网页 / Excel / 数据库 / 本地文件 ↓ RPA 采集与清洗 本地缓存目录raw_data / cache ↓ 批量调度 DeepSeek API单次批量调用 ↓ 结果校验 输出目录results / logsRPA 脚本在本地运行负责采集、清洗、缓存、批量调度、结果归档。DeepSeek API 只负责文本理解、分类、摘要、提取、改写等模型擅长的工作。这样做的好处是拆分清晰采集失败时本地重试不消耗 token。数据去重后重复数据不再发送。批量打包后系统提示词只写一次。结果校验失败时只重试失败项不重跑全量。RPA 工具可以选择影刀 RPA、UiBot、按键精灵这类商业软件也可以用 Python Playwright / Selenium 自建流程。商业 RPA 胜在低代码自建脚本胜在可控和易集成。本文示例以 Python 为主因为 CSDN 读者更关心可复制的代码。5. 环境准备与前置条件5.1 硬件和操作系统RPA 脚本和 API 调用对硬件要求不高普通办公电脑即可。CPU 4 核以上内存 8GB 以上磁盘至少预留 10GB 用于输入输出缓存。不需要 GPU因为所有模型推理都在 DeepSeek 云端完成。操作系统推荐 Windows 10/11 或 Ubuntu 20.04 以上。Windows 上使用影刀等商业 RPA 更顺手Linux 上更适合跑 Python 定时任务。5.2 软件依赖Python 环境建议 3.10 或更高。需要安装以下依赖pip install openai pandas requests beautifulsoup4 pypdf python-dotenv如果做网页自动化可以加装pip install playwright playwright install chromiumRPA 工具如果使用影刀去官网下载客户端创建自动化流程后把按钮逻辑和数据处理绑定到本地 Python 脚本上。影刀支持调用外部程序也支持直接写 Python 代码块。5.3 DeepSeek API Key 准备登录 DeepSeek 开放平台在 API Keys 页面创建密钥。注意密钥只在创建时完整显示一次保存好不要提交到代码仓库。建议用环境变量或 .env 文件管理。# .env 文件示例 DEEPSEEK_API_KEYsk-xxxxxxxxxxxxxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat5.4 端口和目录规划本方案不涉及 Web 服务不需要固定端口。但建议提前规划目录结构project/ ├── raw_data/ # RPA 原始采集数据 ├── cache/ # 去重后的缓存数据 ├── results/ # API 返回结果 ├── logs/ # 运行日志 ├── config.json # 批量任务配置 ├── rpa_collect.py # 采集脚本 ├── call_api.py # API 调用脚本 └── .env # API Key 配置6. 第一步RPA 脚本采集与本地缓存6.1 采集数据的常见方式RPA 采集不只有模拟点击网页这一种方式。从材料看RPA 常被用来做公众号采集、页面自动点击、Excel 数据处理等。实际开发中优先级建议是官方 API 优先。有公开接口就用接口稳定且合规。静态页面解析。requests BeautifulSoup 直接抓 HTML。半自动化。Playwright 加载页面等待渲染后提取字段。全界面模拟。模拟键盘鼠标操作这是最后手段因为容易受页面结构变化影响。6.2 通用采集脚本模板import os import json import hashlib from pathlib import Path import requests from dotenv import load_dotenv load_dotenv() RAW_DIR Path(raw_data) CACHE_DIR Path(cache) RAW_DIR.mkdir(exist_okTrue) CACHE_DIR.mkdir(exist_okTrue) def fetch_page(url: str) - str: 采集页面原始 HTML实际项目需要按目标站点调整请求头 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } resp requests.get(url, headersheaders, timeout15) resp.raise_for_status() return resp.text def save_raw(data: dict, prefix: str item) - Path: 保存原始数据文件名用内容哈希天然去重 content json.dumps(data, ensure_asciiFalse).encode(utf-8) digest hashlib.md5(content).hexdigest() filepath RAW_DIR / f{prefix}_{digest}.json if not filepath.exists(): filepath.write_text(content, encodingutf-8) return filepath def get_cache_key(text: str) - str: return hashlib.md5(text.encode(utf-8)).hexdigest()这里的关键点是先缓存后处理。采集到的每一条数据先计算哈希相同内容不会重复存储后续也不会重复调用 API。6.3 模拟网页操作的 RPA 逻辑如果目标页面没有公开接口需要用 Playwright 模拟点击和滚动import asyncio from playwright.async_api import async_playwright async def collect_dynamic_page(url: str, wait_selector: str): async with async_playwright() as p: browser await p.chromium.launch(headlessFalse) page await browser.new_page() await page.goto(url, timeout30000) await page.wait_for_selector(wait_selector, timeout10000) # 模拟滚动加载更多具体次数按页面实际情况调整 for _ in range(3): await page.mouse.wheel(0, 2000) await page.wait_for_timeout(1000) items await page.query_selector_all(.item) for item in items: text await item.inner_text() print(text) await browser.close() asyncio.run(collect_dynamic_page(https://example.com, .item))注意采集行为必须在目标网站许可范围内进行不要绕过登录、验证码或访问频率限制。7. 第二步批量调用 DeepSeek API7.1 基础调用模板DeepSeek 的 API 接口兼容 OpenAI 协议可以直接用 openai 库。import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com) ) def ask_deepseek(user_content: str, system_prompt: str 你是数据处理助手) - str: try: resp client.chat.completions.create( modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), messages[ {role: system, content: system_prompt}, {role: user, content: user_content} ], temperature0.3, max_tokens2048, streamFalse ) return resp.choices[0].message.content except Exception as e: print(fAPI 调用失败: {e}) raise先跑通这条基础链路再考虑批量。7.2 批量调用与结果落盘批量调用的核心是控制并发和错误隔离。不要一次性把所有数据都发出去建议每批 5 到 10 条逐条发送并保存单条结果这样中途失败时可以续跑。import json import time from pathlib import Path def process_batch(input_dir: Path, output_dir: Path, batch_size: int 10): output_dir.mkdir(exist_okTrue) files list(input_dir.glob(*.json)) total len(files) for i, file in enumerate(files, 1): data json.loads(file.read_text(encodingutf-8)) cache_key file.stem out_file output_dir / fresult_{cache_key}.json if out_file.exists(): print(f[{i}/{total}] 已存在跳过: {cache_key}) continue content data.get(content, ) try: result ask_deepseek(content) out_file.write_text( json.dumps({key: cache_key, result: result}, ensure_asciiFalse), encodingutf-8 ) print(f[{i}/{total}] 完成: {cache_key}) except Exception as e: print(f[{i}/{total}] 失败: {cache_key}, 错误: {e}) # 控制 QPS避免触发限流 time.sleep(0.5)这段代码有几个值得留意的设计每个结果独立文件断点续跑。已存在的输出文件直接跳过不会重复计费。每次调用前后打日志方便定位哪一条失败。sleep 控制请求频率避免短时间大量请求被限流。7.3 失败重试与结果校验批量任务不能只靠一次调用成功就觉得万事大吉。API 可能返回空内容、JSON 解析失败、格式不对这些都要在落盘前校验。def ask_deepseek_with_retry(user_content: str, max_retry: int 3): for attempt in range(1, max_retry 1): try: result ask_deepseek(user_content) if not result or len(result.strip()) 0: raise ValueError(返回内容为空) return result except Exception as e: print(f第 {attempt} 次重试错误: {e}) time.sleep(2 ** attempt) raise RuntimeError(f重试 {max_retry} 次仍失败)在批量循环中调用这个函数就能把临时网络波动和偶发空响应过滤掉。注意最大重试次数不要设太大否则 token 会在故障期间持续消耗。8. token 优化策略与批量任务8.1 先估算再发请求每次调用前先估算 token 量可以避免上下文太长导致成本失控。中文和英文的 token 换算略有不同粗略估法如下def estimate_tokens(text: str) - int: 粗略估算 token 数中文约 1 字符 0.5~1 token英文约 4 字符 1 token if not text: return 0 chinese_chars sum(1 for ch in text if \u4e00 ch \u9fff) other_chars len(text) - chinese_chars return int(chinese_chars * 0.8 other_chars / 4) 8这不是精确算法但足够用来判断单条数据是否超长、批量任务总量是否超预算。8.2 缓存策略同样的内容不要让模型处理两遍。在调用 API 前先查缓存目录CACHE_FILE Path(cache/api_cache.json) def load_cache(): if CACHE_FILE.exists(): return json.loads(CACHE_FILE.read_text(encodingutf-8)) return {} def save_cache(cache: dict): CACHE_FILE.write_text(json.dumps(cache, ensure_asciiFalse), encodingutf-8) def get_cached_result(content_hash: str): return load_cache().get(content_hash) def set_cached_result(content_hash: str, result: str): cache load_cache() cache[content_hash] result save_cache(cache)在实际批量流程里把内容哈希传入命中缓存就直接读结果不调用 API。只有缓存未命中的才算入 token 消耗。8.3 精简系统提示词和输入内容同一批数据只设置一次 system prompt不要每条拼接。用户消息里只保留必要字段不要在 prompt 里堆无关背景。如果模型只需要提取日期和金额就把输入截断到相关段落而不是整篇文章。可以用简单的截断逻辑def trim_content(text: str, max_chars: int 1500) - str: 按字符数截断输入避免一次性发送过长文本 if len(text) max_chars: return text return text[:max_chars] ……[已省略]8.4 批量任务队列设计当 RPA 采集任务和 API 调用任务需要异步跑时可以用一个简单的队列配置{ input_dir: ./raw_data, cache_dir: ./cache, output_dir: ./results, batch_size: 10, max_retry: 3, request_interval: 0.5, model: deepseek-chat, max_input_chars: 1500, enable_cache: true, log_level: INFO }RPA 定期检查 input_dir 新增文件调用 call_api.py 处理处理完的可选项是移动到 done 目录保持输入目录干净。如果使用影刀等可视化 RPA 工具可以直接在流程里加“循环”“判断”“调用 Python 脚本”三个组件把这些逻辑串起来。8.5 定时任务与调度Windows 上可以用计划任务Linux 上可以用 cron。# 每天凌晨 2 点执行采集3 点执行 API 调用 0 2 * * * cd /path/to/project python rpa_collect.py logs/cron.log 21 0 3 * * * cd /path/to/project python call_api.py logs/api.log 21定时任务的好处是把流程固定下来不用每天手动触发也方便记录成本。9. 常见问题与排查方法问题现象可能原因排查方式解决方案API 返回 401 或 403API Key 失效、地域限制、权限不足检查密钥是否过期查看平台账号状态重新生成 API Key更新 .env 配置请求超时网络波动或 prompt 过长查看日志中的耗时分段测试缩短输入文本增加超时时间返回内容为空模型输出被截断或过滤打印原始响应减小 max_tokens简化 prompt上下文超长报错输入超过模型限制调用前估算 token截断输入、分批处理批量任务卡住死循环或缺少失败重试检查日志确认是否有异常堆积增加超时和重试上限token 消耗比预期高缓存失效、重复发送相同内容统计缓存命中率检查内容哈希逻辑确保去重RPA 页面元素找不到页面结构变化查看元素选择器是否过期改用更稳定的选择器或增加等待结果格式不稳定temperature 过高或 prompt 不明确固定 prompt 模板降低 temperature设置 temperature0.2 左右增加格式示例关于材料中反复出现的 token exchange failed、sign-in could not be completed 这类认证报错多见于外部服务接入时的 OAuth 流程问题。如果是在 DeepSeek 直连场景遇到类似错误优先检查 API Key 是否有效、请求头是否正确、网络环境是否被限制。真正的优化方向不是反复刷新 token而是把凭证管理和缓存做好减少不必要的重复认证。10. 合规与安全边界10.1 API 使用合规调用 DeepSeek API 时遵守平台服务条款。不使用非官方工具绕过认证、价格或配额限制。API Key 不要上传到公开仓库不要共享给无关人员。如果密钥泄露立即在平台吊销并重建。10.2 数据采集合规RPA 采集目标网站数据时要先确认目标站点的 robots 协议和服务条款。只采集自己有权限访问的数据不绕过登录限制、验证码和访问频率控制。涉及个人信息的采集和处理要符合数据隐私要求能脱敏的尽量脱敏。10.3 内容版权与生成物复核让 DeepSeek 生成的摘要、文案、报告在对外发布前要做人工复核。不要直接公布未经验证的模型输出。涉及他人肖像、声音、作品内容时必须有明确授权。商业环境中使用模型输出要确认输入素材和输出结果都不侵犯第三方权益。10.4 RPA 操作边界RPA 自动化只用于正当业务场景不用于批量注册、刷量、恶意点击、绕过系统限制等行为。不要因为流程自动化就放松了原有人工操作时的合规审查。11. 最佳实践这里给出几条可以直接用的经验第一先小批量验证再全量跑。准备 5 到 10 条样本跑通缓存、调用、结果校验、输出归档全流程再放开到全部数据。第二保留一套最小可运行配置。把 prompt 模板、批量大小、重试次数、请求间隔全部写入 config.json不要散落在代码里。改配置时只要改一处。第三目录分离。原始数据、缓存、结果、日志分开存放不要混在一个文件夹。出了问题能快速定位哪一步失败。第四日志要有时间戳和关键词。每一条读取、每一条调用成功、每一条失败都记录下来。批量任务跑完先看日志再数输出文件数量两边对得上才算完成。第五接口服务要限定调用范围。如果后续把 RPA 流程提升为 Web 服务只监听 127.0.0.1不要直接暴露公网。需要对外提供服务时做好鉴权和白名单。第六API 请求间隔不要压得太狠。DeepSeek 对请求频率有平台限制把 sleep 时间设为 0.5 到 1 秒批量任务跑慢一点稳定性好很多。第七定期检查缓存目录。缓存文件会随着数据量增长变大设置一个清理策略比如保留最近 30 天避免磁盘被占满。12. 总结与下一步这次内容的核心不是介绍某个新框架而是给出一套量化的 token 成本控制方法。RPA 在这里承担的是“流程自动化”和“数据预处理”角色DeepSeek 承担“文本理解”角色。两者结合重复流程的 token 消耗会明显下降。最先应该验证的功能是缓存去重。把同一条数据发送两次第二次应该直接命中缓存不产生 API 调用。这一步跑通后面整个批量任务才有意义。最容易踩的坑有三个一是 RPA 采集解析不稳定页面结构一变流程就断二是批量任务没有断点续跑失败一次要重跑全量三是 prompt 没固定输出格式来回变。这三个问题在日志和配置上提前做设计都可以规避。后续可以继续扩展的方向包括把采集结果接入数据库、给批量任务加 Web 管理界面、用数据可视化统计每日 token 成本曲线、把 RPA 流程接入企业 IM 实现自动报告推送。建议把这套最小流程先落地跑一周记录每天 API 调用次数、token 消耗和失败率用数据决定下一步要不要上更重的自动化调度。省钱的核心不是不用 API而是让每一次调用都有明确产出、不重复、不浪费。
DeepSeek API成本优化：RPA自动化如何降低token消耗与重复调用
如果你平时只是偶尔问几句大模型那价格波动对你的影响其实很小真正肉疼的是把 API 接进业务流程、每天跑大量重复调用的团队。DeepSeek 的 API 一直按 token 计费网上已经有消息讨论它的价格调整幅度最大能到 350% 左右。这个数字不一定准确最终要以官方定价为准但它指向一个真问题token 消耗是累进式成本单次调用便宜流程一重复就贵跑的人越多成本越失控。这次我们来看一套不复杂但很落地的省钱思路把规则明确的重复流程交给 RPA 自动处理本地负责采集、清洗、缓存只把真正需要模型理解的部分交给 DeepSeek API。目标不是不用 AI而是让每一次 token 调用都花在刀刃上。文章会覆盖 RPA 与普通脚本的区别、DeepSeek API 的调用姿势、token 消耗怎么估算、如何用缓存和批量任务压调用量、常见报错怎么排查以及哪些流程适合自动化、哪些流程不适合。1. 核心能力速览能力项说明项目主题DeepSeek API 成本优化 RPA 重复流程自动化核心思路本地 RPA 负责采集、缓存和规则处理只把必要内容发送给 DeepSeek关键收益减少无效 token 消耗降低 API 调用频率提高重复流程执行效率适用人群使用 DeepSeek API 做业务集成的开发者、运营人员、RPA 工程师自动化程度支持定时任务、批量任务、失败重试、日志记录API 依赖调用 DeepSeek chat 类接口需要 API Key是否支持本地部署不依赖本地模型RPA 脚本本地运行模型调用走云端 API批量任务支持通过队列和缓存设计实现主要风险API 定价变化、token 限额、认证失效、采集合规性参数和价格类信息请以 DeepSeek 官方公告为准本文只讨论成本和流程优化方法。2. 为什么重复流程最烧 token先明确一个概念DeepSeek 这类大模型 API 按 token 计费输入和输出都算钱上下文越长单次成本越高。很多人的误区是只盯着单次对话的价格觉得一次几分钱无所谓但真实场景里成本是这样堆起来的第一同一批数据反复传。处理 100 条数据每条都要带系统提示词和历史上下文等于每次调用都重复支付背景信息的费用。100 条就是 100 份重复 token。第二无效输出太多。让模型做分类、提取、格式化这类规则明确的任务它可能给你回一大段解释文字这些输出 token 完全没必要。第三失败重试浪费。调用超时、认证失效、返回格式异常每重试一次就多一次计费。如果脚本没有重试上限和熔断机制token 会在报错循环里白白烧掉。第四人工操作不统一。不同人操作同一个流程提示词写法不同结果格式不同返工率高调试也烧 token。RPA 解决的是后三个问题尤其是第一点。RPA 全称 Robotic Process Automation机器人流程自动化本质是让软件模拟人工操作按固定规则处理页面、文件、接口调用。和普通 Python 脚本相比RPA 更强调流程编排、界面自动化、定时触发和日志追踪适合接入业务流程。对于 DeepSeek 场景RPA 可以在本地完成数据采集、格式整理、缓存去重只把必须要模型判断的内容打包发给 API。一次批量任务只需要一次系统提示词而不是每条数据都重复一份。3. 适用场景与使用边界3.1 适合自动化的场景数据采集后整理从网页、Excel、数据库抓取字段清洗成统一格式再由 DeepSeek 做摘要或分类。固定格式文档生成本地模板 API 返回文本自动填表、自动归档。定时执行的内容处理每天固定时间处理前一天的日志、评论、工单生成汇总报告。规则明确的信息提取从一批文本里抽日期、金额、人名、订单号这类任务不需要深度推理适合批量调用。小规模测试和回归修改提示词后用同一批本地数据反复验证输出格式减少人工复制粘贴。RPA 最适合“跑一样的流程处理不同的数据”的场景。规则越固定收益越明显。3.2 不适合的场景需要深度交互和多轮推理的任务比如复杂代码调试、头脑风暴这类场景应该让人工主导而不是 RPA 硬跑。对实时性要求极高的场景RPA 流程有调度延迟不适合在线接口直接返回。采集对象有明显反爬或反自动化限制强行自动化可能违反平台规则风险大于收益。涉及未授权数据、隐私内容、版权素材的抓取和处理不要碰。3.3 使用边界提醒所有 RPA 流程都应该在平台规则、数据授权和使用者权限范围内运行。调用 DeepSeek API 时不要尝试通过非官方渠道获取密钥也不要模仿“无限制提示词”之类对抗模型安全机制的用法。处理个人数据要脱敏处理商业数据要确认合同允许做内容生成要复核版权风险。4. 方案架构RPA 负责采集DeepSeek 负责生成整体流程分三层数据源网页 / Excel / 数据库 / 本地文件 ↓ RPA 采集与清洗 本地缓存目录raw_data / cache ↓ 批量调度 DeepSeek API单次批量调用 ↓ 结果校验 输出目录results / logsRPA 脚本在本地运行负责采集、清洗、缓存、批量调度、结果归档。DeepSeek API 只负责文本理解、分类、摘要、提取、改写等模型擅长的工作。这样做的好处是拆分清晰采集失败时本地重试不消耗 token。数据去重后重复数据不再发送。批量打包后系统提示词只写一次。结果校验失败时只重试失败项不重跑全量。RPA 工具可以选择影刀 RPA、UiBot、按键精灵这类商业软件也可以用 Python Playwright / Selenium 自建流程。商业 RPA 胜在低代码自建脚本胜在可控和易集成。本文示例以 Python 为主因为 CSDN 读者更关心可复制的代码。5. 环境准备与前置条件5.1 硬件和操作系统RPA 脚本和 API 调用对硬件要求不高普通办公电脑即可。CPU 4 核以上内存 8GB 以上磁盘至少预留 10GB 用于输入输出缓存。不需要 GPU因为所有模型推理都在 DeepSeek 云端完成。操作系统推荐 Windows 10/11 或 Ubuntu 20.04 以上。Windows 上使用影刀等商业 RPA 更顺手Linux 上更适合跑 Python 定时任务。5.2 软件依赖Python 环境建议 3.10 或更高。需要安装以下依赖pip install openai pandas requests beautifulsoup4 pypdf python-dotenv如果做网页自动化可以加装pip install playwright playwright install chromiumRPA 工具如果使用影刀去官网下载客户端创建自动化流程后把按钮逻辑和数据处理绑定到本地 Python 脚本上。影刀支持调用外部程序也支持直接写 Python 代码块。5.3 DeepSeek API Key 准备登录 DeepSeek 开放平台在 API Keys 页面创建密钥。注意密钥只在创建时完整显示一次保存好不要提交到代码仓库。建议用环境变量或 .env 文件管理。# .env 文件示例 DEEPSEEK_API_KEYsk-xxxxxxxxxxxxxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat5.4 端口和目录规划本方案不涉及 Web 服务不需要固定端口。但建议提前规划目录结构project/ ├── raw_data/ # RPA 原始采集数据 ├── cache/ # 去重后的缓存数据 ├── results/ # API 返回结果 ├── logs/ # 运行日志 ├── config.json # 批量任务配置 ├── rpa_collect.py # 采集脚本 ├── call_api.py # API 调用脚本 └── .env # API Key 配置6. 第一步RPA 脚本采集与本地缓存6.1 采集数据的常见方式RPA 采集不只有模拟点击网页这一种方式。从材料看RPA 常被用来做公众号采集、页面自动点击、Excel 数据处理等。实际开发中优先级建议是官方 API 优先。有公开接口就用接口稳定且合规。静态页面解析。requests BeautifulSoup 直接抓 HTML。半自动化。Playwright 加载页面等待渲染后提取字段。全界面模拟。模拟键盘鼠标操作这是最后手段因为容易受页面结构变化影响。6.2 通用采集脚本模板import os import json import hashlib from pathlib import Path import requests from dotenv import load_dotenv load_dotenv() RAW_DIR Path(raw_data) CACHE_DIR Path(cache) RAW_DIR.mkdir(exist_okTrue) CACHE_DIR.mkdir(exist_okTrue) def fetch_page(url: str) - str: 采集页面原始 HTML实际项目需要按目标站点调整请求头 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } resp requests.get(url, headersheaders, timeout15) resp.raise_for_status() return resp.text def save_raw(data: dict, prefix: str item) - Path: 保存原始数据文件名用内容哈希天然去重 content json.dumps(data, ensure_asciiFalse).encode(utf-8) digest hashlib.md5(content).hexdigest() filepath RAW_DIR / f{prefix}_{digest}.json if not filepath.exists(): filepath.write_text(content, encodingutf-8) return filepath def get_cache_key(text: str) - str: return hashlib.md5(text.encode(utf-8)).hexdigest()这里的关键点是先缓存后处理。采集到的每一条数据先计算哈希相同内容不会重复存储后续也不会重复调用 API。6.3 模拟网页操作的 RPA 逻辑如果目标页面没有公开接口需要用 Playwright 模拟点击和滚动import asyncio from playwright.async_api import async_playwright async def collect_dynamic_page(url: str, wait_selector: str): async with async_playwright() as p: browser await p.chromium.launch(headlessFalse) page await browser.new_page() await page.goto(url, timeout30000) await page.wait_for_selector(wait_selector, timeout10000) # 模拟滚动加载更多具体次数按页面实际情况调整 for _ in range(3): await page.mouse.wheel(0, 2000) await page.wait_for_timeout(1000) items await page.query_selector_all(.item) for item in items: text await item.inner_text() print(text) await browser.close() asyncio.run(collect_dynamic_page(https://example.com, .item))注意采集行为必须在目标网站许可范围内进行不要绕过登录、验证码或访问频率限制。7. 第二步批量调用 DeepSeek API7.1 基础调用模板DeepSeek 的 API 接口兼容 OpenAI 协议可以直接用 openai 库。import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com) ) def ask_deepseek(user_content: str, system_prompt: str 你是数据处理助手) - str: try: resp client.chat.completions.create( modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), messages[ {role: system, content: system_prompt}, {role: user, content: user_content} ], temperature0.3, max_tokens2048, streamFalse ) return resp.choices[0].message.content except Exception as e: print(fAPI 调用失败: {e}) raise先跑通这条基础链路再考虑批量。7.2 批量调用与结果落盘批量调用的核心是控制并发和错误隔离。不要一次性把所有数据都发出去建议每批 5 到 10 条逐条发送并保存单条结果这样中途失败时可以续跑。import json import time from pathlib import Path def process_batch(input_dir: Path, output_dir: Path, batch_size: int 10): output_dir.mkdir(exist_okTrue) files list(input_dir.glob(*.json)) total len(files) for i, file in enumerate(files, 1): data json.loads(file.read_text(encodingutf-8)) cache_key file.stem out_file output_dir / fresult_{cache_key}.json if out_file.exists(): print(f[{i}/{total}] 已存在跳过: {cache_key}) continue content data.get(content, ) try: result ask_deepseek(content) out_file.write_text( json.dumps({key: cache_key, result: result}, ensure_asciiFalse), encodingutf-8 ) print(f[{i}/{total}] 完成: {cache_key}) except Exception as e: print(f[{i}/{total}] 失败: {cache_key}, 错误: {e}) # 控制 QPS避免触发限流 time.sleep(0.5)这段代码有几个值得留意的设计每个结果独立文件断点续跑。已存在的输出文件直接跳过不会重复计费。每次调用前后打日志方便定位哪一条失败。sleep 控制请求频率避免短时间大量请求被限流。7.3 失败重试与结果校验批量任务不能只靠一次调用成功就觉得万事大吉。API 可能返回空内容、JSON 解析失败、格式不对这些都要在落盘前校验。def ask_deepseek_with_retry(user_content: str, max_retry: int 3): for attempt in range(1, max_retry 1): try: result ask_deepseek(user_content) if not result or len(result.strip()) 0: raise ValueError(返回内容为空) return result except Exception as e: print(f第 {attempt} 次重试错误: {e}) time.sleep(2 ** attempt) raise RuntimeError(f重试 {max_retry} 次仍失败)在批量循环中调用这个函数就能把临时网络波动和偶发空响应过滤掉。注意最大重试次数不要设太大否则 token 会在故障期间持续消耗。8. token 优化策略与批量任务8.1 先估算再发请求每次调用前先估算 token 量可以避免上下文太长导致成本失控。中文和英文的 token 换算略有不同粗略估法如下def estimate_tokens(text: str) - int: 粗略估算 token 数中文约 1 字符 0.5~1 token英文约 4 字符 1 token if not text: return 0 chinese_chars sum(1 for ch in text if \u4e00 ch \u9fff) other_chars len(text) - chinese_chars return int(chinese_chars * 0.8 other_chars / 4) 8这不是精确算法但足够用来判断单条数据是否超长、批量任务总量是否超预算。8.2 缓存策略同样的内容不要让模型处理两遍。在调用 API 前先查缓存目录CACHE_FILE Path(cache/api_cache.json) def load_cache(): if CACHE_FILE.exists(): return json.loads(CACHE_FILE.read_text(encodingutf-8)) return {} def save_cache(cache: dict): CACHE_FILE.write_text(json.dumps(cache, ensure_asciiFalse), encodingutf-8) def get_cached_result(content_hash: str): return load_cache().get(content_hash) def set_cached_result(content_hash: str, result: str): cache load_cache() cache[content_hash] result save_cache(cache)在实际批量流程里把内容哈希传入命中缓存就直接读结果不调用 API。只有缓存未命中的才算入 token 消耗。8.3 精简系统提示词和输入内容同一批数据只设置一次 system prompt不要每条拼接。用户消息里只保留必要字段不要在 prompt 里堆无关背景。如果模型只需要提取日期和金额就把输入截断到相关段落而不是整篇文章。可以用简单的截断逻辑def trim_content(text: str, max_chars: int 1500) - str: 按字符数截断输入避免一次性发送过长文本 if len(text) max_chars: return text return text[:max_chars] ……[已省略]8.4 批量任务队列设计当 RPA 采集任务和 API 调用任务需要异步跑时可以用一个简单的队列配置{ input_dir: ./raw_data, cache_dir: ./cache, output_dir: ./results, batch_size: 10, max_retry: 3, request_interval: 0.5, model: deepseek-chat, max_input_chars: 1500, enable_cache: true, log_level: INFO }RPA 定期检查 input_dir 新增文件调用 call_api.py 处理处理完的可选项是移动到 done 目录保持输入目录干净。如果使用影刀等可视化 RPA 工具可以直接在流程里加“循环”“判断”“调用 Python 脚本”三个组件把这些逻辑串起来。8.5 定时任务与调度Windows 上可以用计划任务Linux 上可以用 cron。# 每天凌晨 2 点执行采集3 点执行 API 调用 0 2 * * * cd /path/to/project python rpa_collect.py logs/cron.log 21 0 3 * * * cd /path/to/project python call_api.py logs/api.log 21定时任务的好处是把流程固定下来不用每天手动触发也方便记录成本。9. 常见问题与排查方法问题现象可能原因排查方式解决方案API 返回 401 或 403API Key 失效、地域限制、权限不足检查密钥是否过期查看平台账号状态重新生成 API Key更新 .env 配置请求超时网络波动或 prompt 过长查看日志中的耗时分段测试缩短输入文本增加超时时间返回内容为空模型输出被截断或过滤打印原始响应减小 max_tokens简化 prompt上下文超长报错输入超过模型限制调用前估算 token截断输入、分批处理批量任务卡住死循环或缺少失败重试检查日志确认是否有异常堆积增加超时和重试上限token 消耗比预期高缓存失效、重复发送相同内容统计缓存命中率检查内容哈希逻辑确保去重RPA 页面元素找不到页面结构变化查看元素选择器是否过期改用更稳定的选择器或增加等待结果格式不稳定temperature 过高或 prompt 不明确固定 prompt 模板降低 temperature设置 temperature0.2 左右增加格式示例关于材料中反复出现的 token exchange failed、sign-in could not be completed 这类认证报错多见于外部服务接入时的 OAuth 流程问题。如果是在 DeepSeek 直连场景遇到类似错误优先检查 API Key 是否有效、请求头是否正确、网络环境是否被限制。真正的优化方向不是反复刷新 token而是把凭证管理和缓存做好减少不必要的重复认证。10. 合规与安全边界10.1 API 使用合规调用 DeepSeek API 时遵守平台服务条款。不使用非官方工具绕过认证、价格或配额限制。API Key 不要上传到公开仓库不要共享给无关人员。如果密钥泄露立即在平台吊销并重建。10.2 数据采集合规RPA 采集目标网站数据时要先确认目标站点的 robots 协议和服务条款。只采集自己有权限访问的数据不绕过登录限制、验证码和访问频率控制。涉及个人信息的采集和处理要符合数据隐私要求能脱敏的尽量脱敏。10.3 内容版权与生成物复核让 DeepSeek 生成的摘要、文案、报告在对外发布前要做人工复核。不要直接公布未经验证的模型输出。涉及他人肖像、声音、作品内容时必须有明确授权。商业环境中使用模型输出要确认输入素材和输出结果都不侵犯第三方权益。10.4 RPA 操作边界RPA 自动化只用于正当业务场景不用于批量注册、刷量、恶意点击、绕过系统限制等行为。不要因为流程自动化就放松了原有人工操作时的合规审查。11. 最佳实践这里给出几条可以直接用的经验第一先小批量验证再全量跑。准备 5 到 10 条样本跑通缓存、调用、结果校验、输出归档全流程再放开到全部数据。第二保留一套最小可运行配置。把 prompt 模板、批量大小、重试次数、请求间隔全部写入 config.json不要散落在代码里。改配置时只要改一处。第三目录分离。原始数据、缓存、结果、日志分开存放不要混在一个文件夹。出了问题能快速定位哪一步失败。第四日志要有时间戳和关键词。每一条读取、每一条调用成功、每一条失败都记录下来。批量任务跑完先看日志再数输出文件数量两边对得上才算完成。第五接口服务要限定调用范围。如果后续把 RPA 流程提升为 Web 服务只监听 127.0.0.1不要直接暴露公网。需要对外提供服务时做好鉴权和白名单。第六API 请求间隔不要压得太狠。DeepSeek 对请求频率有平台限制把 sleep 时间设为 0.5 到 1 秒批量任务跑慢一点稳定性好很多。第七定期检查缓存目录。缓存文件会随着数据量增长变大设置一个清理策略比如保留最近 30 天避免磁盘被占满。12. 总结与下一步这次内容的核心不是介绍某个新框架而是给出一套量化的 token 成本控制方法。RPA 在这里承担的是“流程自动化”和“数据预处理”角色DeepSeek 承担“文本理解”角色。两者结合重复流程的 token 消耗会明显下降。最先应该验证的功能是缓存去重。把同一条数据发送两次第二次应该直接命中缓存不产生 API 调用。这一步跑通后面整个批量任务才有意义。最容易踩的坑有三个一是 RPA 采集解析不稳定页面结构一变流程就断二是批量任务没有断点续跑失败一次要重跑全量三是 prompt 没固定输出格式来回变。这三个问题在日志和配置上提前做设计都可以规避。后续可以继续扩展的方向包括把采集结果接入数据库、给批量任务加 Web 管理界面、用数据可视化统计每日 token 成本曲线、把 RPA 流程接入企业 IM 实现自动报告推送。建议把这套最小流程先落地跑一周记录每天 API 调用次数、token 消耗和失败率用数据决定下一步要不要上更重的自动化调度。省钱的核心不是不用 API而是让每一次调用都有明确产出、不重复、不浪费。
