30 分钟搭一条医疗文档处理管线：OCR 加分类智能体加结构化提取全流程

一句话结论

一条”扫描 PDF 进、结构化数据出”的医疗文档管线，用 Mistral 三件套半小时搭出来：OCR 读文档、智能体分类加置信度打分、第二个智能体按类别提字段，置信度不够自动暂停等人点确认——全程防崩溃可续跑。

这条管线做什么

Workflows 是构建、执行和监控复杂 AI 驱动工作流的编排平台。

它提供由久经考验的分布式系统基础设施支撑的持久、容错执行，加上对开发者友好的 SDK。

本教程里你将用三项 Mistral 能力搭一条端到端的医疗文档处理管线：OCR 读 PDF、智能体分类文档并提取结构化数据、Workflows 可靠地编排全流程。

管线接受任何扫描版医疗 PDF——处方、医院账单或影像报告——跑三步：光学字符识别（OCR）提取原始文本，一个 AI 智能体带置信度分数分类文档类型，第二个智能体把患者信息和文档特定字段提取为结构化 JSON。

因为建在 Mistral Workflows 上，每一步都持久且容错：worker 中途重启，工作流从断点恢复而不是从头再来。

你还会加一个人在环步骤：分类器置信度低于可配置阈值时，管线暂停等用户审查确认类别后提取才继续。

Mistral Workflows 支持这种可基于外部输入暂停恢复的长流程——用简单的异步队列或 API 调用链很难可靠实现。

你要构建的东西：

一个这样的工作流：接受医疗文件输入，用 Mistral OCR Processor 识别文档内容，用 Mistral Agents 分类内容类型并格式化数据。

一个用 Streamlit 构建的上传发票前端。

你需要的东西：

机器上装好 Python 3.12。装好 uv。一个 Mistral AI 账户。

步骤总览

设置环境。定义要提取的字段。构建工作流。创建应用前端。加负载测试。更新 Makefile。运行应用和工作流。成功！

设置环境

用下面的命令脚手架一个开箱即跑、Workflows SDK 已配置的 Python 项目，附带辅助命令：uvx mistralai-workflows-cli@latest setup

提示时选默认项目名，按步骤生成 Mistral API key。

检查入门代码

在 IDE 里打开 my-workflow 项目。

注意项目结构：src/workflows/ 目录含 __init__.py、hello.py、start.py；dev_worker.py、discover.py 放在 src/ 下；还有 .gitignore、Makefile、pyproject.toml、README.md。

脚手架项目的关键组件：

包含面向 Workflows 的 Agent Skills，方便用编程智能体开发。

工作流存放在 src/workflows 目录的独立文件里：start.py 包含从命令行触发工作流执行的代码；hello.py 给出最小工作流示例，展示 @workflows.activity()、@workflows.workflow.define() 和 @workflows.workflow.entrypoint() 装饰器的用法。

dev_worker.py 是本地开发用的 worker，监视 src/ 的 .py 变更并按需自动重启工作流 worker。

Makefile 包含精简开发和测试流程的辅助命令。

通过教程步骤，我们将向脚手架项目添加四个 Python 文件：

extraction_fields.py：智能体要消费的有用字符串和常量。

medical_doc_workflow.py：工作流核心的实际功能。

app.py：消费 PDF 跑工作流的 Streamlit 前端。

load_test.py：并行启动多个工作流看负载均衡的功能。

安装依赖

先对依赖要求做些改动。

打开 pyproject.toml 把依赖更新为：mistralai-workflows 加 [mistralai] extra 限定 3.x 版本、pydantic、python-dotenv、streamlit 1.55.0、pymupdf 1.23 以上。

这些更新做了三件事：用 [mistralai] extra 替换裸的 mistralai-workflows，加载提供与 Mistral 模型服务原生集成的插件，包括持久智能体、工具调用和多智能体交接；加 Streamlit 创建上传发票的基础前端；加 PyMuPDF 做 PDF 分析。

用 Makefile 命令装依赖：make installdeps

定义要提取的字段

