实战：数据分析与信息抽取

第二个实战项目聚焦数据分析与信息抽取——从非结构化文本（合同、简历、新闻、报告）中自动抽取结构化数据。这是企业中 LLM 落地最多的场景之一。

项目目标

核心抽取引擎

通用抽取框架

"""
data_extractor.py - 通用信息抽取引擎
支持自定义Schema，适用于各种文档类型的信息抽取。
"""
from pydantic import BaseModel, Field
from openai import OpenAI
from typing import Optional
import json
client = OpenAI()
class ExtractionEngine:
"""
通用信息抽取引擎。
通过 Pydantic Schema 定义抽取目标，
自动生成提示词并结构化抽取结果。
"""
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
def extract(
self,
text: str,
schema: type[BaseModel],
instructions: str = "",
examples: list[dict] = None
) -> BaseModel:
"""
从文本中抽取结构化信息。
Args:
text: 待抽取的原文
schema: Pydantic模型，定义抽取目标
instructions: 额外的抽取指令
examples: Few-shot示例 [{"input": str, "output": dict}]
Returns:
填充好的Pydantic模型实例
"""
system_prompt = self._build_system_prompt(schema, instructions, examples)
response = client.beta.chat.completions.parse(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"请从以下文本中抽取信息：\n\n{text}"}
],
response_format=schema,
temperature=0.0
)
return response.choices[0].message.parsed
def _build_system_prompt(
self,
schema: type[BaseModel],
instructions: str,
examples: list[dict]
) -> str:
"""构建抽取用的System Prompt"""
# 从Pydantic模型提取字段描述
field_descriptions = []
for name, field_info in schema.model_fields.items():
desc = field_info.description or name
field_descriptions.append(f"- {name}: {desc}")
fields_text = "\n".join(field_descriptions)
prompt = f"""你是一个专业的信息抽取助手。从用户提供的文本中准确抽取以下字段：
## 要抽取的字段
{fields_text}
## 抽取规则
1. 严格基于原文抽取，不要编造不存在的信息
2. 如果某个字段在原文中找不到，使用null或空值
3. 日期统一格式: YYYY-MM-DD
4. 金额保留原始货币单位和数值
5. 人名、公司名保持原文写法"""
if instructions:
prompt += f"\n\n## 特殊说明\n{instructions}"
if examples:
prompt += "\n\n## 示例"
for i, ex in enumerate(examples, 1):
prompt += f"\n\n### 示例{i}\n输入: {ex['input'][:200]}\n输出: {json.dumps(ex['output'], ensure_ascii=False, indent=2)}"
return prompt
def batch_extract(
self,
texts: list[str],
schema: type[BaseModel],
instructions: str = ""
) -> list[BaseModel]:
"""批量抽取"""
results = []
for i, text in enumerate(texts, 1):
print(f"  抽取 [{i}/{len(texts)}]...", end=" ")
result = self.extract(text, schema, instructions)
results.append(result)
print("✅")
return results

应用场景1: 合同信息抽取

