Google LangExtract 深度解析：从非结构化文本到结构化知识的工程化桥梁——零微调实现 100% 可溯源的信息提取

引言：为什么结构化信息提取如此重要？

在 AI 大模型时代，我们面临一个看似矛盾的现象：LLM（大语言模型）已经能够理解几乎任何形式的文本，但当企业真正需要将医疗病历、法律合同、财务报告转化为可查询、可分析的结构化数据时，却发现传统的"直接问 LLM"方案远远不够。

问题在于：LLM 的输出不可溯源。

当你问 ChatGPT "从这份病历中提取患者的药物过敏史"，模型可能会返回一个药物列表，但你无法验证这些信息究竟来自文档的哪一行、哪一个词。对于医疗、法律、金融等对数据质量和合规性要求极高的领域，这种"黑箱式"的提取结果无法被信任。

2026 年 4 月，Google 开源了 LangExtract，一个基于 LLM 的结构化信息提取框架。它解决了两个核心痛点：

1. 精准溯源：每个提取结果都精确映射到原文中的字符位置，实现 100% 可溯源
2. 零微调：无需训练模型，只需提供 3-5 个标注示例即可适配任何领域

这不是一个需要部署 GPU 集群的深度学习系统，而是一个基于提示工程（Prompt Engineering）和检索增强（RAG）的轻量级框架。它的工程价值在于：将 LLM 的灵活性与传统信息提取的可靠性结合起来。

本文将深入剖析 LangExtract 的技术架构、核心算法、实战用法，以及它如何改变医疗、法律、金融等领域的信息处理范式。

一、结构化信息提取的演进：从规则到 LLM

1.1 传统方法的困境

在 LLM 出现之前，结构化信息提取经历了三代技术演进：

第一代：基于规则的方法（1990s-2000s）

# 典型的正则表达式提取
import re

text = "患者张三，男，45岁，诊断为高血压"
name_pattern = r"患者([^，]+)"
age_pattern = r"(\d+)岁"
diagnosis_pattern = r"诊断为([^，。]+)"

name = re.search(name_pattern, text).group(1)  # 张三
age = re.search(age_pattern, text).group(1)    # 45
diagnosis = re.search(diagnosis_pattern, text).group(1)  # 高血压

优点：结果 100% 可溯源，完全可控。

缺点：

• 需要针对每个新场景编写大量规则
• 无法处理语义变化（如"确诊为"、"诊断结果为"）
• 维护成本极高，规则之间容易冲突

第二代：统计机器学习方法（2000s-2010s）

基于 CRF（条件随机场）、SVM 等算法的命名实体识别（NER）：

# spaCy 典型用法（预训练模型）
import spacy

nlp = spacy.load("en_core_web_sm")
doc = nlp("Apple was founded by Steve Jobs in California")

for ent in doc.ents:
    print(f"{ent.text}: {ent.label_} ({ent.start_char}-{ent.end_char})")
    # Apple: ORG (0-5)
    # Steve Jobs: PERSON (21-31)
    # California: GPE (35-45)

优点：能够泛化到未见过的表达方式。

缺点：

• 需要大量标注数据（通常数千至数万条）
• 实体类型固定，难以扩展
• 对于跨领域迁移，性能急剧下降

第三代：深度学习方法（2015-2022）

基于 BERT、GPT 等预训练模型的微调：

# Hugging Face Transformers 典型用法
from transformers import pipeline

ner = pipeline("ner", model="dslim/bert-base-NER")
results = ner("Apple is looking at buying U.K. startup for $1 billion")

for r in results:
    print(f"{r['word']}: {r['entity']} (score: {r['score']:.2f})")

优点：语义理解能力强，准确率高。

缺点：

• 微调需要大量标注数据和计算资源
• 模型部署成本高
• 最大问题：输出仍然不可溯源，无法知道模型为什么做出这个判断

1.2 LLM 时代的机遇与挑战

GPT-4、Gemini 等大模型的出现，让"直接问模型"成为可能：