同一类型的文档可能有多种格式。医疗领域里，发票、检验结果、转诊单等的字段名和组织信息的方式可能各不相同。

要处理从文档提取的信息，我们定义字段应如何命名。

先在 src 下建 shared 目录，加 extraction_fields.py 文件：mkdir src/shared 后 touch src/shared/extraction_fields.py。

定义字段常量

向 extraction_fields.py 添加代码。核心是三组常量：

COMMON_FIELDS 列表定义患者通用字段：full_name（患者姓名）、patient_address（患者地址）、social_security_number（社保号）。

SPECIFIC_FIELDS 字典按文档类别定义专属字段——处方类有 doctor_name、medications、prescription_date 等；账单类有 bill_amount、services、care_date；住院报告类有 hospital_name、admission_date、discharge_date、primary_diagnosis；检验分析类有 laboratory、sample_date、abnormal_results；影像类有 exam_type（MRI/CT/X 光）、anatomical_region、conclusion；证明类、互助报销类、社保报销类、门诊报告类、知情同意类各有自己的字段集，other 类为空。

DOCUMENT_CATEGORIES 取 SPECIFIC_FIELDS 的全部键。

CATEGORY_LABELS 给每个类别配图标标签方便前端展示。

这些常量指定提取用的格式，把可能有多种命名和布局的同类输入标准化输出。

构建工作流

字段配置好后，用 Mistral Workflows SDK 搭建工作流。

Workflows 用装饰器定义组件，边建边讲可用的装饰器。

在 workflows 文件夹新建 medical_doc_workflow.py：touch src/workflows/medical_doc_workflow.py。

导入与设置

文件开头是导入和设置：asyncio、logging、os、functools 的 lru_cache、datetime 的 timedelta、typing 的 Optional，加 mistralai.workflows 及其 mistralai 插件、dotenv 的 load_dotenv、pydantic 的 BaseModel/ConfigDict/Field/create_model，再从 shared.extraction_fields 导入三组常量。

调用 load_dotenv 后，把 mistralai_workflows、httpx、httpcore 三个日志器压到 WARNING 级别。

接着定义两个数据模型：ManualCategorySignal 只含 category 字段；DocumentClassification 含 category（描述里列全部分类）、confidence（0 到 1）、explanation 三字段。

然后用 lru_cache 缓存一个工厂函数 get_extraction_output_model：按类别动态创建 Pydantic 模型——common 部分由 COMMON_FIELDS 生成全 Optional 字段、extra 设 forbid；specific 部分按类别从 SPECIFIC_FIELDS 生成；返回把两者嵌套的组合模型。

这是结构化输出的 schema 基础。

创建工作流活动

Workflows 里活动是执行真正计算、API 调用或其他操作的工作单元，用 @workflows.activity 装饰器指定。

默认按 Python 函数名注册。

我们的活动传两个参数：

start_to_close_timeout 设活动从开始到必须返回结果的最长时间，超时则被终止并视为失败（可能触发重试）。

没有超时的话，挂住的活动无限阻塞。

retry_policy_max_attempts 指定活动失败后重试次数。

可用参数见活动基础文档。

我们的工作流包含三个活动：

get_document_signed_url 获取可传给下一活动的文件签名 URL。

classify_document 用 Mistral chat completions 分类文档类型，传入文档类别和文件信息。

extract_patient_info 用 Mistral chat completions 按给定类别从文档提取想要的信息并返回 JSON。

活动一代码：get_document_signed_url 带 5 分钟超时和 2 次重试，内部拿 Mistral 客户端调 get_signed_url_async，把上传文件 ID 换成临时签名 URL，供 Document QnA 使用。

活动二代码：classify_document 带 2 分钟超时和 2 次重试，模型默认 mistral-medium-latest（可用环境变量覆盖），temperature 0.0。

系统提示词写明”你是医疗文档分类专家，只返回匹配 schema 的有效 JSON”；用户消息的 text 部分列出全部候选类别要求精确选一并返回 0 到 1 置信度和简短解释，document_url 部分直接挂签名 URL。

