🎯 大模型应用面试题库¶

# Few-shot模板示例
PROMPT = """你是一个情感分析助手。请判断文本的情感倾向。

示例1:
输入: "这家餐厅的菜品真的太好吃了，环境也很棒"
输出: {"sentiment": "positive", "confidence": 0.95}

示例2:
输入: "等了一个小时还没上菜，服务态度极差"
输出: {"sentiment": "negative", "confidence": 0.92}

示例3:
输入: "菜品还行，就是价格偏贵了点"
输出: {"sentiment": "neutral", "confidence": 0.78}

请分析以下文本:
输入: "{user_text}"
输出: """

# Zero-shot CoT
prompt_zs = f"""
问题：一个商店有45个苹果，卖掉了18个，又进了27个，现在有多少个？
请一步步思考后给出答案。
"""

# Manual CoT (更可靠)
prompt_manual = f"""
问题：小明有12块钱，买了3本书每本2块，还剩多少钱？
思考过程：
1. 3本书的总价 = 3 × 2 = 6块
2. 剩余 = 12 - 6 = 6块
答案：6块

问题：{user_question}
思考过程：
"""

import collections

def self_consistency(client, prompt, n_samples=10, temperature=0.8):
    """自一致性推理"""
    answers = []
    for _ in range(n_samples):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            max_tokens=1024,
        )
        # 提取最终答案（假设格式为 "答案：XXX"）
        text = response.choices[0].message.content
        answer = extract_final_answer(text)
        answers.append(answer)

    # 多数投票
    counter = collections.Counter(answers)  # Counter统计每个答案的出现次数
    best_answer, count = counter.most_common(1)[0]  # most_common(1)取最高频的(答案, 次数)元组并解包
    confidence = count / len(answers)

    return {
        "answer": best_answer,
        "confidence": confidence,
        "vote_distribution": dict(counter),
    }

# 生产级System Prompt模板
SYSTEM_PROMPT = """## 角色
你是一位资深的Python后端开发专家。

## 能力
- 精通Python 3.10+, FastAPI, SQLAlchemy
- 熟悉设计模式和架构原则
- 了解安全最佳实践

## 约束
- 所有代码示例使用Python 3.10+语法
- 遵循PEP 8规范
- 不输出未经验证的安全敏感代码
- 如果不确定，明确说明而非猜测

## 输出格式
- 代码使用```python代码块
- 先给出简要解释，再给代码
- 关键部分添加注释"""

# 方法1: Function Calling (最可靠)
tools = [{
    "type": "function",
    "function": {
        "name": "extract_entities",
        "parameters": {
            "type": "object",
            "properties": {
                "persons": {"type": "array", "items": {"type": "string"}},
                "locations": {"type": "array", "items": {"type": "string"}},
            },
            "required": ["persons", "locations"]
        }
    }
}]

# 方法2: JSON Mode
response = client.chat.completions.create(
    model="gpt-4o-mini",
    response_format={"type": "json_object"},
    messages=[{"role": "user", "content": "以JSON格式提取..."}],
)

# 方法3: Pydantic + Instructor (第三方库)
import instructor
from pydantic import BaseModel

class Sentiment(BaseModel):  # Pydantic BaseModel：自动数据验证和序列化
    label: str
    score: float
    reasoning: str

client = instructor.patch(OpenAI())
result = client.chat.completions.create(
    model="gpt-4o-mini",
    response_model=Sentiment,
    messages=[{"role": "user", "content": "分析情感..."}],
)

# 父子分块 (Parent-Child Chunking)
from langchain_text_splitters import RecursiveCharacterTextSplitter

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)

parent_chunks = parent_splitter.split_documents(docs)
for parent in parent_chunks:
    children = child_splitter.split_documents([parent])
    for child in children:
        child.metadata["parent_id"] = parent.metadata["doc_id"]
        # 检索时匹配child，返回对应parent的完整内容

