尧图网页设计 
1. 为什么批量总结 PDF 这件事卡住的大多是工程细节通义千问 Qwen-Long 是阿里云通义千问系列里专门针对超长上下文场景的模型支持中文、英文等多语言输入最长可处理 1000 万 tokens 的上下文换算下来大约是 1500 万字或 1.5 万页文档。它配套的文档服务能直接解析 pdf、docx、md、epub、mobi 等格式也就是说你不需要自己写 PDF 抽取逻辑把文件传上去模型就能基于文件内容对话。这个能力放在办公自动化场景里非常实用一堆行业报告、产品手册、合同附件、论文合集过去要人工翻半天现在可以写个 Python 脚本批量跑。但真正动手写脚本的人会发现卡住的地方往往不是模型本身而是几个工程细节PDF 上传后首轮解析慢、接口有 100 QPM 的限流、长文档返回可能触发 ResponseTimeout、文件用完要记得删否则占配额、结果要落盘成可读的 docx。这篇就聚焦用 Python 调用 Qwen-Long 批量总结本地 PDF给出可复制的 config.toml 骨架、PDF 解析与分块脚本、批量请求与结果落盘示例并演示一次完整跑通和摘要质量校验。适合有基础 Python 能力、想把手头文档处理流程自动化的办公场景开发者。2. TaoToken 前置统一 Key 与 API 通道怎么配在写业务脚本之前先把调用通道理顺。我习惯把模型调用的 Key 和 base_url 抽到一个 config.toml 里业务代码只读配置不硬编码密钥。这样换通道、换模型、加限流参数都不用改脚本。TaoToken 提供统一的 API 通道官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它的作用是让你用一个 Key 走通多家模型的兼容接口Qwen-Long 这种走 OpenAI 兼容协议的场景可以直接对接。你需要在控制台创建一个 API Key然后把它填进下面的配置。先建一个项目目录结构建议这样pdf_summarizer/ ├── config.toml ├── main.py ├── requirements.txt └── output/config.toml骨架如下把 Key 换成你自己的[api] # TaoToken 统一通道兼容 OpenAI 协议 base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model qwen-long [paths] pdf_dir D:/ABooks output_dir D:/ABooks/output [limits] # Qwen-Long 限流每分钟不超过 100 次调用 qpm 100 max_retries 3 retry_wait_seconds 20 request_timeout 300 [prompt] system You are a helpful assistant. user 请总结这份文档每个章节的核心内容用中文输出结构为章节标题 3到5条要点。依赖装这几个就够pip install openai python-docx tomliPython 3.11 以上自带tomllib3.10 及以下用tomli读配置。下面代码里我做了兼容处理。注意Key 不要提交到 Git建议把config.toml加进.gitignore团队协作时用环境变量覆盖。3. 可复制配置PDF 解析、分块与批量请求脚本Qwen-Long 的文档服务支持直接上传 PDF单文件限制 150M总量 1 万个文件、总大小 100G。上传后返回一个 file id在 messages 里用fileid://引用即可。所以严格来说你不需要自己解析 PDF 文本但有两个场景需要分块一是文件超过 150M 传不上去二是返回内容太长触发超时。下面脚本把上传、请求、重试、落盘、删除串起来。import os import time import sys from pathlib import Path try: import tomllib except ModuleNotFoundError: import tomli as tomllib from openai import OpenAI from docx import Document # ---------- 读配置 ---------- with open(config.toml, rb) as f: cfg tomllib.load(f) api_cfg cfg[api] path_cfg cfg[paths] limit_cfg cfg[limits] prompt_cfg cfg[prompt] client OpenAI( api_keyapi_cfg[api_key], base_urlapi_cfg[base_url], ) os.makedirs(path_cfg[output_dir], exist_okTrue) # ---------- 限流器简单令牌桶 ---------- class RateLimiter: def __init__(self, qpm): self.interval 60.0 / qpm self.last 0.0 def wait(self): now time.time() delta now - self.last if delta self.interval: time.sleep(self.interval - delta) self.last time.time() limiter RateLimiter(limit_cfg[qpm]) # ---------- 带重试的请求 ---------- def chat_with_retry(file_id, max_retries, wait_seconds): messages [ {role: system, content: prompt_cfg[system]}, {role: system, content: ffileid://{file_id}}, {role: user, content: prompt_cfg[user]}, ] for attempt in range(1, max_retries 1): try: limiter.wait() resp client.chat.completions.create( modelapi_cfg[model], messagesmessages, streamFalse, timeoutlimit_cfg[request_timeout], ) return resp.choices[0].message.content except Exception as e: print(f 第 {attempt} 次请求失败{e}) if attempt max_retries: raise time.sleep(wait_seconds) return None # ---------- 单文件处理 ---------- def process_pdf(pdf_path: Path): print(f处理文件{pdf_path.name}) file_object None try: file_object client.files.create( filePath(pdf_path), purposefile-extract, ) print(f 上传成功file id {file_object.id}) summary chat_with_retry( file_object.id, limit_cfg[max_retries], limit_cfg[retry_wait_seconds], ) doc Document() doc.add_heading(pdf_path.stem, level1) doc.add_paragraph(summary or 无返回内容) out_path Path(path_cfg[output_dir]) / f{pdf_path.stem}.docx doc.save(out_path) print(f 摘要已保存{out_path}) return True except Exception as e: print(f 处理失败跳过{e}) return False finally: if file_object is not None: try: client.files.delete(file_object.id) print(f 已删除远端文件{file_object.id}) except Exception as e: print(f 删除文件失败{e}) # ---------- 主流程 ---------- def main(): pdf_dir Path(path_cfg[pdf_dir]) pdfs sorted(pdf_dir.glob(*.pdf)) print(f共发现 {len(pdfs)} 个 PDF 文件) ok, fail 0, 0 for pdf in pdfs: if process_pdf(pdf): ok 1 else: fail 1 print(f完成成功 {ok} 个失败 {fail} 个) if __name__ __main__: main()这段脚本的关键点RateLimiter按 QPM 控制调用间隔避免撞限流chat_with_retry捕获异常后等待重试专门对付ResponseTimeoutfinally里删远端文件防止配额被占满。上传后首轮解析可能较慢request_timeout给到 300 秒比较稳。如果你的 PDF 超过 150M需要先本地拆分再上传。可以用pypdf按页切from pypdf import PdfReader, PdfWriter def split_pdf(src, dst_dir, pages_per_chunk200): reader PdfReader(src) total len(reader.pages) for i in range(0, total, pages_per_chunk): writer PdfWriter() for p in reader.pages[i:i pages_per_chunk]: writer.add_page(p) out Path(dst_dir) / f{Path(src).stem}_part{i // pages_per_chunk 1}.pdf with open(out, wb) as f: writer.write(f) print(f 拆分生成{out})拆分后每个分片单独总结最后在 docx 里按顺序拼接即可。4. 验证请求跑通一次并校验摘要质量配置和脚本就绪后先拿一个 PDF 试跑别一上来就全量。把pdf_dir指向只放一个文件的测试目录执行python main.py正常输出类似共发现 1 个 PDF 文件 处理文件2024行业报告.pdf 上传成功file id file-fe-xxxx 摘要已保存D:/ABooks/output/2024行业报告.docx 已删除远端文件file-fe-xxxx 完成成功 1 个失败 0 个打开生成的 docx检查摘要质量。我一般看三点章节结构是否完整、要点是否覆盖原文核心、有没有明显幻觉。如果发现摘要太笼统调整config.toml里的 user prompt比如要求「每个章节至少 3 条要点标注原文页码」。如果返回被截断说明输出长度接近上限可以把 prompt 改成「先输出前 5 章等我回复继续再输出后续」做分段续写。想快速验证模型对某个文件的理解是否到位可以先用模型对话入口手动传一次文件问几个问题确认解析正常再跑批量。模型对话入口在 https://taotoken.net/api-keys 旁边的对话页创建 Key 后即可使用。5. 本篇常见错排查ResponseTimeout 报错长文档首轮解析慢或输出太长。对策是加大request_timeout并在chat_with_retry里保留重试。如果反复超时把 prompt 拆成「先总结前半部分」再续。400 报错 file id 无效上传后立刻请求有时会撞上解析未完成。可以在上传后time.sleep(5)再发请求或捕获错误后重试。限流失败Qwen-Long 限流是 100 QPM脚本里RateLimiter的 interval 要按 60/1000.6 秒设置。如果你并发跑多个进程总调用量会叠加建议单进程串行。文件删除失败网络抖动或 file id 已失效。删除失败不影响主流程但会占配额建议定期在控制台清理。docx 中文乱码python-docx默认字体对中文支持一般可以在保存前设置样式字体为「微软雅黑」。Key 读不到确认config.toml路径正确Python 3.10 以下装了tomli。用环境变量覆盖时读取逻辑要加os.environ.get兜底。6. 接入方式与后续扩展批量总结跑通后常见的扩展方向有三个一是把结果汇总成一个总表用 pandas 读所有 docx 提取要点二是接入定时任务每天扫一次新入库的 PDF三是把摘要写回知识库或工单系统。这些都不需要改模型调用层只要在process_pdf后面加钩子。如果你要长期跑编码类或 Agent 类任务可以考虑 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入文档和 API Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 和 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。实际用下来把 Key 和 base_url 抽到配置、把重试和限流写进封装是这类批量任务最省心的做法后面换模型或加并发都不用动业务代码。 AI办公自动化：用通义千问Qwen-Long批量总结PDF长文档内容
1. 为什么批量总结 PDF 这件事卡住的大多是工程细节通义千问 Qwen-Long 是阿里云通义千问系列里专门针对超长上下文场景的模型支持中文、英文等多语言输入最长可处理 1000 万 tokens 的上下文换算下来大约是 1500 万字或 1.5 万页文档。它配套的文档服务能直接解析 pdf、docx、md、epub、mobi 等格式也就是说你不需要自己写 PDF 抽取逻辑把文件传上去模型就能基于文件内容对话。这个能力放在办公自动化场景里非常实用一堆行业报告、产品手册、合同附件、论文合集过去要人工翻半天现在可以写个 Python 脚本批量跑。但真正动手写脚本的人会发现卡住的地方往往不是模型本身而是几个工程细节PDF 上传后首轮解析慢、接口有 100 QPM 的限流、长文档返回可能触发 ResponseTimeout、文件用完要记得删否则占配额、结果要落盘成可读的 docx。这篇就聚焦用 Python 调用 Qwen-Long 批量总结本地 PDF给出可复制的 config.toml 骨架、PDF 解析与分块脚本、批量请求与结果落盘示例并演示一次完整跑通和摘要质量校验。适合有基础 Python 能力、想把手头文档处理流程自动化的办公场景开发者。2. TaoToken 前置统一 Key 与 API 通道怎么配在写业务脚本之前先把调用通道理顺。我习惯把模型调用的 Key 和 base_url 抽到一个 config.toml 里业务代码只读配置不硬编码密钥。这样换通道、换模型、加限流参数都不用改脚本。TaoToken 提供统一的 API 通道官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它的作用是让你用一个 Key 走通多家模型的兼容接口Qwen-Long 这种走 OpenAI 兼容协议的场景可以直接对接。你需要在控制台创建一个 API Key然后把它填进下面的配置。先建一个项目目录结构建议这样pdf_summarizer/ ├── config.toml ├── main.py ├── requirements.txt └── output/config.toml骨架如下把 Key 换成你自己的[api] # TaoToken 统一通道兼容 OpenAI 协议 base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model qwen-long [paths] pdf_dir D:/ABooks output_dir D:/ABooks/output [limits] # Qwen-Long 限流每分钟不超过 100 次调用 qpm 100 max_retries 3 retry_wait_seconds 20 request_timeout 300 [prompt] system You are a helpful assistant. user 请总结这份文档每个章节的核心内容用中文输出结构为章节标题 3到5条要点。依赖装这几个就够pip install openai python-docx tomliPython 3.11 以上自带tomllib3.10 及以下用tomli读配置。下面代码里我做了兼容处理。注意Key 不要提交到 Git建议把config.toml加进.gitignore团队协作时用环境变量覆盖。3. 可复制配置PDF 解析、分块与批量请求脚本Qwen-Long 的文档服务支持直接上传 PDF单文件限制 150M总量 1 万个文件、总大小 100G。上传后返回一个 file id在 messages 里用fileid://引用即可。所以严格来说你不需要自己解析 PDF 文本但有两个场景需要分块一是文件超过 150M 传不上去二是返回内容太长触发超时。下面脚本把上传、请求、重试、落盘、删除串起来。import os import time import sys from pathlib import Path try: import tomllib except ModuleNotFoundError: import tomli as tomllib from openai import OpenAI from docx import Document # ---------- 读配置 ---------- with open(config.toml, rb) as f: cfg tomllib.load(f) api_cfg cfg[api] path_cfg cfg[paths] limit_cfg cfg[limits] prompt_cfg cfg[prompt] client OpenAI( api_keyapi_cfg[api_key], base_urlapi_cfg[base_url], ) os.makedirs(path_cfg[output_dir], exist_okTrue) # ---------- 限流器简单令牌桶 ---------- class RateLimiter: def __init__(self, qpm): self.interval 60.0 / qpm self.last 0.0 def wait(self): now time.time() delta now - self.last if delta self.interval: time.sleep(self.interval - delta) self.last time.time() limiter RateLimiter(limit_cfg[qpm]) # ---------- 带重试的请求 ---------- def chat_with_retry(file_id, max_retries, wait_seconds): messages [ {role: system, content: prompt_cfg[system]}, {role: system, content: ffileid://{file_id}}, {role: user, content: prompt_cfg[user]}, ] for attempt in range(1, max_retries 1): try: limiter.wait() resp client.chat.completions.create( modelapi_cfg[model], messagesmessages, streamFalse, timeoutlimit_cfg[request_timeout], ) return resp.choices[0].message.content except Exception as e: print(f 第 {attempt} 次请求失败{e}) if attempt max_retries: raise time.sleep(wait_seconds) return None # ---------- 单文件处理 ---------- def process_pdf(pdf_path: Path): print(f处理文件{pdf_path.name}) file_object None try: file_object client.files.create( filePath(pdf_path), purposefile-extract, ) print(f 上传成功file id {file_object.id}) summary chat_with_retry( file_object.id, limit_cfg[max_retries], limit_cfg[retry_wait_seconds], ) doc Document() doc.add_heading(pdf_path.stem, level1) doc.add_paragraph(summary or 无返回内容) out_path Path(path_cfg[output_dir]) / f{pdf_path.stem}.docx doc.save(out_path) print(f 摘要已保存{out_path}) return True except Exception as e: print(f 处理失败跳过{e}) return False finally: if file_object is not None: try: client.files.delete(file_object.id) print(f 已删除远端文件{file_object.id}) except Exception as e: print(f 删除文件失败{e}) # ---------- 主流程 ---------- def main(): pdf_dir Path(path_cfg[pdf_dir]) pdfs sorted(pdf_dir.glob(*.pdf)) print(f共发现 {len(pdfs)} 个 PDF 文件) ok, fail 0, 0 for pdf in pdfs: if process_pdf(pdf): ok 1 else: fail 1 print(f完成成功 {ok} 个失败 {fail} 个) if __name__ __main__: main()这段脚本的关键点RateLimiter按 QPM 控制调用间隔避免撞限流chat_with_retry捕获异常后等待重试专门对付ResponseTimeoutfinally里删远端文件防止配额被占满。上传后首轮解析可能较慢request_timeout给到 300 秒比较稳。如果你的 PDF 超过 150M需要先本地拆分再上传。可以用pypdf按页切from pypdf import PdfReader, PdfWriter def split_pdf(src, dst_dir, pages_per_chunk200): reader PdfReader(src) total len(reader.pages) for i in range(0, total, pages_per_chunk): writer PdfWriter() for p in reader.pages[i:i pages_per_chunk]: writer.add_page(p) out Path(dst_dir) / f{Path(src).stem}_part{i // pages_per_chunk 1}.pdf with open(out, wb) as f: writer.write(f) print(f 拆分生成{out})拆分后每个分片单独总结最后在 docx 里按顺序拼接即可。4. 验证请求跑通一次并校验摘要质量配置和脚本就绪后先拿一个 PDF 试跑别一上来就全量。把pdf_dir指向只放一个文件的测试目录执行python main.py正常输出类似共发现 1 个 PDF 文件 处理文件2024行业报告.pdf 上传成功file id file-fe-xxxx 摘要已保存D:/ABooks/output/2024行业报告.docx 已删除远端文件file-fe-xxxx 完成成功 1 个失败 0 个打开生成的 docx检查摘要质量。我一般看三点章节结构是否完整、要点是否覆盖原文核心、有没有明显幻觉。如果发现摘要太笼统调整config.toml里的 user prompt比如要求「每个章节至少 3 条要点标注原文页码」。如果返回被截断说明输出长度接近上限可以把 prompt 改成「先输出前 5 章等我回复继续再输出后续」做分段续写。想快速验证模型对某个文件的理解是否到位可以先用模型对话入口手动传一次文件问几个问题确认解析正常再跑批量。模型对话入口在 https://taotoken.net/api-keys 旁边的对话页创建 Key 后即可使用。5. 本篇常见错排查ResponseTimeout 报错长文档首轮解析慢或输出太长。对策是加大request_timeout并在chat_with_retry里保留重试。如果反复超时把 prompt 拆成「先总结前半部分」再续。400 报错 file id 无效上传后立刻请求有时会撞上解析未完成。可以在上传后time.sleep(5)再发请求或捕获错误后重试。限流失败Qwen-Long 限流是 100 QPM脚本里RateLimiter的 interval 要按 60/1000.6 秒设置。如果你并发跑多个进程总调用量会叠加建议单进程串行。文件删除失败网络抖动或 file id 已失效。删除失败不影响主流程但会占配额建议定期在控制台清理。docx 中文乱码python-docx默认字体对中文支持一般可以在保存前设置样式字体为「微软雅黑」。Key 读不到确认config.toml路径正确Python 3.10 以下装了tomli。用环境变量覆盖时读取逻辑要加os.environ.get兜底。6. 接入方式与后续扩展批量总结跑通后常见的扩展方向有三个一是把结果汇总成一个总表用 pandas 读所有 docx 提取要点二是接入定时任务每天扫一次新入库的 PDF三是把摘要写回知识库或工单系统。这些都不需要改模型调用层只要在process_pdf后面加钩子。如果你要长期跑编码类或 Agent 类任务可以考虑 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入文档和 API Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 和 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。实际用下来把 Key 和 base_url 抽到配置、把重试和限流写进封装是这类批量任务最省心的做法后面换模型或加并发都不用动业务代码。