27_依存解析详解：语法结构的精确表示

[主语] ← [动词] → [宾语]

def eisner_algorithm(scores, n):
    # 初始化动态规划表
    # dp[i][j][0] 表示i是j的左依存
    # dp[i][j][1] 表示j是i的右依存
    dp = [[[0.0, 0.0] for _ in range(n+1)] for __ in range(n+1)]
    back = [[[None, None] for _ in range(n+1)] for __ in range(n+1)]
    
    # 基本情况：长度为1的区间
    for i in range(n):
        j = i + 1
        dp[i][j][0] = scores[j][i]  # i是j的左支配词
        dp[i][j][1] = scores[i][j]  # j是i的右支配词
    
    # 填充动态规划表
    for length in range(2, n+1):
        for i in range(n - length + 1):
            j = i + length
            
            # 计算左依存情况
            best_score = -float('inf')
            best_k = -1
            for k in range(i, j):
                score = dp[i][k][1] + dp[k][j][0] + scores[j][i]
                if score > best_score:
                    best_score = score
                    best_k = k
            dp[i][j][0] = best_score
            back[i][j][0] = best_k
            
            # 计算右依存情况
            best_score = -float('inf')
            best_k = -1
            for k in range(i+1, j+1):
                score = dp[i][k][1] + dp[k][j][0] + scores[i][j]
                if score > best_score:
                    best_score = score
                    best_k = k
            dp[i][j][1] = best_score
            back[i][j][1] = best_k
    
    # 构建依存树
    arcs = []
    def backtrack(i, j, direction):
        if j <= i + 1:
            return
        
        k = back[i][j][direction]
        
        if direction == 0:  # i是j的左支配词
            arcs.append((j, i))
            backtrack(i, k, 1)
            backtrack(k, j, 0)
        else:  # j是i的右支配词
            arcs.append((i, j))
            backtrack(i, k, 1)
            backtrack(k, j, 0)
    
    backtrack(0, n, 1)
    return arcs

def mst_algorithm(scores, n):
    # Chu-Liu-Edmonds算法实现
    def find_cycle(parents, n):
        visited = [False] * (n + 1)
        rec_stack = [False] * (n + 1)
        parent = [0] * (n + 1)
        
        def dfs(node):
            visited[node] = True
            rec_stack[node] = True
            
            next_node = parents[node]
            if next_node != 0 and not visited[next_node]:
                parent[next_node] = node
                if dfs(next_node):
                    return True
            elif next_node != 0 and rec_stack[next_node]:
                # 找到环
                cycle = [next_node]
                current = node
                while current != next_node:
                    cycle.append(current)
                    current = parent[current]
                return cycle[::-1]  # 反转得到正确的顺序
            
            rec_stack[node] = False
            return None
        
        for i in range(1, n + 1):
            if not visited[i]:
                cycle = dfs(i)
                if cycle:
                    return cycle
        return None
    
    # 初始化父节点
    parents = [0] * (n + 1)
    
    # 为每个节点选择权重最大的入边
    for i in range(1, n + 1):
        max_score = -float('inf')
        max_parent = 0
        for j in range(0, n + 1):
            if i != j and scores[j][i] > max_score:
                max_score = scores[j][i]
                max_parent = j
        parents[i] = max_parent
    
    # 检测环
    cycle = find_cycle(parents, n)
    
    if not cycle:
        # 无环，直接返回
        return parents
    else:
        # 有环，收缩环并递归
        # 实现略...
        pass

class TransitionParser:
    def __init__(self):
        self.stack = []
        self.buffer = []
        self.arcs = []
    
    def initialize(self, tokens):
        # 初始化栈和缓冲区
        self.stack = [0]  # 0表示虚拟根节点
        self.buffer = list(range(1, len(tokens) + 1))
        self.arcs = []
    
    def is_terminal(self):
        # 检查是否达到终止状态
        return len(self.buffer) == 0 and len(self.stack) == 1
    
    def get_valid_actions(self):
        # 获取当前可用的操作
        actions = []
        
        # 如果缓冲区不为空，可以移进
        if self.buffer:
            actions.append('shift')
        
        # 如果栈中至少有一个实义词，可以左依存
        if len(self.stack) > 1:
            actions.append('left_arc')
        
        # 如果缓冲区不为空且栈中至少有一个词，可以右依存
        if self.buffer and len(self.stack) > 0:
            actions.append('right_arc')
        
        return actions
    
    def do_shift(self):
        # 执行移进操作
        word = self.buffer.pop(0)
        self.stack.append(word)
    
    def do_left_arc(self, label=None):
        # 执行左依存操作
        head = self.buffer[0]
        dependent = self.stack.pop()
        self.arcs.append((dependent, head, label))
    
    def do_right_arc(self, label=None):
        # 执行右依存操作
        head = self.stack[-1]
        dependent = self.buffer.pop(0)
        self.arcs.append((dependent, head, label))
        self.stack.append(dependent)
    
    def parse(self, tokens, oracle):
        # 使用oracle指导解析过程
        self.initialize(tokens)
        
        while not self.is_terminal():
            actions = self.get_valid_actions()
            # 选择操作
            action = oracle.select_action(self, actions)
            
            if action == 'shift':
                self.do_shift()
            elif action == 'left_arc':
                self.do_left_arc(oracle.get_label(self, action))
            elif action == 'right_arc':
                self.do_right_arc(oracle.get_label(self, action))
        
        return self.arcs

