Skip to content

腾讯 Agent 面试题·进阶全新型(下):LLM 工程 / 系统设计(20题) ​

50 道全新高频面试题第 31-50 题,零重复,难度对齐二面深度与三面综合 答题要求:四段式 "问题 → 方案 → 取舍 → 数据佐证"


四、LLM 工程、推理优化与系统底层(10题) ​

L1. 在 LLM 推理服务中,如何实现"请求优先级调度"(VIP 用户优先、实时聊天优先于批量生成任务)? ​

分档:二面深度

高分答题思路:

优先级调度架构:

请求入口
    │
    ▼
[优先级判定]
    ├─ P0(最高):VIP 用户实时对话
    ├─ P1(高):普通用户实时对话
    ├─ P2(中):后台批量生成
    └─ P3(低):离线分析任务
    │
    ▼
[多级队列]
    ├─ P0 Queue ──┐
    ├─ P1 Queue ──┤
    ├─ P2 Queue ──┤ → [调度器] → GPU 推理
    └─ P3 Queue ──┘
    │
    ▼
[调度策略]
    - P0 立即执行(抢占式)
    - P1 在 P0 队列空时执行
    - P2/P3 在 P0+P1 都空时执行
    - P2/P3 有权重比例(如 7:3)

调度器实现:

python
class PriorityScheduler:
    def __init__(self, gpu_pool):
        self.queues = {
            "P0": asyncio.PriorityQueue(),
            "P1": asyncio.PriorityQueue(),
            "P2": asyncio.PriorityQueue(),
            "P3": asyncio.PriorityQueue(),
        }
        self.gpu_pool = gpu_pool
    
    async def submit(self, request, priority="P1"):
        await self.queues[priority].put(request)
    
    async def run(self):
        while True:
            # 1. 按优先级取请求
            request = await self.get_next_request()
            # 2. 分配 GPU
            gpu = await self.gpu_pool.acquire()
            # 3. 执行推理
            asyncio.create_task(self.execute(request, gpu))
    
    async def get_next_request(self):
        # 严格优先级 + 防止低优先级饿死
        for priority in ["P0", "P1", "P2", "P3"]:
            if not self.queues[priority].empty():
                return await self.queues[priority].get()
        
        # 防饿死:每 10 个 P0/P1 请求后,强制取 1 个 P2/P3
        if self.low_priority_counter >= 10:
            self.low_priority_counter = 0
            for priority in ["P2", "P3"]:
                if not self.queues[priority].empty():
                    return await self.queues[priority].get()
        
        # 无请求时等待
        await asyncio.sleep(0.01)

关键技术点:

维度方案
优先级判定请求头/用户标签自动判定,不需要用户手动指定
抢占机制P0 请求到达时,如果有 P2/P3 正在执行 → 标记可中断 → P0 优先
防饿死低优先级请求等待超过 30s → 自动升一级
批量优化P2/P3 请求可以组 batch 提高吞吐,P0/P1 单条优先处理
资源隔离GPU 分两组:实时组(P0/P1)+ 批量组(P2/P3),互不影响

加分点: "优先级调度最大的矛盾是'公平性 vs 延迟'——VIP 用户要低延迟,但普通用户也不能等太久。我们的做法是'资源分组'而不是'抢占'——80% GPU 给实时请求,20% 给批量请求,互不干扰。这样 VIP 延迟 < 500ms,普通用户延迟 < 2s,批量任务不会饿死。"


L2. 设计"投机解码"(Speculative Decoding)方案:用小模型生成草稿 Token,大模型并行校验,如何权衡草稿接受率和额外开销? ​

分档:二面深度

高分答题思路:

投机解码原理:

传统自回归解码:
  Token1 → Token2 → Token3 → Token4 → ...  (串行,每次一次前向传播)

投机解码:
  小模型快速生成 N 个草稿 Token:[T1, T2, T3, T4]
  大模型一次前向传播并行校验:  [T1✓, T2✓, T3✗, ...]
  接受前 2 个(T1, T2),从 T3 位置重新生成
  
  → 4 个 Token 只需 2 次前向传播(小模型 1 次 + 大模型 1 次)
  → 如果草稿全对:4 个 Token 只需 1 次大模型前向传播

关键参数调优:

参数说明典型值
草稿长度 N小模型一次生成的 Token 数4-8
小模型大小草稿模型参数量1B-3B
大模型大小校验模型参数量7B-70B
接受率草稿 Token 被大模型接受的比例60-80%

接受率影响因素:

  • 小模型与大模型的"风格相似度":越相似接受率越高
  • 草稿长度:越长接受率越低(后面的 Token 越可能错)
  • 任务类型:重复性任务(如代码)接受率高,创造性任务接受率低

成本分析:

假设:
  - 大模型前向传播:100ms
  - 小模型前向传播:10ms
  - 草稿长度:5
  - 接受率:70%(平均接受 3.5 个)

传统解码 5 个 Token:5 × 100ms = 500ms
投机解码 5 个 Token:10ms(小模型)+ 100ms(大模型校验)= 110ms
加速比:500/110 ≈ 4.5x

但如果不接受(接受率 0%):
传统解码 5 个 Token:500ms
投机解码:10ms + 100ms + 重新生成 5 × 100ms = 610ms(反而更慢)

适用场景判断:

  • ✅ 适合:代码生成、格式化输出、模板化回复(接受率高)
  • ✅ 适合:有充足 GPU 资源的场景(需要同时加载两个模型)
  • ❌ 不适合:创造性写作、开放性对话(接受率低)
  • ❌ 不适合:GPU 资源紧张(额外显存开销)

加分点: "投机解码的核心权衡是'草稿长度 vs 接受率'。我们在代码生成场景测试发现 N=4 时接受率 75%,N=8 时降到 50%。最优 N 是 4——此时加速比约 3x,且不会因为接受率低导致负优化。"


L3. 当模型需要支持 1M tokens 超长上下文时,除了位置编码外,还有哪些工程手段(稀疏注意力、上下文压缩、分层加载)? ​

分档:二面深度

高分答题思路:

超长上下文工程手段:

手段原理效果代价
稀疏注意力只计算相关位置的 attention减少 O(n²) 到 O(n·log n)可能遗漏远距离依赖
滑动窗口注意力只关注局部窗口O(n·w),w 为窗口大小无法直接访问远距离信息
上下文压缩压缩历史对话为摘要减少 Token 数信息损失
分层加载摘要→检索→精确三层按需加载 Token增加检索延迟
KV Cache 量化压缩 KV Cache减少显存精度损失
分块处理长文本分块独立处理并行化上下文割裂
Flash AttentionIO 优化的注意力计算不改结果,加速 2-3x需要硬件支持

分层加载方案(推荐):

1M Token 文档
    │
    ▼
[预处理] 分层索引
    ├─ Layer 1: 全文摘要(500 token)→ 始终在上下文中
    ├─ Layer 2: 章节摘要(每个 200 token × 50 章 = 10K token)
    └─ Layer 3: 原始分块(每块 2K token × 500 块 = 1M token)
    │
    ▼
