Skip to content

第27章:系统设计真题详解

系统设计面试考的是工程判断力,不是背答案。考官真正想知道的是:你在面对一个开放问题时,能不能在 40 分钟内把需求拆清楚、把架构画对、把权衡说明白。

Agent 类系统设计题之所以特别难,是因为它同时需要两套知识:传统分布式系统的基础(状态管理、并发、容错),加上 LLM 特有的不确定性处理(幻觉、延迟、token 成本、Eval 闭环)。把这两套知识拼在一起,是这一章的核心目标。

这章不给你"标准答案",因为系统设计没有标准答案。我给你的是:每道题的思考框架、关键组件和设计决策,以及面试中最容易失分的点。

27.1 系统设计面试方法论

核心框架

Agent 系统设计面试的最大陷阱,是跳过需求分析直接画图。面试官会在 10 分钟后打断你:"这个系统要支持多少并发用户?"这时候你才发现自己设计的是一个无法定义规模的系统。

正确的顺序是四步走:

需求分析 → 高层架构 → 详细设计 → 权衡讨论

需求分析要区分功能需求和非功能需求。功能需求是系统要做什么,非功能需求是系统要做到多好。对于 Agent 系统,非功能需求尤其重要:

维度要问的问题
规模日活用户数?并发任务数?每次任务平均步骤数?
延迟用户能接受等待多久?任务运行多长时间算"长任务"?
准确性任务成功率要求?允许出错率?
可靠性任务中途崩溃怎么办?重试语义是什么?
成本每次任务预算是多少 token?能接受多高的 API 成本?
安全用户数据隔离要求?工具权限边界?

拿到这些约束之前,你的架构设计没有基础。

高层架构要先画组件图,说清楚五件事:

  1. 用户请求从哪里进入
  2. Agent 核心循环在哪里运行
  3. 状态存在哪里
  4. 工具在哪里执行
  5. 结果从哪里返回

详细设计从高层架构里选 2-3 个最关键的组件深挖。"最关键"通常是系统的瓶颈或失败点。对 Agent 系统来说,多半是:状态管理、上下文窗口策略、工具执行层的错误处理。

权衡讨论是拿分的关键。面试官评判 Senior 和 Staff 的核心差异,就是能不能主动说出"我这么设计是因为……,代价是……,什么情况下我会换另一种方案"。

面试里的容量估算

Agent 题也要做容量估算,只是估算单位不再只有 QPS。建议主动报出四类数字:

估算项为什么重要
请求量DAU、峰值并发、每个会话平均轮次,决定 API Gateway、Session Store 和队列规模
Agent 步数每个任务平均 LLM 调用次数和工具调用次数,决定延迟和成本
Token 预算input/output/reasoning token、检索 chunk 数、历史压缩策略,决定 unit economics
副作用量每小时多少写操作、退款、发信、PR 创建,决定幂等、审批和审计压力

面试中不需要算得特别精确,但要展示你知道 Agent 系统的瓶颈常常在"每个请求内部发生了多少步",而不是外层 HTTP QPS。

Agent 系统的特殊挑战

传统系统设计题里没有的几个维度,在 Agent 题里必须考虑:

非确定性输出:同样的输入,不同时间可能产生不同输出。这意味着你需要 Eval 体系,而不仅仅是监控系统。

Token 成本是一等公民:设计 RAG 管线时不能只考虑准确率,要同时考虑 token 消耗。缓存策略、Chunking 粒度、Retrieval 数量,都直接影响成本。

长任务语义:一个运行 30 分钟的 Agent 任务中途崩溃,是从头重来还是从断点恢复?这不是运维问题,是架构问题,必须在设计阶段决定。

工具副作用:Agent 调用工具可能产生不可逆的副作用(发邮件、扣款、写数据库)。幂等性设计、审批流和审计日志在这里不是可选项,是必须项。

权限继承:Agent 不是一个超级用户。它能查什么、改什么,应该严格继承当前用户、租户、角色和任务目的的权限,而不是拿一个全局 service token 到处调用。


27.2 设计企业知识库对话系统

这是 Agent 系统设计面试里的典型高频题。很多企业都会做内部知识库、文档问答或客服知识助手,考察这道题能快速判断候选人是否真的理解 RAG、权限、引用和评估闭环。