# 直接问 LLM
import openai

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{
        "role": "user",
        "content": """从以下病历中提取患者信息：
        患者李四，男，52岁，2024年3月15日入院。
        主诉：胸闷气短一周。
        既往史：糖尿病10年，高血压5年。
        过敏史：青霉素过敏。
        请以JSON格式返回。"""
    }]
)

优势：无需标注数据，灵活适应任何领域。

致命问题：

1. 幻觉：模型可能编造不存在的信息
2. 不可溯源：无法验证输出结果来自原文何处
3. 格式不稳定：JSON 输出可能格式错误
4. 长文档困难：超出上下文窗口的文档无法处理

LangExtract 的设计目标，就是解决这四个问题。

二、LangExtract 核心架构：四层设计哲学

LangExtract 的架构可以分为四层：

┌─────────────────────────────────────────────────────────┐
│                    用户层 (User Layer)                   │
│  prompt_description + examples + input_text             │
├─────────────────────────────────────────────────────────┤
│                  编排层 (Orchestration Layer)             │
│  分块策略 → 并行调度 → 结果聚合 → 冲突解决                │
├─────────────────────────────────────────────────────────┤
│                   模型层 (Model Layer)                   │
│  Gemini / GPT / Claude / Ollama (本地模型)               │
├─────────────────────────────────────────────────────────┤
│                  验证层 (Validation Layer)               │
│  Source Grounding → Schema 校验 → 可视化生成             │
└─────────────────────────────────────────────────────────┘

2.1 用户层：Schema-Driven 的交互设计

LangExtract 的核心理念是 "Schema-Driven Extraction"。用户不需要写代码定义提取逻辑，而是通过自然语言描述 + 少量示例来定义提取规则。

核心数据结构：

import langextract as lx
from langextract.data import ExampleData, Extraction

# 定义一个提取任务示例
example = ExampleData(
    text="ROMEO. But soft! What light through yonder window breaks?",
    extractions=[
        Extraction(
            extraction_class="character",       # 实体类型
            extraction_text="ROMEO",           # 提取的原文（必须精确匹配）
            attributes={                        # 附加属性
                "emotional_state": "wonder"
            }
        ),
        Extraction(
            extraction_class="emotion",
            extraction_text="But soft!",
            attributes={"feeling": "gentle awe"}
        )
    ]
)

关键设计点：

1. extraction_text 必须是原文中的精确片段（字符级匹配），这是实现溯源的基础
2. extraction_class 定义实体类型，可以是任意字符串
3. attributes 允许添加 LLM 推断的额外信息（这部分可以不是原文）

用户定义提取任务的完整流程：

import textwrap

# 1. 定义提取任务的描述
prompt = textwrap.dedent("""\
Extract characters, emotions, and relationships in order of appearance.
Use exact text for extractions. Do not paraphrase or overlap entities.
Provide meaningful attributes for each entity to add context.""")

# 2. 提供 3-5 个高质量示例
examples = [
    ExampleData(
        text="ROMEO. But soft! What light through yonder window breaks? It is the east, and Juliet is the sun.",
        extractions=[
            Extraction("character", "ROMEO", {"emotional_state": "wonder"}),
            Extraction("emotion", "But soft!", {"feeling": "gentle awe"}),
            Extraction("relationship", "Juliet is the sun", {"type": "metaphor"})
        ]
    ),
    # ... 更多示例
]

# 3. 提供输入文本
input_text = "Lady Juliet gazed longingly at the stars, her heart aching for Romeo"

2.2 编排层：长文档的处理策略

LangExtract 的一个核心创新是解决了"大海捞针"（Needle-in-a-Haystack）问题：当文档长度超过 LLM 的上下文窗口，或者信息分散在文档的各个角落时，如何保证提取的召回率？

三阶段处理流程：

原始长文档 (100+ pages)
       │
       ▼
┌──────────────────┐
│ 第一阶段：分块   │  按语义边界切分为 chunks
│ (Chunking)       │  每个 chunk ~2000 tokens
└──────────────────┘
       │
       ▼
