命名实体识别 NER

# BIO 标注：句子到标签序列的转换
def bio_label_spans(tags):
    """从 BIO 标签序列中提取实体片段"""
    entities = []
    current_type = None
    start = None
    
    for i, tag in enumerate(tags):
        if tag.startswith('B-'):
            # 新实体开始
            if current_type is not None:
                entities.append((current_type, start, i - 1))
            current_type = tag[2:]
            start = i
        elif tag.startswith('I-'):
            tag_type = tag[2:]
            if current_type != tag_type:
                # I- 标签与当前实体类型不匹配，视为新实体
                if current_type is not None:
                    entities.append((current_type, start, i - 1))
                current_type = tag_type
                start = i
        else:
            # O 标签，结束当前实体
            if current_type is not None:
                entities.append((current_type, start, i - 1))
                current_type = None
                start = None
    
    if current_type is not None:
        entities.append((current_type, start, len(tags) - 1))
    
    return entities

# 示例
tokens = ["张", "三", "在", "北", "京", "大", "学", "工", "作"]
tags   = ["B-PER", "I-PER", "O", "B-LOC", "I-LOC", "I-LOC", "I-LOC", "O", "O"]

entities = bio_label_spans(tags)
print("=== BIO 实体提取 ===")
for ent_type, start, end in entities:
    text = "".join(tokens[start:end + 1])
    print(f"  [{ent_type}] {text} (位置: {start}-{end})")

# BIOES 标注体系与 BIO 对比
def bio_to_bioes(tags):
    """将 BIO 标签转换为 BIOES"""
    bioes = []
    n = len(tags)
    
    for i, tag in enumerate(tags):
        if tag == 'O':
            bioes.append('O')
        elif tag.startswith('B-'):
            ent_type = tag[2:]
            if i + 1 < n and tags[i + 1].startswith('I-' + ent_type):
                bioes.append(f'B-{ent_type}')
            else:
                bioes.append(f'S-{ent_type}')  # 单字实体
        elif tag.startswith('I-'):
            ent_type = tag[2:]
            if i + 1 < n and tags[i + 1].startswith('I-' + ent_type):
                bioes.append(f'I-{ent_type}')
            else:
                bioes.append(f'E-{ent_type}')  # 实体结束
        else:
            bioes.append(tag)
    
    return bioes

# 对比 BIO vs BIOES
tokens = ["北", "京", "奥", "运", "会"]
bio    = ["B-EVENT", "I-EVENT", "I-EVENT", "I-EVENT", "I-EVENT"]
bioes  = bio_to_bioes(bio)

print("=== BIO vs BIOES 对比 ===")
for t, b, be in zip(tokens, bio, bioes):
    print(f"  {t}: BIO={b:<12} BIOES={be}")

print("\nBIOES 额外约束:")
print("  1. B- 后面必须是 I- 或 E-（不能直接跟 B-）")
print("  2. I- 后面必须是 I- 或 E-（不能直接跟 B- 或 O）")
print("  3. S- 和 E- 后面必须是 B- 或 S- 或 O")
print("  4. 这些约束在 CRF 解码时自动 enforced")

import numpy as np

# HMM 简化实现
class NERHMM:
    """基于 HMM 的 NER 模型"""
    
    def __init__(self, states, observations):
        self.states = states  # 标签集合 [O, B-PER, I-PER, B-LOC, I-LOC, ...]
        self.obs = observations  # 词汇表
        self.n_states = len(states)
        self.n_obs = len(observations)
        self.s2i = {s: i for i, s in enumerate(states)}
        self.o2i = {w: i for i, w in enumerate(observations)}
    
    def train(self, sequences, labels):
        """用监督数据估计 HMM 参数"""
        # 初始概率
        pi = np.zeros(self.n_states) + 1e-6  # 平滑
        # 转移矩阵
        A = np.zeros((self.n_states, self.n_states)) + 1e-6
        # 发射矩阵
        B = np.zeros((self.n_states, self.n_obs)) + 1e-6
        
        for seq, labs in zip(sequences, labels):
            pi[self.s2i[labs[0]]] += 1
            for t in range(len(seq)):
                s = self.s2i[labs[t]]
                o = self.o2i.get(seq[t], 0)  # UNK → 0
                B[s, o] += 1
                if t > 0:
                    prev_s = self.s2i[labs[t - 1]]
                    A[prev_s, s] += 1
        
        # 归一化
        self.pi = pi / pi.sum()
        self.A = A / A.sum(axis=1, keepdims=True)
        self.B = B / B.sum(axis=1, keepdims=True)
    
    def viterbi_decode(self, obs_seq):
        """Viterbi 解码：找到最优标签序列"""
        T = len(obs_seq)
        N = self.n_states
        
        # δ[t][s]: 时刻 t 到达状态 s 的最大概率
        delta = np.zeros((T, N))
        psi = np.zeros((T, N), dtype=int)
        
        # 初始化
        o0 = self.o2i.get(obs_seq[0], 0)
        delta[0] = np.log(self.pi + 1e-10) + np.log(self.B[:, o0] + 1e-10)
        
        # 递推
        for t in range(1, T):
            o_t = self.o2i.get(obs_seq[t], 0)
            for s in range(N):
                scores = delta[t - 1] + np.log(self.A[:, s] + 1e-10)
                psi[t, s] = np.argmax(scores)
                delta[t, s] = scores[psi[t, s]] + np.log(self.B[s, o_t] + 1e-10)
        
        # 回溯
        path = [0] * T
        path[T - 1] = np.argmax(delta[T - 1])
        for t in range(T - 2, -1, -1):
            path[t] = psi[t + 1, path[t + 1]]
        
        return [self.states[s] for s in path]

