NLP 命名实体识别

SunnyFan大约 17 分钟约 4998 字

NLP 命名实体识别

简介

命名实体识别（Named Entity Recognition, NER）是从非结构化文本中识别并分类命名实体的任务，如人名、地名、组织机构、时间、金额等。NER 是信息抽取、知识图谱构建和智能问答的基础组件，在金融、医疗、法律等领域有广泛的应用需求。

NER 问题的核心挑战在于：同一个词在不同上下文中可能是实体也可能不是（如"苹果"可以指水果也可以指公司），实体边界有时模糊（如"北京大学信息科学技术学院"是一个整体还是多个嵌套实体），以及领域特定实体类型的定义和标注规范需要领域专家参与。随着预训练模型的发展，NER 的效果已经有了质的飞跃，但标注规范和数据质量仍然是决定项目成败的关键因素。

特点

1.序列标注任务 — 对输入序列的每个 token 进行标签预测，通常使用 BIO 或 BIOES 标注体系
2.实体边界敏感 — 不仅需要识别实体类型，还要准确预测实体的起始和结束位置
3.类别多样 — 标准实体包括人名（PER）、地名（LOC）、机构（ORG）等，领域场景可扩展
4.上下文依赖强 — 同一个词在不同上下文中可能是实体也可能不是，需要模型理解语境
5.评估维度多 — 同时关注实体级别的精确率、召回率和 F1，而非仅看 token 级别指标

实现

BIO 标注体系与数据对齐

# 示例1：BIO 标注体系与数据对齐
def convert_to_bio(tokens, entities):
    """
    将 token 列表和实体标注转换为 BIO 标签序列
    tokens: ["张", "三", "在", "北", "京", "工", "作"]
    entities: [(0, 2, "PER"), (3, 5, "LOC")]
    """
    tags = ['O'] * len(tokens)
    for start, end, label in entities:
        tags[start] = f'B-{label}'
        for i in range(start + 1, end):
            tags[i] = f'I-{label}'
    return list(zip(tokens, tags))

# 演示
tokens = ["张", "三", "在", "北", "京", "的", "清", "华", "大", "学", "工", "作"]
entities = [(0, 2, "PER"), (3, 5, "LOC"), (6, 10, "ORG")]
bio_result = convert_to_bio(tokens, entities)
for token, tag in bio_result:
    print(f"  {token:2s} -> {tag}")

# BIO 标签集合
label_list = ['O', 'B-PER', 'I-PER', 'B-LOC', 'I-LOC', 'B-ORG', 'I-ORG']
label2id = {l: i for i, l in enumerate(label_list)}
id2label = {i: l for l, i in label2id.items()}
print(f"\n标签映射: {label2id}")

BIOES 标注体系

def convert_to_bioes(tokens, entities):
    """
    BIOES 标注体系（比 BIO 更严格）
    B: 实体开始
    I: 实体内部
    O: 非实体
    E: 实体结束（单字实体用 S）
    S: 单字实体

    优势：能明确标记实体边界，避免连续实体的歧义
    """
    tags = ['O'] * len(tokens)
    for start, end, label in entities:
        if end - start == 1:
            # 单字实体
            tags[start] = f'S-{label}'
        else:
            tags[start] = f'B-{label}'
            tags[end - 1] = f'E-{label}'
            for i in range(start + 1, end - 1):
                tags[i] = f'I-{label}'
    return list(zip(tokens, tags))

bioes_result = convert_to_bioes(tokens, entities)
print("BIOES 标注结果:")
for token, tag in bioes_result:
    print(f"  {token:2s} -> {tag}")

基于 BiLSTM-CRF 的 NER 模型

# 示例2：基于 BiLSTM-CRF 的 NER 模型
import torch
import torch.nn as nn