[检索时]
    1. 先看 Layer 1 → 判断相关章节
    2. 加载相关 Layer 2 → 定位具体段落
    3. 加载相关 Layer 3 → 获取精确内容
    
    实际加载到模型:500 + 2K + 2K = 4.5K token
    而非全部 1M token

KV Cache 优化:

python
class KVCacheManager:
    def __init__(self, max_cache_size=8192):
        self.max_size = max_cache_size
        self.cache = OrderedDict()
    
    def put(self, token_id, kv_pair):
        if len(self.cache) >= self.max_size:
            # 淘汰最久未访问的 KV
            self.cache.popitem(last=False)
        self.cache[token_id] = kv_pair
    
    def get_relevant(self, query_embedding, top_k=2048):
        """基于注意力分数选择最相关的 KV Cache"""
        scores = {k: compute_attention(v, query_embedding) 
                  for k, v in self.cache.items()}
        top_tokens = sorted(scores, key=scores.get, reverse=True)[:top_k]
        return {k: self.cache[k] for k in top_tokens}

加分点: "1M token 不是'全塞给模型',而是'智能加载'。我们的做法是'摘要常驻 + 按需加载'——500 token 摘要始终在上下文中提供全局视角,具体的原始内容只在需要时检索加载。这样实际 Token 使用量通常不超过 8K,但用户感觉模型能'理解'1M token 的文档。"


L4. 如何在 Python 中实现"零拷贝"数据传输,减少 LLM 推理中的内存复制和序列化开销? ​

分档:二面深度

高分答题思路:

内存复制开销场景:

  • 模型权重从 CPU 加载到 GPU → tensor.to(device)
  • 多进程间传递推理结果 → pickle 序列化/反序列化
  • KV Cache 在请求间共享 → 内存拷贝

零拷贝技术:

技术原理适用场景
内存映射(mmap)文件直接映射到虚拟内存,不拷贝加载模型权重文件
共享内存(shared memory)多进程共享同一物理内存多进程共享模型权重
Tensor 视图创建视图而非拷贝Tensor 切片/转置
ZeroMQ消息传递不拷贝进程间通信
CUDA 零拷贝GPU 直接访问 CPU 内存小批量推理
Pin Memory锁页内存,加速 CPU→GPU 传输数据加载

模型权重零拷贝加载:

python
import mmap

class ZeroCopyModelLoader:
    def load_weights(self, weight_path: str):
        # 传统方式:读取整个文件到内存(拷贝)
        # weights = torch.load(weight_path)  # 慢,占双倍内存
        
        # 零拷贝方式:mmap 映射
        with open(weight_path, "rb") as f:
            # mmap 映射,不实际读取数据
            mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
            # 直接从 mmap 创建 Tensor(零拷贝)
            weights = torch.frombuffer(mm, dtype=torch.float32)
        
        return weights  # 共享同一块物理内存
    
    def to_gpu_zero_copy(self, tensor):
        """CUDA 零拷贝:GPU 直接访问 CPU 锁页内存"""
        pinned = tensor.pin_memory()  # 锁页内存
        # GPU 直接映射这块内存,不实际拷贝
        return pinned.cuda(non_blocking=True)

多进程共享模型权重:

python
import torch.multiprocessing as mp

class SharedModelServer:
    def __init__(self, model_path):
        # 主进程加载模型到共享内存
        self.shared_tensor = self.load_to_shared_memory(model_path)
    
    def worker(self, shared_tensor):
        # 工作进程直接使用共享内存中的权重(零拷贝)
        model = self.build_model_from_shared(shared_tensor)
        while True:
            request = self.get_request()
            result = model.infer(request)
            self.send_result(result)
    
    def start(self, num_workers=4):
        ctx = mp.get_context("spawn")
        workers = [
            ctx.Process(target=self.worker, args=(self.shared_tensor,))
            for _ in range(num_workers)
        ]
        for w in workers:
            w.start()

KV Cache 共享优化:

  • 同一用户的连续请求可以复用前几轮的 KV Cache
  • 用引用计数管理 KV Cache 生命周期
  • 避免在请求间拷贝 KV Cache,传递引用即可

加分点: "零拷贝在 LLM 推理中最大的收益是模型加载——一个 13B 模型权重 26GB,传统加载需要 52GB 内存(文件读入 + Tensor 创建),mmap 方式只需 26GB。加载时间也从 30s 降到 5s。"


L5. 设计"模型热更新"机制:不重启服务、不丢失排队请求的情况下,切换 LLM 模型版本或 LoRA 适配器。 ​

分档:二面深度

高分答题思路:

热更新架构:

[当前版本 v1 运行中]
    ├─ GPU-0: v1 模型,处理中请求 5 个
    ├─ GPU-1: v1 模型,处理中请求 3 个
    └─ 等待队列: 20 个请求
    │
    ▼ 触发热更新(加载 v2)
    │
[阶段 1: v2 加载]
    ├─ GPU-2(备用): 加载 v2 模型
    └─ 新请求路由到 v2
    │
[阶段 2: v1 排空]
    ├─ GPU-0/GPU-1: 继续处理 v1 的进行中请求
    ├─ 不接受新请求到 v1
    └─ v1 队列排空后释放 GPU-0/GPU-1
    │
[阶段 3: v2 扩容]
    ├─ GPU-0/GPU-1 加载 v2
    └─ 所有流量切到 v2
    │
[完成] v1 下线,v2 全量运行

LoRA 热切换(更轻量):

python
class LoraHotSwap:
    """LoRA 适配器热切换(不需要重新加载基础模型)"""
    
    def __init__(self, base_model, gpu_pool):
        self.base_model = base_model
        self.current_lora = None
        self.gpu_pool = gpu_pool
    
    async def swap_lora(self, new_lora_path: str):
        # 1. 加载新 LoRA 到 GPU(不卸载基础模型)
        new_lora = self.load_lora(new_lora_path)
        
        # 2. 原子切换(用锁保护)
        async with self.swap_lock:
            old_lora = self.current_lora
            self.current_lora = new_lora
        
        # 3. 旧 LoRA 延迟卸载(等正在处理的请求完成)
        await self.wait_pending_requests()
        self.unload_lora(old_lora)
        
        # 切换完成,无请求丢失

关键设计点:

维度全量模型热更新LoRA 热切换
切换时间5-10 分钟(需加载模型)< 10 秒
GPU 需求需要备用 GPU不需要额外 GPU
请求影响新请求路由到新版本所有请求立即用新版本
回滚需重新加载旧版本秒级回滚

流量切换策略:

  • 滚动切换:先切 10% → 验证 → 切 50% → 验证 → 100%
  • 蓝绿部署:v1(蓝)全量运行,v2(绿)并行加载,验证后一键切流
  • 金丝雀:v2 只接受 1% 流量,监控指标正常后逐步扩大

加分点: "热更新的核心矛盾是'资源利用率 vs 切换速度'。蓝绿部署最快但需要双倍 GPU,滚动部署省资源但慢。LoRA 热切换是性价比最高的方案——10 秒切换、不占额外 GPU、秒级回滚。如果业务允许用 LoRA,就别做全量模型热更新。"