┌──────────────────┐
│ 第二阶段：并行   │  每个 chunk 独立调用 LLM
│ (Parallel)       │  支持异步 + 重试
└──────────────────┘
       │
       ▼
┌──────────────────┐
│ 第三阶段：聚合   │  去重 + 合并 + 排序
│ (Aggregation)    │  处理跨 chunk 的实体
└──────────────────┘
       │
       ▼
最终提取结果 (带完整溯源信息)

分块策略的核心代码：

# LangExtract 内部的分块逻辑（简化版）
def chunk_text(text: str, max_tokens: int = 2000) -> list[str]:
    """
    语义感知的分块策略：
    1. 优先在段落边界切分
    2. 避免切分表格、代码块等结构化内容
    3. 保留上下文重叠（overlap）以提高跨块实体识别率
    """
    paragraphs = text.split("\n\n")
    chunks = []
    current_chunk = []
    current_length = 0
    
    for para in paragraphs:
        para_tokens = estimate_tokens(para)  # ≈ len(para) / 4
        
        if current_length + para_tokens > max_tokens:
            # 当前块满了，保存并开始新块
            if current_chunk:
                chunks.append("\n\n".join(current_chunk))
            current_chunk = [para]
            current_length = para_tokens
        else:
            current_chunk.append(para)
            current_length += para_tokens
    
    # 处理最后一块
    if current_chunk:
        chunks.append("\n\n".join(current_chunk))
    
    return chunks

并行调度的优化：

import asyncio
from langextract import extraction_engine

async def parallel_extract(chunks: list[str], model_id: str, rate_limit: int):
    """
    并行提取调度器：
    1. 支持 API 速率限制（避免 429 错误）
    2. 自动重试失败请求
    3. 实时进度报告
    """
    semaphore = asyncio.Semaphore(rate_limit)  # 控制并发数
    
    async def extract_chunk(chunk: str, idx: int):
        async with semaphore:
            try:
                result = await extraction_engine.extract_async(
                    text=chunk,
                    model_id=model_id,
                    max_retries=3
                )
                return (idx, result, None)
            except Exception as e:
                return (idx, None, str(e))
    
    # 并发执行所有块
    tasks = [extract_chunk(chunk, i) for i, chunk in enumerate(chunks)]
    results = await asyncio.gather(*tasks)
    
    # 按原始顺序排序
    sorted_results = sorted(results, key=lambda x: x[0])
    
    # 检查错误
    for idx, result, error in sorted_results:
        if error:
            print(f"Chunk {idx} failed: {error}")
    
    return [r for _, r, _ in sorted_results if r is not None]

2.3 模型层：多后端支持的抽象设计

LangExtract 支持多种 LLM 后端，通过统一的抽象接口实现：

from abc import ABC, abstractmethod
from typing import Generator

class BaseModelProvider(ABC):
    """模型提供者的抽象基类"""
    
    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """同步生成"""
        pass
    
    @abstractmethod
    async def generate_async(self, prompt: str, **kwargs) -> str:
        """异步生成"""
        pass
    
    @abstractmethod
    def stream_generate(self, prompt: str, **kwargs) -> Generator[str, None, None]:
        """流式生成"""
        pass

支持的模型后端：

Gemini 模型选择建议：

# 推荐配置
model_id = "gemini-2.5-flash"  # 默认，平衡速度与质量
# model_id = "gemini-2.5-pro"  # 复杂任务，更强推理能力

result = lx.extract(
    text_or_documents=input_text,
    prompt_description=prompt,
    examples=examples,
    model_id=model_id,
)

# 生产环境建议使用 Tier 2 配额以避免速率限制
# 参考：https://ai.google.dev/gemini-api/docs/rate-limits#tier-2

本地模型支持（Ollama）：

# 使用本地 LLM（隐私保护）
# 前置步骤：安装 Ollama 并下载模型
# $ ollama pull llama3.2
# $ ollama serve

result = lx.extract(
    text_or_documents=input_text,
    prompt_description=prompt,
    examples=examples,
    model_id="ollama://llama3.2",  # 通过 ollama:// 前缀指定
)

