fooSynaptic

# 多轮推理 (R-Net 风格)
for layer in range(num_layers):
    # 自注意力
    context = self_attention(context, context)
    # 交叉注意力
    context = cross_attention(context, question)

from transformers import AutoModelForQuestionAnswering

model = AutoModelForQuestionAnswering.from_pretrained("bert-base-uncased")
# Fine-tune on SQuAD

# Few-shot prompting
prompt = """
Context: The Eiffel Tower was built in 1889.
Question: When was the Eiffel Tower built?
Answer: 1889

Context: {context}
Question: {question}
Answer:"""

class ModernMRC:
    def __init__(self, retriever, reader):
        self.retriever = retriever  # Dense retriever
        self.reader = reader        # LLM
    
    def answer(self, question: str, knowledge_base: str = None):
        # 1. 检索
        if knowledge_base:
            docs = self.retriever.retrieve(question, knowledge_base)
            context = "\n\n".join([d.text for d in docs])
        else:
            context = ""
        
        # 2. 阅读理解/生成
        prompt = self._build_prompt(question, context)
        answer = self.reader.generate(prompt)
        
        # 3. 后处理（可选：验证、引用）
        return self._postprocess(answer, docs)
    
    def _build_prompt(self, question, context):
        if context:
            return f"""Based on the following context, answer the question.

Context:
{context}

Question: {question}
Answer:"""
        else:
            return f"Question: {question}\nAnswer:"

class MultiHopReasoner:
    def __init__(self, retriever, llm, max_hops=3):
        self.retriever = retriever
        self.llm = llm
        self.max_hops = max_hops
    
    def reason(self, question):
        reasoning_chain = []
        current_query = question
        
        for hop in range(self.max_hops):
            # 检索
            docs = self.retriever.retrieve(current_query)
            
            # 生成中间推理
            intermediate = self.llm.generate(
                f"Based on: {docs}\nQuestion: {current_query}\n"
                f"Provide intermediate reasoning or the final answer:"
            )
            
            reasoning_chain.append({
                'query': current_query,
                'docs': docs,
                'reasoning': intermediate
            })
            
            # 检查是否已得到答案
            if self._is_final_answer(intermediate):
                break
            
            # 生成下一跳查询
            current_query = self._generate_next_query(question, reasoning_chain)
        
        return self._synthesize_answer(question, reasoning_chain)

class ConversationalQA:
    def __init__(self, mrc_model, history_length=5):
        self.mrc_model = mrc_model
        self.history = []
        self.history_length = history_length
    
    def ask(self, question, context=None):
        # 将对话历史纳入问题
        contextualized_question = self._contextualize(question)
        
        # 获取答案
        answer = self.mrc_model.answer(contextualized_question, context)
        
        # 更新历史
        self.history.append({'q': question, 'a': answer})
        if len(self.history) > self.history_length:
            self.history.pop(0)
        
        return answer
    
    def _contextualize(self, question):
        if not self.history:
            return question
        
        history_text = "\n".join([
            f"Q: {turn['q']}\nA: {turn['a']}"
            for turn in self.history
        ])
        
        return f"Conversation history:\n{history_text}\n\nCurrent question: {question}"

# LLM-as-Judge
def llm_evaluate(question, reference, prediction):
    prompt = f"""Evaluate the answer quality on a scale of 1-5:

Question: {question}
Reference Answer: {reference}
Model Answer: {prediction}

Criteria:
- Correctness: Is the information accurate?
- Completeness: Does it fully answer the question?
- Conciseness: Is it appropriately brief?

Score (1-5):"""
    
    return llm.generate(prompt)

import json
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Example:
    context: str
    question: str
    answer_text: str
    start_position: int
    end_position: int

def load_squad(file_path: str) -> List[Example]:
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    examples = []
    for article in data['data']:
        for paragraph in article['paragraphs']:
            context = paragraph['context']
            for qa in paragraph['qas']:
                question = qa['question']
                if qa.get('is_impossible', False):
                    continue
                answer = qa['answers'][0]
                examples.append(Example(
                    context=context,
                    question=question,
                    answer_text=answer['text'],
                    start_position=answer['answer_start'],
                    end_position=answer['answer_start'] + len(answer['text'])
                ))
    
    return examples