# 混合检索实现 (Reciprocal Rank Fusion)
def reciprocal_rank_fusion(results_list, k=60):
    """RRF融合多路检索结果"""
    fused_scores = {}
    for results in results_list:
        for rank, (doc_id, _) in enumerate(results):  # enumerate返回(序号, 元素)，(doc_id, _)再解包元素，_丢弃第二项
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0
            fused_scores[doc_id] += 1 / (k + rank + 1)

    # 按分数降序排列：items()返回(doc_id, score)对，lambda x: x[1]取score作为排序依据
    return sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)  # lambda匿名函数

# 实际使用
dense_results = vector_store.similarity_search(query, k=20)
sparse_results = bm25_retriever.get_relevant_documents(query)[:20]
fused = reciprocal_rank_fusion([
    [(doc.metadata['id'], doc) for doc in dense_results],
    [(doc.metadata['id'], doc) for doc in sparse_results],
])

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large", max_length=512)

def rerank(query, documents, top_k=5):
    """Cross-Encoder重排序"""
    pairs = [(query, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)

    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)  # zip将文档和分数一一配对，sorted按分数降序
    return [doc for doc, score in ranked[:top_k]]  # 取前K个，解包时用doc保留文档、丢弃score

# HyDE实现
def hyde_retrieval(query, llm, retriever):
    """HyDE: 先生成假设答案，用答案做检索"""
    # Step 1: 生成假设答案
    hypothetical_answer = llm.invoke(
        f"请直接回答以下问题（不需要确认准确性）: {query}"
    ).content

    # Step 2: 用假设答案的embedding检索
    docs = retriever.get_relevant_documents(hypothetical_answer)
    return docs

# Multi-Query实现
def multi_query_retrieval(query, llm, retriever, n_queries=3):
    """多角度Query改写"""
    expanded_queries = llm.invoke(
        f"请将以下问题从{n_queries}个不同角度改写:\n{query}\n只输出改写后的问题，每行一个"
    ).content.strip().split('\n')  # 链式调用：strip去除空白

    all_docs = []
    seen_ids = set()
    for q in [query] + expanded_queries:
        for doc in retriever.get_relevant_documents(q):
            doc_id = doc.metadata.get('doc_id')
            if doc_id not in seen_ids:
                all_docs.append(doc)
                seen_ids.add(doc_id)
    return all_docs

# Iterative RAG简化实现
def iterative_rag(query, retriever, llm, max_steps=3):
    context = ""
    for step in range(max_steps):
        docs = retriever.get_relevant_documents(query if step == 0 else sub_query)
        context += "\n".join([d.page_content for d in docs])

        response = llm.invoke(f"""
        基于已有信息回答问题，如果信息不够，输出"需要更多信息：<子问题>"。
        已有信息：{context}
        问题：{query}
        """).content

        if "需要更多信息" not in response:
            return response
        sub_query = response.split("需要更多信息：")[1].strip()

    return response

# ReAct Agent简化实现
class ReActAgent:
    def __init__(self, llm, tools, max_steps=5):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps

    def run(self, query):
        prompt = self._build_prompt(query)
        history = []

        for step in range(self.max_steps):
            response = self.llm.invoke(prompt + "\n".join(history))

            if "Answer:" in response.content:
                return response.content.split("Answer:")[1].strip()

            # 解析Action
            action, action_input = self._parse_action(response.content)

            if action in self.tools:
                observation = self.tools[action].invoke(action_input)
                history.append(f"Thought: {response.content}")
                history.append(f"Observation: {observation}")
            else:
                history.append(f"Observation: 工具 {action} 不存在")

        return "达到最大步数，无法完成任务"

# OpenAI Function Calling
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "获取股票当前价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "股票代码，如AAPL"},
                },
                "required": ["symbol"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "苹果公司现在股价多少？"}],
    tools=tools,
    tool_choice="auto",
)

# 处理tool call
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)  # json.loads将JSON字符串→Python对象

    # 执行工具
    result = execute_tool(func_name, func_args)

    # 将结果返回给LLM
    messages.append(response.choices[0].message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result)  # json.dumps将Python对象→JSON字符串
    })
    final_response = client.chat.completions.create(model="gpt-4o", messages=messages)