class BiLSTM_CRF(nn.Module):
    def __init__(self, vocab_size, embed_dim=128, hidden_dim=256, num_tags=9):
        super().__init__()
        self.num_tags = num_tags
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim // 2,
                           num_layers=2, bidirectional=True, batch_first=True)
        self.hidden2tag = nn.Linear(hidden_dim, num_tags)

        # CRF 转移矩阵: transitions[i][j] 表示从标签 j 转移到标签 i 的分数
        self.transitions = nn.Parameter(torch.randn(num_tags, num_tags))
        # 开始和结束的转移分数
        self.start_transitions = nn.Parameter(torch.randn(num_tags))
        self.end_transitions = nn.Parameter(torch.randn(num_tags))

    def _get_lstm_features(self, x):
        embeds = self.embedding(x)
        lstm_out, _ = self.lstm(embeds)
        lstm_feats = self.hidden2tag(lstm_out)
        return lstm_feats

    def forward(self, x, tags, mask=None):
        feats = self._get_lstm_features(x)
        # 返回负对数似然损失
        forward_score = self._forward_algorithm(feats, mask)
        gold_score = self._score_sentence(feats, tags, mask)
        return (forward_score - gold_score).mean()

    def decode(self, x, mask=None):
        """维特比解码，输出最优标签序列"""
        feats = self._get_lstm_features(x)
        return self._viterbi_decode(feats, mask)

    def _forward_algorithm(self, feats, mask):
        # 简化实现，实际使用 pytorch-crf 库更高效
        batch_size, seq_len, num_tags = feats.shape
        alpha = self.start_transitions + feats[:, 0]
        for i in range(1, seq_len):
            emit = feats[:, i].unsqueeze(1)  # (B, 1, T)
            trans = self.transitions.unsqueeze(0)  # (1, T, T)
            alpha = torch.logsumexp(alpha.unsqueeze(2) + trans + emit, dim=1)
        return torch.logsumexp(alpha + self.end_transitions, dim=1)

    def _score_sentence(self, feats, tags, mask):
        batch_size, seq_len, _ = feats.shape
        score = self.start_transitions[tags[:, 0]] + feats[:, 0].gather(1, tags[:, 0:1]).squeeze(1)
        for i in range(1, seq_len):
            score += self.transitions[tags[:, i], tags[:, i-1]] + \
                     feats[:, i].gather(1, tags[:, i:i+1]).squeeze(1)
        score += self.end_transitions[tags[:, -1]]
        return score

    def _viterbi_decode(self, feats, mask):
        batch_size, seq_len, num_tags = feats.shape
        score = self.start_transitions + feats[:, 0]
        history = []
        for i in range(1, seq_len):
            broadcast_score = score.unsqueeze(2)
            broadcast_emit = feats[:, i].unsqueeze(1)
            next_score = broadcast_score + self.transitions + broadcast_emit
            max_score, max_idx = next_score.max(dim=1)
            history.append(max_idx)
            score = max_score
        score += self.end_transitions
        best_tag = score.argmax(dim=1)
        return best_tag

model = BiLSTM_CRF(vocab_size=5000, num_tags=len(label_list))
dummy_input = torch.randint(0, 5000, (4, 30))
dummy_tags = torch.randint(0, len(label_list), (4, 30))
loss = model(dummy_input, dummy_tags)
print(f"BiLSTM-CRF Loss: {loss.item():.4f}")

CRF 转移约束解释

def explain_crf_transitions(num_tags=9):
    """
    CRF 转移矩阵的作用：
    学习标签之间的合法转移模式

    合法转移示例（BIO 体系）：
    - O → B-PER  ✓（新实体开始）
    - I-PER → I-PER  ✓（实体继续）
    - I-PER → B-LOC  ✓（新实体开始）
    - O → I-PER  ✗（I 不能直接跟 O）
    - B-PER → I-LOC  ✗（不同类型的 I 不能跟 B）
    """
    print("CRF 通过学习转移矩阵，自动学习标签约束：")
    print("  - I-PER 后面通常跟 I-PER 或 B-xxx，不会跟 O")
    print("  - B-PER 后面通常跟 I-PER，不会跟 I-LOC")
    print("  - 这些约束不是手工定义的，而是从训练数据中学习得到的")

explain_crf_transitions()

基于 BERT 的 NER 微调

# 示例3：基于 BERT 的 NER 微调
from transformers import BertForTokenClassification, BertTokenizer
import torch

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertForTokenClassification.from_pretrained(
    'bert-base-chinese',
    num_labels=len(label_list),
)

text = "李明在北京大学学习计算机科学"
inputs = tokenizer(text, return_tensors='pt', return_offsets_mapping=True)

with torch.no_grad():
    outputs = model(inputs['input_ids'])
    predictions = torch.argmax(outputs.logits, dim=-1)[0]