from transformers import AutoTokenizer

class MRCTokenizer:
    def __init__(self, model_name: str, max_length: int = 384, doc_stride: int = 128):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.max_length = max_length
        self.doc_stride = doc_stride
    
    def encode(self, example: Example):
        # Tokenize question and context
        encoding = self.tokenizer(
            example.question,
            example.context,
            max_length=self.max_length,
            truncation='only_second',
            stride=self.doc_stride,
            return_overflowing_tokens=True,
            return_offsets_mapping=True,
            padding='max_length',
        )
        
        # 找到答案在 token 序列中的位置
        offset_mapping = encoding['offset_mapping'][0]
        
        start_token = None
        end_token = None
        
        for idx, (start, end) in enumerate(offset_mapping):
            if start <= example.start_position < end:
                start_token = idx
            if start < example.end_position <= end:
                end_token = idx
                break
        
        return {
            'input_ids': encoding['input_ids'][0],
            'attention_mask': encoding['attention_mask'][0],
            'start_position': start_token or 0,
            'end_position': end_token or 0,
        }

import torch
import torch.nn as nn
from transformers import AutoModel

class MRCModel(nn.Module):
    def __init__(self, model_name: str, dropout: float = 0.1):
        super().__init__()
        self.bert = AutoModel.from_pretrained(model_name)
        hidden_size = self.bert.config.hidden_size
        
        self.dropout = nn.Dropout(dropout)
        self.start_classifier = nn.Linear(hidden_size, 1)
        self.end_classifier = nn.Linear(hidden_size, 1)
    
    def forward(
        self,
        input_ids: torch.Tensor,
        attention_mask: torch.Tensor,
        start_positions: Optional[torch.Tensor] = None,
        end_positions: Optional[torch.Tensor] = None,
    ):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        sequence_output = self.dropout(outputs.last_hidden_state)
        
        # (batch, seq_len, 1) -> (batch, seq_len)
        start_logits = self.start_classifier(sequence_output).squeeze(-1)
        end_logits = self.end_classifier(sequence_output).squeeze(-1)
        
        # Mask padding tokens
        start_logits = start_logits.masked_fill(~attention_mask.bool(), -1e9)
        end_logits = end_logits.masked_fill(~attention_mask.bool(), -1e9)
        
        loss = None
        if start_positions is not None and end_positions is not None:
            loss_fct = nn.CrossEntropyLoss()
            start_loss = loss_fct(start_logits, start_positions)
            end_loss = loss_fct(end_logits, end_positions)
            loss = (start_loss + end_loss) / 2
        
        return {
            'loss': loss,
            'start_logits': start_logits,
            'end_logits': end_logits,
        }

class JointMRCModel(nn.Module):
    """联合预测 start 和 end，考虑 start-end 依赖"""
    
    def __init__(self, model_name: str, max_answer_length: int = 30):
        super().__init__()
        self.bert = AutoModel.from_pretrained(model_name)
        hidden_size = self.bert.config.hidden_size
        self.max_answer_length = max_answer_length
        
        self.start_classifier = nn.Linear(hidden_size, 1)
        self.end_classifier = nn.Linear(hidden_size * 2, 1)
    
    def forward(self, input_ids, attention_mask, start_positions=None, end_positions=None):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        H = outputs.last_hidden_state  # (batch, seq_len, hidden)
        
        # Start prediction
        start_logits = self.start_classifier(H).squeeze(-1)
        
        if self.training and start_positions is not None:
            # 训练时使用真实的 start 位置
            start_indices = start_positions.unsqueeze(-1).unsqueeze(-1)
            start_repr = H.gather(1, start_indices.expand(-1, -1, H.size(-1))).squeeze(1)
        else:
            # 推理时使用预测的 start 位置
            start_indices = start_logits.argmax(dim=-1, keepdim=True).unsqueeze(-1)
            start_repr = H.gather(1, start_indices.expand(-1, -1, H.size(-1))).squeeze(1)
        
        # End prediction conditioned on start
        start_repr_expanded = start_repr.unsqueeze(1).expand(-1, H.size(1), -1)
        end_input = torch.cat([H, start_repr_expanded], dim=-1)
        end_logits = self.end_classifier(end_input).squeeze(-1)
        
        # 只允许 end >= start 且在 max_answer_length 范围内
        # 这里简化处理，完整实现需要更复杂的 mask
        
        return {'start_logits': start_logits, 'end_logits': end_logits}