import torch
import torch.nn as nn
import torch.nn.functional as F

class GraphDependencyParser(nn.Module):
    def __init__(self, vocab_size, emb_dim, hidden_dim, num_labels):
        super(GraphDependencyParser, self).__init__()
        self.embedding = nn.Embedding(vocab_size, emb_dim)
        self.lstm = nn.LSTM(emb_dim, hidden_dim, bidirectional=True, batch_first=True)
        self.arc_scorer = nn.Linear(2 * hidden_dim, 2 * hidden_dim)
        self.label_scorer = nn.Linear(4 * hidden_dim, num_labels)
    
    def forward(self, tokens):
        # 词嵌入
        emb = self.embedding(tokens)
        # BiLSTM编码
        lstm_out, _ = self.lstm(emb)
        # 计算弧得分
        arc_scores = self._compute_arc_scores(lstm_out)
        # 计算标签得分
        label_scores = self._compute_label_scores(lstm_out)
        return arc_scores, label_scores
    
    def _compute_arc_scores(self, lstm_out):
        # 计算所有词对的弧得分
        # 使用双线性模型：h_j^T W h_i
        head = self.arc_scorer(lstm_out)
        # 计算得分矩阵
        # arc_scores[i, j] 表示i是j的支配词的得分
        return torch.matmul(head, lstm_out.transpose(1, 2))
    
    def _compute_label_scores(self, lstm_out):
        # 计算依存标签得分
        batch_size, seq_len, _ = lstm_out.size()
        # 为所有词对计算特征
        h_head = lstm_out.unsqueeze(2).expand(-1, -1, seq_len, -1)
        h_dep = lstm_out.unsqueeze(1).expand(-1, seq_len, -1, -1)
        combined = torch.cat([h_head, h_dep], dim=-1)
        # 计算标签得分
        return self.label_scorer(combined)
    
    def parse(self, tokens):
        # 解析函数
        arc_scores, label_scores = self.forward(tokens)
        # 使用MST算法寻找最优依存树
        # 这里简化实现，实际应调用完整的MST算法
        batch_size, seq_len, _ = tokens.size()
        arcs = []
        labels = []
        
        for i in range(batch_size):
            # 找到每个词的最佳支配词
            batch_arcs = []
            batch_labels = []
            for j in range(1, seq_len):  # 跳过虚拟根节点
                head_scores = arc_scores[i, :, j]
                best_head = torch.argmax(head_scores).item()
                batch_arcs.append((j, best_head))
                # 找到最佳标签
                best_label = torch.argmax(label_scores[i, best_head, j]).item()
                batch_labels.append(best_label)
            arcs.append(batch_arcs)
            labels.append(batch_labels)
        
        return arcs, labels