# 训练示例
states = ["O", "B-PER", "I-PER", "B-LOC", "I-LOC"]
vocab = ["张", "三", "李", "四", "在", "北", "京", "上", "工", "作", "<UNK>"]

hmm = NERHMM(states, vocab)
seqs = [["张", "三", "在", "北", "京"], ["李", "四", "在", "上", "海"]]
labs = [["B-PER", "I-PER", "O", "B-LOC", "I-LOC"],
        ["B-PER", "I-PER", "O", "B-LOC", "I-LOC"]]
hmm.train(seqs, labs)

# 测试
test = ["张", "三", "在", "北", "京"]
pred = hmm.viterbi_decode(test)
print("=== HMM Viterbi 解码 ===")
for t, p in zip(test, pred):
    print(f"  {t} → {p}")

import numpy as np

# CRF 特征函数设计
class CRFFeatureExtractor:
    """CRF 特征提取器"""
    
    def __init__(self):
        self.feature_funcs = [
            self._unigram_word,       # 当前词
            self._unigram_suffix,     # 当前词后缀
            self._unigram_capital,    # 当前词是否大写
            self._bigram_prev_word,   # 前一个词
            self._bigram_next_word,   # 后一个词
            self._word_shape,         # 词形模式
        ]
    
    def _unigram_word(self, tokens, i, tag):
        return f"UW={tokens[i]},{tag}"
    
    def _unigram_suffix(self, tokens, i, tag):
        word = tokens[i]
        suffix = word[-2:] if len(word) >= 2 else word
        return f"US={suffix},{tag}"
    
    def _unigram_capital(self, tokens, i, tag):
        is_cap = tokens[i][0].isupper() if tokens[i] else False
        return f"UC={is_cap},{tag}"
    
    def _bigram_prev_word(self, tokens, i, tag):
        if i == 0:
            return None
        return f"BPW={tokens[i-1]},{tag}"
    
    def _bigram_next_word(self, tokens, i, tag):
        if i == len(tokens) - 1:
            return None
        return f"BNW={tokens[i+1]},{tag}"
    
    def _word_shape(self, tokens, i, tag):
        word = tokens[i]
        shape = "".join(
            "X" if c.isupper() else "x" if c.islower() else "d" if c.isdigit() else c
            for c in word
        )
        return f"WS={shape},{tag}"
    
    def extract(self, tokens, tag_seq):
        """提取所有特征"""
        features = {}
        for i, tag in enumerate(tag_seq):
            for func in self.feature_funcs:
                f = func(tokens, i, tag)
                if f:
                    features[f] = features.get(f, 0) + 1
            # 转移特征
            if i > 0:
                trans = f"T:{tag_seq[i-1]}→{tag}"
                features[trans] = features.get(trans, 0) + 1
        return features

# 演示特征提取
extractor = CRFFeatureExtractor()
tokens = ["John", "Smith", "works", "in", "Beijing"]
tags = ["B-PER", "I-PER", "O", "O", "B-LOC"]

features = extractor.extract(tokens, tags)
print("=== CRF 特征提取示例 ===")
for feat, count in sorted(features.items())[:15]:
    print(f"  {feat}: {count}")
print(f"  ... 共 {len(features)} 个特征")

import numpy as np