L6. 在分布式推理中,"张量并行"和"流水线并行"的适用场景和通信开销对比? ​

分档:二面深度

高分答题思路:

维度张量并行(TP)流水线并行(PP)
切分方式每层的权重矩阵切分到多 GPU不同层分配到不同 GPU
通信频率每层都要通信(All-Reduce)只在层边界通信(Send/Recv)
通信量大(与隐藏维度成正比)小(只传激活值)
GPU 利用率高(所有 GPU 同时计算同一层)低(有气泡,后层等前层)
延迟低(层内并行)高(串行通过流水线)
吞吐中(通信开销限制)高(批次填充气泡)
最佳 GPU 数2-8(通信开销随数量增加)4-16+(通信开销小)
适用场景单机多 GPU、低延迟跨机多 GPU、高吞吐

张量并行示意:

GPU-0: 计算 W₁ × X(W 的前半部分)
GPU-1: 计算 W₂ × X(W 的后半部分)
        ↓
All-Reduce: GPU-0 和 GPU-1 结果相加
        ↓
进入下一层(两个 GPU 同时计算下一层)

流水线并行示意:

GPU-0: Layer 1-10  → 传给 GPU-1
GPU-1: Layer 11-20 → 传给 GPU-2
GPU-2: Layer 21-30 → 传给 GPU-3
GPU-3: Layer 31-40 → 输出

气泡问题:GPU-1 等 GPU-0 算完才能开始
解决:微批次填充(Micro-batching)
  Batch1 → GPU-0 → GPU-1 → GPU-2 → GPU-3
  Batch2 → GPU-0 → GPU-1 → GPU-2 → GPU-3
  (流水线填充后,所有 GPU 同时工作)

混合并行(推荐):

单机内:张量并行(TP=4,4 GPU 同机,通信快)
跨机:流水线并行(PP=2,2 台机器,通信少)
总共:TP=4 × PP=2 = 8 GPU

GPU 0-3(机器A):TP 组 1 → 处理 Layer 1-20
GPU 4-7(机器B):TP 组 2 → 处理 Layer 21-40

加分点: "选择 TP 还是 PP 的核心标准是'通信带宽'——同机 GPU 间 NVLink 带宽 600GB/s,适合 TP;跨机网络带宽 50-100GB/s,适合 PP。所以典型配置是'机内 TP + 机间 PP'。如果只有单机多卡,TP 就够了。"


L7. 如何设计"Token 级流控":在 SSE 流式输出中实现精确的速率限制、字数控制和实时断句? ​

分档:二面深度

高分答题思路:

流控需求场景:

  • 速率限制:免费用户 50 token/s,付费用户 200 token/s
  • 字数控制:最多输出 500 字,到了就停
  • 实时断句:在句子边界暂停,提升阅读体验

Token 级流控架构:

python
class TokenStreamController:
    def __init__(self, rate_limit=50, max_tokens=500):
        self.rate_limit = rate_limit  # token/秒
        self.max_tokens = max_tokens
        self.token_count = 0
        self.last_send_time = time.time()
    
    async def stream(self, token_generator):
        buffer = ""
        
        async for token in token_generator:
            self.token_count += 1
            
            # 1. 字数控制
            if self.token_count >= self.max_tokens:
                buffer += "..."  # 截断标记
                yield buffer
                break
            
            # 2. 速率限制
            elapsed = time.time() - self.last_send_time
            expected_elapsed = 1.0 / self.rate_limit
            if elapsed < expected_elapsed:
                await asyncio.sleep(expected_elapsed - elapsed)
            
            # 3. 断句控制
            buffer += token
            if self.is_sentence_boundary(token):
                yield buffer
                buffer = ""
            
            self.last_send_time = time.time()
        
        if buffer:  # 发送剩余内容
            yield buffer
    
    def is_sentence_boundary(self, token: str) -> bool:
        """检测是否在句子边界"""
        sentence_endings = {"。", "!", "?", ".", "!", "?", "\n"}
        return any(ending in token for ending in sentence_endings)

SSE 实现细节:

data: {"token": "今天", "index": 0}

data: {"token": "天气", "index": 1}

data: {"token": "不错", "index": 2, "sentence_end": true}

data: {"token": "。", "index": 3, "sentence_end": true}

data: [DONE]

断句优化策略:

  • 不逐 token 发送(太碎),而是在句子边界批量发送
  • 设置最大缓冲:即使没有句子结束符,缓冲超过 20 token 也强制发送
  • 标点检测:中文句号、英文句号、换行符都作为断句点

速率限制实现:

python
class TokenBucket:
    """令牌桶算法实现速率限制"""
    def __init__(self, rate=50, capacity=50):
        self.rate = rate  # token/秒
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()
    
    async def consume(self, n=1):
        # 补充令牌
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now
        
        # 等待可用
        while self.tokens < n:
            wait_time = (n - self.tokens) / self.rate
            await asyncio.sleep(wait_time)
            self.tokens += wait_time * self.rate
        
        self.tokens -= n

加分点: "Token 级流控最大的价值是'差异化体验'——免费用户限速 50 token/s 让他们感受到'还能用',付费用户 200 token/s 感受到'真快'。断句控制让流式输出更像'打字机效果',阅读体验好很多。"


L8. 解释 Python 的"pymalloc 内存池"机制,以及它在高频 LLM 推理服务中的内存碎片和性能影响。 ​

分档:二面深度

高分答题思路:

pymalloc 原理:

Python 内存管理三层架构:

1. pymalloc(Python 内存分配器)
   - 管理 ≤ 512 字节的小对象
   - 按大小分类(8/16/24/.../512 字节)
   - 每类有独立的内存池(arena → pool → block)
   - 分配/释放不经过系统 malloc,减少系统调用

2. malloc/free(C 标准库)
   - 管理 > 512 字节的对象
   - 直接调用系统 malloc

3. mmap(操作系统)
   - 大块内存分配(如数组、模型权重)
   - 直接映射虚拟内存

在 LLM 推理中的影响:

场景pymalloc 影响问题
高频创建/销毁小对象内存池碎片化Token 计数器、中间结果
长字符串处理超过 512B 走 malloc推理结果字符串
Tensor 操作不经过 pymallocPyTorch 有自己的内存管理
多线程 GILpymalloc 线程不安全 → GIL 保护并发推理受限

内存碎片问题:

python
# 问题场景:高频推理服务中
while True:
    # 每次推理创建大量小对象(dict, str, list)
    result = model.infer(request)  # 创建 ~1000 个小对象
    
    # 处理完后销毁
    del result  # pymalloc 回收到池中,但可能产生碎片
    
    # 长时间运行后:
    # - pymalloc 池有很多碎片(空闲块不连续)
    # - 实际内存占用 > 真实需要
    # - 极端情况下触发频繁 GC

优化方案:

python
# 1. 对象复用(减少创建/销毁)
class ResultPool:
    """复用推理结果对象"""
    _pool = []
    
    @classmethod
    def get(cls):
        if cls._pool:
            obj = cls._pool.pop()
            obj.reset()
            return obj
        return cls()
    
    @classmethod
    def put(cls, obj):
        cls._pool.append(obj)