2.4 验证层：Source Grounding 的核心技术

这是 LangExtract 最具创新性的部分。传统的 LLM 信息提取是"黑箱"，而 LangExtract 实现了 字符级溯源（Character-Level Source Grounding）。

核心数据结构：

from dataclasses import dataclass
from typing import Optional

@dataclass
class CharInterval:
    """字符区间：[start, end)，左闭右开"""
    start: int
    end: int
    
    def extract_text(self, full_text: str) -> str:
        return full_text[self.start:self.end]

@dataclass  
class Extraction:
    extraction_class: str           # 实体类型
    extraction_text: str           # 提取的文本
    char_interval: Optional[CharInterval]  # 字符区间（核心）
    attributes: dict               # 附加属性
    
    def is_grounded(self) -> bool:
        """验证是否可溯源（文本是否精确匹配）"""
        return self.char_interval is not None
    
    def verify(self, full_text: str) -> bool:
        """验证提取文本与原文是否一致"""
        if not self.char_interval:
            return False
        extracted = self.char_interval.extract_text(full_text)
        return extracted == self.extraction_text

溯源验证的完整流程：

LLM 输出：
{
  "extraction_class": "medication",
  "extraction_text": "Aspirin 100mg",
  "attributes": {"dosage": "once daily"}
}
        │
        ▼
┌──────────────────────────────┐
│ 在原文中搜索 "Aspirin 100mg"  │
│ 找到位置：char_start=245     │
│           char_end=259       │
└──────────────────────────────┘
        │
        ▼
验证：text[245:259] == "Aspirin 100mg" ?
        │
   ┌────┴────┐
   │ Yes      │ No
   ▼          ▼
创建 CharInterval   char_interval = None
(245, 259)          标记为不可信
        │
        ▼
返回带溯源信息的 Extraction 对象

处理 LLM 幻觉的机制：

# 过滤掉不可信的提取结果
grounded_extractions = [
    e for e in result.extractions 
    if e.char_interval is not None
]

# 统计可信度
total = len(result.extractions)
grounded = len(grounded_extractions)
print(f"可信度: {grounded/total*100:.1f}%")

# 输出示例：
# 可信度: 94.2%  （5.8% 的提取无法在原文中定位，可能是幻觉）

三、实战：从医疗病历提取结构化信息

让我们通过一个完整的医疗信息提取案例来演示 LangExtract 的实际应用。

3.1 场景描述

假设我们有一批电子病历（EHR），需要提取：

• 患者基本信息（姓名、年龄、性别）
• 诊断信息（主诊断、合并症）
• 用药信息（药物名称、剂量、频次）
• 过敏史（过敏原、反应类型）

样本病历：

患者姓名：张三
性别：男
年龄：65岁
住院号：20240315001
入院日期：2024-03-15

主诉：胸闷气短一周，加重三天。

现病史：患者一周前无明显诱因出现胸闷、气短，活动后加重，
休息后可缓解。近三天症状加重，夜间不能平卧，端坐呼吸。
伴双下肢水肿，尿量减少。

既往史：
1. 高血压病史15年，长期服用氨氯地平 5mg qd
2. 2型糖尿病10年，服用二甲双胍 500mg tid
3. 冠心病5年，行PCI术，服用阿司匹林 100mg qd、氯吡格雷 75mg qd

过敏史：青霉素过敏（表现为皮疹），磺胺类药物过敏（表现为过敏性休克）

诊断：
1. 心力衰竭（NYHA III级）
2. 冠状动脉粥样硬化性心脏病
3. 高血压病3级（极高危）
4. 2型糖尿病

医嘱：
1. 呋塞米 20mg iv qd
2. 螺内酯 20mg po qd  
3. 培哚普利 4mg po qd
4. 美托洛尔缓释片 23.75mg po qd

3.2 定义提取 Schema

import langextract as lx
from langextract.data import ExampleData, Extraction