需求澄清

开口之前先问几个问题:

  • 知识库规模?(100 万文档 vs 1 亿文档,架构差异很大)
  • 数据来源?(内部 Wiki、PDF 合同、API 文档,解析策略不同)
  • 用户角色?(公开查询 vs 按权限隔离的文档)
  • 对话记忆?(单轮问答 vs 多轮上下文保持)
  • 答案要引用来源?(需要追溯性)

假设我们设计的是:企业内部 HR 知识库,5 万份文档(政策文件、合同模板、FAQ),按租户、部门和角色隔离权限,支持多轮对话,答案必须引用具体文件段落。

高层架构

这张图里有三条关键路径:

  1. 查询路径:用户输入 → 向量检索 + 关键词检索 → 重排序 → LLM 生成 → 输出
  2. 摄入路径:文档 → 解析 → 切片 → Embedding → 入库
  3. 权限路径:每次检索前过滤只属于该用户角色的文档

文档摄入管线详细设计

这是整个系统质量的根基,也是面试中最容易被忽视的部分。

解析层:不同文件类型用不同解析器。PDF 用 pymupdfunstructured,HTML 用 beautifulsoup,Word 用 python-docx。实际工程中 OCR 质量参差不齐,扫描版 PDF 往往产生乱码,需要质量检查步骤。

Chunking 策略:固定长度切片(512 token)简单但会切断句子;语义切片(按标题/段落边界)保留上下文但难以对齐到统一长度。实践中常用的是递归字符切片(先按段落,再按句子,最后按字符数),同时保留相邻块的 overlap(约 10-15%)防止跨块信息丢失。

python
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", "。", ".", " "]  # 中文优先按段落切
)

元数据设计:每个 chunk 必须携带足够的元数据,才能支持权限过滤、版本管理和来源引用。最小集合:doc_idchunk_idtenant_idacl_hash / allowed_rolessource_urldoc_versioncreated_atsection_title。只用 department 做权限过滤通常不够,因为同一部门里也可能有经理文档、HRBP 文档、法务文档等更细粒度权限。

增量更新:文档更新时,不能全量重新 Embedding(成本太高),需要按 doc_id 做增量替换。删除文档时要传播到向量库,避免幽灵文档。

检索层详细设计

单纯的向量检索在企业场景里往往不够用,原因是:

  • 用户经常用准确术语查询("第12条款"、"HR政策2024年版"),向量检索对关键词匹配效果差
  • 向量检索的召回集合(top-50)里往往混杂大量不相关内容

混合检索(Hybrid Search)是现在的主流方案:

稠密检索(向量余弦相似度)
    +
稀疏检索(BM25 关键词匹配)

融合排序(RRF: Reciprocal Rank Fusion)

重排序(Cross-Encoder)

Top-K 精选

重排序这一步是分水岭。用 Cross-Encoder 模型重新计算 query 和每个 chunk 的相关性,可以显著提升最终 Top-5 的质量,但会增加额外延迟,具体取决于模型大小、硬件和候选数量。在延迟敏感场景下,可以只对 Top-20 做 Rerank 而不是 Top-50。

权限前置过滤:在向量检索时就按 tenant_id、角色、ACL 版本等元数据过滤,而不是检索后再过滤。原因有两个:第一,检索后过滤会导致实际返回的 chunk 数量不可预测;第二,未授权 chunk 即使没有展示给用户,也可能通过日志、reranker prompt 或调试 trace 泄露。大多数向量数据库(Qdrant、Weaviate、pgvector)都支持带过滤条件的向量检索,但大规模 ACL 过滤会影响召回和索引性能,生产里常见做法是按 tenant / data domain 分区,再叠加细粒度 metadata filter。

多轮对话的上下文管理

多轮对话的核心难题是:如何让 Agent 在第 5 轮问题里仍然记得第 1 轮提到的合同编号?

两种主流策略:

策略一:完整历史压缩

保留最近 N 轮完整对话,超出时做摘要压缩。

