构建可信赖的LLM信息抽取系统：从提示词到生产部署

要让大语言模型（LLM）的信息抽取结果足够可靠，能够直接用于业务决策或自动化流程，需要解决的核心问题是：如何确保模型输出的结构化数据在准确性、一致性和完整性上达到生产环境要求。单纯依赖提示词工程或简单后处理往往不够，必须建立一套覆盖数据验证、错误处理、质量评估和流程控制的完整机制。

1. 理解 LLM 信息抽取的信任挑战

信息抽取（Information Extraction）任务要求模型从非结构化文本中识别并提取特定类型的实体、关系或事件，并以结构化格式（如 JSON、XML 或数据库记录）输出。LLM 在此类任务中表现出色，但直接用于生产环境仍面临几个关键信任问题：

• 输出格式不稳定：同一提示词下，LLM 可能返回 JSON、文本描述或混合格式，需要稳定解析。
• 内容幻觉（Hallucination）：模型可能生成原文中不存在的信息。
• 部分抽取与遗漏：长文本或复杂结构中容易漏掉部分实体或关系。
• 数值与日期格式不一致：如“2023年10月1日”可能被输出为“2023-10-01”“2023/10/1”或保留原文格式。
• 上下文依赖歧义：同一实体在不同段落中的指代可能被错误关联或重复计数。

这些问题的根源在于 LLM 本质是生成模型，而非精确的查询-应答系统。构建可信赖的抽取流程，必须从提示词设计、输出解析、验证规则和异常处理四个层面系统化解决。

2. 设计高可靠性的抽取提示词

提示词的质量直接决定抽取任务的基线成功率。以下是关键设计原则和示例。

2.1 明确输出格式与约束

使用 JSON Schema 或类似结构定义输出格式，并在提示词中明确说明：

请从以下文本中抽取所有人物姓名、职务和所属公司。严格按照以下 JSON 格式输出，不要添加任何解释性文字： { "persons": [ { "name": "字符串，必须为中文全名", "title": "字符串，职位名称", "company": "字符串，公司全称" } ] } 如果文本中没有相关信息，返回空数组。不要推断或添加文本中未明确提及的内容。 待处理文本：{{input_text}}

关键约束点：

• 指定根字段和数组结构
• 定义每个字段的数据类型和格式要求
• 明确空值处理规则
• 禁止模型自由发挥

2.2 提供少量示例（Few-shot Learning）

在提示词中包含 1-3 个正例和反例，帮助模型理解边界：

示例1： 输入："张三现任ABC科技公司首席技术官，负责产品研发。" 输出：{"persons": [{"name": "张三", "title": "首席技术官", "company": "ABC科技公司"}]} 示例2： 输入："会议讨论了市场趋势和产品规划。" 输出：{"persons": []} # 没有人名信息时返回空数组 示例3： 输入："李四曾在多家互联网公司工作。" 输出：{"persons": []} # 公司名称不明确，不抽取 待处理文本：{{input_text}}

示例的作用是让模型看到：

• 完整抽取的标准案例
• 无相关信息时的正确处理
• 模糊信息的保守策略（宁可漏抽，不要错抽）

2.3 分步骤思考（Chain-of-Thought）

对于复杂抽取任务，让模型先分析再输出：

请按以下步骤处理文本： 1. 识别文本中提到的所有人物 2. 确定每个人物的职务（如有） 3. 确定每个人物所属公司（如有） 4. 只抽取明确提及的信息，不要推断 5. 按指定JSON格式输出结果 待处理文本：{{input_text}}

这种方法能降低模型一次性生成完整答案的复杂度，提高关键信息的抽取准确率。

3. 实现健壮的输出解析与验证

即使有良好的提示词，LLM 输出仍需要程序化验证。以下是核心验证层设计。

3.1 多格式解析适配器

首先构建能处理多种响应格式的解析器：

