腾讯 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)调度器实现:
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 Attention | IO 优化的注意力计算 | 不改结果,加速 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 tokenKV Cache 优化:
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 传输 | 数据加载 |
模型权重零拷贝加载:
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)多进程共享模型权重:
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 热切换(更轻量):
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 级流控架构:
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 也强制发送
- 标点检测:中文句号、英文句号、换行符都作为断句点
速率限制实现:
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 操作 | 不经过 pymalloc | PyTorch 有自己的内存管理 |
| 多线程 GIL | pymalloc 线程不安全 → GIL 保护 | 并发推理受限 |
内存碎片问题:
# 问题场景:高频推理服务中
while True:
# 每次推理创建大量小对象(dict, str, list)
result = model.infer(request) # 创建 ~1000 个小对象
# 处理完后销毁
del result # pymalloc 回收到池中,但可能产生碎片
# 长时间运行后:
# - pymalloc 池有很多碎片(空闲块不连续)
# - 实际内存占用 > 真实需要
# - 极端情况下触发频繁 GC优化方案:
# 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 实现:
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(推理时更常用):
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
梯度检查点在推理中的变体(重计算策略):
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工程实践建议:
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 模型太大放不下 GPU | CPU 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 |
| 向量空间 | 按租户分 collection | Milvus collection per tenant / 或 partition + tenant_id 过滤 |
| 对话历史 | 按租户分表/分库 | PostgreSQL row-level security / 分表 |
| 模型配置 | 租户级配置 | 每租户独立的 Prompt/模型参数/工具配置 |
| 资源配额 | 租户级限制 | QPS 限制 / Token 配额 / 存储配额 |
| 网络隔离 | VPC 级隔离(高安全) | 独立 VPC / 安全组 |
三种隔离级别:
Level 1: 逻辑隔离(共享资源,按 tenant_id 区分)
优点:成本低、运维简单
缺点:隔离弱,可能数据泄露
适用:中小企业客户
Level 2: 物理隔离(独立数据库/向量库,共享计算)
优点:隔离强、性能好
缺点:成本中、运维中
适用:中大型企业客户
Level 3: 完全隔离(独立 VPC、独立集群)
优点:隔离最强
缺点:成本高、运维复杂
适用:金融/政府客户租户管理实现:
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:
{
"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 刷新 │ │
│ │ - 凭证轮转 │ │
│ │ - 失效熔断 │ │
│ └───────────────────────┘ │
└──────────────────────────────────┘凭证管理:
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 审计日志"系统:完整记录决策链路(输入→路由→检索→工具→生成),满足金融/医疗行业的合规溯源要求。
分档:三面综合
高分答题思路:
审计日志结构:
{
"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 年(按法规配置) |
| 访问控制 | 审计日志只有合规人员可查看 |
| 脱敏 | 敏感字段(身份证/银行卡)脱敏后存储 |
技术实现:
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)统计显著性判断:
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 维度汇总
- 按时间维度趋势
- 异常成本告警成本追踪实现:
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 平台,技术选型、人员分工、里程碑规划和风险兜底如何设计?
分档:三面综合
高分答题思路:
技术选型:
| 层次 | 选型 | 理由 |
|---|---|---|
| LLM | GPT-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(推翻过程): "原方案问题:
- Chain 中的每一步错误都会中断整个流程,没有优雅降级
- 调试困难——15 步 Chain 出错时不知道哪步出了问题
- 灵活性差——改变流程顺序需要重写 Chain
推翻论据:
- 收集了 2 周的 Badcase 数据,35% 的错误是 Chain 中间步骤失败导致整个流程中断
- 竞品分析发现主流 Agent 框架都在用 DAG/状态机模式
- 做 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 | 未知 | 泛化能力 |
评分维度:
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),有数据有反思 |
| 通用加分 | 主动提到方案的成本/风险/适用边界,展示工程判断力 |
| 通用减分 | 只讲理论不讲实践 / 只讲优点不讲缺点 / 没有量化数据 |