Chandra OCR部署教程Airflow调度PDF批量解析任务支持失败重试与告警你是不是也遇到过这样的烦恼手头有一堆扫描的合同、发票、报告PDF需要把里面的文字、表格、公式都提取出来还要保留原来的排版格式。手动复制粘贴效率太低还容易出错。用传统的OCR工具对复杂排版、表格和公式的支持又很差出来的结果一团糟。今天我就来分享一个实战方案用Chandra OCR Airflow搭建一个自动化、高可靠的PDF批量解析系统。这个方案不仅能一键将PDF/图片转换成结构清晰的Markdown、HTML或JSON还能通过Airflow进行任务调度、失败重试和自动告警非常适合处理海量文档。简单来说你只需要准备一台有GPU的机器甚至RTX 3060都行跟着教程走就能拥有一个“文档解析机器人”7x24小时为你工作。1. 为什么选择Chandra OCR在开始动手之前我们先搞清楚为什么要用Chandra。市面上OCR工具不少但Chandra有几个杀手锏让它成为处理复杂文档的首选。1.1 核心优势布局感知与高精度Chandra不是一个简单的文字识别工具它是一个“布局感知”的视觉语言模型。这意味着它不仅能认出字还能理解文档的结构。保留完整排版它能识别标题、段落、列表、分栏并在输出的Markdown/HTML中完美保留这些结构。你得到的不是一堆乱序的文字而是有层次、可读性强的文档。复杂元素识别表格、数学公式、手写体、表单复选框这些让传统OCR头疼的元素Chandra都能很好地处理。官方在权威的olmOCR基准测试中拿到了83.1的综合分在表格、手写、数学公式等单项上甚至领先GPT-4o和Gemini Flash 2。多格式输出一次推理同时生成Markdown、HTML和JSON三种格式。Markdown方便编辑和阅读HTML适合网页展示JSON则包含了每个元素的坐标和类型信息非常适合后续接入RAG检索增强生成系统或进行自动化处理。1.2 技术栈选型vLLM后端 Airflow调度为了构建一个稳定、高效的生产级系统我们选择以下组合Chandra with vLLM后端vLLM是一个高性能的推理服务框架能极大提升模型吞吐量并支持多GPU并行。相比本地HuggingFace加载vLLM服务化后更稳定也方便其他系统调用。Apache Airflow一个强大的工作流调度平台。我们将用它来编排PDF解析任务实现定时触发、任务依赖、失败自动重试、失败告警等功能。这是保证批量任务可靠运行的关键。简单来说vLLM负责提供强大的“识别大脑”Airflow负责指挥这个“大脑”何时、以何种方式、处理哪些“文档”并确保任务万无一失。2. 环境准备与快速部署我们的部署分为两步先启动Chandra OCR服务大脑再配置Airflow调度系统指挥官。2.1 第一步部署Chandra OCR (vLLM服务)首先确保你的环境有一张至少8GB显存的NVIDIA GPU例如RTX 3060 12GB, RTX 4090等。我们将使用Docker来快速部署这是最省心的方法。1. 拉取镜像并启动服务打开终端执行以下命令# 拉取集成了vLLM的Chandra OCR镜像 docker pull datalabto/chandra-ocr-vllm:latest # 运行容器将服务端口7860映射到宿主机 docker run -d --gpus all \ -p 7860:7860 \ --name chandra-ocr \ datalabto/chandra-ocr-vllm:latest--gpus all将宿主机的所有GPU分配给容器。-p 7860:7860将容器内的7860端口映射出来我们通过这个端口访问OCR服务。--name chandra-ocr给容器起个名字方便管理。2. 验证服务是否正常容器启动后稍等片刻首次运行需要加载模型。然后在浏览器中访问http://你的服务器IP:7860。 你应该能看到一个简洁的Streamlit交互界面。上传一张图片或PDF试试如果能正常返回识别结果说明Chandra OCR服务已经成功运行。3. 服务API接口除了网页界面该服务主要提供HTTP API供程序调用。核心接口是URL:http://你的服务器IP:7860/ocr方法: POST参数: 通过表单form-data上传文件字段名为file。你可以用curl命令快速测试curl -X POST -F file/path/to/your/document.pdf http://localhost:7860/ocr如果返回一串包含markdown、html等字段的JSON说明API工作正常。至此我们的“识别大脑”已经在线待命。2.2 第二步安装与配置Apache AirflowAirflow的安装方式很多这里我们使用官方推荐的pip安装方式并配置本地Executor适合单机调度。对于更复杂的生产环境可以考虑Celery或Kubernetes Executor。1. 创建Airflow专属环境推荐使用conda或venv创建一个独立的Python环境避免包冲突。# 使用venv python -m venv airflow_venv source airflow_venv/bin/activate # Linux/macOS # 或 airflow_venv\Scripts\activate # Windows # 安装Airflow指定版本和依赖 pip install apache-airflow2.10.0 --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.10.0/constraints-3.10.txt # 安装必要的Provider和依赖 pip install apache-airflow-providers-http pip install requests pandas # 用于我们的任务脚本2. 初始化Airflow数据库Airflow需要一个元数据库默认是SQLite来存储任务状态、变量等信息。# 设置Airflow的家目录所有配置文件、DAG文件都会在这里 export AIRFLOW_HOME~/airflow # 初始化数据库 airflow db init3. 创建管理员用户airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com执行后会提示你输入密码。4. 启动Web服务器与调度器打开两个终端窗口分别运行# 终端1启动Web服务器访问 http://localhost:8080 airflow webserver --port 8080 # 终端2启动调度器 airflow scheduler现在打开浏览器访问http://localhost:8080用刚才创建的用户名密码登录你就能看到Airflow的Web UI了。我们的“指挥官”平台也准备就绪了。接下来就是编写任务脚本让两者协同工作。3. 编写Airflow DAGPDF批量解析任务DAG有向无环图是Airflow的核心概念它定义了一组任务及其依赖关系。我们将在$AIRFLOW_HOME/dags目录下创建一个Python文件例如pdf_ocr_pipeline.py。这个DAG主要做以下几件事扫描指定目录下的PDF文件。调用我们部署好的Chandra OCR服务逐个解析PDF。将解析结果Markdown保存到指定目录。如果解析失败自动重试并在最终失败时发送告警。下面是完整的DAG代码示例from datetime import datetime, timedelta import os from pathlib import Path import requests from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.dummy import DummyOperator from airflow.exceptions import AirflowException # 定义默认参数 default_args { owner: data_team, depends_on_past: False, email_on_failure: True, # 失败时发邮件 email: [your_alert_emailexample.com], # 替换为你的告警邮箱 email_on_retry: False, retries: 3, # 失败后重试3次 retry_delay: timedelta(minutes2), # 每次重试间隔2分钟 } # OCR服务地址替换为你的实际地址 OCR_SERVICE_URL http://localhost:7860/ocr # 本地目录路径 SOURCE_PDF_DIR /data/pdfs/to_process # 待处理的PDF目录 OUTPUT_MD_DIR /data/pdfs/processed_md # 处理后的Markdown输出目录 def process_single_pdf(pdf_path, output_dir): 处理单个PDF文件 try: with open(pdf_path, rb) as f: files {file: f} response requests.post(OCR_SERVICE_URL, filesfiles, timeout120) # 设置超时时间 if response.status_code 200: result response.json() markdown_content result.get(markdown, ) # 生成输出文件名同PDF名后缀改为.md output_filename Path(pdf_path).stem .md output_path Path(output_dir) / output_filename # 保存Markdown文件 with open(output_path, w, encodingutf-8) as md_file: md_file.write(markdown_content) print(f成功处理: {pdf_path} - {output_path}) return True else: print(fOCR服务调用失败 ({response.status_code}): {pdf_path}) return False except Exception as e: print(f处理文件 {pdf_path} 时发生异常: {str(e)}) return False def batch_process_pdfs(**context): 批量处理PDF的主函数 source_dir Path(SOURCE_PDF_DIR) output_dir Path(OUTPUT_MD_DIR) # 确保输出目录存在 output_dir.mkdir(parentsTrue, exist_okTrue) # 查找所有PDF文件 pdf_files list(source_dir.glob(*.pdf)) if not pdf_files: print(f在目录 {SOURCE_PDF_DIR} 中未找到PDF文件。) return print(f开始批量处理共 {len(pdf_files)} 个PDF文件。) success_count 0 failed_files [] for pdf_file in pdf_files: success process_single_pdf(pdf_file, output_dir) if success: success_count 1 else: failed_files.append(pdf_file.name) # 推送处理结果到Airflow XCom方便后续任务或告警使用 context[ti].xcom_push(keysuccess_count, valuesuccess_count) context[ti].xcom_push(keyfailed_files, valuefailed_files) context[ti].xcom_push(keytotal_files, valuelen(pdf_files)) print(f处理完成。成功: {success_count}, 失败: {len(failed_files)}) if failed_files: print(f失败文件列表: {failed_files}) # 如果有失败文件可以选择抛出异常让Airflow触发重试或标记失败 # raise AirflowException(f部分文件处理失败: {failed_files}) def send_slack_alert(**context): 发送告警到Slack示例需配置Webhook # 这里以Slack为例你也可以替换成邮件、钉钉、企业微信等 success_count context[ti].xcom_pull(keysuccess_count, task_idsbatch_ocr_task) failed_files context[ti].xcom_pull(keyfailed_files, task_idsbatch_ocr_task) total_files context[ti].xcom_pull(keytotal_files, task_idsbatch_ocr_task) if failed_files: slack_webhook_url https://hooks.slack.com/services/你的WEBHOOK_URL message { text: f⚠️ PDF批量OCR任务告警, blocks: [ { type: section, text: { type: mrkdwn, text: f*任务状态*: 部分失败\n*处理总数*: {total_files}\n*成功数*: {success_count}\n*失败数*: {len(failed_files)}\n*失败文件*: {, .join(failed_files)[:500]}... # 截断长消息 } } ] } try: requests.post(slack_webhook_url, jsonmessage) print(Slack告警已发送。) except Exception as e: print(f发送Slack告警失败: {e}) # 定义DAG with DAG( pdf_batch_ocr_pipeline, default_argsdefault_args, description使用Chandra OCR批量处理PDF并支持重试与告警, schedule_intervaltimedelta(hours6), # 每6小时运行一次可按需调整 start_datedatetime(2024, 1, 1), catchupFalse, # 不追补历史任务 tags[ocr, batch_processing], ) as dag: start DummyOperator(task_idstart) # 核心任务批量OCR处理 batch_ocr_task PythonOperator( task_idbatch_ocr_task, python_callablebatch_process_pdfs, provide_contextTrue, ) # 告警任务仅在核心任务成功后执行用于汇总结果并发送成功/失败通知 alert_task PythonOperator( task_idsend_alert, python_callablesend_slack_alert, provide_contextTrue, ) end DummyOperator(task_idend) # 定义任务依赖关系 start batch_ocr_task alert_task end关键点解释重试机制在default_args中设置了retries: 3和retry_delay。如果batch_ocr_task因网络波动、OCR服务暂时不可用等原因失败Airflow会自动重试3次每次间隔2分钟。告警机制email_on_failure: True如果任务最终失败重试后仍失败Airflow会自动发送邮件到指定邮箱。send_slack_alert任务这是一个自定义的告警任务无论成功与否都会运行。它会从上游任务拉取处理结果通过XCom如果有失败文件则发送详细告警到Slack。你可以根据团队习惯将其改为邮件、钉钉、企业微信等。任务编排流程很简单开始 - 批量处理 - 发送告警 - 结束。alert_task被设置为必须在batch_ocr_task之后执行以获取其运行结果。将上述代码保存为$AIRFLOW_HOME/dags/pdf_ocr_pipeline.py。Airflow调度器会自动加载这个DAG。4. 在Airflow中触发与监控任务回到Airflow的Web UI (http://localhost:8080)。找到你的DAG在DAGs列表里你应该能看到名为pdf_batch_ocr_pipeline的DAG。激活DAG点击DAG名称左侧的开关按钮将其状态变为“On”绿色。手动触发测试点击DAG名称进入详情页点击右上角的“Trigger DAG”按钮可以立即运行一次任务。这是测试你的配置是否正确的好方法。监控任务状态Tree View以树状图展示最近几次DAG运行的任务状态成功、失败、运行中、重试等。Graph View图形化展示DAG结构和当前运行状态。Task Instance Details点击任意一个任务实例可以查看详细日志。这是排查问题最重要的地方。如果OCR调用失败日志里会显示具体的HTTP错误码或异常信息。查看定时任务由于我们设置了schedule_intervaltimedelta(hours6)这个DAG会每6小时自动运行一次无需手动干预。5. 实用技巧与进阶配置系统跑起来后你可能会遇到一些实际问题这里提供一些解决思路和进阶玩法。5.1 性能与稳定性优化调整vLLM参数如果处理速度慢或显存不足可以在启动Chandra容器时调整vLLM参数例如限制最大并发数、调整批处理大小等。需要查阅vLLM和Chandra的文档。Airflow并行度上述DAG是单文件顺序处理。如果PDF数量巨大可以改造batch_process_pdfs函数使用PythonOperator的executor_config配合Kubernetes Pod Operator或者使用Dynamic Task Mapping来并行处理多个文件。错误处理细化当前的错误处理比较粗略。你可以根据OCR服务返回的具体错误码如文件格式不支持、服务超时、内容过长等实现更精细的重试策略例如格式错误不重试超时则重试。5.2 扩展应用场景与RAG管道集成将输出的Markdown或JSON直接送入向量数据库如Chroma、Milvus构建企业知识库的索引管道。结果后处理在process_single_pdf函数中增加后处理步骤比如用正则表达式提取关键信息合同金额、日期等或调用另一个LLM对解析结果进行摘要、分类。触发方式多样化除了定时调度还可以配置Airflow监听某个目录如使用FileSensor一旦有新PDF放入立即触发OCR任务实现近实时处理。5.3 常见问题排查任务一直处于“排队”状态检查Airflow调度器(scheduler)是否正常运行。检查Executor的并行工作线程数是否已满默认是16。调用OCR服务超时检查Chandra容器日志看模型是否加载成功。增大requests.post的timeout参数。检查网络是否通畅。显存不足(OOM)对于特别大或复杂的PDFChandra可能需要更多显存。考虑使用更高显存的GPU或在调用API时限制传入图片的分辨率如果Chandra API支持。告警未发出检查send_slack_alert任务日志看Slack Webhook URL是否正确网络是否可达。检查Airflow的邮箱配置smtp相关配置是否正确。6. 总结通过本教程我们成功搭建了一个基于Chandra OCR和Apache Airflow的自动化文档解析系统。这个方案的优势非常明显高精度借助Chandra领先的布局感知能力复杂文档的转换质量远超传统OCR。自动化与可靠Airflow提供了强大的调度、监控、重试和告警能力让批量处理任务无需人工值守即使出错也能自动恢复或及时通知。开箱即用与可扩展使用Docker和pip安装部署简单。整个管道是代码定义的非常容易修改和扩展以适应各种业务场景。从一堆杂乱无章的扫描件到结构清晰、可直接利用的数字化文本中间繁琐的解析工作现在可以完全交给这个自动化的“数字员工”。你可以放心地去处理更重要的业务逻辑而把重复的文档处理工作交给这个7x24小时稳定运行的管道。希望这个实战教程能帮你解决实际的文档处理难题。如果在部署过程中遇到问题欢迎在评论区交流讨论。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Chandra OCR部署教程：Airflow调度PDF批量解析任务，支持失败重试与告警
发布时间：2026/7/10 20:13:08
Chandra OCR部署教程Airflow调度PDF批量解析任务支持失败重试与告警你是不是也遇到过这样的烦恼手头有一堆扫描的合同、发票、报告PDF需要把里面的文字、表格、公式都提取出来还要保留原来的排版格式。手动复制粘贴效率太低还容易出错。用传统的OCR工具对复杂排版、表格和公式的支持又很差出来的结果一团糟。今天我就来分享一个实战方案用Chandra OCR Airflow搭建一个自动化、高可靠的PDF批量解析系统。这个方案不仅能一键将PDF/图片转换成结构清晰的Markdown、HTML或JSON还能通过Airflow进行任务调度、失败重试和自动告警非常适合处理海量文档。简单来说你只需要准备一台有GPU的机器甚至RTX 3060都行跟着教程走就能拥有一个“文档解析机器人”7x24小时为你工作。1. 为什么选择Chandra OCR在开始动手之前我们先搞清楚为什么要用Chandra。市面上OCR工具不少但Chandra有几个杀手锏让它成为处理复杂文档的首选。1.1 核心优势布局感知与高精度Chandra不是一个简单的文字识别工具它是一个“布局感知”的视觉语言模型。这意味着它不仅能认出字还能理解文档的结构。保留完整排版它能识别标题、段落、列表、分栏并在输出的Markdown/HTML中完美保留这些结构。你得到的不是一堆乱序的文字而是有层次、可读性强的文档。复杂元素识别表格、数学公式、手写体、表单复选框这些让传统OCR头疼的元素Chandra都能很好地处理。官方在权威的olmOCR基准测试中拿到了83.1的综合分在表格、手写、数学公式等单项上甚至领先GPT-4o和Gemini Flash 2。多格式输出一次推理同时生成Markdown、HTML和JSON三种格式。Markdown方便编辑和阅读HTML适合网页展示JSON则包含了每个元素的坐标和类型信息非常适合后续接入RAG检索增强生成系统或进行自动化处理。1.2 技术栈选型vLLM后端 Airflow调度为了构建一个稳定、高效的生产级系统我们选择以下组合Chandra with vLLM后端vLLM是一个高性能的推理服务框架能极大提升模型吞吐量并支持多GPU并行。相比本地HuggingFace加载vLLM服务化后更稳定也方便其他系统调用。Apache Airflow一个强大的工作流调度平台。我们将用它来编排PDF解析任务实现定时触发、任务依赖、失败自动重试、失败告警等功能。这是保证批量任务可靠运行的关键。简单来说vLLM负责提供强大的“识别大脑”Airflow负责指挥这个“大脑”何时、以何种方式、处理哪些“文档”并确保任务万无一失。2. 环境准备与快速部署我们的部署分为两步先启动Chandra OCR服务大脑再配置Airflow调度系统指挥官。2.1 第一步部署Chandra OCR (vLLM服务)首先确保你的环境有一张至少8GB显存的NVIDIA GPU例如RTX 3060 12GB, RTX 4090等。我们将使用Docker来快速部署这是最省心的方法。1. 拉取镜像并启动服务打开终端执行以下命令# 拉取集成了vLLM的Chandra OCR镜像 docker pull datalabto/chandra-ocr-vllm:latest # 运行容器将服务端口7860映射到宿主机 docker run -d --gpus all \ -p 7860:7860 \ --name chandra-ocr \ datalabto/chandra-ocr-vllm:latest--gpus all将宿主机的所有GPU分配给容器。-p 7860:7860将容器内的7860端口映射出来我们通过这个端口访问OCR服务。--name chandra-ocr给容器起个名字方便管理。2. 验证服务是否正常容器启动后稍等片刻首次运行需要加载模型。然后在浏览器中访问http://你的服务器IP:7860。 你应该能看到一个简洁的Streamlit交互界面。上传一张图片或PDF试试如果能正常返回识别结果说明Chandra OCR服务已经成功运行。3. 服务API接口除了网页界面该服务主要提供HTTP API供程序调用。核心接口是URL:http://你的服务器IP:7860/ocr方法: POST参数: 通过表单form-data上传文件字段名为file。你可以用curl命令快速测试curl -X POST -F file/path/to/your/document.pdf http://localhost:7860/ocr如果返回一串包含markdown、html等字段的JSON说明API工作正常。至此我们的“识别大脑”已经在线待命。2.2 第二步安装与配置Apache AirflowAirflow的安装方式很多这里我们使用官方推荐的pip安装方式并配置本地Executor适合单机调度。对于更复杂的生产环境可以考虑Celery或Kubernetes Executor。1. 创建Airflow专属环境推荐使用conda或venv创建一个独立的Python环境避免包冲突。# 使用venv python -m venv airflow_venv source airflow_venv/bin/activate # Linux/macOS # 或 airflow_venv\Scripts\activate # Windows # 安装Airflow指定版本和依赖 pip install apache-airflow2.10.0 --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.10.0/constraints-3.10.txt # 安装必要的Provider和依赖 pip install apache-airflow-providers-http pip install requests pandas # 用于我们的任务脚本2. 初始化Airflow数据库Airflow需要一个元数据库默认是SQLite来存储任务状态、变量等信息。# 设置Airflow的家目录所有配置文件、DAG文件都会在这里 export AIRFLOW_HOME~/airflow # 初始化数据库 airflow db init3. 创建管理员用户airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com执行后会提示你输入密码。4. 启动Web服务器与调度器打开两个终端窗口分别运行# 终端1启动Web服务器访问 http://localhost:8080 airflow webserver --port 8080 # 终端2启动调度器 airflow scheduler现在打开浏览器访问http://localhost:8080用刚才创建的用户名密码登录你就能看到Airflow的Web UI了。我们的“指挥官”平台也准备就绪了。接下来就是编写任务脚本让两者协同工作。3. 编写Airflow DAGPDF批量解析任务DAG有向无环图是Airflow的核心概念它定义了一组任务及其依赖关系。我们将在$AIRFLOW_HOME/dags目录下创建一个Python文件例如pdf_ocr_pipeline.py。这个DAG主要做以下几件事扫描指定目录下的PDF文件。调用我们部署好的Chandra OCR服务逐个解析PDF。将解析结果Markdown保存到指定目录。如果解析失败自动重试并在最终失败时发送告警。下面是完整的DAG代码示例from datetime import datetime, timedelta import os from pathlib import Path import requests from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.dummy import DummyOperator from airflow.exceptions import AirflowException # 定义默认参数 default_args { owner: data_team, depends_on_past: False, email_on_failure: True, # 失败时发邮件 email: [your_alert_emailexample.com], # 替换为你的告警邮箱 email_on_retry: False, retries: 3, # 失败后重试3次 retry_delay: timedelta(minutes2), # 每次重试间隔2分钟 } # OCR服务地址替换为你的实际地址 OCR_SERVICE_URL http://localhost:7860/ocr # 本地目录路径 SOURCE_PDF_DIR /data/pdfs/to_process # 待处理的PDF目录 OUTPUT_MD_DIR /data/pdfs/processed_md # 处理后的Markdown输出目录 def process_single_pdf(pdf_path, output_dir): 处理单个PDF文件 try: with open(pdf_path, rb) as f: files {file: f} response requests.post(OCR_SERVICE_URL, filesfiles, timeout120) # 设置超时时间 if response.status_code 200: result response.json() markdown_content result.get(markdown, ) # 生成输出文件名同PDF名后缀改为.md output_filename Path(pdf_path).stem .md output_path Path(output_dir) / output_filename # 保存Markdown文件 with open(output_path, w, encodingutf-8) as md_file: md_file.write(markdown_content) print(f成功处理: {pdf_path} - {output_path}) return True else: print(fOCR服务调用失败 ({response.status_code}): {pdf_path}) return False except Exception as e: print(f处理文件 {pdf_path} 时发生异常: {str(e)}) return False def batch_process_pdfs(**context): 批量处理PDF的主函数 source_dir Path(SOURCE_PDF_DIR) output_dir Path(OUTPUT_MD_DIR) # 确保输出目录存在 output_dir.mkdir(parentsTrue, exist_okTrue) # 查找所有PDF文件 pdf_files list(source_dir.glob(*.pdf)) if not pdf_files: print(f在目录 {SOURCE_PDF_DIR} 中未找到PDF文件。) return print(f开始批量处理共 {len(pdf_files)} 个PDF文件。) success_count 0 failed_files [] for pdf_file in pdf_files: success process_single_pdf(pdf_file, output_dir) if success: success_count 1 else: failed_files.append(pdf_file.name) # 推送处理结果到Airflow XCom方便后续任务或告警使用 context[ti].xcom_push(keysuccess_count, valuesuccess_count) context[ti].xcom_push(keyfailed_files, valuefailed_files) context[ti].xcom_push(keytotal_files, valuelen(pdf_files)) print(f处理完成。成功: {success_count}, 失败: {len(failed_files)}) if failed_files: print(f失败文件列表: {failed_files}) # 如果有失败文件可以选择抛出异常让Airflow触发重试或标记失败 # raise AirflowException(f部分文件处理失败: {failed_files}) def send_slack_alert(**context): 发送告警到Slack示例需配置Webhook # 这里以Slack为例你也可以替换成邮件、钉钉、企业微信等 success_count context[ti].xcom_pull(keysuccess_count, task_idsbatch_ocr_task) failed_files context[ti].xcom_pull(keyfailed_files, task_idsbatch_ocr_task) total_files context[ti].xcom_pull(keytotal_files, task_idsbatch_ocr_task) if failed_files: slack_webhook_url https://hooks.slack.com/services/你的WEBHOOK_URL message { text: f⚠️ PDF批量OCR任务告警, blocks: [ { type: section, text: { type: mrkdwn, text: f*任务状态*: 部分失败\n*处理总数*: {total_files}\n*成功数*: {success_count}\n*失败数*: {len(failed_files)}\n*失败文件*: {, .join(failed_files)[:500]}... # 截断长消息 } } ] } try: requests.post(slack_webhook_url, jsonmessage) print(Slack告警已发送。) except Exception as e: print(f发送Slack告警失败: {e}) # 定义DAG with DAG( pdf_batch_ocr_pipeline, default_argsdefault_args, description使用Chandra OCR批量处理PDF并支持重试与告警, schedule_intervaltimedelta(hours6), # 每6小时运行一次可按需调整 start_datedatetime(2024, 1, 1), catchupFalse, # 不追补历史任务 tags[ocr, batch_processing], ) as dag: start DummyOperator(task_idstart) # 核心任务批量OCR处理 batch_ocr_task PythonOperator( task_idbatch_ocr_task, python_callablebatch_process_pdfs, provide_contextTrue, ) # 告警任务仅在核心任务成功后执行用于汇总结果并发送成功/失败通知 alert_task PythonOperator( task_idsend_alert, python_callablesend_slack_alert, provide_contextTrue, ) end DummyOperator(task_idend) # 定义任务依赖关系 start batch_ocr_task alert_task end关键点解释重试机制在default_args中设置了retries: 3和retry_delay。如果batch_ocr_task因网络波动、OCR服务暂时不可用等原因失败Airflow会自动重试3次每次间隔2分钟。告警机制email_on_failure: True如果任务最终失败重试后仍失败Airflow会自动发送邮件到指定邮箱。send_slack_alert任务这是一个自定义的告警任务无论成功与否都会运行。它会从上游任务拉取处理结果通过XCom如果有失败文件则发送详细告警到Slack。你可以根据团队习惯将其改为邮件、钉钉、企业微信等。任务编排流程很简单开始 - 批量处理 - 发送告警 - 结束。alert_task被设置为必须在batch_ocr_task之后执行以获取其运行结果。将上述代码保存为$AIRFLOW_HOME/dags/pdf_ocr_pipeline.py。Airflow调度器会自动加载这个DAG。4. 在Airflow中触发与监控任务回到Airflow的Web UI (http://localhost:8080)。找到你的DAG在DAGs列表里你应该能看到名为pdf_batch_ocr_pipeline的DAG。激活DAG点击DAG名称左侧的开关按钮将其状态变为“On”绿色。手动触发测试点击DAG名称进入详情页点击右上角的“Trigger DAG”按钮可以立即运行一次任务。这是测试你的配置是否正确的好方法。监控任务状态Tree View以树状图展示最近几次DAG运行的任务状态成功、失败、运行中、重试等。Graph View图形化展示DAG结构和当前运行状态。Task Instance Details点击任意一个任务实例可以查看详细日志。这是排查问题最重要的地方。如果OCR调用失败日志里会显示具体的HTTP错误码或异常信息。查看定时任务由于我们设置了schedule_intervaltimedelta(hours6)这个DAG会每6小时自动运行一次无需手动干预。5. 实用技巧与进阶配置系统跑起来后你可能会遇到一些实际问题这里提供一些解决思路和进阶玩法。5.1 性能与稳定性优化调整vLLM参数如果处理速度慢或显存不足可以在启动Chandra容器时调整vLLM参数例如限制最大并发数、调整批处理大小等。需要查阅vLLM和Chandra的文档。Airflow并行度上述DAG是单文件顺序处理。如果PDF数量巨大可以改造batch_process_pdfs函数使用PythonOperator的executor_config配合Kubernetes Pod Operator或者使用Dynamic Task Mapping来并行处理多个文件。错误处理细化当前的错误处理比较粗略。你可以根据OCR服务返回的具体错误码如文件格式不支持、服务超时、内容过长等实现更精细的重试策略例如格式错误不重试超时则重试。5.2 扩展应用场景与RAG管道集成将输出的Markdown或JSON直接送入向量数据库如Chroma、Milvus构建企业知识库的索引管道。结果后处理在process_single_pdf函数中增加后处理步骤比如用正则表达式提取关键信息合同金额、日期等或调用另一个LLM对解析结果进行摘要、分类。触发方式多样化除了定时调度还可以配置Airflow监听某个目录如使用FileSensor一旦有新PDF放入立即触发OCR任务实现近实时处理。5.3 常见问题排查任务一直处于“排队”状态检查Airflow调度器(scheduler)是否正常运行。检查Executor的并行工作线程数是否已满默认是16。调用OCR服务超时检查Chandra容器日志看模型是否加载成功。增大requests.post的timeout参数。检查网络是否通畅。显存不足(OOM)对于特别大或复杂的PDFChandra可能需要更多显存。考虑使用更高显存的GPU或在调用API时限制传入图片的分辨率如果Chandra API支持。告警未发出检查send_slack_alert任务日志看Slack Webhook URL是否正确网络是否可达。检查Airflow的邮箱配置smtp相关配置是否正确。6. 总结通过本教程我们成功搭建了一个基于Chandra OCR和Apache Airflow的自动化文档解析系统。这个方案的优势非常明显高精度借助Chandra领先的布局感知能力复杂文档的转换质量远超传统OCR。自动化与可靠Airflow提供了强大的调度、监控、重试和告警能力让批量处理任务无需人工值守即使出错也能自动恢复或及时通知。开箱即用与可扩展使用Docker和pip安装部署简单。整个管道是代码定义的非常容易修改和扩展以适应各种业务场景。从一堆杂乱无章的扫描件到结构清晰、可直接利用的数字化文本中间繁琐的解析工作现在可以完全交给这个自动化的“数字员工”。你可以放心地去处理更重要的业务逻辑而把重复的文档处理工作交给这个7x24小时稳定运行的管道。希望这个实战教程能帮你解决实际的文档处理难题。如果在部署过程中遇到问题欢迎在评论区交流讨论。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
![为什么选择NVIDIA GLM-5.1-NVFP4？7大核心功能让你的AI应用脱颖而出 [特殊字符]](http://pic.xiahunao.cn/yaotu/为什么选择NVIDIA GLM-5.1-NVFP4？7大核心功能让你的AI应用脱颖而出 [特殊字符])