# 2. 大块数据用 numpy/array(绕过 pymalloc)
# 不用 list,用 numpy array
import numpy as np
tokens = np.empty(2048, dtype=np.int64)  # 连续内存,不走 pymalloc

# 3. 定期重启 worker(暴力但有效)
# gunicorn --max-requests 1000  # 每个 worker 处理 1000 请求后重启

性能影响数据:

  • 不优化:运行 24 小时后内存从 2G 膨胀到 8G(碎片 + 泄漏)
  • 对象池 + numpy:24 小时后内存稳定在 3G
  • 定期重启:内存永远不超过 4G(简单粗暴)

加分点: "pymalloc 在 LLM 推理服务中的最大问题不是性能而是'内存膨胀'——看起来没有内存泄漏,但 RSS 持续增长。这是因为 pymalloc 的 arena 不会归还给操作系统。解决方案是'定期重启 worker'——配置 --max-requests 1000,每个 worker 处理 1000 请求后自动重启,内存归零。"


L9. 设计"多模态推理"的批处理策略:图文混合输入如何统一编码、动态组 Batch 并处理不同模态的序列长度差异? ​

分档:二面深度

高分答题思路:

多模态批处理挑战:

  • 图像和文本的序列长度差异大(图像可能 196-1024 token,文本 10-500 token)
  • 不同图片分辨率不同 → token 数不同
  • batch 内有的请求有图、有的没有
  • 模态间注意力计算复杂

统一编码方案:

原始输入:
  请求A: [图片(224×224)] + "描述这张图"
  请求B: [图片(512×512)] + "图中有几个人?"
  请求C: "今天天气怎么样"(纯文本)

统一编码:
  请求A: [IMG_TOKEN × 196] + [TXT_TOKEN × 5]   → 总长 201
  请求B: [IMG_TOKEN × 1024] + [TXT_TOKEN × 8]  → 总长 1032
  请求C: [TXT_TOKEN × 10]                        → 总长 10

动态组 Batch 策略:

策略说明优点缺点
同模态分组纯文本一组、图文混合一组简单灵活性差
长度分桶按总序列长度分桶(0-200/200-500/500-1K/1K+)减少 padding需要等待
模态+长度双维度先按模态分、再按长度分桶最优利用率实现复杂
动态 padding取 batch 内最长序列做 padding简单短序列浪费计算

动态组 Batch 实现:

python
class MultimodalBatcher:
    BUCKETS = [(0, 256), (256, 512), (512, 1024), (1024, 2048)]
    
    def __init__(self, max_batch_size=8, max_wait_ms=50):
        self.max_batch = max_batch_size
        self.max_wait = max_wait_ms
        self.queues = {bucket: [] for bucket in self.BUCKETS}
    
    async def add(self, request):
        seq_len = request.image_tokens + request.text_tokens
        bucket = self.find_bucket(seq_len)
        self.queues[bucket].append(request)
    
    async def get_batch(self):
        while True:
            # 1. 找到有请求的桶
            for bucket, queue in self.queues.items():
                if len(queue) >= self.max_batch:
                    return queue[:self.max_batch]  # 凑满一批
                
                # 2. 等待超时也出发(避免长等待)
                if queue and self.wait_time(queue[0]) > self.max_wait:
                    return queue
            
            await asyncio.sleep(0.001)
    
    def pad_batch(self, batch):
        """动态 padding 到 batch 内最长序列"""
        max_len = max(req.total_tokens for req in batch)
        for req in batch:
            req.pad_to(max_len)  # padding 到最长
        return batch

注意力优化:

  • 图像 token 之间做局部注意力(只关注相邻图像 patch)
  • 文本 token 做全局注意力(关注所有 token)
  • 图文交叉注意力(文本关注图像)

加分点: "多模态批处理的核心矛盾是'吞吐 vs 延迟'——等凑批提高吞吐但增加延迟。我们的做法是'弹性批大小'——低负载时 batch=1(低延迟),高负载时 batch=8(高吞吐),50ms 超时强制出发。这样在不同负载下都能平衡。"


L10. 当 GPU 显存不足时,"CPU Offloading"和"梯度检查点"在推理阶段的取舍和工程实践? ​

分档:二面深度

高分答题思路:

两种方案对比:

维度CPU Offloading梯度检查点
原理不活跃的模型层权重存 CPU,用时加载 GPU只存部分中间结果,需要时重新计算
显存节省大(可节省 50-80%)中(可节省 30-50%)
延迟增加大(CPU↔GPU 传输 ~10ms/层)中(重新计算 ~5ms/层)
适用阶段推理 + 训练主要是训练
实现复杂度中(需要管理层调度)低(框架原生支持)

推理阶段的取舍:

CPU Offloading(推理时更常用):

python
class CPUOffloadInfer:
    """推理时 CPU Offloading 方案"""
    
    def __init__(self, model, gpu_layers=20, cpu_layers=20):
        # 将模型分成 GPU 部分和 CPU 部分
        self.gpu_layers = model.layers[:gpu_layers].to("cuda")
        self.cpu_layers = model.layers[gpu_layers:].to("cpu")
    
    async def forward(self, input):
        # 1. GPU 层计算
        hidden = self.gpu_layers(input)
        
        # 2. 需要时从 CPU 加载到 GPU
        for layer in self.cpu_layers:
            # 异步预加载下一层到 GPU
            layer.to("cuda", non_blocking=True)
            hidden = layer(hidden)
            # 用完移回 CPU
            layer.to("cpu")
        
        return hidden

优化策略:

  • 层级 offloading:不活跃的层放 CPU,活跃的层放 GPU
  • 预取:计算当前层时异步加载下一层
  • KV Cache offloading:历史 KV Cache 放 CPU,当前轮放 GPU

梯度检查点在推理中的变体(重计算策略):

python
class RecomputeInfer:
    """推理时重计算策略(少存多算)"""
    
    def __init__(self, model, checkpoint_every=5):
        self.model = model
        self.checkpoint_every = checkpoint_every  # 每5层存一次
    
    def forward(self, input):
        hidden = input
        checkpoints = []
        
        for i, layer in enumerate(self.model.layers):
            hidden = layer(hidden)
            
            # 每5层存一次中间结果
            if i % self.checkpoint_every == 0:
                checkpoints.append(hidden.detach())
                # 释放中间 hidden 的梯度
                # (推理时没有梯度,但释放中间结果省显存)
        
        return hidden

工程实践建议:

场景推荐方案原因
模型太大放不下 GPUCPU Offloading唯一选择
模型刚好放得下但 KV Cache 太大KV Cache Offloading权重在 GPU,Cache 在 CPU
batch size 不够大梯度检查点(重计算)省中间结果显存
延迟敏感减少层数/用更小模型Offloading 延迟太大
吞吐优先CPU Offloading + 大 batch用延迟换吞吐

