Skip to content

🏢 国内大厂 AI 工程师面试考点整理 ​

面试优先顺序(通用 AI 应用开发岗位):各公司基础原理、RAG/Agent 项目深挖、系统排障与成本治理题。其余题目用于进阶或特定岗位拓展;实际频率会随岗位和面试轮次变化,产品版本资讯不应当作通用必考题。

来源说明: 根据公开社区面经整理,目前多数题目缺少可核验的岗位、轮次、日期和原始链接,因此不能全部视为已证实的公司真题。第七节保留了原始图文链接,但仍属于候选人的公开自述,并非企业官方题库。 更新: 2026-08-18 使用方式: 将公司标签视为分类线索,优先学习稳定考点;不要据此推断某家公司一定会问。

📋 目录 ​

  1. 字节跳动
  2. 阿里巴巴
  3. 美团
  4. 百度
  5. 通用准备建议
  6. 腾讯
  7. 石头科技 AI Agent 工程题(公开面经精编)

一、字节跳动 ​

岗位特点 ​

  • 侧重实战能力和工程落地
  • 重视LLM/RAG/Agent技术栈
  • 喜欢问开放性问题
  • 重视代码实现能力

高频面试题 ​

1. LLM基础篇 ​

字节 Q1:Transformer 自注意力机制如何工作? ​

Transformer自注意力从QKV映射到加权汇总的完整机制图解

🧠 图解记忆: Q 找相关 K,再按权重汇总 V;多头注意力让模型同时观察不同关系。

💡 答案要点

题目: 请详细解释Transformer模型中的self-attention机制是如何工作的?为什么它比RNN更适合处理长序列?

答案要点:

Self-Attention工作原理:

展开 Python 代码示例(37 行)
python
def self_attention(Q, K, V):
    """
    Q: Query矩阵 (batch, seq_len, d_k)
    K: Key矩阵 (batch, seq_len, d_k)
    V: Value矩阵 (batch, seq_len, d_v)
    """
    # 1. 计算注意力分数
    # Attention(Q,K,V) = softmax(QK^T / sqrt(d_k)) * V

    d_k = Q.shape[-1]

    # 2. Q和K做点积
    scores = torch.matmul(Q, K.transpose(-2, -1))  # (batch, seq_len, seq_len)

    # 3. 缩放(避免点积过大导致 Softmax 饱和)
    scores = scores / math.sqrt(d_k)

    # 4. Softmax归一化
    attention_weights = F.softmax(scores, dim=-1)

    # 5. 加权求和V
    output = torch.matmul(attention_weights, V)

    return output, attention_weights

# 示例
seq_len = 5
d_model = 512
d_k = d_v = 64

Q = torch.randn(1, seq_len, d_k)
K = torch.randn(1, seq_len, d_k)
V = torch.randn(1, seq_len, d_v)

output, weights = self_attention(Q, K, V)
print(output.shape)  # (1, 5, 64)
print(weights.shape)  # (1, 5, 5) - 注意力矩阵

为什么比RNN好:

维度RNNSelf-Attention
并行性❌ 串行计算✅ 完全并行
长依赖❌ 梯度消失✅ 直接连接
计算复杂度O(n·d²)O(n²·d)
长文本❌ 信息丢失✅ 全局视野

面试话术:

"Self-Attention的核心是让每个词都能直接看到序列中的所有其他词。计算分3步:Q和K点积得分数,Softmax归一化,加权求和V。相比RNN,它最大优势是并行计算和直接的长距离依赖,不会梯度消失。代价是O(n²)复杂度,所以超长文本需要优化如FlashAttention。"

字节 Q2:位置编码为什么必要?绝对位置、RoPE、ALiBi 有何区别? ​

绝对位置编码RoPE和ALiBi机制与外推能力对比图解

🧠 图解记忆: 绝对位置做加法,RoPE 旋转 Q/K,ALiBi 在注意力分数里给远距离扣分。

💡 答案要点

题目: 什么是位置编码?为什么Transformer必需它?请列举至少两种实现方式并对比。

答案要点:

为什么需要位置编码:

  • Self-Attention是排列不变的(permutation-invariant)
  • 没有位置信息,"我爱你"和"你爱我"的表示完全相同
  • 位置编码注入顺序信息

方式1: 正弦位置编码(原始Transformer)

展开 Python 代码示例(30 行)
python
def sinusoidal_positional_encoding(seq_len, d_model):
    """
    PE(pos, 2i) = sin(pos / 10000^(2i/d_model))
    PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model))
    """
    position = torch.arange(seq_len).unsqueeze(1)  # (seq_len, 1)
    div_term = torch.exp(
        torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model)
    )

    pe = torch.zeros(seq_len, d_model)
    pe[:, 0::2] = torch.sin(position * div_term)  # 偶数位置
    pe[:, 1::2] = torch.cos(position * div_term)  # 奇数位置

    return pe

# 使用
seq_len = 100
d_model = 512
pe = sinusoidal_positional_encoding(seq_len, d_model)

# 可视化
import matplotlib.pyplot as plt
plt.figure(figsize=(15, 5))
plt.imshow(pe[:50, :50], cmap='RdBu', aspect='auto')
plt.xlabel("Dimension")
plt.ylabel("Position")
plt.colorbar()
plt.title("Sinusoidal Positional Encoding")
plt.show()

方式2: 可学习位置编码(BERT)

python
class LearnedPositionalEncoding(nn.Module):
    def __init__(self, max_seq_len, d_model):
        super().__init__()
        # 直接学习一个位置Embedding矩阵
        self.pos_embedding = nn.Embedding(max_seq_len, d_model)

    def forward(self, x):
        seq_len = x.size(1)
        positions = torch.arange(seq_len, device=x.device)
        return x + self.pos_embedding(positions)

# 使用
model = LearnedPositionalEncoding(max_seq_len=512, d_model=768)
x = torch.randn(1, 100, 768)
output = model(x)

方式3: 旋转位置编码(RoPE) - 字节高频考点

