面灵AI→

快手 Agent 开发一面:工具兜底、多 Agent 与流式解析

轮次
一面
结果
已挂
时间
2026-09
来源
牛客网

《面试题目》

  1. 工具调用失败怎么兜底?
  2. 多 Agent 怎么分工?
  3. 上下文越来越长怎么办?
  4. 怎么衡量输出质量?
  5. 手撕:流式 JSON 解析(处理半包和粘包)
  6. 反问:你们有没有评测集?

《参考解析》

工具调用失败的兜底 要分层,不能一律重试。① 分类:网络超时、限流、瞬时 5xx → 指数退避重试 2~3 次;参数不合法 → 把结构化错误回填让模型重新生成参数;权限不足或资源不存在 → 直接失败,重试无意义;空结果 → 要区分「正常无数据」和「接口异常」。② 降级:重试耗尽后回到模板兜底(生成类任务退到模板文案并打标 degraded=true),保证链路有产出而不是直接报错。③ 幂等:所有写操作带 request_id 幂等键,否则重试会造成重复下单、重复发消息。④ 可观测:每次失败落一条结构化日志(工具名、参数摘要、错误类型、重试次数、是否降级),这是后续判断「该不该继续重试」的唯一依据。⑤ 硬上限:总步数、总耗时、总 token 三重熔断,防止 Agent 在错误反馈里死循环——这是线上最贵的一种故障。

多 Agent 分工 常见形态是流水线 + 共享黑板:策划(出选题和结构)、写稿(生成脚本)、审核(合规与质量校验)三段,中间产物写在共享的黑板或状态对象上,而不是靠消息互相等待,这样各段可以独立重试和替换。工程要点:① 产出契约——每个 Agent 的输出必须是结构化的(JSON schema),否则下游要花大量 token 去理解自然语言;② 不互相阻塞——用任务队列或状态机驱动,而不是一个 Agent 同步调用另一个,否则一个慢节点拖住整条链;③ 上下文隔离——每个 Agent 只拿自己需要的输入,不要把所有历史都传下去,既省 token 又减少干扰;④ 失败隔离——审核不通过应回到写稿重写(限定重试次数),而不是整条链重跑。如果被追问「为什么不用一个大 Agent」,答案是职责单一便于评测和定位问题、可以给不同阶段配不同的模型与温度、阶段之间可以并行。

上下文越来越长的三档压缩 按代价从低到高:① 滑动窗口——只保留最近 N 轮,最便宜但会丢早期信息,适合闲聊型会话;② 摘要压缩——把较早的对话滚动摘要成一段结论,省 token 但必然丢精度,所以金额、时间、编号、硬性约束这类字段要单独结构化保留,不能进摘要;③ 关键记忆抽离——把用户偏好、已确认的事实、未完成事项抽成结构化记忆存外部(键值库或向量库),每轮按相关性召回注入。

配套的是 token 预算管理:给 system prompt、硬约束、检索结果、最近对话各分一个额度,超出时按优先级裁剪,而不是让某一项无限膨胀。选哪一档取决于任务对「记错」的容忍度——生成类可以激进压缩,事务类(订单、退款)必须保留原始结构化事实,摘要里丢掉一个订单号就是事故。

怎么衡量输出质量 单一指标一定会被骗,至少三层:① 规则校验(硬性可判定的部分)——格式是否符合 schema、长度、敏感词、必含要素、有没有编造不存在的品牌或事实;② 模型打分(软性质量)——用小模型或 LLM-as-judge 按维度打分(相关性、流畅度、信息量、指令遵循),注意它的位置偏见和长度偏见,要用成对比较加参考答案做校准;③ 人工抽检与线上指标——人工抽检 badcase、用户反馈(点赞、举报、修改率)、下游业务指标(脚本采纳率)。

再往下一层是评测集本身,要有三类集合:固定回归集(每次改动都跑,防止改好一个坏一个)、困难集(专挑边界样本)、线上采样集(保证分布一致)。原帖提到团队有内部红蓝对抗评测,这属于对抗样本层的补充——主动构造绕过攻击,能发现常规评测集覆盖不到的问题。反问时问「有没有评测集」是个好问题,它直接决定这个团队能不能持续迭代。

手撕流式 JSON 解析 网络分块不保证对齐消息边界,所以必须自己维护缓冲区,处理半包(一条消息被拆到多个 chunk)和粘包(一个 chunk 里有多条消息)。正确做法是字符扫描状态机,而不是「按换行切」:维护 in_string(是否在字符串内)、escape(上一个字符是否是未配对的转义符)、depth(大括号嵌套深度)和一个 buf;每收到一个 chunk 追加到 buf,逐字符扫描,在 depth 归零且不在字符串内时切出一个完整 JSON 片段去解析,剩下的继续留在缓冲区。

buf = ""
in_str = esc = False
depth = 0
BACKSLASH = chr(92)

def feed(chunk):
    global buf, in_str, esc, depth
    buf += chunk
    out = []
    start = 0
    for i, ch in enumerate(buf):
        if in_str:
            if esc:
                esc = False
            elif ch == BACKSLASH:
                esc = True
            elif ch == '"':
                in_str = False
        else:
            if ch == '"':
                in_str = True
            elif ch in "{[":
                depth += 1
            elif ch in "}]":
                depth -= 1
                if depth == 0:
                    out.append(buf[start:i + 1])
                    start = i + 1
    buf = buf[start:]
    return out

容易翻车的点:① 忽略转义,会导致字符串里出现引号时被误判为边界;② 流结束时必须 flush 缓冲区里剩下的数据,否则最后一条丢;③ 单条消息超过缓冲上限要报错,否则内存被打爆;④ 解析失败不要直接丢弃,把原始片段记下来便于排查;⑤ 如果需求是「部分字段先到先用」,要用增量 JSON 解析器,而不是等整条完整。

这场面试的复盘 原帖作者自评「答得一般」,并给出一条很实在的建议:被拷打时不要硬编,「这块我没做过」比扯概念体面。面试官追问的往往是「你有没有真踩过」,编造出来的方案经不住第二层追问(比如「那你当时的缓存命中率是多少」)。没做过的部分可以说清「我会怎么设计、需要验证什么」,把不确定性讲明白,可信度反而更高。