python
def compress_history(messages: list, max_tokens: int = 4000) -> list:
    if count_tokens(messages) <= max_tokens:
        return messages
    # 保留最近 3 轮,其余压缩成摘要
    recent = messages[-6:]
    older = messages[:-6]
    summary = llm.summarize(older, 
                            instruction="保留关键实体:合同编号、人名、决策项")
    return [{"role": "system", "content": f"历史摘要:{summary}"}] + recent

策略二:外置状态存储

把对话中提取的关键实体(用户提到的文档编号、上次的查询主题)存入 Redis,每轮对话开始时注入。

实践中两种策略常组合使用:外置存储处理"硬实体"(文档编号必须精确),压缩摘要处理"软语境"(上下文气氛)。

权限与引用的关键边界

RAG 系统的权限检查不能只发生在检索层。一个更稳的设计是三道闸:

  1. 检索前:根据用户身份和租户生成 filter,只从可访问集合召回。
  2. 生成前:把进入 prompt 的 chunk 重新做一次 ACL 校验,避免缓存或索引延迟导致越权。
  3. 返回前:引用列表只返回用户有权查看的 source_url 和段落,且 trace / 日志中不记录完整敏感 chunk。

这也是企业知识库和普通公开 RAG demo 的根本区别。

常见误区

误区一:用余弦相似度阈值做 Guardrail

"相似度低于 0.7 就不回答"这个规则在实践中几乎不可用。相似度是模型内部的相对度量,不同 Embedding 模型的数值范围差异很大,0.7 对 A 模型可能很高,对 B 模型可能很低。更好的方式是用校准集确定阈值,并结合引用覆盖率、answerability 分类、LLM-as-Judge faithfulness 检查和人工抽检;不要把单一相似度阈值当安全边界。

误区二:忽视 Chunking 质量对答案质量的影响

大多数 RAG 系统的问题不在于 LLM 不够聪明,而在于检索到了质量低的 chunk。调优顺序应该是:先提升 Chunking 质量和检索精度,再考虑换更强的 LLM。

误区三:忘记引用追溯

企业 HR 场景里用户问"这个政策是哪一年的?"没有来源引用的答案没有信任基础。系统必须在生成答案时把 source_urlsection_title 一起返回给前端。

面试加分点

主动提出 Eval 体系:"这个系统上线后,我会设计四类指标:Context Recall(关键信息有没有被检索到)、Faithfulness(答案是否被引用支撑)、Answer Relevance(是否回答了用户问题)、Refusal Accuracy(文档没有答案时是否拒答)。用人工标注集校准 LLM-as-Judge,再用 Ragas / DeepEval 这类工具自动化跑,接入 CI/CD。"


27.3 设计智能客服 Agent

客服 Agent 和知识库问答系统的核心区别:它有副作用。查询订单是读操作,申请退款是写操作,申请退款之后还要触发外部系统。这一步跨过去,整个系统的复杂度至少上升一个量级。

需求澄清

  • 业务范围:只回答问题,还是能执行操作(退款、换货、取消订单)?
  • 人工接管:什么时候需要转人工?用什么信号触发?
  • 多渠道:Web、App、微信/企业微信、电话(语音)?
  • SLA:响应延迟要求?每小时能处理多少并发对话?

假设:电商客服 Agent,支持订单查询、退款申请、物流查询,需要人工接管能力,日处理量 5 万对话。SLA 要拆开说:首响 P95 3 秒以内,完整解决可以是 10-60 秒;退款、取消订单这类写操作以正确性和审批为优先,不应该为了 3 秒 SLA 牺牲安全。

高层架构

意图路由:在进入 Agent 之前分流

不是所有问题都需要跑一遍完整的 Agent 循环。分三层处理:

第一层:FAQ 直接匹配。高频问题("快递几天到"、"退款多久到账")可以用向量相似度直接匹配预置答案,延迟 <100ms,成本极低。这类问题占客服问题量的 40-60%(基于行业经验,具体数字因业务不同而异,需通过实际日志分析)。

第二层:工具调用 Agent。需要查询订单状态、发起退款申请的问题进入 Agent 循环。这类问题有明确的输入输出,工具调用成功率高。

第三层:人工升级。情绪激动、涉及赔偿协商、超出 Agent 权限范围的问题直接转人工。升级触发信号设计是这道题的加分项:

  • 情绪信号:用 Sentiment 分类器检测愤怒/失望情绪
  • 失败信号:同一问题用户连续否定 3 次
  • 金额信号:涉及金额超过阈值自动升级
  • 显式请求:用户说"我要投诉"、"转人工"