# ===== 合同抽取Schema =====
class ContractParty(BaseModel):
"""合同方"""
name: str = Field(description="公司或个人名称")
role: str = Field(description="角色：甲方/乙方/丙方")
representative: Optional[str] = Field(default=None, description="法定代表人")
class ContractInfo(BaseModel):
"""合同关键信息"""
contract_title: str = Field(description="合同标题")
contract_number: Optional[str] = Field(default=None, description="合同编号")
parties: list[ContractParty] = Field(description="合同各方")
sign_date: Optional[str] = Field(default=None, description="签订日期 YYYY-MM-DD")
effective_date: Optional[str] = Field(default=None, description="生效日期")
expiry_date: Optional[str] = Field(default=None, description="到期日期")
total_amount: Optional[str] = Field(default=None, description="合同总金额（含币种）")
payment_terms: Optional[str] = Field(default=None, description="付款条件摘要")
key_obligations: list[str] = Field(default_factory=list, description="核心义务条款，最多5条")
termination_clause: Optional[str] = Field(default=None, description="终止条款摘要")
# ===== 使用示例 =====
engine = ExtractionEngine()
contract_text = """
技术服务合同
合同编号：TS-2025-0042
甲方：北京智联科技有限公司
法定代表人：张明
地址：北京市海淀区中关村大街1号
乙方：深圳云端数据服务有限公司
法定代表人：李华
地址：深圳市南山区科技园南路88号
签订日期：2025年3月15日
第一条 服务内容
乙方为甲方提供AI模型部署与运维服务，包括但不限于模型训练、API部署、性能监控。
第二条 服务期限
自2025年4月1日起至2026年3月31日止。
第三条 合同金额与支付
合同总额为人民币480,000元（大写：肆拾捌万元整）。
支付方式：签订后支付30%，项目中期支付40%，验收后支付30%。
第四条 违约责任
任何一方违反本合同约定，应向守约方支付合同总额10%的违约金。
"""
result = engine.extract(
contract_text,
ContractInfo,
instructions="中国法律合同格式，注意区分甲方乙方"
)
print(json.dumps(result.model_dump(), ensure_ascii=False, indent=2))

输出结果

{
"contract_title": "技术服务合同",
"contract_number": "TS-2025-0042",
"parties": [
{"name": "北京智联科技有限公司", "role": "甲方", "representative": "张明"},
{"name": "深圳云端数据服务有限公司", "role": "乙方", "representative": "李华"}
],
"sign_date": "2025-03-15",
"effective_date": "2025-04-01",
"expiry_date": "2026-03-31",
"total_amount": "人民币480,000元",
"payment_terms": "签订后30%，中期40%，验收后30%",
"key_obligations": [
"乙方为甲方提供AI模型部署与运维服务",
"服务包括模型训练、API部署、性能监控",
"违约方支付合同总额10%违约金"
],
"termination_clause": null
}

应用场景2: 简历信息抽取

class Education(BaseModel):
"""教育经历"""
school: str = Field(description="学校名称")
degree: str = Field(description="学历：本科/硕士/博士/...")
major: str = Field(description="专业")
start_year: Optional[int] = Field(default=None, description="入学年份")
end_year: Optional[int] = Field(default=None, description="毕业年份")
class WorkExperience(BaseModel):
"""工作经历"""
company: str = Field(description="公司名称")
position: str = Field(description="职位")
start_date: Optional[str] = Field(default=None, description="入职日期")
end_date: Optional[str] = Field(default=None, description="离职日期，在职填'至今'")
responsibilities: list[str] = Field(description="主要职责，2-4条")
class ResumeInfo(BaseModel):
"""简历信息"""
name: str = Field(description="姓名")
phone: Optional[str] = Field(default=None, description="手机号")
email: Optional[str] = Field(default=None, description="邮箱")
location: Optional[str] = Field(default=None, description="所在城市")
years_of_experience: Optional[int] = Field(default=None, description="工作年限")
education: list[Education] = Field(description="教育经历")
work_experience: list[WorkExperience] = Field(description="工作经历")
skills: list[str] = Field(description="核心技能")
summary: str = Field(description="一句话概括此人的专业背景和优势")
# 使用
resume_text = """
王小明
13800138000 | xiaoming@email.com | 北京
10年软件开发经验，专注于AI和后端系统开发。
教育背景
清华大学  计算机科学与技术  硕士  2012-2015
武汉大学  软件工程  本科  2008-2012
工作经历
字节跳动 | 高级技术专家 | 2020.06 - 至今
- 负责推荐系统的AI模型开发与优化
- 带领10人团队完成大模型落地项目
- 系统QPS从5000提升到20000
阿里巴巴 | 资深开发工程师 | 2015.07 - 2020.05
- 负责淘宝搜索后端架构设计
- 主导微服务改造，降低40%运维成本
- 获2019年技术突破奖
技能
Python, Java, Go, PyTorch, TensorFlow, Kubernetes, AWS
"""
resume = engine.extract(resume_text, ResumeInfo)
print(f"姓名: {resume.name}")
print(f"经验: {resume.years_of_experience}年")
print(f"总结: {resume.summary}")