# 主从模式示例（简化）
class OrchestratorAgent:
    def __init__(self, planner_llm, worker_agents):
        self.planner = planner_llm
        self.workers = worker_agents  # {"researcher": ..., "coder": ..., "reviewer": ...}

    def run(self, task):
        # 1. 规划子任务
        plan = self.planner.invoke(f"将以下任务分解为子任务，指定负责的Agent:\n{task}")
        subtasks = self._parse_plan(plan)

        # 2. 分派执行
        results = {}
        for subtask in subtasks:
            agent_name = subtask["agent"]
            result = self.workers[agent_name].run(subtask["description"])
            results[subtask["id"]] = result

        # 3. 汇总
        summary = self.planner.invoke(f"汇总以下子任务结果:\n{results}")
        return summary

# 带记忆的Agent简化实现
class MemoryAgent:
    def __init__(self, llm, vector_store, max_history=20):
        self.llm = llm
        self.long_term = vector_store  # 长期记忆
        self.short_term = []            # 短期记忆(对话历史)
        self.max_history = max_history

    def chat(self, user_message):
        # 1. 从长期记忆检索相关信息
        relevant_memories = self.long_term.similarity_search(user_message, k=3)
        memory_context = "\n".join([m.page_content for m in relevant_memories])

        # 2. 构建Prompt
        messages = [
            {"role": "system", "content": f"相关记忆:\n{memory_context}"},
            *self.short_term[-self.max_history:],
            {"role": "user", "content": user_message}
        ]

        # 3. 生成回复
        response = self.llm.invoke(messages).content

        # 4. 更新记忆
        self.short_term.append({"role": "user", "content": user_message})
        self.short_term.append({"role": "assistant", "content": response})

        # 5. 重要信息存入长期记忆
        if self._is_important(user_message, response):
            self.long_term.add_texts([f"用户: {user_message}\n助手: {response}"])

        return response

from peft import LoraConfig, get_peft_model, TaskType

lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                      # 秩
    lora_alpha=32,             # 缩放因子 (alpha/r 为实际缩放)
    lora_dropout=0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],  # 作用模块
    bias="none",
)

model = get_peft_model(base_model, lora_config)
model.print_trainable_parameters()
# trainable params: 8,388,608 || all params: 6,746,804,224 || trainable%: 0.124%

# 数据质量检查Pipeline
def validate_training_data(data):
    """验证微调数据质量"""
    issues = []

    for i, item in enumerate(data):  # enumerate同时获取索引和元素
        # 1. 格式检查
        if not all(k in item for k in ['instruction', 'output']):  # all()全部为True才返回True
            issues.append(f"[{i}] 缺少必要字段")
            continue

        # 2. 长度检查
        total_tokens = count_tokens(item['instruction'] + item.get('input','') + item['output'])  # get('input',''):若该键不存在则返回空字符串，避免KeyError
        if total_tokens > 2048:
            issues.append(f"[{i}] 超长: {total_tokens} tokens")
        if len(item['output'].strip()) < 10:
            issues.append(f"[{i}] 回答过短")

        # 3. 重复检查
        # ... MinHash去重

    print(f"共 {len(data)} 条, {len(issues)} 个问题")
    return issues

# vLLM使用示例
from vllm import LLM, SamplingParams

llm = LLM(
    model="meta-llama/Llama-2-7b-chat-hf",
    tensor_parallel_size=1,
    gpu_memory_utilization=0.90,
    max_num_seqs=256,          # 最大并发
    max_model_len=4096,
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512,
)

outputs = llm.generate(["你好，请介绍一下自己"], sampling_params)

JUDGE_PROMPT = """请评估以下AI回答的质量。

用户问题: {question}
AI回答: {answer}
参考答案: {reference}

请从以下维度打分(1-5分):
1. 准确性: 信息是否正确
2. 完整性: 是否覆盖关键点
3. 清晰度: 表述是否清楚
4. 有用性: 对用户的帮助程度

输出JSON: {"accuracy": x, "completeness": x, "clarity": x, "helpfulness": x, "overall": x}
"""