响应经 response_format=DocumentClassification 结构化解析，解析失败抛 RuntimeError。

活动三代码：extract_patient_info 同样 2 分钟超时 2 次重试，按类别取动态提取模型，把通用字段和类别专属字段列进提示词，指示缺失值填 null，经 parse_async 拿到结构化 JSON 后返回。

定义工作流

定义工作流用 @workflows.workflow.define 装饰器传名称。

SDK 提供三个装饰器运行和追踪工作流：

@workflows.workflow.query 用于查询工作流步骤状态。查询用同步通信且只读。

@workflows.workflow.signal 用于标记用户选择了文档类别。

信号用异步通信、可在执行中任意时刻发送。

@workflows.workflow.entrypoint 定义工作流的 run。

它依次走过每个活动、沿途更新步骤直到完成。

核心代码分四块解读：

PdfOcrWorkflow 继承 InteractiveWorkflow。

__init__ 里维护 steps 字典——ocr、classify、extract 各带 status 和 result——经 get_steps 查询暴露给前端；再加一个 _manual_category 暂存人工覆盖。

get_steps 查询方法同步只读返回 steps，前端轮询进度用它。

manual_category_signal 信号方法异步写入，把用户手选类别存进 _manual_category——run 活跃时也能改状态。

run 入口方法签名收 file_id、filename、confidence_threshold（默认 0.9）和可选的 manual_review_timeout_seconds，返回 ChatAssistantWorkflowOutput。

方法体先建三个 TodoListItem（准备文档、分类、提取）让客户端显示粗粒度进度，然后在 TodoList 上下文里顺序执行：OCR 步骤调签名 URL 活动；分类步骤调 classify 活动，拿到置信度后判断——低于阈值就把步骤状态改为 waiting_human 并进入人在环分支：workflows.workflow.wait_condition 确定性地暂停直到信号到达或超时，超时就给 explanation 追加”人工审查超时，使用模型预测类别”继续走，信号按时到达则用人工类别覆盖、置信度改为 1.0；提取步骤按最终类别调 extract 活动。

结束时返回带 content 文本和 structuredContent（filename、classification、patient_info）的输出。

文件尾部是 main 函数：打印 “Worker ready” 后调 workflows.run_worker([PdfOcrWorkflow]) 启动 worker 等任务。

这个 InteractiveWorkflow 编排了三个 Mistral 驱动的阶段：准备文档访问、分类文档类型、提取结构化字段。

创建应用前端

工作流不强制要前端，但演示用途我们加一个。

在 entrypoints 文件夹新建 app.py。

我们建的 Streamlit 应用是处理医疗 PDF 的端到端人在环管线：上传文档、跑 OCR、分类文档类型、提取结构化患者信息；实时显示工作流进度，模型置信度过低时让用户手动确认类别，保证提取对下游可靠。

app.py 的导入和设置部分：streamlit、dotenv、pydantic 加 PyMuPDF（fitz），从 shared.extraction_fields 导入 CATEGORY_LABELS。

API_KEY 从环境变量取，BASE_URL 默认官方 API 地址。

COMMON_FIELD_LABELS 定义患者三字段的中文展示名。STEPS_CONFIG 列出三步的键和标题。

PdfOcrInput 模型含 file_id、filename、confidence_threshold。

接下来是 Streamlit 与 Workflows 的集成层：

get_workflows_client 每次新建客户端——注释强调不要跨 run_async 调用缓存：每次调用创建并关闭自己的事件循环，跨循环复用异步客户端触发 “Event loop is closed” 错误。

run_async 帮手为 Streamlit 的同步模型桥接异步调用：新建事件循环跑完协程，finally 里先排空异步生成器（避免 “Task was destroyed” 错误）再关循环。

upload_pdf 用 Mistral 文件接口把 PDF 字节以 purpose=“ocr” 上传，返回文件 ID。

trigger_workflow 生成带 uuid 的 execution_id，调 execute_workflow_async 触发 pdf_ocr_workflow，传入模型化的输入。

poll_steps 用 query_workflow_execution 调 get_steps 查询拿实时进度。