# 对齐 token 和预测结果（跳过 [CLS] 和 [SEP]）
tokens = tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
for token, pred_id in zip(tokens[1:-1], predictions[1:-1]):
    tag = id2label.get(pred_id.item(), 'O')
    if tag != 'O':
        print(f"  {token:3s} -> {tag}")

print(f"\n模型参数量: {sum(p.numel() for p in model.parameters()):,}")

BERT NER 完整训练流程

def train_bert_ner(model, tokenizer, train_dataset, val_dataset,
                   label2id, id2label, epochs=3, lr=2e-5, batch_size=16):
    """
    BERT NER 完整训练流程
    """
    from torch.utils.data import DataLoader
    from torch.optim import AdamW

    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    optimizer = AdamW(model.parameters(), lr=lr)

    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size)

    for epoch in range(epochs):
        model.train()
        total_loss = 0

        for batch in train_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)

            optimizer.zero_grad()
            outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
            loss = outputs.loss
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
            optimizer.step()
            total_loss += loss.item()

        avg_loss = total_loss / len(train_loader)

        # 验证
        model.eval()
        val_loss = 0
        with torch.no_grad():
            for batch in val_loader:
                input_ids = batch['input_ids'].to(device)
                attention_mask = batch['attention_mask'].to(device)
                labels = batch['labels'].to(device)
                outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
                val_loss += outputs.loss.item()

        print(f"Epoch {epoch+1}/{epochs}, Train Loss: {avg_loss:.4f}, Val Loss: {val_loss/len(val_loader):.4f}")

    return model

print("BERT NER 训练流程已定义")

NER 评估（实体级别）

# 示例4：NER 评估（实体级别）
def evaluate_ner(y_true_tags, y_pred_tags, tokens_list):
    """
    严格实体级别评估：实体边界和类型完全匹配才算正确
    """
    tp, fp, fn = 0, 0, 0

    for tokens, true_tags, pred_tags in zip(tokens_list, y_true_tags, y_pred_tags):
        true_entities = extract_entities(tokens, true_tags)
        pred_entities = extract_entities(tokens, pred_tags)

        true_set = set(true_entities)
        pred_set = set(pred_entities)

        tp += len(true_set & pred_set)
        fp += len(pred_set - true_set)
        fn += len(true_set - pred_set)

    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

    print(f"实体级别评估: P={precision:.4f}, R={recall:.4f}, F1={f1:.4f}")
    return precision, recall, f1

def extract_entities(tokens, tags):
    """从 BIO 标签中提取实体"""
    entities = []
    current_entity = None
    for i, (token, tag) in enumerate(zip(tokens, tags)):
        if tag.startswith('B-'):
            if current_entity:
                entities.append(current_entity)
            current_entity = {'type': tag[2:], 'start': i, 'tokens': [token]}
        elif tag.startswith('I-') and current_entity and current_entity['type'] == tag[2:]:
            current_entity['tokens'].append(token)
        else:
            if current_entity:
                entities.append(current_entity)
                current_entity = None
    if current_entity:
        entities.append(current_entity)
    return [(e['type'], e['start'], e['start'] + len(e['tokens'])) for e in entities]

# 模拟评估
tokens_list = [["张", "三", "在", "北", "京"]]
y_true = [['B-PER', 'I-PER', 'O', 'B-LOC', 'I-LOC']]
y_pred = [['B-PER', 'I-PER', 'O', 'B-ORG', 'I-ORG']]  # LOC 被误判为 ORG
evaluate_ner(y_true, y_pred, tokens_list)

每类实体详细评估

def detailed_ner_evaluation(y_true_tags, y_pred_tags, tokens_list, label_list):
    """每类实体的详细评估"""
    all_labels = [l for l in label_list if l != 'O']
    results = {}

    for label_prefix in set(l.split('-')[1] for l in all_labels):
        tp, fp, fn = 0, 0, 0
        for tokens, true_tags, pred_tags in zip(tokens_list, y_true_tags, y_pred_tags):
            true_entities = [e for e in extract_entities(tokens, true_tags) if e[0] == label_prefix]
            pred_entities = [e for e in extract_entities(tokens, pred_tags) if e[0] == label_prefix]

            true_set = set(true_entities)
            pred_set = set(pred_entities)
            tp += len(true_set & pred_set)
            fp += len(pred_set - true_set)
            fn += len(true_set - pred_set)

        p = tp / (tp + fp) if (tp + fp) > 0 else 0
        r = tp / (tp + fn) if (tp + fn) > 0 else 0
        f1 = 2 * p * r / (p + r) if (p + r) > 0 else 0
        results[label_prefix] = {"precision": p, "recall": r, "f1": f1, "support": tp + fn}

    print("各类实体评估结果:")
    for label, metrics in results.items():
        print(f"  {label:5s}: P={metrics['precision']:.3f} R={metrics['recall']:.3f} "
              f"F1={metrics['f1']:.3f} (样本数: {metrics['support']})")
    return results