import json import re from typing import Dict, List, Any, Optional class LLMOutputParser: def __init__(self, expected_schema: Dict[str, Any]): self.expected_schema = expected_schema def parse_json_response(self, response_text: str) -> Optional[Dict]: """尝试解析纯JSON、代码块中的JSON或文本中的JSON""" # 清理响应文本 cleaned_text = response_text.strip() # 情况1：直接JSON响应 try: return json.loads(cleaned_text) except json.JSONDecodeError: pass # 情况2：代码块中的JSON（如 ```json {...} ```） code_block_pattern = r'```(?:json)?\s*(\{.*?\})\s*```' match = re.search(code_block_pattern, cleaned_text, re.DOTALL) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass # 情况3：文本中可能包含JSON对象 json_pattern = r'\{[^{}]*"[^"]*"[^{}]*\}' matches = re.findall(json_pattern, cleaned_text) for match in matches: try: candidate = json.loads(match) # 验证基本结构是否符合预期 if self._validate_structure(candidate): return candidate except json.JSONDecodeError: continue return None def _validate_structure(self, data: Dict) -> bool: """验证数据基本结构是否符合预期schema""" if not isinstance(data, dict): return False for key, expected_type in self.expected_schema.items(): if key not in data: return False if not isinstance(data[key], expected_type): return False return True

3.2 数据质量验证规则

针对抽取内容实现业务规则验证：

class DataValidator: def __init__(self, validation_rules: Dict[str, List[callable]]): self.rules = validation_rules def validate_extraction(self, extracted_data: Dict) -> Dict[str, List[str]]: """验证抽取数据，返回错误信息""" errors = {} for field, rules in self.rules.items(): field_errors = [] value = extracted_data.get(field) for rule in rules: try: if not rule(value): field_errors.append(f"违反规则: {rule.__name__}") except Exception as e: field_errors.append(f"验证异常: {str(e)}") if field_errors: errors[field] = field_errors return errors # 定义具体验证规则 def validate_person_name(name: str) -> bool: """验证人名格式""" if not name or not isinstance(name, str): return False # 中文姓名通常2-4个字符，不包含特殊字符 if not re.match(r'^[\u4e00-\u9fa5]{2,4}$', name.strip()): return False return True def validate_company_name(company: str) -> bool: """验证公司名格式""" if not company or not isinstance(company, str): return False # 公司名应包含公司、集团、科技等关键词 keywords = ['公司', '集团', '科技', '有限', '股份'] return any(keyword in company for keyword in keywords) def validate_title_format(title: str) -> bool: """验证职务格式""" if not title or not isinstance(title, str): return False # 职务应包含常见职务关键词 title_keywords = ['总监', '经理', '工程师', '顾问', '专家', '代表'] return any(keyword in title for keyword in title_keywords) # 配置验证规则 validation_rules = { 'name': [validate_person_name], 'company': [validate_company_name], 'title': [validate_title_format] }

3.3 一致性检查与去重

处理同一实体的多次出现和冲突信息：

class ConsistencyChecker: def deduplicate_entities(self, entities: List[Dict]) -> List[Dict]: """基于关键字段去重实体""" seen = set() unique_entities = [] for entity in entities: # 创建实体的唯一标识 identifier = self._create_entity_identifier(entity) if identifier not in seen: seen.add(identifier) unique_entities.append(entity) return unique_entities def _create_entity_identifier(self, entity: Dict) -> str: """创建实体唯一标识（可根据业务调整）""" name = entity.get('name', '').lower().strip() company = entity.get('company', '').lower().strip() return f"{name}::{company}" def resolve_conflicts(self, entities: List[Dict]) -> List[Dict]: """解决同一实体的冲突信息""" grouped = {} for entity in entities: identifier = self._create_entity_identifier(entity) if identifier not in grouped: grouped[identifier] = [] grouped[identifier].append(entity) resolved_entities = [] for identifier, variants in grouped.items(): if len(variants) == 1: resolved_entities.append(variants[0]) else: # 选择信息最完整的版本 best_variant = max(variants, key=lambda x: len(str(x))) resolved_entities.append(best_variant) return resolved_entities

4. 构建生产级抽取流水线

将上述组件组合成完整的处理流水线，包含重试、监控和降级策略。

4.1 核心流水线实现