加分点: "推理阶段 CPU Offloading 最有价值的不是'offload 权重'而是'offload KV Cache'——模型权重是固定的可以优化布局,但 KV Cache 随对话增长。我们把最近 5 轮的 KV Cache 放 GPU,之前的放 CPU,这样长对话也能跑。延迟增加约 15%,但支持的最大对话长度翻了 3 倍。"


五、系统设计、工程实践与软技能(10题) ​

S1. 设计一个"Agent 即服务"(Agent-as-a-Service)平台,支持第三方开发者注册自定义 Agent、工具和市场分成。 ​

分档:三面综合

高分答题思路:

平台架构:

┌─────────────────────────────────────────────────────┐
│                   开发者控制台                        │
│  Agent 注册 │ 工具发布 │ 版本管理 │ 收益分析          │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│                   API Gateway                         │
│  认证 │ 限流 │ 计费 │ 路由                            │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│              Agent 运行时引擎                         │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │ Agent A │ │ Agent B │ │ Agent C │  (沙箱隔离)   │
│  └────┬────┘ └────┬────┘ └────┬────┘               │
│       │           │           │                      │
│  ┌────▼───────────▼───────────▼────┐               │
│  │       工具市场(Tool Market)     │               │
│  │  搜索API │ 数据库 │ 翻译 │ ...    │               │
│  └─────────────────────────────────┘               │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│              基础设施层                               │
│  LLM 推理 │ 向量数据库 │ 对象存储 │ 计费系统          │
└─────────────────────────────────────────────────────┘

核心模块设计:

模块设计要点
Agent 注册开发者提交 Prompt + 工具配置 + 评测集 → 平台审核 → 上架
工具市场统一工具 API 规范(OpenAPI 3.0)→ 开发者注册工具 → Agent 按需调用
沙箱隔离每个 Agent 运行在独立容器,不能访问其他 Agent 的数据
计费分成按调用量计费 → 平台抽成 30% → 开发者得 70%
版本管理Agent 版本号 + 灰度发布 + 回滚机制
质量保障上架前必须通过评测集 + 安全审核

计费模型:

用户调用 Agent A(开发者 Alice 的 Agent)
  - Token 消耗:2K input + 1K output
  - 工具调用:3 次(搜索 × 2,翻译 × 1)
  - 总费用:Token 费 ¥0.05 + 工具费 ¥0.03 = ¥0.08
  
费用分配:
  - LLM 推理成本:¥0.02
  - 平台抽成(30%):¥0.018
  - 开发者 Alice 得到:¥0.042

安全审核:

  • Agent Prompt 审查(不能包含恶意指令)
  • 工具安全审查(不能访问用户隐私数据)
  • 上架后持续监控(异常调用模式检测)

加分点: "Agent-as-a-Service 最大的挑战不是技术而是'生态冷启动'——没有开发者就没有 Agent,没有 Agent 就没有用户。解决方案是'平台先做 10 个标杆 Agent',用优质 Agent 吸引用户,再吸引开发者。"


S2. 如何设计"多租户 Agent"的硬隔离方案:知识库、模型配置、对话历史、向量空间完全隔离且可独立计费? ​

分档:三面综合

高分答题思路:

隔离层次:

层次隔离方案实现方式
知识库每租户独立存储S3 按租户分桶 / PostgreSQL 按租户分 schema
向量空间按租户分 collectionMilvus collection per tenant / 或 partition + tenant_id 过滤
对话历史按租户分表/分库PostgreSQL row-level security / 分表
模型配置租户级配置每租户独立的 Prompt/模型参数/工具配置
资源配额租户级限制QPS 限制 / Token 配额 / 存储配额
网络隔离VPC 级隔离(高安全)独立 VPC / 安全组

三种隔离级别:

Level 1: 逻辑隔离(共享资源,按 tenant_id 区分)
  优点:成本低、运维简单
  缺点:隔离弱,可能数据泄露
  适用:中小企业客户

Level 2: 物理隔离(独立数据库/向量库,共享计算)
  优点:隔离强、性能好
  缺点:成本中、运维中
  适用:中大型企业客户

Level 3: 完全隔离(独立 VPC、独立集群)
  优点:隔离最强
  缺点:成本高、运维复杂
  适用:金融/政府客户

租户管理实现:

python
class TenantManager:
    def __init__(self):
        self.tenants = {}  # tenant_id → TenantConfig
    
    def create_tenant(self, tenant_id: str, tier: str = "standard"):
        config = TenantConfig(
            tenant_id=tenant_id,
            tier=tier,
            # 独立资源
            db_schema=f"tenant_{tenant_id}",
            vector_collection=f"tenant_{tenant_id}",
            storage_bucket=f"tenant-{tenant_id}",
            # 配额
            qps_limit=100 if tier == "standard" else 1000,
            token_quota=1_000_000,  # 月 Token 配额
            storage_quota=10 * 1024**3,  # 10GB
        )
        self.tenants[tenant_id] = config
        
        # 初始化租户资源
        await self.init_tenant_resources(config)
        return config
    
    def get_context(self, tenant_id: str) -> TenantContext:
        """获取当前请求的租户上下文"""
        config = self.tenants[tenant_id]
        return TenantContext(
            db_session=self.get_db_session(config.db_schema),
            vector_store=self.get_vector_store(config.vector_collection),
            storage=self.get_storage(config.storage_bucket),
        )

计费实现:

  • 每次请求记录:tenant_id + token 消耗 + 工具调用 + 时间戳
  • 按月汇总生成账单
  • 超出配额 → 降级或拒绝服务

加分点: "多租户隔离的核心权衡是'隔离强度 vs 资源利用率'。我们的做法是'分级隔离'——普通客户逻辑隔离(成本低),企业客户物理隔离(安全高),金融客户完全隔离(合规要求)。这样不同客户付不同价格,资源利用率最大化。"


S3. 设计"Agent 工作流"的可视化编排系统:支持拖拽节点、条件分支、循环、异常处理和版本回滚。 ​

分档:三面综合

高分答题思路:

可视化编排系统架构:

┌──────────────────────────────────────────────┐
│              拖拽式画布编辑器                   │
│                                              │
│  [开始] → [意图识别] → ┬─ [搜索] → [总结]     │
│                       ├─ [数据库] → [格式化]  │
│                       └─ [反问]               │
│                                              │
│  节点类型:LLM │ 工具 │ 条件 │ 循环 │ 延迟     │
└──────────────────┬───────────────────────────┘
                   │ 导出 JSON
┌──────────────────▼───────────────────────────┐
│              工作流引擎                        │
│  解析 JSON → 构建 DAG → 调度执行               │
└──────────────────┬───────────────────────────┘
                   │
┌──────────────────▼───────────────────────────┐
│              版本管理                          │
│  v1.0 → v1.1 → v2.0 → 回滚到 v1.1            │
└──────────────────────────────────────────────┘

工作流 JSON Schema:

