当前位置：首页 > news >正文

程序员必备技能：使用本地LLM提取非结构化医疗数据，收藏这篇就够了

news 2026/3/26 18:14:55

本文介绍使用本地部署的LLM(vLLM/Ollama)替代传统正则表达式，从非结构化医疗病历中提取结构化信息。通过Pandas、Pydantic和Instructor工具实现数据读取、字段定义和JSON转换，能理解语义而非简单匹配。该方法保证数据安全且零API成本，文章提供了完整代码实现和推荐模型，帮助开发者高效处理医疗文本数据。

以前我试图用正则表达式（Regex）提取信息。但在复杂的病历面前，正则就有点力不从心。“病人无过敏史”和“病人对青霉素过敏”，关键词都有“过敏”，正则很难分辨语义。一旦文本格式稍微变动，写好的代码就全废了，特别是在跨地域、跨文化等多中心的研究项目背景下。然而80% 的高价值数据都是“非结构化”的。它们躺在 PDF 里、聊天记录里、医生手写的病历里。对于计算机来说，这些不是数据，只是“噪点”。为了分析它们，我们过去只能靠人工——盯着屏幕，Ctrl+C，Ctrl+V，直到眼花缭乱。

通过LLM等方法提取信息能替代这部分繁琐、重复的工作。大语言模型的出现，不是为了陪你聊天，而是为了理解。我构建了一个基于本地 vLLM/Ollama 的工具，它不像传统程序那样“匹配字符”，而是像人类专家一样“阅读文本”。

它能读懂“血压偏高”意味着数值异常。
它能从几千字的病程记录中，精准抓取“阿司匹林”归类为药物，抓取“头晕”归类为症状。
这不是简单的抓取，这是认知重组。

通过 vLLM 在本地构建推理服务，数据不出内网，显卡就在手边。既享受了 AI 的智力，又守住了数据的边界。零API成本，无限次调用，这才是LLM落地应用的正确打开方式。

以下是如何进行提取的核心流程，本案例基于病历数据，其他的数据可以参考，我认为方法都是一样的，就是需求有所变化：

Pandas: 读取 Excel 文件，其他的文件格式用类似的方法读入即可。
Pydantic: 定义你想要提取的病历字段（如：主诉、诊断、用药）。
Instructor + Local LLM: 强制模型将非结构化文本转换为符合定义的 JSON。
Loop: 遍历每一行数据进行处理。
Pandas: 将提取结果与原始 ID 合并，保存为新文件。

第一步：准备环境，必要的环境的不啰嗦了

pip install pandas openpyxl openai instructor pydantic tqdm

第二步：编写批量处理脚本

在这个脚本中，我模拟了一个常见的病历提取需求（提取诊断、症状、药物）。你可以根据实际需求修改 MedicalRecord 类中的字段。

创建文件 batch_extract.py:

import pandas as pd import instructor from openai import OpenAI from pydantic import BaseModel, Field from typing import List, Optional from tqdm import tqdm # 进度条库 # ================= 配置区域 ================= # 1. 设置输入输出文件路径 INPUT_FILE = "raw_data.xlsx" # 你的原始文件 OUTPUT_FILE = "structured_data.xlsx" # 输出结果文件 # 2. 连接本地 LLM (vLLM 或 Ollama) # vLLM 默认通常是 http://localhost:8000/v1 # Ollama 默认通常是 http://localhost:11434/v1 client = instructor.from_openai( OpenAI( base_url="http://localhost:11434/v1", api_key="ollama" ), mode=instructor.Mode.JSON, ) MODEL_NAME = "qwen2.5:14b" # 建议使用通义千问等中文能力强的模型 # ================= 定义提取结构 ================= # 这是最重要的一步：定义你想从病历中挖出什么信息，这部分根据自己的需求进行设定 class MedicalRecord(BaseModel): patient_id: Optional[str] = Field(description="病人ID，如果在文本中未提及则留空") diagnosis: List[str] = Field(description="确诊的疾病名称列表") symptoms: List[str] = Field(description="病人主诉及症状表现") medications: List[str] = Field(description="医生开具的药物名称列表") surgeries: List[str] = Field(description="曾做过的手术名称，无则为空列表") allergies: Optional[str] = Field(description="过敏史，如果未提及填'无'") # 你可以继续添加字段，比如 'admission_date', 'doctor_advice' 等 # ================= 核心处理逻辑 ================= def process_medical_text(text: str): """调用 LLM 进行提取""" if not text or pd.isna(text): return None try: resp = client.chat.completions.create( model=MODEL_NAME, response_model=MedicalRecord, messages=[ { "role": "system", "content": "你是一个专业的医疗数据结构化助手。请分析病历文本，提取关键医疗实体。保持客观，不要推测。" }, { "role": "user", "content": f"病历内容：\n{text}" }, ], max_retries=2, # 如果生成的 JSON 格式不对，自动重试 2 次 ) return resp.model_dump() # 转为字典格式 except Exception as e: print(f"Error processing row: {e}") return None def main(): # 1. 读取 Excel (假设没有表头，或者第一行是表头) # header=0 表示第一行是列名，如果是纯数据没有列名，用 header=None print(f"正在读取 {INPUT_FILE}...") df = pd.read_excel(INPUT_FILE) # 假设 Excel 结构：第一列是 ID，第二列是 文本 # 我们为了通用，直接使用 iloc 按位置索引 # col 0: ID, col 1: Context extracted_results = [] # 2. 使用 tqdm 显示进度条进行循环 print("开始提取数据...") for index, row in tqdm(df.iterrows(), total=df.shape[0]): row_id = row.iloc[0] # 第一列 ID raw_text = row.iloc[1] # 第二列 文本 # 调用 LLM data = process_medical_text(raw_text) if data: # 将原始 ID 放进去，确保一一对应 data['original_id'] = row_id extracted_results.append(data) else: # 处理失败或空行的情况 extracted_results.append({ 'original_id': row_id, 'error': '提取失败或内容为空' }) # 3. 转换为 DataFrame 并保存 print("提取完成，正在保存...") result_df = pd.DataFrame(extracted_results) # 调整列顺序，把 ID 放在第一列 cols = ['original_id'] + [c for c in result_df.columns if c != 'original_id'] result_df = result_df[cols] result_df.to_excel(OUTPUT_FILE, index=False) print(f"文件已保存至: {OUTPUT_FILE}") if __name__ == "__main__": main()

推荐模型：

Qwen2.5 (14B 或 32B): 通义千问在中文医疗实体抽取上表现非常好，且指令跟随能力强。
HuatuoGPT: 如果你有能力加载微调过的医疗模型。
Llama3-70B: 逻辑能力最强。
…

提示词工程 (System Prompt)

在代码中的 messages 部分，建议根据你的具体病历类型进行微调。例如：

“你是一个三甲医院的病案整理专家。请从杂乱的文本中提取结构化信息。对于药品名称，请统一转换为通用名。如果没有提及某项信息，请严格返回空列表，严禁编造。”

如果使用的是vLLM:

vLLM 最大的优势是支持高并发吞吐。你可以修改 Python 脚本使用 asyncio 并发发送请求，而不是等一条处理完再发下一条。

import asyncio async def process_batch(rows): tasks = [process_single_row_async(row) for row in rows] return await asyncio.gather(*tasks) # 这样可以一次性发 10-20 个请求给 vLLM，速度会快 10 倍以上

如果 Excel 有 10000000000 行，跑到第 9000000 行程序崩了怎么办？
建议：