class BiLSTMCRFSimplified:
    """BiLSTM-CRF 简化演示（非训练，仅展示架构）"""
    
    def __init__(self, vocab_size, embed_dim, hidden_dim, n_tags):
        self.vocab_size = vocab_size
        self.embed_dim = embed_dim
        self.hidden_dim = hidden_dim
        self.n_tags = n_tags
        
        # 词嵌入层
        self.embedding = np.random.randn(vocab_size, embed_dim) * 0.1
        
        # BiLSTM 权重 (简化: 单层)
        self.W_f = np.random.randn(embed_dim + hidden_dim, hidden_dim) * 0.1
        self.W_b = np.random.randn(embed_dim + hidden_dim, hidden_dim) * 0.1
        
        # 发射层: BiLSTM 输出 → 标签分数
        self.W_emit = np.random.randn(2 * hidden_dim, n_tags) * 0.1
        
        # CRF 转移矩阵
        self.trans = np.random.randn(n_tags, n_tags) * 0.1
        # 强制设置非法转移为大负数
        # I-PER 不能跟在 O 后面 (假设 O=0, B-PER=1, I-PER=2)
        self.trans[0, 2] = -100  # O → I-PER 非法
        self.trans[0, 4] = -100  # O → I-LOC 非法
    
    def get_emission_scores(self, sentence):
        """BiLSTM 生成发射分数矩阵"""
        T = len(sentence)
        d = self.hidden_dim
        
        # 简化: 用随机模拟 BiLSTM 输出
        # 实际中这里跑前向和反向 LSTM
        forward_h = np.random.randn(T, d) * 0.5
        backward_h = np.random.randn(T, d) * 0.5
        
        # 拼接双向隐状态
        bi_h = np.concatenate([forward_h, backward_h], axis=1)  # (T, 2d)
        
        # 发射分数: (T, n_tags)
        emissions = bi_h @ self.W_emit
        return emissions
    
    def viterbi_decode(self, emissions):
        """CRF Viterbi 解码"""
        T, K = emissions.shape
        
        # DP 表
        dp = np.full((T, K), -np.inf)
        backpointers = np.zeros((T, K), dtype=int)
        
        # 初始化
        dp[0] = emissions[0]
        
        # 递推
        for t in range(1, T):
            for j in range(K):
                scores = dp[t - 1] + self.trans[:, j] + emissions[t, j]
                backpointers[t, j] = np.argmax(scores)
                dp[t, j] = scores[backpointers[t, j]]
        
        # 回溯
        path = [0] * T
        path[T - 1] = np.argmax(dp[T - 1])
        for t in range(T - 2, -1, -1):
            path[t] = backpointers[t + 1, path[t + 1]]
        
        return path, dp[T - 1, path[T - 1]]

# 演示
model = BiLSTMCRFSimplified(vocab_size=1000, embed_dim=100, hidden_dim=64, n_tags=5)
tags = ["O", "B-PER", "I-PER", "B-LOC", "I-LOC"]
sentence = [0, 1, 2, 3, 4]  # 词索引

emissions = model.get_emission_scores(sentence)
path, score = model.viterbi_decode(emissions)

print("=== BiLSTM-CRF 解码 ===")
for i, p in enumerate(sentence):
    print(f"  词 {p} → {tags[path[i]]} (发射分数: {emissions[i, path[i]]:.3f})")
print(f"  总得分: {score:.3f}")

import numpy as np

# 字符级 CNN 编码器
class CharCNN:
    """字符级 CNN: 从字符序列生成词表示"""
    
    def __init__(self, char_vocab_size, char_embed_dim, num_filters=30,
                 filter_sizes=(3, 4, 5)):
        self.char_embed = np.random.randn(char_vocab_size, char_embed_dim) * 0.1
        self.num_filters = num_filters
        self.filter_sizes = filter_sizes
        
        # 每种 filter size 一组卷积核
        self.conv_weights = {}
        self.conv_biases = {}
        for fs in filter_sizes:
            self.conv_weights[fs] = np.random.randn(fs, char_embed_dim, num_filters) * 0.1
            self.conv_biases[fs] = np.zeros(num_filters)
    
    def encode_word(self, chars):
        """对单个词的字符序列编码"""
        # 字符嵌入: (L, char_d)
        char_indices = [min(c, self.char_embed.shape[0] - 1) for c in chars]
        char_embeds = self.char_embed[char_indices]  # (L, char_d)
        
        # 多尺度卷积 + max pooling
        representations = []
        for fs in self.filter_sizes:
            if len(chars) < fs:
                continue
            # 卷积: 滑动窗口
            conv_out = []
            for i in range(len(chars) - fs + 1):
                window = char_embeds[i:i + fs]  # (fs, char_d)
                # 简化的卷积操作
                feature_map = np.tensordot(window, self.conv_weights[fs], axes=([0, 1], [0, 1]))
                conv_out.append(feature_map)
            
            if conv_out:
                conv_out = np.array(conv_out)  # (L-fs+1, num_filters)
                # Max pooling over time
                pooled = conv_out.max(axis=0)  # (num_filters,)
                representations.append(pooled)
        
        # 拼接所有 filter size 的 pooled 向量
        if representations:
            return np.concatenate(representations)
        return np.zeros(self.num_filters * len(self.filter_sizes))
    
    def encode_sentence(self, sentence_chars):
        """对句子中所有词编码"""
        return np.array([self.encode_word(chars) for chars in sentence_chars])

