AI 大模型：从专业角度的系统实践与代码实现

原创

用户12339161

修改于 2026-09-18 14:36:46

1160

摘要

AI 大模型（Large Language Model，LLM）不是单一算法，而是由数据、算力、算法、工程、评估与安全共同构成的复杂系统。它以 Transformer 为骨架，通过自监督预训练获得通用语言能力，再经过指令微调、对齐、检索增强、工具调用与推理优化落地到生产。专业视角下，大模型的核心矛盾始终是：能力、延迟、成本、可控性与安全之间的权衡。本文从架构原理、训练对齐、推理优化、微调、RAG、Agent、文章生成流水线、评估与安全等维度，给出一套可落地的工程方法，并附带可直接运行的 Python 代码。

关键词：AI 大模型；Transformer；MoE；LoRA；vLLM；RAG；Agent；评估；安全护栏

1. 技术定位：大模型是什么

大模型的本质是：

给定上下文，预测下一个 token

它通过大规模自监督学习获得语言、知识、推理、代码与翻译能力。但它不是数据库，不是搜索引擎，也不是逻辑引擎。它输出概率分布，因此存在幻觉、偏差、时效性不足和不可解释性。

生产级 LLM 应用可抽象为：

LLM App = Model + Context + Retrieval + Tools + Evaluation + Guardrails

专业原则：

通用任务优先使用通用模型；
复杂推理再切换推理模型；
事实性内容必须依赖 RAG；
结构化输出必须校验；
所有关键操作必须有人工或规则兜底。

2. 核心架构：Transformer、注意力与 MoE

主流大模型采用 Decoder-only Transformer。核心组件包括多头自注意力、前馈网络、残差连接、层归一化和位置编码。

注意力公式：

Attention(Q, K, V) = softmax(QK^T / sqrt(d_k)) V

推理时使用 KV Cache 避免重复计算。MoE 通过路由只激活部分专家，降低每 token 计算量，但增加显存与通信复杂度。

下面用 PyTorch 实现缩放点积注意力：

import torch
import torch.nn as nn
import math

class ScaledDotProductAttention(nn.Module):
    def __init__(self, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(dropout)

    def forward(self, q, k, v, mask=None):
        d_k = q.size(-1)
        scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(d_k)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, float("-inf"))
        attn = torch.softmax(scores, dim=-1)
        attn = self.dropout(attn)
        return torch.matmul(attn, v), attn

多头注意力：

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model=256, nhead=8, dropout=0.1):
        super().__init__()
        assert d_model % nhead == 0
        self.d_model = d_model
        self.nhead = nhead
        self.d_k = d_model // nhead
        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)
        self.attn = ScaledDotProductAttention(dropout)

    def forward(self, x, mask=None):
        B, L, _ = x.size()
        q = self.w_q(x).view(B, L, self.nhead, self.d_k).transpose(1, 2)
        k = self.w_k(x).view(B, L, self.nhead, self.d_k).transpose(1, 2)
        v = self.w_v(x).view(B, L, self.nhead, self.d_k).transpose(1, 2)
        out, _ = self.attn(q, k, v, mask)
        out = out.transpose(1, 2).contiguous().view(B, L, self.d_model)
        return self.w_o(out)

3. 训练与对齐：从预训练到 DPO

典型流程：

预训练 → 指令微调 SFT → 对齐 RLHF / DPO → 蒸馏 / 量化 → 部署

预训练：自监督下一 token 预测，消耗最多算力；
SFT：用指令-回答数据教会模型遵循指令；
RLHF：奖励模型 + 强化学习，对齐人类偏好；
DPO：直接偏好优化，简化 RLHF；
蒸馏：大模型教小模型；
量化：GPTQ、AWQ、GGUF、bitsandbytes。

对齐不是一次性工作，而是持续评估与迭代。

4. 微调：LoRA 参数高效微调

当 RAG 无法满足风格、格式或领域术语要求时，考虑微调。LoRA 通过低秩矩阵大幅减少可训练参数。

# pip install torch transformers peft trl datasets accelerate bitsandbytes
import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer

model_name = "Qwen/Qwen2.5-1.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True,
)

lora = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora)

dataset = load_dataset("json", data_files="sft.jsonl", split="train")

args = TrainingArguments(
    output_dir="./lora-out",
    per_device_train_batch_size=2,
    gradient_accumulation_steps=8,
    learning_rate=1e-4,
    num_train_epochs=3,
    bf16=True,
    logging_steps=10,
    save_strategy="epoch",
)

trainer = SFTTrainer(
    model=model,
    args=args,
    train_dataset=dataset,
    tokenizer=tokenizer,
    dataset_text_field="text",
    max_seq_length=1024,
)
trainer.train()
model.save_pretrained("./lora-adapter")