print("详细 NER 评估函数已定义")

标注一致性评估

def inter_annotator_agreement(annotator1_tags, annotator2_tags):
    """
    标注者间一致性（IAA）评估——Cohen's Kappa
    在开始大规模标注前，必须先评估标注者间一致性
    Kappa > 0.8: 一致性很好
    Kappa 0.6-0.8: 一致性可接受
    Kappa < 0.6: 需要重新定义标注规范
    """
    from sklearn.metrics import cohen_kappa_score

    # 将 BIO 标签展平
    tags1 = [t for tags in annotator1_tags for t in tags]
    tags2 = [t for tags in annotator2_tags for t in tags]

    kappa = cohen_kappa_score(tags1, tags2)
    print(f"Cohen's Kappa: {kappa:.4f}")
    if kappa > 0.8:
        print("标注一致性很好，可以开始大规模标注")
    elif kappa > 0.6:
        print("标注一致性可接受，建议对不一致的样本进行讨论")
    else:
        print("标注一致性不足，需要重新培训标注员或修改标注规范")

print("标注一致性评估函数已定义")

Span-based NER 方法

传统的 BIO 标注方法无法处理嵌套实体问题。Span-based 方法将 NER 转化为 span 分类任务，枚举文本中所有可能的片段并判断其是否为实体。

# Span-based NER 实现思路
import torch
import torch.nn as nn

class SpanClassifier(nn.Module):
    """基于 Span 分类的 NER 模型，可以处理嵌套实体"""
    def __init__(self, hidden_dim=768, num_labels=10, max_span_length=10):
        super().__init__()
        self.max_span_length = max_span_length
        self.span_embedding = nn.Sequential(
            nn.Linear(hidden_dim * 3, hidden_dim),  # start + end + span_mean
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(hidden_dim, num_labels + 1),  # +1 for non-entity
        )

    def get_span_representations(self, hidden_states, max_span_length=10):
        """
        从 BERT hidden states 中提取所有 span 的表示
        span 表示 = [start_token, end_token, mean(span_tokens)]
        """
        batch_size, seq_len, hidden_dim = hidden_states.shape
        span_reprs = []
        span_indices = []

        for start in range(seq_len):
            for end in range(start + 1, min(start + max_span_length + 1, seq_len + 1)):
                start_repr = hidden_states[:, start, :]
                end_repr = hidden_states[:, end - 1, :]
                # span 内 token 的平均表示
                span_tokens = hidden_states[:, start:end, :]
                mean_repr = span_tokens.mean(dim=1)
                span_repr = torch.cat([start_repr, end_repr, mean_repr], dim=-1)
                span_reprs.append(span_repr)
                span_indices.append((start, end))

        if span_reprs:
            span_reprs = torch.stack(span_reprs, dim=1)  # (B, num_spans, hidden*3)
        return span_reprs, span_indices

    def forward(self, hidden_states):
        span_reprs, span_indices = self.get_span_representations(hidden_states)
        logits = self.span_embedding(span_reprs)  # (B, num_spans, num_labels+1)
        return logits, span_indices

# 使用示例
hidden_states = torch.randn(2, 20, 768)  # batch=2, seq_len=20, hidden=768
model = SpanClassifier()
logits, span_indices = model(hidden_states)
print(f"Span 数量: {len(span_indices)}")
print(f"Logits 形状: {logits.shape}")

嵌套实体处理

# 处理嵌套实体示例
nested_text = "北京大学信息科学技术学院"
# 嵌套结构：
# [北京大学信息科学技术学院]ORG
#   [北京大学]ORG
#     [北京]LOC