from torch.utils.data import DataLoader, Dataset
from transformers import get_linear_schedule_with_warmup
from tqdm import tqdm

def train(model, train_dataloader, val_dataloader, epochs=3, lr=3e-5):
    optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01)
    
    total_steps = len(train_dataloader) * epochs
    scheduler = get_linear_schedule_with_warmup(
        optimizer, 
        num_warmup_steps=int(0.1 * total_steps),
        num_training_steps=total_steps
    )
    
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    
    best_f1 = 0
    for epoch in range(epochs):
        model.train()
        total_loss = 0
        
        for batch in tqdm(train_dataloader, desc=f'Epoch {epoch+1}'):
            batch = {k: v.to(device) for k, v in batch.items()}
            
            outputs = model(**batch)
            loss = outputs['loss']
            
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
            
            optimizer.step()
            scheduler.step()
            optimizer.zero_grad()
            
            total_loss += loss.item()
        
        avg_loss = total_loss / len(train_dataloader)
        print(f'Epoch {epoch+1}, Loss: {avg_loss:.4f}')
        
        # Validation
        f1 = evaluate(model, val_dataloader, device)
        print(f'Validation F1: {f1:.4f}')
        
        if f1 > best_f1:
            best_f1 = f1
            torch.save(model.state_dict(), 'best_model.pt')
    
    return model

import re
import string
from collections import Counter

def normalize_answer(s):
    """标准化答案用于评估"""
    def remove_articles(text):
        return re.sub(r'\b(a|an|the)\b', ' ', text)
    
    def white_space_fix(text):
        return ' '.join(text.split())
    
    def remove_punc(text):
        exclude = set(string.punctuation)
        return ''.join(ch for ch in text if ch not in exclude)
    
    def lower(text):
        return text.lower()
    
    return white_space_fix(remove_articles(remove_punc(lower(s))))

def compute_f1(pred: str, gold: str) -> float:
    pred_tokens = normalize_answer(pred).split()
    gold_tokens = normalize_answer(gold).split()
    
    common = Counter(pred_tokens) & Counter(gold_tokens)
    num_same = sum(common.values())
    
    if num_same == 0:
        return 0
    
    precision = num_same / len(pred_tokens)
    recall = num_same / len(gold_tokens)
    
    return 2 * precision * recall / (precision + recall)

def predict(model, tokenizer, context: str, question: str, device):
    """单条推理"""
    model.eval()
    
    encoding = tokenizer(
        question, context,
        max_length=384,
        truncation='only_second',
        return_tensors='pt'
    )
    
    encoding = {k: v.to(device) for k, v in encoding.items()}
    
    with torch.no_grad():
        outputs = model(**encoding)
    
    start_idx = outputs['start_logits'].argmax().item()
    end_idx = outputs['end_logits'].argmax().item()
    
    # 确保 end >= start
    if end_idx < start_idx:
        end_idx = start_idx
    
    # 解码答案
    answer_tokens = encoding['input_ids'][0][start_idx:end_idx+1]
    answer = tokenizer.decode(answer_tokens, skip_special_tokens=True)
    
    return answer

from openai import OpenAI

def llm_qa(context: str, question: str) -> str:
    client = OpenAI()
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个问答助手。根据给定的上下文回答问题。如果答案不在上下文中，请说'无法回答'。"},
            {"role": "user", "content": f"上下文：{context}\n\n问题：{question}"}
        ],
        temperature=0
    )
    
    return response.choices[0].message.content

import torch
import torch.nn as nn