# 演示
char_vocab = {"<PAD>": 0, "<UNK>": 1}
for c in "abcdefghijklmnopqrstuvwxyz":
    char_vocab[c] = len(char_vocab)

char_cnn = CharCNN(char_vocab_size=len(char_vocab), char_embed_dim=10)

words = [[char_vocab.get(c, 1) for c in "unhappiness"],
         [char_vocab.get(c, 1) for c in "running"],
         [char_vocab.get(c, 1) for c in "the"]]

print("=== 字符级 CNN 编码 ===")
for chars in words:
    rep = char_cnn.encode_word(chars)
    print(f"  词长={len(chars):2d} → 表示维度={len(rep):3d} (前5维: {rep[:5]})")

from transformers import BertTokenizer, BertForTokenClassification
import torch

# BERT NER 推理示例
def ner_with_bert(text, model_name="bert-base-chinese"):
    """用 BERT 做中文 NER"""
    tokenizer = BertTokenizer.from_pretrained(model_name)
    model = BertForTokenClassification.from_pretrained(
        model_name, num_labels=9  # O, B-PER, I-PER, B-LOC, I-LOC, B-ORG, I-ORG, B-MISC, I-MISC
    )
    model.eval()
    
    # 分词
    tokens = tokenizer.tokenize(text)
    input_ids = tokenizer.convert_tokens_to_ids(tokens)
    
    # 添加 [CLS] 和 [SEP]
    input_ids = [tokenizer.cls_token_id] + input_ids + [tokenizer.sep_token_id]
    attention_mask = [1] * len(input_ids)
    
    # 推理
    with torch.no_grad():
        outputs = model(
            input_ids=torch.tensor([input_ids]),
            attention_mask=torch.tensor([attention_mask])
        )
        predictions = torch.argmax(outputs.logits, dim=2)[0]
    
    # 标签映射
    label_map = {
        0: "O", 1: "B-PER", 2: "I-PER", 3: "B-LOC", 4: "I-LOC",
        5: "B-ORG", 6: "I-ORG", 7: "B-MISC", 8: "I-MISC"
    }
    
    # 提取实体（跳过 [CLS] 和 [SEP]）
    results = []
    for token, pred in zip(tokens, predictions[1:-1]):
        results.append((token, label_map[pred.item()]))
    
    return results

# 模拟输出
print("=== BERT NER 推理模拟 ===")
text = "张三在北京大学工作"
tokens = ["张", "三", "在", "北", "京", "大", "学", "工", "作"]
preds  = ["B-PER", "I-PER", "O", "B-ORG", "I-ORG", "I-ORG", "I-ORG", "O", "O"]

print(f"输入: {text}")
for t, p in zip(tokens, preds):
    print(f"  {t} → {p}")

import numpy as np

# WordPiece 分词对 NER 的影响
def simulate_wordpiece_split(word, max_len=10):
    """模拟 WordPiece 分词"""
    # 简化的子词拆分规则
    prefixes = ["un", "re", "dis", "pre", "mis", "over", "under"]
    suffixes = ["ness", "tion", "ment", "ing", "ed", "ly", "er", "able"]
    
    subwords = []
    rest = word
    
    # 检查前缀
    for p in prefixes:
        if rest.startswith(p) and len(rest) > len(p):
            subwords.append(p)
            rest = rest[len(p):]
            break
    
    # 剩余部分按音节拆分
    while len(rest) > 3:
        subwords.append("##" + rest[:3])
        rest = rest[3:]
    if rest:
        subwords.append("##" + rest if subwords else rest)
    
    return subwords if subwords else [word]

# 分析 NER 标签对齐问题
print("=== WordPiece 子词对齐 ===")
words = ["unhappiness", "restructuring", "preprocessing", "Beijing"]
ner_labels = ["B-MISC", "B-MISC", "B-MISC", "B-LOC"]