# 定义提取任务的描述
medical_prompt = """
Extract patient information from medical records:
- patient_info: name, age, gender, admission date
- diagnosis: all diagnoses with ICD-10 codes if mentioned
- medication: drug name, dosage, frequency, route
- allergy: allergen and reaction type

Use EXACT text from the document for extractions.
List medications in order of appearance.
For each allergy, include reaction severity.
"""

# 定义示例（few-shot learning）
medical_examples = [
    ExampleData(
        text="""患者李四，女，58岁。
        诊断：高血压病
        用药：厄贝沙坦 150mg qd
        过敏史：无""",
        extractions=[
            Extraction("patient_info", "李四", {"gender": "女", "age": "58"}),
            Extraction("diagnosis", "高血压病", {"icd10": "I10"}),
            Extraction("medication", "厄贝沙坦 150mg qd", 
                      {"drug": "厄贝沙坦", "dosage": "150mg", "frequency": "qd"}),
        ]
    ),
    # 可以添加更多示例以提高准确率
]

3.3 执行提取

# 读取病历
with open("medical_record.txt", "r", encoding="utf-8") as f:
    record_text = f.read()

# 执行提取
result = lx.extract(
    text_or_documents=record_text,
    prompt_description=medical_prompt,
    examples=medical_examples,
    model_id="gemini-2.5-flash",
)

# 输出结果
print(f"提取实体数: {len(result.extractions)}")
for ext in result.extractions:
    if ext.char_interval:
        print(f"[{ext.extraction_class}] {ext.extraction_text}")
        print(f"  位置: {ext.char_interval.start}-{ext.char_interval.end}")
        print(f"  属性: {ext.attributes}")

输出示例：

提取实体数: 12

[patient_info] 张三
  位置: 5-7
  属性: {'gender': '男', 'age': '65', 'admission_date': '2024-03-15'}

[diagnosis] 心力衰竭（NYHA III级）
  位置: 312-325
  属性: {'icd10': 'I50.9', 'type': 'primary'}

[medication] 氨氯地平 5mg qd
  位置: 180-192
  属性: {'drug': '氨氯地平', 'dosage': '5mg', 'frequency': 'qd', 'route': 'oral'}

[allergy] 青霉素过敏（表现为皮疹）
  位置: 245-261
  属性: {'allergen': '青霉素', 'reaction': '皮疹', 'severity': 'moderate'}

[allergy] 磺胺类药物过敏（表现为过敏性休克）
  位置: 263-287
  属性: {'allergen': '磺胺类药物', 'reaction': '过敏性休克', 'severity': 'severe'}

3.4 可视化验证

LangExtract 的杀手锏功能是 交互式 HTML 可视化：

# 保存结果到 JSONL
lx.io.save_annotated_documents(
    [result], 
    output_name="medical_extraction.jsonl",
    output_dir="./output"
)

# 生成可视化 HTML
html_content = lx.visualize("output/medical_extraction.jsonl")

with open("medical_visualization.html", "w", encoding="utf-8") as f:
    if hasattr(html_content, 'data'):
        f.write(html_content.data)
    else:
        f.write(html_content)

print("可视化文件已生成: medical_visualization.html")

打开生成的 HTML 文件，你会看到：

┌─────────────────────────────────────────────────────────┐
│  LangExtract Visualization                               │
├─────────────────────────────────────────────────────────┤
│  [实体类型过滤器] [置信度阈值] [搜索框]                    │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  患者姓名：[张三] ← 鼠标悬停显示属性                        │
│  性别：男                                                │
│  年龄：[65岁]                                            │
│  ...                                                    │
│                                                         │
│  过敏史：[青霉素过敏（表现为皮疹）] ← 点击跳转到原文位置      │
│          [磺胺类药物过敏（表现为过敏性休克）] ← 高亮显示      │
│                                                         │
└─────────────────────────────────────────────────────────┘

每个提取结果都被高亮显示，点击即可跳转到原文中的精确位置。这对于医学审核人员验证提取结果至关重要。

3.5 批量处理大规模病历

import glob
import json

# 批量处理 1000+ 份病历
record_files = glob.glob("./records/*.txt")
all_results = []