应用场景3: 批量评论分析

class ReviewAnalysis(BaseModel):
"""产品评论分析"""
sentiment: str = Field(description="情感倾向: positive/negative/neutral")
rating_estimate: int = Field(description="推测评分1-5", ge=1, le=5)
product_aspects: list[str] = Field(description="提到的产品方面（如价格、质量、外观等）")
pain_points: list[str] = Field(default_factory=list, description="用户痛点")
praise_points: list[str] = Field(default_factory=list, description="用户好评点")
purchase_intent: str = Field(description="复购意愿: high/medium/low")
key_quote: str = Field(description="最有代表性的一句原文")
# 批量分析
reviews = [
"这款耳机音质确实不错，降噪效果杠杠的。就是续航有点短，4小时就没电了。价格有点贵但物有所值。",
"垃圾产品，买回来第三天就坏了，售后态度也差，强烈不推荐！",
"中规中矩吧，没有特别惊喜也没什么槽点。包装挺精美的，送人还行。",
]
results = engine.batch_extract(reviews, ReviewAnalysis)
# 汇总分析
print("\n📊 评论分析汇总\n")
print(f"{'情感':>8} {'评分':>4} {'复购':>8} 核心引用")
print("-" * 60)
for r in results:
emoji = {"positive": "😊", "negative": "😡", "neutral": "😐"}.get(r.sentiment, "❓")
print(f"{emoji} {r.sentiment:>8} {r.rating_estimate}/5  {r.purchase_intent:>8}  {r.key_quote[:30]}...")

抽取 Pipeline：复杂文档处理

对于长文档，需要分段处理再合并结果：

def extract_from_long_document(
document: str,
schema: type[BaseModel],
chunk_size: int = 2000,
overlap: int = 200
) -> list[BaseModel]:
"""
处理超长文档的分段抽取。
Args:
document: 长文档文本
schema: 抽取Schema
chunk_size: 每段最大字符数
overlap: 段间重叠字符数（防止信息被截断）
Returns:
合并后的抽取结果列表
"""
engine = ExtractionEngine()
# Step 1: 分段
chunks = []
start = 0
while start < len(document):
end = start + chunk_size
# 尝试在段落边界截断
if end < len(document):
boundary = document.rfind("\n\n", start, end)
if boundary > start:
end = boundary
chunks.append(document[start:end])
start = end - overlap  # 保留重叠
print(f"文档分为 {len(chunks)} 段")
# Step 2: 分段抽取
all_results = engine.batch_extract(chunks, schema)
# Step 3: 去重（基于关键字段）
seen = set()
unique_results = []
for r in all_results:
key = json.dumps(r.model_dump(), sort_keys=True, ensure_ascii=False)
if key not in seen:
seen.add(key)
unique_results.append(r)
print(f"抽取结果: {len(all_results)} → 去重后 {len(unique_results)}")
return unique_results

项目中的提示工程技术对照

动手练习

练习：构建你的信息抽取器

1. 选择文档类型 — 合同、简历、新闻、或你工作中常见的文档
2. 定义 Schema — 用 Pydantic 定义需要抽取的字段
3. 准备测试数据 — 至少3份不同的文档样本
4. 测试与优化 — 运行抽取，检查缺失字段和错误，调整提示词
5. 加入 Few-shot — 对复杂字段添加抽取示例

本节要点

• ✅ 通用抽取引擎通过 Pydantic Schema 定义目标，适用于各种文档类型
• ✅ 合同、简历、评论等是信息抽取最常见的落地场景
• ✅ Few-shot 示例能显著提升复杂格式的抽取准确率
• ✅ 长文档需要分段处理：分段→抽取→合并→去重
• ✅ 批量处理时注意成本控制和速率限制

下一步：实战：智能客服对话系统 🚀