工具层设计:客服场景的权限控制

客服 Agent 的工具权限必须严格分级,不能给 Agent 超出业务场景的权限:

python
# 工具权限矩阵
TOOL_PERMISSIONS = {
    "query_order": {
        "scope": "read",
        "max_orders_per_query": 10,
        "require_auth": True  # 必须验证用户身份
    },
    "initiate_refund": {
        "scope": "write",
        "max_refund_amount": 1000,  # 超过此金额需人工审批
        "require_auth": True,
        "require_confirmation": True  # 执行前必须向用户确认
    },
    "cancel_order": {
        "scope": "write",
        "require_auth": True,
        "require_confirmation": True,
        "time_window": "24h"  # 只能取消 24 小时内的订单
    }
}

执行前确认模式(Confirmation Pattern)是高风险操作的标准做法:

Agent:"我将为您申请退款,金额 ¥299,退款方式为原路退回。确认操作吗?"
用户:"确认"
Agent: → 调用 initiate_refund API
Agent:"退款申请已提交,预计 3-5 个工作日到账,工单号 TK-20240501-001234。"

工具调用的幂等性在这里尤为关键。用户网络抖动可能导致前端重发请求,退款接口必须根据 request_id 做幂等去重,避免重复退款。

生产工具调用还要带完整的调用上下文,而不是只传业务参数:

python
class ToolCallContext(BaseModel):
    user_id: str
    tenant_id: str
    conversation_id: str
    auth_level: Literal["anonymous", "verified", "step_up_verified"]
    scopes: list[str]
    idempotency_key: str

class RefundRequest(BaseModel):
    order_id: str
    amount: float
    reason: str
    confirmed_by_user: bool

退款工具执行前应检查:用户是否已验证、订单是否属于该用户、金额是否低于自动处理阈值、是否已得到用户确认、是否已有相同 idempotency key 的成功记录。Agent 只能提出工具调用请求,最终授权应由工具层和策略层决定。

Session 管理与多轮对话

客服场景的对话往往比较短(3-8 轮),但需要跨渠道保持状态(用户从 App 切到 Web 续接)。

Session 数据可以存 Redis,TTL 设 30 分钟(用户离开后自动清理),包含:

  • 当前对话历史
  • 已验证身份的引用(不要把身份证号、手机号、token 原文放进 session)
  • 本次对话中已查询过的订单缓存

如果涉及支付、退款、账户修改,通常还需要 step-up verification:用户在本轮对话中重新通过短信、App push、Passkey 或登录态校验,而不是复用很久以前的登录状态。

人工接管的状态转移

人工接管时,系统必须把完整的对话历史和 Agent 已完成的操作摘要推给坐席,让坐席能立即上下文而不需要让用户重复描述问题。

延迟优化

首响 P95 3 秒的要求对 LLM 调用来说比较紧。几个关键优化点:

  1. FAQ 分流优先:把能直接回答的问题挡在 LLM 调用之前
  2. 流式输出:用 SSE 把首 token 延迟控制在 1 秒以内,用户感知延迟比实际延迟重要
  3. 工具调用并行化:如果 Agent 需要同时查订单和物流,两个工具调用并行执行
  4. 模型分级:简单查询用小模型(降延迟),复杂退款协商用大模型(提准确率)

客服 Agent 的上线指标也要提前说清楚:自动解决率、误操作率、转人工率、用户满意度、平均处理时长、每单成本、投诉升级率。只报"回答准确率"不够,因为客服系统真正关心的是问题是否被安全解决。


27.4 设计代码生成与自动修复 Agent

Coding Agent 是工程师最容易感知、也最能暴露 Agent 工程细节的一类系统(Cursor、GitHub Copilot、Devin 都属于这一类),面试中很适合考察候选人对上下文构建、沙箱、验证和安全边界的理解。

需求澄清

  • 任务类型:单文件代码生成,还是跨文件/跨仓库的修改?
  • 验证方式:用测试用例验证,还是 linter/静态分析?
  • 代码执行环境:在哪里运行生成的代码?
  • 输入格式:自然语言描述,还是 Issue/PR?