for file_path in record_files:
    with open(file_path, "r", encoding="utf-8") as f:
        text = f.read()
    
    result = lx.extract(
        text_or_documents=text,
        prompt_description=medical_prompt,
        examples=medical_examples,
        model_id="gemini-2.5-flash",
    )
    
    # 添加文件元数据
    result.metadata = {"file_path": file_path}
    all_results.append(result)

# 批量保存
lx.io.save_annotated_documents(
    all_results,
    output_name="batch_medical_extraction.jsonl",
    output_dir="./output"
)

# 统计分析
total_entities = sum(len(r.extractions) for r in all_results)
grounded_entities = sum(
    1 for r in all_results 
    for e in r.extractions 
    if e.char_interval is not None
)

print(f"总实体数: {total_entities}")
print(f"可信实体: {grounded_entities} ({grounded_entities/total_entities*100:.1f}%)")

四、高级特性：长文档处理与多轮提取

4.1 大海捞针问题的解决

当处理 100 页以上的医疗记录或法律合同时，简单的分块策略会丢失跨块信息。LangExtract 采用了 多轮提取策略：

# 第一轮：粗粒度分块提取
def coarse_extraction(long_document: str, chunk_size: int = 10000) -> list:
    """粗粒度提取：快速扫描全文档"""
    chunks = split_into_chunks(long_document, chunk_size)
    all_extractions = []
    
    for chunk in chunks:
        result = lx.extract(
            text_or_documents=chunk,
            prompt_description="快速识别所有实体名称和位置",
            examples=coarse_examples,
            model_id="gemini-2.5-flash",
        )
        all_extractions.extend(result.extractions)
    
    # 去重
    unique_extractions = deduplicate(all_extractions)
    return unique_extractions

# 第二轮：精确定位
def fine_extraction(long_document: str, 
                    coarse_results: list, 
                    context_window: int = 500) -> list:
    """精细提取：对每个粗结果进行上下文扩展"""
    refined_extractions = []
    
    for ext in coarse_results:
        # 提取上下文窗口
        start = max(0, ext.char_interval.start - context_window)
        end = min(len(long_document), ext.char_interval.end + context_window)
        context = long_document[start:end]
        
        # 精细提取
        result = lx.extract(
            text_or_documents=context,
            prompt_description="提取实体的详细属性",
            examples=fine_examples,
            model_id="gemini-2.5-pro",  # 使用更强的模型
        )
        
        # 调整字符偏移
        for r in result.extractions:
            if r.char_interval:
                r.char_interval.start += start
                r.char_interval.end += start
            refined_extractions.append(r)
    
    return refined_extractions

4.2 处理表格和结构化数据

医疗记录通常包含复杂的表格结构。LangExtract 可以通过定制 Prompt 来处理：

table_prompt = """
Extract information from medical tables:
- For each table row, extract all column values
- Maintain table structure in output
- Handle merged cells and nested tables

Table format:
| 药品名称 | 剂量 | 用法 | 频次 |
|---------|------|-----|------|
"""

# 预处理：将表格转换为 Markdown 格式
def extract_table_info(text_with_tables: str) -> list:
    # 假设已经识别出表格区域
    table_sections = identify_tables(text_with_tables)
    
    results = []
    for table in table_sections:
        result = lx.extract(
            text_or_documents=table,
            prompt_description=table_prompt,
            examples=table_examples,
            model_id="gemini-2.5-flash",
        )
        results.extend(result.extractions)
    
    return results

五、与其他工具的对比分析

5.1 LangExtract vs spaCy

5.2 LangExtract vs 直接问 LLM

5.3 LangExtract vs 传统 OCR + NER

传统流程：扫描件 → OCR → NER → 规则提取 → 人工校验
                   ↓
                多处可能出错，无法定位错误源头

LangExtract 流程：文档 → LLM 提取 → 自动验证 → 可视化校验
                              ↓
                        每个结果可溯源到原始字符位置

六、生产部署的最佳实践

