DeepSeek 技术突围，分层架构 + 前沿算法 + 代码实操的多维破局

import re

text = "这是一则#热闻@，关于科技发展*，但格式有点乱"
clean_text = re.sub(r'[#@*]', '', text)
print(clean_text)

import nltk
from nltk.tokenize import word_tokenize

text = "DeepSeek is a powerful language model"
tokens = word_tokenize(text)
print(tokens)

from gensim.models import Word2Vec
from nltk.tokenize import word_tokenize

sentences = [
    word_tokenize("I love natural language processing"),
    word_tokenize("DeepSeek can process language well")
]
model = Word2Vec(sentences, vector_size=100, window=5, min_count=1, workers=4)
print(model.wv['language'])

import torch
import torch.nn as nn


class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def scaled_dot_product_attention(self, Q, K, V, mask=None):
        attn_scores = torch.matmul(Q, K.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.d_k, dtype=torch.float32))
        if mask is not None:
            attn_scores = attn_scores.masked_fill(mask == 0, float('-inf'))
        attn_probs = torch.softmax(attn_scores, dim=-1)
        output = torch.matmul(attn_probs, V)
        return output

    def split_heads(self, x):
        batch_size, seq_length, d_model = x.size()
        return x.view(batch_size, seq_length, self.num_heads, self.d_k).transpose(1, 2)

    def forward(self, Q, K, V, mask=None):
        Q = self.split_heads(self.W_q(Q))
        K = self.split_heads(self.W_k(K))
        V = self.split_heads(self.W_v(V))

        attn_output = self.scaled_dot_product_attention(Q, K, V, mask)
        attn_output = attn_output.transpose(1, 2).contiguous().view(-1, attn_output.size(-2), attn_output.size(-1))
        output = self.W_o(attn_output)
        return output


class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff):
        super(TransformerEncoderLayer, self).__init__()
        self.attn = MultiHeadAttention(d_model, num_heads)
        self.norm1 = nn.LayerNorm(d_model)
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model)
        )
        self.norm2 = nn.LayerNorm(d_model)

    def forward(self, x, mask=None):
        attn_output = self.attn(x, x, x, mask)
        x = self.norm1(x + attn_output)
        ffn_output = self.ffn(x)
        x = self.norm2(x + ffn_output)
        return x

import torch
import torch.nn as nn


class VisualModel(nn.Module):
    def __init__(self):
        super(VisualModel, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, kernel_size=3, padding=1)
        self.pool = nn.MaxPool2d(2, 2)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=3, padding=1)
        self.fc1 = nn.Linear(32 * 8 * 8, 128)

    def forward(self, x):
        x = self.pool(torch.relu(self.conv1(x)))
        x = self.pool(torch.relu(self.conv2(x)))
        x = x.view(-1, 32 * 8 * 8)
        x = torch.relu(self.fc1(x))
        return x


class LanguageModel(nn.Module):
    def __init__(self, d_model, num_heads, d_ff):
        super(LanguageModel, self).__init__()
        self.encoder_layer = TransformerEncoderLayer(d_model, num_heads, d_ff)

    def forward(self, x):
        return self.encoder_layer(x)


class EarlyFusionModel(nn.Module):
    def __init__(self, d_model, num_heads, d_ff):
        super(EarlyFusionModel, self).__init__()
        self.visual_model = VisualModel()
        self.language_model = LanguageModel(d_model, num_heads, d_ff)
        self.fc = nn.Linear(128 + d_model, 10)

    def forward(self, visual_input, language_input):
        visual_feature = self.visual_model(visual_input)
        language_feature = self.language_model(language_input)
        fused_feature = torch.cat([visual_feature, language_feature], dim=1)
        output = self.fc(fused_feature)
        return output

from transformers import BertForSequenceClassification, BertTokenizer, AdamW
import torch
import torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler

# 加载预训练模型和分词器
model_name = 'bert-base-uncased'
tokenizer = BertTokenizer.from_pretrained(model_name)
model = BertForSequenceClassification.from_pretrained(model_name, num_labels=2)

# 模拟数据
texts = ["This is a positive sentence", "This is a negative sentence"]
labels = [1, 0]
encoded_inputs = tokenizer(texts, padding=True, truncation=True, return_tensors='pt')
input_ids = encoded_inputs['input_ids']
attention_mask = encoded_inputs['attention_mask']
labels = torch.tensor(labels)

# 创建数据集和数据加载器
dataset = TensorDataset(input_ids, attention_mask, labels)
dataloader = DataLoader(dataset, sampler=RandomSampler(dataset), batch_size=1)

# 定义优化器和损失函数
optimizer = AdamW(model.parameters(), lr=2e-5)
criterion = nn.CrossEntropyLoss()

# 微调模型
for epoch in range(3):
    model.train()
    total_loss = 0
    for batch in dataloader:
        batch_input_ids, batch_attention_mask, batch_labels = batch
        optimizer.zero_grad()
        outputs = model(batch_input_ids, attention_mask=batch_attention_mask, labels=batch_labels)
        loss = outputs.loss
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f'Epoch {epoch + 1}, Loss: {total_loss / len(dataloader)}')

from flask import Flask, request, jsonify

app = Flask(__name__)


@app.route('/chat', methods=['POST'])
def chat():
    data = request.get_json()
    # 这里模拟调用DeepSeek服务，实际需替换为真实API调用
    messages = data.get('messages', [])
    # 假设DeepSeek回复内容
    reply = "这是模拟的DeepSeek回复"
    result = {
        "choices": [
            {
                "message": {
                    "content": reply
                }
            }
        ]
    }
    return jsonify(result)


if __name__ == '__main__':
    app.run(debug=True)

import requests
import json

# 设置API端点
API_ENDPOINT = "https://api.deepseek.com/v1/chat/completions"
# 设置请求头，包含内容类型和授权信息（假设API_KEY为你的密钥，实际需替换）
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer API_KEY"
}
# 定义请求数据
data = {
    "messages": [
        {
            "role": "user",
            "content": "请介绍一下Python语言的特点"
        }
    ]
}
# 发送POST请求
response = requests.post(API_ENDPOINT, headers=headers, data=json.dumps(data))
if response.status_code == 200:
    result = response.json()
    reply = result["choices"][0]["message"]["content"]
    print("DeepSeek的回复：", reply)
else:
    print(f"请求失败，状态码：{response.status_code}，错误信息：{response.text}")

import requests
import json

API_ENDPOINT = "https://api.deepseek.com/v1/chat/completions"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer API_KEY"
}
# 初始化消息列表，用于存储对话历史
messages = []
while True:
    user_input = input("你：")
    if user_input.lower() == "退出":
        break
    messages.append({"role": "user", "content": user_input})
    data = {"messages": messages}
    response = requests.post(API_ENDPOINT, headers=headers, data=json.dumps(data))
    if response.status_code == 200:
        result = response.json()
        reply = result["choices"][0]["message"]["content"]
        messages.append({"role": "assistant", "content": reply})
        print("DeepSeek：", reply)
    else:
        print(f"请求失败，状态码：{response.status_code}，错误信息：{response.text}")