假设:自动修复 CI 失败的 Coding Agent。输入是失败的单元测试和对应的代码仓库,输出是修复后的代码变更(PR 或 patch)。

高层架构

上下文构建:Coding Agent 的核心挑战

一个典型的生产代码仓库有几十万行代码,不可能全部塞进上下文窗口。上下文构建是 Coding Agent 质量的决定性因素。

第一步:定位失败原因

解析测试失败日志,提取:

  • 失败的测试函数名
  • 报错的文件和行号
  • 异常栈追踪
python
def parse_test_failure(log: str) -> dict:
    # 提取测试名、报错位置、错误类型
    return {
        "test_name": extract_test_name(log),
        "failed_files": extract_file_locations(log),
        "error_type": classify_error(log),  # AssertionError, TypeError, etc.
        "stack_trace": extract_stack(log)
    }

第二步:相关代码检索

不能只看报错行,还需要:

  • 失败测试的完整函数体
  • 被测试函数的实现
  • 被测试函数调用的其他函数(调用图展开 1-2 层)
  • 相关的类定义和接口

这里推荐使用符号索引(Tree-sitter 做 AST 解析,ctags 做符号表),而不是纯文本搜索——可以精确找到函数定义、调用关系,而不是字符串匹配。

第三步:上下文裁剪

收集到的代码片段可能超过 100K token。需要裁剪:

  • 保留完整的失败测试函数
  • 保留直接相关函数的完整实现
  • 间接依赖函数只保留签名(函数名 + 参数 + 返回类型 + 文档注释)

编辑-执行-验证循环

生成 patch

应用 patch 到工作副本

在沙箱中运行测试

解析测试结果
    ↓ 全部通过
生成最终 PR
    ↓ 还有失败
分析新的失败(是修复了问题还是引入了新问题?)
    ↓ 达到最大轮次
失败:生成诊断报告

防止无限循环的关键:不是简单地限制最大轮次,而是检测进展信号。每轮之后比较失败的测试数量:如果下降了,继续;如果持平或上升了(修复引入了新 bug),触发不同的处理策略——回退上一个 patch,换一种修复方向。

python
def has_progress(before: TestResult, after: TestResult) -> bool:
    return after.failed_count < before.failed_count

def agent_loop(context: Context, max_rounds: int = 10):
    prev_result = run_tests_in_sandbox(context.original_code)
    
    for round_num in range(max_rounds):
        patch = llm.generate_patch(context)
        new_code = apply_patch(context.code, patch)
        new_result = run_tests_in_sandbox(new_code)
        
        if new_result.all_passed:
            return create_pr(patch)
        
        if not has_progress(prev_result, new_result):
            # 没有进展,换策略
            context.add_hint("上一次修复没有减少失败数,请尝试不同的方向")
            context.code = context.original_code  # 回退
        else:
            context.code = new_code
            prev_result = new_result
    
    return create_diagnostic_report(context, prev_result)

沙箱设计

