1. 这不是考算法题,是考你能不能把AI真正用起来
“Agent 工程师面试到底考察什么?”——这个问题我被问过至少37次,提问者里有刚刷完LeetCode准备转AI的后端工程师,有带团队落地过5个智能客服项目的PM,也有在大厂做了一年半RAG却突然被要求“加个工单自动分派Agent”的算法同学。他们共同的困惑是:明明简历上写了“熟悉LangChain、调通了Llama3、做过Function Calling”,可一到面试现场,面试官扔过来一道“设计一个处理客户报修工单的Agent”,当场就卡壳了。
这道题背后根本不是考你会不会写prompt,也不是考你能不能背出ReAct、Plan-and-Execute、MRKL这些架构名词。它是在考:你有没有亲手把AI从Demo状态推到生产环境里跑过一圈?是不是真的理解“工单”这个业务实体背后藏着多少非标字段、多少人工判断逻辑、多少系统间的数据断点?是不是清楚当一个用户发来“打印机卡纸还冒烟了”,你的Agent该先查设备型号还是先触发安全告警流程?
我用自己带过的6个真实工单Agent项目(覆盖制造业MES、电商售后中台、SaaS运维平台三类场景)和作为面试官参与的42场Agent岗位终面经历,把这道高频题拆解成一张可执行的检查清单。它不教你怎么背答案,而是告诉你:当面试官说“请设计一个工单Agent”,他其实在等你主动说出这五个关键问题——
“这个工单系统当前有没有API?如果有,返回字段里‘故障描述’是纯文本还是结构化JSON?”
“维修人员排班数据存在哪个库?是MySQL实时查,还是每天凌晨同步到ES?”
“历史工单里,有多少比例的‘已解决’状态其实是客服手动标记的,而非系统自动闭环?”
“当Agent建议派给张三时,张三手机App是否能实时收到推送?如果收不到,降级方案是什么?”
“如果用户补充说‘上次修完三天又坏了’,这个上下文要存多久?存在向量库还是直接追加到当前工单备注?”
这些细节,才是区分“会调API的Prompt工程师”和“能扛住线上流量的Agent工程师”的分水岭。接下来,我会以一道典型面试题为锚点,逐层展开真实项目里踩过的坑、验证过的方案、以及那些从来不会写在招聘JD里,但决定你能否通过终面的隐性能力。
2. 面试题还原:一道真实的工单Agent设计题
我们先看这道被多家公司反复使用的面试题原文(已脱敏):
题目:某家电企业售后服务系统每日接收约8000条客户报修工单,来源包括APP提交、400电话转录、微信小程序。当前工单需由人工坐席完成三件事:① 判断故障类型(如“制冷失效”“噪音异常”“无法开机”);② 根据设备型号、安装年限、保修状态匹配维修师傅;③ 生成初步处理建议(如“先检查电源线”“预约上门检测”)。现计划用AI Agent替代人工初筛环节,请设计该Agent的整体架构,并说明关键模块如何实现。
这道题表面在考架构设计,实则是一张多维度的能力压力测试表。我把它拆解为五个不可回避的硬核子问题,每个都对应着Agent工程师在真实项目中必须直面的战场:
2.1 子问题一:冷启动怎么破?没有标注数据,连baseline都建不起来
几乎所有面试者第一反应都是“上微调”。但现实是:这家企业的历史工单库里,82%的“故障类型”字段为空,剩下18%里,客服随手填的“坏了”“不工作”“有问题”占63%。你拿这种数据去微调Qwen2-7B?模型学出来的不是故障分类,是客服的摸鱼话术分布。
我们实际采用的方案是三层冷启动策略:
第一层:规则兜底(上线前3天必须跑通)
抓取工单标题里的高频关键词:“不制冷”→“制冷失效”,“嗡嗡响”→“噪音异常”,“指示灯不亮”→“无法开机”。用正则+同义词库(我们自建了217个家电故障口语表达映射表)覆盖57%的工单。这部分代码必须能在1小时内写完、测通、上线,因为这是你向业务方证明“AI真能干活”的第一块敲门砖。第二层:小样本学习(第4-14天)
让3个资深维修师傅用半天时间标注200条工单,重点覆盖长尾故障(如“制热时有塑料烧焦味”“除湿模式下出风带白雾”)。用这些数据训练一个轻量级TextCNN模型(参数量<50万),准确率做到89%,比纯规则提升12个百分点。这里的关键不是模型多先进,而是标注成本可控、迭代周期短——师傅们反馈“标200条比写100条SOP还轻松”。第三层:在线学习闭环(第15天起)
当Agent对某条工单给出“制冷失效”判断后,在坐席确认界面增加一个“判断是否正确”按钮。所有被点击“错误”的样本,自动进入待审核队列,由技术同学每天花15分钟清洗后加入训练集。实测运行3个月后,模型在长尾故障上的F1值从61%升至79%,而新增标注成本几乎为零。
提示:面试时如果你只说“用few-shot learning”,面试官大概率会追问“200条样本怎么选?用K-means聚类还是基于故障树采样?”——真实项目里,我们用的是后者。因为家电故障有强物理约束:压缩机故障必然伴随高压管温度异常,所以按设备部件树分层采样,比随机采样效果好23%。
2.2 子问题二:工单字段全是“人话”,怎么变成机器能吃的结构化数据
原始工单字段示例(脱敏):
设备型号:KFR-35GW/BpR3TYA1-B1 购买日期:2022.03.15 故障描述:空调开了半天没冷气,外机风扇转但压缩机不启动,听不到咔哒声 附件:1张外机照片(模糊)、2段15秒语音(有电流声)问题来了:
- “外机风扇转但压缩机不启动”——这是两个独立事实,还是因果关系?
- “听不到咔哒声”——是正常现象(某些机型无启动声)还是故障特征?
- 语音附件里的电流声,需要转文字还是直接喂给音频模型?
我们最终落地的方案是混合解析流水线:
- 文本主干提取:用定制化NER模型识别“外机风扇”“压缩机”“咔哒声”为设备部件,“转”“不启动”“听不到”为状态动词,构建(部件,状态)二元组。这里不用通用大模型,而是用spaCy训练的轻量模型(推理延迟<80ms),因为工单文本长度稳定在30-200字,过度参数化反而拖慢吞吐。
- 语音增强处理:不直接ASR,而是先用开源VAD(Voice Activity Detection)切出有效语音段,再用Whisper-tiny提取MFCC特征,最后输入到一个3层MLP判断“电流声强度等级”(0-5级)。实测发现,电流声>3级时,压缩机启动电容故障概率达89%,这个信号比ASR文字更可靠。
- 图像辅助验证:外机照片不做目标检测(成本高),而是用CLIP-ViT提取图像特征,与“压缩机锈蚀”“散热片堵塞”等12个预设故障图库做余弦相似度比对。当相似度>0.62时,触发“建议检查散热系统”动作。
这套流水线在压测中达到单实例QPS 127(AWS c6i.2xlarge),远超业务要求的80 QPS。关键经验是:不要幻想一个模型解决所有问题,要把工单当成多模态体检报告,每个模态用最适合的工具切一刀。
2.3 子问题三:派工逻辑藏在Excel里,Agent怎么读得懂业务规则
这是最常被忽略的致命点。面试者往往假设“派工=查数据库”,但真实情况是:
- 维修师傅张三的接单范围写着“仅限2020年后生产的变频空调”,但他的技能标签库里只有“格力”“美的”“海尔”;
- 某区域因台风导致200台设备集中报修,系统要求优先派给有高空作业证的师傅,但证书信息存在另一个HR系统里;
- 保修期计算规则是“购买日+365天”,但遇到闰年要额外加1天——这个逻辑写在财务部共享的Excel里,从未接入任何系统。
我们的解法是规则即服务(Rules-as-a-Service):
- 用Python脚本每天凌晨解析Excel规则表,生成JSON Schema描述的规则引擎DSL;
- Agent在决策时,不直接调用数据库,而是向规则引擎发送请求:
{ "rule_id": "assign_priority", "context": { "device_model": "KFR-35GW/BpR3TYA1-B1", "install_date": "2022-03-15", "fault_type": "compressor_not_start" } } - 规则引擎返回结构化结果:
{"priority": 1, "required_cert": ["high_altitude"], "max_distance_km": 15}
这个设计让业务方能自主修改Excel,技术同学只需保证DSL解析器健壮。上线后,规则变更平均耗时从3天缩短到12分钟,且每次变更都有完整审计日志——这点在金融、医疗类工单系统中是强制要求。
注意:很多候选人会提“用LLM解释规则”,这是危险信号。我们做过AB测试:当规则含嵌套条件(如“若设备在保修期且故障属人为损坏,则派单给二级服务商”),LLM解析准确率仅64%,而DSL引擎是100%。AI在这里的角色是增强规则执行,不是替代规则制定。
2.4 子问题四:并发扛不住?别怪模型,先查查你的Redis连接池
“AI Agent怎么扛并发”是热搜词,但90%的面试者答偏了方向。他们狂讲模型量化、vLLM推理优化,却没人提一句:当8000条工单涌进来,你的Agent框架底层用的是什么HTTP客户端?连接池配置多少?Redis锁是用SETNX还是Redlock?
我们真实压测数据:
- 初始版本(LangChain + requests + 单Redis连接):QPS 23,错误率17%(超时+连接拒绝)
- 优化后(FastAPI + httpx + 连接池100 + Redis集群+连接池20):QPS 112,错误率0.3%
关键改动只有三处:
- HTTP客户端换血:requests在高并发下会创建大量TIME_WAIT连接,httpx的异步连接复用让单实例吞吐翻倍;
- Redis连接池扩容:原配置
max_connections=20,压测时发现85%的请求卡在redis.connection.Connection.connect(),调到max_connections=50后延迟下降62%; - 锁粒度细化:不用全局锁控制工单状态更新,而是按“设备型号前缀”分片(如KFR-35*、KFR-51*),把锁竞争从100%降到12%。
这些细节不会出现在任何Agent框架文档里,但它们决定了你的Agent是玩具还是生产系统。面试时如果说“我用LangChain搭了个demo”,不如说“我把LangChain的AsyncCallbackHandler重写了,避免了事件循环阻塞”。
2.5 子问题五:安全不是加个防火墙,是设计时就砍掉攻击面
“Agent安全”在热搜里排第五,但多数人只想到“防越狱prompt”。真实工单场景里,最大的安全风险来自数据泄露链路:
- 用户在故障描述里写了“家里老人独居,地址在XX小区3栋201”;
- Agent生成建议时,可能把“建议上门检测”和地址拼在一起,存进日志;
- 日志系统权限配置失误,导致实习生能查到所有用户地址。
我们的防御体系是纵深防御四层:
- 输入层脱敏:在工单进入Agent前,用正则+NER双校验识别身份证号、手机号、详细地址,替换为
[PHONE]、[ADDRESS]; - 决策层隔离:Agent所有内部思考过程(Chain-of-Thought)禁止出现任何PII字段,用设备ID代替用户ID;
- 输出层过滤:生成的维修建议必须通过规则引擎校验,如含“上门”字样,则强制添加“需用户授权后提供地址”;
- 审计层留痕:所有脱敏操作记录原始值哈希、操作人、时间戳,满足GDPR审计要求。
特别提醒:当面试官问“怎么防prompt注入”,别急着讲对抗样本。先反问:“工单系统是否允许用户上传任意格式附件?”——如果支持PDF,攻击者可能在PDF元数据里埋恶意prompt,这才是真实战场。
3. 架构设计:为什么不用LangChain/LlamaIndex,而选自研编排层
回到面试题核心:“请设计该Agent的整体架构”。几乎所有候选人画的都是这张图:User Input → LLM → Tool Calling → Database → Response
但真实工单Agent的架构图,长得像这样:
[工单接入网关] ↓(HTTP/AMQP) [协议适配层] ← 支持APP/400/微信多源格式转换 ↓ [多模态解析流水线] ← 文本NER+语音VAD+图像CLIP ↓ [结构化工单中心] ← 统一Schema:{device_id, fault_tree_path, urgency_score} ↓ [决策编排引擎] ← 规则DSL + LLM增强 + 人工干预开关 ↓ [执行调度中心] ← 派单API + 短信网关 + App Push SDK ↓ [可观测性总线] ← 全链路Trace + 决策日志 + 数据漂移监控这个架构放弃LangChain不是因为它不好,而是因为LangChain的抽象层级和工单业务的耦合点错位了。举三个具体例子:
3.1 错位一:Tool Calling的语义鸿沟
LangChain的Tool定义是“函数名+描述+参数”,但工单场景里,一个“派单”动作涉及:
- 调用CRM系统API(需Bearer Token)
- 同步更新Elasticsearch工单索引(需Bulk请求)
- 发送企业微信消息(需模板ID+审批流)
- 记录操作日志到MongoDB(需事务一致性)
如果硬套LangChain的Tool,你要写4个独立Tool,然后在prompt里让LLM决定调用顺序——这等于把业务逻辑交给不可控的LLM。我们改成原子动作(Atomic Action)+ 编排策略(Orchestration Policy):
- 原子动作:
assign_to_technician,send_wecom_notice,update_es_index - 编排策略:用JSON Schema定义执行条件,如
{ "action": "assign_to_technician", "condition": "fault_tree_path.startsWith('compressor/') && urgency_score > 3" }
这样,业务逻辑在JSON里可读、可测、可灰度,LLM只负责最擅长的事:从非结构化文本里提取fault_tree_path和urgency_score。
3.2 错位二:记忆管理的性能陷阱
LangChain的ConversationBufferMemory默认把整个对话存Redis,工单场景下,一条工单平均交互5轮,每轮含200字文本+3个JSON字段,单条内存占用超1.2KB。当并发100时,Redis内存暴涨120MB,且LRANGE命令成为瓶颈。
我们改用分层记忆架构:
- 短期记忆(<5分钟):存在本地LRU Cache(1000条),用设备ID哈希分片;
- 中期记忆(<7天):存在Redis Hash,key为
memory:{device_id},field为last_fault_type、repair_history; - 长期记忆(>7天):存入向量库,但只存故障特征向量(768维float),不是原始文本。
最关键的是:所有记忆读写都走异步队列。Agent决策时只读短期记忆,后台Worker定时合并中长期记忆——这让我们在P99延迟<200ms的前提下,支撑了单集群日均12万工单。
3.3 错位三:可观测性的缺失
LangChain的日志是DEBUG级别字符串,而工单Agent必须回答:
- 为什么这条工单被派给了李四而不是王五?
- 故障类型判断依据是文本关键词还是语音分析结果?
- 决策延迟高的10%请求,瓶颈在NER还是规则引擎?
我们自研的可观测性总线包含三个核心组件:
- 决策追踪器(Decision Tracer):每条工单生成唯一trace_id,记录每个决策节点的输入/输出/耗时/置信度;
- 数据漂移监控器(Data Drift Monitor):每天对比新工单的故障类型分布与基线,当“噪音异常”占比突增30%,自动告警并冻结相关规则;
- 人工干预看板(Human-in-the-loop Dashboard):坐席可随时查看Agent决策依据,点击“Override”后,系统自动记录差异点用于模型迭代。
这套设计让故障排查时间从平均47分钟缩短到8分钟,这才是工程化的价值。
4. 实操落地:从面试题到可运行代码的5个关键步骤
现在,我们把面试题转化为可立即运行的最小可行代码(MVP)。这不是玩具Demo,而是我们真实项目第一版上线的精简版,已在测试环境稳定运行23天。
4.1 步骤一:定义工单核心Schema(15分钟)
不写一行LLM代码前,先用Pydantic定义业务契约。这是防止后期返工的最重要一步:
from pydantic import BaseModel, Field, validator from typing import List, Optional, Dict, Any class DeviceInfo(BaseModel): model: str = Field(..., description="设备型号,如KFR-35GW/BpR3TYA1-B1") install_date: str = Field(..., description="安装日期,YYYY-MM-DD格式") warranty_status: str = Field(..., description="保修状态:in_warranty/out_of_warranty") class FaultEvidence(BaseModel): text: str = Field(..., description="故障描述文本") audio_features: Optional[Dict[str, float]] = Field(default=None, description="语音MFCC特征") image_similarity: Optional[Dict[str, float]] = Field(default=None, description="图像与故障图库相似度") class WorkOrder(BaseModel): id: str = Field(..., description="工单ID") device: DeviceInfo evidence: FaultEvidence fault_tree_path: str = Field(default="", description="故障树路径,如compressor/start_failure") urgency_score: float = Field(default=0.0, ge=0.0, le=10.0) @validator('fault_tree_path') def validate_fault_path(cls, v): valid_paths = [ 'compressor/start_failure', 'compressor/overheat', 'fan/noise', 'fan/stop', 'power/no_power' ] if v and v not in valid_paths: raise ValueError(f'Invalid fault_tree_path: {v}') return v实操心得:这个Schema我们和维修主管一起花了2小时敲定。他指着“compressor/start_failure”说:“这个要拆成两级,因为启动失败可能是电容、接触器、主板三个原因”,于是我们加了
root_cause: Optional[str]字段。Schema定义过程就是业务对齐过程,比写代码重要十倍。
4.2 步骤二:实现多模态解析器(45分钟)
重点不是模型多大,而是响应快、错误少:
import re from transformers import pipeline import numpy as np class MultimodalParser: def __init__(self): # 轻量NER模型,3MB,加载<1s self.ner_pipeline = pipeline( "token-classification", model="dslim/bert-base-NER", tokenizer="dslim/bert-base-NER", aggregation_strategy="simple" ) def parse_text(self, text: str) -> Dict[str, Any]: """解析文本,返回结构化故障证据""" # 规则兜底:抓取高频故障词 keywords = { "不制冷": "compressor/no_cooling", "嗡嗡响": "fan/noise", "不启动": "compressor/start_failure", "没风": "fan/stop" } for kw, path in keywords.items(): if kw in text: return {"fault_tree_path": path, "confidence": 0.85} # NER增强:识别部件和状态 ner_results = self.ner_pipeline(text[:512]) parts = [r['word'] for r in ner_results if r['entity_group'] == 'ORG'] states = [r['word'] for r in ner_results if r['entity_group'] == 'MISC'] if parts and states: # 简单映射逻辑(真实项目用决策树) if "压缩机" in parts and "不启动" in states: return {"fault_tree_path": "compressor/start_failure", "confidence": 0.72} return {"fault_tree_path": "", "confidence": 0.0} # 测试 parser = MultimodalParser() result = parser.parse_text("空调开了半天没冷气,外机风扇转但压缩机不启动") print(result) # {'fault_tree_path': 'compressor/start_failure', 'confidence': 0.72}注意:这里故意不用大模型,因为工单文本长度固定、领域封闭。我们实测BERT-base-NER在家电故障NER任务上F1=0.89,而Qwen2-7B是0.91——但延迟从1200ms降到85ms,吞吐提升14倍。在确定性高的场景,小模型是更优解。
4.3 步骤三:构建规则引擎DSL(60分钟)
用JSON Schema定义业务规则,比写代码更贴近业务语言:
import jsonschema from jsonschema import validate import json RULE_SCHEMA = { "type": "object", "properties": { "id": {"type": "string"}, "description": {"type": "string"}, "conditions": { "type": "array", "items": { "type": "object", "properties": { "field": {"type": "string"}, "operator": {"enum": ["eq", "gt", "lt", "startswith", "in"]}, "value": {"type": ["string", "number", "array"]} } } }, "actions": { "type": "array", "items": { "type": "object", "properties": { "type": {"enum": ["assign", "notify", "escalate"]}, "target": {"type": "string"}, "params": {"type": "object"} } } } }, "required": ["id", "conditions", "actions"] } # 示例规则:压缩机故障且紧急度>5,派给高级技师 assign_rule = { "id": "compressor_high_urgency", "description": "压缩机故障高紧急度派单规则", "conditions": [ {"field": "fault_tree_path", "operator": "startswith", "value": "compressor/"}, {"field": "urgency_score", "operator": "gt", "value": 5.0} ], "actions": [ {"type": "assign", "target": "senior_tech", "params": {"cert": "high_voltage"}} ] } # 验证规则合法性 validate(instance=assign_rule, schema=RULE_SCHEMA)实操心得:把规则写成JSON,业务方能直接在Git里PR,技术同学只需写一个DSL解析器。我们用Jinja2模板渲染规则执行SQL,让DBA也能看懂——降低协作门槛,比追求技术炫酷重要得多。
4.4 步骤四:实现决策编排器(90分钟)
这是整个Agent的大脑,用状态机思想设计:
from enum import Enum from dataclasses import dataclass from typing import List, Dict, Any, Optional class DecisionState(Enum): PARSING = "parsing" RULE_MATCHING = "rule_matching" LLM_ENHANCEMENT = "llm_enhancement" EXECUTION = "execution" COMPLETE = "complete" @dataclass class DecisionContext: work_order: WorkOrder state: DecisionState = DecisionState.PARSING rule_matches: List[Dict] = None llm_output: Optional[Dict] = None execution_result: Optional[Dict] = None class DecisionOrchestrator: def __init__(self, rules: List[Dict]): self.rules = rules def run(self, work_order: WorkOrder) -> DecisionContext: ctx = DecisionContext(work_order=work_order) # Step 1: 多模态解析 parser = MultimodalParser() parse_result = parser.parse_text(work_order.evidence.text) work_order.fault_tree_path = parse_result["fault_tree_path"] work_order.urgency_score = self._calc_urgency(parse_result["confidence"]) ctx.state = DecisionState.RULE_MATCHING # Step 2: 规则匹配 ctx.rule_matches = self._match_rules(work_order) if ctx.rule_matches: ctx.state = DecisionState.EXECUTION ctx.execution_result = self._execute_actions(ctx.rule_matches[0], work_order) else: # Step 3: LLM增强(真实项目用Qwen2-1.5B,此处简化) ctx.llm_output = {"suggested_action": "escalate_to_human"} ctx.state = DecisionState.LLM_ENHANCEMENT ctx.state = DecisionState.COMPLETE return ctx def _match_rules(self, wo: WorkOrder) -> List[Dict]: matches = [] for rule in self.rules: match = True for cond in rule["conditions"]: field_val = getattr(wo, cond["field"], None) if cond["operator"] == "startswith": match = match and (field_val and field_val.startswith(cond["value"])) elif cond["operator"] == "gt": match = match and (field_val and field_val > cond["value"]) if match: matches.append(rule) return matches[:1] # 只取最高优先级规则 def _calc_urgency(self, confidence: float) -> float: # 真实项目用更复杂的公式,此处简化 return min(10.0, 5.0 + confidence * 5.0) # 运行测试 orchestrator = DecisionOrchestrator([assign_rule]) wo = WorkOrder( id="WO-2024-001", device=DeviceInfo(model="KFR-35GW/BpR3TYA1-B1", install_date="2022-03-15", warranty_status="in_warranty"), evidence=FaultEvidence(text="空调开了半天没冷气,外机风扇转但压缩机不启动") ) result = orchestrator.run(wo) print(f"Decision: {result.execution_result}")关键点:这个编排器不依赖任何LLM框架,所有逻辑清晰可见。当业务方说“把紧急度阈值从5改成6”,你只需要改一行代码——可维护性是生产系统的生命线。
4.5 步骤五:部署与监控(30分钟)
用Docker Compose一键部署,附带基础监控:
# docker-compose.yml version: '3.8' services: workorder-agent: build: . ports: - "8000:8000" environment: - REDIS_URL=redis://redis:6379/0 - RULES_PATH=/app/rules.json depends_on: - redis healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/health"] interval: 30s timeout: 10s retries: 3 redis: image: redis:7-alpine command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 10s timeout: 5s retries: 5配套健康检查接口:
from fastapi import FastAPI from starlette.responses import JSONResponse app = FastAPI() @app.get("/health") async def health_check(): # 检查Redis连接 try: import redis r = redis.Redis.from_url("redis://localhost:6379/0") r.ping() redis_ok = True except: redis_ok = False # 检查规则加载 try: with open("/app/rules.json") as f: rules = json.load(f) rules_ok = len(rules) > 0 except: rules_ok = False status = "healthy" if redis_ok and rules_ok else "unhealthy" return JSONResponse({ "status": status, "checks": {"redis": redis_ok, "rules": rules_ok} })提示:面试时如果说“我用Docker部署”,不如说“我给健康检查加了Redis连通性和规则完整性双校验,避免容器启动成功但服务不可用”。工程细节才是区分真伪的试金石。
5. 面试避坑指南:那些让你当场出局的致命错误
根据42场终面观察,以下行为出现一次,基本意味着面试结束。这不是主观评判,而是真实项目血泪教训的映射:
5.1 致命错误一:把Agent当成黑盒,不谈数据流向
当面试官问“工单数据怎么进Agent”,你说“用户输入文本,LLM处理后输出结果”,这就踩雷了。真实数据链路是:APP前端 → Nginx负载均衡 → 工单API网关(鉴权/限流) → Kafka Topic(解耦) → Flink实时ETL(清洗/补全) → Agent服务(消费Kafka)
我们曾因忽略Kafka分区策略,导致同一设备的所有工单被分配到不同Agent实例,造成状态不一致。必须说清每个环节的中间件选型理由:
- 为什么用Kafka不用RabbitMQ?—— 因为需要精确一次(exactly-once)语义,保障工单不丢不重;
- 为什么Flink做ETL不用Spark Streaming?—— 因为Flink的事件时间窗口能精准处理“用户补传语音”的乱序问题。
5.2 致命错误二:混淆“能跑通”和“能交付”
很多人演示时说:“我用LangChain搭了个Demo,能处理工单”。但面试官真正想听的是:
- 这个Demo的P95延迟是多少?在什么硬件上测的?
- 当输入含emoji的工单(如“空调❄️不制冷🔥”),你的文本解析会崩溃吗?
- 如果用户发来一段30秒的嘈杂语音,你的VAD模块会切出几个片段?最长片段多长?
我们要求所有Demo必须附带性能基线报告:
| 场景 | 并发数 | P95延迟 | 错误率 | 硬件配置 |
|---|---|---|---|---|
| 纯文本工单 | 50 | 128ms | 0.1% | c6i.2xlarge |
| 含语音工单 | 20 | 842ms | 1.2% | c6i.2xlarge + g5.xlarge GPU |
没有这份报告,你的Demo只是玩具。
5.3 致命错误三:忽视人工协同机制
所有成功的工单Agent,都不是取代人,而是让人做更有价值的事。但90%的候选人只谈“自动派单”,不谈“人机协同”。真实设计必须包含:
- 降级开关:当Agent连续3次判断错误,自动切换到人工坐席队列;
- 置信度透出:在坐席界面上显示“故障类型:压缩机不启动(置信度72%)”,并高亮判断依据(如“文本含‘不启动’,语音MFCC特征匹配压缩机故障谱”);
- 反馈闭环:坐席点击“修正”后,系统自动生成diff报告,供算法同学快速定位模型弱点。
我们上线后,坐席对AI的信任度从31%升至79%,关键不是准确率多高,而是让人类始终掌握最终决策权,并理解AI的思考过程。
5.4 致命错误四:对“安全”的理解停留在prompt层面
当被问“怎么防prompt注入”,如果说“加个system prompt限制”,这暴露了你没碰过真实生产环境。工单场景的安全威胁是:
- 数据泄露:工单附件里的用户身份证照片,被Agent自动OCR后存入日志;
- 越权访问:攻击者构造特殊工单,触发Agent调用内部API获取其他用户信息;
- 资源耗尽:上传1GB无效PDF,撑爆Agent内存。
我们的防护措施是:
- 所有附件在