LLM 工具与生态面试题

# LangChain LCEL 风格（Runnable 协议）
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("用一句话解释 {concept}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser   # 管道组合
result = chain.invoke({"concept": "PagedAttention"})

# LlamaIndex QueryEngine 风格
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(llm=OpenAI(model="gpt-4o-mini"))
response = query_engine.query("什么是 PagedAttention？")

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    question: str
    plan: list[str]
    current_step: int
    context: str
    final_answer: str

def planner(state: AgentState) -> AgentState:
    # LLM 分解问题为步骤
    state["plan"] = ["检索", "分析", "总结"]
    return state

def retriever(state: AgentState) -> AgentState:
    state["context"] = "检索结果..."
    state["current_step"] += 1
    return state

def responder(state: AgentState) -> AgentState:
    state["final_answer"] = "最终回答..."
    return state

workflow = StateGraph(AgentState)
workflow.add_node("planner", planner)
workflow.add_node("retriever", retriever)
workflow.add_node("responder", responder)
workflow.add_edge("planner", "retriever")
workflow.add_edge("retriever", "responder")
workflow.add_edge("responder", END)
workflow.set_entry_point("planner")

memory = MemorySaver()
app = workflow.compile(checkpointer=memory, interrupt_before=["responder"])

from autogen import GroupChat, GroupChatManager, ConversableAgent

assistant = ConversableAgent(
    name="coder",
    system_message="你是 Python 工程师",
    llm_config={"model": "gpt-4o-mini"},
)
reviewer = ConversableAgent(
    name="reviewer",
    system_message="你是 Code Reviewer",
    llm_config={"model": "gpt-4o-mini"},
)
user_proxy = ConversableAgent(
    name="user",
    human_input_mode="ALWAYS",
)

group_chat = GroupChat(
    agents=[user_proxy, assistant, reviewer],
    messages=[],
    max_round=10,
)
manager = GroupChatManager(groupchat=group_chat, llm_config={"model": "gpt-4o-mini"})
user_proxy.initiate_chat(manager, message="实现快速排序")

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="研究员",
    goal="搜集 PagedAttention 相关资料",
    backstory="你是 LLM 推理领域专家",
    tools=[search_tool],
)
writer = Agent(
    role="技术写作者",
    goal="撰写技术博客",
    backstory="你是资深技术博主",
)

task1 = Task(description="搜集 PagedAttention 资料", agent=researcher)
task2 = Task(description="撰写 2000 字技术博客", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential,  # 顺序执行
)
result = crew.kickoff()

# promptfooconfig.yaml
prompts:
  - id: prompt-v1
    raw: |
      解释 {{concept}}，用一句话回答。
  - id: prompt-v2
    raw: |
      你是一个 SRE 专家。请用一句话解释 {{concept}}。

providers:
  - id: openai:gpt-4o-mini
  - id: openai:gpt-4o
  - id: anthropic:claude-3-5-sonnet

tests:
  - vars:
      concept: PagedAttention
    assert:
      - type: contains
        value: "KV Cache"
      - type: llm-rubric
        value: "回答准确且简洁"
      - type: javascript
        value: "output.split(' ').length <= 50"

  - vars:
      concept: Continuous Batching
    assert:
      - type: similar
        value: "连续批处理"
        threshold: 0.7

import dspy

# 1. 定义 Signature（声明输入输出）
class GenerateAnswer(dspy.Signature):
    """用一句话解释 LLM 概念。"""
    concept = dspy.InputField()
    answer = dspy.OutputField()

# 2. 定义 Module
class QABot(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, concept):
        return self.generate(concept=concept)

# 3. 准备训练集
trainset = [
    dspy.Example(concept="PagedAttention", answer="将 KV Cache 分页管理的注意力机制").with_inputs("concept"),
    dspy.Example(concept="Continuous Batching", answer="动态插入新请求到批处理的调度策略").with_inputs("concept"),
]

# 4. 编译（自动优化 Prompt）
tp = dspy.teleprompt.BootstrapFewShot(metric=dspy.evaluate.answer_exact_match)
optimized = tp.compile(QABot(), trainset=trainset)

# 5. 推理
result = optimized(concept="RadixAttention")
print(result.answer)

from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch

# 1. 一行加载模型（Hub ID 或本地路径）
model_id = "meta-llama/Meta-Llama-3-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto",          # Accelerate 自动分配 GPU
    attn_implementation="flash_attention_2",
)

# 2. Pipeline 一行推理
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)
output = pipe("解释 PagedAttention", max_new_tokens=200)
print(output[0]["generated_text"])

# 3. Chat Template
messages = [
    {"role": "system", "content": "你是 SRE 专家"},
    {"role": "user", "content": "什么是 Continuous Batching？"},
]
input_ids = tokenizer.apply_chat_template(messages, return_tensors="pt").to("cuda")
output_ids = model.generate(input_ids, max_new_tokens=200)
print(tokenizer.decode(output_ids[0], skip_special_tokens=True))