for word, label in zip(words, ner_labels):
    subwords = simulate_wordpiece_split(word)
    print(f"\n  词: {word} → {subwords}")
    print(f"  首子词策略:")
    for i, sw in enumerate(subwords):
        if i == 0:
            print(f"    {sw:<20} → {label}")
        else:
            print(f"    {sw:<20} → X (忽略)")

# 统计子词膨胀率
all_words = ["The", "quick", "brown", "fox", "unhappiness", "restructuring",
             "preprocessing", "telecommunications", "internationalization"]
total_tokens = 0
total_words = len(all_words)
for w in all_words:
    sws = simulate_wordpiece_split(w)
    total_tokens += len(sws)
print(f"\n  词数: {total_words} → Token数: {total_tokens} (膨胀率: {total_tokens/total_words:.1f}x)")

import numpy as np

# Span-based NER: 枚举所有可能的实体片段
def span_based_ner(tokens, max_len=10):
    """基于片段枚举的 NER（支持嵌套）"""
    n = len(tokens)
    spans = []
    
    # 枚举所有可能的 (start, end) 片段
    for start in range(n):
        for end in range(start, min(start + max_len, n)):
            spans.append((start, end))
    
    return spans

def classify_span(tokens, start, end, model=None):
    """对单个片段做实体分类"""
    span_text = "".join(tokens[start:end + 1])
    # 模拟分类
    entity_types = ["PER", "LOC", "ORG", "MISC", "NEG"]
    
    # 实际中这里用 BERT 编码 span 并分类
    # 模拟: 用长度启发式
    np.random.seed(hash(span_text) % (2**32))
    probs = np.random.dirichlet([1, 1, 1, 1, 3])  # 偏向 NEG
    
    best_idx = np.argmax(probs)
    return entity_types[best_idx], probs

# 演示嵌套实体识别
print("=== Span-based 嵌套 NER ===")
tokens = ["北", "京", "大", "学", "的", "计", "算", "机", "科", "学", "系"]
spans = span_based_ner(tokens, max_len=5)

print(f"句子: {''.join(tokens)}")
print(f"总片段数: {len(spans)}\n")

# 找出可能的实体
entities = []
for start, end in spans:
    entity_type, probs = classify_span(tokens, start, end)
    if entity_type != "NEG":
        text = "".join(tokens[start:end + 1])
        confidence = probs.max()
        entities.append((entity_type, start, end, text, confidence))
        print(f"  [{entity_type}] {text} (位置: {start}-{end}, 置信度: {confidence:.2f})")

# 检测嵌套
print("\n嵌套关系:")
for i, (t1, s1, e1, text1, _) in enumerate(entities):
    for j, (t2, s2, e2, text2, _) in enumerate(entities):
        if i < j and s1 <= s2 and e2 <= e1 and (s1 != s2 or e1 != e2):
            print(f"  [{text1}] 嵌套包含 [{text2}]")

import numpy as np

# 文档级 NER: 跨句实体识别
class DocumentNER:
    """文档级 NER - 处理跨句实体"""
    
    def __init__(self, max_doc_len=512, sentence_overlap=50):
        self.max_doc_len = max_doc_len
        self.sentence_overlap = sentence_overlap
    
    def split_document(self, sentences):
        """将文档切分为可处理的片段（带重叠）"""
        chunks = []
        current_chunk = []
        current_len = 0
        
        for sent in sentences:
            sent_len = len(sent.split())
            if current_len + sent_len > self.max_doc_len:
                # 当前 chunk 已满，保存并添加重叠
                overlap_tokens = []
                for prev_sent in reversed(current_chunk):
                    tokens = prev_sent.split()
                    overlap_tokens = tokens[-self.sentence_overlap:] + overlap_tokens
                    if len(overlap_tokens) >= self.sentence_overlap:
                        break
                
                chunks.append({
                    "tokens": current_chunk,
                    "overlap": " ".join(overlap_tokens) if overlap_tokens else "",
                })
                current_chunk = [sent]
                current_len = sent_len
            else:
                current_chunk.append(sent)
                current_len += sent_len
        
        if current_chunk:
            chunks.append({"tokens": current_chunk, "overlap": ""})
        
        return chunks
    
    def merge_entities(self, chunk_entities):
        """合并各 chunk 的实体结果，去重"""
        all_entities = {}
        
        for chunk_idx, entities in enumerate(chunk_entities):
            for ent_type, start, end, text, conf in entities:
                # 全局位置索引
                global_key = (ent_type, start, end)
                if global_key not in all_entities:
                    all_entities[global_key] = []
                all_entities[global_key].append((text, conf, chunk_idx))
        
        # 合并: 同一实体取最高置信度
        merged = []
        for (ent_type, start, end), mentions in all_entities.items():
            best_mention = max(mentions, key=lambda x: x[1])
            text, conf, _ = best_mention
            merged.append((ent_type, start, end, text, conf))
        
        return merged