def extract_nested_entities(text, entities):
    """
    提取嵌套实体
    entities: [(start, end, label), ...] 可以有重叠
    """
    # 按长度排序，最外层实体在前
    sorted_entities = sorted(entities, key=lambda x: x[1] - x[0], reverse=True)

    result = []
    for start, end, label in sorted_entities:
        entity_text = text[start:end]
        # 查找内部的嵌套实体
        inner = [e for e in entities
                 if e[0] >= start and e[1] <= end
                 and (e[0], e[1]) != (start, end)]
        result.append({
            'text': entity_text,
            'label': label,
            'start': start,
            'end': end,
            'inner_entities': inner
        })
    return result

# 使用 GlobalPointer 思路处理嵌套 NER
# GlobalPointer 利用相对位置编码，为每个 (start, end, label) 三元组打分
# 可以自然地处理嵌套实体

使用大模型做 Few-Shot NER

# 使用 GPT/大模型做 Few-Shot NER
# 优势：无需训练，仅通过提示词适配新领域

few_shot_prompt = """从以下文本中提取命名实体，按 JSON 格式输出。

实体类型：PER（人名）、LOC（地名）、ORG（机构名）、TIME（时间）、MONEY（金额）

示例1：
输入：张三于2023年5月在北京百度总部参加会议
输出：{"PER": ["张三"], "LOC": ["北京"], "ORG": ["百度"], "TIME": ["2023年5月"]}

示例2：
输入：李四向王五借款人民币50万元，约定2024年底归还
输出：{"PER": ["李四", "王五"], "MONEY": ["人民币50万元"], "TIME": ["2024年底"]}

示例3：
输入：华为技术有限公司在深圳注册成立，注册资本5000万元
输出：{"ORG": ["华为技术有限公司"], "LOC": ["深圳"], "MONEY": ["5000万元"]}

请提取以下文本的实体：
输入：{input_text}
输出："""

def few_shot_ner(text, prompt_template=few_shot_prompt):
    """使用大模型进行 Few-Shot NER"""
    prompt = prompt_template.format(input_text=text)
    # 实际使用时调用大模型 API
    # response = openai.ChatCompletion.create(
    #     model="gpt-3.5-turbo",
    #     messages=[{"role": "user", "content": prompt}],
    #     temperature=0.0,
    # )
    # return parse_json(response.choices[0].message.content)
    print(f"Prompt 长度: {len(prompt)} 字符")
    return prompt

# 使用
few_shot_ner("阿里巴巴集团宣布在杭州建立新总部，投资金额达100亿元")

数据增强策略

# NER 数据增强方法

# 方法1：实体替换 — 替换同类型的实体
entity_pool = {
    'PER': ['张三', '李四', '王五', '赵六', '钱七', '孙八'],
    'LOC': ['北京', '上海', '广州', '深圳', '杭州', '成都'],
    'ORG': ['阿里巴巴', '腾讯', '百度', '字节跳动', '华为', '小米'],
}

def entity_replacement_augment(tokens, tags, entity_pool, replace_prob=0.3):
    """随机替换同类型的实体"""
    import random
    new_tokens = tokens.copy()
    entities = extract_entities(tokens, tags)

    for entity_type, start, end in entities:
        if random.random() < replace_prob:
            new_entity = random.choice(entity_pool.get(entity_type, []))
            new_entity_tokens = list(new_entity)
            # 注意：替换后长度可能变化，需要重新对齐标签
            if len(new_entity_tokens) == end - start:
                for i, t in enumerate(new_entity_tokens):
                    new_tokens[start + i] = t

    return new_tokens, tags

# 方法2：同义词替换 — 替换非实体词
def synonym_replacement(tokens, tags, synonym_dict, replace_prob=0.1):
    """替换非实体词为同义词"""
    import random
    new_tokens = tokens.copy()
    for i, (token, tag) in enumerate(zip(tokens, tags)):
        if tag == 'O' and token in synonym_dict and random.random() < replace_prob:
            new_tokens[i] = random.choice(synonym_dict[token])
    return new_tokens, tags

# 方法3：回译增强 — 利用翻译 API 生成同义句
def back_translation_augment(text, src_lang='zh', mid_lang='en'):
    """
    回译增强：原文 -> 英文 -> 中文
    生成语义相同但表述不同的训练样本
    """
    # 实际使用翻译 API
    # translated = translate(text, src=src_lang, tgt=mid_lang)
    # back_translated = translate(translated, src=mid_lang, tgt=src_lang)
    # 需要重新标注 NER 标签（可使用已有模型预测）
    pass