get_execution_status 和 get_execution_details 分别拿执行状态和完整详情。

send_signal 调 signal_workflow_execution 发 manual_category 信号，载荷就是用户选的类别。

backfill_steps_from_execution_result 从最终结果回填步骤状态：依次尝试 structuredContent、structured_content 键位，把 ocr、classify、extract 标记为 done。

这段代码创建并复用工作流客户端、提供从 Streamlit 同步执行模型跑异步 API 的帮手、上传 PDF 文件、启动新工作流执行、轮询逐步进度、检查整体执行状态、用户覆盖低置信度分类时发送人工类别信号。

完成 Streamlit 应用

补上界面元素。这段代码块很长，但对理解 Mistral Workflows 不重要。

PDF 渲染部分：get_pdf_first_page 用 PyMuPDF 把 PDF 首页转图片，1.5 倍缩放求清晰，失败返回 None。

步骤渲染部分：render_step 按状态分支——pending 显示待处理；running 显示进行中；waiting_human 是关键分支，警告置信度不足、用下拉框列出全部类别让用户选、点 Validate 按钮后调 send_signal 发信号并刷新页面；done 按 ocr/classify/extract 分别渲染——classify 显示类别标签、解释和置信度进度条，extract 把通用和专属字段渲染成表格，空值跳过。

主界面：页面配置宽布局，标题加副标题”上传 PDF → OCR → 分类 → 患者信息提取”。

侧边栏放置信度阈值滑块（0 到 1 步进 0.05），滑到 1.0 提示永远要人工验证，滑到 0 提示从不需要。

会话状态初始化 execution_id、done、steps、poll_error、signal_sent 五个键。

文件上传器限 PDF 类型。

上传后交互流：点 Start Workflow 主按钮时清空会话状态、读文件字节、用 st.status 显示上传进度、调 upload_pdf 拿 file_id、调 trigger_workflow 拿 execution_id、刷新页面。

有 execution_id 且未完成时：轮询 poll_steps 更新状态，左右分栏——左边渲染 PDF 首页预览，右边逐个渲染三步卡片；全部 done 就标记完成显示成功，waiting_human 且信号未发就等用户操作，发过就 0.5 秒后刷新；其余情况查执行状态，COMPLETED 就用 backfill 回填并显示成功，FAILED/CANCELED/TERMINATED 报错，还在跑就 0.5 秒后自刷新重轮询。

已完成的会话重新渲染结果表格。

这段代码实现应用的前端行为：渲染 PDF 预览、在界面显示每个工作流步骤、按步骤状态（pending、running、waiting_human、done）更新用户所见。

还在分类置信度低时处理人工审查、把提取的通用/专属字段展示成表格、管理 Streamlit 会话状态和重跑，让界面持续反映工作流实时进度直到完成或失败。

加负载测试

创建负载测试机制来演示 Mistral Workflows 的一个关键优势：并发跑大量长时、AI 密集任务而无需自己管队列或线程。

它从 input_doc 文件夹读一个或多个 PDF，上传到 Mistral Files API，再用几行异步代码并行触发指定数量的工作流执行。

在 entrypoints 目录新建 load_test.py。

用 workflows.execute_workflow_async() 函数并行启动指定数量的工作流，默认 10 个。

负载测试脚本：并行启动 10 个工作流演示负载均衡。每个工作流会被分发到可用 worker。

跨行业复用要点

这条管线的模式可以平移到任何”文档进、结构化数据出”的场景：把 CATEGORY_LABELS 换成合同类型、发票类别或简历板块，SPECIFIC_FIELDS 换成对应字段，三步活动结构原样保留。

置信度阈值是精度与人工成本的旋钮：医疗场景建议 0.9 起步，错分代价低的场景可以降到 0.7 减少人工介入。

wait_condition 加信号的人在环模式适用于一切”低置信度必须人拍板”的 AI 流程——审批、风控、质检通用。

原文信息

• 作者：MistralDevs（@MistralDevs），Mistral AI 官方开发者账号
• 发布时间：2026-06-29 原文地址：