# 演示
doc_ner = DocumentNER(max_doc_len=100, sentence_overlap=20)
sentences = [
    "张三毕业于北京大学计算机科学系。",
    "他在该校攻读博士学位期间发表了多篇论文。",
    "毕业后，他加入了微软亚洲研究院。",
    "张三目前主要从事自然语言处理研究。",
]

chunks = doc_ner.split_document(sentences)
print("=== 文档级 NER: 跨句实体 ===")
for i, chunk in enumerate(chunks):
    print(f"\n  Chunk {i}:")
    print(f"    句子: {chunk['tokens']}")
    if chunk['overlap']:
        print(f"    重叠: {chunk['overlap'][:30]}...")

print("\n  跨句实体合并:")
print("  '张三' 在句子 1 和句子 4 中都出现")
print("  '他'(句子 2) 指代 '张三' → 需要共指消解")
print("  '该校'(句子 2) 指代 '北京大学' → 需要共指消解")

from collections import defaultdict

class NEREvaluator:
    """NER 实体级评估器"""
    
    def __init__(self):
        self.tp = 0
        self.fp = 0
        self.fn = 0
        self.per_type = defaultdict(lambda: {"tp": 0, "fp": 0, "fn": 0})
        self.confusion = defaultdict(lambda: defaultdict(int))
    
    def evaluate(self, gold_entities, pred_entities):
        """评估一组实体的精确率/召回率/F1
        实体格式: (type, start, end) 集合
        """
        gold_set = set(gold_entities)
        pred_set = set(pred_entities)
        
        # 严格匹配
        correct = gold_set & pred_set
        self.tp += len(correct)
        self.fp += len(pred_set - gold_set)
        self.fn += len(gold_set - pred_set)
        
        # 按类型统计
        for ent_type in set(e[0] for e in gold_entities) | set(e[0] for e in pred_entities):
            gold_type = {e for e in gold_entities if e[0] == ent_type}
            pred_type = {e for e in pred_entities if e[0] == ent_type}
            self.per_type[ent_type]["tp"] += len(gold_type & pred_type)
            self.per_type[ent_type]["fp"] += len(pred_type - gold_type)
            self.per_type[ent_type]["fn"] += len(gold_type - pred_type)
        
        # 混淆矩阵
        for pred_ent in pred_entities:
            if pred_ent not in gold_set:
                # 预测错误: 找最相似的真实实体
                best_gold = self._find_best_overlap(pred_ent, gold_entities)
                if best_gold and best_gold[0] != pred_ent[0]:
                    self.confusion[best_gold[0]][pred_ent[0]] += 1
    
    def _find_best_overlap(self, pred, gold_entities):
        """找与预测实体重叠最大的真实实体"""
        best = None
        best_overlap = 0
        for g in gold_entities:
            overlap_start = max(pred[1], g[1])
            overlap_end = min(pred[2], g[2])
            overlap = max(0, overlap_end - overlap_start + 1)
            if overlap > best_overlap:
                best_overlap = overlap
                best = g
        return best
    
    def metrics(self):
        """计算所有指标"""
        def f1(tp, fp, fn):
            p = tp / (tp + fp) if (tp + fp) > 0 else 0
            r = tp / (tp + fn) if (tp + fn) > 0 else 0
            f = 2 * p * r / (p + r) if (p + r) > 0 else 0
            return p, r, f
        
        overall_p, overall_r, overall_f1 = f1(self.tp, self.fp, self.fn)
        
        per_type_metrics = {}
        for ent_type, counts in self.per_type.items():
            p, r, f = f1(counts["tp"], counts["fp"], counts["fn"])
            per_type_metrics[ent_type] = {"P": p, "R": r, "F1": f}
        
        return {
            "overall": {"P": overall_p, "R": overall_r, "F1": overall_f1},
            "per_type": per_type_metrics,
            "confusion": dict(self.confusion),
        }

# 演示评估
evaluator = NEREvaluator()
gold = [
    ("PER", 0, 1), ("LOC", 3, 4), ("ORG", 7, 10),
]
pred = [
    ("PER", 0, 1), ("LOC", 3, 5), ("ORG", 7, 9), ("MISC", 12, 13),
]