class TransitionDependencyParser(nn.Module):
    def __init__(self, vocab_size, emb_dim, hidden_dim, num_actions, num_labels):
        super(TransitionDependencyParser, self).__init__()
        self.embedding = nn.Embedding(vocab_size, emb_dim)
        self.lstm = nn.LSTM(emb_dim, hidden_dim, bidirectional=True, batch_first=True)
        self.state_encoder = nn.Linear(4 * hidden_dim, hidden_dim)
        self.action_classifier = nn.Linear(hidden_dim, num_actions)
        self.label_classifier = nn.Linear(hidden_dim, num_labels)
    
    def forward(self, tokens, stack, buffer):
        # 词嵌入
        emb = self.embedding(tokens)
        # BiLSTM编码
        lstm_out, _ = self.lstm(emb)
        # 构建状态表示
        state_repr = self._build_state_representation(lstm_out, stack, buffer)
        # 预测操作
        action_logits = self.action_classifier(state_repr)
        # 预测标签
        label_logits = self.label_classifier(state_repr)
        return action_logits, label_logits
    
    def _build_state_representation(self, lstm_out, stack, buffer):
        # 从栈和缓冲区中提取关键位置的表示
        # 简化实现，实际应根据解析器状态提取
        batch_size = lstm_out.size(0)
        state_reprs = []
        
        for i in range(batch_size):
            # 获取栈顶两个元素和缓冲区第一个元素的表示
            stack_elems = stack[i][-2:] if len(stack[i]) >= 2 else []
            buffer_elem = buffer[i][0] if buffer[i] else 0
            
            # 提取表示
            representations = []
            for elem in stack_elems:
                representations.append(lstm_out[i, elem])
            if buffer_elem:
                representations.append(lstm_out[i, buffer_elem])
            
            # 补齐长度
            while len(representations) < 3:
                representations.append(torch.zeros_like(lstm_out[i, 0]))
            
            # 组合表示
            state_repr = torch.cat(representations[:3], dim=0)
            state_reprs.append(state_repr)
        
        # 批处理
        state_repr_batch = torch.stack(state_reprs, dim=0)
        return F.relu(self.state_encoder(state_repr_batch))

UAS = 正确识别的无标记依存弧数量 / 总依存弧数量

LAS = 正确识别的有标记依存弧数量 / 总依存弧数量

Root Accuracy = 正确识别根节点的句子数量 / 总句子数量

def extract_relations(text, parser, entity_recognizer):
    # 识别实体
    entities = entity_recognizer.recognize(text)
    
    # 依存解析
    tokens, arcs, labels = parser.parse(text)
    
    # 构建依存图
    dependency_graph = build_dependency_graph(tokens, arcs, labels)
    
    # 提取实体关系
    relations = []
    for i, (e1_start, e1_end, e1_type) in enumerate(entities):
        for j, (e2_start, e2_end, e2_type) in enumerate(entities):
            if i != j:
                # 找到实体之间的最短依存路径
                path = find_shortest_path(dependency_graph, e1_end, e2_start)
                if path and is_valid_relation_path(path):
                    # 提取关系类型
                    relation_type = classify_relation_type(path)
                    relations.append({
                        'head': (e1_start, e1_end, e1_type),
                        'tail': (e2_start, e2_end, e2_type),
                        'type': relation_type,
                        'path': path
                    })
    
    return relations

def answer_question(question, context, parser):
    # 解析问题
    q_tokens, q_arcs, q_labels = parser.parse(question)
    
    # 识别问题类型
    question_type = identify_question_type(q_tokens, q_arcs, q_labels)
    
    # 识别问题焦点
    focus = identify_question_focus(q_tokens, q_arcs, q_labels)
    
    # 解析上下文
    c_tokens, c_arcs, c_labels = parser.parse(context)
    
    # 构建上下文的依存图
    context_graph = build_dependency_graph(c_tokens, c_arcs, c_labels)
    
    # 根据问题类型和焦点，从上下文提取答案
    # 简化实现，实际应使用更复杂的匹配策略
    candidate_answers = []
    if question_type == 'what':
        # 寻找名词短语
        noun_phrases = find_noun_phrases(context_graph)
        # 根据焦点过滤
        candidate_answers = filter_by_focus(noun_phrases, focus, context_graph)
    elif question_type == 'who':
        # 寻找人物实体
        person_entities = find_person_entities(context_graph)
        candidate_answers = filter_by_focus(person_entities, focus, context_graph)
    # 其他问题类型...
    
    # 选择最佳答案
    best_answer = select_best_answer(candidate_answers, question, context)
    
    return best_answer

def extractive_summary(text, parser, num_sentences=3):
    # 分句
    sentences = split_sentences(text)
    
    # 对每个句子进行依存解析
    parsed_sentences = []
    for sentence in sentences:
        tokens, arcs, labels = parser.parse(sentence)
        parsed_sentences.append((tokens, arcs, labels))
    
    # 计算句子得分
    scores = []
    for i, (tokens, arcs, labels) in enumerate(parsed_sentences):
        # 计算句子长度得分
        length_score = min(len(tokens) / 20, 1.0)
        
        # 计算句法中心性得分
        centrality_score = calculate_syntactic_centrality(arcs)
        
        # 计算位置得分
        position_score = 1.0 - (i / len(sentences))
        
        # 综合得分
        score = 0.4 * centrality_score + 0.3 * position_score + 0.3 * length_score
        scores.append(score)
    
    # 选择得分最高的句子
    top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:num_sentences]
    top_indices.sort()  # 保持原文顺序
    
    # 生成摘要
    summary = ' '.join([sentences[i] for i in top_indices])
    
    return summary