微调不是知识注入的首选，RAG 更适合事实更新。微调适合“行为塑造”。

5. 推理优化：vLLM 与 OpenAI 兼容服务

生产推理需要高吞吐、低延迟和成本可控。vLLM 通过 PagedAttention、Continuous Batching 和 Prefix Caching 显著提升吞吐。

vllm serve Qwen/Qwen2.5-7B-Instruct \
  --port 8000 \
  --dtype bfloat16 \
  --max-model-len 8192

使用 OpenAI 兼容客户端调用：

from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")

resp = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "解释一下 KV Cache 和 PagedAttention。"}],
    temperature=0.3,
    max_tokens=512,
)
print(resp.choices[0].message.content)

推理优化清单：

KV Cache；
PagedAttention；
Continuous Batching；
Prefix Caching；
Speculative Decoding；
量化：GPTQ、AWQ、GGUF、FP8；
多卡：Tensor Parallel、Pipeline Parallel。

6. RAG：检索增强生成

RAG 用检索提供事实，用生成组织语言。最小实现如下：

import numpy as np
from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")

def embed(texts: list[str]) -> np.ndarray:
    # 真实场景可替换为 embedding 模型或本地向量模型
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data], dtype=np.float32)

class VectorStore:
    def __init__(self):
        self.docs: list[str] = []
        self.mat: np.ndarray | None = None

    def add(self, docs: list[str]):
        self.docs.extend(docs)
        vecs = embed(docs)
        self.mat = vecs if self.mat is None else np.vstack([self.mat, vecs])

    def search(self, query: str, top_k: int = 3) -> list[str]:
        q = embed([query])[0]
        sims = self.mat @ q / (
            np.linalg.norm(self.mat, axis=1) * np.linalg.norm(q) + 1e-8
        )
        idx = np.argsort(-sims)[:top_k]
        return [self.docs[i] for i in idx]

store = VectorStore()
store.add([
    "大模型推理时使用 KV Cache 避免重复计算注意力键值。",
    "LoRA 通过低秩矩阵微调，大幅减少可训练参数。",
    "RAG 用检索提供事实，用生成组织语言。",
])

def rag_answer(question: str) -> str:
    context = "\n".join(store.search(question))
    resp = client.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[
            {"role": "system", "content": "只基于给定资料回答，不知道就说不知道。"},
            {"role": "user", "content": f"资料：\n{context}\n\n问题：{question}"},
        ],
        temperature=0.2,
    )
    return resp.choices[0].message.content

print(rag_answer("LoRA 是什么？"))

生产级 RAG 还需要文档解析、切分、元数据、混合检索、重排序、查询改写、权限过滤与引用溯源。

7. Agent：工具调用与循环

Agent 的本质是 LLM + 工具 + 记忆 + 循环。

import json
from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")

TOOLS = {
    "search": lambda q: f"[搜索结果] 关于「{q}」的资料...",
    "calculator": lambda expr: str(eval(expr, {"__builtins__": {}}, {})),
}

def run_agent(question: str, max_steps: int = 5) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "你可以调用工具：search(query), calculator(expr)。"
                "需要时输出 JSON：{\"tool\":\"...\",\"args\":{...}}，否则直接回答。"
            ),
        },
        {"role": "user", "content": question},
    ]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=messages,
            temperature=0,
        )
        content = resp.choices[0].message.content.strip()
        try:
            call = json.loads(content)
            if "tool" in call:
                fn = TOOLS[call["tool"]]
                result = fn(**call["args"])
                messages.append({"role": "assistant", "content": content})
                messages.append({"role": "user", "content": f"工具结果：{result}"})
                continue
        except json.JSONDecodeError:
            pass
        return content
    return "达到最大步数限制"

print(run_agent("计算 12*(3+4)，并搜索 ReAct 的核心思想。"))

生产 Agent 需要工具白名单、参数校验、超时、最大步数、成本上限、人工确认高风险操作和全链路日志。

8. 文章生成流水线：大纲 → 分节 → 合并 → 润色

直接让模型生成长文容易结构松散。专业做法是拆成流水线。

import json
from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")
MODEL = "Qwen/Qwen2.5-7B-Instruct"

def generate_outline(topic: str, sections: int = 5) -> dict:
    resp = client.chat.completions.create(
        model=MODEL,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你只输出 JSON。"},
            {
                "role": "user",
                "content": (
                    f"为主题《{topic}》生成文章大纲，包含 {sections} 个章节。"
                    'JSON: {"title":"","summary":"","keywords":[],'
                    '"sections":[{"heading":"","points":["",""]}]}'
                ),
            },
        ],
        temperature=0.3,
    )
    return json.loads(resp.choices[0].message.content)