evaluator.evaluate(gold, pred)
metrics = evaluator.metrics()

print("=== NER 评估结果 ===")
print(f"总体: P={metrics['overall']['P']:.3f}, R={metrics['overall']['R']:.3f}, F1={metrics['overall']['F1']:.3f}")
print(f"  TP={evaluator.tp}, FP={evaluator.fp}, FN={evaluator.fn}")
print("\n按类别:")
for t, m in metrics["per_type"].items():
    print(f"  {t}: P={m['P']:.3f}, R={m['R']:.3f}, F1={m['F1']:.3f}")

import numpy as np
from collections import defaultdict

# 严格匹配 vs 宽松匹配 vs Token 级评估对比
def compare_evaluation_methods(gold_entities, pred_entities, seq_len, gold_labels, pred_labels):
    """对比三种评估方式"""
    gold_set = set(gold_entities)
    pred_set = set(pred_entities)
    
    # 1. 严格匹配 (Exact Match)
    exact_tp = len(gold_set & pred_set)
    exact_fp = len(pred_set - gold_set)
    exact_fn = len(gold_set - pred_set)
    
    # 2. 宽松匹配 (Partial Match) - 只要有重叠就算 TP
    partial_tp = 0
    for p in pred_set:
        for g in gold_set:
            if p[0] == g[0]:  # 类型相同
                # 检查是否有重叠
                if max(p[1], g[1]) <= min(p[2], g[2]):
                    partial_tp += 1
                    break
    partial_fp = len(pred_set) - partial_tp
    partial_fn = len(gold_set) - partial_tp
    
    # 3. Token 级别评估
    correct_tokens = sum(1 for g, p in zip(gold_labels, pred_labels) if g == p)
    token_acc = correct_tokens / seq_len if seq_len > 0 else 0
    
    def f1(tp, fp, fn):
        p = tp / (tp + fp) if (tp + fp) > 0 else 0
        r = tp / (tp + fn) if (tp + fn) > 0 else 0
        f = 2 * p * r / (p + r) if (p + r) > 0 else 0
        return p, r, f
    
    ep, er, ef = f1(exact_tp, exact_fp, exact_fn)
    pp, pr, pf = f1(partial_tp, partial_fp, partial_fn)
    
    print("=== 三种评估方式对比 ===")
    print(f"{'评估方式':<15} {'精确率':>8} {'召回率':>8} {'F1':>8}")
    print("-" * 45)
    print(f"{'严格匹配':<12} {ep:>8.3f} {er:>8.3f} {ef:>8.3f}")
    print(f"{'宽松匹配':<12} {pp:>8.3f} {pr:>8.3f} {pf:>8.3f}")
    print(f"{'Token准确率':<11} {'':>8} {'':>8} {token_acc:>8.3f}")

# 模拟示例
gold_ents = [("PER", 0, 1), ("LOC", 4, 7)]
pred_ents = [("PER", 0, 1), ("LOC", 4, 6)]  # 边界差 1 个 token
gold_labels = ["B-PER", "I-PER", "O", "O", "B-LOC", "I-LOC", "I-LOC", "I-LOC", "O", "O"]
pred_labels = ["B-PER", "I-PER", "O", "O", "B-LOC", "I-LOC", "I-LOC", "O", "O", "O"]

compare_evaluation_methods(gold_ents, pred_ents, 10, gold_labels, pred_labels)
print("\n分析:")
print("  宽松匹配 F1 高于严格匹配 → 边界有轻微偏差但实体基本正确")
print("  Token 准确率很高 → 但掩盖了实体边界错误")
print("  结论: 必须看严格匹配的 F1!")

# spaCy NER: 训练自定义 NER 模型
import spacy
from spacy.training import Example
import random

def train_spacy_ner(train_data, model_name="zh_core_web_sm", n_iter=20):
    """用 spaCy 训练自定义 NER 模型"""
    # 加载预训练模型
    try:
        nlp = spacy.load(model_name)
    except OSError:
        nlp = spacy.blank("zh")
    
    # 创建或更新 NER pipeline
    if "ner" not in nlp.pipe_names:
        ner = nlp.add_pipe("ner")
    else:
        ner = nlp.get_pipe("ner")
    
    # 添加实体标签
    for _, annotations in train_data:
        for ent in annotations.get("entities", []):
            ner.add_label(ent[2])
    
    # 训练
    optimizer = nlp.begin_training()
    
    for i in range(n_iter):
        random.shuffle(train_data)
        losses = {}
        
        for text, annotations in train_data:
            doc = nlp.make_doc(text)
            example = Example.from_dict(doc, annotations)
            nlp.update([example], drop=0.5, losses=losses)
        
        if i % 5 == 0:
            print(f"  Iteration {i:3d} | Loss: {losses.get('ner', 0):.4f}")
    
    return nlp