class CRF(nn.Module):
    def __init__(self, num_tags, batch_first=True):
        super().__init__()
        self.num_tags = num_tags
        self.batch_first = batch_first
        
        # 转移矩阵: 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 forward(self, emissions, tags, mask=None):
        """计算负对数似然损失"""
        if mask is None:
            mask = torch.ones_like(tags, dtype=torch.bool)
        
        if self.batch_first:
            emissions = emissions.transpose(0, 1)
            tags = tags.transpose(0, 1)
            mask = mask.transpose(0, 1)
        
        # 计算分子（正确路径的分数）
        numerator = self._compute_score(emissions, tags, mask)
        
        # 计算分母（配分函数）
        denominator = self._compute_normalizer(emissions, mask)
        
        # 负对数似然
        return (denominator - numerator).mean()
    
    def _compute_score(self, emissions, tags, mask):
        """计算给定标签序列的分数"""
        seq_len, batch_size = tags.shape
        
        # 起始分数
        score = self.start_transitions[tags[0]]
        score += emissions[0, torch.arange(batch_size), tags[0]]
        
        for i in range(1, seq_len):
            # 转移分数 + 发射分数
            score += self.transitions[tags[i], tags[i-1]] * mask[i]
            score += emissions[i, torch.arange(batch_size), tags[i]] * mask[i]
        
        # 结束分数
        last_tag_idx = mask.sum(dim=0) - 1
        last_tags = tags.gather(0, last_tag_idx.unsqueeze(0)).squeeze(0)
        score += self.end_transitions[last_tags]
        
        return score
    
    def _compute_normalizer(self, emissions, mask):
        """前向算法计算配分函数"""
        seq_len, batch_size, num_tags = emissions.shape
        
        # 初始化
        score = self.start_transitions + emissions[0]
        
        for i in range(1, seq_len):
            # broadcast: (batch, num_tags, 1) + (num_tags, num_tags) + (batch, 1, num_tags)
            broadcast_score = score.unsqueeze(2)
            broadcast_emissions = emissions[i].unsqueeze(1)
            
            next_score = broadcast_score + self.transitions + broadcast_emissions
            next_score = torch.logsumexp(next_score, dim=1)
            
            # 应用 mask
            score = torch.where(mask[i].unsqueeze(1), next_score, score)
        
        # 添加结束分数
        score += self.end_transitions
        
        return torch.logsumexp(score, dim=1)
    
    def decode(self, emissions, mask=None):
        """Viterbi 解码"""
        if mask is None:
            mask = torch.ones(emissions.shape[:2], dtype=torch.bool, device=emissions.device)
        
        if self.batch_first:
            emissions = emissions.transpose(0, 1)
            mask = mask.transpose(0, 1)
        
        return self._viterbi_decode(emissions, mask)
    
    def _viterbi_decode(self, emissions, mask):
        """Viterbi 算法"""
        seq_len, batch_size, num_tags = emissions.shape
        
        # 初始化
        score = self.start_transitions + emissions[0]
        history = []
        
        for i in range(1, seq_len):
            broadcast_score = score.unsqueeze(2)
            broadcast_emissions = emissions[i].unsqueeze(1)
            
            next_score = broadcast_score + self.transitions + broadcast_emissions
            next_score, indices = next_score.max(dim=1)
            
            score = torch.where(mask[i].unsqueeze(1), next_score, score)
            history.append(indices)
        
        # 添加结束分数
        score += self.end_transitions
        
        # 回溯
        best_tags_list = []
        _, best_last_tag = score.max(dim=1)
        
        for idx in range(batch_size):
            best_tags = [best_last_tag[idx].item()]
            seq_length = int(mask[:, idx].sum().item())
            
            for hist in reversed(history[:seq_length-1]):
                best_last_tag_idx = best_tags[-1]
                best_tags.append(hist[idx, best_last_tag_idx].item())
            
            best_tags.reverse()
            best_tags_list.append(best_tags)
        
        return best_tags_list

class BiLSTM_CRF(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_tags):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim // 2, 
                           num_layers=2, bidirectional=True, batch_first=True)
        self.fc = nn.Linear(hidden_dim, num_tags)
        self.crf = CRF(num_tags)
    
    def forward(self, x, tags, mask=None):
        embeddings = self.embedding(x)
        lstm_out, _ = self.lstm(embeddings)
        emissions = self.fc(lstm_out)
        
        return self.crf(emissions, tags, mask)
    
    def predict(self, x, mask=None):
        embeddings = self.embedding(x)
        lstm_out, _ = self.lstm(embeddings)
        emissions = self.fc(lstm_out)
        
        return self.crf.decode(emissions, mask)