6.1 API 速率限制处理

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class RateLimitedExtractor:
    def __init__(self, requests_per_minute: int = 60):
        self.rpm = requests_per_minute
        self.semaphore = asyncio.Semaphore(requests_per_minute)
    
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=4, max=60)
    )
    async def extract_with_retry(self, text: str, **kwargs):
        async with self.semaphore:
            return await lx.extract_async(
                text_or_documents=text,
                **kwargs
            )
    
    async def batch_extract(self, texts: list[str], **kwargs):
        tasks = [self.extract_with_retry(t, **kwargs) for t in texts]
        return await asyncio.gather(*tasks)

6.2 成本优化策略

def optimize_cost_estimate(text_length: int, model: str) -> float:
    """
    估算 API 调用成本
    
    Gemini 2.5 Flash: $0.000125 / 1K input tokens
    Gemini 2.5 Pro:  $0.00125 / 1K input tokens
    
    经验公式：prompt_tokens ≈ text_length / 2 + system_prompt_tokens
    """
    input_tokens = text_length / 4 + 500  # 中文字符约等于 0.5 tokens
    
    if model == "gemini-2.5-flash":
        return input_tokens * 0.000125 / 1000
    elif model == "gemini-2.5-pro":
        return input_tokens * 0.00125 / 1000
    else:
        return 0

# 使用示例
doc_length = 10000  # 10,000 字符
cost_flash = optimize_cost_estimate(doc_length, "gemini-2.5-flash")
cost_pro = optimize_cost_estimate(doc_length, "gemini-2.5-pro")

print(f"Flash 成本: ${cost_flash:.6f}")
print(f"Pro 成本: ${cost_pro:.6f}")

6.3 准确率评估

def evaluate_extraction_accuracy(
    predictions: list[Extraction],
    ground_truth: list[Extraction],
    tolerance: int = 2  # 允许的字符偏移误差
) -> dict:
    """
    评估提取准确率
    
    指标：
    - Precision: 正确提取数 / 总提取数
    - Recall: 正确提取数 / 真实实体数
    - F1: 2 * P * R / (P + R)
    """
    true_positives = 0
    
    for pred in predictions:
        for gt in ground_truth:
            # 检查是否匹配（类型 + 文本 + 位置）
            if (pred.extraction_class == gt.extraction_class and
                pred.extraction_text == gt.extraction_text and
                abs(pred.char_interval.start - gt.char_interval.start) <= tolerance):
                true_positives += 1
                break
    
    precision = true_positives / len(predictions) if predictions else 0
    recall = true_positives / len(ground_truth) if ground_truth else 0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
    
    return {
        "precision": precision,
        "recall": recall,
        "f1": f1,
        "true_positives": true_positives,
        "false_positives": len(predictions) - true_positives,
        "false_negatives": len(ground_truth) - true_positives
    }

七、真实应用场景

7.1 医疗：药物相互作用检测

# 从多份病历中提取用药信息，检测潜在相互作用
def detect_drug_interactions(patient_records: list[str]) -> list[dict]:
    all_medications = []
    
    for record in patient_records:
        result = lx.extract(
            text_or_documents=record,
            prompt_description="提取所有药物名称、剂量、给药途径",
            examples=medication_examples,
            model_id="gemini-2.5-flash",
        )
        
        for ext in result.extractions:
            if ext.extraction_class == "medication":
                all_medications.append({
                    "drug": ext.attributes.get("drug"),
                    "dosage": ext.attributes.get("dosage"),
                    "source": ext.char_interval.extract_text(record)
                })
    
    # 调用药物相互作用数据库
    interactions = check_ddi_database(all_medications)
    return interactions

7.2 法律：合同条款提取

legal_prompt = """
Extract contract clauses:
- party: names and roles of all parties
- term: duration, start date, end date
- payment: amount, currency, due dates
- liability: limitation clauses, indemnification
- termination: conditions and notice periods

Quote exact text for each extraction.
"""