print("数据增强方法已定义")

NER 与实体链接（Entity Linking）

# NER 的下游任务：实体链接
# 实体链接将识别到的实体映射到知识库中的标准实体

class EntityLinker:
    """简易实体链接器"""
    def __init__(self, knowledge_base):
        """
        knowledge_base: {
            "entity_id": {
                "name": "标准名称",
                "aliases": ["别名1", "别名2"],
                "type": "ORG",
                "description": "描述"
            }
        }
        """
        self.kb = knowledge_base
        self.alias_index = {}
        for eid, info in knowledge_base.items():
            self.alias_index[info['name']] = eid
            for alias in info.get('aliases', []):
                self.alias_index[alias] = eid

    def link(self, entity_text, entity_type):
        """将实体文本链接到知识库"""
        # 精确匹配
        if entity_text in self.alias_index:
            return self.alias_index[entity_text]

        # 模糊匹配（编辑距离）
        best_match = None
        best_score = 0.8  # 阈值
        for alias, eid in self.alias_index.items():
            if self.kb[eid]['type'] == entity_type:
                score = self._similarity(entity_text, alias)
                if score > best_score:
                    best_score = score
                    best_match = eid
        return best_match

    def _similarity(self, s1, s2):
        """简单的字符级相似度"""
        if not s1 or not s2:
            return 0
        common = len(set(s1) & set(s2))
        total = len(set(s1) | set(s2))
        return common / total if total > 0 else 0

# 使用
kb = {
    "ORG_001": {"name": "阿里巴巴集团", "aliases": ["阿里", "阿里巴巴", "Alibaba"], "type": "ORG"},
    "ORG_002": {"name": "北京大学", "aliases": ["北大", "燕京大学"], "type": "ORG"},
    "LOC_001": {"name": "北京市", "aliases": ["北京", "北平"], "type": "LOC"},
}

linker = EntityLinker(kb)
print(f"阿里 -> {linker.link('阿里', 'ORG')}")
print(f"北大 -> {linker.link('北大', 'ORG')}")

多模态 NER

# 多模态 NER：结合文本和图像信息
# 在社交媒体、新闻等场景中，图片信息可以辅助实体识别

# 思路：
# 1. 使用 BERT 提取文本特征
# 2. 使用 ResNet/CLIP 提取图像特征
# 3. 融合两种特征进行 NER 预测

class MultimodalNER(nn.Module):
    """多模态 NER 模型（文本 + 图像）"""
    def __init__(self, bert_model, num_labels=9, image_dim=2048, hidden_dim=256):
        super().__init__()
        self.bert = bert_model
        self.image_projection = nn.Linear(image_dim, hidden_dim)
        self.fusion_layer = nn.Linear(768 + hidden_dim, 768)
        self.classifier = nn.Linear(768, num_labels)

    def forward(self, input_ids, attention_mask, image_features):
        # 文本特征
        text_output = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        text_hidden = text_output.last_hidden_state  # (B, L, 768)

        # 图像特征
        image_proj = self.image_projection(image_features)  # (B, hidden_dim)
        image_proj = image_proj.unsqueeze(1).expand(-1, text_hidden.size(1), -1)

        # 特征融合
        fused = torch.cat([text_hidden, image_proj], dim=-1)
        fused = self.fusion_layer(fused)

        # 分类
        logits = self.classifier(fused)
        return logits

常见标注工具与工作流

NER 标注工具推荐：

1. Doccano（开源）
   - 支持序列标注、文本分类、机器翻译
   - 轻量级，易于部署
   - pip install doccano

2. Label Studio（开源）
   - 支持多种标注任务（文本、图像、音频）
   - 可集成 ML Backend 进行主动学习
   - 支持团队协作和质量管理

3. Prodogy（商业）
   - 支持主动学习（模型辅助标注）
   - 内置 NER、文本分类等模板
   - 支持规则和模型混合标注

标注工作流建议：
1. 制定标注规范（含边界样例和争议处理规则）
2. 两位标注员独立标注 100 条样本
3. 计算 IAA（Cohen's Kappa），目标 > 0.8
4. 讨论 and 解决不一致的标注
5. 修订标注规范
6. 大规模标注（分配给标注团队）
7. 质量抽检（抽 10% 双人标注复核）