json
{
  "version": "1.0.0",
  "nodes": [
    {
      "id": "start",
      "type": "start",
      "next": "intent"
    },
    {
      "id": "intent",
      "type": "llm",
      "config": {
        "prompt": "识别用户意图",
        "model": "gpt-4"
      },
      "next": "branch"
    },
    {
      "id": "branch",
      "type": "condition",
      "branches": [
        {"condition": "intent == 'search'", "next": "search"},
        {"condition": "intent == 'query'", "next": "database"},
        {"condition": "intent == 'unknown'", "next": "ask_user"}
      ]
    },
    {
      "id": "search",
      "type": "tool",
      "config": {"tool_name": "web_search"},
      "next": "summarize"
    }
  ],
  "edges": [...],
  "error_handling": {
    "default": "retry_3_times_then_fallback"
  }
}

版本管理:

  • 每次保存自动生成新版本
  • 支持对比两个版本的差异
  • 一键回滚到任意历史版本
  • 版本发布需要通过测试(在测试环境跑通)

关键技术选型:

  • 前端画布:React Flow / Vue Flow(拖拽节点编辑器)
  • 后端引擎:自研 DAG 执行器 / Temporal / Airflow
  • 版本存储:Git-like(每次保存为一个 commit)

加分点: "可视化编排的最大价值是'降低门槛'——非技术人员也能设计 Agent 工作流。但核心挑战是'表达力 vs 易用性'——过于简单无法表达复杂逻辑,过于复杂则失去可视化优势。我们的做法是'常见模式模板化'——预置 20+ 常用工作流模板,用户基于模板修改。"


S4. 当 Agent 需要调用外部 SaaS API(企业微信、钉钉、飞书)时,如何设计统一的 OAuth 授权、凭证轮转和失效熔断? ​

分档:三面综合

高分答题思路:

统一 OAuth 架构:

┌──────────────────────────────────┐
│        统一 SaaS 连接器           │
│                                  │
│  ┌──────────┐ ┌──────────┐      │
│  │企微Connector│ │钉钉Connector│  │
│  └─────┬────┘ └─────┬────┘      │
│        │            │            │
│  ┌─────▼────────────▼────┐      │
│  │   统一凭证管理器       │      │
│  │  - OAuth 授权流程      │      │
│  │  - Token 刷新          │      │
│  │  - 凭证轮转             │      │
│  │  - 失效熔断             │      │
│  └───────────────────────┘      │
└──────────────────────────────────┘

凭证管理:

python
class CredentialManager:
    def __init__(self):
        self.credentials = {}  # tenant_id + saas_type → Credential
    
    async def get_valid_token(self, tenant_id, saas_type) -> str:
        cred = self.credentials[(tenant_id, saas_type)]
        
        # 1. 检查是否过期
        if cred.is_expired():
            # 2. 尝试刷新
            try:
                cred = await self.refresh_token(cred)
            except RefreshFailedError:
                # 3. 刷新失败 → 熔断
                self.circuit_breaker.trip(tenant_id, saas_type)
                raise SaaSUnavailableError(f"{saas_type} 授权已失效,请重新授权")
        
        return cred.access_token
    
    async def refresh_token(self, cred) -> Credential:
        """用 refresh_token 刷新 access_token"""
        # 每个平台的刷新流程不同
        handler = self.get_handler(cred.saas_type)
        new_cred = await handler.refresh(cred.refresh_token)
        
        # 更新存储
        self.credentials[(cred.tenant_id, cred.saas_type)] = new_cred
        return new_cred

熔断机制:

  • 连续 3 次刷新失败 → 熔断 5 分钟
  • 熔断期间直接返回"服务暂不可用"
  • 熔断后自动尝试恢复(半开状态)
  • 通知管理员重新授权

凭证轮转:

  • access_token 有效期通常 2 小时
  • 提前 5 分钟刷新(不等过期才刷新)
  • refresh_token 有效期 30 天 → 接近过期时通知用户重新授权

加分点: "统一 SaaS 连接器最大的坑是'各平台 OAuth 实现不一致'——企微用 corpid+corpsecret、钉钉用 appkey+appsecret、飞书用 app_id+app_secret。我们的做法是'适配器模式'——每个平台一个 Connector 实现,对外统一接口。凭证管理器只管'有没有有效 token',不关心是哪个平台的。"


S5. 设计"Agent 审计日志"系统:完整记录决策链路(输入→路由→检索→工具→生成),满足金融/医疗行业的合规溯源要求。 ​

分档:三面综合

高分答题思路:

审计日志结构:

json
{
  "trace_id": "req_20240715_001",
  "timestamp": "2024-07-15T10:30:00Z",
  "user_id": "user_123",
  "tenant_id": "bank_abc",
  "session_id": "sess_456",
  
  "input": {
    "user_message": "查询我的贷款余额",
    "context": {"previous_intent": "greeting"}
  },
  
  "decision_chain": [
    {
      "step": 1,
      "type": "intent_classification",
      "model": "gpt-4",
      "input": "查询我的贷款余额",
      "output": {"intent": "loan_balance_query", "confidence": 0.95},
      "latency_ms": 120,
      "tokens": {"input": 50, "output": 20}
    },
    {
      "step": 2,
      "type": "tool_call",
      "tool": "loan_api",
      "params": {"user_id": "user_123"},
      "result": {"balance": 50000, "currency": "CNY"},
      "latency_ms": 350
    },
    {
      "step": 3,
      "type": "generation",
      "model": "gpt-4",
      "input": "贷款余额 50000 元",
      "output": "您的贷款余额为 5 万元人民币。",
      "latency_ms": 200,
      "tokens": {"input": 100, "output": 30}
    }
  ],
  
  "output": "您的贷款余额为 5 万元人民币。",
  "total_latency_ms": 670,
  "total_cost_cny": 0.008,
  "compliance": {
    "data_classification": "confidential",
    "retention_days": 365,
    "audit_required": true
  }
}

合规要求满足:

要求实现
可溯源trace_id 贯穿全链路,每步有输入/输出记录
不可篡改日志写入后只读(WORM 存储 / 区块链存证)
数据分级自动标注数据敏感级别(公开/内部/机密/绝密)
保留周期金融:7 年 / 医疗:10 年(按法规配置)
访问控制审计日志只有合规人员可查看
脱敏敏感字段(身份证/银行卡)脱敏后存储

技术实现:

python
class AuditLogger:
    def __init__(self, storage):
        self.storage = storage  # Elasticsearch + S3 (WORM)
    
    async def log_decision(self, trace_id, step, decision):
        # 脱敏处理
        decision = self.desensitize(decision)
        
        # 写入 ES(可搜索)
        await self.storage.es.index(
            index=f"audit-{datetime.now():%Y.%m}",
            id=trace_id,
            document=decision
        )
        
        # 写入 S3(不可篡改,WORM 模式)
        await self.storage.s3.put_object(
            Bucket="audit-logs-worm",
            Key=f"{trace_id}/{step}.json",
            Body=json.dumps(decision),
            # S3 Object Lock: Compliance mode, Retain 7 years
        )
    
    def desensitize(self, data):
        """脱敏处理"""
        SENSITIVE_PATTERNS = {
            r"\d{18}": "[身份证已脱敏]",      # 身份证
            r"\d{16,19}": "[银行卡已脱敏]",    # 银行卡
            r"\d{11}": "[手机号已脱敏]",       # 手机号
        }
        # 递归处理所有字符串字段
        ...