# 训练数据
TRAIN_DATA = [
    ("张三在北京大学工作", {"entities": [(0, 2, "PER"), (3, 7, "ORG")]}),
    ("李明是上海人", {"entities": [(0, 2, "PER"), (4, 6, "LOC")]}),
    ("腾讯总部在深圳", {"entities": [(0, 2, "ORG"), (5, 7, "LOC")]}),
    ("王五毕业于清华大学", {"entities": [(0, 2, "PER"), (5, 9, "ORG")]}),
    ("阿里巴巴在杭州", {"entities": [(0, 4, "ORG"), (5, 7, "LOC")]}),
]

print("=== spaCy NER 训练 ===")
print("（模拟输出，需要安装 spacy 和中文模型）")
for i in range(0, 20, 5):
    loss = max(0, 50 - i * 8 + random.uniform(-2, 2))
    print(f"  Iteration {i:3d} | Loss: {loss:.4f}")

print("\n模型保存与加载:")
print("  nlp.to_disk('./ner_model')")
print("  nlp = spacy.load('./ner_model')")
print("  doc = nlp('新句子')")
print("  for ent in doc.ents:")
print("      print(ent.text, ent.label_)")

# HuggingFace Transformers: 微调 BERT 做 NER
from transformers import (
    AutoTokenizer, AutoModelForTokenClassification,
    TrainingArguments, Trainer, DataCollatorForTokenClassification
)
import numpy as np

# 1. 准备数据
def prepare_ner_dataset(tokenized_texts, tokenized_labels, label_list):
    """将文本和标签转换为 HuggingFace Dataset"""
    label2id = {label: i for i, label in enumerate(label_list)}
    id2label = {i: label for label, i in label2id.items()}
    
    # 对齐标签到 token 级别
    def align_labels(words, labels, tokens, tokenizer):
        word_ids = tokens.word_ids()
        aligned_labels = []
        previous_word_idx = None
        
        for word_idx in word_ids:
            if word_idx is None:
                aligned_labels.append(-100)  # 忽略 [CLS], [SEP]
            elif word_idx != previous_word_idx:
                aligned_labels.append(label2id[labels[word_idx]])
            else:
                # 同一词的后续子词: 继承前一个子词的标签
                aligned_labels.append(label2id[labels[word_idx]])
            previous_word_idx = word_idx
        
        return aligned_labels
    
    # 示例数据
    texts = ["张三在北京大学工作", "李明是上海人"]
    words_list = [list(t) for t in texts]
    labels_list = [
        ["B-PER", "I-PER", "O", "B-ORG", "I-ORG", "I-ORG", "I-ORG", "O", "O"],
        ["B-PER", "I-PER", "O", "B-LOC", "I-LOC", "O"],
    ]
    label_list = ["O", "B-PER", "I-PER", "B-ORG", "I-ORG", "B-LOC", "I-LOC"]
    
    tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
    encodings = tokenizer(texts, truncation=True, padding=True, is_split_into_words=False)
    
    # 简化的标签对齐（中文字符级）
    all_labels = []
    for encoding, labels in zip(encodings.encodings, labels_list):
        # 添加 -100 给 [CLS] 和 [SEP]
        aligned = [-100] + [label2id.get(l, 0) for l in labels] + [-100]
        all_labels.append(aligned)
    
    encodings["labels"] = all_labels
    
    return encodings, label2id, id2label

# 2. 训练配置
print("=== HuggingFace NER 微调 ===")
print("训练参数:")
print("  model: bert-base-chinese")
print("  learning_rate: 2e-5")
print("  per_device_train_batch_size: 16")
print("  num_train_epochs: 3")
print("  weight_decay: 0.01")
print("  warmup_ratio: 0.1")
print("\n代码流程:")
print("  tokenizer = AutoTokenizer.from_pretrained('bert-base-chinese')")
print("  model = AutoModelForTokenClassification.from_pretrained(")
print("      'bert-base-chinese', num_labels=7, id2label=id2label")
print("  )")
print("  trainer = Trainer(")
print("      model=model, args=training_args,")
print("      train_dataset=train_dataset,")
print("      data_collator=data_collator")
print("  )")
print("  trainer.train()")