def extract_contract_terms(contract_text: str) -> dict:
    result = lx.extract(
        text_or_documents=contract_text,
        prompt_description=legal_prompt,
        examples=legal_examples,
        model_id="gemini-2.5-pro",  # 法律文档建议用更强的模型
    )
    
    # 按条款类型组织
    clauses = {}
    for ext in result.extractions:
        clause_type = ext.extraction_class
        if clause_type not in clauses:
            clauses[clause_type] = []
        clauses[clause_type].append({
            "text": ext.extraction_text,
            "location": ext.char_interval,
            "attributes": ext.attributes
        })
    
    return clauses

7.3 金融：财务报表解析

financial_prompt = """
Extract financial metrics from reports:
- revenue: amount, period, segment
- expense: category, amount, period
- asset: type, value, depreciation
- liability: type, amount, due_date
- ratio: name, value, calculation_basis

Extract exact numbers and percentages.
"""

def extract_financial_data(report_text: str) -> list[dict]:
    result = lx.extract(
        text_or_documents=report_text,
        prompt_description=financial_prompt,
        examples=financial_examples,
        model_id="gemini-2.5-flash",
    )
    
    # 验证数值提取的准确性
    validated_data = []
    for ext in result.extractions:
        # 检查提取的数值是否在原文中存在
        if ext.char_interval:
            original_text = ext.char_interval.extract_text(report_text)
            # 检查数值一致性
            if validate_number_extraction(ext.attributes, original_text):
                validated_data.append(ext)
    
    return validated_data

八、局限性与未来展望

8.1 当前局限

1. 依赖 LLM API：云模型存在隐私和延迟问题
2. 长文档成本：100 页文档可能需要数十次 API 调用
3. 表格处理：复杂表格仍需预处理
4. 手写内容：需要先进行 OCR

8.2 改进建议

# 策略 1：使用本地模型处理敏感数据
sensitive_result = lx.extract(
    text_or_documents=sensitive_document,
    model_id="ollama://llama3.2",  # 本地运行
)

# 策略 2：分阶段处理降低成本
# 第一阶段：用 Flash 快速筛选
coarse_result = lx.extract(..., model_id="gemini-2.5-flash")
# 第二阶段：仅对关键实体用 Pro 精细提取
for entity in coarse_result.extractions[:10]:  # 只处理前 10 个关键实体
    fine_result = lx.extract(..., model_id="gemini-2.5-pro")

8.3 与向量数据库结合

# 将提取结果存入 Milvus，支持语义搜索
from pymilvus import Collection

def store_extractions_to_milvus(
    extractions: list[Extraction],
    collection: Collection
):
    entities = []
    for ext in extractions:
        entities.append({
            "id": hash(ext.extraction_text),
            "class": ext.extraction_class,
            "text": ext.extraction_text,
            "embedding": get_embedding(ext.extraction_text),
            "attributes": json.dumps(ext.attributes),
            "source_location": f"{ext.char_interval.start}-{ext.char_interval.end}"
        })
    
    collection.insert(entities)

结语

Google LangExtract 代表了信息提取技术的一个重要转折点：从"训练模型"转向"编排模型"。

在传统范式下，如果你想从医疗病历中提取结构化信息，你需要：

1. 收集数千条标注数据
2. 训练一个 NER 模型
3. 部署推理服务
4. 处理模型漂移和领域迁移问题

而 LangExtract 只需要你：

1. 写一段自然语言描述
2. 提供 3-5 个示例
3. 调用一次 API

这不是说深度学习不重要，而是说对于大多数企业应用场景，工程化的可靠性比模型的复杂度更重要。LangExtract 的价值在于：它把"让 LLM 的输出可信"这个问题，用软件工程的方式解决了。

对于医疗、法律、金融等领域的从业者，LangExtract 提供了一条从"非结构化数据"到"结构化知识"的可行路径。你不需要成为机器学习专家，只需要理解你的业务逻辑，然后用自然语言告诉模型你要什么。

参考资源：

• LangExtract GitHub: https://github.com/google/langextract
• LangExtract PyPI: https://pypi.org/project/langextract/
• Gemini API 文档: https://ai.google.dev/docs
• LangExtract 论文: arXiv:2404.16812

本文所有代码均基于 LangExtract 1.0 版本。实际使用时请参考官方最新文档。