加分点: "审计日志的核心不是'记录什么'而是'怎么保证不被篡改'。金融场景我们用 S3 Object Lock(Compliance Mode)——写入后 7 年内任何人都不能删除/修改,包括 root 账号。这是合规审计的硬性要求。"


S6. 如何设计"Agent A/B 测试"框架:对比两个 Prompt 版本或两个模型版本在业务指标(转化率、满意度、成本)上的显著性差异? ​

分档:三面综合

高分答题思路:

A/B 测试框架:

用户请求
    │
    ▼
[流量分配]
    - 用户 ID hash → 50% A组 / 50% B组
    - 同一用户始终在同一组(避免体验不一致)
    ▼
[执行]
    ├─ A组 → Agent v1(旧 Prompt)
    └─ B组 → Agent v2(新 Prompt)
    ▼
[指标采集]
    - 转化率:用户是否完成目标行为
    - 满意度:点赞/点踩 + 代理指标
    - 成本:Token 消耗 + 工具调用费用
    - 延迟:端到端响应时间
    ▼
[统计分析]
    - 假设检验(t-test / Mann-Whitney U)
    - 置信区间
    - 最小样本量计算
    - 显著性判断(p < 0.05)

统计显著性判断:

python
from scipy import stats

class ABTestAnalyzer:
    def analyze(self, group_a: list, group_b: list, metric: str) -> dict:
        # 1. 描述性统计
        a_mean, a_std = np.mean(group_a), np.std(group_a)
        b_mean, b_std = np.mean(group_b), np.std(group_b)
        
        # 2. 假设检验
        if self.is_normal(group_a) and self.is_normal(group_b):
            # 正态分布 → t-test
            stat, p_value = stats.ttest_ind(group_a, group_b)
        else:
            # 非正态 → Mann-Whitney U
            stat, p_value = stats.mannwhitneyu(group_a, group_b)
        
        # 3. 效果量
        effect_size = (b_mean - a_mean) / a_std  # Cohen's d
        
        # 4. 结论
        significant = p_value < 0.05 and abs(effect_size) > 0.1
        
        return {
            "metric": metric,
            "a_mean": a_mean,
            "b_mean": b_mean,
            "improvement": (b_mean - a_mean) / a_mean * 100,  # %
            "p_value": p_value,
            "effect_size": effect_size,
            "significant": significant,
            "recommendation": "deploy_B" if significant and b_mean > a_mean else "keep_A"
        }
    
    def min_sample_size(self, effect_size=0.1, alpha=0.05, power=0.8):
        """计算最小样本量"""
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)
        n = 2 * ((z_alpha + z_beta) / effect_size) ** 2
        return int(np.ceil(n))

多指标综合决策:

  • 转化率提升 5%(p=0.02)✓
  • 满意度提升 3%(p=0.08)✗ 不显著
  • 成本降低 10%(p=0.01)✓
  • 延迟增加 15%(p=0.001)✗ 变差

综合判断:转化率提升+成本降低 > 延迟增加 → 部署 B 但优化延迟

加分点: "A/B 测试最常见的坑是'提前结束'——看到 B 组指标好就立刻全量,但样本量不够时统计显著性不稳定。我们要求至少跑 7 天 + 每组 1000+ 样本才能下结论。另一个坑是'多指标比较'——同时看 10 个指标,即使没有差异也可能有 1 个碰巧显著(多重比较问题),需要 Bonferroni 校正。"


S7. 设计"Agent 成本监控"体系:精确追踪每个请求在 Token 消耗、模型调用次数、工具执行时长和向量检索上的成本分摊。 ​

分档:三面综合

高分答题思路:

成本追踪架构:

请求开始 → 创建 CostTracker
    │
    ├─ LLM 调用 #1 → 记录 input_tokens + output_tokens + model
    ├─ LLM 调用 #2 → 记录 ...
    ├─ 工具调用 #1 → 记录 tool_name + duration
    ├─ 向量检索 #1 → 记录 collection + query_count
    └─ 请求结束 → 汇总总成本
    │
    ▼
[成本计算]
    LLM 成本 = Σ(tokens × 单价)
    工具成本 = Σ(调用次数 × 单价)
    向量成本 = Σ(查询次数 × 单价)
    基础设施成本 = 分摊(GPU折旧 + 网络带宽)
    │
    ▼
[报表]
    - 按租户/用户/Agent 维度汇总
    - 按时间维度趋势
    - 异常成本告警

成本追踪实现:

python
class CostTracker:
    PRICING = {
        "gpt-4": {"input": 0.03, "output": 0.06},  # per 1K tokens
        "gpt-3.5": {"input": 0.001, "output": 0.002},
        "embedding": 0.0001,  # per 1K tokens
        "vector_search": 0.0001,  # per query
    }
    
    def __init__(self, tenant_id: str):
        self.tenant_id = tenant_id
        self.records = []
    
    def record_llm_call(self, model, input_tokens, output_tokens):
        cost = (input_tokens / 1000 * self.PRICING[model]["input"] +
                output_tokens / 1000 * self.PRICING[model]["output"])
        self.records.append({
            "type": "llm", "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        })
    
    def record_tool_call(self, tool_name, duration_ms):
        # 工具成本按调用次数计
        cost = self.PRICING.get(tool_name, 0.001)
        self.records.append({
            "type": "tool", "name": tool_name,
            "duration_ms": duration_ms,
            "cost": cost
        })
    
    def record_vector_search(self, collection, num_queries):
        cost = num_queries * self.PRICING["vector_search"]
        self.records.append({
            "type": "vector", "collection": collection,
            "queries": num_queries, "cost": cost
        })
    
    def summary(self) -> dict:
        total = sum(r["cost"] for r in self.records)
        by_type = defaultdict(float)
        for r in self.records:
            by_type[r["type"]] += r["cost"]
        return {
            "tenant_id": self.tenant_id,
            "total_cost": total,
            "by_type": dict(by_type),
            "call_count": len(self.records),
        }

成本优化策略:

  • 高频简单请求 → 用小模型(成本降 90%)
  • 缓存常见查询 → 减少重复 LLM 调用
  • 批量处理 → 减少 API 调用次数
  • 向量检索优化 → 减少不必要的检索

加分点: "成本监控的核心价值不是'算账'而是'优化'——发现某个 Agent 的成本异常高,深挖发现是每轮都在重新检索整个知识库。加了一层查询缓存后成本降低 60%。没有成本追踪,这种问题根本发现不了。"


S8. 当团队从 0 到 1 搭建 Agent 平台,技术选型、人员分工、里程碑规划和风险兜底如何设计? ​

分档:三面综合

高分答题思路:

技术选型:

层次选型理由
LLMGPT-4(主)+ Qwen-7B(蒸馏)主力用大模型,简单场景用小模型降本
框架LangChain + 自研编排层LangChain 快速起步,自研层处理定制需求
向量库Milvus开源、支持大规模、有云服务
后端Python + FastAPI生态好、异步支持
前端Vue 3 + Vite轻量、快速
部署Docker + K8s标准化、可扩展
监控Prometheus + Grafana开源标准

人员分工(6人团队):

Tech Lead(1人):架构设计 + 技术决策 + 代码审查
后端工程师(2人):Agent 引擎 + 工具集成 + API
前端工程师(1人):管理后台 + 用户界面
算法工程师(1人):Prompt 工程 + 模型评测 + RAG 优化
测试/运维(1人):自动化测试 + CI/CD + 监控

里程碑规划:

Phase 1(1-2月):MVP
  - 单 Agent 基础对话
  - 3-5 个核心工具
  - 基础 RAG
  - 目标:内部 Demo 可用

Phase 2(3-4月):Beta
  - 多 Agent 编排
  - 工具市场
  - 评测体系
  - 目标:内部试用

Phase 3(5-6月):GA
  - 多租户
  - 计费系统
  - 安全合规
  - 目标:外部客户可用

Phase 4(7-12月):Scale
  - 性能优化
  - 生态建设
  - 高可用
  - 目标:支撑 100+ 租户

风险兜底:

风险影响兜底方案
LLM 供应商涨价/停服服务中断多模型适配层(GPT/Claude/开源)
核心成员离职进度延迟文档+代码审查+知识共享
技术方向错误返工每月技术评审 + 快速验证
性能不达标无法上线降级方案(小模型+缓存)
安全漏洞数据泄露安全审计 + 渗透测试

加分点: "0 到 1 最大的坑是'过度设计'——一上来就做多 Agent 编排、多租户隔离,结果 3 个月还没出 MVP。正确做法是'先做能用再做好'——Phase 1 用最简单的单 Agent + 3 个工具,验证核心价值,然后再迭代。"


S9. 描述一次你"推翻原有技术方案"的经历:原方案是什么、推翻的核心论据是什么、新方案的量化收益? ​

分档:三面综合

高分答题思路:

STAR 结构答题:

Situation: "我们的 Agent 最初用 LangChain 做 Prompt 编排,所有逻辑通过 Chain 串联。"

Task: "随着业务复杂度增加,Chain 越来越长,最长的 Chain 有 15 步,维护困难。"

Action(推翻过程): "原方案问题:

  1. Chain 中的每一步错误都会中断整个流程,没有优雅降级
  2. 调试困难——15 步 Chain 出错时不知道哪步出了问题
  3. 灵活性差——改变流程顺序需要重写 Chain

推翻论据:

  1. 收集了 2 周的 Badcase 数据,35% 的错误是 Chain 中间步骤失败导致整个流程中断
  2. 竞品分析发现主流 Agent 框架都在用 DAG/状态机模式
  3. 做 POC 验证:用状态机重写了核心流程,Badcase 率从 35% 降到 8%

新方案:

  • 改为状态机模式(自研)
  • 每个状态独立可测试
  • 状态间转换有明确的条件判断
  • 失败时可以从最后一个成功状态恢复"

Result:

| 指标 | 原方案(LangChain) | 新方案(状态机) |
|------|---------------------|-----------------|
| Badcase 率 | 35% | 8% |
| 平均调试时间 | 30min | 5min |
| 流程变更时间 | 2天 | 2小时 |
| 代码复用率 | 20% | 60% |

复盘反思: "推翻方案的勇气来自数据——不是'感觉不好'而是'35% 的 Badcase 证明方案有结构性缺陷'。如果 Badcase 率只有 10%,可能优化就够了,不需要推翻。"

加分点: "推翻方案最大的阻力不是技术而是'沉没成本'——已经投入 2 个月写的代码要扔掉,团队有抵触。我的做法是'用数据说话 + POC 验证'——先做一个小规模 POC 证明新方案更好,用结果说服团队,而不是用'感觉'说服。"


S10. 如果让你设计一个"Agent 开发者大赛"的评测标准和防作弊机制,你会如何设计任务集、评分维度和提交审计? ​

分档:三面综合

高分答题思路:

评测任务集设计:

任务类型题目数量难度考察能力
基础对话5简单意图理解、多轮对话
工具调用5中等工具选择、参数提取
RAG 检索5中等知识检索、信息整合
复杂推理5困难多跳推理、逻辑链
创意生成3困难创造性、多样性
安全对抗2困难防注入、边界处理
隐藏题5未知泛化能力

评分维度:

python
SCORING_RUBRIC = {
    "正确性": {
        "weight": 0.35,
        "metrics": ["事实准确率", "逻辑正确性", "工具调用准确率"]
    },
    "完整性": {
        "weight": 0.20,
        "metrics": ["信息覆盖度", "多角度分析", "边界处理"]
    },
    "效率": {
        "weight": 0.15,
        "metrics": ["Token 消耗", "工具调用次数", "响应时间"]
    },
    "安全性": {
        "weight": 0.15,
        "metrics": ["防注入能力", "敏感信息保护", "拒绝不当请求"]
    },
    "创新性": {
        "weight": 0.10,
        "metrics": ["解决方案新颖度", "用户体验设计"]
    },
    "工程质量": {
        "weight": 0.05,
        "metrics": ["代码结构", "错误处理", "可扩展性"]
    }
}

防作弊机制:

作弊方式检测方法处罚
硬编码答案用隐藏题(选手不知道的题)测试硬编码无法通过隐藏题
人工代答提交运行日志 + 代码审查无完整运行日志 → 取消资格
抄袭他人代码相似度检测(MOSS/plagiarism)相似度 > 60% → 取消资格
调用外部 API网络请求审计(沙箱内执行)检测到未授权网络请求 → 取消资格
利用评测集漏洞评测集有陷阱题(诱导作弊的假答案)掉入陷阱 → 扣分

提交审计流程:

1. 选手提交代码 + 运行日志
2. 自动评测
   - 在沙箱中执行选手代码
   - 跑公开题 + 隐藏题
   - 自动评分
3. 人工审查
   - Top 10 选手代码审查
   - 检查是否有硬编码/作弊
4. 答辩
   - Top 5 选手现场答辩
   - 解释技术方案 + 现场改题
5. 公示
   - 评分明细公示 3 天
   - 接受申诉

评测集保密:

  • 公开题(选手可提前看到):50% → 用于开发调试
  • 隐藏题(比赛时才公布):30% → 测试泛化能力
  • 私密题(只有评委知道):20% → 最终排名用

加分点: "Agent 大赛评测的核心矛盾是'公平 vs 全面'——完全自动化的评测可能遗漏创意性,人工评测主观性太强。我们的做法是'自动评测打基础分 + 人工评审打加分项'——基础分决定能否进 Top 10,加分项决定最终排名。这样既有客观数据又有人文判断。"


答题策略补充 ​

要点说明
LLM 工程题特点需要展示底层理解(CUDA/内存/并行),面试官可能追问实现细节
系统设计题特点画架构图是必备,先画全局再讲细节,注意层次感
软技能题特点用 STAR 结构(Situation-Task-Action-Result),有数据有反思
通用加分主动提到方案的成本/风险/适用边界,展示工程判断力
通用减分只讲理论不讲实践 / 只讲优点不讲缺点 / 没有量化数据