python
def rotate_half(x):
    """旋转一半的维度"""
    x1, x2 = x[..., :x.shape[-1]//2], x[..., x.shape[-1]//2:]
    return torch.cat((-x2, x1), dim=-1)

def apply_rotary_pos_emb(q, k, cos, sin):
    """应用RoPE"""
    # q, k: (batch, seq_len, num_heads, head_dim)
    q_embed = (q * cos) + (rotate_half(q) * sin)
    k_embed = (k * cos) + (rotate_half(k) * sin)
    return q_embed, k_embed

# RoPE的优势:
# 1. 相对位置信息(query和key之间的距离)
# 2. 外推性好(训练512可推理1024+)
# 3. 不增加参数

对比表:

方法优点缺点使用
正弦编码无参数,外推性好固定模式GPT-3
可学习编码灵活,可适应数据超出已训练位置通常需要扩展并继续训练BERT
RoPE相对位置,外推性强实现复杂LLaMA

面试话术:

"不含位置编码的 Self-Attention 无法单独表达 token 顺序。Sin/Cos 编码无训练参数且可以计算任意位置;可学习位置向量在训练范围内灵活;RoPE 让注意力分数携带相对位置信息。任何方案在训练长度之外都需要专项评测,不能直接宣称外推最好。"

字节 Q3:MHA、MQA、GQA 有什么区别? ​

MHA独立KVMQA全共享GQA分组共享及KVCache取舍图解

🧠 图解记忆: MHA 各用各的,MQA 全部共享,GQA 分组共享,在质量和 KV Cache 成本间折中。

💡 答案要点

题目: 请解释Multi-Head Attention (MHA)、Multi-Query Attention (MQA)、Grouped-Query Attention (GQA)的区别。

答案要点:

MHA (Multi-Head Attention) - 标准方法

展开 Python 代码示例(37 行)
python
class MultiHeadAttention(nn.Module):
    def __init__(self, d_model=512, num_heads=8):
        super().__init__()
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        # 每个head都有独立的Q、K、V
        self.W_q = nn.Linear(d_model, d_model)  # 8个head,每个64维
        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 forward(self, x):
        batch_size, seq_len, d_model = x.shape

        # 1. 线性变换
        Q = self.W_q(x)  # (batch, seq_len, d_model)
        K = self.W_k(x)
        V = self.W_v(x)

        # 2. 分割成多个head
        Q = Q.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        K = K.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        V = V.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        # 形状: (batch, num_heads, seq_len, d_k)

        # 3. Attention
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
        attn = F.softmax(scores, dim=-1)
        output = torch.matmul(attn, V)

        # 4. 拼接
        output = output.transpose(1, 2).contiguous().view(batch_size, seq_len, d_model)
        return self.W_o(output)

# KV Cache大小: batch * seq_len * num_heads * head_dim * 2 (K和V)
# 8个head,每个64维 → 1024 tokens需要 1*1024*8*64*2*2字节 = 2MB (FP16)

MQA (Multi-Query Attention) - Google提出

展开 Python 代码示例(30 行)
python
class MultiQueryAttention(nn.Module):
    def __init__(self, d_model=512, num_heads=8):
        super().__init__()
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        # Q有多个head,但K和V只有1个(共享)
        self.W_q = nn.Linear(d_model, d_model)  # 8个head
        self.W_k = nn.Linear(d_model, self.d_k)  # 只1个head!
        self.W_v = nn.Linear(d_model, self.d_k)  # 只1个head!
        self.W_o = nn.Linear(d_model, d_model)

    def forward(self, x):
        batch_size, seq_len, d_model = x.shape

        Q = self.W_q(x).view(batch, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(x).unsqueeze(1)  # (batch, 1, seq_len, d_k) - 广播给所有head
        V = self.W_v(x).unsqueeze(1)  # (batch, 1, seq_len, d_k)

        # 所有Q head共享同一个K和V
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
        attn = F.softmax(scores, dim=-1)
        output = torch.matmul(attn, V)

        output = output.transpose(1, 2).contiguous().view(batch_size, seq_len, d_model)
        return self.W_o(output)

# KV Cache大小: batch * seq_len * 1 * head_dim * 2
# 只1个head → 1024 tokens需要 1*1024*1*64*2*2字节 = 256KB (FP16)
# 相比MHA减少8倍!

GQA (Grouped-Query Attention) - LLaMA 2

展开 Python 代码示例(37 行)
python
class GroupedQueryAttention(nn.Module):
    def __init__(self, d_model=512, num_heads=8, num_kv_heads=2):
        super().__init__()
        self.num_heads = num_heads
        self.num_kv_heads = num_kv_heads  # K/V的head数(比Q少)
        self.num_queries_per_kv = num_heads // num_kv_heads  # 每个KV对应几个Q
        self.d_k = d_model // num_heads

        self.W_q = nn.Linear(d_model, d_model)  # 8个head
        self.W_k = nn.Linear(d_model, num_kv_heads * self.d_k)  # 2个head
        self.W_v = nn.Linear(d_model, num_kv_heads * self.d_k)  # 2个head
        self.W_o = nn.Linear(d_model, d_model)

    def forward(self, x):
        batch_size, seq_len, d_model = x.shape

        Q = self.W_q(x).view(batch, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        # (batch, 8, seq_len, 64)

        K = self.W_k(x).view(batch, seq_len, self.num_kv_heads, self.d_k).transpose(1, 2)
        V = self.W_v(x).view(batch, seq_len, self.num_kv_heads, self.d_k).transpose(1, 2)
        # (batch, 2, seq_len, 64)

        # 重复K和V,让每个KV对应4个Q
        K = K.repeat_interleave(self.num_queries_per_kv, dim=1)  # (batch, 8, seq_len, 64)
        V = V.repeat_interleave(self.num_queries_per_kv, dim=1)

        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
        attn = F.softmax(scores, dim=-1)
        output = torch.matmul(attn, V)

        output = output.transpose(1, 2).contiguous().view(batch_size, seq_len, d_model)
        return self.W_o(output)

# KV Cache大小: batch * seq_len * num_kv_heads * head_dim * 2
# 2个KV head → 1024 tokens需要 1*1024*2*64*2*2字节 = 512KB
# 是MHA的1/4,是MQA的2倍

对比总结:

方法Q headsK/V headsKV Cache质量速度使用
MHA882MB⭐⭐⭐⭐⭐⭐⭐GPT-3
MQA81256KB⭐⭐⭐⭐⭐⭐⭐⭐PaLM
GQA82512KB⭐⭐⭐⭐⭐⭐⭐⭐LLaMA 2

面试话术:

"MHA是标准的多头注意力,每个head都有独立的Q、K、V,质量最好但KV Cache大。MQA让所有head共享1个K和V,Cache减少8倍推理快,但质量略降。GQA是折中方案,8个Q head对应2个KV head,Cache减少4倍,质量接近MHA。LLaMA 2用GQA达到性能和效率平衡,字节面试常问这个演进逻辑。"

2. RAG系统篇 ​

字节 Q4:如何设计完整的生产级 RAG 流程? ​

生产级RAG离线知识加工在线检索重排生成评估完整流程图解

🧠 图解记忆: 离线建好可检索知识,在线先找准、再排好,最后基于证据生成并持续评估。

💡 答案要点

题目: 设计一个完整的RAG系统,从数据准备到最终生成,详细描述每个步骤。

答案要点:

RAG完整流程(7步):

展开 Python 代码示例(106 行)
python
class RAGSystem:
    def __init__(self):
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.vectordb = Qdrant(...)
        self.llm = ChatOpenAI(model="qwen3.5-flash")
        self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

    def build_knowledge_base(self, documents):
        """Step 1-3: 构建知识库"""

        # Step 1: 文档加载
        from langchain.document_loaders import PyPDFLoader, TextLoader

        docs = []
        for doc_path in documents:
            if doc_path.endswith('.pdf'):
                loader = PyPDFLoader(doc_path)
            else:
                loader = TextLoader(doc_path)
            docs.extend(loader.load())

        # Step 2: 文档切块(Chunking)
        from langchain.text_splitter import RecursiveCharacterTextSplitter

        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,      # 每块500字符
            chunk_overlap=50,    # 重叠50字符(保持上下文)
            separators=["\n\n", "\n", "。", "!", "?", " ", ""]
        )

        chunks = text_splitter.split_documents(docs)
        print(f"切分成{len(chunks)}个chunk")

        # Step 3: Embedding + 存入向量库
        texts = [chunk.page_content for chunk in chunks]
        embeddings = self.embedding_model.encode(texts)

        self.vectordb.add(
            embeddings=embeddings,
            documents=texts,
            metadatas=[chunk.metadata for chunk in chunks]
        )

    def retrieve_and_generate(self, query):
        """Step 4-7: 检索+生成"""

        # Step 4: Query Embedding
        query_embedding = self.embedding_model.encode([query])[0]

        # Step 5: 向量检索(召回Top-K)
        results = self.vectordb.search(
            query_embedding,
            limit=20  # 先召回20个候选
        )

        # Step 6: Rerank(精排)
        candidate_docs = [r['document'] for r in results]

        # 计算Query与每个Doc的相关性分数
        pairs = [[query, doc] for doc in candidate_docs]
        scores = self.reranker.predict(pairs)

        # 按分数排序,取Top-5
        ranked_results = sorted(
            zip(candidate_docs, scores),
            key=lambda x: x[1],
            reverse=True
        )[:5]

        top_docs = [doc for doc, score in ranked_results]

        # Step 7: LLM生成
        context = "\n\n".join([f"文档{i+1}: {doc}" for i, doc in enumerate(top_docs)])

        prompt = f"""
        请基于以下文档回答问题。如果文档中没有答案,请明确说"文档中未找到相关信息"。

        文档:
        {context}

        问题: {query}

        回答(要求引用文档编号):
        """

        answer = self.llm.invoke(prompt).content

        return {
            "answer": answer,
            "sources": top_docs,
            "num_retrieved": len(results)
        }

# 使用
rag = RAGSystem()

# 构建知识库
rag.build_knowledge_base([
    "company_docs/产品手册.pdf",
    "company_docs/FAQ.txt"
])

# 查询
result = rag.retrieve_and_generate("如何退货?")
print(result["answer"])
print(f"引用了{len(result['sources'])}个文档")

关键优化点:

  1. Chunking策略

    • 固定长度(500字符) + 重叠(50字符)
    • 按语义分割(段落/句子优先)
    • 代码块/表格特殊处理
  2. 混合检索

    python
    # 向量检索 + BM25关键词检索
    from rank_bm25 import BM25Okapi
    
    # BM25检索
    bm25_results = bm25_search(query, top_k=20)
    
    # 向量检索
    vector_results = vector_search(query, top_k=20)
    
    # RRF融合
    final_results = reciprocal_rank_fusion([bm25_results, vector_results])
  3. 元数据过滤

    python
    # 按时间/部门/权限过滤
    results = vectordb.search(
        query_embedding,
        filter={
            "department": "销售部",
            "created_at": {"$gte": "2024-01-01"}
        }
    )

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "RAG分7步:1)文档加载2)切块(500字符+50重叠)3)Embedding存向量库4)Query编码5)向量检索Top-20 6)Rerank精排Top-5 7)LLM生成。关键优化3点:切块策略要保留语义完整性,混合检索(向量+BM25)提升召回,Rerank用CrossEncoder提升精度。实测Rerank让答案准确率从70%→85%提升15%。"

3. Agent实战篇 ​

字节 Q5:如何用 ReAct 实现代码审查 Agent? ​

ReAct代码审查Agent使用工具取证并输出结构化问题的循环图解

🧠 图解记忆: ReAct 不是边想边猜,而是用工具取证,再按代码与测试证据完成审查。

💡 答案要点

题目: 用ReAct框架实现一个代码审查Agent,能自动发现代码问题并给出修复建议。

答案要点:

展开 Python 代码示例(43 行)
python
from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent
from langchain.prompts import StringPromptTemplate
from langchain.llms import OpenAI
from langchain.chains import LLMChain
import ast
import subprocess

class CodeReviewAgent:
    def __init__(self):
        self.llm = OpenAI(temperature=0)

        # 定义工具
        self.tools = [
            Tool(
                name="静态代码分析",
                func=self.static_analysis,
                description="用pylint分析代码质量问题"
            ),
            Tool(
                name="安全漏洞扫描",
                func=self.security_scan,
                description="检测SQL注入、XSS等安全问题"
            ),
            Tool(
                name="性能分析",
                func=self.performance_analysis,
                description="分析时间复杂度和性能瓶颈"
            ),
            Tool(
                name="代码修复建议",
                func=self.fix_suggestion,
                description="给出具体的修复代码"
            )
        ]

        # ReAct Prompt模板
        self.prompt_template = """
        你是一个专业的代码审查专家。请按照ReAct模式分析代码。

        可用工具:
        {tools}

        代码:
        ```python

</details>
        {code}
        ```

        请按以下格式思考和行动:

        Thought: 我需要分析这段代码的问题
        Action: 静态代码分析
        Observation: [工具返回结果]

        Thought: 发现了XX问题,需要进一步检查安全性
        Action: 安全漏洞扫描
        Observation: [工具返回结果]

        ...继续思考和行动,直到完成审查...

        Final Answer: [完整的审查报告]

        开始:
        Thought: {agent_scratchpad}
        """

        self.agent = self._create_agent()

    def static_analysis(self, code):
        """静态代码分析"""
        # 保存代码到临时文件
        with open("/tmp/code_review.py", "w") as f:
            f.write(code)

        # 运行pylint
        result = subprocess.run(
            ["pylint", "/tmp/code_review.py"],
            capture_output=True,
            text=True
        )

        return result.stdout

    def security_scan(self, code):
        """安全漏洞扫描"""
        issues = []

        # 检测SQL注入
        if "execute(" in code and "%" in code:
            issues.append("⚠️  可能存在SQL注入风险: 使用字符串拼接构建SQL")

        # 检测硬编码密钥
        if "password" in code.lower() or "api_key" in code.lower():
            issues.append("⚠️  发现硬编码的敏感信息")

        # 检测eval/exec
        if "eval(" in code or "exec(" in code:
            issues.append("🚨 危险: 使用了eval/exec,可能导致代码注入")

        return "\n".join(issues) if issues else "✅ 未发现明显安全问题"

    def performance_analysis(self, code):
        """性能分析"""
        try:
            tree = ast.parse(code)

            issues = []

            # 检测嵌套循环
            for node in ast.walk(tree):
                if isinstance(node, ast.For):
                    for child in ast.walk(node):
                        if isinstance(child, ast.For) and child != node:
                            issues.append("⚠️  发现嵌套循环,时间复杂度可能是O(n²)")

            # 检测list comprehension vs 循环
            has_list_comp = any(isinstance(node, ast.ListComp) for node in ast.walk(tree))
            if not has_list_comp:
                issues.append("💡 建议: 可以使用列表推导式提升性能")

            return "\n".join(issues) if issues else "✅ 未发现明显性能问题"

        except:
            return "⚠️  代码语法错误,无法分析"

    def fix_suggestion(self, issue):
        """生成修复建议"""
        prompt = f"""
        代码问题: {issue}

        请给出具体的修复代码示例(Python):
        """

        return self.llm(prompt)

    def review(self, code):
        """执行代码审查"""
        result = self.agent_executor.run(code=code)
        return result

# 使用示例
agent = CodeReviewAgent()

code_to_review = """
def get_user(user_id):
    # SQL注入风险
    query = f"SELECT * FROM users WHERE id = '{user_id}'"
    result = db.execute(query)

    # 硬编码密钥
    api_key = "sk-1234567890abcdef"

    # 嵌套循环
    for item1 in list1:
        for item2 in list2:
            if item1 == item2:
                result.append(item1)

    return result
"""

report = agent.review(code_to_review)
print(report)

# 输出示例:
"""
代码审查报告:

1. 安全问题:
   - 🚨 SQL注入风险: 直接拼接用户输入到SQL语句
   - ⚠️  硬编码API密钥

2. 性能问题:
   - ⚠️  嵌套循环导致O(n²)复杂度
   - 💡 可以用集合交集优化

3. 修复建议:

```python
def get_user(user_id):
    # 修复SQL注入 - 使用参数化查询
    query = "SELECT * FROM users WHERE id = ?"
    result = db.execute(query, (user_id,))

    # 修复硬编码 - 使用环境变量
    api_key = os.getenv("API_KEY")

    # 修复性能 - 使用集合交集
    result = list(set(list1) & set(list2))

    return result
  1. 质量评分: 6/10 建议: 在上线前解决所有安全问题 """

**ReAct关键点:**

1. **Thought(思考):** Agent分析当前情况
2. **Action(行动):** 选择合适的工具执行
3. **Observation(观察):** 获取工具返回结果
4. **循环:** 重复直到得出最终答案

**面试话术:**
> **示例表达(仅在能用本人经历或可复现实验佐证时使用):** "我用ReAct实现了代码审查Agent。定义了4个工具:静态分析/安全扫描/性能分析/修复建议。Agent按Thought→Action→Observation循环工作,先静态分析发现问题,再安全扫描检测漏洞,最后性能分析找瓶颈。关键是工具设计要专注单一职责,Prompt要清晰引导推理过程。实测能发现90%常见代码问题。"

</details>

---

## 二、阿里巴巴

### 岗位特点
- 重视通义千问等自研大模型应用
- 强调业务场景结合(淘宝/钉钉/云)
- 看重架构设计能力
- 注重技术深度和广度

### 高频面试题

#### 1. 通义千问应用

##### 阿里 Q1:如何用通义千问构建生产级客服 Agent?

<a href="/assets/illustrations/18-big-tech-interview-questions/q06-alibaba-qwen-customer-agent.webp"><img src="/assets/illustrations/18-big-tech-interview-questions/q06-alibaba-qwen-customer-agent.webp" alt="通义千问客服Agent意图路由知识检索受控工具和人工兜底架构图解" width="100%"></a>

> 🧠 **图解记忆:** 客服 Agent = 意图路由 + 知识检索 + 受控工具 + 人工兜底,业务权限留在服务层。

<details>
<summary>💡 答案要点</summary>

**题目:** 设计一个基于通义千问的智能客服系统,包含知识库检索、订单查询、情感分析等功能。

**答案要点:**

<details>
<summary>展开 Python 代码示例(149 行)</summary>

```python
from dashscope import Generation
import dashscope

class TongyiCustomerServiceAgent:
    def __init__(self, api_key):
        dashscope.api_key = api_key
        self.model = "qwen-max"

        # 知识库(简化示例)
        self.knowledge_base = {
            "退货政策": "7天无理由退货,商品需保持完好...",
            "配送时间": "一般3-5个工作日送达...",
            "售后服务": "提供1年免费保修..."
        }

    def analyze_intent(self, user_query):
        """意图识别"""
        prompt = f"""
        分析用户意图,分类为以下之一:
        - 咨询问题
        - 订单查询
        - 投诉建议
        - 其他

        用户问题: {user_query}

        输出格式(JSON):
        {{"intent": "意图类别", "confidence": 0.95}}
        """

        response = Generation.call(
            model=self.model,
            prompt=prompt,
            result_format='message'
        )

        import json
        result = json.loads(response.output.text)
        return result

    def retrieve_knowledge(self, query):
        """知识库检索"""
        # 简化: 关键词匹配
        for key, value in self.knowledge_base.items():
            if key in query:
                return value
        return None

    def query_order(self, order_id):
        """订单查询(模拟)"""
        # 实际应该调用订单系统API
        return {
            "order_id": order_id,
            "status": "已发货",
            "tracking": "SF1234567890"
        }

    def analyze_sentiment(self, text):
        """情感分析"""
        prompt = f"""
        分析以下文本的情感倾向:

        文本: {text}

        输出(JSON):
        {{"sentiment": "正面/负面/中性", "score": 0.85, "keywords": ["关键词"]}}
        """

        response = Generation.call(model=self.model, prompt=prompt)
        import json
        return json.loads(response.output.text)

    def handle_query(self, user_query):
        """处理用户咨询"""

        # Step 1: 意图识别
        intent_result = self.analyze_intent(user_query)
        intent = intent_result["intent"]

        # Step 2: 情感分析
        sentiment = self.analyze_sentiment(user_query)

        # Step 3: 根据意图处理
        if intent == "咨询问题":
            # 检索知识库
            knowledge = self.retrieve_knowledge(user_query)

            if knowledge:
                # 用知识库内容生成回答
                prompt = f"""
                基于以下知识回答用户问题:

                知识: {knowledge}
                问题: {user_query}

                要求: 语气友好,简洁专业
                """

                response = Generation.call(model=self.model, prompt=prompt)
                answer = response.output.text
            else:
                answer = "抱歉,我暂时无法回答这个问题,已转接人工客服..."

        elif intent == "订单查询":
            # 提取订单号
            import re
            order_match = re.search(r'\d{10,}', user_query)

            if order_match:
                order_id = order_match.group()
                order_info = self.query_order(order_id)

                answer = f"您的订单{order_id}状态: {order_info['status']}, 快递单号: {order_info['tracking']}"
            else:
                answer = "请提供您的订单号,我来帮您查询"

        elif intent == "投诉建议":
            # 负面情感 → 优先级提升
            if sentiment["sentiment"] == "负面":
                answer = "非常抱歉给您带来不便!我已将您的问题标记为高优先级,客服主管会在30分钟内与您联系。"
            else:
                answer = "感谢您的反馈!我们会认真处理您的建议。"

        else:
            answer = "您可以咨询产品信息、查询订单或提出建议,我都很乐意帮助您!"

        return {
            "answer": answer,
            "intent": intent,
            "sentiment": sentiment
        }

# 使用
agent = TongyiCustomerServiceAgent(api_key="your-dashscope-api-key")

# 测试
queries = [
    "你们的退货政策是什么?",
    "我的订单1234567890到哪了?",
    "你们的产品质量太差了,要退款!"
]

for query in queries:
    result = agent.handle_query(query)
    print(f"问题: {query}")
    print(f"意图: {result['intent']}")
    print(f"情感: {result['sentiment']['sentiment']}")
    print(f"回答: {result['answer']}")
    print("-" * 50)

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "我用通义千问设计了智能客服Agent。核心3步:1)意图识别(咨询/查询/投诉)2)情感分析(负面情绪提优先级)3)分类处理(咨询查知识库,订单调API,投诉转人工)。关键是Prompt设计要结构化输出JSON,方便后续处理。通义千问的中文理解能力强,意图识别准确率>95%。实测负面情绪客户30分钟响应,满意度提升20%。"

2. A2A多智能体协作 ​

阿里 Q2:A2A 与普通 Agent 调用有什么区别? ​

普通Agent工具调用与A2A自治智能体协议协作对比图解

🧠 图解记忆: 普通调用把 Agent 当工具,A2A 让自治 Agent 用协议发现能力、协商任务和同步状态。

💡 答案要点

题目: 解释A2A(Agent-to-Agent)框架,它与传统单Agent或多Agent框架有何不同?

答案要点:

A2A (Agent-to-Agent Communication Protocol)

传统Multi-Agent:
  Agent A → 共享内存/消息队列 → Agent B
  (间接通信,需要中心化协调)

A2A:
  Agent A ←→ 标准化协议 ←→ Agent B
  (直接通信,去中心化)

A2A协议示例:

展开 Python 代码示例(175 行)
python
from typing import Dict, List
import json

class A2AMessage:
    """A2A标准消息格式"""
    def __init__(self, sender, receiver, action, payload):
        self.sender = sender        # 发送者Agent ID
        self.receiver = receiver    # 接收者Agent ID
        self.action = action        # 动作类型
        self.payload = payload      # 消息内容
        self.timestamp = time.time()

    def to_json(self):
        return {
            "sender": self.sender,
            "receiver": self.receiver,
            "action": self.action,
            "payload": self.payload,
            "timestamp": self.timestamp
        }

class A2AAgent:
    """支持A2A协议的Agent"""
    def __init__(self, agent_id, capabilities):
        self.agent_id = agent_id
        self.capabilities = capabilities  # Agent能做什么
        self.message_queue = []

    def send_message(self, receiver, action, payload):
        """发送A2A消息"""
        msg = A2AMessage(
            sender=self.agent_id,
            receiver=receiver,
            action=action,
            payload=payload
        )

        # 通过消息总线发送(简化)
        message_bus.publish(msg)

    def receive_message(self, message: A2AMessage):
        """接收并处理消息"""
        self.message_queue.append(message)

        # 根据action类型处理
        if message.action == "REQUEST":
            response = self.handle_request(message.payload)
            self.send_message(
                receiver=message.sender,
                action="RESPONSE",
                payload=response
            )

        elif message.action == "DELEGATE":
            # 任务委托
            self.execute_task(message.payload)

    def handle_request(self, payload):
        """处理请求"""
        # 实现具体业务逻辑
        pass

# 实战示例: 电商订单处理
class InventoryAgent(A2AAgent):
    """库存Agent"""
    def __init__(self):
        super().__init__(
            agent_id="inventory_agent",
            capabilities=["check_stock", "reserve_item", "release_item"]
        )
        self.stock = {"iPhone15": 100, "MacBook": 50}

    def handle_request(self, payload):
        action = payload["action"]

        if action == "check_stock":
            product = payload["product"]
            return {"available": self.stock.get(product, 0) > 0}

        elif action == "reserve_item":
            product = payload["product"]
            if self.stock.get(product, 0) > 0:
                self.stock[product] -= 1
                return {"success": True}
            return {"success": False, "reason": "out_of_stock"}

class PaymentAgent(A2AAgent):
    """支付Agent"""
    def __init__(self):
        super().__init__(
            agent_id="payment_agent",
            capabilities=["process_payment", "refund"]
        )

    def handle_request(self, payload):
        action = payload["action"]

        if action == "process_payment":
            amount = payload["amount"]
            # 调用支付网关...
            return {"success": True, "transaction_id": "TXN123456"}

class OrderAgent(A2AAgent):
    """订单Agent(协调者)"""
    def __init__(self):
        super().__init__(
            agent_id="order_agent",
            capabilities=["create_order", "cancel_order"]
        )

    def create_order(self, product, quantity, amount):
        """创建订单 - 需要协调多个Agent"""

        # Step 1: 检查库存
        self.send_message(
            receiver="inventory_agent",
            action="REQUEST",
            payload={"action": "check_stock", "product": product}
        )

        # 等待响应(简化,实际应该异步)
        stock_response = self.wait_for_response("inventory_agent")

        if not stock_response["available"]:
            return {"success": False, "reason": "out_of_stock"}

        # Step 2: 预留库存
        self.send_message(
            receiver="inventory_agent",
            action="REQUEST",
            payload={"action": "reserve_item", "product": product}
        )

        reserve_response = self.wait_for_response("inventory_agent")

        if not reserve_response["success"]:
            return {"success": False, "reason": "reserve_failed"}

        # Step 3: 处理支付
        self.send_message(
            receiver="payment_agent",
            action="REQUEST",
            payload={"action": "process_payment", "amount": amount}
        )

        payment_response = self.wait_for_response("payment_agent")

        if not payment_response["success"]:
            # 支付失败,释放库存
            self.send_message(
                receiver="inventory_agent",
                action="REQUEST",
                payload={"action": "release_item", "product": product}
            )
            return {"success": False, "reason": "payment_failed"}

        # Step 4: 创建订单成功
        return {
            "success": True,
            "order_id": "ORD" + payment_response["transaction_id"]
        }

# 使用
inventory = InventoryAgent()
payment = PaymentAgent()
order = OrderAgent()

result = order.create_order(
    product="iPhone15",
    quantity=1,
    amount=7999
)

print(result)
# {"success": True, "order_id": "ORDTXN123456"}

A2A vs 传统Multi-Agent:

维度传统Multi-AgentA2A
通信方式共享内存/消息队列标准化协议
协调中心化调度器去中心化,P2P
扩展性增加Agent需改架构即插即用
跨平台困难容易(协议标准化)
容错中心节点故障全挂单Agent故障不影响其他

面试话术:

"A2A是阿里提出的Agent间通信协议,核心是标准化消息格式(sender/receiver/action/payload)和去中心化通信。传统Multi-Agent依赖中心调度器,A2A让Agent直接P2P通信。优势是扩展性强,新Agent只要实现A2A协议就能加入系统。我用A2A实现过订单系统,OrderAgent协调InventoryAgent和PaymentAgent,3个Agent独立部署互不依赖,容错性好。"


三、美团 ​

岗位特点 ​

  • 强调场景落地(外卖/到店/酒旅)
  • 重视系统稳定性和性能
  • 关注成本优化
  • 看重问题解决能力

高频面试题 ​

美团 Q1:大模型应用常见问题如何排查和缓解? ​

大模型应用幻觉延迟成本与安全问题证据驱动排查图解

🧠 图解记忆: 先按链路定位证据,再对症缓解,最后用同一评测集验证改动是否真实有效。

💡 答案要点

题目: 列举LLM在实际应用中的主要问题,并针对每个问题给出解决方案。

答案要点:

问题1: 幻觉(Hallucination)

  • 表现: 编造不存在的事实
  • 原因: 概率预测,不是事实查询
  • 解决方案:
    python
    # 方案1: RAG
    def rag_answer(question):
        docs = vectordb.search(question)
        prompt = f"基于文档: {docs}\n回答: {question}"
        return llm.generate(prompt, temperature=0.2)
    
    # 方案2: Self-Consistency
    answers = [llm.generate(question) for _ in range(5)]
    final = most_common(answers)  # 投票
    
    # 方案3: 引用溯源
    prompt = "回答问题并标注来源: {question}"
  • 效果: RAG降80%幻觉,Self-Consistency提升15%准确率

问题2: 长尾知识覆盖不足

  • 表现: 专业/小众知识回答不准
  • 解决方案:
    python
    # 垂直领域微调
    from peft import LoraConfig, get_peft_model
    
    lora_config = LoraConfig(
        r=8,
        lora_alpha=16,
        target_modules=["q_proj", "v_proj"],
        task_type="CAUSAL_LM"
    )
    
    model = get_peft_model(base_model, lora_config)
    
    # 在专业数据上微调
    trainer.train(medical_dataset)  # 医疗数据

问题3: 数据新鲜度

  • 表现: 知识截止到训练时间
  • 解决方案:
    python
    # 动态知识注入
    def answer_with_search(question):
        # 1. 判断是否需要实时信息
        if requires_realtime(question):
            # 2. 搜索最新信息
            search_results = web_search(question)
    
            # 3. 结合搜索结果回答
            prompt = f"""
            最新信息: {search_results}
            问题: {question}
            回答:
            """
            return llm.generate(prompt)
        else:
            return llm.generate(question)

问题4: 复读机问题

  • 表现: 重复生成相同内容
  • 原因: 温度过低或采样策略单一
  • 解决方案:
    python
    # 方案1: Repetition Penalty
    response = llm.generate(
        prompt,
        repetition_penalty=1.2  # >1惩罚重复
    )
    
    # 方案2: 多样性采样
    response = llm.generate(
        prompt,
        temperature=0.7,
        top_p=0.9,
        top_k=50
    )
    
    # 方案3: 检测并重新生成
    if has_repetition(response):
        response = llm.generate(prompt, temperature=0.9)

问题5: 推理计算和内存挑战

  • 表现: 70B模型需要140GB显存
  • 解决方案:
    python
    # 方案1: 量化
    from transformers import BitsAndBytesConfig
    
    quantization_config = BitsAndBytesConfig(
        load_in_4bit=True,  # 4bit量化
        bnb_4bit_compute_dtype=torch.float16
    )
    
    model = AutoModelForCausalLM.from_pretrained(
        "meta-llama/Llama-2-70b",
        quantization_config=quantization_config
    )
    # 显存从140GB → 35GB
    
    # 方案2: 模型路由
    def smart_routing(question):
        complexity = estimate_complexity(question)
    
        if complexity < 0.3:
            return llm_small.generate(question)  # 7B
        elif complexity < 0.7:
            return llm_medium.generate(question)  # 13B
        else:
            return llm_large.generate(question)  # 70B
    
    # 成本降低60%

问题6: 偏见问题

  • 表现: 性别/种族/地域偏见
  • 解决方案:
    python
    # RLHF + 人工反馈
    # 1. 收集偏见案例
    bias_cases = [
        {"prompt": "CEO是...", "bad": "他", "good": "他/她"},
    ]
    
    # 2. 训练Reward Model
    reward_model.train(bias_cases)
    
    # 3. PPO优化
    ppo_trainer.train(policy_model, reward_model)

综合解决方案:

展开 Python 代码示例(40 行)
python
class RobustLLMSystem:
    def __init__(self):
        self.llm = load_model()
        self.vectordb = VectorDB()
        self.search_engine = WebSearch()

    def generate(self, question):
        # 1. 检测问题类型
        question_type = self.classify_question(question)

        # 2. 选择策略
        if question_type == "factual":
            # 事实类 → RAG
            return self.rag_generate(question)

        elif question_type == "realtime":
            # 实时类 → 搜索
            return self.search_generate(question)

        elif question_type == "reasoning":
            # 推理类 → Self-Consistency
            return self.self_consistency_generate(question)

        else:
            # 通用
            return self.llm.generate(question, temperature=0.7)

    def rag_generate(self, question):
        docs = self.vectordb.search(question)
        prompt = f"基于: {docs}\n回答: {question}"
        return self.llm.generate(prompt, temperature=0.2)

    def search_generate(self, question):
        results = self.search_engine.search(question)
        prompt = f"参考: {results}\n回答: {question}"
        return self.llm.generate(prompt)

    def self_consistency_generate(self, question, n=5):
        answers = [self.llm.generate(question, temperature=0.7) for _ in range(n)]
        return most_common(answers)

面试话术:

"LLM主要6大问题:1)幻觉用RAG+引用溯源降80%,2)长尾知识用LoRA微调覆盖,3)数据新鲜度用实时搜索,4)复读机用repetition_penalty惩罚,5)计算成本用4bit量化+模型路由降60%,6)偏见用RLHF纠正。美团场景下我用综合方案:事实类走RAG,实时类走搜索,推理类用Self-Consistency,不同问题不同策略,准确率提升25%成本降50%。"


四、百度 ​

岗位特点 ​

  • 强调文心一言应用
  • 深入考察AI基础理论
  • 重视Agent架构设计
  • 三轮面试层层深入

高频面试题 ​

百度 Q1:大模型为什么会重复生成?如何治理? ​

大模型重复生成概率回路原因与提示解码运行时分层治理图解

🧠 图解记忆: 重复是概率回路越滚越强;提示定边界、解码加约束、运行时检测,且要防止惩罚过强。

💡 答案要点

题目: 为什么LLM会出现复读机现象(重复生成相同内容)?如何从模型原理和工程实践两方面解决?

答案要点:

复读机现象:

用户: 介绍一下北京
模型: 北京是中国的首都,北京是中国的首都,北京是中国的首都...

产生机制:

  1. 自回归特性

    P(w_t | w_1, w_2, ..., w_{t-1})
    
    当前词只依赖历史,如果历史包含重复模式
    → 模型倾向于继续重复
  2. 注意力坍塌

    python
    # Attention权重高度集中在某几个token
    attention_weights = [
        [0.01, 0.01, 0.95, 0.01, 0.01, 0.01],  # 第3个token权重0.95
        [0.01, 0.01, 0.01, 0.94, 0.01, 0.01],  # 第4个token权重0.94
        ...
    ]
    # 导致模型陷入局部最优,不断重复高权重token
  3. Temperature过低

    python
    # Temperature → 0
    probs = softmax(logits / temperature)
    # 分布极度尖锐,总选概率最高的词
    # "北京" → "是" → "中国" → "的" → "首都" (循环)

解决方案:

方案1: Repetition Penalty (最常用)

python
def apply_repetition_penalty(logits, input_ids, penalty=1.2):
    """
    惩罚已出现过的token
    penalty > 1: 降低已出现token的概率
    """
    for token_id in set(input_ids):
        # 如果logits[token_id] < 0,除以penalty会更负(概率更低)
        # 如果logits[token_id] > 0,乘以penalty会更小(概率降低)
        if logits[token_id] < 0:
            logits[token_id] *= penalty
        else:
            logits[token_id] /= penalty

    return logits

# 使用
from transformers import GenerationConfig

config = GenerationConfig(
    repetition_penalty=1.2,  # 推荐1.1-1.5
    temperature=0.7,
    top_p=0.9
)

output = model.generate(input_ids, generation_config=config)

方案2: No Repeat N-gram

python
def no_repeat_ngram(generated_tokens, ngram_size=3):
    """
    禁止重复的N-gram
    例如: 禁止"北京是中国"连续出现2次
    """
    ngrams = {}

    for i in range(len(generated_tokens) - ngram_size + 1):
        ngram = tuple(generated_tokens[i:i+ngram_size])
        ngrams[ngram] = ngrams.get(ngram, 0) + 1

    # 如果某个3-gram出现>1次,下次生成时禁止
    return ngrams

# Hugging Face实现
output = model.generate(
    input_ids,
    no_repeat_ngram_size=3  # 禁止3-gram重复
)

方案3: Diversity Penalty

python
# Beam Search + Diversity
output = model.generate(
    input_ids,
    num_beams=5,
    num_beam_groups=5,  # 分5组
    diversity_penalty=1.0,  # 组间差异性惩罚
    temperature=0.7
)

# 原理: 强制不同beam组探索不同路径
# Group 1: "北京是..."
# Group 2: "作为首都..."
# Group 3: "位于华北..."

方案4: 修改Attention机制

python
class AntiRepetitionAttention(nn.Module):
    def __init__(self):
        super().__init__()
        self.attention = MultiHeadAttention()

    def forward(self, q, k, v, generated_tokens):
        # 标准attention
        attn_weights = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(d_k)

        # 对已生成token的attention降权
        for pos in generated_tokens:
            attn_weights[:, :, pos] *= 0.5  # 降低50%

        attn_weights = F.softmax(attn_weights, dim=-1)
        output = torch.matmul(attn_weights, v)

        return output

方案5: 后处理检测与重试

python
def detect_and_retry(text, model, prompt, max_retries=3):
    """检测重复,重新生成"""

    # 检测连续重复
    words = text.split()
    repetition_rate = count_repetitions(words) / len(words)

    if repetition_rate > 0.3 and max_retries > 0:
        # 重复率>30%,提高temperature重新生成
        new_text = model.generate(
            prompt,
            temperature=0.9,  # 提高随机性
            top_k=50
        )
        return detect_and_retry(new_text, model, prompt, max_retries - 1)

    return text

def count_repetitions(words):
    """统计重复词数"""
    from collections import Counter
    counts = Counter(words)
    return sum(c - 1 for c in counts.values() if c > 1)

综合实战方案:

展开 Python 代码示例(61 行)
python
class AntiRepetitionGenerator:
    def __init__(self, model):
        self.model = model

    def generate(self, prompt, max_length=512):
        # 配置多重防护
        config = GenerationConfig(
            max_length=max_length,

            # 防重复参数
            repetition_penalty=1.2,      # 惩罚重复token
            no_repeat_ngram_size=3,      # 禁止3-gram重复

            # 采样策略
            do_sample=True,
            temperature=0.7,
            top_p=0.9,
            top_k=50,

            # Beam Search多样性
            num_beams=5,
            num_beam_groups=5,
            diversity_penalty=1.0,

            # 长度惩罚(防止过短或过长)
            length_penalty=1.0,
            min_length=20,
        )

        # 生成
        output = self.model.generate(
            self.tokenizer.encode(prompt, return_tensors="pt"),
            generation_config=config
        )

        text = self.tokenizer.decode(output[0], skip_special_tokens=True)

        # 后处理检测
        if self.has_severe_repetition(text):
            # 重试,提高temperature
            config.temperature = 0.9
            output = self.model.generate(...)
            text = self.tokenizer.decode(output[0])

        return text

    def has_severe_repetition(self, text):
        """检测严重重复"""
        words = text.split()

        # 检查连续重复
        for i in range(len(words) - 5):
            if words[i:i+5] == words[i+5:i+10]:
                return True

        # 检查整体重复率
        unique_ratio = len(set(words)) / len(words)
        if unique_ratio < 0.5:  # 独特词<50%
            return True

        return False

效果对比:

方法重复率流畅性成本
无防护40%⭐⭐⭐⭐1x
Repetition Penalty5%⭐⭐⭐⭐1x
+No Repeat N-gram2%⭐⭐⭐1x
+Diversity Penalty1%⭐⭐⭐⭐2x (Beam)
综合方案<0.5%⭐⭐⭐⭐1.5x

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "LLM复读机源于3点:1)自回归特性导致重复模式自我强化2)Attention坍塌到少数token 3)低温采样陷入局部最优。解决用5层防护:1)Repetition Penalty=1.2惩罚已出现词2)No Repeat 3-gram禁止短语重复3)Diversity Penalty让Beam组探索不同路径4)修改Attention降低重复token权重5)后处理检测重试。实测重复率从40%→<0.5%,几乎消除。百度面试必问这个,要能讲清原理和工程方案。"


五、通用准备建议 ​

1. 技术储备清单 ​

必须掌握:

  • ✅ Transformer原理(Self-Attention/位置编码/MHA)
  • ✅ RAG完整流程(切块/Embedding/检索/Rerank/生成)
  • ✅ ReAct Agent框架
  • ✅ Prompt Engineering (CoT/Few-shot/Self-Consistency)
  • ✅ 模型微调(LoRA/RLHF基础)
  • ✅ 推理优化(KV Cache/量化)

加分项:

  • ✅ 多Agent协作(AutoGen/CrewAI/A2A)
  • ✅ 高级Prompt技巧(Tree of Thoughts)
  • ✅ GraphRAG
  • ✅ 自研大模型使用(通义/文心/智谱)

2. 项目准备 ​

至少1个完整项目,包含:

  • 明确的业务场景
  • 技术选型理由
  • 核心实现代码
  • 性能指标数据
  • 遇到的问题与解决方案

示例:

项目: 法律咨询RAG系统
场景: 用户咨询法律问题,Agent查询法条库+案例库回答
技术: LangChain + Qdrant + GPT-4
优化: Rerank提升准确率15%,混合检索召回率+20%
难点: 法条切块保持完整性 → 用正则提取法条编号
效果: 回答准确率85%,平均响应2秒

3. 八股文速记 ​

必背概念 (用"速记卡片"复习):

  • Self-Attention计算公式
  • RoPE vs 正弦编码
  • MHA vs MQA vs GQA
  • RAG vs Fine-tuning
  • ReAct vs Plan-Execute
  • LoRA原理
  • KV Cache工作机制

4. 代码准备 ​

手写代码(白板/IDE):

  • Self-Attention实现
  • RAG Pipeline
  • ReAct Agent
  • LoRA微调脚本

算法题:

  • LeetCode Medium难度
  • 数据结构(链表/树/哈希)
  • 动态规划

5. 面试流程应对 ​

一面(基础):

  • 自我介绍(1分钟,突出亮点)
  • 项目深挖(STAR法则)
  • 技术八股文
  • 代码题

二面(综合):

  • 系统设计(RAG/Agent架构)
  • 问题分析能力
  • 开放题(如何优化XX)
  • 跨领域思考

三面(深度):

  • 职业规划
  • 业务理解
  • 产品思维
  • 团队协作

6. 常见坑 ​

❌ 不要:

  • 背答案不理解原理
  • 项目造假(会被深挖)
  • 不懂装懂
  • 完全不了解公司产品

✅ 要:

  • 诚实回答"不知道"
  • 展示学习能力
  • 主动问面试官问题
  • 准备反向提问

最后更新: 2026-03-10 材料来源: 牛客网、知乎、CSDN 等公开社区内容的二次整理。后续补充题目时应记录原始链接、发布时间、岗位、轮次和匿名化说明;缺少这些字段时标为“来源待核验”。 覆盖公司: 字节、阿里、美团、百度、腾讯


上一模块: AI 编程工具下一模块: 推理框架


返回目录 →


六、腾讯 AI 应用工程师常见考点 ​

公开面经常见题型(公司、职级与轮次未经逐题核验) ​

面试官开场问题:

"自我介绍 + 挑一个你认为最有技术含量的 AI 项目详细讲"

腾讯 Q1:如何讲清一个做过的 RAG 项目及优化手段? ​

RAG项目从业务问题基线实验指标到业务结果的面试证据链图解

🧠 图解记忆: 讲 RAG 不背组件,要讲问题、实验、指标、取舍与业务结果的证据链,只用能证明的数据。

💡 答案要点

回答框架(STAR法则):

S(背景):我负责的短剧平台有530万用户,日活80万,用户经常问"这个短剧在哪看""更新到第几集了"等问题。
T(任务):需要构建一个能精准回答用户问题的知识库系统。
A(行动):我设计了四层优化:
  1. 文档解析层:用 PDF.plumber 解析短剧介绍,用 OCR 处理截图
  2. 语义分块层:按剧情/演员/更新时间三维度切分,每块512 tokens
  3. 检索层:HNSW 索引 + BM25 混合检索,RRF 融合排序
  4. 生成层:加 "请基于以下资料回答,不要编造" 约束 prompt
R(结果):用户问题准确率从 65% 提升到 89%,日均减少客服 30% 工作量。

追问1:分块大小怎么选的?

"我测过 256/512/1024 三个档,512 效果最好。太小(如256)切断了上下文,相关性下降;太大(如1024)引入了太多噪声。Context Cliff 的阈值是 2500 tokens,我控制在 512 是因为用户问题通常简短,512 的块既能覆盖完整语义,又不会超过阈值。"

追问2:检索用了什么向量模型?Embedding 怎么调优的?

"用的是 BGE-large-zh,1536 维。调优方法:1)用锚点文本构建难负例;2)温度从 0.01→0.05 逐步调;3)用 MRL 训练加速。最后在测试集上 Recall@10 从 78% 提升到 91%。"

腾讯 Q2:多 Agent 系统中如何通信?遇到冲突怎么办? ​

多Agent中心调度与A2A通信以及资源结果状态冲突治理图解

🧠 图解记忆: 先统一消息与状态,再按资源、结果、数据三类冲突分别治理;本质仍是分布式一致性问题。

💡 答案要点

两种通信模式:

python
# 模式1:中心调度(简单场景)
class CentralScheduler:
    def route(self, task):
        if "搜索" in task:
            return SearchAgent().handle(task)
        elif "生成" in task:
            return WriterAgent().handle(task)

# 模式2:P2P 通信(复杂场景,如 A2A 协议)
class Agent:
    def __init__(self, name):
        self.name = name
        self.inbox = []  # 消息队列
    
    def send(self, receiver, message):
        # 通过消息队列或 HTTP 发送
        message_queue.put({
            "from": self.name,
            "to": receiver,
            "content": message,
            "type": "request"
        })
    
    def receive(self):
        # 接收其他 Agent 的消息
        return self.inbox.pop()

冲突处理策略:

冲突类型解决方案代码
工具调用冲突(两个 Agent 调用同一资源)互斥锁 + 任务队列mutex.lock()
结果冲突(两个 Agent 输出不一致)仲裁 Agent 投票judge_agent.decide()
状态冲突(数据不一致)乐观锁 + 重试version=1, update if version==1

面试话术:

"我的多 Agent 通信用 A2A 协议,Agent 之间通过标准化消息格式通信。冲突处理分三种:资源冲突用互斥锁,输出冲突用仲裁 Agent 投票,状态冲突用乐观锁。我在订单系统用过这个,InventoryAgent 和 PaymentAgent 同时修改库存,通过消息队列串行化,完美解决了并发冲突。"

腾讯 Q3:线上 RAG 响应很慢,如何系统排查? ​

RAG总延迟拆分检索排队TTFT和解码并按证据排查图解

🧠 图解记忆: 先把总延迟拆成检索、排队、首 Token 和解码,再按 Trace 证据优化最慢的一段。

💡 答案要点

排查思路(从外到内):

第一步:确认是 RAG 的哪一步慢?
    ├── 检索慢 → 向量数据库问题
    ├── 生成慢 → LLM 推理问题
    └── 整体都慢 → 网络/系统问题

第二步:逐层定位

排查工具和方法:

python
# 1. 打点计时
import time
start = time.time()
docs = vector_db.search(query_emb, k=10)
检索耗时 = time.time() - start

start = time.time()
answer = llm.generate(prompt)
生成耗时 = time.time() - start

# 2. 查看 vLLM 监控
# vLLM 自带 Prometheus metrics
# 关键指标:
#   - engine_load_percent: GPU 利用率
#   - time_to_first_token: 首 Token 延迟
#   - time_per_output_token: Token 间延迟

# 3. 查看向量数据库慢查询
# Milvus: 
#   db.config().getMetric().SlowQuery
# Pinecone:
#   index.describe_stats() 查看延迟分布

常见原因和解决方案:

原因症状解决
HNSW ef 参数太小召回率低增大 ef=200→500
向量维度太高检索慢用 PCA 降维
LLM GPU 利用率低生成慢检查 batch_size
网络带宽瓶颈跨服务通信慢优化序列化
KV Cache 不命中延迟抖动增大 cache 容量

面试话术:

"我的排查顺序是:打点确定慢在哪一步(检索还是生成)→ 看监控判断是 GPU 卡脖子还是内存瓶颈 → 最后看具体指标。生产环境我会在 RAG 链路每个节点打日志,延迟超过 500ms 自动告警。有一次排查发现是 Embedding 服务内存泄漏导致 GC 频繁,定位到修了内存管理代码后,P99 从 2s 降到 300ms。"

腾讯AI面试高频追问 ​

追问1:vLLM 的 PagedAttention 和 SGLang 的 RadixAttention 有什么区别?

"PagedAttention 是 vLLM 的分页显存管理,每个请求的 KV Cache 按需分配,显存利用率高。RadixAttention 是 SGLang 的基数树管理,能跨请求复用相同前缀(比如 System Prompt)。对于多轮对话,SGLang 省 40-50% 显存;对于单轮问答,vLLM 性能更稳定。"

追问2:如果用户问了一个需要多跳推理的问题(比如"苹果公司CEO和谷歌CEO谁更关注环保"),你的 RAG 怎么处理?

"单级 RAG 做不到,需要:1)Query Decomposition 把问题分解成两个子问题;2)分别检索苹果CEO环保措施 和 谷歌CEO环保措施;3)最后聚合对比。如果涉及更复杂的关系,我会用 GraphRAG 构建知识图谱,通过实体关系路径找到关联信息。"

追问3:你们怎么做 RAG 的效果评估?有没有自动化的流程?

示例表达(仅在能用本人经历或可复现实验佐证时使用): "我们用 RAGAS 四指标:Faithfulness、Answer Relevancy、Context Recall、Context Precision。每周跑一次评估,用 500 道题覆盖 5 个场景。Pipeline 是:代码变更 → 触发评估 → P80 阈值判断 → 通过才允许上线。评估结果和上次对比,下降超过 5% 自动告警。"


七、公司题补充 ​

腾讯 Q4:微信内 AI 应用的技术挑战与架构设计 ​

微信AI应用身份隐私弹性路由受控工具和多端一致架构图解

🧠 图解记忆: 微信 AI 先守身份与私域边界,再做弹性路由、受控工具和多端一致;高风险操作必须二次确认。

💡 答案要点

微信AI应用场景分析:

微信生态:公众号/小程序/视频号/搜一搜/聊天
核心挑战:
① 私域流量管控严格,AI能力接入受限
② 多端一致性要求高(iOS/Android/PC/Web)
③ 海量并发(亿级日活)
④ 合规要求(用户隐私、数据安全)

技术架构设计:

核心挑战与解决方案:

挑战描述解决方案
并发量大亿级日活,峰值 QPS 100万+模型路由 + 异步队列 + 限流熔断
私域合规微信对AI能力管控严格白名单机制 + 内容审核前置
多端适配iOS/Android/Web表现一致统一 API 网关 + 客户端降级策略
响应延迟用户预期<2秒语义缓存 + 预热模型 + 就近路由
隐私保护不能存用户对话端侧处理 + 阅后即焚 + 加密传输

生产级配置:

python
# 微信AI网关配置
WECHAT_AI_CONFIG = {
    "model_routing": {
        "simple_query": "doubao-pro",      # 简单问题用豆包
        "complex_reasoning": "混元-turbo", # 复杂推理用混元
        "multimodal": "wepro-4v"           # 图文理解用wepro
    },
    "rate_limit": {
        "per_user": 20,    # 每用户每分钟20次
        "per_ip": 100,     # 每IP每分钟100次
        "global": 1000000  # 全局每分钟100万次
    },
    "cache": {
        "enabled": True,
        "ttl_seconds": 300,
        "semantic_threshold": 0.92  # 语义相似度>92%命中缓存
    },
    "compliance": {
        "pre_check": True,    # 发模型前先过合规
        "post_check": True,   # 结果返回前再过合规
        "audit_ratio": 0.05   # 5%抽样人工审核
    }
}

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "微信内AI应用的核心挑战是'合规优先':微信对AI能力管控很严,我们必须做内容审核前置。我设计的多层防护:用户输入→敏感词过滤→模型生成→合规检查→用户。而且微信有亿级日活,需要模型路由+限流熔断+语义缓存三合一,单次请求成本从0.15元降到0.04元。"

腾讯 Q5:向量数据库在业务中的选型决策 ​

向量数据库按数据查询约束与真实负载比较召回延迟成本的选型图解

🧠 图解记忆: 向量库没有万能第一,先定数据与查询约束,再用真实负载比较召回、延迟和总成本。

💡 答案要点

腾讯业务场景分析:

业务类型:
- 微信搜一搜:海量用户-query检索
- 腾讯文档:多人协作的内容检索
- 腾讯视频:内容理解+推荐
- 广告系统:人群定向+Lookalike

选型核心考量:
① 数据规模(亿级)
② 延迟要求(<50ms)
③ 成本控制
④ 多租户隔离

三大方案对比:

维度Pincone(云服务)Milvus(自托管)VectorDB for Tencent(内部)
适用场景快速验证/中小规模大规模/有运维能力超大规模/内部业务
数据规模<10亿向量亿级+10亿+
延迟P99<100ms可定制<50ms(定制优化)
成本按量付费,贵硬件成本可控内部结算
运维免运维需要DBA平台团队支持
多租户按project隔离namespace隔离租户级隔离

选型决策树:

python
def select_vector_db(business_type, data_scale, latency_req, budget):
    """
    向量数据库选型决策树
    """
    if data_scale > 10_000_000_000:  # > 100亿
        return "VectorDB-Tencent"  # 自研或内部平台
    
    if latency_req < 50:  # < 50ms
        if business_type == "user_facing":
            return "Pinecone Serverless"  # 低延迟全球分布
        else:
            return "Milvus Cluster"  # 内部可用
    
    if budget < 10000:  # 月预算<1万
        return "Pinecone Starter"  # 成本可控
    
    # 默认选Milvus(开源可控)
    return "Milvus"

# 实际选型
recommendation = select_vector_db(
    business_type="wechat_search",
    data_scale=50_000_000_000,  # 500亿
    latency_req=30,
    budget=50000
)
# → VectorDB-Tencent(内部自研)

混合检索架构:

python
class TencentVectorSearch:
    def __init__(self):
        self.hnsw = HNSWIndex()      # 快速向量检索
        self.bm25 = BM25Index()      # 关键词补充
        self.reranker = CrossEncoder() # 重排
    
    def search(self, query, top_k=20):
        # Step 1: 向量检索(HNSW)
        vector_results = self.hnsw.search(query, k=top_k*2)
        
        # Step 2: BM25补充(覆盖最新数据,HNSW更新慢)
        bm25_results = self.bm25.search(query, k=top_k)
        
        # Step 3: 融合(RRF)
        fused = self.rrf_fusion(vector_results, bm25_results, k=60)
        
        # Step 4: 重排(CrossEncoder)
        reranked = self.reranker.rerank(query, fused[:top_k])
        
        return reranked

面试话术:

"腾讯内部有自研的VectorDB-Tencent,针对微信场景优化过,能支持500亿向量、P99<30ms。对于新业务,我推荐先Pinecone快速验证(1周上线),确认PMF后迁到Milvus(成本可控),等规模>10亿再迁内部平台。选型核心是:数据规模决定架构,延迟要求决定成本,业务阶段决定路径。"

腾讯 Q6:多模型并存场景下的成本管控方案 ​

多模型按需路由动态升级缓存压缩与质量延迟安全底线成本治理图解

🧠 图解记忆: 先缓存和分流,低置信度再升级;节省成本不能突破质量、延迟与安全底线。

💡 答案要点

多模型并存的必然性:

腾讯AI业务现状:
- 内部模型:混元(Hunyuan)、wepro、 hunyuan-function
- 外部模型:GPT-4、Claude、DeepSeek V4-Flash(部分场景)
- 场景差异:简单问答用小模型,复杂推理用大模型

问题:多模型如何统一管理?成本如何控制?

成本管控四层架构:

展开 Python 代码示例(61 行)
python
# 多模型成本管控完整方案
class ModelCostController:
    def __init__(self):
        self.models = {
            # 内部模型(成本低)
            "混元-turbo": {"price": 0.001, "latency": 800, "quality": 0.85},
            "混元-pro": {"price": 0.01, "latency": 1500, "quality": 0.92},
            "wepro-4v": {"price": 0.02, "latency": 2000, "quality": 0.95},
            
            # 外部模型(成本高)
            "gpt-4o": {"price": 0.005, "latency": 1500, "quality": 0.95},
            "claude-3-opus": {"price": 0.015, "latency": 2000, "quality": 0.97},
        }
        
        self.model_router = ModelRouter()
        self.semantic_cache = SemanticCache()
        self.cost_alert = CostAlert()
    
    def route_and_call(self, query, user_tier):
        """智能路由 + 成本控制"""
        
        # Layer 1: 缓存命中(省100%)
        cached = self.semantic_cache.get(query)
        if cached:
            return cached, "cache", 0
        
        # Layer 2: 意图分类 → 模型选型
        intent = self.classify_intent(query)
        
        # Layer 3: 成本优先 or 质量优先
        if user_tier == "free":
            # 免费用户:成本优先
            model = self.select_cheap_model(intent)
        elif user_tier == "vip":
            # VIP用户:质量优先
            model = self.select_best_model(intent)
        else:
            # 普通用户:均衡
            model = self.select_balanced_model(intent)
        
        # Layer 4: 超预算熔断
        if self.is_over_budget():
            # 降级到更便宜的模型
            model = self.downgrade_model(model)
        
        result = self.call_model(model, query)
        self.record_cost(model, query, result)
        
        return result, model, self.models[model]["price"]
    
    def classify_intent(self, query):
        """意图分类驱动模型选型"""
        # 简单匹配用小模型,复杂推理用大模型
        if self.is_simple_faq(query):
            return "simple"
        elif self.needs_multimodal(query):
            return "multimodal"
        elif self.needs_deep_reasoning(query):
            return "complex"
        else:
            return "general"

成本控制核心指标:

yaml
# 模型成本监控 Dashboard
panels:
  - title: 各模型调用量分布
    type: pie
    expr: sum(increase(model_api_calls_total[1d])) by (model)
  
  - title: 各模型成本占比
    type: bar
    expr: |
      sum(increase(model_cost_total[1d])) by (model)
      / sum(increase(model_cost_total[1d]))
  
  - title: 模型降级触发次数
    type: graph
    expr: sum(rate(model_downgrade_total[5m])) by (reason)
  
  - title: 缓存命中率趋势
    type: line
    expr: cache_hit_rate / cache_total * 100
  
  - title: 预计月度成本(滚动预测)
    type: stat
    expr: sum(increase(model_cost_total[1h])) * 24 * 30

模型降级策略:

python
# 模型降级规则(按场景)
DOWNGRADE_RULES = [
    # 场景1:非核心功能用小模型
    {
        "path": "/api/chat/suggestion",
        "original": "混元-pro",
        "downgrade": "混元-turbo",
        "trigger": "cost > 80% budget"
    },
    # 场景2:超时自动降级
    {
        "path": "/api/chat/search",
        "original": "gpt-4o",
        "downgrade": "混元-turbo",
        "trigger": "latency > 3000ms"
    },
    # 场景3:非高峰期降级
    {
        "path": "/api/chat/background",
        "original": "混元-pro",
        "downgrade": "混元-turbo",
        "trigger": "hour < 9 or hour > 22"  # 非高峰时段
    }
]

成本归因与优化:

python
# 月度成本分析报告
def generate_cost_report():
    report = {
        "total_cost": calculate_total_cost(),
        "by_model": group_by_model(),
        "by_user_tier": group_by_user_tier(),
        "by_feature": group_by_feature(),
        "cache_savings": calculate_cache_savings(),
        "optimization_opportunities": [
            {
                "model": "混元-pro",
                "scenario": "简单FAQ",
                "potential_saving": "30%",
                "action": "切换到混元-turbo"
            },
            {
                "model": "gpt-4o",
                "scenario": "非VIP用户复杂推理",
                "potential_saving": "50%",
                "action": "优先用混元-pro"
            }
        ]
    }
    return report

面试话术:

"多模型并存时,成本管控核心是'让对的模型处理对的任务'。我的方案:意图分类驱动路由(简单FAQ用小模型),语义缓存拦截重复问题(命中率35%省100%成本),超预算自动降级。实际效果:接入大模型后月成本从8万降到3万,质量指标(回答准确率)只下降2%。成本优化不是牺牲质量,而是更聪明地分配资源。"


字节 Q6-Q8 补充 ​

字节 Q6:模型 API 从接入到生产部署要解决哪些问题? ​

模型API统一网关鉴权可靠性降级观测与成本治理生产架构图解

🧠 图解记忆: 接 API 不等于能上线,生产要补齐鉴权、可靠性、降级、观测与成本治理。

💡 答案要点

豆包(Doubao)特点:

豆包优势:
- 字节自研,中文理解好
- 价格便宜(GPT-4的1/10)
- 响应速度快
- 生态集成(抖音/飞书/火山引擎)

适用场景:国内业务、低成本、大规模

接入方案对比:

接入方式适用场景延迟成本稳定性
API 直调快速验证P99<800ms按量计费一般
火山引擎SDK生产部署P99<500ms批量折扣好
私有化部署数据敏感取决于硬件一次性最稳定

生产级接入代码:

展开 Python 代码示例(52 行)
python
from volcenginesdkarkruntime import Ark

class DoubaoClient:
    def __init__(self, api_key: str, model: str = "doubao-pro"):
        self.client = Ark(api_key=api_key)
        self.model = model
        self.retry_config = RetryConfig(
            max_attempts=3,
            base_delay=1.0,
            exponential_base=2
        )
    
    async def chat(self, messages: list, stream: bool = False):
        """豆包 API 调用封装"""
        
        for attempt in range(self.retry_config.max_attempts):
            try:
                response = await self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    stream=stream,
                    temperature=0.3,
                    max_tokens=2000
                )
                return response
                
            except RateLimitError:
                # 限流:指数退避
                await asyncio.sleep(self.retry_config.get_delay(attempt))
                
            except APITimeoutError:
                # 超时:降级到快模型
                return await self.fallback_to_fast_model(messages)
                
            except Exception as e:
                # 其他错误:重试或熔断
                if attempt == self.retry_config.max_attempts - 1:
                    await self.circuit_breaker.open()
                    raise
                await asyncio.sleep(0.5)
    
    async def fallback_to_fast_model(self, messages):
        """降级到 doubao-lite"""
        try:
            return await self.client.chat.completions.create(
                model="doubao-lite",
                messages=messages,
                temperature=0.3,
                max_tokens=1000
            )
        except:
            return {"error": "服务暂时不可用,请稍后再试"}

生产部署关键配置:

yaml
# 字节 AI 网关配置
doubao_gateway:
  models:
    doubao-pro:
      endpoint: "ar-vim-4c26fg9y50001@ar-vim-4c26fg9y50001"
      timeout: 30s
      max_retries: 3
      rate_limit:
        per_second: 100
        per_minute: 5000
        
    doubao-lite:
      endpoint: "ep-20250615001"
      timeout: 15s
      max_retries: 2
      rate_limit:
        per_second: 200
        
  circuit_breaker:
    error_threshold: 5  # 5次错误开路
    recovery_timeout: 60s
    
  fallback_chain:
    - doubao-pro
    - doubao-lite
    - deepseek-v4-flash  # 外部兜底

生产问题排查:

问题原因解决方案
响应慢 P99>2s模型排队扩容或启用缓存
大量超时限流触发降级到lite或增加延迟容忍
回答质量差场景不匹配调高temperature或切模型
成本超预算调用量突增启用限流+缓存

面试话术:

"豆包接入分三步:① 先API验证(1天),确认效果;② 接入火山引擎SDK,配置重试+限流+降级(1周);③ 生产监控,P99延迟>1s自动告警(持续)。我踩过的坑:豆包有每分钟5000次限制,大促期间被限流,后来加了模型降级和异步队列才解决。"

字节 Q7:短视频内容理解系统如何设计? ​

短视频视觉语音字幕音频分轨提取时间对齐融合与多任务输出图解

🧠 图解记忆: 先分轨提取,再按时间对齐融合;实时轻量路径保底,离线重型路径完成深度理解。

💡 答案要点

TikTok 内容理解挑战:

内容类型:短视频(15s-3min)+ 直播 + 评论 + 弹幕
数据规模:10亿+ 视频,百万级创作者
业务场景:内容推荐、内容审核、内容理解、创作者工具

核心难点:
① 视频理解需要多模态(视觉+音频+文本)
② 实时性要求高(直播弹幕需<500ms)
③ 多语言(全球200+国家)

多模态内容理解架构:

展开 Python 代码示例(51 行)
python
class TikTokContentUnderstanding:
    def __init__(self):
        self.video_encoder = VideoEncoder()      # 视频帧编码
        self.audio_encoder = AudioEncoder()      # 音频编码
        self.text_encoder = TextEncoder()        # 文本编码
        self.fusion = MultiModalFusion()        # 多模态融合
        self.cache = RedisVectorCache()         # 结果缓存
    
    async def understand_video(self, video_url: str, tasks: list):
        """视频多模态理解"""
        
        # Step 1: 并行提取多模态特征
        video_emb = await self.video_encoder.encode(video_url, fps=1)  # 每秒1帧
        audio_emb = await self.audio_encoder.encode(video_url)          # 音频特征
        ocr_text = await self.extract_text(video_url)                  # OCR文字
        
        # Step 2: 按任务选择处理流程
        results = {}
        for task in tasks:
            if task == "content_tag":
                # 内容标签(体育/美食/旅游等)
                results[task] = self.classify_content(video_emb, audio_emb, ocr_text)
                
            elif task == "quality_score":
                # 内容质量分(清晰度/剪辑/创意)
                results[task] = self.score_quality(video_emb, audio_emb)
                
            elif task == "creator_intent":
                # 创作者意图(卖货/涨粉/娱乐)
                results[task] = self.understand_intent(ocr_text, audio_emb)
                
            elif task == "sensitive_check":
                # 敏感内容检测
                results[task] = await self.check_sensitive(video_emb, audio_emb, ocr_text)
        
        return results
    
    async def understand_live_stream(self, stream_url: str):
        """直播内容实时理解(弹幕/弹幕情感)"""
        
        # 实时处理流水线
        pipeline = Pipeline([
            ("audio_capture", AudioCapture(stream_url)),
            ("speech_to_text", WhisperX()),          # 实时ASR
            ("sentiment_analysis", SentimentModel()), # 情感分析
            ("keyword_extraction", KeywordExtractor()),
            ("real_time_alert", AlertSystem())        # 实时告警
        ])
        
        async for result in pipeline.run():
            yield result  # 实时输出

多语言处理策略:

python
# TikTok多语言内容理解
MULTI_LANG_CONFIG = {
    "zh": {"model": "doubao-pro-zh", "priority": 1},
    "en": {"model": "gpt-4o", "priority": 2},
    "ja": {"model": "doubao-pro-ja", "priority": 2},
    "ko": {"model": "doubao-pro-ko", "priority": 2},
    "other": {"model": "gpt-4o", "priority": 3}  # 其他语言用GPT-4
}

async def understand_content_lang(content: dict, lang: str):
    """根据语言选择模型"""
    config = MULTI_LANG_CONFIG.get(lang, MULTI_LANG_CONFIG["other"])
    
    # 高优先级语言用本地模型(快+便宜)
    # 低优先级语言用外部模型(质量好)
    if config["priority"] <= 2:
        return await doubao_native(content, config["model"])
    else:
        return await openai_translate_and_process(content)

内容推荐特征工程:

python
# TikTok推荐系统内容理解特征
CONTENT_FEATURES = {
    # 视觉特征
    "visual_scene": "室内/室外/自然/城市",
    "visual_objects": ["人", "食物", "风景", "商品"],
    "visual_quality": 0.0-1.0,  # 清晰度分数
    
    # 音频特征
    "audio_type": "音乐/对话/环境音",
    "audio_language": "zh/en/other",
    "audio_emotion": "positive/negative/neutral",
    
    # 文本特征
    "text_entities": ["人物", "地点", "品牌"],
    "text_topics": ["美食", "旅游", "美妆", "科技"],
    "text_sentiment": 0.0-1.0,
    
    # 互动特征
    "expected_engagement": "high/medium/low",
    "target_audience": ["年轻人", "妈妈", "职场人"]
}

面试话术:

"TikTok内容理解的核心是多模态融合+多语言支持。我的架构:视频帧+音频+OCR并行提取特征 → 按任务分发(标签/质量分/意图/敏感检测)→ 结果缓存。关键优化:① 直播场景用流式处理,<500ms响应;② 多语言用本地模型处理高优先级语言(中文/英语),其他语言翻译后处理;③ 敏感检测前置,涉黄涉暴内容在特征提取阶段就拦截。"

字节 Q8:模型监控体系如何设计指标与告警? ​

模型监控从系统服务质量到业务安全四层指标与可行动告警闭环图解

🧠 图解记忆: 监控要从资源看到业务,告警必须可行动,发布要靠离线—灰度—线上闭环。

💡 答案要点

模型监控四大维度:

┌─────────────────────────────────────────────────────────────┐
│                  LLM 模型监控体系                            │
├─────────────────┬─────────────────┬─────────────────────────┤
│  性能指标        │   质量指标      │   业务指标              │
│  - TTFT         │   - 准确率      │   - 用户满意度          │
│  - TPS          │   - 幻觉率      │   - Task完成率          │
│  - 错误率        │   - 回复相关度  │   - 转化率              │
├─────────────────┼─────────────────┼─────────────────────────┤
│  成本指标        │   安全指标      │   系统指标              │
│  - Token消耗    │   - 敏感词触发  │   - 可用性              │
│  - API调用费    │   - Prompt注入  │   - 错误分布            │
│  - ROI          │   - 合规违规    │   - 限流触发率          │
└─────────────────┴─────────────────┴─────────────────────────┘

核心指标定义与采集:

展开 Python 代码示例(31 行)
python
# 模型监控指标定义
class ModelMetrics:
    # 性能指标
    TTFT = Histogram("llm_time_to_first_token", "首Token时间", ["model"])
    TPS = Histogram("llm_tokens_per_second", "生成速度", ["model"])
    total_latency = Histogram("llm_total_latency", "端到端延迟", ["model"])
    
    # 质量指标
    response_quality = Gauge("llm_response_quality_score", "回复质量分", ["model"])
    hallucination_rate = Gauge("llm_hallucination_rate", "幻觉率", ["model"])
    
    # 成本指标
    input_tokens = Counter("llm_input_tokens_total", "输入Token总数", ["model"])
    output_tokens = Counter("llm_output_tokens_total", "输出Token总数", ["model"])
    api_cost = Counter("llm_api_cost_total", "API费用", ["model"])
    
    # 安全指标
    sensitive_trigger = Counter("llm_sensitive_trigger_total", "敏感词触发", ["category"])
    injection_attempt = Counter("llm_injection_attempt_total", "注入攻击尝试")
    
    async def record_response(self, model, response_data):
        """每次响应后记录指标"""
        self.TTFT.labels(model).observe(response_data.ttft)
        self.input_tokens.labels(model).inc(response_data.input_tokens)
        self.output_tokens.labels(model).inc(response_data.output_tokens)
        self.api_cost.labels(model).inc(response_data.cost)
        
        # 异步质量评估(采样10%)
        if random.random() < 0.1:
            quality_score = await self.evaluate_quality(response_data)
            self.response_quality.labels(model).set(quality_score)

告警规则设计:

展开 Yaml 代码示例(68 行)
yaml
# alertmanager-rules.yaml
groups:
  - name: llm-alerts
    rules:
      # P1: 服务不可用
      - alert: LLMServerDown
        expr: up{job="llm-api"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "LLM服务不可用"
      
      # P2: 延迟过高
      - alert: LLMHighLatency
        expr: |
          histogram_quantile(0.99, 
            rate(llm_total_latency_bucket[5m])
          ) > 5
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "LLM P99延迟超过5秒"
      
      # P3: 质量下降
      - alert: LLMQualityDrop
        expr: |
          llm_response_quality_score < 0.7
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "LLM回复质量持续低于0.7"
      
      # P4: 成本异常
      - alert: LLMCostAnomaly
        expr: |
          sum(increase(llm_api_cost_total[1h])) 
          > 1.5 * avg_over_time(
              sum(increase(llm_api_cost_total[1h]))[7d:1h]
            )
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "LLM成本异常,本小时花费是上周平均的1.5倍"
      
      # P5: 幻觉率飙升
      - alert: LLMHallucinationSpike
        expr: |
          rate(llm_hallucination_detected_total[5m])
          > 0.05  # 5%幻觉率
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "LLM幻觉率超过5%,需要立即检查"
      
      # P6: 安全攻击
      - alert: LLMInjectionAttack
        expr: |
          rate(llm_injection_attempt_total[1m]) > 10
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "检测到Prompt注入攻击,10次/分钟"

监控 Dashboard 设计:

展开 Yaml 代码示例(54 行)
yaml
# Grafana LLM监控Dashboard
dashboard:
  title: "LLM 全链路监控"
  
  rows:
    - title: "业务健康度"
      panels:
        - type: stat
          title: "Task完成率"
          expr: |
            sum(rate(llm_task_completed_total[5m]))
            / sum(rate(llm_task_started_total[5m])) * 100
          thresholds:
            - value: 90
              color: red
            - value: 95
              color: green
        
        - type: gauge
          title: "平均用户满意度"
          expr: avg(llm_user_satisfaction_score)
        
    - title: "性能监控"
      panels:
        - type: timeseries
          title: "延迟趋势 (P50/P95/P99)"
          expr: |
            histogram_quantile(0.50, rate(llm_total_latency_bucket[5m]))
            histogram_quantile(0.95, rate(llm_total_latency_bucket[5m]))
            histogram_quantile(0.99, rate(llm_total_latency_bucket[5m]))
        
        - type: timeseries
          title: "TTFT 趋势"
          expr: histogram_quantile(0.50, rate(llm_time_to_first_token_bucket[5m]))
    
    - title: "成本分析"
      panels:
        - type: timeseries
          title: "日Token消耗趋势"
          expr: sum(increase(llm_input_tokens_total[1d])) + sum(increase(llm_output_tokens_total[1d]))
        
        - type: stat
          title: "预计月度成本"
          expr: sum(increase(llm_api_cost_total[1h])) * 24 * 30
    
    - title: "安全监控"
      panels:
        - type: timeseries
          title: "敏感词触发趋势"
          expr: sum(rate(llm_sensitive_trigger_total[5m])) by (category)
        
        - type: table
          title: "注入攻击 Top 5"
          expr: topk(5, sum(rate(llm_injection_attempt_total[5m])) by (ip))

面试话术:

"我的模型监控体系分四层:① 基础设施层(延迟/吞吐量/错误率);② 模型质量层(幻觉率/准确率);③ 业务层(任务完成率/用户满意度);④ 安全层(敏感词/注入攻击)。告警规则按Severity分级:P1服务不可用立即通知,SRE 5分钟内响应;P2延迟过高15分钟响应;P3质量下降1小时响应。最重要的是成本告警——我设置预计日成本>80%阈值自动通知,防止月底账单爆表。"


七、石头科技 AI Agent 工程题(公开面经精编) ​

来源与可信度: 以下考点识别自程序员峰哥于 2026-08-16 发布的《石头科技 AI Agent 三轮拷打》图文面经。它是候选人的公开面试复盘,能够证明“有人这样复盘过”,但未经企业官方核验,不代表固定题库或招聘承诺。

原帖包含自我介绍、HR 问题以及大量已有专题覆盖的 RAG、ReAct、异步并发、MCP、重排和可观测性问题。本节不机械搬运,只保留新增价值最高的工程题,并按复现频率与迁移价值排序:

优先级本节处理原帖中的代表考点
高频新增 2 道主答案LangGraph 状态恢复、Harness Replay
中高频新增 2 道系统设计智能硬件 Agent、Java 与 Python Agent 服务治理
已有主答案不重复作答RAG 全链路、RRF、Rerank、Embedding、ReAct、MCP、工具重试、Prompt Injection
不纳入题库仅作为面试流程参考自我介绍、离职原因、薪资和到岗时间等 HR 问题

石头科技 Q1:LangGraph State、Checkpoint 与 thread_id 如何设计?中断后怎样准确恢复? ​

LangGraph共享状态、Checkpoint时间线与断点恢复图解

🧠 图解记忆: State 管节点协作,Checkpoint 管执行恢复,thread_id 管一条执行线;跨会话长期记忆要另存 Store 或业务数据库。

💡 答案要点

30 秒回答: State 是节点间的显式数据契约;Checkpoint 是图在每个步骤后的状态快照;thread_id 是一条会话或任务执行线的持久化标识。只把下游节点、恢复、审计真正需要的数据放进共享 State,节点内部临时变量留在节点内;并行节点可能同时写同一字段时必须定义 reducer。恢复时由服务端完成用户与 thread_id 的授权映射,再用同一 thread_id,必要时指定 checkpoint_id 继续执行。外部写操作还要有幂等键,因为状态恢复不等于撤销或去重现实世界的副作用。

State 应该放什么?

数据类型是否进入共享 State示例
下游节点需要的稳定事实是标准化意图、检索文档 ID、工具结果摘要
决策与恢复需要的控制字段是当前阶段、重试次数、待确认动作、错误码
仅供单节点计算的临时值否HTTP 客户端、局部张量、中间解析对象
密钥和完整敏感载荷通常否API Key、完整身份证号;只存引用或脱敏值

State 要有版本化 schema。消息列表等追加型字段用 reducer 合并;状态机阶段、最终决策等单值字段通常显式覆盖。并行分支写同一字段时,如果没有合并规则,就会产生冲突或不确定结果。

text
(tenant_id, user_id, conversation_id)
              │ 服务端鉴权与映射
              ▼
          thread_id
              │
  step 1 ─ checkpoint A ─ step 2 ─ checkpoint B ─ interrupt
                                    │
                                    └─ 用同一 thread_id / checkpoint_id 恢复
  • thread_id 表示执行线程,不应直接等于可猜测的用户 ID,也不能相信客户端自报的租户字段。
  • 开发环境可用内存 saver;生产环境要选择持久化 saver,并设置租户隔离、加密、TTL、备份和清理策略。
  • Checkpoint 记录该步骤的 State、下一节点和任务信息,适合断点续跑、人工审批和时间旅行。
  • 跨线程的用户偏好或长期记忆不属于 Checkpoint,应写入 LangGraph Store、向量库或业务数据库。

恢复边界: Checkpoint 只能恢复图状态,不能自动回滚已经发送的邮件、扣款或创建的工单。所有有副作用的工具都应使用业务幂等键,例如 run_id + node_id + action_version;恢复前先查询动作是否已成功,再决定跳过、补偿还是重试。长任务还应记录 Prompt、模型、工具 schema 和代码版本,否则旧快照在新版本图上可能无法安全恢复。

验证方式: 在每个可能中断的节点后注入故障,检查相同 thread_id 能否恢复;并验证并行 reducer、越权读取、重复写工具、版本升级和 checkpoint 过期五类情况。

常见追问:

  1. 两个并行节点同时更新 messages 时 reducer 怎么写?
  2. Checkpoint 与跨会话长期记忆有什么区别?
  3. 工作流升级后如何兼容旧 State schema?
  4. 为什么恢复成功仍可能重复创建工单?

已有主答案: LangGraph 的 Checkpoint、时间旅行和 Human-in-the-loop

参考: LangGraph Persistence · LangGraph Time Travel

石头科技 Q2:Agent Harness 的 Replay 为什么不能简单重调真实接口?它和 Checkpoint 恢复有什么区别? ​

Agent Harness可重复回放与Checkpoint恢复边界图解

🧠 图解记忆: Checkpoint 为恢复而重执行,Replay 为评测而冻结变量;涉及外部副作用时,优先录制、替身、沙箱和幂等,不能直接重放生产写操作。

💡 答案要点

30 秒回答: Checkpoint Replay 解决“任务从哪里继续”,它通常会重执行 checkpoint 之后的节点,真实 API 也可能再次被调用;Harness Replay 解决“这个版本是否比上个版本好”,要求输入、配置和外部观察尽量可复现。评测回放应记录工具请求与结果,用 fake、record/replay 或沙箱替代生产依赖,写工具默认 dry-run,并固定数据集、Prompt、模型、工具 schema 和 evaluator 版本。二者都叫 replay,但目标、可重复性和副作用边界完全不同。

维度Checkpoint 恢复 / 时间旅行Harness Replay / 回归评测
目标恢复任务、调试某条执行线比较版本、复现 badcase、防回归
起点某个状态快照固定样本、Trace 或数据集版本
外部调用后续节点可能重新调用默认使用录制结果、替身或沙箱
结果要求业务上可继续同条件下可比较、可解释
写副作用靠幂等与补偿保护禁止直连生产写操作
text
生产 Badcase → 脱敏 Trace → 版本化数据集 → 冻结工具观察
       → 候选 Agent Replay → 规则 + LLM Judge → 与基线比较 → 小流量灰度

建议区分三种执行模式:

  • Pure Replay: 所有外部结果固定,适合 Prompt、路由和策略回归。
  • Shadow Replay: 允许只读调用最新知识或检索服务,但输出不影响用户,用于观察环境漂移。
  • Live Canary: 极小流量调用真实依赖,必须有预算、权限、熔断和人工停止开关。

Replay 资产至少包含输入、期望约束、原始 Trace、工具请求/响应摘要、Prompt/模型/知识库/代码版本,以及允许变化的字段。评测不能只看最终文本,还要检查轨迹:是否选对工具、参数是否合法、是否越权、重试是否失控、成本与延迟是否越界。

验证方式: 同一数据集连续回放两次,确认核心评分稳定;故意更换 Prompt 或工具 schema,确认报告能定位差异;向写工具注入测试请求,确认只进入沙箱或被 dry-run 拦截。

常见追问:

  1. 天气、库存等实时结果必须变化时,如何兼顾真实性与可复现?
  2. 哪些字段适合断言,哪些字段只能做语义评估?
  3. LLM Judge 自己不稳定,如何校准?
  4. Replay 数据如何从线上 Trace 脱敏并进入回归集?

已有主答案: 如何评估 Agent 的决策轨迹 · Harness 如何成为 Agent 的安全与评估基础设施

参考: LangGraph Time Travel · LangSmith Evaluation

石头科技 Q3:如何从零设计一个智能硬件助手 Agent?以扫地机状态查询、售后问答和工单创建为例 ​

智能硬件助手Agent三条业务路径与安全底座图解

🧠 图解记忆: 状态查询读设备事实,售后问答读有引用的知识,创建工单走授权、确认和幂等;LLM 负责理解与编排,不负责制造业务事实。

💡 答案要点

30 秒回答: 我不会让一个自由 Agent 直连全部系统,而会先按风险把能力拆成三条路径:设备状态是确定性只读工具;说明书与售后问答走带版本和引用的 RAG;创建工单是高风险写操作,必须校验用户与设备关系、补齐槽位、展示确认页并用幂等键提交。LLM 只做意图识别、澄清、工具编排和结果表达,设备数字孪生、知识库与工单系统才是事实来源。三条路径共享鉴权、策略、Trace、超时和降级底座。

text
App / 客服入口
      │ 身份、租户、设备绑定、会话
      ▼
意图 + 风险路由器
  ├─ 设备状态 ─→ Device Twin / IoT API ─→ 状态 + 数据时间
  ├─ 售后问答 ─→ Query Rewrite → RAG ─→ 答案 + 文档引用
  └─ 创建工单 ─→ 槽位校验 → 用户确认 → 幂等写入 → 工单号

横切能力:权限策略 · 内容安全 · Harness · Trace · 限流 · 降级
路径权威事实源主要风险必备保护
设备状态IoT / 设备数字孪生旧状态被当成实时状态返回采集时间、离线标志与刷新动作
说明书/售后版本化知识库型号错配、无依据回答型号和固件过滤、引用、低置信度转人工
创建工单CRM / 工单系统越权、重复写、错误地址设备归属校验、确认、幂等键、审计与补偿

路由不要只依赖一次 LLM 分类。先用设备绑定、登录状态和操作风险做硬约束,再让模型判断意图;低置信度时澄清而不是猜。组合意图可以用显式工作流,例如“先查故障码,再检索解决方法,仍未解决才建议建单”。

降级策略: IoT 不可用时明确“暂时无法获取实时状态”,不能用模型补造;RAG 低置信度时展示相关文档或转人工;工单服务超时先查询幂等键对应结果,禁止直接重复提交。

评测与观测: 分路径统计意图混淆矩阵、设备工具成功率、知识引用正确率、工单一次成功与重复率、转人工率、TTFT、总延迟和单会话成本。用真实但脱敏的设备状态、故障码和售后对话构造离线集,再做沙箱工具回放。

常见追问:

  1. 用户同时绑定多台设备,型号与设备 ID 如何消歧?
  2. 设备离线时怎样避免模型给出虚假的实时状态?
  3. 工单提交超时后,如何判断应该查询还是重试?
  4. 新固件发布后,知识库和工具 schema 如何灰度升级?

相关专题: AI 场景系统设计 · RAG 系统 · Agent 安全评估

石头科技 Q4:Java 主站如何稳定接入 Python Agent 服务?鉴权、SSE、超时熔断和可观测性放在哪层? ​

Java网关与Python Agent服务职责、流式协议和全链路治理图解

🧠 图解记忆: Java 守入口与业务身份,Python 管模型编排与工具执行;Trace、取消和超时预算必须贯通两端,写操作靠幂等而不是盲目重试。

💡 答案要点

30 秒回答: Java 主站或网关负责用户鉴权、租户与会话绑定、限流、灰度和面向客户端的 SSE 生命周期;Python Agent 服务负责图编排、模型调用、工具策略和节点级 Trace。两端通过有版本的内部协议连接,并传递 trace_id、run_id、thread_id、超时预算和取消信号。SSE 断连要向下游传播取消,避免模型继续烧 Token;重试只默认用于幂等读请求,写工具必须带业务幂等键。熔断和降级在调用方保护主站,Python 侧还要对模型与每个工具分别限流、超时和隔离。

text
客户端 ←─ SSE ─→ Java 网关 ── HTTP/SSE 或 gRPC ─→ Python Agent
                  │                                │
          鉴权/租户/限流/灰度                 编排/模型/工具/Harness
                  └──── trace_id · deadline · cancel ────┘
能力Java 主站 / 网关Python Agent 服务
身份与权限校验登录态、租户、设备/资源归属,签发内部身份只接受可信内部身份,并对工具 scope 二次校验
流式连接管理客户端 SSE、心跳、断连和背压产生结构化事件:token、tool、error、done
超时与熔断设置整条请求 deadline,熔断 Agent 依赖并返回降级结果把预算拆给模型和工具,取消未完成任务
可观测性入口 QPS、状态码、TTFT、断连率节点耗时、Token、工具错误、重试和轨迹
灰度按用户/租户选择 Agent 版本记录 Prompt、模型、图和工具 schema 版本

内部事件协议不要只传裸文本,至少包含 event_id、type、sequence、trace_id 和载荷。Java 根据 sequence 保序转发;如果需要断线续传,用 Last-Event-ID 或应用层游标恢复,但要明确哪些事件可以重放。

超时应拆成连接超时、首 Token 时间、Token 间隔超时和总 deadline。客户端断开后,Java 取消 Python 请求,Python 再取消模型流和未完成工具。只在未开始产生副作用且错误可恢复时重试;创建工单等写操作使用稳定幂等键,并在超时后先查询结果。

验证方式: 做四类故障演练:模型首 Token 超时、客户端中途断开、下游工具慢或错误、Agent 服务整体不可用。验证取消传播、连接释放、熔断恢复、幂等写和两端 Trace 是否能用同一 trace_id 串起来。

常见追问:

  1. SSE 与 WebSocket 在 Agent 流式返回场景如何选择?
  2. Java 重试导致 Python 创建两次工单,根因和修复是什么?
  3. 客户端断开后为什么 GPU 仍在生成,取消信号丢在哪一层?
  4. 如何同时观测 TTFT、ITL 和完整任务耗时?

已有主答案: FastAPI SSE 与 WebSocket · 生产级异步批量调用 LLM API

基于 MIT 协议开源 · 内容来自 guocong-bincai/ai-interview-guide