import logging from datetime import datetime from typing import Dict, List, Any, Optional class TrustworthyExtractionPipeline: def __init__(self, llm_client, parser, validator, checker): self.llm = llm_client self.parser = parser self.validator = validator self.checker = checker self.logger = logging.getLogger(__name__) def extract_with_retry(self, text: str, max_retries: int = 3) -> Dict[str, Any]: """带重试的抽取流程""" for attempt in range(max_retries): try: self.logger.info(f"抽取尝试 {attempt + 1}/{max_retries}") # 1. LLM调用 response = self.llm.generate(prompt=self._build_prompt(text)) # 2. 输出解析 parsed_data = self.parser.parse_json_response(response) if not parsed_data: raise ValueError("无法解析LLM响应") # 3. 数据验证 validation_errors = self.validator.validate_extraction(parsed_data) if validation_errors: self.logger.warning(f"验证错误: {validation_errors}") # 根据错误类型决定是否重试 if self._should_retry(validation_errors): continue # 4. 一致性处理 if 'persons' in parsed_data: deduplicated = self.checker.deduplicate_entities(parsed_data['persons']) resolved = self.checker.resolve_conflicts(deduplicated) parsed_data['persons'] = resolved # 5. 质量评分 quality_score = self._calculate_quality_score(parsed_data, validation_errors) parsed_data['_metadata'] = { 'extraction_timestamp': datetime.now().isoformat(), 'quality_score': quality_score, 'validation_errors': validation_errors, 'attempts': attempt + 1 } return parsed_data except Exception as e: self.logger.error(f"抽取失败 (尝试 {attempt + 1}): {str(e)}") if attempt == max_retries - 1: return self._get_fallback_result(str(e)) return self._get_fallback_result("超过最大重试次数") def _should_retry(self, errors: Dict) -> bool: """根据错误类型决定是否重试""" # 格式错误通常可重试，内容错误可能需要调整提示词 retryable_errors = ['格式错误', '解析失败'] for error_list in errors.values(): for error in error_list: if any(retryable in error for retryable in retryable_errors): return True return False def _calculate_quality_score(self, data: Dict, errors: Dict) -> float: """计算抽取质量评分（0-1）""" base_score = 1.0 # 根据错误数量扣分 error_penalty = len(errors) * 0.1 base_score -= min(error_penalty, 0.5) # 最多扣0.5分 # 根据数据完整性加分/扣分 if 'persons' in data: person_count = len(data['persons']) if person_count == 0: base_score *= 0.8 # 无结果可能表示原文无信息或抽取失败 elif person_count > 5: base_score = min(base_score * 1.1, 1.0) # 多结果可能表示成功 return round(base_score, 2) def _get_fallback_result(self, error_msg: str) -> Dict: """降级结果""" return { "persons": [], "_metadata": { "extraction_timestamp": datetime.now().isoformat(), "quality_score": 0.0, "error": error_msg, "fallback": True } }

4.2 监控与指标收集

生产环境需要实时监控抽取质量：

class ExtractionMonitor: def __init__(self): self.metrics = { 'total_requests': 0, 'successful_extractions': 0, 'failed_extractions': 0, 'average_quality_score': 0.0, 'common_errors': {} } def record_extraction_result(self, result: Dict): """记录单次抽取结果""" self.metrics['total_requests'] += 1 metadata = result.get('_metadata', {}) quality_score = metadata.get('quality_score', 0.0) if quality_score > 0.7: # 质量阈值可调整 self.metrics['successful_extractions'] += 1 else: self.metrics['failed_extractions'] += 1 errors = metadata.get('validation_errors', {}) self._record_errors(errors) # 更新平均质量分 current_total = (self.metrics['average_quality_score'] * (self.metrics['total_requests'] - 1)) self.metrics['average_quality_score'] = ( (current_total + quality_score) / self.metrics['total_requests'] ) def _record_errors(self, errors: Dict): """记录错误类型分布""" for field, error_list in errors.items(): if field not in self.metrics['common_errors']: self.metrics['common_errors'][field] = {} for error in error_list: self.metrics['common_errors'][field][error] = ( self.metrics['common_errors'][field].get(error, 0) + 1 ) def get_health_report(self) -> Dict: """生成健康度报告""" success_rate = (self.metrics['successful_extractions'] / self.metrics['total_requests'] if self.metrics['total_requests'] > 0 else 0) return { 'success_rate': round(success_rate, 3), 'average_quality': round(self.metrics['average_quality_score'], 3), 'total_processed': self.metrics['total_requests'], 'common_issues': self.metrics['common_errors'] }

