production-agentic-rag-course Week 2 实战：arXiv API 集成与科学 PDF 处理管线构建

RAG人工智能大模型本地部署后端AI Agent教程【免费下载链接】production-agentic-rag-course项目地址https://gitcode.com/GitHub_Trending/ar/production-agentic-rag-course点击查看免费下载本篇文章围绕开源仓库 production-agentic-rag-course 的 Week 2 教学材料notebooks/week2/README.md与notebooks/week2/week2_arxiv_integration.ipynb展开系统讲解如何为 RAG 系统构建一条生产级学术数据摄入管线从带限速与重试的 arXiv API 客户端、Docling 结构化 PDF 解析到 PostgreSQL 去重存储再到 Airflow 每日自动化调度。读完本文你将掌握完整的数据摄入流水线架构、各环节的源码级实现原理以及可直接复制的验证与调试方法。Week 2 数据摄入架构从 arXiv API 检索到 PDF 解析再到数据库存储的完整链路。Week 2 在项目中的定位与核心目标Week 2 是 arXiv Paper Curator 项目的第二周课程目标是为 RAG检索增强生成系统搭建内容供给层——自动获取、处理并存储 arXiv 上的最新学术论文。这是后续所有检索、索引与问答能力的原料基础其产出论文元数据 结构化全文将被 Week 3 的 OpenSearch 全文检索、Week 4 的混合检索直接消费。本周要交付的核心能力包括五方面能力域具体内容对应源码arXiv API 集成带限速与重试的健壮客户端、日期过滤、cs.AI 分类检索src/services/arxiv/client.pyPDF 处理管线PDF 下载与缓存、Docling 结构化内容抽取、优雅降级src/services/pdf_parser/数据库存储论文元数据与全文内容持久化到 PostgreSQL、upsert 去重src/repositories/paper.py错误处理全链路异常隔离单篇失败不阻断整批处理src/services/metadata_fetcher.py自动化就绪为 Airflow 编排准备的模块化组件与 DAGairflow/dags/arxiv_paper_ingestion.py从源码结构看本周核心组件形成了清晰的职责分层ArxivClient数据获取→PDFParserService/DoclingParser内容解析→PaperRepository持久化→MetadataFetcher编排。每个组件都通过factory.py中的工厂函数创建统一从src/config.py读取配置保持了依赖注入与可测试性。环境准备基础设施校验与全新容器构建前置条件检查开始 Week 2 前需要确认三项前置条件Week 1 的基础设施已完成PostgreSQL 16、FastAPI、Airflow、Docker Compose 服务栈UV 虚拟环境已激活notebook 的 kernel 为.venvPython 3.12Docker Desktop 正常运行为什么必须重建容器Week 2 引入了新的 Python 依赖docling、arxiv 客户端以及更新的 Airflow DAG必须重建镜像而非使用缓存层。README 给出了两种重建方式方式一常规重建保留数据docker compose down docker compose up --build方式二全新开始推荐新用户或出现 schema 冲突时# 完全清空数据卷确保使用正确的 Week 2 schema docker compose down -v # 构建最新代码的全新容器 docker compose up --build -d方式二会销毁已有数据但能确保数据库拥有 Week 2 新增的 PDF 处理与 arXiv 元数据相关列详见下文 数据库模式设计。建议在以下场景使用首次运行 Week 2、出现 schema 错误或列缺失错误、希望从干净数据库开始、Week 1 数据不再重要。服务健康检查notebook 的第一批代码单元通过subprocess与requests实现了自动化的容器与健康检查覆盖五个服务端点服务健康检查端点FastAPIhttp://localhost:8000/api/v1/healthPostgreSQL经 API 验证http://localhost:8000/api/v1/healthOllamaLLM 服务http://localhost:11434/api/versionOpenSearchhttp://localhost:9200/_cluster/healthAirflowhttp://localhost:8080/health检查脚本会逐项输出✓ Healthy或✗及具体原因只有全部健康才进入下一步。提示刚重建完容器后 Airflow 与 OpenSearch 启动最慢可等待 1-2 分钟再重跑检查单元。arXiv API 集成限速、重试与日期过滤客户端的配置化设计ArxivClient的所有行为参数都来自 src/config.py 中的ArxivSettingsPydantic 配置类支持.env中的ARXIV__前缀环境变量覆盖配置项默认值作用base_urlhttps://export.arxiv.org/api/queryarXiv API 查询端点pdf_cache_dir./data/arxiv_pdfsPDF 本地缓存目录字段校验器会自动创建rate_limit_delay3.0秒请求间隔限速遵循 arXiv 官方建议timeout_seconds30HTTP 超时阈值max_results15单次默认拉取数量search_categorycs.AI默认检索分类download_max_retries3PDF 下载最大重试次数download_retry_delay_base5.0秒重试退避基准指数退避max_concurrent_downloads5并发下载上限max_concurrent_parsing1并发解析上限客户端通过 src/services/arxiv/factory.py 的make_arxiv_client()创建notebook 中一行即可完成实例化并打印关键参数from src.services.arxiv.factory import make_arxiv_client arxiv_client make_arxiv_client() print(f✓ Client created: {arxiv_client.base_url}) print(f Rate limit: {arxiv_client.rate_limit_delay}s) print(f Max results: {arxiv_client.max_results}) print(f Category: {arxiv_client.search_category})核心方法fetch_papers 的请求构造fetch_papers() 是主抓取入口其签名如下async def fetch_papers( self, max_results: Optional[int] None, start: int 0, sort_by: str submittedDate, sort_order: str descending, from_date: Optional[str] None, to_date: Optional[str] None, ) - List[ArxivPaper]:内部实现要点查询构造基础查询为cat:cs.AI若提供日期参数则转换为 arXiv API 的submittedDate区间语法起始日补0000、结束日补2359并用号连接区间例如cat:cs.AI AND submittedDate:[202508080000TO202508092359]。参数编码使用urlencode(..., safe:[])刻意不转义:、、[、]这些 arXiv 查询语法所需字符。结果上限max_results被min(max_results, 2000)钳制防止单次请求过大。响应解析通过标准库xml.etree.ElementTree解析 Atom XML逐一提取 arXiv ID、标题清理换行、作者列表、摘要、分类、发布日期与 PDF 链接_parse_single_entry并强制将http://arxiv.org/的 PDF 链接升级为 HTTPS_get_pdf_url。限速与重试的工程细节限速逻辑位于请求发送前client.py记录上一次请求时间若间隔不足rate_limit_delay默认 3 秒则asyncio.sleep补齐差值再更新时间戳。这一机制保证无论并发场景如何对 arXiv 的实际请求频率始终受限速约束——这也是 README 中约 20 篇/分钟吞吐量预期的由来。异常处理采用分层设计超时抛ArxivAPITimeoutError、HTTP 错误抛ArxivAPIException、XML 解析失败抛ArxivParseError定义见 src/exceptions.py。notebook 中专门演示了 503 场景——arXiv 临时不可用属于正常现象捕获异常后优雅返回空列表恰好验证了限速与错误处理在真实环境中的工作状态。notebook 还提供了另外两个检索入口的调用示例fetch_papers_with_query直接传入自定义查询串支持按作者au:LeCun AND cat:cs.AI、标题关键词ti:transformer AND cat:cs.AI等高级检索client.py。fetch_paper_by_id按 arXiv ID 精确取单篇自动剥离版本号如2507.17748v1→2507.17748后以id_list参数查询client.py。日期过滤的 notebook 实测from_date 20250808 to_date 20250809 date_papers await arxiv_client.fetch_papers( max_results5, from_datefrom_date, to_dateto_date, )测试结果将打印该日期窗口内论文的 ID、标题截断 60 字符、前两位作者、分类与发布日期可直接验证日期过滤是否生效。PDF 处理管线下载缓存与 Docling 结构化解析下载与缓存download_pdf() 实现下载与本地缓存论文 arXiv ID 中的/会被替换为_生成安全文件名如2508.12345.pdf存入pdf_cache_dir。若缓存文件已存在且未指定force_downloadTrue直接复用缓存避免重复下载。下载使用流式写入client.stream(GET, url)配合aiter_bytes()分块落盘不一次性载入内存。重试采用指数退避第 N 次重试等待download_retry_delay_base * N秒重试耗尽后清理不完整的残留文件并抛PDFDownloadException/PDFDownloadTimeoutError_download_with_retry。notebook 验证代码会打印 PDF 文件名与大小MB确认下载与缓存正常工作pdf_path await arxiv_client.download_pdf(test_paper) if pdf_path and pdf_path.exists(): size_mb pdf_path.stat().st_size / (1024 * 1024) print(f✓ PDF downloaded: {pdf_path.name} ({size_mb:.2f} MB))Docling 解析从 PDF 到结构化内容PDFParserService内部封装了 DoclingParser通过 src/services/pdf_parser/factory.py 的make_pdf_parser_service()带lru_cache的单例工厂创建配置来自PDFParserSettings配置项默认值作用max_pages30单篇最大解析页数防止大论文耗尽内存max_file_size_mb20最大文件体积do_ocrFalse是否启用 OCR扫描版 PDF默认关闭以保速度do_table_structureTrue是否抽取表格结构解析前会执行四道校验_validate_pdf文件非空、大小不超限、文件头为%PDF-魔数、页数不超max_pages借助pypdfium2读取页数。超限文件被优雅跳过返回None而非抛错损坏文件则抛出明确异常。解析流程使用 Docling 的DocumentConverter现代 APImax_num_pages与max_file_size双重限制防内存溢出随后按文档结构元素标签title/section_header切分章节生成PaperSection列表并用doc.export_to_text()输出全文raw_textdocling.py。最终产出PdfContent其中parser_used标记为ParserType.DOCLING。notebook 的解析测试会展示章节数与原始文本长度pdf_content await pdf_parser.parse_pdf(test_pdf) print(f Sections: {len(pdf_content.sections)}) print(f Raw text length: {len(pdf_content.raw_text)} characters) print(f Parser used: {pdf_content.parser_used})关于解析成功率的合理预期README 明确提示并非所有 PDF 都能成功解析。Docling 对标准学术论文格式效果最好80-90% 的成功率属于正常水平。系统设计上对此有充分预期——解析失败会记录错误并继续处理下一篇绝不因单篇失败中断整批任务。这一点在下一节的全管线编排中得到落实。数据库集成PostgreSQL 存储与 upsert 去重数据库模式设计Week 2 的 schema 更新也是必须重建容器的根本原因体现在 src/models/paper.py 的Paper模型中包含三类字段arXiv 元数据arxiv_id唯一索引、title、authorsJSON、abstract、categoriesJSON、published_date、pdf_url解析内容raw_text全文、sectionsJSON 章节、referencesJSON 引用处理状态parser_used、parser_metadataJSON、pdf_processed布尔、pdf_processing_date以及created_at/updated_at时间戳数据库连接由 src/db/factory.py 的make_database()创建默认连接串为postgresql://rag_user:rag_passwordlocalhost:5432/rag_db可通过POSTGRES_前缀环境变量覆盖见 src/schemas/database/config.py。upsert 去重逻辑PaperRepository.upsert() 实现存在即更新、不存在即新建def upsert(self, paper_create: PaperCreate) - Paper: existing_paper self.get_by_arxiv_id(paper_create.arxiv_id) if existing_paper: # 仅更新本次提交中出现的字段exclude_unset for key, value in paper_create.model_dump(exclude_unsetTrue).items(): setattr(existing_paper, key, value) return self.update(existing_paper) else: return self.create(paper_create)exclude_unsetTrue保证每日重复摄入同一 arXiv ID 时只覆盖实际变更的字段配合arxiv_id唯一索引从应用层与数据库层双重杜绝重复记录。仓库还提供了丰富的查询方法get_by_arxiv_id、get_processed_papers已解析全文、get_unprocessed_papers待补解析、get_papers_with_raw_text已有全文、get_processing_stats处理率统计。notebook 存储测试的完整流程构造PaperCreate注意用dateutil.parser.parse将字符串日期转为datetime→paper_repo.upsert(paper_create)→paper_repo.get_by_arxiv_id(...)回读验证。全管线编排MetadataFetcher 的端到端处理编排器设计MetadataFetcher 是 README 中标记为 的主编排器将三个子系统串成一条流水线Step 1 抓取arxiv_client.fetch_papers(...)默认按提交时间倒序Step 2 解析_process_pdfs_batch(papers)并发处理 PDFStep 3 存储_store_papers_to_db(papers, parsed_papers, db_session)fetch_and_process_papers()的签名支持精细控制async def fetch_and_process_papers( self, max_results: Optional[int] None, from_date: Optional[str] None, to_date: Optional[str] None, process_pdfs: bool True, store_to_db: bool True, db_session: Optional[Session] None, ) - Dict[str, Any]:返回的统计字典包含papers_fetched、pdfs_downloaded、pdfs_parsed、papers_stored、processing_time、errors等键notebook 据此打印管线结果与成功率的计算依据。并发模型与优雅降级PDF 批处理_process_pdfs_batch采用下载-解析重叠流水线用asyncio.Semaphore分别控制下载并发默认 5与解析并发默认 1每篇论文的下载完成后立即进入解析期间其它下载继续——这一设计专门面向每天上百篇的生产负载。每篇论文的下载解析在_download_and_parse_pipeline中执行并遵循降级存储策略下载失败 → 记录download_failures不阻塞其它论文下载成功但解析失败 → 仍保留元数据入库标记pdf_processed: False解析成功 → 全文内容序列化后随元数据一并写入标记pdf_processed: True。_store_papers_to_db对每篇论文单独 try/except最后统一commit()出错时rollback()并归零计数。这意味着即使一批中有若干论文失败成功部分依然完整落库——这正是错误处理优雅继续这一性能特征在源码层面的体现。notebook 提供了两种全管线测试Test 6process_pdfsFalse快速验证抓取入库与 Test 8process_pdfsTruefrom_date20250813to_date20250814验证含 PDF 的完整链路并计算下载成功率与解析成功率download_rate (results[pdfs_downloaded] / results[papers_fetched]) * 100 parse_rate (results[pdfs_parsed] / results[pdfs_downloaded]) * 100生产就绪Airflow 每日自动化Week 2 的最终验收点之一是 Airflow DAG 就绪。notebook 通过docker exec检查 DAG 状态docker exec rag-airflow airflow dags list docker exec rag-airflow airflow dags list-import-errorsWeb UI 访问方式http://localhost:8080用户名/密码均为admin。仓库中的 arxiv_paper_ingestion.py 定义了核心 DAGarxiv_paper_ingestion调度schedule0 6 * * 1-5工作日 6:00 UTC 每日抓取catchupFalse、max_active_runs1任务链setup_environment → fetch_daily_papers → index_papers_hybrid → generate_daily_report → cleanup_temp_files健壮性retries2、retry_delay30 分钟cleanup_temp_files会自动删除/tmp下超过 30 天的 PDF 释放磁盘模块化任务函数分布在airflow/dags/arxiv_ingestion/下的setup.py、fetching.py、indexing.py、reporting.py中notebook 同时提醒了一个已知问题若list-import-errors输出包含docling说明 Airflow 容器内尚未安装 Docling——这是 Week 2 的预期现象DAG 结构已完整只需在容器启动脚本中补齐依赖可参考 airflow/requirements-airflow.txt即可在运行时修复。性能特征与成功标准本周系统能力基线README 给出的性能特征作为项目自身文档中的设计预期具体数值会随网络与硬件环境浮动arXiv API约 20 篇/分钟遵守 3 秒限速PDF 处理单篇 2-5 秒取决于 PDF 复杂度数据库存储约 100 篇/秒批量操作错误处理单点失败时优雅继续成功率论文抓取 95%PDF 解析 80-90%Week 2 完成判定标准✅ 全部满足即视为本周达成arXiv API 客户端在正确限速下抓取论文PDF 下载与缓存稳定可靠Docling 从学术论文中抽取结构化内容数据库存储完整论文元数据及关系全管线端到端处理论文具备错误处理能力Airflow DAG 配置完成、可自动化运行所有组件具备生产级错误处理与监控时间投入参考全新容器构建10-15 分钟Week 2 依赖必需notebook 完成45-60 分钟管线测试30-45 分钟总计约 1.5-2 小时故障排查速查表按 README 的 Support Resources 整理遇到问题依次排查确认容器为全新构建执行docker compose up --buildWeek 2 依赖docling、arxiv 客户端与新 DAG 必须重新构建镜像复查 Week 1 基础设施所有服务健康检查全部✓见上文健康检查表检查 notebook 的排障章节针对每个测试单元的输出定位失败环节查看服务健康与日志docker compose ps与docker logs service定位容器级问题Airflow 特定问题确认 DAG 无 import 错误若报docling缺失在 Airflow 容器中补齐依赖后重启完成 Week 2 后你已经具备理解生产级数据管线架构、应对真实 API 集成挑战、实现健壮错误处理与监控的能力可以无缝进入 Week 3OpenSearch 集成与全文检索——届时本周入库的论文元数据与解析全文将成为混合检索索引的数据源。赞分享RAG人工智能大模型本地部署后端AI Agent教程【免费下载链接】production-agentic-rag-course项目地址https://gitcode.com/GitHub_Trending/ar/production-agentic-rag-course点击查看免费下载相关推荐production-agentic-rag-course 实战Week 5 完整 RAG 系统 —— Ollama 本地 LLM、混合检索与流式 API 的端到端集成production agentic rag course 实战Week 5 完整 RAG 系统 —— Ollama 本地 LLM、混合检索与流式 API 的RAG人工智能大模型本地部署后端AI Agent教程production-agentic-rag-course Week 7 实战用 LangGraph 构建 Agentic RAG 智能检索并以 Telegram Bot 打造移动端论文问答助手production agentic rag course Week 7 实战用 LangGraph 构建 Agentic RAG 智能检索并以 TelegRAG人工智能大模型本地部署后端AI Agent教程production-agentic-rag-course 的 Airflow 配置指南基于 Apache Airflow 的 arXiv 论文每日摄取管道实战production agentic rag course 的 Airflow 配置指南基于 Apache Airflow 的 arXiv 论文每日摄取管道实RAG人工智能大模型本地部署后端AI Agent教程上一篇BMAD-METHOD 快速修复指南用 bmad-build 无规划直入 Build 阶段处理 Bug 与小型改动下一篇ZeroClaw SOP Fan-In: 用 Telegram / Discord / Slack 等对话渠道消息触发 SOP 运行创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考