from transformers import BertModel

class BERT_CRF(nn.Module):
    def __init__(self, bert_name, num_tags):
        super().__init__()
        self.bert = BertModel.from_pretrained(bert_name)
        self.dropout = nn.Dropout(0.1)
        self.fc = nn.Linear(self.bert.config.hidden_size, num_tags)
        self.crf = CRF(num_tags)
    
    def forward(self, input_ids, attention_mask, tags=None):
        outputs = self.bert(input_ids, attention_mask=attention_mask)
        sequence_output = self.dropout(outputs.last_hidden_state)
        emissions = self.fc(sequence_output)
        
        if tags is not None:
            return self.crf(emissions, tags, attention_mask.bool())
        else:
            return self.crf.decode(emissions, attention_mask.bool())

def label_smoothing_loss(crf, emissions, tags, mask, epsilon=0.1):
    """带标签平滑的 CRF 损失"""
    nll_loss = crf(emissions, tags, mask)
    
    # 均匀分布的损失
    uniform_loss = -torch.logsumexp(emissions, dim=-1).mean()
    
    return (1 - epsilon) * nll_loss + epsilon * uniform_loss

# 添加硬约束：B-X 后面只能接 I-X 或 O
def add_constraints(transitions, tag2idx):
    for tag_from, idx_from in tag2idx.items():
        for tag_to, idx_to in tag2idx.items():
            if tag_from.startswith('B-') or tag_from.startswith('I-'):
                entity = tag_from[2:]
                if tag_to.startswith('I-') and tag_to[2:] != entity:
                    transitions.data[idx_to, idx_from] = -1e9

import torch
import torch.nn as nn

class BiDAFAttention(nn.Module):
    def __init__(self, hidden_size):
        super().__init__()
        self.W = nn.Linear(hidden_size * 3, 1, bias=False)
    
    def forward(self, context, query, c_mask, q_mask):
        """
        Args:
            context: (batch, c_len, hidden)
            query: (batch, q_len, hidden)
            c_mask: (batch, c_len)
            q_mask: (batch, q_len)
        """
        batch, c_len, hidden = context.size()
        q_len = query.size(1)
        
        # 扩展维度以计算所有 (i, j) 对
        c_expand = context.unsqueeze(2).expand(-1, -1, q_len, -1)
        q_expand = query.unsqueeze(1).expand(-1, c_len, -1, -1)
        
        # 计算相似度矩阵 S
        cq = torch.cat([c_expand, q_expand, c_expand * q_expand], dim=-1)
        S = self.W(cq).squeeze(-1)  # (batch, c_len, q_len)
        
        # Mask
        q_mask_expand = q_mask.unsqueeze(1).expand(-1, c_len, -1)
        S = S.masked_fill(~q_mask_expand, -1e9)
        
        # C2Q attention
        a = torch.softmax(S, dim=-1)
        c2q = torch.bmm(a, query)  # (batch, c_len, hidden)
        
        # Q2C attention
        b = torch.softmax(S.max(dim=-1)[0], dim=-1)
        q2c = torch.bmm(b.unsqueeze(1), context)  # (batch, 1, hidden)
        q2c = q2c.expand(-1, c_len, -1)
        
        # 融合
        G = torch.cat([context, c2q, context * c2q, context * q2c], dim=-1)
        
        return G

class CrossAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        self.mha = nn.MultiheadAttention(d_model, n_heads)
    
    def forward(self, query, key_value):
        # query 来自一个序列，key/value 来自另一个序列
        return self.mha(query, key_value, key_value)

class FiD(nn.Module):
    def __init__(self, encoder, decoder):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder
    
    def forward(self, question, passages):
        # 独立编码每个 passage
        encoded = []
        for passage in passages:
            enc = self.encoder(question + passage)
            encoded.append(enc)
        
        # 融合解码
        fused = torch.cat(encoded, dim=1)
        return self.decoder(fused)