5. 处理常见问题与优化策略

5.1 质量阈值与人工审核流程

根据业务风险设定质量阈值，建立人工审核通道：

class QualityGate: def __init__(self, auto_approve_threshold: float = 0.8, review_threshold: float = 0.5): self.auto_approve_threshold = auto_approve_threshold self.review_threshold = review_threshold def evaluate_result(self, result: Dict) -> str: """评估结果并返回处理建议""" metadata = result.get('_metadata', {}) quality_score = metadata.get('quality_score', 0.0) if quality_score >= self.auto_approve_threshold: return "auto_approve" elif quality_score >= self.review_threshold: return "human_review" else: return "reject" def should_trigger_human_review(self, result: Dict) -> bool: """判断是否需要人工审核""" evaluation = self.evaluate_result(result) return evaluation == "human_review" # 人工审核接口示例 class HumanReviewService: def submit_for_review(self, original_text: str, extraction_result: Dict, confidence: float) -> str: """提交人工审核并返回审核结果""" # 实现审核队列管理、分配审核任务、记录审核结果 pass

5.2 持续优化提示词

基于监控数据定期优化提示词：

class PromptOptimizer: def __init__(self, monitor: ExtractionMonitor): self.monitor = monitor def analyze_common_failures(self) -> List[str]: """分析常见失败模式""" recommendations = [] common_errors = self.monitor.metrics['common_errors'] for field, errors in common_errors.items(): for error_type, count in errors.items(): if count > 10: # 阈值可调整 if "格式" in error_type: recommendations.append( f"字段'{field}'格式错误频繁，考虑在提示词中加强格式说明") elif "验证" in error_type: recommendations.append( f"字段'{field}'内容验证失败，考虑提供更明确的示例") return recommendations def generate_improved_prompt(self, current_prompt: str, issues: List[str]) -> str: """基于问题分析生成改进版提示词""" improved_prompt = current_prompt for issue in issues: if "格式说明" in issue: # 在提示词中增加格式约束 if "严格按照以下JSON格式" not in improved_prompt: improved_prompt = improved_prompt.replace( "按指定JSON格式输出", "严格按照以下JSON格式输出，不要添加任何解释性文字" ) elif "明确示例" in issue: # 增加反例说明 if "不要推断" not in improved_prompt: improved_prompt += "\n不要推断或添加文本中未明确提及的内容。" return improved_prompt

6. 生产环境部署建议

6.1 基础设施要求

组件	最低要求	推荐配置	说明
LLM API客户端	支持重试机制	熔断器+降级策略	防止上游服务故障影响业务
解析验证服务	2核4GB内存	4核8GB内存+自动扩缩容	处理峰值请求
数据存储	关系型数据库	时序数据库+监控告警	记录抽取历史和质量指标
缓存层	本地内存缓存	Redis集群	缓存提示词模板和验证规则

6.2 关键监控指标

建立以下监控仪表盘：

• 服务质量指标

  • 请求成功率（>95%）
  • 平均响应时间（<5秒）
  • 质量分分布（按小时统计）

• 业务指标

  • 自动通过率 vs 人工审核率
  • 各字段抽取准确率
  • 常见错误类型趋势

• 系统指标

  • API调用频次和配额使用
  • 解析失败率
  • 内存和CPU使用率

6.3 灾备与降级方案

制定分级降级策略：

1. 一级降级：质量阈值从0.8降至0.6，减少人工审核量
2. 二级降级：关闭复杂验证规则，只进行基本格式检查
3. 三级降级：切换到规则基抽取或返回空结果并记录异常
4. 完全降级：服务不可用时的友好错误提示和人工处理通道

6.4 版本管理与回滚

• 提示词版本化：每次修改保存版本号和测试结果
• A/B测试：新提示词先在小流量测试，对比质量指标
• 快速回滚：保留最近3个稳定版本，支持分钟级回滚

构建可信赖的LLM信息抽取系统需要在前端提示词设计、中端解析验证和后端监控优化三个层面同时投入。通过系统化的质量控制和持续迭代，能够将抽取结果的可靠性提升到足以支撑业务决策的水平。关键是要建立数据驱动的优化循环，基于实际使用反馈不断调整技术方案和业务规则。