代码执行必须在沙箱里进行,原因很简单:Agent 生成的代码可能包含危险操作(os.system("rm -rf /"、网络请求、文件系统修改)。

沙箱的核心要求:

  • 网络控制:默认禁网或走 allowlist proxy。完全禁网更安全,但会影响依赖下载、文档查询和私有包安装;允许联网时必须限制域名、记录请求并屏蔽 secret。
  • 文件系统隔离:只能读写项目工作目录,不能访问宿主机文件
  • 资源限制:CPU 时间上限、内存上限、磁盘使用上限
  • 可快速克隆:每次测试运行需要一个干净的环境,冷启动时间要短

生产中常见的选型:E2B(Firecracker microVM,专为 AI 代码执行设计)用于托管场景,Docker 容器用于自托管场景。Docker 的文件系统隔离可以通过只读挂载基础镜像 + 可写工作目录来实现。

状态持久化

修复任务可能运行 10-30 分钟,进程崩溃不能丢失进度。每轮编辑-执行后做检查点:

检查点内容:
- 基础 commit hash、当前 patch、必要时保存修改后文件快照或工作区 artifact
- 已运行的轮次
- 每轮的 patch 和测试结果
- Agent 的推理过程摘要

恢复时,可以重建上下文,从最后一个成功检查点继续。

常见误区

误区一:直接把测试输出塞进 Prompt

测试失败输出可能包含大量无关堆栈信息,直接塞进去会污染 Agent 的上下文。应该先做结构化解析,提取关键信息。

误区二:忽视 patch 冲突

如果仓库在 Agent 修复期间有其他提交,patch 会产生冲突。需要在应用 patch 之前检查是否有并发修改。

误区三:假设测试环境与生产一致

沙箱里测试通过不等于生产环境通过。沙箱镜像需要与 CI 环境保持一致,特别是依赖版本。

误区四:让 Agent 覆盖用户未提交改动

Coding Agent 开始前必须读取 git status,确认工作区是否干净;如果不干净,要么创建独立 worktree / branch,要么明确把用户未提交改动当作只读上下文,不能静默覆盖。


27.5 设计多 Agent 数据分析 Pipeline

这道题考察多 Agent 架构的实际工程落地,而不是单纯的 Agent 概念。用多个专用 Agent 协作,比单一 Agent 更适合数据分析场景——原因是分析任务天然可以分解成独立子任务,而且不同子任务需要不同的工具集和专业能力。

需求澄清

  • 数据源:关系型数据库、数据仓库、CSV 文件,还是混合?
  • 分析类型:探索性分析(EDA)、报告生成,还是异常检测?
  • 输出格式:文字摘要、图表、完整报告?
  • 交互模式:单次分析,还是支持追问的对话式分析?

假设:自然语言驱动的数据分析系统,用户用中文提问,系统分析 PostgreSQL 数据库,生成包含图表的分析报告,支持追问。

多 Agent 架构设计

规划 Agent 的设计

规划 Agent 是整个系统的入口,负责两件事:

  1. 理解分析需求:把自然语言需求分解成具体的分析子任务
  2. 协调执行顺序:有些子任务有依赖关系(要先查数据才能画图),有些可以并行(多个独立的数据查询)
python
# 规划 Agent 的输出结构
class AnalysisPlan(BaseModel):
    tasks: list[AnalysisTask]
    dependencies: dict[str, list[str]]  # task_id -> [depends_on_task_ids]
    
class AnalysisTask(BaseModel):
    task_id: str
    agent_type: Literal["sql", "visualization", "report"]
    description: str
    inputs: list[str]  # 依赖的上游 task_id
    
# 示例:用户问"过去 30 天各品类销售额对比"
# 规划结果:
# - task1: sql_agent → 查询过去30天各品类销售额
# - task2: sql_agent → 查询各品类同期对比(依赖 task1 确定品类列表)
# - task3: visualization_agent → 生成柱状图(依赖 task1, task2)
# - task4: report_agent → 撰写分析报告(依赖 task1, task2, task3)

DAG 依赖图让系统可以安全地并行执行无依赖的任务(task1 和 task2 如果相互独立可以同时执行),提升吞吐。

SQL 生成 Agent 的工程细节

SQL Agent 不能直接执行 LLM 生成的 SQL,必须经过验证层:

python
async def sql_agent_execute(task: AnalysisTask, db: Database) -> DataResult:
    # 1. 构建 Schema 上下文(只传相关表,不传全部 schema)
    relevant_tables = retrieve_relevant_tables(task.description, db.schema_index)
    schema_context = db.get_schema(relevant_tables)
    
    # 2. 生成 SQL
    sql = await llm.generate(
        system=SQL_SYSTEM_PROMPT,
        user=f"需求:{task.description}\n\nSchema:{schema_context}"
    )
    
    # 3. 静态验证:用 SQL parser,不要只做字符串匹配
    if not is_valid_sql(sql):
        sql = await llm.fix_sql(sql, validation_errors)
    
    # 4. 安全检查(只允许 SELECT,不允许 DDL/DML)
    if not is_readonly_sql(sql):
        raise SecurityError("只允许只读查询")
    
    # 5. 成本与权限检查
    plan = await db.explain(sql)
    if plan.estimated_rows > 1_000_000 or plan.estimated_cost > MAX_QUERY_COST:
        raise QueryTooExpensive("查询代价过高,请缩小时间范围或过滤条件")
    assert_user_can_access_tables(task.user_id, plan.tables)
    
    # 6. 执行(只读角色、statement timeout、行数限制、RLS)
    result = await db.execute_readonly(sql, row_limit=10000, timeout_seconds=30)
    
    # 7. 结果验证(结果集是否符合预期?列名是否对齐?)
    if not validate_result_schema(result, task.expected_schema):
        return await retry_with_feedback(task, sql, result)
    
    return result

Schema 上下文裁剪是这个 Agent 的关键优化。一个企业数仓可能有几百张表,全部塞进上下文会导致:

  • 超出 token 限制
  • LLM 注意力被稀释,SQL 质量下降

解决方案:用向量检索和业务语义层找到与用户需求最相关的 3-5 张表,只传这些表的 Schema、字段业务含义、常用 join path 和指标定义。这需要提前对所有表的描述(表名 + 列名 + 业务含义)建索引,也需要数据团队维护指标口径,避免模型自己发明 GMV、活跃用户、留存率的定义。

Agent 间数据传递

多 Agent 系统里,Agent 之间传递的数据不能直接放在消息体里——数据量太大,会导致上下文爆炸。正确的方式是:

  • SQL 查询结果存入中间结果存储(Redis 或对象存储),传递的是引用(result_id
  • 图表存入图表存储,传递的是 URL
  • 每个下游 Agent 只在需要时才加载实际数据
python
# 数据引用模式
class DataRef:
    ref_id: str
    ref_type: Literal["query_result", "chart", "text"]
    metadata: dict  # 行数、列名等摘要信息
    
    def load(self) -> Any:
        return storage.get(self.ref_id)

下游 Agent(比如报告 Agent)可以只用 metadata 来决定如何引用数据,而不需要加载完整数据。

评审 Agent:质量保证层

报告生成后,评审 Agent 做一遍检查:

  • 数字是否与查询结果一致(防止 LLM 幻觉)
  • 图表是否与文字描述对应
  • 有没有超出数据范围的结论(比如数据只到昨天,但报告说"当前最新情况")

评审 Agent 的输出是结构化的改进意见,如果评分不达标,规划 Agent 可以触发重写特定部分的子任务。

这个循环就是 Anthropic 文档里描述的 Evaluator-Optimizer 模式 [1],在数据分析场景里特别有效,因为有明确的事实基准(查询结果)可以做核对。

这里的"评审 Agent"不应该只是另一个 LLM 给主观评分。能确定的检查要尽量确定化:数字是否等于查询结果、百分比是否能复算、图表数据是否来自同一个 result_id、报告是否引用了正确时间范围。LLM 更适合检查叙述是否清晰、结论是否过度外推。

常见误区

误区一:所有任务都用同一个 LLM

SQL 生成用大模型质量高但贵。报告摘要可以用中型模型。图表参数提取用小模型或规则就够了。模型路由在多 Agent 场景里的收益比单 Agent 更显著。

误区二:忽视并发安全

多个 Agent 并发写中间结果存储时,需要用 task_id 做命名隔离,防止 task1 和 task2 的结果互相覆盖。

误区三:不设置 Agent 数量上限

规划 Agent 如果把一个简单问题分解成 20 个子 Agent 任务,成本会失控。需要设置最大 Agent 数量和最大总 token 预算,让规划 Agent 在约束条件下做分解。


27.6 关键权衡

前四道题每道都涉及相似的权衡,这里统一梳理,方便面试时快速调取。

Embedding 延迟 vs 检索准确度 vs Token 成本

┌──────────────────┬────────────┬────────────┬────────────┐
│ 方案              │ 延迟        │ 准确度      │ 成本        │
├──────────────────┼────────────┼────────────┼────────────┤
│ 向量检索(仅)    │ 低          │ 中          │ 低          │
│ 混合检索         │ 中          │ 高          │ 中          │
│ 混合 + Rerank    │ 中高        │ 很高        │ 中高        │
│ 全文精确匹配      │ 低          │ 低(语义弱)│ 极低        │
└──────────────────┴────────────┴────────────┴────────────┘

决策原则:如果延迟容忍度高(异步报告生成场景),用混合 + Rerank;如果要求实时(客服对话),用混合检索但做 Rerank 的行数裁剪(Top-20 而不是 Top-50)。

模型能力 vs 推理延迟 vs 成本

这个权衡在 Multi-Agent 场景里最明显——不同子任务可以用不同大小的模型:

任务复杂度高(复杂 SQL 生成、多步规划)→ 大模型
任务明确(简单分类、参数提取)→ 小模型
高频低价值(意图路由)→ 本地模型或规则

面试中说出"模型路由"这个词,并能举出具体的路由规则,是 Senior 工程师的标志。

Agent 自主权 vs 安全风险 vs 用户体验

这三个指标构成一个典型的三角权衡:

  • 提高自主权(减少确认弹窗)→ 更流畅的用户体验,但安全风险上升
  • 降低自主权(每步都确认)→ 安全性高,但用户疲劳,体验差
  • 最优点因业务而异

推荐的分级框架:

操作类型示例处理方式
只读操作查询订单、搜索文档直接执行,不需确认
低风险写操作加购物车、创建草稿执行后通知
中风险写操作提交退款申请执行前确认
高风险写操作删除账号数据、大额支付人工审批
不可逆操作不在 Agent 权限范围内拒绝,给出理由

这个框架来自 OpenAI 的 Safety in Building Agents 指南 [2],在面试中引用官方来源是加分项。

评估闭环 vs 上线速度

Agent 系统设计题里,"怎么证明它能上线"本身就是架构的一部分。一个实用的上线闭环:

  1. 离线评估:用历史真实样本构建 eval set,覆盖正常、边界、恶意和权限场景。
  2. 影子模式:Agent 只生成建议,不执行动作,与人工处理结果对比。
  3. 小流量灰度:只开放低风险用户/低风险工具,观察失败类型。
  4. 持续回归:每次 prompt、模型、工具、检索索引变更都跑 regression eval。

面试时能说出这条路径,说明你不是只会画组件图,而是知道 Agent 的可靠性要靠评估和迭代建立。

幂等性:容易被忽视的生产必需品

任何有副作用的工具调用都必须是幂等的。幂等性的工程实现:

python
import hashlib

def generate_idempotency_key(user_id: str, action: str, params: dict) -> str:
    """生成幂等键:相同的用户、操作、参数组合生成相同的 key"""
    content = f"{user_id}:{action}:{json.dumps(params, sort_keys=True)}"
    return hashlib.sha256(content.encode()).hexdigest()[:32]

async def initiate_refund(order_id: str, amount: float, user_id: str):
    idempotency_key = generate_idempotency_key(
        user_id, "refund", {"order_id": order_id, "amount": amount}
    )
    
    # 原子占位,避免两个 Worker 同时执行退款。
    # SET key value NX EX 86400: 只有第一次请求能拿到执行权。
    acquired = await redis.set(
        f"refund:{idempotency_key}:lock",
        "running",
        nx=True,
        ex=300,
    )
    if not acquired:
        cached = await redis.get(f"refund:{idempotency_key}:result")
        if cached:
            return json.loads(cached)
        raise RetryLater("相同退款请求正在处理中")
    
    # 执行退款
    try:
        result = await payment_api.refund(
            order_id,
            amount,
            idempotency_key=idempotency_key,
        )
    except Exception:
        await redis.delete(f"refund:{idempotency_key}:lock")
        raise
    
    # 缓存结果(TTL 24 小时)
    await redis.setex(f"refund:{idempotency_key}:result", 86400, json.dumps(result))
    return result

这段代码在面试白板上写出来,能直接展示你对副作用处理的理解深度。

真实支付/退款系统里,幂等最好由下游业务系统用唯一约束或官方 idempotency key 兜底。Redis 锁只能减少重复提交,不能替代支付系统的最终幂等保证。


参考资料

[1] Building Effective Agents - Anthropic Engineering (https://www.anthropic.com/engineering/building-effective-agents)

[2] Safety in Building Agents - OpenAI Platform (https://platform.openai.com/docs/guides/agent-builder-safety)

[3] A Practical Guide to Building Agents - OpenAI (https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf)

[4] OpenTelemetry AI agent observability (https://opentelemetry.io/blog/2025/ai-agent-observability/)

[5] Ragas 文档 (https://docs.ragas.io/)

[6] DeepEval 文档 (https://docs.confident-ai.com/)