def generate_section(topic: str, heading: str, points: list[str]) -> str:
    point_text = "\n".join(f"- {p}" for p in points)
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "你是资深技术作者，结构清晰，不编造事实。"},
            {
                "role": "user",
                "content": (
                    f"主题：{topic}\n章节：{heading}\n要点：\n{point_text}\n"
                    "输出 Markdown，约 500-800 字。"
                ),
            },
        ],
        temperature=0.5,
        max_tokens=2048,
    )
    return resp.choices[0].message.content

def generate_article(topic: str, sections: int = 5) -> str:
    outline = generate_outline(topic, sections)

    parts = [
        f"# {outline['title']}\n",
        f"> 摘要：{outline.get('summary', '')}\n",
        f"**关键词**：{'、'.join(outline.get('keywords', []))}\n",
    ]

    for sec in outline["sections"]:
        content = generate_section(topic, sec["heading"], sec.get("points", []))
        parts.append(f"\n## {sec['heading']}\n\n{content}\n")

    draft = "\n".join(parts)

    final = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "你是资深编辑，负责润色与统一风格。"},
            {
                "role": "user",
                "content": "润色以下文章，保持 Markdown，消除重复，不改变事实。\n\n" + draft,
            },
        ],
        temperature=0.3,
        max_tokens=8192,
    )
    return final.choices[0].message.content

article = generate_article("AI 大模型", sections=5)
with open("ai_llm_article.md", "w", encoding="utf-8") as f:
    f.write(article)

print(article[:800])

9. 评估与可观测性

没有评估，就没有优化。LLM 评估分三层：单元评估、组件评估、端到端评估。

import json
import time

def evaluate_article(article: str, topic: str) -> dict:
    resp = client.chat.completions.create(
        model=MODEL,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你是严格的技术评审，只输出 JSON。"},
            {
                "role": "user",
                "content": (
                    f"评估文章《{topic}》。输出 "
                    '{"relevance":0-10,"factuality":0-10,'
                    '"structure":0-10,"readability":0-10,"safety":0-10,'
                    '"comments":""}\n\n文章：\n' + article[:6000]
                ),
            },
        ],
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

def traced_chat(messages, **kwargs):
    start = time.time()
    resp = client.chat.completions.create(messages=messages, **kwargs)
    usage = resp.usage
    print({
        "model": kwargs.get("model"),
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "latency_ms": int((time.time() - start) * 1000),
    })
    return resp

可观测性应记录：任务成功率、事实性、相关性、格式合规率、P50/P95 延迟、token 成本、安全违规率、全链路 trace。

10. 安全与合规

大模型安全风险包括提示注入、工具越权、数据泄露、无限循环、幻觉行动、版权与肖像权问题。

护栏示例：

BANNED_WORDS = ["违法", "暴力", "仇恨", "隐私泄露"]
DANGEROUS_TOOLS = {"shell", "delete_file", "execute_sql"}

def guard_output(text: str) -> bool:
    return not any(word in text for word in BANNED_WORDS)

def guard_tool_call(tool_name: str, user_role: str):
    if tool_name in DANGEROUS_TOOLS and user_role != "admin":
        raise PermissionError(f"无权调用工具: {tool_name}")
    return True

生产清单：

工具白名单与权限模型；
参数校验与输入过滤；
最大步数、最大 token、最大成本；
超时、重试、熔断、降级；
人工确认高风险操作；
全链路日志与审计；
评估集与回归测试；
灰度发布与回滚；
数据脱敏与隔离；
遵守 OWASP LLM Top 10 与行业规范。

11. 常见反模式

把大模型当数据库，靠模型记忆回答事实；
不做评估就上线；
不做 RAG，直接问模型最新知识；
让模型直接执行 Shell、SQL、文件删除；
不控制 token 与成本；
无超时、无重试、无降级；
微调解决知识更新；
忽略提示注入与权限隔离；
无日志、无回放、无版本管理；
生成内容不审核就发布。

12. 结论

AI 大模型的专业实践，是从 Transformer 原理出发，经过预训练、对齐、微调、量化与推理优化，最终以 RAG、Agent、Function Calling 和评估安全护栏落地为生产系统。代码上，可以从自注意力实现、LoRA 微调、vLLM 服务、RAG 检索、Agent 循环和文章生成流水线起步；工程上，必须建立评估、可观测性、成本治理与安全合规。真正可用的大模型应用，不是最大模型，而是在质量、延迟、成本与安全之间持续权衡的系统。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

人工智能