from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from transformers import AutoModelForCausalLM, BitsAndBytesConfig
from trl import SFTTrainer, SFTConfig

# 1. 4-bit 量化加载
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Meta-Llama-3-8B-Instruct",
    quantization_config=bnb_config,
    device_map="auto",
)
model = prepare_model_for_kbit_training(model)

# 2. LoRA 配置
lora_config = LoraConfig(
    r=64,
    lora_alpha=128,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()  # 仅 0.5% 参数可训练

# 3. SFT 训练
trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    args=SFTConfig(
        output_dir="./llama3-8b-qlora",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        fp16=True,
    ),
)
trainer.train()
trainer.save_model()

# 启动 vLLM 服务
vllm serve meta-llama/Meta-Llama-3-70B-Instruct \
  --host 0.0.0.0 \
  --port 8000 \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9 \
  --max-model-len 8192 \
  --quantization awq \
  --dtype bfloat16 \
  --enable-prefix-caching \
  --enforce-eager

# OpenAI 兼容客户端调用
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3-70B-Instruct",
    messages=[{"role": "user", "content": "解释 PagedAttention"}],
    temperature=0.7,
    max_tokens=500,
)
print(response.choices[0].message.content)

单卡显存 ≈ (P + G + OS) / N
  其中 P = 参数量（fp16 2 bytes / 参数）
       G = 梯度（fp16 2 bytes / 参数）
       OS = 优化器状态（Adam fp32 8 bytes / 参数）
  每参数总计 12 bytes，分片后单卡 12/N bytes

{
  "bf16": {"enabled": true},
  "optimizer": {
    "type": "AdamW",
    "params": {
      "lr": "auto",
      "betas": "auto",
      "weight_decay": "auto"
    }
  },
  "scheduler": {
    "type": "WarmupDecayLR",
    "params": {"warmup_min_ratio": 0.0, "warmup_num_steps": "auto"}
  },
  "zero_optimization": {
    "stage": 3,
    "offload_optimizer": {
      "device": "cpu",
      "pin_memory": true
    },
    "offload_param": {
      "device": "cpu",
      "pin_memory": true
    },
    "overlap_comm": true,
    "contiguous_gradients": true,
    "sub_group_size": 1e9,
    "reduce_bucket_size": "auto",
    "stage3_prefetch_bucket_size": "auto",
    "stage3_param_persistence_threshold": "auto",
    "stage3_max_live_parameters": 1e9,
    "stage3_max_reuse_distance": 1e9,
    "stage3_gather_16bit_weights_on_model_save": true
  },
  "gradient_accumulation_steps": "auto",
  "gradient_clipping": "auto",
  "train_micro_batch_size_per_gpu": "auto",
  "train_batch_size": "auto"
}

from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
from torch.distributed.fsdp import MixedPrecision, BackwardPrefetch
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy

# 1. FSDP 包装
fsdp_model = FSDP(
    model,
    mixed_precision=MixedPrecision(
        param_dtype=torch.bfloat16,
        reduce_dtype=torch.bfloat16,
        buffer_dtype=torch.bfloat16,
    ),
    auto_wrap_policy=transformer_auto_wrap_policy,  # 按 Transformer 层包装
    backward_prefetch=BackwardPrefetch.BACKWARD_PRE,
    device_id=torch.cuda.current_device(),
    limit_all_gathers=True,
    use_orig_params=True,
)

# 2. 启动
torch.distributed.launch \
  --nproc_per_node=8 \
  --nnodes=2 \
  --node_rank=0 \
  --master_addr=$MASTER_ADDR \
  train_fsdp.py

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_relevancy,
    answer_correctness,
)
from datasets import Dataset

# 1. 准备评估数据集
eval_data = {
    "question": [
        "什么是 PagedAttention？",
        "vLLM 的核心创新是什么？",
    ],
    "answer": [
        "PagedAttention 是 vLLM 的核心算法，将 KV Cache 按页管理...",
        "vLLM 的核心创新是 PagedAttention 与 Continuous Batching。",
    ],
    "contexts": [
        ["PagedAttention 借鉴操作系统虚拟内存分页机制..."],
        ["vLLM 通过 PagedAttention 算法实现显存零浪费..."],
    ],
    "ground_truth": [
        "PagedAttention 是 vLLM 的核心创新，将 KV Cache 分页管理以减少显存浪费。",
        "vLLM 核心创新是 PagedAttention。",
    ],
}
dataset = Dataset.from_dict(eval_data)

# 2. 评估
result = evaluate(
    dataset,
    metrics=[
        faithfulness,        # 忠实度
        answer_relevancy,    # 答案相关性
        context_precision,   # 上下文精确率
        context_recall,      # 上下文召回率
        context_relevancy,   # 上下文相关性
        answer_correctness,  # 答案正确性
    ],
    llm=evaluator_llm,         # LLM-as-Judge
    embeddings=evaluator_embeddings,  # Embedding 模型
)
print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 'context_precision': 0.85, ...}