from transformers import BertModel, BertTokenizer
import torch
import torch.nn as nn

class BertDependencyParser(nn.Module):
    def __init__(self, bert_model_name, num_labels):
        super(BertDependencyParser, self).__init__()
        self.bert = BertModel.from_pretrained(bert_model_name)
        self.arc_scorer = nn.Linear(self.bert.config.hidden_size * 2, self.bert.config.hidden_size * 2)
        self.label_scorer = nn.Linear(self.bert.config.hidden_size * 4, num_labels)
    
    def forward(self, input_ids, attention_mask):
        # 获取BERT输出
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        last_hidden_state = outputs.last_hidden_state
        
        # 计算弧得分
        batch_size, seq_len, _ = last_hidden_state.size()
        head = self.arc_scorer(last_hidden_state)
        arc_scores = torch.bmm(head, last_hidden_state.transpose(1, 2))
        
        # 计算标签得分
        h_head = last_hidden_state.unsqueeze(2).expand(-1, -1, seq_len, -1)
        h_dep = last_hidden_state.unsqueeze(1).expand(-1, seq_len, -1, -1)
        combined = torch.cat([h_head, h_dep], dim=-1)
        label_scores = self.label_scorer(combined)
        
        return arc_scores, label_scores

请分析以下句子的依存句法结构，以"依存词	支配词	依存关系"的格式输出：
他迅速地完成了这项复杂的任务。

from transformers import AutoModelForTokenClassification, AutoTokenizer
from peft import get_peft_model, LoraConfig

# 加载预训练模型
model_name = "bert-base-uncased"
model = AutoModelForTokenClassification.from_pretrained(model_name, num_labels=num_dependency_labels)

# 配置LoRA
lora_config = LoraConfig(
    r=8,  # 秩
    lora_alpha=32,  # 缩放因子
    target_modules=["query", "value"],  # 目标模块
    lora_dropout=0.1,  # Dropout概率
    bias="none"  # 偏置处理方式
)

# 应用LoRA
peft_model = get_peft_model(model, lora_config)

# 训练模型
# 注意：此时只有LoRA参数是可训练的

def augment_dependency_tree(sentence, arcs, labels, parser):
    # 使用回译进行数据增强
    augmented_sentences = []
    augmented_arcs = []
    augmented_labels = []
    
    # 回译
    backtranslated = backtranslate(sentence)
    if backtranslated != sentence:
        # 解析回译后的句子
        try:
            _, new_arcs, new_labels = parser.parse(backtranslated)
            augmented_sentences.append(backtranslated)
            augmented_arcs.append(new_arcs)
            augmented_labels.append(new_labels)
        except:
            pass
    
    # 同义词替换
    synonym_replaced = replace_synonyms(sentence)
    if synonym_replaced != sentence:
        try:
            _, new_arcs, new_labels = parser.parse(synonym_replaced)
            augmented_sentences.append(synonym_replaced)
            augmented_arcs.append(new_arcs)
            augmented_labels.append(new_labels)
        except:
            pass
    
    return augmented_sentences, augmented_arcs, augmented_labels

def train_dependency_parser(model, train_data, dev_data, optimizer, scheduler, epochs=10):
    best_dev_las = 0.0
    best_model = None
    
    for epoch in range(epochs):
        # 训练模式
        model.train()
        total_loss = 0
        
        for batch in train_data:
            optimizer.zero_grad()
            
            # 前向传播
            arc_scores, label_scores = model(
                input_ids=batch['input_ids'],
                attention_mask=batch['attention_mask']
            )
            
            # 计算损失
            arc_loss = compute_arc_loss(arc_scores, batch['heads'])
            label_loss = compute_label_loss(label_scores, batch['labels'], batch['heads'])
            loss = arc_loss + label_loss
            
            # 反向传播
            loss.backward()
            
            # 梯度裁剪
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            
            # 更新参数
            optimizer.step()
            scheduler.step()
            
            total_loss += loss.item()
        
        # 验证
        model.eval()
        with torch.no_grad():
            dev_uas, dev_las = evaluate_parser(model, dev_data)
        
        print(f'Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_data):.4f}, '\
              f'Dev UAS: {dev_uas:.4f}, Dev LAS: {dev_las:.4f}')
        
        # 保存最佳模型
        if dev_las > best_dev_las:
            best_dev_las = dev_las
            best_model = copy.deepcopy(model)
    
    return best_model, best_dev_las